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];

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

@@ -198,6 +198,77 @@ public void inclusiveRange() {
         });
     }
 
+    @Test
+    public void prefixString() {
+        fdb.run(context -> {
+            fdb.database().run(tr -> {
+                for (int i = 0; i < 5; i++) {
+                    tr.set(subspace.pack(Tuple.from("apple", i)), Tuple.from("apple", i).pack());
+                }
+                return null;
+            });
+            TupleRange range = new TupleRange(
+                    Tuple.from("a"),
+                    Tuple.from("a"),
+                    EndpointType.PREFIX_STRING,
+                    EndpointType.PREFIX_STRING
+            );
+
+            KeyValueCursor cursor = KeyValueCursor.Builder.withSubspace(subspace)
+                    .setContext(context)
+                    .setRange(range)
+                    .setContinuation(null)
+                    .setScanProperties(ScanProperties.FORWARD_SCAN)
+                    .build();
+            for (int j = 0; j < 5; j++) {
+                KeyValue kv = cursor.getNext().get();
+                assertArrayEquals(subspace.pack(Tuple.from("apple", j)), kv.getKey());
+                assertArrayEquals(Tuple.from("apple", j).pack(), kv.getValue());
+            }
+            assertThat(cursor.getNext().hasNext(), is(false));
+
+            cursor = KeyValueCursor.Builder.withSubspace(subspace)
+                    .setContext(context)
+                    .setRange(range)
+                    .setContinuation(null)
+                    .setScanProperties(new ScanProperties(ExecuteProperties.newBuilder().setReturnedRowLimit(2).build()))
+                    .build();
+            for (int j = 0; j < 2; j++) {
+                KeyValue kv = cursor.getNext().get();
+                assertArrayEquals(subspace.pack(Tuple.from("apple", j)), kv.getKey());
+                assertArrayEquals(Tuple.from("apple", j).pack(), kv.getValue());
+            }
+
+            cursor = KeyValueCursor.Builder.withSubspace(subspace)
+                    .setContext(context)
+                    .setRange(range)
+                    .setContinuation(cursor.getNext().getContinuation().toBytes())
+                    .setScanProperties(new ScanProperties(ExecuteProperties.newBuilder().setReturnedRowLimit(2).build()))
+                    .build();
+
+            for (int j = 2; j < 4; j++) {
+                KeyValue kv = cursor.getNext().get();
+                assertArrayEquals(subspace.pack(Tuple.from("apple", j)), kv.getKey());
+                assertArrayEquals(Tuple.from("apple", j).pack(), kv.getValue());
+            }
+
+            cursor = KeyValueCursor.Builder.withSubspace(subspace)
+                    .setContext(context)
+                    .setRange(range)
+                    .setContinuation(cursor.getNext().getContinuation().toBytes())
+                    .setScanProperties(new ScanProperties(ExecuteProperties.newBuilder().setReturnedRowLimit(1).build()))
+                    .build();
+
+            for (int j = 4; j < 5; j++) {
+                KeyValue kv = cursor.getNext().get();
+                assertArrayEquals(subspace.pack(Tuple.from("apple", j)), kv.getKey());
+                assertArrayEquals(Tuple.from("apple", j).pack(), kv.getValue());
+            }
+
+            return null;
+        });
+    }
+
     @Test
     public void exclusiveRange() {
         fdb.run(context -> {