@SuppressWarnings("unchecked") @Test public void testFetchRange() throws IOException { windowStore = createWindowStore(context, false, true); long startTime = segmentSize - 4L; putFirstBatch(windowStore, startTime, context); final KeyValue<Windowed<Integer>, String> zero = windowedPair(0, "zero", startTime + 0); final KeyValue<Windowed<Integer>, String> one = windowedPair(1, "one", startTime + 1); final KeyValue<Windowed<Integer>, String> two = windowedPair(2, "two", startTime + 2); final KeyValue<Windowed<Integer>, String> four = windowedPair(4, "four", startTime + 4); final KeyValue<Windowed<Integer>, String> five = windowedPair(5, "five", startTime + 5); assertEquals( Utils.mkList(zero, one), StreamsTestUtils.toList(windowStore.fetch(0, 1, startTime + 0L - WINDOW_SIZE, startTime + 0L + WINDOW_SIZE)) ); assertEquals( Utils.mkList(one), StreamsTestUtils.toList(windowStore.fetch(1, 1, startTime + 0L - WINDOW_SIZE, startTime + 0L + WINDOW_SIZE)) ); assertEquals( Utils.mkList(one, two), StreamsTestUtils.toList(windowStore.fetch(1, 3, startTime + 0L - WINDOW_SIZE, startTime + 0L + WINDOW_SIZE)) ); assertEquals( Utils.mkList(zero, one, two), StreamsTestUtils.toList(windowStore.fetch(0, 5, startTime + 0L - WINDOW_SIZE, startTime + 0L + WINDOW_SIZE)) ); assertEquals( Utils.mkList(zero, one, two, four, five), StreamsTestUtils.toList(windowStore.fetch(0, 5, startTime + 0L - WINDOW_SIZE, startTime + 0L + WINDOW_SIZE + 5L)) ); assertEquals( Utils.mkList(two, four, five), StreamsTestUtils.toList(windowStore.fetch(0, 5, startTime + 2L, startTime + 0L + WINDOW_SIZE + 5L)) ); assertEquals( Utils.mkList(), StreamsTestUtils.toList(windowStore.fetch(4, 5, startTime + 2L, startTime + WINDOW_SIZE)) ); assertEquals( Utils.mkList(), StreamsTestUtils.toList(windowStore.fetch(0, 3, startTime + 3L, startTime + WINDOW_SIZE + 5)) ); }
@Override public WindowStoreIterator<V> fetch(K key, long timeFrom, long timeTo) { final KeyValueIterator<Bytes, byte[]> bytesIterator = bytesStore.fetch(Bytes.wrap(serdes.rawKey(key)), timeFrom, timeTo); return new WindowStoreIteratorWrapper<>(bytesIterator, serdes, windowSize).valuesIterator(); }
RocksDBWindowStore extends WrappedStateStore.AbstractStateStore implements WindowStore<K, V> { @Override public WindowStoreIterator<V> fetch(K key, long timeFrom, long timeTo) { final KeyValueIterator<Bytes, byte[]> bytesIterator = bytesStore.fetch(Bytes.wrap(serdes.rawKey(key)), timeFrom, timeTo); return new WindowStoreIteratorWrapper<>(bytesIterator, serdes, windowSize).valuesIterator(); } }
RocksDBWindowStore extends WrappedStateStore.AbstractStateStore implements WindowStore<K, V> { @Override public WindowStoreIterator<V> fetch(K key, long timeFrom, long timeTo) { final KeyValueIterator<Bytes, byte[]> bytesIterator = bytesStore.fetch(Bytes.wrap(serdes.rawKey(key)), timeFrom, timeTo); return new WindowStoreIteratorWrapper<>(bytesIterator, serdes, windowSize).valuesIterator(); } RocksDBWindowStore(final SegmentedBytesStore bytesStore,
                       final Serde<K> keySerde,
                       final Serde<V> valueSerde,
                       final boolean retainDuplicates,
                       final long windowSize); }
RocksDBWindowStore extends WrappedStateStore.AbstractStateStore implements WindowStore<K, V> { @Override public WindowStoreIterator<V> fetch(K key, long timeFrom, long timeTo) { final KeyValueIterator<Bytes, byte[]> bytesIterator = bytesStore.fetch(Bytes.wrap(serdes.rawKey(key)), timeFrom, timeTo); return new WindowStoreIteratorWrapper<>(bytesIterator, serdes, windowSize).valuesIterator(); } RocksDBWindowStore(final SegmentedBytesStore bytesStore,
                       final Serde<K> keySerde,
                       final Serde<V> valueSerde,
                       final boolean retainDuplicates,
                       final long windowSize); @Override @SuppressWarnings("unchecked") void init(final ProcessorContext context, final StateStore root); @Override void put(K key, V value); @Override void put(K key, V value, long timestamp); @Override WindowStoreIterator<V> fetch(K key, long timeFrom, long timeTo); @Override KeyValueIterator<Windowed<K>, V> fetch(K from, K to, long timeFrom, long timeTo); }
RocksDBWindowStore extends WrappedStateStore.AbstractStateStore implements WindowStore<K, V> { @Override public WindowStoreIterator<V> fetch(K key, long timeFrom, long timeTo) { final KeyValueIterator<Bytes, byte[]> bytesIterator = bytesStore.fetch(Bytes.wrap(serdes.rawKey(key)), timeFrom, timeTo); return new WindowStoreIteratorWrapper<>(bytesIterator, serdes, windowSize).valuesIterator(); } RocksDBWindowStore(final SegmentedBytesStore bytesStore,
                       final Serde<K> keySerde,
                       final Serde<V> valueSerde,
                       final boolean retainDuplicates,
                       final long windowSize); @Override @SuppressWarnings("unchecked") void init(final ProcessorContext context, final StateStore root); @Override void put(K key, V value); @Override void put(K key, V value, long timestamp); @Override WindowStoreIterator<V> fetch(K key, long timeFrom, long timeTo); @Override KeyValueIterator<Windowed<K>, V> fetch(K from, K to, long timeFrom, long timeTo);  }