This is an automated email from the ASF dual-hosted git repository. merlimat pushed a commit to branch branch-4.18 in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
commit 8ee3a9024f54ed7f738c255153388131f9942f6d Author: Matteo Merli <[email protected]> AuthorDate: Thu Oct 15 16:27:11 2026 -0700 Partition index and filter blocks of the entry location RocksDB (#4909) * Partition index and filter blocks of the entry location RocksDB The entry location index caches index and filter blocks in the block cache but keeps one full Bloom filter block per SST. For a 64 MiB SST that block is ~8.7 MiB, larger than a shard of the LRU block cache (capacity / 64; 3.2 MiB with the default 2 GiB of direct memory). The LRU cache still inserts it, evicting the rest of the shard, and frees it as soon as the reader releases it, so every point lookup reads, checksums and parses the whole filter again. Use kTwoLevelIndexSearch with partitioned filters and pin only the top-level index and filter blocks, as RocksDB recommends when index and filter blocks live in the block cache. Partitions are ~4 KiB. The layout is readable by RocksDB 7.9.2, and SSTs stay at format_version 5. The new dbStorage_rocksDB_partitionedIndexAndFiltersEnabled setting (default true) disables it; the shipped entry_location_rocksdb.conf template sets the same options. * Exercise lookups across several index and filter partitions in the layout test Write 10 ledgers of 1000 entries instead of a single key, so the filter spans several 4 KiB partitions and, with 256-byte data blocks, the index spans several partitions too. Read back every entry, absent entries and the last entry of each ledger, with and without partitioning. --- .../bookie/storage/ldb/KeyValueStorageRocksDB.java | 13 ++++ .../storage/ldb/KeyValueStorageRocksDBTest.java | 87 ++++++++++++++++++++++ .../resources/conf/entry_location_rocksdb.conf | 13 +++- conf/bk_server.conf | 3 + conf/entry_location_rocksdb.conf.default | 13 +++- site3/website/docs/reference/config.md | 1 + 6 files changed, 128 insertions(+), 2 deletions(-) diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/KeyValueStorageRocksDB.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/KeyValueStorageRocksDB.java index 97ce8d02df..d6c685715c 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/KeyValueStorageRocksDB.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/KeyValueStorageRocksDB.java @@ -53,6 +53,7 @@ import org.rocksdb.CompressionType; import org.rocksdb.ConfigOptions; import org.rocksdb.DBOptions; import org.rocksdb.Env; +import org.rocksdb.IndexType; import org.rocksdb.InfoLogLevel; import org.rocksdb.LRUCache; import org.rocksdb.LiveFileMetaData; @@ -100,6 +101,8 @@ public class KeyValueStorageRocksDB implements KeyValueStorage { private static final String ROCKSDB_SST_SIZE_MB = "dbStorage_rocksDB_sstSizeInMB"; private static final String ROCKSDB_BLOCK_SIZE = "dbStorage_rocksDB_blockSize"; private static final String ROCKSDB_BLOOM_FILTERS_BITS_PER_KEY = "dbStorage_rocksDB_bloomFilterBitsPerKey"; + private static final String ROCKSDB_PARTITIONED_INDEX_AND_FILTERS_ENABLED = + "dbStorage_rocksDB_partitionedIndexAndFiltersEnabled"; private static final String ROCKSDB_BLOCK_CACHE_SIZE = "dbStorage_rocksDB_blockCacheSize"; private static final String ROCKSDB_NUM_LEVELS = "dbStorage_rocksDB_numLevels"; private static final String ROCKSDB_NUM_FILES_IN_LEVEL0 = "dbStorage_rocksDB_numFilesInLevel0"; @@ -210,6 +213,8 @@ public class KeyValueStorageRocksDB implements KeyValueStorage { int blockSize = conf.getInt(ROCKSDB_BLOCK_SIZE, 64 * 1024); int bloomFilterBitsPerKey = conf.getInt(ROCKSDB_BLOOM_FILTERS_BITS_PER_KEY, 10); boolean lz4CompressionEnabled = conf.getBoolean(ROCKSDB_LZ4_COMPRESSION_ENABLED, true); + boolean partitionedIndexAndFiltersEnabled = + conf.getBoolean(ROCKSDB_PARTITIONED_INDEX_AND_FILTERS_ENABLED, true); if (lz4CompressionEnabled) { options.setCompressionType(CompressionType.LZ4_COMPRESSION); @@ -240,6 +245,14 @@ public class KeyValueStorageRocksDB implements KeyValueStorage { // Options best suited for HDDs tableOptions.setCacheIndexAndFilterBlocks(true); + if (partitionedIndexAndFiltersEnabled) { + // The full filter of a 64 MiB SST is several MiB, larger than one block cache shard: the LRU + // cache drops it as soon as it is released and every lookup reads it again. Partitioned + // index and filter blocks are ~4 KiB each; only the small top-level blocks are pinned. + tableOptions.setIndexType(IndexType.kTwoLevelIndexSearch); + tableOptions.setPartitionFilters(true); + tableOptions.setPinTopLevelIndexAndFilter(true); + } options.setLevelCompactionDynamicLevelBytes(true); options.setTableFormatConfig(tableOptions); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/KeyValueStorageRocksDBTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/KeyValueStorageRocksDBTest.java index 97be7ae7c7..d6962016bc 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/KeyValueStorageRocksDBTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/KeyValueStorageRocksDBTest.java @@ -18,6 +18,7 @@ */ package org.apache.bookkeeper.bookie.storage.ldb; +import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; @@ -27,11 +28,18 @@ import static org.junit.Assert.assertTrue; import java.io.File; import java.io.IOException; import java.net.URL; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; import java.nio.file.Files; +import java.nio.file.Path; import java.nio.file.Paths; import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.Stream; import org.apache.bookkeeper.conf.ServerConfiguration; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TemporaryFolder; import org.rocksdb.BlockBasedTableConfig; import org.rocksdb.ChecksumType; import org.rocksdb.ColumnFamilyDescriptor; @@ -39,9 +47,88 @@ import org.rocksdb.ColumnFamilyOptions; import org.rocksdb.CompressionType; import org.rocksdb.DBOptions; import org.rocksdb.Options; +import org.rocksdb.SstFileReader; public class KeyValueStorageRocksDBTest { + @Rule + public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + @Test + public void testEntryLocationPartitionedIndexAndFilters() throws Exception { + File directory = temporaryFolder.newFolder(); + ServerConfiguration configuration = new ServerConfiguration(); + configuration.setEntryLocationRocksdbConf(new File(directory, "entry.conf").toString()); + // Small data blocks give the SST enough index entries to span several 4 KiB index partitions. + configuration.setProperty("dbStorage_rocksDB_blockSize", 256); + assertTrue(writeAndCheckEntryLocationSst(configuration, true) > 1); + + configuration.setProperty("dbStorage_rocksDB_partitionedIndexAndFiltersEnabled", false); + assertEquals(0, writeAndCheckEntryLocationSst(configuration, false)); + + configuration = new ServerConfiguration(); + configuration.setEntryLocationRocksdbConf( + getClass().getClassLoader().getResource("conf/entry_location_rocksdb.conf").getPath()); + assertTrue(writeAndCheckEntryLocationSst(configuration, true) > 0); + } + + /** + * Writes 10 ledgers of 1000 entries, compacts them into SSTs and checks their index and filter layout and + * lookups. Returns the number of index partitions. + */ + private long writeAndCheckEntryLocationSst(ServerConfiguration configuration, boolean partitioned) + throws Exception { + final int ledgers = 10; + final int entries = 1000; + File directory = temporaryFolder.newFolder(); + try (KeyValueStorageRocksDB storage = new KeyValueStorageRocksDB( + directory.toString(), "db", KeyValueStorageFactory.DbConfigType.EntryLocation, configuration)) { + for (int ledgerId = 0; ledgerId < ledgers; ledgerId++) { + for (int entryId = 0; entryId < entries; entryId++) { + storage.put(entryLocationKey(ledgerId, entryId), location(ledgerId, entryId)); + } + } + storage.compact(); + List<Path> ssts; + try (Stream<Path> files = Files.list(directory.toPath().resolve("db"))) { + ssts = files.filter(path -> path.toString().endsWith(".sst")).collect(Collectors.toList()); + } + assertTrue("Compaction must produce an SST", !ssts.isEmpty()); + long indexPartitions = 0; + for (Path sst : ssts) { + try (Options options = new Options(); SstFileReader reader = new SstFileReader(options)) { + reader.open(sst.toString()); + indexPartitions += reader.getTableProperties().getIndexPartitions(); + // 10 bits per key: the filter needs several 4 KiB partitions when it is partitioned. + assertTrue(reader.getTableProperties().getFilterSize() > 2 * 4096); + } + // The metaindex block is not compressed and names the filter block by its kind. + String content = new String(Files.readAllBytes(sst), StandardCharsets.ISO_8859_1); + assertEquals(partitioned, content.contains("partitionedfilter.")); + assertEquals(!partitioned, content.contains("fullfilter.")); + } + + for (int ledgerId = 0; ledgerId < ledgers; ledgerId++) { + for (int entryId = 0; entryId < entries; entryId++) { + assertArrayEquals(location(ledgerId, entryId), storage.get(entryLocationKey(ledgerId, entryId))); + } + assertNull(storage.get(entryLocationKey(ledgerId, entries))); + assertNull(storage.get(entryLocationKey(ledgerId + ledgers, 0))); + assertArrayEquals(entryLocationKey(ledgerId, entries - 1), + storage.getFloor(entryLocationKey(ledgerId, Long.MAX_VALUE)).getKey()); + } + return indexPartitions; + } + } + + private static byte[] entryLocationKey(long ledgerId, long entryId) { + return ByteBuffer.allocate(16).putLong(ledgerId).putLong(entryId).array(); + } + + private static byte[] location(long ledgerId, long entryId) { + return ByteBuffer.allocate(8).putLong(ledgerId << 32 | entryId * 1024).array(); + } + @Test public void testRocksDBInitiateWithBookieConfiguration() throws Exception { ServerConfiguration configuration = new ServerConfiguration(); diff --git a/bookkeeper-server/src/test/resources/conf/entry_location_rocksdb.conf b/bookkeeper-server/src/test/resources/conf/entry_location_rocksdb.conf index 976e53b47b..6cf416bfc2 100644 --- a/bookkeeper-server/src/test/resources/conf/entry_location_rocksdb.conf +++ b/bookkeeper-server/src/test/resources/conf/entry_location_rocksdb.conf @@ -66,4 +66,15 @@ # set by jni: tableOptions.setFilterPolicy, bloomfilter:[bits_per_key]:[use_block_based_builder] filter_policy=rocksdb.BloomFilter:10:false # set by jni: tableOptions.setCacheIndexAndFilterBlocks - cache_index_and_filter_blocks=true \ No newline at end of file + cache_index_and_filter_blocks=true + # Split index and filter blocks into partitions of ~metadata_block_size + # (4 KiB). An unpartitioned filter of a 64 MiB SST is several MiB, larger + # than a block cache shard, so the cache drops it after each use and every + # lookup reads it again. Only the small top-level blocks are pinned. + # Partitioned index and filters are readable by 7.9.2. + # set by jni: tableOptions.setIndexType + index_type=kTwoLevelIndexSearch + # set by jni: tableOptions.setPartitionFilters + partition_filters=true + # set by jni: tableOptions.setPinTopLevelIndexAndFilter + pin_top_level_index_and_filter=true diff --git a/conf/bk_server.conf b/conf/bk_server.conf index d680e750cd..1e721ac2ed 100644 --- a/conf/bk_server.conf +++ b/conf/bk_server.conf @@ -795,6 +795,9 @@ gcEntryLogMetadataCacheEnabled=false # dbStorage_rocksDB_sstSizeInMB=64 # dbStorage_rocksDB_blockSize=65536 # dbStorage_rocksDB_bloomFilterBitsPerKey=10 +# Split index and Bloom filter blocks into ~4 KiB partitions. An unpartitioned filter +# can exceed a block cache shard and is then read again on every lookup +# dbStorage_rocksDB_partitionedIndexAndFiltersEnabled=true # dbStorage_rocksDB_numLevels=-1 # dbStorage_rocksDB_numFilesInLevel0=4 # dbStorage_rocksDB_maxSizeInLevel1MB=256 diff --git a/conf/entry_location_rocksdb.conf.default b/conf/entry_location_rocksdb.conf.default index e4dc394243..93847732b7 100644 --- a/conf/entry_location_rocksdb.conf.default +++ b/conf/entry_location_rocksdb.conf.default @@ -70,4 +70,15 @@ # set by jni: tableOptions.setFilterPolicy, bloomfilter:[bits_per_key]:[use_block_based_builder] filter_policy=rocksdb.BloomFilter:10:false # set by jni: tableOptions.setCacheIndexAndFilterBlocks - cache_index_and_filter_blocks=true \ No newline at end of file + cache_index_and_filter_blocks=true + # Split index and filter blocks into partitions of ~metadata_block_size + # (4 KiB). An unpartitioned filter of a 64 MiB SST is several MiB, larger + # than a block cache shard, so the cache drops it after each use and every + # lookup reads it again. Only the small top-level blocks are pinned. + # Partitioned index and filters are readable by 7.9.2. + # set by jni: tableOptions.setIndexType + index_type=kTwoLevelIndexSearch + # set by jni: tableOptions.setPartitionFilters + partition_filters=true + # set by jni: tableOptions.setPinTopLevelIndexAndFilter + pin_top_level_index_and_filter=true diff --git a/site3/website/docs/reference/config.md b/site3/website/docs/reference/config.md index 44385497c8..94726c4ade 100644 --- a/site3/website/docs/reference/config.md +++ b/site3/website/docs/reference/config.md @@ -236,6 +236,7 @@ The table below lists parameters that you can set to configure bookies. All conf | dbStorage_rocksDB_sstSizeInMB | Size of RocksDB sst file size in MB. RocksDB is used for storing ledger indexes.<br /> | 64 | | dbStorage_rocksDB_blockSize | | 65536 | | dbStorage_rocksDB_bloomFilterBitsPerKey | | 10 | +| dbStorage_rocksDB_partitionedIndexAndFiltersEnabled | Partition the index and Bloom filter blocks of each SST file and pin only their small top-level blocks. Without partitioning, the filter block of a large SST file can exceed a block cache shard and be read again on every lookup. | true | | dbStorage_rocksDB_numLevels | | -1 | | dbStorage_rocksDB_numFilesInLevel0 | | 10 | | dbStorage_rocksDB_maxSizeInLevel1MB | | 256 |
