You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by za...@apache.org on 2025/02/11 06:55:35 UTC
(flink) 02/05: [FLINK-36439][docs] Documents for ForSt State Backend
This is an automated email from the ASF dual-hosted git repository.
zakelly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 381be9cf9ec628f255e13962d2601c858303f10f
Author: Zakelly <za...@gmail.com>
AuthorDate: Thu Feb 6 12:49:23 2025 +0800
[FLINK-36439][docs] Documents for ForSt State Backend
---
docs/content.zh/docs/deployment/config.md | 30 ++-
docs/content.zh/docs/ops/metrics.md | 7 +-
docs/content/docs/deployment/config.md | 30 ++-
docs/content/docs/ops/metrics.md | 7 +-
docs/content/docs/ops/state/large_state_tuning.md | 16 +-
docs/content/docs/ops/state/state_backends.md | 57 ++++-
.../shortcodes/generated/expert_forst_section.html | 54 +++++
.../forst_configurable_configuration.html | 168 ++++++++++++++
.../shortcodes/generated/forst_configuration.html | 120 ++++++++++
.../forst_native_metric_configuration.html | 246 +++++++++++++++++++++
.../generated/state_backend_forst_section.html | 66 ++++++
.../flink/annotation/docs/Documentation.java | 2 +
flink-docs/pom.xml | 5 +
.../configuration/ConfigOptionsDocGenerator.java | 2 +
.../docs/util/ConfigurationOptionLocator.java | 3 +
.../state/forst/ForStConfigurableOptions.java | 52 +++--
.../state/forst/ForStNativeMetricOptions.java | 84 +++----
.../org/apache/flink/state/forst/ForStOptions.java | 64 ++++--
18 files changed, 917 insertions(+), 96 deletions(-)
diff --git a/docs/content.zh/docs/deployment/config.md b/docs/content.zh/docs/deployment/config.md
index 8fbf550db8f..8922586e2b3 100644
--- a/docs/content.zh/docs/deployment/config.md
+++ b/docs/content.zh/docs/deployment/config.md
@@ -160,7 +160,7 @@ These values are configured as memory sizes, for example *1536m* or *2g*.
You can configure checkpointing directly in code within your Flink job or application. Putting these values here in the configuration defines them as defaults in case the application does not configure anything.
- - `state.backend.type`: The state backend to use. This defines the data structure mechanism for taking snapshots. Common values are `hashmap` or `rocksdb`.
+ - `state.backend.type`: The state backend to use. This defines the data structure mechanism for taking snapshots. Common values are `hashmap`, `rocksdb` or `forst`.
- `execution.checkpointing.dir`: The directory to write checkpoints to. This takes a path URI like *s3://mybucket/flink-app/checkpoints* or *hdfs://namenode:port/flink/checkpoints*.
- `execution.checkpointing.savepoint-dir`: The default directory for savepoints. Takes a path URI, similar to `execution.checkpointing.dir`.
- `execution.checkpointing.interval`: The base interval setting. To enable checkpointing, you need to set this value larger than 0.
@@ -352,6 +352,12 @@ These are the options commonly needed to configure the RocksDB state backend. Se
{{< generated/state_backend_rocksdb_section >}}
+### ForSt State Backend
+
+These are the options commonly needed to configure the ForSt state backend. See the [Advanced ForSt Backend Section](#advanced-forst-state-backends-options) for options necessary for advanced low level configurations and trouble-shooting.
+
+{{< generated/state_backend_forst_section >}}
+
----
----
@@ -374,6 +380,16 @@ Enabling RocksDB's native metrics may cause degraded performance and should be s
{{< generated/rocksdb_native_metric_configuration >}}
+### ForSt Native Metrics
+
+ForSt has similar native metric mechanism to RocksDB.
+
+{{< hint warning >}}
+Enabling ForSt's native metrics may cause degraded performance and should be set carefully.
+{{< /hint >}}
+
+{{< generated/forst_native_metric_configuration >}}
+
----
----
@@ -474,6 +490,12 @@ Advanced options to tune RocksDB and RocksDB checkpoints.
{{< generated/expert_rocksdb_section >}}
+### Advanced ForSt State Backends Options
+
+Advanced options to tune ForSt and ForSt checkpoints.
+
+{{< generated/expert_forst_section >}}
+
### State Changelog Options
Please refer to [State Backends]({{< ref "docs/ops/state/state_backends#enabling-changelog" >}}) for information on
@@ -484,13 +506,17 @@ using State Changelog. {{< generated/state_changelog_section >}}
These settings take effect when the `state.changelog.storage` is set to `filesystem` (see [above](#state-changelog-storage)).
{{< generated/fs_state_changelog_configuration >}}
-**RocksDB Configurable Options**
+### RocksDB Configurable Options
These options give fine-grained control over the behavior and resources of ColumnFamilies.
With the introduction of `state.backend.rocksdb.memory.managed` and `state.backend.rocksdb.memory.fixed-per-slot` (Apache Flink 1.10), it should be only necessary to use the options here for advanced performance tuning. These options here can also be specified in the application program via `RocksDBStateBackend.setRocksDBOptions(RocksDBOptionsFactory)`.
{{< generated/rocksdb_configurable_configuration >}}
+### ForSt State Backend Configurable Options
+
+{{< generated/forst_configurable_configuration >}}
+
### Advanced Fault Tolerance Options
*These parameters can help with problems related to failover and to components erroneously considering each other as failed.*
diff --git a/docs/content.zh/docs/ops/metrics.md b/docs/content.zh/docs/ops/metrics.md
index 852893b2c52..bb3fc9c9317 100644
--- a/docs/content.zh/docs/ops/metrics.md
+++ b/docs/content.zh/docs/ops/metrics.md
@@ -1541,7 +1541,12 @@ Note that for failed checkpoints, metrics are updated on a best efforts basis an
### RocksDB
Certain RocksDB native metrics are available but disabled by default, you can find full documentation [here]({{< ref "docs/deployment/config" >}}#rocksdb-native-metrics)
-### ForStDB
+### ForSt
+
+Certain ForSt native metrics are available but disabled by default, you can find full documentation [here]({{< ref "docs/deployment/config" >}}#forst-native-metrics)
+
+Besides that, we support the following metrics:
+
<table class="table table-bordered">
<thead>
<tr>
diff --git a/docs/content/docs/deployment/config.md b/docs/content/docs/deployment/config.md
index 7c149528963..a804fdc34a3 100644
--- a/docs/content/docs/deployment/config.md
+++ b/docs/content/docs/deployment/config.md
@@ -160,7 +160,7 @@ These values are configured as memory sizes, for example *1536m* or *2g*.
You can configure checkpointing directly in code within your Flink job or application. Putting these values here in the configuration defines them as defaults in case the application does not configure anything.
- - `state.backend.type`: The state backend to use. This defines the data structure mechanism for taking snapshots. Common values are `hashmap` or `rocksdb`.
+ - `state.backend.type`: The state backend to use. This defines the data structure mechanism for taking snapshots. Common values are `hashmap`, `rocksdb` or `forst`.
- `execution.checkpointing.dir`: The directory to write checkpoints to. This takes a path URI like *s3://mybucket/flink-app/checkpoints* or *hdfs://namenode:port/flink/checkpoints*.
- `execution.checkpointing.savepoint-dir`: The default directory for savepoints. Takes a path URI, similar to `execution.checkpointing.dir`.
- `execution.checkpointing.interval`: The base interval setting. To enable checkpointing, you need to set this value larger than 0.
@@ -354,6 +354,12 @@ These are the options commonly needed to configure the RocksDB state backend. Se
{{< generated/state_backend_rocksdb_section >}}
+### ForSt State Backend
+
+These are the options commonly needed to configure the ForSt state backend. See the [Advanced ForSt Backend Section](#advanced-forst-state-backends-options) for options necessary for advanced low level configurations and trouble-shooting.
+
+{{< generated/state_backend_forst_section >}}
+
----
----
@@ -376,6 +382,16 @@ Enabling RocksDB's native metrics may cause degraded performance and should be s
{{< generated/rocksdb_native_metric_configuration >}}
+### ForSt Native Metrics
+
+ForSt has similar native metric mechanism to RocksDB.
+
+{{< hint warning >}}
+Enabling ForSt's native metrics may cause degraded performance and should be set carefully.
+{{< /hint >}}
+
+{{< generated/forst_native_metric_configuration >}}
+
----
----
@@ -476,6 +492,12 @@ Advanced options to tune RocksDB and RocksDB checkpoints.
{{< generated/expert_rocksdb_section >}}
+### Advanced ForSt State Backends Options
+
+Advanced options to tune ForSt and ForSt checkpoints.
+
+{{< generated/expert_forst_section >}}
+
### State Changelog Options
Please refer to [State Backends]({{< ref "docs/ops/state/state_backends#enabling-changelog" >}}) for information on
@@ -486,13 +508,17 @@ using State Changelog. {{< generated/state_changelog_section >}}
These settings take effect when the `state.changelog.storage` is set to `filesystem` (see [above](#state-changelog-storage)).
{{< generated/fs_state_changelog_configuration >}}
-**RocksDB Configurable Options**
+### RocksDB Configurable Options
These options give fine-grained control over the behavior and resources of ColumnFamilies.
With the introduction of `state.backend.rocksdb.memory.managed` and `state.backend.rocksdb.memory.fixed-per-slot` (Apache Flink 1.10), it should be only necessary to use the options here for advanced performance tuning. These options here can also be specified in the application program via `RocksDBStateBackend.setRocksDBOptions(RocksDBOptionsFactory)`.
{{< generated/rocksdb_configurable_configuration >}}
+### ForSt State Backend Configurable Options
+
+{{< generated/forst_configurable_configuration >}}
+
### Advanced Fault Tolerance Options
*These parameters can help with problems related to failover and to components erroneously considering each other as failed.*
diff --git a/docs/content/docs/ops/metrics.md b/docs/content/docs/ops/metrics.md
index f091e2d9d31..70a99457e80 100644
--- a/docs/content/docs/ops/metrics.md
+++ b/docs/content/docs/ops/metrics.md
@@ -1531,7 +1531,12 @@ Note that for failed checkpoints, metrics are updated on a best efforts basis an
### RocksDB
Certain RocksDB native metrics are available but disabled by default, you can find full documentation [here]({{< ref "docs/deployment/config" >}}#rocksdb-native-metrics)
-### ForStDB
+### ForSt
+
+Certain ForSt native metrics are available but disabled by default, you can find full documentation [here]({{< ref "docs/deployment/config" >}}#forst-native-metrics)
+
+Besides that, we support the following metrics:
+
<table class="table table-bordered">
<thead>
<tr>
diff --git a/docs/content/docs/ops/state/large_state_tuning.md b/docs/content/docs/ops/state/large_state_tuning.md
index e783d2b56c1..d743a0f05cf 100644
--- a/docs/content/docs/ops/state/large_state_tuning.md
+++ b/docs/content/docs/ops/state/large_state_tuning.md
@@ -97,13 +97,27 @@ the same time. For applications with large state in Flink, this often ties up to
When a savepoint is manually triggered, it may be in process concurrently with an ongoing checkpoint.
-## Tuning RocksDB
+## Tuning RocksDB or ForSt
The state storage workhorse of many large scale Flink streaming applications is the *RocksDB State Backend*.
The backend scales well beyond main memory and reliably stores large [keyed state]({{< ref "docs/dev/datastream/fault-tolerance/state" >}}).
+If you are handling very large state, even exceeding the local disk space of the TaskManagers, you may want to consider
+using the disaggregated state store [ForStStateBackend]({{< ref "docs/deployment/config" >}}#forst-state-backend).
+This backend stores the state in a separate storage system, such as HDFS or S3, and only keeps the
+state metadata and cache in the TaskManagers. And the [State API V2]({{< ref "docs/dev/datastream/fault-tolerance/state_v2" >}})
+is also recommended to cooperate with ForStStateBackend for large state applications.
+
RocksDB's performance can vary with configuration, this section outlines some best-practices for tuning jobs that use the RocksDB State Backend.
+{{< hint info >}}
+The design of ForSt is very similar to RocksDB, and the configurable options are almost the same,
+so you can refer to following sections to configure ForSt.
+
+The following article is introduced from the perspective of RocksDB. If you want to configure ForSt
+in a similar way, you need to use the corresponding configuration under ForSt.
+{{< /hint >}}
+
### Incremental Checkpoints
When it comes to reducing the time that checkpoints take, activating incremental checkpoints should be one of the first considerations.
diff --git a/docs/content/docs/ops/state/state_backends.md b/docs/content/docs/ops/state/state_backends.md
index 6594018422f..b3cb7462333 100644
--- a/docs/content/docs/ops/state/state_backends.md
+++ b/docs/content/docs/ops/state/state_backends.md
@@ -95,12 +95,50 @@ Certain RocksDB native metrics are available but disabled by default, you can fi
The total memory amount of RocksDB instance(s) per slot can also be bounded, please refer to documentation [here]({{< ref "docs/ops/state/large_state_tuning" >}}#bounding-rocksdb-memory-usage) for details.
+## The ForStStateBackend
+
+The *ForStStateBackend* is a state backend that is based on [ForSt project](https://github.com/ververica/ForSt),
+which is also a LSM-tree structured key-value store and built on top of the RocksDB.
+It is designed to provide a more efficient way to store and access state in Flink applications.
+Most importantly, it can hold its sst files on remote file systems that Flink supports, such as HDFS, S3, etc.
+This allows Flink to scale the state size beyond the local disk capacity of the TaskManager.
+Moreover, by putting the sst files on remote file systems, it can also provide a more lightweight
+way to perform checkpoint and recovery.
+
+The ForStStateBackend is still in the experimental stage and is not fully available for production.
+It always performs asynchronous incremental snapshots.
+
+The ForStStateBackend is encouraged for:
+
+- Jobs with very large state, long windows, large key/value states. Local disk may not be enough to
+store the state.
+- All high-availability setups.
+- Asynchronous state access is preferred. Since the ForStStateBackend is the only one supporting
+asynchronous state access.
+- Jobs that require lightweight checkpoint and recovery, such as cloud-native applications.
+
+Limitations of the ForStStateBackend (for now):
+
+- Same as EmbeddedRocksDBStateBackend, the maximum supported size per key and per value is 2^31 bytes each.
+- Does not support canonical savepoint, full snapshot, changelog and file-merging checkpoints.
+Always perform incremental snapshots.
+
+Compared with EmbeddedRocksDBStateBackend, ForStStateBackend stores data on remote file system, thus
+the amount of state that you can keep is unlimited. The local disk of TaskManager is only used to
+store cache of file, to provide better performance. Note that when most of the active state is on
+remote file system, the performance of state access may be affected by the network latency. Flink
+introduces asynchronous state access to mitigate this issue. If you are using the asynchronous state
+methods in State API V2, you can benefit from the asynchronous state access. To get familiar with the
+State API V2, please refer to the [State API V2 documentation]({{< ref "docs/dev/datastream/fault-tolerance/state_v2" >}}).
+
## Choose The Right State Backend
When deciding between `HashMapStateBackend` and `RocksDB`, it is a choice between performance and scalability.
`HashMapStateBackend` is very fast as each state access and update operates on objects on the Java heap; however, state size is limited by available memory within the cluster.
-On the other hand, `RocksDB` can scale based on available disk space and is the only state backend to support incremental snapshots.
+On the other hand, `RocksDB` can scale based on available disk space.
However, each state access and update requires (de-)serialization and potentially reading from disk which leads to average performance that is an order of magnitude slower than the memory state backends.
+If you are handling very large state even exceeding the available disk space,
+or you prefer a fast rescale under cloud-native setup, you should consider using `ForStStateBackend`.
{{< hint info >}}
In Flink 1.13 we unified the binary format of Flink's savepoints. That means you can take a savepoint and then restore from it using a different state backend.
@@ -150,8 +188,18 @@ If you want to use the `EmbeddedRocksDBStateBackend` in your IDE or configure it
</dependency>
```
+Same for `ForStStateBackend`:
+```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-statebackend-forst</artifactId>
+ <version>{{< version >}}</version>
+ <scope>provided</scope>
+</dependency>
+```
+
{{< hint info >}}
-Since RocksDB is part of the default Flink distribution, you do not need this dependency if you are not using any RocksDB code in your job and configure the state backend via `state.backend.type` and further [checkpointing]({{< ref "docs/deployment/config" >}}#checkpointing) and [RocksDB-specific]({{< ref "docs/deployment/config" >}}#rocksdb-state-backend) parameters in your [Flink configuration file]({{< ref "docs/deployment/config#flink-configuration-file" >}}).
+Since RocksDB and ForSt is part of the default Flink distribution, you do not need this dependency if you are not using any RocksDB code in your job and configure the state backend via `state.backend.type` and further [checkpointing]({{< ref "docs/deployment/config" >}}#checkpointing) and [RocksDB-specific]({{< ref "docs/deployment/config" >}}#rocksdb-state-backend) or [ForSt-specific]({{< ref "docs/deployment/config" >}}#forst-state-backend) parameters in your [Flink configuration file] [...]
{{< /hint >}}
@@ -159,9 +207,10 @@ Since RocksDB is part of the default Flink distribution, you do not need this de
A default state backend can be configured in the [Flink configuration file]({{< ref "docs/deployment/config#flink-configuration-file" >}}), using the configuration key `state.backend.type`.
-Possible values for the config entry are *hashmap* (HashMapStateBackend), *rocksdb* (EmbeddedRocksDBStateBackend), or the fully qualified class
+Possible values for the config entry are *hashmap* (HashMapStateBackend), *rocksdb* (EmbeddedRocksDBStateBackend), *forst* (ForStStateBackend) or the fully qualified class
name of the class that implements the state backend factory {{< gh_link file="flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendFactory.java" name="StateBackendFactory" >}},
-such as `org.apache.flink.state.rocksdb.EmbeddedRocksDBStateBackendFactory` for EmbeddedRocksDBStateBackend.
+such as `org.apache.flink.state.rocksdb.EmbeddedRocksDBStateBackendFactory` for EmbeddedRocksDBStateBackend
+and `org.apache.flink.state.forst.ForStStateBackendFactory` for ForStStateBackend.
The `execution.checkpointing.dir` option defines the directory to which all backends write checkpoint data and meta data files.
You can find more details about the checkpoint directory structure [here]({{< ref "docs/ops/state/checkpoints" >}}#directory-structure).
diff --git a/docs/layouts/shortcodes/generated/expert_forst_section.html b/docs/layouts/shortcodes/generated/expert_forst_section.html
new file mode 100644
index 00000000000..ae289ad4582
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/expert_forst_section.html
@@ -0,0 +1,54 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>state.backend.forst.executor.inline-coordinator</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to let the task thread be the coordinator thread responsible for distributing requests. If set to 'true', the task thread will be responsible for distributing requests, otherwise, a dedicated coordinator thread will be used. The default value is 'false'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.executor.inline-write</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to let write requests be executed within the coordinator thread. If set to 'true', write requests will be executed within the coordinator thread, otherwise, a dedicated write thread will be used. The default value is 'true'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.local-dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The local directory (on the TaskManager) where ForSt puts some metadata files. By default, it will be <WORKING_DIR>/tmp. See <code class="highlighter-rouge">process.taskmanager.working-dir</code> for more details.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.fixed-per-slot</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>The fixed total amount of memory per slot, shared among all ForSt instances.This option overrides the 'state.backend.forst.memory.managed' option.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.fixed-per-tm</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>The fixed total amount of memory per Task Manager, shared among all ForSt instances. This is a cluster-level option. This option only takes effect if 'state.backend.forst.memory.managed' is set to false and 'state.backend.forst.memory.fixed-per-slot' is not configured. If so, 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 s [...]
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.managed</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>If set true, 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.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.options-factory</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>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.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/forst_configurable_configuration.html b/docs/layouts/shortcodes/generated/forst_configurable_configuration.html
new file mode 100644
index 00000000000..a6aac0a00f4
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/forst_configurable_configuration.html
@@ -0,0 +1,168 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>state.backend.forst.block.blocksize</h5></td>
+ <td style="word-wrap: break-word;">4 kb</td>
+ <td>MemorySize</td>
+ <td>The approximate size (in bytes) of user data packed per block. The default blocksize is '4KB'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.block.cache-size</h5></td>
+ <td style="word-wrap: break-word;">8 mb</td>
+ <td>MemorySize</td>
+ <td>The amount of the cache for data blocks in ForSt. The default block-cache size is '8MB'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.block.metadata-blocksize</h5></td>
+ <td style="word-wrap: break-word;">4 kb</td>
+ <td>MemorySize</td>
+ <td>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'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.bloom-filter.bits-per-key</h5></td>
+ <td style="word-wrap: break-word;">10.0</td>
+ <td>Double</td>
+ <td>Bits per key that the bloom filter will use, this only takes effect when the bloom filter is used. The default value is 10.0.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.bloom-filter.block-based-mode</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If set 'true', ForSt will use block-based filter instead of full filter, this only takes effect when bloom filter is used. The default value is 'false'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.compaction.filter.periodic-compaction-time</h5></td>
+ <td style="word-wrap: break-word;">30 d</td>
+ <td>Duration</td>
+ <td>Periodic compaction could speed up expired state entries cleanup, especially for state entries rarely accessed. Files older than this value will be picked up for compaction, and re-written to the same level as they were before. It makes sure a file goes through compaction filters periodically. 0 means turning off periodic compaction.The default value is '30 d' (30 days).</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.compaction.filter.query-time-after-num-entries</h5></td>
+ <td style="word-wrap: break-word;">1000</td>
+ <td>Long</td>
+ <td>Number of state entries to process by compaction filter before updating current timestamp. Updating the timestamp more often can improve cleanup speed, but it decreases compaction performance because it uses JNI calls from native code.The default value is '1000L'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.compaction.level.max-size-level-base</h5></td>
+ <td style="word-wrap: break-word;">256 mb</td>
+ <td>MemorySize</td>
+ <td>The upper-bound of the total size of level base files in bytes. The default value is '256 mb'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.compaction.level.target-file-size-base</h5></td>
+ <td style="word-wrap: break-word;">64 mb</td>
+ <td>MemorySize</td>
+ <td>The target file size for compaction, which determines a level-1 file size. The default value is '64 mb'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.compaction.level.use-dynamic-size</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If true, ForSt will pick target size of each level dynamically. From an empty key-value store, ForSt would make last level the base level, which means merging L0 data into the last level, until it exceeds max_bytes_for_level_base. And then repeat this process for second last level and so on. The default value is 'false'. For more information, please refer to <a href="https://github.com/facebook/rocksdb/wiki/Leveled-Compaction#level_compaction_dynamic_level_bytes-is-true"> [...]
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.compaction.style</h5></td>
+ <td style="word-wrap: break-word;">LEVEL</td>
+ <td><p>Enum</p></td>
+ <td>The specified compaction style for DB. Candidate compaction style is LEVEL, FIFO, UNIVERSAL or NONE, and Flink chooses 'LEVEL' as default style.<br /><br />Possible values:<ul><li>"LEVEL"</li><li>"UNIVERSAL"</li><li>"FIFO"</li><li>"NONE"</li></ul></td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.compression.per.level</h5></td>
+ <td style="word-wrap: break-word;">SNAPPY_COMPRESSION</td>
+ <td><p>List<Enum></p></td>
+ <td>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 <code class="highlighter-rouge">state.backend.forst.compaction.level.use-dynamic-size</code> is true, compression_per_level[0] still [...]
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.files.open</h5></td>
+ <td style="word-wrap: break-word;">-1</td>
+ <td>Integer</td>
+ <td>The maximum number of open files (per stateful operator) that can be used by the ForSt, '-1' means no limit. The default value is '-1'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.log.dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>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.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.log.file-num</h5></td>
+ <td style="word-wrap: break-word;">4</td>
+ <td>Integer</td>
+ <td>The maximum number of files ForSt should keep for information logging (Default setting: 4).</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.log.level</h5></td>
+ <td style="word-wrap: break-word;">INFO_LEVEL</td>
+ <td><p>Enum</p></td>
+ <td>The specified information logging level for ForSt. If unset, Flink will use <code class="highlighter-rouge">INFO_LEVEL</code>.<br />Note: ForSt info logs will not be written to the TaskManager logs and there is no rolling strategy, unless you configure <code class="highlighter-rouge">state.backend.forst.log.dir</code>, <code class="highlighter-rouge">state.backend.forst.log.max-file-size</code>, and <code class="highlighter-rouge">state.backend.forst.log.file-num</code> a [...]
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.log.max-file-size</h5></td>
+ <td style="word-wrap: break-word;">25 mb</td>
+ <td>MemorySize</td>
+ <td>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'. </td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.rescaling.use-delete-files-in-range</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If true, during rescaling, the deleteFilesInRange API will be invoked to clean up the useless key-values so that primary storage space can be reclaimed more promptly.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.restore-overlap-fraction-threshold</h5></td>
+ <td style="word-wrap: break-word;">0.0</td>
+ <td>Double</td>
+ <td>The threshold of overlap fraction between the state handle's key-group range and target key-group range. When restore base DB, only the handle which overlap fraction greater than or equal to threshold has a chance to be an initial handle. The default value is 0.0, there is always a handle will be selected for initialization. </td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.thread.num</h5></td>
+ <td style="word-wrap: break-word;">2</td>
+ <td>Integer</td>
+ <td>The maximum number of concurrent background flush and compaction jobs (per stateful operator). The default value is '2'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.use-bloom-filter</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether every newly created SST file will contain a Bloom filter. Default 'false'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.use-ingest-db-restore-mode</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>A recovery mode that directly clips and ingests multiple DBs during state recovery if the keys in the SST files does not exceed the declared key-group range.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.write-batch-size</h5></td>
+ <td style="word-wrap: break-word;">2 mb</td>
+ <td>MemorySize</td>
+ <td>The max size of the consumed memory for ForSt batch write, will flush just based on item count if this config set to 0.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.writebuffer.count</h5></td>
+ <td style="word-wrap: break-word;">2</td>
+ <td>Integer</td>
+ <td>The maximum number of write buffers that are built up in memory. The default value is '2'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.writebuffer.number-to-merge</h5></td>
+ <td style="word-wrap: break-word;">1</td>
+ <td>Integer</td>
+ <td>The minimum number of write buffers that will be merged together before writing to storage. The default value is '1'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.writebuffer.size</h5></td>
+ <td style="word-wrap: break-word;">64 mb</td>
+ <td>MemorySize</td>
+ <td>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'.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/forst_configuration.html b/docs/layouts/shortcodes/generated/forst_configuration.html
new file mode 100644
index 00000000000..56d6855ad82
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/forst_configuration.html
@@ -0,0 +1,120 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>state.backend.forst.cache.dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The directory where ForSt caches its SST files, fallback to state.backend.forst.local-dir/cache if not configured.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.cache.reserve-size</h5></td>
+ <td style="word-wrap: break-word;">-1</td>
+ <td>Long</td>
+ <td>The reserved size of cache, when set to a positive number. Meaning that the cache will reserve the specified size of disk space. This option and the 'state.backend.forst.cache.size-based-limit' option can be set simultaneously, the smaller cache limit will be used as the upper limit.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.cache.size-based-limit</h5></td>
+ <td style="word-wrap: break-word;">-1</td>
+ <td>Long</td>
+ <td>The size-based capacity limit of cache, a non-positive number indicates that there is no limit. The default value is '-1', which means that the cache size is not limited by size.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.executor.inline-coordinator</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to let the task thread be the coordinator thread responsible for distributing requests. If set to 'true', the task thread will be responsible for distributing requests, otherwise, a dedicated coordinator thread will be used. The default value is 'false'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.executor.inline-write</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to let write requests be executed within the coordinator thread. If set to 'true', write requests will be executed within the coordinator thread, otherwise, a dedicated write thread will be used. The default value is 'true'.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.executor.read-io-parallelism</h5></td>
+ <td style="word-wrap: break-word;">3</td>
+ <td>Integer</td>
+ <td>The number of threads used for read IO operations in the executor.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.executor.write-io-parallelism</h5></td>
+ <td style="word-wrap: break-word;">1</td>
+ <td>Integer</td>
+ <td>The number of threads used for write IO operations in the executor. Only valid when 'state.backend.forst.executor.inline-write' is false.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.local-dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The local directory (on the TaskManager) where ForSt puts some metadata files. By default, it will be <WORKING_DIR>/tmp. See <code class="highlighter-rouge">process.taskmanager.working-dir</code> for more details.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.fixed-per-slot</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>The fixed total amount of memory per slot, shared among all ForSt instances.This option overrides the 'state.backend.forst.memory.managed' option.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.fixed-per-tm</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>The fixed total amount of memory per Task Manager, shared among all ForSt instances. This is a cluster-level option. This option only takes effect if 'state.backend.forst.memory.managed' is set to false and 'state.backend.forst.memory.fixed-per-slot' is not configured. If so, 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 s [...]
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.high-prio-pool-ratio</h5></td>
+ <td style="word-wrap: break-word;">0.1</td>
+ <td>Double</td>
+ <td>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 'state.backend.forst.memory.managed' or 'state.backend.forst.memory.fixed-per-slot' are configured.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.managed</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>If set true, 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.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.partitioned-index-filters</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>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 'state.backend.forst.memory.managed' or 'state.backend.forst.mem [...]
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.write-buffer-ratio</h5></td>
+ <td style="word-wrap: break-word;">0.5</td>
+ <td>Double</td>
+ <td>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 'state.backend.forst.memory.managed' or 'state.backend.forst.memory.fixed-per-slot' are configured.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.options-factory</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>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.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.remote-dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The remote directory where ForSt puts its SST files, fallback to state.backend.forst.local-dir if not configured. Recognized shortcut name is 'checkpoint-dir', which means that forst shares the directory with checkpoint.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.timer-service.cache-size</h5></td>
+ <td style="word-wrap: break-word;">128</td>
+ <td>Integer</td>
+ <td>The cache size per keyGroup of ForSt timer service factory. This option only has an effect when 'state.backend.forst.timer-service.factory' is configured to 'ForStDB'. Increasing this value can improve the performance of ForSt timer service, but consumes more heap memory at the same time.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.timer-service.factory</h5></td>
+ <td style="word-wrap: break-word;">ForStDB</td>
+ <td><p>Enum</p></td>
+ <td>This determines the factory for timer service state implementation.<br /><br />Possible values:<ul><li>"HEAP": Heap-based</li><li>"ForStDB": Implementation based on RocksDB</li></ul></td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/forst_native_metric_configuration.html b/docs/layouts/shortcodes/generated/forst_native_metric_configuration.html
new file mode 100644
index 00000000000..96c522050a8
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/forst_native_metric_configuration.html
@@ -0,0 +1,246 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>state.backend.forst.metrics.actual-delayed-write-rate</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the current actual delayed write rate. 0 means no delay.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.background-errors</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of background errors in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.block-cache-capacity</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor block cache capacity.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.block-cache-hit</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total count of block cache hit in ForSt (BLOCK_CACHE_HIT == BLOCK_CACHE_INDEX_HIT + BLOCK_CACHE_FILTER_HIT + BLOCK_CACHE_DATA_HIT).</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.block-cache-miss</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total count of block cache misses in ForSt (BLOCK_CACHE_MISS == BLOCK_CACHE_INDEX_MISS + BLOCK_CACHE_FILTER_MISS + BLOCK_CACHE_DATA_MISS).</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.block-cache-pinned-usage</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the memory size for the entries being pinned in block cache.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.block-cache-usage</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the memory size for the entries residing in block cache.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.bloom-filter-full-positive</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total count of reads not avoided by bloom full filter.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.bloom-filter-full-true-positive</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total count of reads not avoided by bloom full filter and the data actually exists in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.bloom-filter-useful</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total count of reads avoided by bloom filter.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.bytes-read</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of uncompressed bytes read (from memtables/cache/sst) from Get() operation in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.bytes-written</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of uncompressed bytes written by DB::{Put(), Delete(), Merge(), Write()} operations, which does not include the compaction written bytes, in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.column-family-as-variable</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to expose the column family as a variable for ForSt property based metrics.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.compaction-pending</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Track pending compactions in ForSt. Returns 1 if a compaction is pending, 0 otherwise.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.compaction-read-bytes</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the bytes read during compaction in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.compaction-write-bytes</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the bytes written during compaction in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.cur-size-active-mem-table</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the approximate size of the active memtable in bytes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.cur-size-all-mem-tables</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the approximate size of the active and unflushed immutable memtables in bytes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.estimate-live-data-size</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Estimate of the amount of live data in bytes (usually smaller than sst files size due to space amplification).</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.estimate-num-keys</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Estimate the number of keys in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.estimate-pending-compaction-bytes</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Estimated total number of bytes compaction needed to rewrite to get all levels down to under target size. Not valid for other compactions than level-based.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.estimate-table-readers-mem</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Estimate the memory used for reading SST tables, excluding memory used in block cache (e.g. filter and index blocks) in bytes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.is-write-stopped</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Track whether write has been stopped in ForSt. The metric will return 1 if write has been stopped, 0 otherwise.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.iter-bytes-read</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of uncompressed bytes read (from memtables/cache/sst) from an iterator operation in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.live-sst-files-size</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total size (bytes) of all SST files belonging to the latest version. WARNING: may slow down online queries if there are too many files.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.mem-table-flush-pending</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of pending memtable flushes in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-deletes-active-mem-table</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total number of delete entries in the active memtable.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-deletes-imm-mem-tables</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total number of delete entries in the unflushed immutable memtables.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-entries-active-mem-table</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total number of entries in the active memtable.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-entries-imm-mem-tables</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total number of entries in the unflushed immutable memtables.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-files-at-level</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of files at each level.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-immutable-mem-table</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of immutable memtables in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-live-versions</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor number of live versions. Version is an internal data structure. See ForSt file version_set.h for details. More live versions often mean more SST files are held from being deleted, by iterators or unfinished compactions.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-running-compactions</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of currently running compactions.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-running-flushes</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of currently running flushes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.num-snapshots</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the number of unfinished snapshots of the ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.size-all-mem-tables</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the approximate size of the active, unflushed immutable, and pinned immutable memtables in bytes.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.stall-micros</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the writer wait duration for compaction or flush to finish in ForSt.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.metrics.total-sst-files-size</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Monitor the total size (bytes) of all SST files of all versions. WARNING: may slow down online queries if there are too many files.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/state_backend_forst_section.html b/docs/layouts/shortcodes/generated/state_backend_forst_section.html
new file mode 100644
index 00000000000..29459c4157d
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/state_backend_forst_section.html
@@ -0,0 +1,66 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>state.backend.forst.cache.dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The directory where ForSt caches its SST files, fallback to state.backend.forst.local-dir/cache if not configured.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.cache.reserve-size</h5></td>
+ <td style="word-wrap: break-word;">-1</td>
+ <td>Long</td>
+ <td>The reserved size of cache, when set to a positive number. Meaning that the cache will reserve the specified size of disk space. This option and the 'state.backend.forst.cache.size-based-limit' option can be set simultaneously, the smaller cache limit will be used as the upper limit.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.cache.size-based-limit</h5></td>
+ <td style="word-wrap: break-word;">-1</td>
+ <td>Long</td>
+ <td>The size-based capacity limit of cache, a non-positive number indicates that there is no limit. The default value is '-1', which means that the cache size is not limited by size.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.executor.read-io-parallelism</h5></td>
+ <td style="word-wrap: break-word;">3</td>
+ <td>Integer</td>
+ <td>The number of threads used for read IO operations in the executor.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.executor.write-io-parallelism</h5></td>
+ <td style="word-wrap: break-word;">1</td>
+ <td>Integer</td>
+ <td>The number of threads used for write IO operations in the executor. Only valid when 'state.backend.forst.executor.inline-write' is false.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.memory.partitioned-index-filters</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>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 'state.backend.forst.memory.managed' or 'state.backend.forst.mem [...]
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.remote-dir</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The remote directory where ForSt puts its SST files, fallback to state.backend.forst.local-dir if not configured. Recognized shortcut name is 'checkpoint-dir', which means that forst shares the directory with checkpoint.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.timer-service.cache-size</h5></td>
+ <td style="word-wrap: break-word;">128</td>
+ <td>Integer</td>
+ <td>The cache size per keyGroup of ForSt timer service factory. This option only has an effect when 'state.backend.forst.timer-service.factory' is configured to 'ForStDB'. Increasing this value can improve the performance of ForSt timer service, but consumes more heap memory at the same time.</td>
+ </tr>
+ <tr>
+ <td><h5>state.backend.forst.timer-service.factory</h5></td>
+ <td style="word-wrap: break-word;">ForStDB</td>
+ <td><p>Enum</p></td>
+ <td>This determines the factory for timer service state implementation.<br /><br />Possible values:<ul><li>"HEAP": Heap-based</li><li>"ForStDB": Implementation based on RocksDB</li></ul></td>
+ </tr>
+ </tbody>
+</table>
diff --git a/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java b/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java
index 8b766624152..e469fd8c386 100644
--- a/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java
+++ b/flink-annotations/src/main/java/org/apache/flink/annotation/docs/Documentation.java
@@ -77,6 +77,7 @@ public final class Documentation {
public static final String SECURITY_AUTH_ZOOKEEPER = "security_auth_zk";
public static final String STATE_BACKEND_ROCKSDB = "state_backend_rocksdb";
+ public static final String STATE_BACKEND_FORST = "state_backend_forst";
public static final String STATE_LATENCY_TRACKING = "state_latency_tracking";
@@ -95,6 +96,7 @@ public final class Documentation {
"expert_high_availability_k8s";
public static final String EXPERT_SECURITY_SSL = "expert_security_ssl";
public static final String EXPERT_ROCKSDB = "expert_rocksdb";
+ public static final String EXPERT_FORST = "expert_forst";
public static final String EXPERT_CLUSTER = "expert_cluster";
public static final String EXPERT_JOB_MANAGER = "expert_jobmanager";
diff --git a/flink-docs/pom.xml b/flink-docs/pom.xml
index efa4fdee02e..160f35d65d3 100644
--- a/flink-docs/pom.xml
+++ b/flink-docs/pom.xml
@@ -134,6 +134,11 @@ under the License.
<artifactId>flink-statebackend-rocksdb</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-statebackend-forst</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-python</artifactId>
diff --git a/flink-docs/src/main/java/org/apache/flink/docs/configuration/ConfigOptionsDocGenerator.java b/flink-docs/src/main/java/org/apache/flink/docs/configuration/ConfigOptionsDocGenerator.java
index 0af7bbcf9f0..a392ae3a0e5 100644
--- a/flink-docs/src/main/java/org/apache/flink/docs/configuration/ConfigOptionsDocGenerator.java
+++ b/flink-docs/src/main/java/org/apache/flink/docs/configuration/ConfigOptionsDocGenerator.java
@@ -260,6 +260,8 @@ public class ConfigOptionsDocGenerator {
@VisibleForTesting
static String toSnakeCase(String name) {
+ // We prefer lower case 'forst' for 'ForSt' instead of 'for_st'
+ name = name.replaceAll("(ForSt)([A-Z]|$)", "forst$2");
return name.replaceAll("(.)([A-Z][a-z])", "$1_$2").toLowerCase();
}
diff --git a/flink-docs/src/main/java/org/apache/flink/docs/util/ConfigurationOptionLocator.java b/flink-docs/src/main/java/org/apache/flink/docs/util/ConfigurationOptionLocator.java
index 00e4bb65fc7..d3deb0fd777 100644
--- a/flink-docs/src/main/java/org/apache/flink/docs/util/ConfigurationOptionLocator.java
+++ b/flink-docs/src/main/java/org/apache/flink/docs/util/ConfigurationOptionLocator.java
@@ -68,6 +68,9 @@ public class ConfigurationOptionLocator {
new OptionsClassLocation(
"flink-state-backends/flink-statebackend-rocksdb",
"org.apache.flink.state.rocksdb"),
+ new OptionsClassLocation(
+ "flink-state-backends/flink-statebackend-forst",
+ "org.apache.flink.state.forst"),
new OptionsClassLocation(
"flink-table/flink-table-api-java", "org.apache.flink.table.api.config"),
new OptionsClassLocation("flink-python", "org.apache.flink.python"),
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
index 3b96bf89164..2e6c9aa19b3 100644
--- 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
@@ -39,6 +39,7 @@ 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.apache.flink.configuration.description.TextElement.text;
import static org.apache.flink.state.forst.ForStOptions.CACHE_DIRECTORY;
import static org.apache.flink.state.forst.ForStOptions.CACHE_RESERVED_SIZE;
import static org.apache.flink.state.forst.ForStOptions.CACHE_SIZE_BASE_LIMIT;
@@ -81,7 +82,7 @@ public class ForStConfigurableOptions implements Serializable {
.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 maximum number of open files (per stateful operator) that can be used by the ForSt, '-1' means no limit. "
+ "The default value is '-1'.");
public static final ConfigOption<MemorySize> LOG_MAX_FILE_SIZE =
@@ -109,7 +110,7 @@ public class ForStConfigurableOptions implements Serializable {
"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.");
+ + "If setting this option as a non-existing location, e.g '/dev/null', ForSt will then create the log under its own database folder.");
public static final ConfigOption<InfoLogLevel> LOG_LEVEL =
key("state.backend.forst.log.level")
@@ -160,7 +161,7 @@ public class ForStConfigurableOptions implements Serializable {
.withDescription(
Description.builder()
.text(
- "If true, ForSt will pick target size of each level dynamically. From an empty DB, ")
+ "If true, ForSt will pick target size of each level dynamically. From an empty key-value store, ")
.text(
"ForSt would make last level the base level, which means merging L0 data into the last level, ")
.text(
@@ -182,27 +183,27 @@ public class ForStConfigurableOptions implements Serializable {
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"
+ + "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",
+ + "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")
+ "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",
+ "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.",
+ "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.",
+ "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.",
@@ -216,16 +217,24 @@ public class ForStConfigurableOptions implements Serializable {
.memoryType()
.defaultValue(MemorySize.parse("64mb"))
.withDescription(
- "The target file size for compaction, which determines a level-1 file size. "
- + "The default value is '64MB'.");
+ Description.builder()
+ .text(
+ "The target file size for compaction, which determines a level-1 file size. "
+ + "The default value is '%s'.",
+ text(MemorySize.parse("64mb").toString()))
+ .build());
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'.");
+ Description.builder()
+ .text(
+ "The upper-bound of the total size of level base files in bytes. "
+ + "The default value is '%s'.",
+ text(MemorySize.parse("256mb").toString()))
+ .build());
public static final ConfigOption<MemorySize> WRITE_BUFFER_SIZE =
key("state.backend.forst.writebuffer.size")
@@ -289,15 +298,14 @@ public class ForStConfigurableOptions implements Serializable {
.booleanType()
.defaultValue(false)
.withDescription(
- "If true, every newly created SST file will contain a Bloom filter. "
- + "It is disabled by default.");
+ "Whether every newly created SST file will contain a Bloom filter. Default 'false'.");
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. "
+ "Bits per key that the bloom filter will use, this only takes effect when the bloom filter is used. "
+ "The default value is 10.0.");
public static final ConfigOption<Boolean> BLOOM_FILTER_BLOCK_BASED_MODE =
@@ -305,7 +313,7 @@ public class ForStConfigurableOptions implements Serializable {
.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. "
+ "If set 'true', ForSt will use block-based filter instead of full filter, this only takes effect when bloom filter is used. "
+ "The default value is 'false'.");
public static final ConfigOption<Long> COMPACT_FILTER_QUERY_TIME_AFTER_NUM_ENTRIES =
@@ -325,14 +333,14 @@ public class ForStConfigurableOptions implements Serializable {
"Periodic compaction could speed up expired state entries cleanup, especially for state"
+ " entries rarely accessed. Files older than this value will be picked up for compaction,"
+ " and re-written to the same level as they were before. It makes sure a file goes through"
- + " compaction filters periodically. 0 means turning off periodic compaction.The default value is '30days'.");
+ + " compaction filters periodically. 0 means turning off periodic compaction.The default value is '30 d' (30 days).");
public static final ConfigOption<Double> RESTORE_OVERLAP_FRACTION_THRESHOLD =
key("state.backend.forst.restore-overlap-fraction-threshold")
.doubleType()
.defaultValue(0.0)
.withDescription(
- "The threshold of overlap fraction between the handle's key-group range and target key-group range. "
+ "The threshold of overlap fraction between the state handle's key-group range and target key-group range. "
+ "When restore base DB, only the handle which overlap fraction greater than or equal to threshold "
+ "has a chance to be an initial handle. "
+ "The default value is 0.0, there is always a handle will be selected for initialization. ");
@@ -351,7 +359,7 @@ public class ForStConfigurableOptions implements Serializable {
.defaultValue(Boolean.FALSE)
.withDescription(
"If true, during rescaling, the deleteFilesInRange API will be invoked "
- + "to clean up the useless files so that local disk space can be reclaimed more promptly.");
+ + "to clean up the useless key-values so that primary storage space can be reclaimed more promptly.");
static final ConfigOption<?>[] CANDIDATE_CONFIGS =
new ConfigOption<?>[] {
diff --git a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricOptions.java b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricOptions.java
index 47603f505b8..103e700b460 100644
--- a/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricOptions.java
+++ b/flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStNativeMetricOptions.java
@@ -43,8 +43,8 @@ import java.util.Set;
* <p>Statistics based metrics would report at the database level, it can return ticker or histogram
* kind results.
*
- * <p>Properties and doc comments are taken from RocksDB documentation. See <a
- * href="https://github.com/facebook/rocksdb/blob/64324e329eb0a9b4e77241a425a1615ff524c7f1/include/rocksdb/db.h#L429">
+ * <p>Properties and doc comments are taken from ForSt documentation. See <a
+ * href="https://github.com/ververica/ForSt/blob/64324e329eb0a9b4e77241a425a1615ff524c7f1/include/rocksdb/db.h#L429">
* db.h</a> for more information.
*/
@Experimental
@@ -52,36 +52,36 @@ public class ForStNativeMetricOptions implements Serializable {
private static final long serialVersionUID = 1L;
// --------------------------------------------------------------------------------------------
- // RocksDB property based metrics, report at column family level
+ // ForSt property based metrics, report at column family level
// --------------------------------------------------------------------------------------------
public static final String METRICS_COLUMN_FAMILY_AS_VARIABLE_KEY =
- "state.backend.rocksdb.metrics" + ".column-family-as-variable";
+ "state.backend.forst.metrics" + ".column-family-as-variable";
public static final ConfigOption<Boolean> MONITOR_NUM_IMMUTABLE_MEM_TABLES =
ConfigOptions.key(ForStProperty.NumImmutableMemTable.getConfigKey())
.booleanType()
.defaultValue(false)
- .withDescription("Monitor the number of immutable memtables in RocksDB.");
+ .withDescription("Monitor the number of immutable memtables in ForSt.");
public static final ConfigOption<Boolean> MONITOR_MEM_TABLE_FLUSH_PENDING =
ConfigOptions.key(ForStProperty.MemTableFlushPending.getConfigKey())
.booleanType()
.defaultValue(false)
- .withDescription("Monitor the number of pending memtable flushes in RocksDB.");
+ .withDescription("Monitor the number of pending memtable flushes in ForSt.");
public static final ConfigOption<Boolean> TRACK_COMPACTION_PENDING =
ConfigOptions.key(ForStProperty.CompactionPending.getConfigKey())
.booleanType()
.defaultValue(false)
.withDescription(
- "Track pending compactions in RocksDB. Returns 1 if a compaction is pending, 0 otherwise.");
+ "Track pending compactions in ForSt. Returns 1 if a compaction is pending, 0 otherwise.");
public static final ConfigOption<Boolean> MONITOR_BACKGROUND_ERRORS =
ConfigOptions.key(ForStProperty.BackgroundErrors.getConfigKey())
.booleanType()
.defaultValue(false)
- .withDescription("Monitor the number of background errors in RocksDB.");
+ .withDescription("Monitor the number of background errors in ForSt.");
public static final ConfigOption<Boolean> MONITOR_CUR_SIZE_ACTIVE_MEM_TABLE =
ConfigOptions.key(ForStProperty.CurSizeActiveMemTable.getConfigKey())
@@ -137,7 +137,7 @@ public class ForStNativeMetricOptions implements Serializable {
ConfigOptions.key(ForStProperty.EstimateNumKeys.getConfigKey())
.booleanType()
.defaultValue(false)
- .withDescription("Estimate the number of keys in RocksDB.");
+ .withDescription("Estimate the number of keys in ForSt.");
public static final ConfigOption<Boolean> ESTIMATE_TABLE_READERS_MEM =
ConfigOptions.key(ForStProperty.EstimateTableReadersMem.getConfigKey())
@@ -145,13 +145,13 @@ public class ForStNativeMetricOptions implements Serializable {
.defaultValue(false)
.withDescription(
"Estimate the memory used for reading SST tables, excluding memory"
- + " used in block cache (e.g.,filter and index blocks) in bytes.");
+ + " used in block cache (e.g. filter and index blocks) in bytes.");
public static final ConfigOption<Boolean> MONITOR_NUM_SNAPSHOTS =
ConfigOptions.key(ForStProperty.NumSnapshots.getConfigKey())
.booleanType()
.defaultValue(false)
- .withDescription("Monitor the number of unreleased snapshots of the database.");
+ .withDescription("Monitor the number of unfinished snapshots of the ForSt.");
public static final ConfigOption<Boolean> MONITOR_NUM_LIVE_VERSIONS =
ConfigOptions.key(ForStProperty.NumLiveVersions.getConfigKey())
@@ -159,7 +159,7 @@ public class ForStNativeMetricOptions implements Serializable {
.defaultValue(false)
.withDescription(
"Monitor number of live versions. Version is an internal data structure. "
- + "See RocksDB file version_set.h for details. More live versions often mean more SST files are held "
+ + "See ForSt file version_set.h for details. More live versions often mean more SST files are held "
+ "from being deleted, by iterators or unfinished compactions.");
public static final ConfigOption<Boolean> ESTIMATE_LIVE_DATA_SIZE =
@@ -174,7 +174,7 @@ public class ForStNativeMetricOptions implements Serializable {
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the total size (bytes) of all SST files of all versions."
+ "Monitor the total size (bytes) of all SST files of all versions. "
+ "WARNING: may slow down online queries if there are too many files.");
public static final ConfigOption<Boolean> MONITOR_LIVE_SST_FILES_SIZE =
@@ -182,7 +182,7 @@ public class ForStNativeMetricOptions implements Serializable {
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the total size (bytes) of all SST files belonging to the latest version."
+ "Monitor the total size (bytes) of all SST files belonging to the latest version. "
+ "WARNING: may slow down online queries if there are too many files.");
public static final ConfigOption<Boolean> ESTIMATE_PENDING_COMPACTION_BYTES =
@@ -190,7 +190,7 @@ public class ForStNativeMetricOptions implements Serializable {
.booleanType()
.defaultValue(false)
.withDescription(
- "Estimated total number of bytes compaction needs to rewrite to get all levels "
+ "Estimated total number of bytes compaction needed to rewrite to get all levels "
+ "down to under target size. Not valid for other compactions than level-based.");
public static final ConfigOption<Boolean> MONITOR_NUM_RUNNING_COMPACTIONS =
@@ -217,7 +217,7 @@ public class ForStNativeMetricOptions implements Serializable {
.booleanType()
.defaultValue(false)
.withDescription(
- "Track whether write has been stopped in RocksDB. Returns 1 if write has been stopped, 0 otherwise.");
+ "Track whether write has been stopped in ForSt. The metric will return 1 if write has been stopped, 0 otherwise.");
public static final ConfigOption<Boolean> BLOCK_CACHE_CAPACITY =
ConfigOptions.key(ForStProperty.BlockCacheCapacity.getConfigKey())
@@ -244,91 +244,91 @@ public class ForStNativeMetricOptions implements Serializable {
.booleanType()
.defaultValue(false)
.withDescription(
- "Whether to expose the column family as a variable for RocksDB property based metrics.");
+ "Whether to expose the column family as a variable for ForSt property based metrics.");
public static final ConfigOption<Boolean> MONITOR_NUM_FILES_AT_LEVEL =
- ConfigOptions.key("state.backend.rocksdb.metrics.num-files-at-level")
+ ConfigOptions.key("state.backend.forst.metrics.num-files-at-level")
.booleanType()
.defaultValue(false)
.withDescription("Monitor the number of files at each level.");
// --------------------------------------------------------------------------------------------
- // RocksDB statistics based metrics, report at database level
+ // ForSt statistics based metrics, report at database level
// --------------------------------------------------------------------------------------------
public static final ConfigOption<Boolean> MONITOR_BLOCK_CACHE_HIT =
- ConfigOptions.key("state.backend.rocksdb.metrics.block-cache-hit")
+ ConfigOptions.key("state.backend.forst.metrics.block-cache-hit")
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the total count of block cache hit in RocksDB (BLOCK_CACHE_HIT == BLOCK_CACHE_INDEX_HIT + BLOCK_CACHE_FILTER_HIT + BLOCK_CACHE_DATA_HIT).");
+ "Monitor the total count of block cache hit in ForSt (BLOCK_CACHE_HIT == BLOCK_CACHE_INDEX_HIT + BLOCK_CACHE_FILTER_HIT + BLOCK_CACHE_DATA_HIT).");
public static final ConfigOption<Boolean> MONITOR_BLOCK_CACHE_MISS =
- ConfigOptions.key("state.backend.rocksdb.metrics.block-cache-miss")
+ ConfigOptions.key("state.backend.forst.metrics.block-cache-miss")
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the total count of block cache misses in RocksDB (BLOCK_CACHE_MISS == BLOCK_CACHE_INDEX_MISS + BLOCK_CACHE_FILTER_MISS + BLOCK_CACHE_DATA_MISS).");
+ "Monitor the total count of block cache misses in ForSt (BLOCK_CACHE_MISS == BLOCK_CACHE_INDEX_MISS + BLOCK_CACHE_FILTER_MISS + BLOCK_CACHE_DATA_MISS).");
public static final ConfigOption<Boolean> MONITOR_BLOOM_FILTER_USEFUL =
- ConfigOptions.key("state.backend.rocksdb.metrics.bloom-filter-useful")
+ ConfigOptions.key("state.backend.forst.metrics.bloom-filter-useful")
.booleanType()
.defaultValue(false)
.withDescription("Monitor the total count of reads avoided by bloom filter.");
public static final ConfigOption<Boolean> MONITOR_BLOOM_FILTER_FULL_POSITIVE =
- ConfigOptions.key("state.backend.rocksdb.metrics.bloom-filter-full-positive")
+ ConfigOptions.key("state.backend.forst.metrics.bloom-filter-full-positive")
.booleanType()
.defaultValue(false)
.withDescription(
"Monitor the total count of reads not avoided by bloom full filter.");
public static final ConfigOption<Boolean> MONITOR_BLOOM_FILTER_FULL_TRUE_POSITIVE =
- ConfigOptions.key("state.backend.rocksdb.metrics.bloom-filter-full-true-positive")
+ ConfigOptions.key("state.backend.forst.metrics.bloom-filter-full-true-positive")
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the total count of reads not avoided by bloom full filter and the data actually exists in RocksDB.");
+ "Monitor the total count of reads not avoided by bloom full filter and the data actually exists in ForSt.");
public static final ConfigOption<Boolean> MONITOR_BYTES_READ =
- ConfigOptions.key("state.backend.rocksdb.metrics.bytes-read")
+ ConfigOptions.key("state.backend.forst.metrics.bytes-read")
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the number of uncompressed bytes read (from memtables/cache/sst) from Get() operation in RocksDB.");
+ "Monitor the number of uncompressed bytes read (from memtables/cache/sst) from Get() operation in ForSt.");
public static final ConfigOption<Boolean> MONITOR_ITER_BYTES_READ =
- ConfigOptions.key("state.backend.rocksdb.metrics.iter-bytes-read")
+ ConfigOptions.key("state.backend.forst.metrics.iter-bytes-read")
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the number of uncompressed bytes read (from memtables/cache/sst) from an iterator operation in RocksDB.");
+ "Monitor the number of uncompressed bytes read (from memtables/cache/sst) from an iterator operation in ForSt.");
public static final ConfigOption<Boolean> MONITOR_BYTES_WRITTEN =
- ConfigOptions.key("state.backend.rocksdb.metrics.bytes-written")
+ ConfigOptions.key("state.backend.forst.metrics.bytes-written")
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the number of uncompressed bytes written by DB::{Put(), Delete(), Merge(), Write()} operations, which does not include the compaction written bytes, in RocksDB.");
+ "Monitor the number of uncompressed bytes written by DB::{Put(), Delete(), Merge(), Write()} operations, which does not include the compaction written bytes, in ForSt.");
public static final ConfigOption<Boolean> MONITOR_COMPACTION_READ_BYTES =
- ConfigOptions.key("state.backend.rocksdb.metrics.compaction-read-bytes")
+ ConfigOptions.key("state.backend.forst.metrics.compaction-read-bytes")
.booleanType()
.defaultValue(false)
- .withDescription("Monitor the bytes read during compaction in RocksDB.");
+ .withDescription("Monitor the bytes read during compaction in ForSt.");
public static final ConfigOption<Boolean> MONITOR_COMPACTION_WRITE_BYTES =
- ConfigOptions.key("state.backend.rocksdb.metrics.compaction-write-bytes")
+ ConfigOptions.key("state.backend.forst.metrics.compaction-write-bytes")
.booleanType()
.defaultValue(false)
- .withDescription("Monitor the bytes written during compaction in RocksDB.");
+ .withDescription("Monitor the bytes written during compaction in ForSt.");
public static final ConfigOption<Boolean> MONITOR_STALL_MICROS =
- ConfigOptions.key("state.backend.rocksdb.metrics.stall-micros")
+ ConfigOptions.key("state.backend.forst.metrics.stall-micros")
.booleanType()
.defaultValue(false)
.withDescription(
- "Monitor the duration of writer requiring to wait for compaction or flush to finish in RocksDB.");
+ "Monitor the writer wait duration for compaction or flush to finish in ForSt.");
/** Creates a {@link ForStNativeMetricOptions} based on an external configuration. */
public static ForStNativeMetricOptions fromConfig(ReadableConfig config) {
@@ -666,14 +666,14 @@ public class ForStNativeMetricOptions implements Serializable {
}
/**
- * @return the enabled RocksDB property-based metrics
+ * @return the enabled ForSt property-based metrics
*/
public Collection<ForStProperty> getProperties() {
return Collections.unmodifiableCollection(properties);
}
/**
- * @return the enabled RocksDB statistics metrics.
+ * @return the enabled ForSt statistics metrics.
*/
public Collection<TickerType> getMonitorTickerTypes() {
return Collections.unmodifiableCollection(monitorTickerTypes);
@@ -689,7 +689,7 @@ public class ForStNativeMetricOptions implements Serializable {
}
/**
- * @return true if RocksDB statistics metrics are enabled, false otherwise.
+ * @return true if ForSt statistics metrics are enabled, false otherwise.
*/
public boolean isStatisticsEnabled() {
return !monitorTickerTypes.isEmpty();
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
index 73e43443332..8c38ed83038 100644
--- 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
@@ -35,6 +35,7 @@ import static org.apache.flink.state.forst.ForStStateBackend.REMOTE_SHORTCUT_CHE
public class ForStOptions {
/** The local directory (on the TaskManager) where ForSt puts some meta files. */
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
public static final ConfigOption<String> LOCAL_DIRECTORIES =
ConfigOptions.key("state.backend.forst.local-dir")
.stringType()
@@ -42,7 +43,7 @@ public class ForStOptions {
.withDescription(
Description.builder()
.text(
- "The local directory (on the TaskManager) where ForSt puts some meta files. Per default, it will be <WORKING_DIR>/tmp. See %s for more details.",
+ "The local directory (on the TaskManager) where ForSt puts some metadata files. By default, it will be <WORKING_DIR>/tmp. See %s for more details.",
TextElement.code(
ClusterOptions
.TASK_MANAGER_PROCESS_WORKING_DIR_BASE
@@ -50,6 +51,7 @@ public class ForStOptions {
.build());
/** The remote directory where ForSt puts its SST files. */
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<String> REMOTE_DIRECTORY =
ConfigOptions.key("state.backend.forst.remote-dir")
.stringType()
@@ -60,6 +62,7 @@ public class ForStOptions {
+ " Recognized shortcut name is '%s', which means that forst shares the directory with checkpoint.",
LOCAL_DIRECTORIES.key(), REMOTE_SHORTCUT_CHECKPOINT));
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<String> CACHE_DIRECTORY =
ConfigOptions.key("state.backend.forst.cache.dir")
.stringType()
@@ -69,24 +72,29 @@ public class ForStOptions {
"The directory where ForSt caches its SST files, fallback to %s/cache if not configured.",
LOCAL_DIRECTORIES.key()));
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<Long> CACHE_SIZE_BASE_LIMIT =
ConfigOptions.key("state.backend.forst.cache.size-based-limit")
.longType()
.defaultValue(-1L)
.withDescription(
- "The size-based capacity limit of cache, a non-positive number indicates that there is no limit.");
+ "The size-based capacity limit of cache, a non-positive number indicates that there is no limit. "
+ + "The default value is '-1', which means that the cache size is not limited by size.");
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<Long> CACHE_RESERVED_SIZE =
ConfigOptions.key("state.backend.forst.cache.reserve-size")
.longType()
.defaultValue(-1L)
.withDescription(
- "The reserved size of cache, when set to a positive number, dynamic space checking will be leveraged. "
- + "This option and the "
+ "The reserved size of cache, when set to a positive number. Meaning that "
+ + "the cache will reserve the specified size of disk space. "
+ + "This option and the '"
+ CACHE_SIZE_BASE_LIMIT.key()
- + "option can be set simultaneously, the smaller cache limit will be used as the upper limit.");
+ + "' option can be set simultaneously, the smaller cache limit will be used as the upper limit.");
/** The options factory class for ForSt to create DBOptions and ColumnFamilyOptions. */
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
public static final ConfigOption<String> OPTIONS_FACTORY =
ConfigOptions.key("state.backend.forst.options-factory")
.stringType()
@@ -96,37 +104,40 @@ public class ForStOptions {
+ "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.");
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
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 "
+ "If set true, 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.");
+ + "block caches, etc.");
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
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.",
+ "The fixed total amount of memory per slot, shared among all ForSt instances."
+ + "This option overrides the '%s' option.",
USE_MANAGED_MEMORY.key()));
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
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 "
+ "The fixed total amount of memory per Task Manager, shared among all ForSt instances. This is a cluster-level option. "
+ + "This option only takes effect if '%s' is set to false and '%s' is not configured. If so, "
+ "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",
+ + "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 =
@@ -149,10 +160,11 @@ public class ForStOptions {
+ "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()));
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<Boolean> USE_PARTITIONED_INDEX_FILTERS =
ConfigOptions.key("state.backend.forst.memory.partitioned-index-filters")
.booleanType()
- .defaultValue(false)
+ .defaultValue(true)
.withDescription(
String.format(
"With partitioning, the index/filter block of an SST file is partitioned into smaller blocks with "
@@ -163,7 +175,7 @@ public class ForStOptions {
USE_MANAGED_MEMORY.key(), FIX_PER_SLOT_MEMORY_SIZE.key()));
/** Choice of timer service implementation. */
- @Documentation.Section(Documentation.Sections.STATE_BACKEND_ROCKSDB)
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<ForStStateBackend.PriorityQueueStateType>
TIMER_SERVICE_FACTORY =
ConfigOptions.key("state.backend.forst.timer-service.factory")
@@ -172,33 +184,42 @@ public class ForStOptions {
.withDescription(
"This determines the factory for timer service state implementation.");
- /** The cache size per key-group for ROCKSDB timer service factory implementation. */
- @Documentation.Section(Documentation.Sections.STATE_BACKEND_ROCKSDB)
+ /** The cache size per key-group for ForSt timer service factory implementation. */
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<Integer> FORST_TIMER_SERVICE_FACTORY_CACHE_SIZE =
ConfigOptions.key("state.backend.forst.timer-service.cache-size")
.intType()
.defaultValue(128)
.withDescription(
String.format(
- "The cache size per keyGroup of rocksdb timer service factory. This option only has an effect "
+ "The cache size per keyGroup of ForSt timer service factory. This option only has an effect "
+ "when '%s' is configured to '%s'. Increasing this value can improve the performance "
- + "of rocksdb timer service, but consumes more heap memory at the same time.",
+ + "of ForSt timer service, but consumes more heap memory at the same time.",
TIMER_SERVICE_FACTORY.key(), ForStDB.name()));
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
public static final ConfigOption<Boolean> EXECUTOR_COORDINATOR_INLINE =
ConfigOptions.key("state.backend.forst.executor.inline-coordinator")
.booleanType()
.defaultValue(false)
.withDescription(
- "Whether to let the task thread be the coordinator thread responsible for distributing requests.");
+ "Whether to let the task thread be the coordinator thread responsible for distributing requests. "
+ + "If set to 'true', the task thread will be responsible for distributing requests, "
+ + "otherwise, a dedicated coordinator thread will be used. "
+ + "The default value is 'false'.");
+ @Documentation.Section(Documentation.Sections.EXPERT_FORST)
public static final ConfigOption<Boolean> EXECUTOR_WRITE_IO_INLINE =
ConfigOptions.key("state.backend.forst.executor.inline-write")
.booleanType()
.defaultValue(true)
.withDescription(
- "Whether to let write requests executed within the coordinator thread.");
+ "Whether to let write requests be executed within the coordinator thread. "
+ + "If set to 'true', write requests will be executed within the coordinator thread, "
+ + "otherwise, a dedicated write thread will be used. "
+ + "The default value is 'true'.");
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<Integer> EXECUTOR_READ_IO_PARALLELISM =
ConfigOptions.key("state.backend.forst.executor.read-io-parallelism")
.intType()
@@ -206,6 +227,7 @@ public class ForStOptions {
.withDescription(
"The number of threads used for read IO operations in the executor.");
+ @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
public static final ConfigOption<Integer> EXECUTOR_WRITE_IO_PARALLELISM =
ConfigOptions.key("state.backend.forst.executor.write-io-parallelism")
.intType()