@SuppressWarnings("unchecked") @Test public void testInitialLoading() throws IOException { File storeDir = new File(baseDir, windowName); windowStore = createWindowStore(context, false, true); new File(storeDir, segments.segmentName(0L)).mkdir(); new File(storeDir, segments.segmentName(1L)).mkdir(); new File(storeDir, segments.segmentName(2L)).mkdir(); new File(storeDir, segments.segmentName(3L)).mkdir(); new File(storeDir, segments.segmentName(4L)).mkdir(); new File(storeDir, segments.segmentName(5L)).mkdir(); new File(storeDir, segments.segmentName(6L)).mkdir(); windowStore.close(); windowStore = createWindowStore(context, false, true); assertEquals( Utils.mkSet(segments.segmentName(4L), segments.segmentName(5L), segments.segmentName(6L)), segmentDirs(baseDir) ); try (WindowStoreIterator iter = windowStore.fetch(0, 0L, 1000000L)) { while (iter.hasNext()) { iter.next(); } } assertEquals( Utils.mkSet(segments.segmentName(4L), segments.segmentName(5L), segments.segmentName(6L)), segmentDirs(baseDir) ); }
@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);  }