Resolves #1076: [bug] Continuation unused in scanIndex with EndpointType.PREFIX_STRING

Author: hoangmaihuy

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

@@ -246,6 +246,11 @@ protected void prepare() {
             // left (inclusive) and another on the right (exclusive).
             prefixLength = calculatePrefixLength();
 
+            // When both endpoints are prefix strings, we should strip off the last byte \x00 from Tuple
+            if (lowEndpoint == EndpointType.PREFIX_STRING && highEndpoint == EndpointType.PREFIX_STRING) {
+                prefixLength--;
+            }
+
             reverse = scanProperties.isReverse();
             if (continuation != null) {
                 final byte[] continuationBytes = new byte[prefixLength + continuation.length];
@@ -260,6 +265,7 @@ protected void prepare() {
                 }
             }
             final Range byteRange = TupleRange.toRange(lowBytes, highBytes, lowEndpoint, highEndpoint);
+            System.out.println("Range: " + byteRange);
             lowBytes = byteRange.begin;
             highBytes = byteRange.end;
 

fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/KeyValueCursorTest.java

@@ -198,6 +198,75 @@ public void inclusiveRange() {
         });
     }
 
+    private KeyValueCursor scanPrefixString(FDBRecordContext context, Tuple prefixRange, ScanProperties scanProperties, byte[] continuation) {
+        TupleRange range = new TupleRange(
+                prefixRange,
+                prefixRange,
+                EndpointType.PREFIX_STRING,
+                EndpointType.PREFIX_STRING
+        );
+        KeyValueCursor cursor = KeyValueCursor.Builder.withSubspace(subspace)
+                .setContext(context)
+                .setRange(range)
+                .setContinuation(continuation)
+                .setScanProperties(scanProperties)
+                .build();
+        return cursor;
+    }
+
+    private void assertKeyValue(KeyValue kv, Tuple key, Tuple value) {
+        assertArrayEquals(subspace.pack(key), kv.getKey());
+        assertArrayEquals(value.pack(), kv.getValue());
+    }
+
+    @Test
+    public void prefixString() {
+        // Populate data
+        fdb.database().run(tr -> {
+            for (int i = 0; i < 5; i++) {
+                tr.set(subspace.pack(Tuple.from("apple", i)), Tuple.from("apple", i).pack());
+                tr.set(subspace.pack(Tuple.from("banana", i)), Tuple.from("banana", i).pack());
+                tr.set(subspace.pack(Tuple.from("app", i)), Tuple.from("app", i).pack());
+                tr.set(subspace.pack(Tuple.from("a", i)), Tuple.from("a", i).pack());
+            }
+            return null;
+        });
+        fdb.run(context -> {
+            // Scan by empty prefix string, limit 10 rows
+            KeyValueCursor cursor = scanPrefixString(context, Tuple.from(""), new ScanProperties(ExecuteProperties.newBuilder().setReturnedRowLimit(10).build()), null);
+            for (int i = 0; i < 5; i++) {
+                assertKeyValue(cursor.getNext().get(), Tuple.from("a", i), Tuple.from("a", i));
+            }
+            for (int i = 0; i < 5; i++) {
+                assertKeyValue(cursor.getNext().get(), Tuple.from("app", i), Tuple.from("app", i));
+            }
+            // Continue scan next 10 rows
+            cursor = scanPrefixString(context, Tuple.from(""), new ScanProperties(ExecuteProperties.newBuilder().setReturnedRowLimit(10).build()), cursor.getNext().getContinuation().toBytes());
+            for (int i = 0; i < 5; i++) {
+                assertKeyValue(cursor.getNext().get(), Tuple.from("apple", i), Tuple.from("apple", i));
+            }
+            for (int i = 0; i < 5; i++) {
+                assertKeyValue(cursor.getNext().get(), Tuple.from("banana", i), Tuple.from("banana", i));
+            }
+            // Reverse scan with "app" prefix
+            cursor = scanPrefixString(context, Tuple.from("app"), new ScanProperties(ExecuteProperties.newBuilder().setReturnedRowLimit(5).build(), true), null);
+            for (int i = 4; i >= 0; i--) {
+                assertKeyValue(cursor.getNext().get(), Tuple.from("apple", i), Tuple.from("apple", i));
+            }
+            cursor = scanPrefixString(context, Tuple.from("app"), ScanProperties.REVERSE_SCAN, cursor.getNext().getContinuation().toBytes());
+            for (int i = 4; i >= 0; i--) {
+                assertKeyValue(cursor.getNext().get(), Tuple.from("app", i), Tuple.from("app", i));
+            }
+            // Scan with null character
+            cursor = scanPrefixString(context, Tuple.from("a\0"), ScanProperties.FORWARD_SCAN, null);
+            assertEquals(0, cursor.getCount().join());
+            cursor = scanPrefixString(context, Tuple.from("ap\0le"), ScanProperties.FORWARD_SCAN, null);
+            assertEquals(0, cursor.getCount().join());
+
+            return null;
+        });
+    }
+
     @Test
     public void exclusiveRange() {
         fdb.run(context -> {