KeyValueCursor extends AsyncIteratorCursor<KeyValue> implements BaseCursor<KeyValue> { @Override @Nonnull public RecordCursorResult<KeyValue> getNext() { return context.asyncToSync(FDBStoreTimer.Waits.WAIT_ADVANCE_CURSOR, onNext()); } private  KeyValueCursor(@Nonnull final FDBRecordContext context,
                           @Nonnull final AsyncIterator<KeyValue> iterator,
                           int prefixLength,
                           @Nonnull final CursorLimitManager limitManager,
                           int valuesLimit); @Nonnull @Override CompletableFuture<RecordCursorResult<KeyValue>> onNext(); @Override @Nonnull RecordCursorResult<KeyValue> getNext();  }
@Test public void inclusiveRange() { fdb.run(context -> { KeyValueCursor cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setLow(Tuple.from(3, 3), EndpointType.RANGE_INCLUSIVE) .setHigh(Tuple.from(4, 2), EndpointType.RANGE_INCLUSIVE) .setContinuation(null) .setScanProperties(ScanProperties.FORWARD_SCAN) .build(); assertEquals(Arrays.asList(Tuple.from(3L, 3L), Tuple.from(3L, 4L), Tuple.from(4L, 0L), Tuple.from(4L, 1L), Tuple.from(4L, 2L)), cursor.map(KeyValue::getValue).map(Tuple::fromBytes).asList().join()); cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setLow(Tuple.from(3, 3), EndpointType.RANGE_INCLUSIVE) .setHigh(Tuple.from(4, 2), EndpointType.RANGE_INCLUSIVE) .setContinuation(null) .setScanProperties(new ScanProperties(ExecuteProperties.newBuilder().setReturnedRowLimit(2).build())) .build(); assertEquals(Arrays.asList(Tuple.from(3L, 3L), Tuple.from(3L, 4L)), cursor.map(KeyValue::getValue).map(Tuple::fromBytes).asList().join()); cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setLow(Tuple.from(3, 3), EndpointType.RANGE_INCLUSIVE) .setHigh(Tuple.from(4, 2), EndpointType.RANGE_INCLUSIVE) .setContinuation(cursor.getNext().getContinuation().toBytes()) .setScanProperties(ScanProperties.FORWARD_SCAN) .build(); assertEquals(Arrays.asList(Tuple.from(4L, 0L), Tuple.from(4L, 1L), Tuple.from(4L, 2L)), cursor.map(KeyValue::getValue).map(Tuple::fromBytes).asList().join()); return null; }); }
@Test public void exclusiveRange() { fdb.run(context -> { KeyValueCursor cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setLow(Tuple.from(3, 3), EndpointType.RANGE_EXCLUSIVE) .setHigh(Tuple.from(4, 2), EndpointType.RANGE_EXCLUSIVE) .setContinuation(null) .setScanProperties(ScanProperties.FORWARD_SCAN) .build(); assertEquals(Arrays.asList(Tuple.from(3L, 4L), Tuple.from(4L, 0L), Tuple.from(4L, 1L)), cursor.map(KeyValue::getValue).map(Tuple::fromBytes).asList().join()); cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setLow(Tuple.from(3, 3), EndpointType.RANGE_EXCLUSIVE) .setHigh(Tuple.from(4, 2), EndpointType.RANGE_EXCLUSIVE) .setContinuation(null) .setScanProperties(new ScanProperties(ExecuteProperties.newBuilder().setReturnedRowLimit(2).build())) .build(); assertEquals(Arrays.asList(Tuple.from(3L, 4L), Tuple.from(4L, 0L)), cursor.map(KeyValue::getValue).map(Tuple::fromBytes).asList().join()); cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setLow(Tuple.from(3, 3), EndpointType.RANGE_EXCLUSIVE) .setHigh(Tuple.from(4, 2), EndpointType.RANGE_EXCLUSIVE) .setContinuation(cursor.getNext().getContinuation().toBytes()) .setScanProperties(ScanProperties.FORWARD_SCAN) .build(); assertEquals(Collections.singletonList(Tuple.from(4L, 1L)), cursor.map(KeyValue::getValue).map(Tuple::fromBytes).asList().join()); return null; }); }
@Test public void noNextReasons() { fdb.run(context -> { KeyValueCursor cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setRange(TupleRange.allOf(Tuple.from(3))) .setContinuation(null) .setScanProperties(ScanProperties.FORWARD_SCAN.with(props -> props.setReturnedRowLimit(3))) .build(); assertEquals(Arrays.asList(Tuple.from(3L, 0L), Tuple.from(3L, 1L), Tuple.from(3L, 2L)), cursor.map(KeyValue::getValue).map(Tuple::fromBytes).asList().join()); RecordCursorResult<KeyValue> result = cursor.getNext(); assertEquals(RecordCursor.NoNextReason.RETURN_LIMIT_REACHED, result.getNoNextReason()); cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setRange(TupleRange.allOf(Tuple.from(3))) .setContinuation(result.getContinuation().toBytes()) .setScanProperties(ScanProperties.FORWARD_SCAN.with(props -> props.setReturnedRowLimit(3))) .build(); assertEquals(Arrays.asList(Tuple.from(3L, 3L), Tuple.from(3L, 4L)), cursor.map(KeyValue::getValue).map(Tuple::fromBytes).asList().join()); result = cursor.getNext(); assertEquals(RecordCursor.NoNextReason.SOURCE_EXHAUSTED, result.getNoNextReason()); assertNull(result.getContinuation().toBytes()); return null; }); }
@Test public void simpleScanLimit() { fdb.run(context -> { RecordScanLimiter limiter = RecordScanLimiterFactory.enforce(2); KeyValueCursor cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setRange(TupleRange.ALL) .setScanProperties(forwardScanWithLimiter(limiter)) .build(); assertEquals(2, (int) cursor.getCount().join()); RecordCursorResult<KeyValue> result = cursor.getNext(); assertThat("no next reason should be SCAN_LIMIT_REACHED", result.getNoNextReason(), equalTo(RecordCursor.NoNextReason.SCAN_LIMIT_REACHED)); return null; }); }
@Test public void limitNotReached() { fdb.run(context -> { RecordScanLimiter limiter = RecordScanLimiterFactory.enforce(4); KeyValueCursor cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setLow(Tuple.from(3, 3), EndpointType.RANGE_EXCLUSIVE) .setHigh(Tuple.from(4, 2), EndpointType.RANGE_EXCLUSIVE) .setScanProperties(forwardScanWithLimiter(limiter)) .build(); assertEquals(3, (int) cursor.getCount().join()); RecordCursorResult<?> result = cursor.getNext(); assertThat("no next reason should be SOURCE_EXHAUSTED", result.getNoNextReason(), equalTo(RecordCursor.NoNextReason.SOURCE_EXHAUSTED)); return null; }); }
@Test public void sharedLimiter() { fdb.run(context -> { RecordScanLimiter limiter = RecordScanLimiterFactory.enforce(4); KeyValueCursor.Builder builder = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setLow(Tuple.from(3, 3), EndpointType.RANGE_EXCLUSIVE) .setHigh(Tuple.from(4, 2), EndpointType.RANGE_EXCLUSIVE) .setScanProperties(forwardScanWithLimiter(limiter)); KeyValueCursor cursor1 = builder.build(); KeyValueCursor cursor2 = builder.build(); assertThat(hasNextAndAdvance(cursor1), is(true)); assertThat(hasNextAndAdvance(cursor2), is(true)); assertThat(hasNextAndAdvance(cursor1), is(true)); assertThat(hasNextAndAdvance(cursor2), is(true)); assertThat(cursor1.getNext().hasNext(), is(false)); assertThat(cursor2.getNext().hasNext(), is(false)); assertThat("no next reason should be SCAN_LIMIT_REACHED", cursor1.getNext().getNoNextReason(), equalTo(RecordCursor.NoNextReason.SCAN_LIMIT_REACHED)); assertThat("no next reason should be SCAN_LIMIT_REACHED", cursor2.getNext().getNoNextReason(), equalTo(RecordCursor.NoNextReason.SCAN_LIMIT_REACHED)); return null; }); }
@Test public void limiterWithLookahead() { fdb.run(context -> { RecordScanLimiter limiter = RecordScanLimiterFactory.enforce(1); KeyValueCursor kvCursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setLow(Tuple.from(3, 3), EndpointType.RANGE_EXCLUSIVE) .setHigh(Tuple.from(4, 2), EndpointType.RANGE_EXCLUSIVE) .setScanProperties(forwardScanWithLimiter(limiter)) .build(); RecordCursor<KeyValue> cursor = kvCursor.skip(2); RecordCursorResult<KeyValue> result = cursor.getNext(); assertThat("skipped items should exhaust limit", result.hasNext(), is(false)); assertThat("no next reason should be SCAN_LIMIT_REACHED", result.getNoNextReason(), equalTo(RecordCursor.NoNextReason.SCAN_LIMIT_REACHED)); return null; }); }
@Test public void emptyScan() { fdb.run(context -> { RecordCursor<KeyValue> cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setRange(TupleRange.allOf(Tuple.from(9))) .setContinuation(null) .setScanProperties(ScanProperties.FORWARD_SCAN) .build(); RecordCursorResult<KeyValue> result = cursor.getNext(); assertFalse(result.hasNext()); assertEquals(RecordCursor.NoNextReason.SOURCE_EXHAUSTED, result.getNoNextReason()); assertTrue(result.getContinuation().isEnd()); return null; }); }
@Test public void emptyScanSplit() { fdb.run(context -> { RecordCursor<KeyValue> kvCursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setRange(TupleRange.allOf(Tuple.from(9))) .setContinuation(null) .setScanProperties(ScanProperties.FORWARD_SCAN) .build(); RecordCursor<?> cursor = new SplitHelper.KeyValueUnsplitter(context, subspace, kvCursor, false, null, false, new CursorLimitManager(context, ScanProperties.FORWARD_SCAN)); RecordCursorResult<?> result = cursor.getNext(); assertFalse(result.hasNext()); assertEquals(RecordCursor.NoNextReason.SOURCE_EXHAUSTED, result.getNoNextReason()); assertTrue(result.getContinuation().isEnd()); return null; }); }
@Test public void all() { fdb.run(context -> { KeyValueCursor cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setRange(TupleRange.ALL) .setContinuation(null) .setScanProperties(ScanProperties.FORWARD_SCAN) .build(); for (int i = 0; i < 5; i++) { for (int j = 0; j < 5; j++) { KeyValue kv = cursor.getNext().get(); assertArrayEquals(subspace.pack(Tuple.from(i, j)), kv.getKey()); assertArrayEquals(Tuple.from(i, j).pack(), kv.getValue()); } } assertThat(cursor.getNext().hasNext(), is(false)); cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setRange(TupleRange.ALL) .setContinuation(null) .setScanProperties(new ScanProperties(ExecuteProperties.newBuilder().setReturnedRowLimit(10).build())) .build(); assertEquals(10, (int)cursor.getCount().join()); cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setRange(TupleRange.ALL) .setContinuation(cursor.getNext().getContinuation().toBytes()) .setScanProperties(ScanProperties.FORWARD_SCAN) .build(); assertEquals(15, (int)cursor.getCount().join()); return null; }); }
@Test public void beginsWith() { fdb.run(context -> { KeyValueCursor cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setRange(TupleRange.allOf(Tuple.from(3))) .setContinuation(null) .setScanProperties(ScanProperties.FORWARD_SCAN) .build(); for (int j = 0; j < 5; j++) { KeyValue kv = cursor.getNext().get(); assertArrayEquals(subspace.pack(Tuple.from(3, j)), kv.getKey()); assertArrayEquals(Tuple.from(3, j).pack(), kv.getValue()); } assertThat(cursor.getNext().hasNext(), is(false)); cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setRange(TupleRange.allOf(Tuple.from(3))) .setContinuation(null) .setScanProperties(new ScanProperties(ExecuteProperties.newBuilder().setReturnedRowLimit(2).build())) .build(); assertEquals(2, (int)cursor.getCount().join()); cursor = KeyValueCursor.Builder.withSubspace(subspace) .setContext(context) .setRange(TupleRange.allOf(Tuple.from(3))) .setContinuation(cursor.getNext().getContinuation().toBytes()) .setScanProperties(new ScanProperties(ExecuteProperties.newBuilder().setReturnedRowLimit(3).build())) .build(); assertEquals(3, (int)cursor.getCount().join()); return null; }); }