This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/master by this push:
new 6751f970e9 Partition index and filter blocks of the entry location
RocksDB (#4909)
6751f970e9 is described below
commit 6751f970e911e89a636ff2b1f7b8a5551f398e40
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 | 77 ++++++++++++++++++++++
.../resources/conf/entry_location_rocksdb.conf | 11 ++++
conf/bk_server.conf | 3 +
conf/entry_location_rocksdb.conf.default | 11 ++++
site3/website/docs/reference/config.md | 1 +
6 files changed, 116 insertions(+)
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 dca6ab5efc..d2af22f2bc 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
@@ -30,6 +30,7 @@ import java.io.IOException;
import java.net.URL;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
@@ -47,6 +48,7 @@ import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.CompressionType;
import org.rocksdb.DBOptions;
import org.rocksdb.Options;
+import org.rocksdb.SstFileReader;
public class KeyValueStorageRocksDBTest {
@@ -97,6 +99,81 @@ public class KeyValueStorageRocksDBTest {
}
}
+ @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 5653cb4a2d..65feb198fa 100644
--- a/bookkeeper-server/src/test/resources/conf/entry_location_rocksdb.conf
+++ b/bookkeeper-server/src/test/resources/conf/entry_location_rocksdb.conf
@@ -91,3 +91,14 @@
filter_policy=rocksdb.BloomFilter:10:false
# set by jni: tableOptions.setCacheIndexAndFilterBlocks
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 386c2b1d91..b9adbbfb14 100644
--- a/conf/entry_location_rocksdb.conf.default
+++ b/conf/entry_location_rocksdb.conf.default
@@ -95,3 +95,14 @@
filter_policy=rocksdb.BloomFilter:10:false
# set by jni: tableOptions.setCacheIndexAndFilterBlocks
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 |
