@Test public void compareSegmentedKeys() throws Exception { assertThat( "same key in same segment should be ranked the same", cacheFunction.compareSegmentedKeys( cacheFunction.cacheKey(THE_KEY), THE_KEY ) == 0 ); final Bytes sameKeyInPriorSegment = WindowStoreUtils.toBinaryKey(new byte[]{0xA, 0xB, 0xC}, 1234, 42); assertThat( "same keys in different segments should be ordered according to segment", cacheFunction.compareSegmentedKeys( cacheFunction.cacheKey(sameKeyInPriorSegment), THE_KEY ) < 0 ); assertThat( "same keys in different segments should be ordered according to segment", cacheFunction.compareSegmentedKeys( cacheFunction.cacheKey(THE_KEY), sameKeyInPriorSegment ) > 0 ); final Bytes lowerKeyInSameSegment = WindowStoreUtils.toBinaryKey(new byte[]{0xA, 0xB, 0xB}, TIMESTAMP - 1, 0); assertThat( "different keys in same segments should be ordered according to key", cacheFunction.compareSegmentedKeys( cacheFunction.cacheKey(THE_KEY), lowerKeyInSameSegment ) > 0 ); assertThat( "different keys in same segments should be ordered according to key", cacheFunction.compareSegmentedKeys( cacheFunction.cacheKey(lowerKeyInSameSegment), THE_KEY ) < 0 ); }
int compareSegmentedKeys(Bytes cacheKey, Bytes storeKey) { long storeSegmentId = segmentId(storeKey); long cacheSegmentId = ByteBuffer.wrap(cacheKey.get()).getLong(); final int segmentCompare = Long.compare(cacheSegmentId, storeSegmentId); if (segmentCompare == 0) { byte[] cacheKeyBytes = cacheKey.get(); byte[] storeKeyBytes = storeKey.get(); return Bytes.BYTES_LEXICO_COMPARATOR.compare( cacheKeyBytes, SEGMENT_ID_BYTES, cacheKeyBytes.length - SEGMENT_ID_BYTES, storeKeyBytes, 0, storeKeyBytes.length ); } else { return segmentCompare; } }
SegmentedCacheFunction implements CacheFunction { int compareSegmentedKeys(Bytes cacheKey, Bytes storeKey) { long storeSegmentId = segmentId(storeKey); long cacheSegmentId = ByteBuffer.wrap(cacheKey.get()).getLong(); final int segmentCompare = Long.compare(cacheSegmentId, storeSegmentId); if (segmentCompare == 0) { byte[] cacheKeyBytes = cacheKey.get(); byte[] storeKeyBytes = storeKey.get(); return Bytes.BYTES_LEXICO_COMPARATOR.compare( cacheKeyBytes, SEGMENT_ID_BYTES, cacheKeyBytes.length - SEGMENT_ID_BYTES, storeKeyBytes, 0, storeKeyBytes.length ); } else { return segmentCompare; } } }
SegmentedCacheFunction implements CacheFunction { int compareSegmentedKeys(Bytes cacheKey, Bytes storeKey) { long storeSegmentId = segmentId(storeKey); long cacheSegmentId = ByteBuffer.wrap(cacheKey.get()).getLong(); final int segmentCompare = Long.compare(cacheSegmentId, storeSegmentId); if (segmentCompare == 0) { byte[] cacheKeyBytes = cacheKey.get(); byte[] storeKeyBytes = storeKey.get(); return Bytes.BYTES_LEXICO_COMPARATOR.compare( cacheKeyBytes, SEGMENT_ID_BYTES, cacheKeyBytes.length - SEGMENT_ID_BYTES, storeKeyBytes, 0, storeKeyBytes.length ); } else { return segmentCompare; } } SegmentedCacheFunction(KeySchema keySchema, long segmentInterval); }
SegmentedCacheFunction implements CacheFunction { int compareSegmentedKeys(Bytes cacheKey, Bytes storeKey) { long storeSegmentId = segmentId(storeKey); long cacheSegmentId = ByteBuffer.wrap(cacheKey.get()).getLong(); final int segmentCompare = Long.compare(cacheSegmentId, storeSegmentId); if (segmentCompare == 0) { byte[] cacheKeyBytes = cacheKey.get(); byte[] storeKeyBytes = storeKey.get(); return Bytes.BYTES_LEXICO_COMPARATOR.compare( cacheKeyBytes, SEGMENT_ID_BYTES, cacheKeyBytes.length - SEGMENT_ID_BYTES, storeKeyBytes, 0, storeKeyBytes.length ); } else { return segmentCompare; } } SegmentedCacheFunction(KeySchema keySchema, long segmentInterval); @Override Bytes key(Bytes cacheKey); @Override Bytes cacheKey(Bytes key); long segmentId(Bytes key); }
SegmentedCacheFunction implements CacheFunction { int compareSegmentedKeys(Bytes cacheKey, Bytes storeKey) { long storeSegmentId = segmentId(storeKey); long cacheSegmentId = ByteBuffer.wrap(cacheKey.get()).getLong(); final int segmentCompare = Long.compare(cacheSegmentId, storeSegmentId); if (segmentCompare == 0) { byte[] cacheKeyBytes = cacheKey.get(); byte[] storeKeyBytes = storeKey.get(); return Bytes.BYTES_LEXICO_COMPARATOR.compare( cacheKeyBytes, SEGMENT_ID_BYTES, cacheKeyBytes.length - SEGMENT_ID_BYTES, storeKeyBytes, 0, storeKeyBytes.length ); } else { return segmentCompare; } } SegmentedCacheFunction(KeySchema keySchema, long segmentInterval); @Override Bytes key(Bytes cacheKey); @Override Bytes cacheKey(Bytes key); long segmentId(Bytes key);  }