@SuppressWarnings("unchecked") @Test public void testPutAndFetchBefore() throws IOException { windowStore = createWindowStore(context, false, true); long startTime = segmentSize - 4L; putFirstBatch(windowStore, startTime, context); assertEquals(Utils.mkList("zero"), toList(windowStore.fetch(0, startTime + 0L - WINDOW_SIZE, startTime + 0L))); assertEquals(Utils.mkList("one"), toList(windowStore.fetch(1, startTime + 1L - WINDOW_SIZE, startTime + 1L))); assertEquals(Utils.mkList("two"), toList(windowStore.fetch(2, startTime + 2L - WINDOW_SIZE, startTime + 2L))); assertEquals(Utils.mkList(), toList(windowStore.fetch(3, startTime + 3L - WINDOW_SIZE, startTime + 3L))); assertEquals(Utils.mkList("four"), toList(windowStore.fetch(4, startTime + 4L - WINDOW_SIZE, startTime + 4L))); assertEquals(Utils.mkList("five"), toList(windowStore.fetch(5, startTime + 5L - WINDOW_SIZE, startTime + 5L))); putSecondBatch(windowStore, startTime, context); assertEquals(Utils.mkList(), toList(windowStore.fetch(2, startTime - 1L - WINDOW_SIZE, startTime - 1L))); assertEquals(Utils.mkList(), toList(windowStore.fetch(2, startTime + 0L - WINDOW_SIZE, startTime + 0L))); assertEquals(Utils.mkList(), toList(windowStore.fetch(2, startTime + 1L - WINDOW_SIZE, startTime + 1L))); assertEquals(Utils.mkList("two"), toList(windowStore.fetch(2, startTime + 2L - WINDOW_SIZE, startTime + 2L))); assertEquals(Utils.mkList("two", "two+1"), toList(windowStore.fetch(2, startTime + 3L - WINDOW_SIZE, startTime + 3L))); assertEquals(Utils.mkList("two", "two+1", "two+2"), toList(windowStore.fetch(2, startTime + 4L - WINDOW_SIZE, startTime + 4L))); assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3"), toList(windowStore.fetch(2, startTime + 5L - WINDOW_SIZE, startTime + 5L))); assertEquals(Utils.mkList("two+1", "two+2", "two+3", "two+4"), toList(windowStore.fetch(2, startTime + 6L - WINDOW_SIZE, startTime + 6L))); assertEquals(Utils.mkList("two+2", "two+3", "two+4", "two+5"), toList(windowStore.fetch(2, startTime + 7L - WINDOW_SIZE, startTime + 7L))); assertEquals(Utils.mkList("two+3", "two+4", "two+5", "two+6"), toList(windowStore.fetch(2, startTime + 8L - WINDOW_SIZE, startTime + 8L))); assertEquals(Utils.mkList("two+4", "two+5", "two+6"), toList(windowStore.fetch(2, startTime + 9L - WINDOW_SIZE, startTime + 9L))); assertEquals(Utils.mkList("two+5", "two+6"), toList(windowStore.fetch(2, startTime + 10L - WINDOW_SIZE, startTime + 10L))); assertEquals(Utils.mkList("two+6"), toList(windowStore.fetch(2, startTime + 11L - WINDOW_SIZE, startTime + 11L))); assertEquals(Utils.mkList(), toList(windowStore.fetch(2, startTime + 12L - WINDOW_SIZE, startTime + 12L))); assertEquals(Utils.mkList(), toList(windowStore.fetch(2, startTime + 13L - WINDOW_SIZE, startTime + 13L))); windowStore.flush(); Map<Integer, Set<String>> entriesByKey = entriesByKey(changeLog, startTime); assertEquals(Utils.mkSet("zero@0"), entriesByKey.get(0)); assertEquals(Utils.mkSet("one@1"), entriesByKey.get(1)); assertEquals(Utils.mkSet("two@2", "two+1@3", "two+2@4", "two+3@5", "two+4@6", "two+5@7", "two+6@8"), entriesByKey.get(2)); assertNull(entriesByKey.get(3)); assertEquals(Utils.mkSet("four@4"), entriesByKey.get(4)); assertEquals(Utils.mkSet("five@5"), entriesByKey.get(5)); assertNull(entriesByKey.get(6)); }
@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);  }