You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ta...@apache.org on 2021/09/30 16:38:43 UTC
[flink] branch release-1.14 updated: [FLINK-24148][state] Add bloom
filter option in RocksDBConfiguredOptions
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.14 by this push:
new 7882d5e [FLINK-24148][state] Add bloom filter option in RocksDBConfiguredOptions
7882d5e is described below
commit 7882d5e5e7d6c6bd0c12f659b4c92829703a2b29
Author: Jiayi Liao <li...@bytedance.com>
AuthorDate: Sun Sep 5 16:40:28 2021 +0800
[FLINK-24148][state] Add bloom filter option in RocksDBConfiguredOptions
this fix #17142.
---
.../rocksdb_configurable_configuration.html | 18 +++++++
.../state/DefaultConfigurableOptionsFactory.java | 58 +++++++++++++++++++++-
.../state/RocksDBConfigurableOptions.java | 21 ++++++++
.../state/RocksDBStateBackendConfigTest.java | 11 +++-
4 files changed, 106 insertions(+), 2 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/rocksdb_configurable_configuration.html b/docs/layouts/shortcodes/generated/rocksdb_configurable_configuration.html
index 1b9762a..523efaa 100644
--- a/docs/layouts/shortcodes/generated/rocksdb_configurable_configuration.html
+++ b/docs/layouts/shortcodes/generated/rocksdb_configurable_configuration.html
@@ -27,6 +27,18 @@
<td>Approximate size of partitioned metadata packed per block. Currently applied to indexes block when partitioned index/filters option is enabled. RocksDB has default metadata blocksize as '4KB'.</td>
</tr>
<tr>
+ <td><h5>state.backend.rocksdb.bloom-filter.bits-per-key</h5></td>
+ <td style="word-wrap: break-word;">10.0</td>
+ <td>Double</td>
+ <td>Bits per key that bloom filter will use, this only take effect when bloom filter is used.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.rocksdb.bloom-filter.block-based-mode</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If true, RocksDB will use block-based filter instead of full filter, this only take effect when bloom filter is used.</td>
+ </tr>
+ <tr>
<td><h5>state.backend.rocksdb.compaction.level.max-size-level-base</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>MemorySize</td>
@@ -87,6 +99,12 @@
<td>The maximum number of concurrent background flush and compaction jobs (per stateful operator). RocksDB has default configuration as '1'.</td>
</tr>
<tr>
+ <td><h5>state.backend.rocksdb.use-bloom-filter</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If true, every newly created SST file will contain a Bloom filter. RocksDB disables it by default.</td>
+ </tr>
+ <tr>
<td><h5>state.backend.rocksdb.write-batch-size</h5></td>
<td style="word-wrap: break-word;">2 mb</td>
<td>MemorySize</td>
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/DefaultConfigurableOptionsFactory.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/DefaultConfigurableOptionsFactory.java
index 4e151dd..7308720 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/DefaultConfigurableOptionsFactory.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/DefaultConfigurableOptionsFactory.java
@@ -24,6 +24,7 @@ import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.util.Preconditions;
import org.rocksdb.BlockBasedTableConfig;
+import org.rocksdb.BloomFilter;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.CompactionStyle;
import org.rocksdb.DBOptions;
@@ -42,6 +43,8 @@ import java.util.Set;
import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.BLOCK_CACHE_SIZE;
import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.BLOCK_SIZE;
+import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.BLOOM_FILTER_BITS_PER_KEY;
+import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.BLOOM_FILTER_BLOCK_BASED_MODE;
import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.COMPACTION_STYLE;
import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.LOG_DIR;
import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.LOG_FILE_NUM;
@@ -54,6 +57,7 @@ import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOption
import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.METADATA_BLOCK_SIZE;
import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.MIN_WRITE_BUFFER_NUMBER_TO_MERGE;
import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.TARGET_FILE_SIZE_BASE;
+import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.USE_BLOOM_FILTER;
import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.USE_DYNAMIC_LEVEL_SIZE;
import static org.apache.flink.contrib.streaming.state.RocksDBConfigurableOptions.WRITE_BUFFER_SIZE;
@@ -156,6 +160,24 @@ public class DefaultConfigurableOptionsFactory implements ConfigurableRocksDBOpt
blockBasedTableConfig.setBlockCacheSize(getBlockCacheSize());
}
+ if (isOptionConfigured(USE_BLOOM_FILTER)) {
+ final boolean enabled = Boolean.parseBoolean(getInternal(USE_BLOOM_FILTER.key()));
+ if (enabled) {
+ final double bitsPerKey =
+ isOptionConfigured(BLOOM_FILTER_BITS_PER_KEY)
+ ? Double.parseDouble(getInternal(BLOOM_FILTER_BITS_PER_KEY.key()))
+ : BLOOM_FILTER_BITS_PER_KEY.defaultValue();
+ final boolean blockBasedMode =
+ isOptionConfigured(BLOOM_FILTER_BLOCK_BASED_MODE)
+ ? Boolean.parseBoolean(
+ getInternal(BLOOM_FILTER_BLOCK_BASED_MODE.key()))
+ : BLOOM_FILTER_BLOCK_BASED_MODE.defaultValue();
+ BloomFilter bloomFilter = new BloomFilter(bitsPerKey, blockBasedMode);
+ handlesToClose.add(bloomFilter);
+ blockBasedTableConfig.setFilterPolicy(bloomFilter);
+ }
+ }
+
return currentOptions.setTableFormatConfig(blockBasedTableConfig);
}
@@ -429,6 +451,37 @@ public class DefaultConfigurableOptionsFactory implements ConfigurableRocksDBOpt
return this;
}
+ // --------------------------------------------------------------------------
+ // Filter policy in RocksDB
+ // --------------------------------------------------------------------------
+
+ private boolean getUseBloomFilter() {
+ return Boolean.parseBoolean(getInternal(USE_BLOOM_FILTER.key()));
+ }
+
+ public DefaultConfigurableOptionsFactory setUseBloomFilter(boolean useBloomFilter) {
+ setInternal(USE_BLOOM_FILTER.key(), String.valueOf(useBloomFilter));
+ return this;
+ }
+
+ private double getBloomFilterBitsPerKey() {
+ return Double.parseDouble(getInternal(BLOOM_FILTER_BITS_PER_KEY.key()));
+ }
+
+ public DefaultConfigurableOptionsFactory setBloomFilterBitsPerKey(double bitsPerKey) {
+ setInternal(BLOOM_FILTER_BITS_PER_KEY.key(), String.valueOf(bitsPerKey));
+ return this;
+ }
+
+ private boolean getBloomFilterBlockBasedMode() {
+ return Boolean.parseBoolean(getInternal(BLOOM_FILTER_BLOCK_BASED_MODE.key()));
+ }
+
+ public DefaultConfigurableOptionsFactory setBloomFilterBlockBasedMode(boolean blockBasedMode) {
+ setInternal(BLOOM_FILTER_BLOCK_BASED_MODE.key(), String.valueOf(blockBasedMode));
+ return this;
+ }
+
private static final ConfigOption<?>[] CANDIDATE_CONFIGS =
new ConfigOption<?>[] {
// configurable DBOptions
@@ -449,7 +502,10 @@ public class DefaultConfigurableOptionsFactory implements ConfigurableRocksDBOpt
MIN_WRITE_BUFFER_NUMBER_TO_MERGE,
BLOCK_SIZE,
METADATA_BLOCK_SIZE,
- BLOCK_CACHE_SIZE
+ BLOCK_CACHE_SIZE,
+ USE_BLOOM_FILTER,
+ BLOOM_FILTER_BITS_PER_KEY,
+ BLOOM_FILTER_BLOCK_BASED_MODE
};
private static final Set<ConfigOption<?>> POSITIVE_INT_CONFIG_SET =
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBConfigurableOptions.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBConfigurableOptions.java
index db9f6ba..807c0e7 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBConfigurableOptions.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBConfigurableOptions.java
@@ -227,4 +227,25 @@ public class RocksDBConfigurableOptions implements Serializable {
.withDescription(
"The max size of the consumed memory for RocksDB batch write, "
+ "will flush just based on item count if this config set to 0.");
+
+ public static final ConfigOption<Boolean> USE_BLOOM_FILTER =
+ key("state.backend.rocksdb.use-bloom-filter")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "If true, every newly created SST file will contain a Bloom filter. RocksDB disables it by default.");
+
+ public static final ConfigOption<Double> BLOOM_FILTER_BITS_PER_KEY =
+ key("state.backend.rocksdb.bloom-filter.bits-per-key")
+ .doubleType()
+ .defaultValue(10.0)
+ .withDescription(
+ "Bits per key that bloom filter will use, this only take effect when bloom filter is used.");
+
+ public static final ConfigOption<Boolean> BLOOM_FILTER_BLOCK_BASED_MODE =
+ key("state.backend.rocksdb.bloom-filter.block-based-mode")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "If true, RocksDB will use block-based filter instead of full filter, this only take effect when bloom filter is used.");
}
diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
index 5fc83b0..5fcfe3e 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackendConfigTest.java
@@ -51,6 +51,7 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.rocksdb.BlockBasedTableConfig;
+import org.rocksdb.BloomFilter;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.CompactionStyle;
import org.rocksdb.DBOptions;
@@ -481,7 +482,10 @@ public class RocksDBStateBackendConfigTest {
.setMinWriteBufferNumberToMerge(3)
.setBlockSize("64KB")
.setMetadataBlockSize("16KB")
- .setBlockCacheSize("512mb");
+ .setBlockCacheSize("512mb")
+ .setUseBloomFilter(true)
+ .setBloomFilterBitsPerKey(12.0)
+ .setBloomFilterBlockBasedMode(false);
try (RocksDBResourceContainer optionsContainer =
new RocksDBResourceContainer(PredefinedOptions.DEFAULT, customizedOptions)) {
@@ -507,6 +511,7 @@ public class RocksDBStateBackendConfigTest {
assertEquals(64 * SizeUnit.KB, tableConfig.blockSize());
assertEquals(16 * SizeUnit.KB, tableConfig.metadataBlockSize());
assertEquals(512 * SizeUnit.MB, tableConfig.blockCacheSize());
+ assertTrue(tableConfig.filterPolicy() instanceof BloomFilter);
}
}
@@ -540,6 +545,8 @@ public class RocksDBStateBackendConfigTest {
verifyIllegalArgument(RocksDBConfigurableOptions.USE_DYNAMIC_LEVEL_SIZE, "1");
verifyIllegalArgument(RocksDBConfigurableOptions.COMPACTION_STYLE, "LEV");
+ verifyIllegalArgument(RocksDBConfigurableOptions.USE_BLOOM_FILTER, "NO");
+ verifyIllegalArgument(RocksDBConfigurableOptions.BLOOM_FILTER_BLOCK_BASED_MODE, "YES");
}
// verify legal configuration
@@ -561,6 +568,7 @@ public class RocksDBStateBackendConfigTest {
configuration.setString(RocksDBConfigurableOptions.BLOCK_SIZE.key(), "4 kb");
configuration.setString(RocksDBConfigurableOptions.METADATA_BLOCK_SIZE.key(), "8 kb");
configuration.setString(RocksDBConfigurableOptions.BLOCK_CACHE_SIZE.key(), "512 mb");
+ configuration.setString(RocksDBConfigurableOptions.USE_BLOOM_FILTER.key(), "TRUE");
DefaultConfigurableOptionsFactory optionsFactory =
new DefaultConfigurableOptionsFactory();
@@ -590,6 +598,7 @@ public class RocksDBStateBackendConfigTest {
assertEquals(4 * SizeUnit.KB, tableConfig.blockSize());
assertEquals(8 * SizeUnit.KB, tableConfig.metadataBlockSize());
assertEquals(512 * SizeUnit.MB, tableConfig.blockCacheSize());
+ assertTrue(tableConfig.filterPolicy() instanceof BloomFilter);
}
}
}