This is an automated email from the ASF dual-hosted git repository. pnowojski pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2d30c0350685510501d432ef6e37ba004ef26def Author: Aleksandr Savonin <[email protected]> AuthorDate: Tue Sep 15 19:12:16 2026 +0200 [FLINK-40584][state/rocksdb] Clear inherited seek restrictions on bounded iterators The bounded iterator copies the backend's ReadOptions, so a prefixSameAsStart or iterate_lower_bound set globally by an options factory would carry over and silently drop keys inside the bound: the first stops at the next extracted prefix, the second clamps the seek. Clear both on the private copy, the shared options stay untouched. --- .../flink/state/rocksdb/RocksDBOperationUtils.java | 18 +++-- .../state/rocksdb/RocksDBPrefixIteratorTest.java | 79 ++++++++++++++++++++++ 2 files changed, 92 insertions(+), 5 deletions(-) diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBOperationUtils.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBOperationUtils.java index d3404641227..8c8e5cdb196 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBOperationUtils.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBOperationUtils.java @@ -124,11 +124,11 @@ public class RocksDBOperationUtils { * before the next map's keys. The prefix only sets the bound; callers still position the * iterator with {@code seek}. * - * <p>The iterator uses a private copy of {@code readOptions} with the upper bound set and - * auto-prefix mode enabled. The caller's options are not modified. If the prefix has no - * exclusive upper bound (see {@code getPrefixEnd}), the caller's options are used as they are - * and the iterator is unbounded: no key outside such a prefix can follow it, so nothing extra - * is visited. + * <p>The iterator uses a private copy of {@code readOptions} with the upper bound set, + * auto-prefix mode enabled, and {@code prefixSameAsStart} and any lower bound cleared. The + * caller's options are not modified. If the prefix has no exclusive upper bound (see {@code + * getPrefixEnd}), the caller's options are used as they are and the iterator is unbounded: no + * key outside such a prefix can follow it, so nothing extra is visited. */ static RocksIteratorWrapper getRocksIteratorBoundedByPrefix( RocksDB db, @@ -148,6 +148,14 @@ public class RocksDBOperationUtils { // Auto-prefix mode is needed because, with a prefix extractor configured, the default // seek mode honors the upper bound only if it shares the seek key's extracted prefix. boundedReadOptions.setAutoPrefixMode(true); + // prefix_same_as_start would invalidate the iterator as soon as the extracted prefix + // changes. With an extractor longer than the seek prefix that happens inside the map + // or key group, or immediately for the initial seek from the bare prefix. Neither + // auto_prefix_mode nor total_order_seek disables it, so it is cleared on the copy. + boundedReadOptions.setPrefixSameAsStart(false); + // The seek key is the lower end of every bounded scan. A configured lower bound above + // it would silently skip entries, so the copy carries none. + boundedReadOptions.setIterateLowerBound(null); return new RocksIteratorWrapper( db.newIterator(columnFamilyHandle, boundedReadOptions), boundedReadOptions, diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBPrefixIteratorTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBPrefixIteratorTest.java index 9552a9adb53..c920c532473 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBPrefixIteratorTest.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBPrefixIteratorTest.java @@ -242,6 +242,85 @@ class RocksDBPrefixIteratorTest { } } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testResumedSeekDoesNotStopAtExtractedPrefix(boolean totalOrderSeek) + throws RocksDBException { + final RocksDB db = rocksDBExtension.getRocksDB(); + try (ColumnFamilyOptions options = + new ColumnFamilyOptions().useFixedLengthPrefixExtractor(2); + ColumnFamilyHandle columnFamily = + db.createColumnFamily(new ColumnFamilyDescriptor(bytes(42), options)); + ReadOptions readOptions = + new ReadOptions() + .setPrefixSameAsStart(true) + .setTotalOrderSeek(totalOrderSeek)) { + db.put(columnFamily, bytes(1, 0), bytes()); + db.put(columnFamily, bytes(1, 1), bytes()); + db.put(columnFamily, bytes(2, 0), bytes()); + + try (RocksIteratorWrapper iterator = + RocksDBOperationUtils.getRocksIteratorBoundedByPrefix( + db, columnFamily, readOptions, bytes(1))) { + // The map prefix is [1], but the resume key's extracted prefix is [1, 0]. + assertKeys(iterator, bytes(1, 0), bytes(1, 0), bytes(1, 1)); + } + + assertThat(readOptions.prefixSameAsStart()).isTrue(); + } + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testInitialSeekDoesNotStopAtExtractedPrefix(boolean totalOrderSeek) + throws RocksDBException { + final RocksDB db = rocksDBExtension.getRocksDB(); + // RocksDB enforces prefixSameAsStart regardless of totalOrderSeek, so both are covered. + try (ColumnFamilyOptions options = new ColumnFamilyOptions().useCappedPrefixExtractor(2); + ColumnFamilyHandle columnFamily = + db.createColumnFamily(new ColumnFamilyDescriptor(bytes(42), options)); + ReadOptions readOptions = + new ReadOptions() + .setPrefixSameAsStart(true) + .setTotalOrderSeek(totalOrderSeek)) { + db.put(columnFamily, bytes(1, 0), bytes()); + db.put(columnFamily, bytes(1, 1), bytes()); + db.put(columnFamily, bytes(2, 0), bytes()); + + try (RocksIteratorWrapper iterator = + RocksDBOperationUtils.getRocksIteratorBoundedByPrefix( + db, columnFamily, readOptions, bytes(1))) { + // The bare prefix [1] is its own extracted prefix, which no stored key shares. + assertKeys(iterator, bytes(1), bytes(1, 0), bytes(1, 1)); + } + + assertThat(readOptions.prefixSameAsStart()).isTrue(); + } + } + + @Test + void testConfiguredLowerBoundDoesNotClampSeek() throws RocksDBException { + final RocksDB db = rocksDBExtension.getRocksDB(); + try (ColumnFamilyOptions options = new ColumnFamilyOptions(); + ColumnFamilyHandle columnFamily = + db.createColumnFamily(new ColumnFamilyDescriptor(bytes(42), options)); + Slice lowerBound = new Slice(bytes(1, 1)); + ReadOptions readOptions = new ReadOptions().setIterateLowerBound(lowerBound)) { + db.put(columnFamily, bytes(1, 0), bytes()); + db.put(columnFamily, bytes(1, 1), bytes()); + db.put(columnFamily, bytes(2, 0), bytes()); + + try (RocksIteratorWrapper iterator = + RocksDBOperationUtils.getRocksIteratorBoundedByPrefix( + db, columnFamily, readOptions, bytes(1))) { + // An inherited lower bound would clamp the seek to [1, 1]. + assertKeys(iterator, bytes(1, 0), bytes(1, 0), bytes(1, 1)); + } + + assertThat(readOptions.iterateLowerBound().data()).isEqualTo(bytes(1, 1)); + } + } + @ParameterizedTest @MethodSource("prefixBloomRanges") void testPrefixBloomFiltersRespectBoundsAndConfiguredTotalOrderSeek(
