You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ha...@apache.org on 2024/04/26 01:48:52 UTC
(flink) 02/07: [FLINK-35047][state] Introduce ForSt related options
This is an automated email from the ASF dual-hosted git repository.
hangxiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 92d7f6538607a0189f9f7f2bff0470f8c8cad408
Author: Hangxiang Yu <ma...@gmail.com>
AuthorDate: Mon Apr 15 21:05:07 2024 +0800
[FLINK-35047][state] Introduce ForSt related options
---
.../forst/ConfigurableForStOptionsFactory.java | 39 +++
.../state/forst/ForStConfigurableOptions.java | 377 +++++++++++++++++++++
.../org/apache/flink/state/forst/ForStOptions.java | 123 +++++++
.../flink/state/forst/ForStOptionsFactory.java | 106 ++++++
4 files changed, 645 insertions(+)
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ConfigurableForStOptionsFactory.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ConfigurableForStOptionsFactory.java
new file mode 100644
index 00000000000..181d60ba603
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ConfigurableForStOptionsFactory.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.forst;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.configuration.ReadableConfig;
+
+/** An interface for options factory that pick up additional parameters from a configuration. */
+@Experimental
+public interface ConfigurableForStOptionsFactory extends ForStOptionsFactory {
+
+ /**
+ * Creates a variant of the options factory that applies additional configuration parameters.
+ *
+ * <p>If no configuration is applied, or if the method directly applies configuration values to
+ * the (mutable) options factory object, this method may return the original options factory
+ * object. Otherwise it typically returns a modified copy.
+ *
+ * @param configuration The configuration to pick the values from.
+ * @return A reconfigured options factory.
+ */
+ ForStOptionsFactory configure(ReadableConfig configuration);
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java
new file mode 100644
index 00000000000..d811199e5e0
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java
@@ -0,0 +1,377 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.forst;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.util.Preconditions;
+
+import org.rocksdb.CompactionStyle;
+import org.rocksdb.CompressionType;
+import org.rocksdb.InfoLogLevel;
+
+import java.io.File;
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+import static org.apache.flink.configuration.description.LinkElement.link;
+import static org.apache.flink.configuration.description.TextElement.code;
+import static org.rocksdb.CompactionStyle.FIFO;
+import static org.rocksdb.CompactionStyle.LEVEL;
+import static org.rocksdb.CompactionStyle.NONE;
+import static org.rocksdb.CompactionStyle.UNIVERSAL;
+import static org.rocksdb.CompressionType.LZ4_COMPRESSION;
+import static org.rocksdb.CompressionType.NO_COMPRESSION;
+import static org.rocksdb.CompressionType.SNAPPY_COMPRESSION;
+import static org.rocksdb.InfoLogLevel.INFO_LEVEL;
+
+/**
+ * This class contains the configuration options for the ForStStateBackend.
+ *
+ * <p>Currently, Options of ForSt would be configured by values here, and a user-defined {@link
+ * ForStOptionsFactory} may override the configurations here.
+ */
+@Experimental
+public class ForStConfigurableOptions implements Serializable {
+
+ // --------------------------------------------------------------------------
+ // Provided configurable DBOptions within Flink
+ // --------------------------------------------------------------------------
+
+ public static final ConfigOption<Integer> MAX_BACKGROUND_THREADS =
+ key("state.backend.forst.thread.num")
+ .intType()
+ .defaultValue(2)
+ .withDescription(
+ "The maximum number of concurrent background flush and compaction jobs (per stateful operator). "
+ + "The default value is '2'.");
+
+ public static final ConfigOption<Integer> MAX_OPEN_FILES =
+ key("state.backend.forst.files.open")
+ .intType()
+ .defaultValue(-1)
+ .withDescription(
+ "The maximum number of open files (per stateful operator) that can be used by the DB, '-1' means no limit. "
+ + "The default value is '-1'.");
+
+ public static final ConfigOption<MemorySize> LOG_MAX_FILE_SIZE =
+ key("state.backend.forst.log.max-file-size")
+ .memoryType()
+ .defaultValue(MemorySize.parse("25mb"))
+ .withDescription(
+ "The maximum size of ForSt's file used for information logging. "
+ + "If the log files becomes larger than this, a new file will be created. "
+ + "If 0, all logs will be written to one log file. "
+ + "The default maximum file size is '25MB'. ");
+
+ public static final ConfigOption<Integer> LOG_FILE_NUM =
+ key("state.backend.forst.log.file-num")
+ .intType()
+ .defaultValue(4)
+ .withDescription(
+ "The maximum number of files ForSt should keep for information logging (Default setting: 4).");
+
+ public static final ConfigOption<String> LOG_DIR =
+ key("state.backend.forst.log.dir")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The directory for ForSt's information logging files. "
+ + "If empty (Flink default setting), log files will be in the same directory as the Flink log. "
+ + "If non-empty, this directory will be used and the data directory's absolute path will be used as the prefix of the log file name. "
+ + "If setting this option as a non-existing location, e.g '/dev/null', ForSt will then create the log under its own database folder as before.");
+
+ public static final ConfigOption<InfoLogLevel> LOG_LEVEL =
+ key("state.backend.forst.log.level")
+ .enumType(InfoLogLevel.class)
+ .defaultValue(INFO_LEVEL)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The specified information logging level for ForSt. "
+ + "If unset, Flink will use %s.",
+ code(INFO_LEVEL.name()))
+ .linebreak()
+ .text(
+ "Note: ForSt info logs will not be written to the TaskManager logs and there "
+ + "is no rolling strategy, unless you configure %s, %s, and %s accordingly. "
+ + "Without a rolling strategy, long-running tasks may lead to uncontrolled "
+ + "disk space usage if configured with increased log levels!",
+ code(LOG_DIR.key()),
+ code(LOG_MAX_FILE_SIZE.key()),
+ code(LOG_FILE_NUM.key()))
+ .linebreak()
+ .text(
+ "There is no need to modify the ForSt log level, unless for troubleshooting ForSt.")
+ .build());
+
+ // --------------------------------------------------------------------------
+ // Provided configurable ColumnFamilyOptions within Flink
+ // --------------------------------------------------------------------------
+
+ public static final ConfigOption<CompactionStyle> COMPACTION_STYLE =
+ key("state.backend.forst.compaction.style")
+ .enumType(CompactionStyle.class)
+ .defaultValue(LEVEL)
+ .withDescription(
+ String.format(
+ "The specified compaction style for DB. Candidate compaction style is %s, %s, %s or %s, "
+ + "and Flink chooses '%s' as default style.",
+ LEVEL.name(),
+ FIFO.name(),
+ UNIVERSAL.name(),
+ NONE.name(),
+ LEVEL.name()));
+
+ public static final ConfigOption<Boolean> USE_DYNAMIC_LEVEL_SIZE =
+ key("state.backend.forst.compaction.level.use-dynamic-size")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ Description.builder()
+ .text(
+ "If true, ForSt will pick target size of each level dynamically. From an empty DB, ")
+ .text(
+ "ForSt would make last level the base level, which means merging L0 data into the last level, ")
+ .text(
+ "until it exceeds max_bytes_for_level_base. And then repeat this process for second last level and so on. ")
+ .text("The default value is 'false'. ")
+ .text(
+ "For more information, please refer to %s",
+ link(
+ "https://github.com/facebook/rocksdb/wiki/Leveled-Compaction#level_compaction_dynamic_level_bytes-is-true",
+ "RocksDB's doc."))
+ .build());
+
+ public static final ConfigOption<List<CompressionType>> COMPRESSION_PER_LEVEL =
+ key("state.backend.forst.compression.per.level")
+ .enumType(CompressionType.class)
+ .asList()
+ .defaultValues(SNAPPY_COMPRESSION)
+ .withDescription(
+ Description.builder()
+ .text(
+ "A semicolon-separated list of Compression Type. Different levels can have different "
+ + "compression policies. In many cases, lower levels use fast compression algorithms,"
+ + " while higher levels with more data use slower but more effective compression algorithms. "
+ + "The N th element in the List corresponds to the compression type of the level N-1"
+ + "When %s is true, compression_per_level[0] still determines L0, but other "
+ + "elements are based on the base level and may not match the level seen in the info log",
+ code(USE_DYNAMIC_LEVEL_SIZE.key()))
+ .linebreak()
+ .text(
+ "Note: If the List size is smaller than the level number, the undefined lower level uses the last Compression Type in the List")
+ .linebreak()
+ .text(
+ "Some commonly used compression algorithms for candidates include %s ,%s and %s",
+ code(NO_COMPRESSION.name()),
+ code(SNAPPY_COMPRESSION.name()),
+ code(LZ4_COMPRESSION.name()))
+ .linebreak()
+ .text(
+ "The default value is %s, which means that all data uses the Snappy compression algorithm.",
+ code(SNAPPY_COMPRESSION.name()))
+ .text(
+ "Likewise, if set to %s , means that all data is not compressed, which will achieve faster speed but will bring some space amplification.",
+ code(NO_COMPRESSION.name()))
+ .text(
+ "In addition, if we need to consider both spatial amplification and performance, we can also set it to '%s;%s;%s', which means that L0 and L1 data will not be compressed, and other data will be compressed using LZ4.",
+ code(NO_COMPRESSION.name()),
+ code(NO_COMPRESSION.name()),
+ code(LZ4_COMPRESSION.name()))
+ .build());
+
+ public static final ConfigOption<MemorySize> TARGET_FILE_SIZE_BASE =
+ key("state.backend.forst.compaction.level.target-file-size-base")
+ .memoryType()
+ .defaultValue(MemorySize.parse("64mb"))
+ .withDescription(
+ "The target file size for compaction, which determines a level-1 file size. "
+ + "The default value is '64MB'.");
+
+ public static final ConfigOption<MemorySize> MAX_SIZE_LEVEL_BASE =
+ key("state.backend.forst.compaction.level.max-size-level-base")
+ .memoryType()
+ .defaultValue(MemorySize.parse("256mb"))
+ .withDescription(
+ "The upper-bound of the total size of level base files in bytes. "
+ + "The default value is '256MB'.");
+
+ public static final ConfigOption<MemorySize> WRITE_BUFFER_SIZE =
+ key("state.backend.forst.writebuffer.size")
+ .memoryType()
+ .defaultValue(MemorySize.parse("64mb"))
+ .withDescription(
+ "The amount of data built up in memory (backed by an unsorted log on disk) "
+ + "before converting to a sorted on-disk files. The default writebuffer size is '64MB'.");
+
+ public static final ConfigOption<Integer> MAX_WRITE_BUFFER_NUMBER =
+ key("state.backend.forst.writebuffer.count")
+ .intType()
+ .defaultValue(2)
+ .withDescription(
+ "The maximum number of write buffers that are built up in memory. "
+ + "The default value is '2'.");
+
+ public static final ConfigOption<Integer> MIN_WRITE_BUFFER_NUMBER_TO_MERGE =
+ key("state.backend.forst.writebuffer.number-to-merge")
+ .intType()
+ .defaultValue(1)
+ .withDescription(
+ "The minimum number of write buffers that will be merged together before writing to storage. "
+ + "The default value is '1'.");
+
+ public static final ConfigOption<MemorySize> BLOCK_SIZE =
+ key("state.backend.forst.block.blocksize")
+ .memoryType()
+ .defaultValue(MemorySize.parse("4kb"))
+ .withDescription(
+ "The approximate size (in bytes) of user data packed per block. "
+ + "The default blocksize is '4KB'.");
+
+ public static final ConfigOption<MemorySize> METADATA_BLOCK_SIZE =
+ key("state.backend.forst.block.metadata-blocksize")
+ .memoryType()
+ .defaultValue(MemorySize.parse("4kb"))
+ .withDescription(
+ "Approximate size of partitioned metadata packed per block. "
+ + "Currently applied to indexes block when partitioned index/filters option is enabled. "
+ + "The default blocksize is '4KB'.");
+
+ public static final ConfigOption<MemorySize> BLOCK_CACHE_SIZE =
+ key("state.backend.forst.block.cache-size")
+ .memoryType()
+ .defaultValue(MemorySize.parse("8mb"))
+ .withDescription(
+ "The amount of the cache for data blocks in ForSt. "
+ + "The default block-cache size is '8MB'.");
+
+ public static final ConfigOption<MemorySize> WRITE_BATCH_SIZE =
+ key("state.backend.forst.write-batch-size")
+ .memoryType()
+ .defaultValue(MemorySize.parse("2mb"))
+ .withDescription(
+ "The max size of the consumed memory for ForSt 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.forst.use-bloom-filter")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "If true, every newly created SST file will contain a Bloom filter. "
+ + "It is disabled by default.");
+
+ public static final ConfigOption<Double> BLOOM_FILTER_BITS_PER_KEY =
+ key("state.backend.forst.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. "
+ + "The default value is 10.0.");
+
+ public static final ConfigOption<Boolean> BLOOM_FILTER_BLOCK_BASED_MODE =
+ key("state.backend.forst.bloom-filter.block-based-mode")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "If true, ForSt will use block-based filter instead of full filter, this only take effect when bloom filter is used. "
+ + "The default value is 'false'.");
+
+ static final ConfigOption<?>[] CANDIDATE_CONFIGS =
+ new ConfigOption<?>[] {
+ // configurable DBOptions
+ MAX_BACKGROUND_THREADS,
+ MAX_OPEN_FILES,
+ LOG_LEVEL,
+ LOG_MAX_FILE_SIZE,
+ LOG_FILE_NUM,
+ LOG_DIR,
+
+ // configurable ColumnFamilyOptions
+ COMPACTION_STYLE,
+ COMPRESSION_PER_LEVEL,
+ USE_DYNAMIC_LEVEL_SIZE,
+ TARGET_FILE_SIZE_BASE,
+ MAX_SIZE_LEVEL_BASE,
+ WRITE_BUFFER_SIZE,
+ MAX_WRITE_BUFFER_NUMBER,
+ MIN_WRITE_BUFFER_NUMBER_TO_MERGE,
+ BLOCK_SIZE,
+ METADATA_BLOCK_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 =
+ new HashSet<>(
+ Arrays.asList(
+ MAX_BACKGROUND_THREADS,
+ LOG_FILE_NUM,
+ MAX_WRITE_BUFFER_NUMBER,
+ MIN_WRITE_BUFFER_NUMBER_TO_MERGE));
+
+ private static final Set<ConfigOption<?>> SIZE_CONFIG_SET =
+ new HashSet<>(
+ Arrays.asList(
+ TARGET_FILE_SIZE_BASE,
+ MAX_SIZE_LEVEL_BASE,
+ WRITE_BUFFER_SIZE,
+ BLOCK_SIZE,
+ METADATA_BLOCK_SIZE,
+ BLOCK_CACHE_SIZE));
+
+ /**
+ * Helper method to check whether the (key,value) is valid through given configuration and
+ * returns the formatted value.
+ *
+ * @param option The configuration key which is configurable in {@link
+ * ForStConfigurableOptions}.
+ * @param value The value within given configuration.
+ */
+ static void checkArgumentValid(ConfigOption<?> option, Object value) {
+ final String key = option.key();
+
+ if (POSITIVE_INT_CONFIG_SET.contains(option)) {
+ Preconditions.checkArgument(
+ (Integer) value > 0,
+ "Configured value for key: " + key + " must be larger than 0.");
+ } else if (SIZE_CONFIG_SET.contains(option)) {
+ Preconditions.checkArgument(
+ ((MemorySize) value).getBytes() > 0,
+ "Configured size for key" + key + " must be larger than 0.");
+ } else if (LOG_MAX_FILE_SIZE.equals(option)) {
+ Preconditions.checkArgument(
+ ((MemorySize) value).getBytes() >= 0,
+ "Configured size for key " + key + " must be larger than or equal to 0.");
+ } else if (LOG_DIR.equals(option)) {
+ Preconditions.checkArgument(
+ new File((String) value).isAbsolute(),
+ "Configured path for key " + key + " is not absolute.");
+ }
+ }
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
new file mode 100644
index 00000000000..ba8ce6b2c62
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptions.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.forst;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.configuration.ClusterOptions;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.configuration.description.TextElement;
+
+/** Configuration options for the ForStStateBackend. */
+@Experimental
+public class ForStOptions {
+
+ /** The local directory (on the TaskManager) where ForSt puts its some files. */
+ public static final ConfigOption<String> LOCAL_DIRECTORIES =
+ ConfigOptions.key("state.backend.forst.localdir")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The local directory (on the TaskManager) where ForSt puts its some files. Per default, it will be <WORKING_DIR>/tmp. See %s for more details.",
+ TextElement.code(
+ ClusterOptions
+ .TASK_MANAGER_PROCESS_WORKING_DIR_BASE
+ .key()))
+ .build());
+
+ /** The options factory class for ForSt to create DBOptions and ColumnFamilyOptions. */
+ public static final ConfigOption<String> OPTIONS_FACTORY =
+ ConfigOptions.key("state.backend.forst.options-factory")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The options factory class for users to add customized options in DBOptions and ColumnFamilyOptions for ForSt. "
+ + "If set, the ForSt state backend will load the class and apply configs to DBOptions and ColumnFamilyOptions "
+ + "after loading ones from 'ForStConfigurableOptions' and pre-defined options.");
+
+ public static final ConfigOption<Boolean> USE_MANAGED_MEMORY =
+ ConfigOptions.key("state.backend.forst.memory.managed")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "If set, the ForSt state backend will automatically configure itself to use the "
+ + "managed memory budget of the task slot, and divide the memory over write buffers, indexes, "
+ + "block caches, etc. That way, the three major uses of memory of ForSt will be capped.");
+
+ public static final ConfigOption<MemorySize> FIX_PER_SLOT_MEMORY_SIZE =
+ ConfigOptions.key("state.backend.forst.memory.fixed-per-slot")
+ .memoryType()
+ .noDefaultValue()
+ .withDescription(
+ String.format(
+ "The fixed total amount of memory, shared among all ForSt instances per slot. "
+ + "This option overrides the '%s' option when configured.",
+ USE_MANAGED_MEMORY.key()));
+
+ public static final ConfigOption<MemorySize> FIX_PER_TM_MEMORY_SIZE =
+ ConfigOptions.key("state.backend.forst.memory.fixed-per-tm")
+ .memoryType()
+ .noDefaultValue()
+ .withDescription(
+ String.format(
+ "The fixed total amount of memory, shared among all ForSt instances per Task Manager (cluster-level option). "
+ + "This option only takes effect if neither '%s' nor '%s' are not configured. If none is configured "
+ + "then each ForSt column family state has its own memory caches (as controlled by the column "
+ + "family options). "
+ + "The relevant options for the shared resources (e.g. write-buffer-ratio) can be set on the same level (config.yaml)."
+ + "Note, that this feature breaks resource isolation between the slots",
+ USE_MANAGED_MEMORY.key(), FIX_PER_SLOT_MEMORY_SIZE.key()));
+
+ public static final ConfigOption<Double> WRITE_BUFFER_RATIO =
+ ConfigOptions.key("state.backend.forst.memory.write-buffer-ratio")
+ .doubleType()
+ .defaultValue(0.5)
+ .withDescription(
+ String.format(
+ "The maximum amount of memory that write buffers may take, as a fraction of the total shared memory. "
+ + "This option only has an effect when '%s' or '%s' are configured.",
+ USE_MANAGED_MEMORY.key(), FIX_PER_SLOT_MEMORY_SIZE.key()));
+
+ public static final ConfigOption<Double> HIGH_PRIORITY_POOL_RATIO =
+ ConfigOptions.key("state.backend.forst.memory.high-prio-pool-ratio")
+ .doubleType()
+ .defaultValue(0.1)
+ .withDescription(
+ String.format(
+ "The fraction of cache memory that is reserved for high-priority data like index, filter, and "
+ + "compression dictionary blocks. This option only has an effect when '%s' or '%s' are configured.",
+ USE_MANAGED_MEMORY.key(), FIX_PER_SLOT_MEMORY_SIZE.key()));
+
+ public static final ConfigOption<Boolean> USE_PARTITIONED_INDEX_FILTERS =
+ ConfigOptions.key("state.backend.forst.memory.partitioned-index-filters")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ String.format(
+ "With partitioning, the index/filter block of an SST file is partitioned into smaller blocks with "
+ + "an additional top-level index on them. When reading an index/filter, only top-level index is loaded into memory. "
+ + "The partitioned index/filter then uses the top-level index to load on demand into the block cache "
+ + "the partitions that are required to perform the index/filter query. "
+ + "This option only has an effect when '%s' or '%s' are configured.",
+ USE_MANAGED_MEMORY.key(), FIX_PER_SLOT_MEMORY_SIZE.key()));
+}
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptionsFactory.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptionsFactory.java
new file mode 100644
index 00000000000..7fbc432e659
--- /dev/null
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStOptionsFactory.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.forst;
+
+import org.apache.flink.annotation.Experimental;
+
+import org.rocksdb.ColumnFamilyOptions;
+import org.rocksdb.DBOptions;
+import org.rocksdb.ReadOptions;
+import org.rocksdb.WriteOptions;
+
+import java.util.Collection;
+
+/**
+ * A factory for {@link DBOptions} and {@link ColumnFamilyOptions} to be passed to the
+ * ForStStateBackend. Options have to be created lazily by this factory, because the {@code Options}
+ * class is not serializable and holds pointers to native code.
+ */
+@Experimental
+public interface ForStOptionsFactory extends java.io.Serializable {
+
+ /**
+ * This method should set the additional options on top of the current options object. The
+ * current options object may contain pre-defined options based on flags that have been
+ * configured on the state backend.
+ *
+ * <p>It is important to set the options on the current object and return the result from the
+ * setter methods, otherwise the pre-defined options may get lost.
+ *
+ * @param currentOptions The options object with the pre-defined options.
+ * @param handlesToClose The collection to register newly created {@link
+ * org.rocksdb.RocksObject}s.
+ * @return The options object on which the additional options are set.
+ */
+ DBOptions createDBOptions(DBOptions currentOptions, Collection<AutoCloseable> handlesToClose);
+
+ /**
+ * This method should set the additional options on top of the current options object. The
+ * current options object may contain pre-defined options based on flags that have been
+ * configured on the state backend.
+ *
+ * <p>It is important to set the options on the current object and return the result from the
+ * setter methods, otherwise the pre-defined options may get lost.
+ *
+ * @param currentOptions The options object with the pre-defined options.
+ * @param handlesToClose The collection to register newly created {@link
+ * org.rocksdb.RocksObject}s.
+ * @return The options object on which the additional options are set.
+ */
+ ColumnFamilyOptions createColumnOptions(
+ ColumnFamilyOptions currentOptions, Collection<AutoCloseable> handlesToClose);
+
+ // TODO: Support Metrics Options
+
+ /**
+ * This method should set the additional options on top of the current options object. The
+ * current options object may contain pre-defined options based on flags that have been
+ * configured on the state backend.
+ *
+ * <p>It is important to set the options on the current object and return the result from the
+ * setter methods, otherwise the pre-defined options may get lost.
+ *
+ * @param currentOptions The options object with the pre-defined options.
+ * @param handlesToClose The collection to register newly created {@link
+ * org.rocksdb.RocksObject}s.
+ * @return The options object on which the additional options are set.
+ */
+ default WriteOptions createWriteOptions(
+ WriteOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
+ return currentOptions;
+ }
+
+ /**
+ * This method should set the additional options on top of the current options object. The
+ * current options object may contain pre-defined options based on flags that have been
+ * configured on the state backend.
+ *
+ * <p>It is important to set the options on the current object and return the result from the
+ * setter methods, otherwise the pre-defined options may get lost.
+ *
+ * @param currentOptions The options object with the pre-defined options.
+ * @param handlesToClose The collection to register newly created {@link
+ * org.rocksdb.RocksObject}s.
+ * @return The options object on which the additional options are set.
+ */
+ default ReadOptions createReadOptions(
+ ReadOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
+ return currentOptions;
+ }
+}