Fix bug of scan aggregate index returning empty non-end continuation

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/KeyValueCursorBase.java

@@ -145,7 +145,20 @@
         private final byte[] lastKey;
         private final int prefixLength;
         private final SerializationMode serializationMode;
-        private static final long MAGIC_NUMBER = 1234L;
+        /*
+        how we chose this "magic number":
+        The goal is to make sure an old continuation won't be accidentally parsed as: {magic_number = 1234567890L, inner_continuation = some byte array}
+        An example that can be parsed as above is:
+        byte[] data = new byte[] {0x11, (byte) 0xD2, 0x02, (byte) 0x96, 0x49, 0x00, 0x00, 0x00, 0x00, 0x0A, 0x01, 0x14};
+        where 0x11 is wire tag for sfixed64,
+        [(byte) 0xD2, 0x02, (byte) 0x96, 0x49, 0x00, 0x00, 0x00, 0x00] is 1234567890L
+        0x0A is wire tag for bytes
+        0x01 represents that the byte array contains 1 byte
+        0x14 is the value of the byte array
+
+        Because 0xD2 is a negative number in java, it is not possible for a continuation to reach this number.
+         */
+        private static final long MAGIC_NUMBER = 1234567890L;
 
         public Continuation(@Nullable final byte[] lastKey, final int prefixLength, final SerializationMode serializationMode) {
             // Note that doing this without a full copy is dangerous if the array is ever mutated.
@@ -188,13 +201,10 @@
             return byteString.isEmpty() ? new byte[0] : byteString.toByteArray();
         }
 
-        public static byte[] fromRawBytes(@Nullable byte[] rawBytes, SerializationMode serializationMode) {
+        public static byte[] fromRawBytes(@Nullable byte[] rawBytes) {
             if (rawBytes == null) {
                 return null;
             }
-            if (serializationMode == SerializationMode.TO_OLD) {
-                return rawBytes;
-            }
             try {
                 RecordCursorProto.KeyValueCursorContinuation continuationProto = RecordCursorProto.KeyValueCursorContinuation.parseFrom(rawBytes);
                 if (continuationProto.getMagicNumber() != MAGIC_NUMBER) {
@@ -219,7 +229,7 @@
                 return builder.setMagicNumber(MAGIC_NUMBER).build();
             } else {
                 ByteString base = ZeroCopyByteString.wrap(Objects.requireNonNull(lastKey));
                 // even when prefixLength = lastKey.length, proto.getInnerContinuation() = ByteString.EMPTY, proto.hasContinuation() = true
                 return builder.setInnerContinuation(base.substring(prefixLength, lastKey.length)).setMagicNumber(MAGIC_NUMBER).build();
             }
         }
@@ -313,7 +323,7 @@
             reverse = scanProperties.isReverse();
 
             if (continuation != null) {
-                byte[] realContinuation = KeyValueCursorBase.Continuation.fromRawBytes(continuation, serializationMode);
+                byte[] realContinuation = KeyValueCursorBase.Continuation.fromRawBytes(continuation);
                 final byte[] continuationBytes = new byte[prefixLength + realContinuation.length];
                 System.arraycopy(lowBytes, 0, continuationBytes, 0, prefixLength);
                 System.arraycopy(realContinuation, 0, continuationBytes, prefixLength, realContinuation.length);

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/RecordQueryIndexPlan.java

@@ -133,7 +133,7 @@
  */
 @API(API.Status.INTERNAL)
 @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
 public class RecordQueryIndexPlan implements RecordQueryPlanWithNoChildren,
                                              RecordQueryPlanWithComparisons,
                                              RecordQueryPlanWithIndex,
                                              PlannerGraphRewritable,
@@ -180,7 +180,7 @@
         this(indexName, commonPrimaryKey, scanParameters, useIndexPrefetch, fetchIndexRecords, reverse, strictlySorted, Optional.empty(), new Type.Any(), QueryPlanConstraint.noConstraint());
     }
 
     public RecordQueryIndexPlan(@Nonnull final String indexName,
                                 @Nullable final KeyExpression commonPrimaryKey,
                                 @Nonnull final IndexScanParameters scanParameters,
                                 @Nonnull final IndexFetchMethod indexFetchMethod,
@@ -222,7 +222,7 @@
     }
 
     @VisibleForTesting
     public RecordQueryIndexPlan(@Nonnull final String indexName,
                                 @Nullable final KeyExpression commonPrimaryKey,
                                 @Nonnull final IndexScanParameters scanParameters,
                                 @Nonnull final IndexFetchMethod indexFetchMethod,
@@ -310,7 +310,7 @@
         final RecordMetaData metaData = store.getRecordMetaData();
         final Index index = metaData.getIndex(indexName);
         final IndexScanBounds scanBounds = scanParameters.bind(store, index, context);
-        byte[] innerContinuation = KeyValueCursorBase.Continuation.fromRawBytes(continuation, serializationMode);
+        byte[] innerContinuation = KeyValueCursorBase.Continuation.fromRawBytes(continuation);
 
         return store.scanIndexRemoteFetch(index, scanBounds, innerContinuation, executeProperties.asScanProperties(isReverse()), IndexOrphanBehavior.ERROR)
                 .map(store::queriedRecord)
@@ -324,7 +324,7 @@
         final RecordMetaData metaData = store.getRecordMetaData();
         final Index index = metaData.getIndex(indexName);
         final IndexScanBounds scanBounds = scanParameters.bind(store, index, context);
-        byte[] innerContinuation = KeyValueCursorBase.Continuation.fromRawBytes(continuation, serializationMode);
+        byte[] innerContinuation = KeyValueCursorBase.Continuation.fromRawBytes(continuation);
 
         if (!IndexScanType.BY_VALUE_OVER_SCAN.equals(getScanType())) {
             return store.scanIndex(index, scanBounds, innerContinuation, executeProperties.asScanProperties(reverse));
@@ -793,7 +793,7 @@
                 return null;
             }
             // Add the prefix back to the inner continuation
-            byte[] innerContinuation = KeyValueCursorBase.Continuation.fromRawBytes(continuation, serializationMode);
+            byte[] innerContinuation = KeyValueCursorBase.Continuation.fromRawBytes(continuation);
             return ByteArrayUtil.join(prefixBytes, innerContinuation);
         }
 
@@ -802,7 +802,7 @@
             if (continuation.isEnd()) {
                 return continuation;
             }
-            byte[] continuationBytes = KeyValueCursorBase.Continuation.fromRawBytes(continuation.toBytes(), serializationMode);
+            byte[] continuationBytes = KeyValueCursorBase.Continuation.fromRawBytes(continuation.toBytes());
             if (continuationBytes != null && ByteArrayUtil.startsWith(continuationBytes, prefixBytes)) {
                 // Strip away the prefix. Note that ByteStrings re-use the underlying ByteArray, so this can
                 // save a copy.
@@ -835,7 +835,7 @@
                 if (bytes == null) {
                     synchronized (this) {
                         if (bytes == null) {
-                            byte[] baseContinuationBytes = KeyValueCursorBase.Continuation.fromRawBytes(baseContinuation.toBytes(), serializationMode);
+                            byte[] baseContinuationBytes = KeyValueCursorBase.Continuation.fromRawBytes(baseContinuation.toBytes());
                             if (baseContinuationBytes == null) {
                                 return null;
                             }
@@ -849,7 +849,7 @@
             @Nonnull
             @Override
             public ByteString toByteString() {
-                byte[] result = KeyValueCursorBase.Continuation.fromRawBytes(baseContinuation.toBytes(), serializationMode);
+                byte[] result = KeyValueCursorBase.Continuation.fromRawBytes(baseContinuation.toBytes());
                 return result == null ? ByteString.EMPTY : ByteString.copyFrom(result).substring(prefixLength);
             }
 

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/RecordQueryScanPlan.java

@@ -95,8 +95,6 @@ public class RecordQueryScanPlan implements RecordQueryPlanWithNoChildren, Recor
     private final Optional<? extends WithPrimaryKeyMatchCandidate> matchCandidateOptional;
     @Nonnull
     private final Supplier<ComparisonRanges> comparisonRangesSupplier;
-    @Nonnull
-    private final KeyValueCursorBase.SerializationMode serializationMode;
 
     /**
      * Overloaded constructor.
@@ -146,17 +144,6 @@ public RecordQueryScanPlan(@Nullable Set<String> recordTypes,
         this(recordTypes, flowedType, commonPrimaryKey, comparisons, reverse, strictlySorted, Optional.of(matchCandidate));
     }
 
-    public RecordQueryScanPlan(@Nullable Set<String> recordTypes,
-                               @Nonnull Type flowedType,
-                               @Nullable KeyExpression commonPrimaryKey,
-                               @Nonnull ScanComparisons comparisons,
-                               boolean reverse,
-                               boolean strictlySorted,
-                               @Nonnull final Optional<? extends WithPrimaryKeyMatchCandidate> matchCandidateOptional) {
-        this(recordTypes, flowedType, commonPrimaryKey, comparisons, reverse, strictlySorted, matchCandidateOptional, KeyValueCursorBase.SerializationMode.TO_OLD);
-    }
-
-
     /**
      * Overloaded constructor.
      * @param recordTypes a super set of record types of the records that this scan operator can produce
@@ -175,8 +162,7 @@ public RecordQueryScanPlan(@Nullable Set<String> recordTypes,
                                @Nonnull ScanComparisons comparisons,
                                boolean reverse,
                                boolean strictlySorted,
-                               @Nonnull final Optional<? extends WithPrimaryKeyMatchCandidate> matchCandidateOptional,
-                               @Nonnull final KeyValueCursorBase.SerializationMode serializationMode) {
+                               @Nonnull final Optional<? extends WithPrimaryKeyMatchCandidate> matchCandidateOptional) {
         this.recordTypes = recordTypes == null ? null : ImmutableSet.copyOf(recordTypes);
         this.flowedType = flowedType;
         this.commonPrimaryKey = commonPrimaryKey;
@@ -185,7 +171,6 @@ public RecordQueryScanPlan(@Nullable Set<String> recordTypes,
         this.strictlySorted = strictlySorted;
         this.matchCandidateOptional = matchCandidateOptional;
         this.comparisonRangesSupplier = Suppliers.memoize(this::computeComparisonRanges);
-        this.serializationMode = serializationMode;
     }
 
     @Nonnull
@@ -195,7 +180,7 @@ public <M extends Message> RecordCursor<QueryResult> executePlan(@Nonnull final
                                                                      @Nullable final byte[] continuation,
                                                                      @Nonnull final ExecuteProperties executeProperties) {
         final TupleRange range = comparisons.toTupleRange(store, context);
-        byte[] innerContinuation = KeyValueCursorBase.Continuation.fromRawBytes(continuation, serializationMode);
+        byte[] innerContinuation = KeyValueCursorBase.Continuation.fromRawBytes(continuation);
 
         return store.scanRecords(
                 range.getLow(), range.getHigh(), range.getLowEndpoint(), range.getHighEndpoint(), innerContinuation,

fdb-record-layer-core/src/main/proto/record_cursor.proto

@@ -168,5 +168,5 @@ message RangeCursorContinuation {
 
 message KeyValueCursorContinuation {
     optional bytes inner_continuation = 1;
-    optional int64 magic_number = 2;
+    optional sfixed64 magic_number = 2;
 }