[FLINK-37668][state] Make compaction filter related config take effect for RocksDBStateBackend

Author: masteryhx

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBConfigurableOptions.java

@@ -380,7 +380,9 @@ public class RocksDBConfigurableOptions implements Serializable {
                 BLOOM_FILTER_BLOCK_BASED_MODE,
                 RESTORE_OVERLAP_FRACTION_THRESHOLD,
                 USE_INGEST_DB_RESTORE_MODE,
-                INCREMENTAL_RESTORE_ASYNC_COMPACT_AFTER_RESCALE
+                INCREMENTAL_RESTORE_ASYNC_COMPACT_AFTER_RESCALE,
+                COMPACT_FILTER_QUERY_TIME_AFTER_NUM_ENTRIES,
+                COMPACT_FILTER_PERIODIC_COMPACTION_TIME
             };
 
     private static final Set<ConfigOption<?>> POSITIVE_INT_CONFIG_SET =

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBResourceContainer.java

@@ -402,6 +402,11 @@ private ColumnFamilyOptions setColumnFamilyOptionsFromConfigurableOptions(
         currentOptions.setMinWriteBufferNumberToMerge(
                 internalGetOption(RocksDBConfigurableOptions.MIN_WRITE_BUFFER_NUMBER_TO_MERGE));
 
+        currentOptions.setPeriodicCompactionSeconds(
+                internalGetOption(
+                                RocksDBConfigurableOptions.COMPACT_FILTER_PERIODIC_COMPACTION_TIME)
+                        .getSeconds());
+
         TableFormatConfig tableFormatConfig = currentOptions.tableFormatConfig();
 
         BlockBasedTableConfig blockBasedTableConfig;

flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/state/rocksdb/RocksDBStateBackendConfigTest.java

@@ -68,6 +68,7 @@
 
 import java.io.File;
 import java.nio.file.Files;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -624,6 +625,10 @@ public void testConfigurableOptionsFromConfig() throws Exception {
             verifyIllegalArgument(RocksDBConfigurableOptions.BLOOM_FILTER_BLOCK_BASED_MODE, "YES");
             verifyIllegalArgument(
                     RocksDBConfigurableOptions.RESTORE_OVERLAP_FRACTION_THRESHOLD, "2");
+            verifyIllegalArgument(
+                    RocksDBConfigurableOptions.COMPACT_FILTER_PERIODIC_COMPACTION_TIME, "-1s");
+            verifyIllegalArgument(
+                    RocksDBConfigurableOptions.COMPACT_FILTER_QUERY_TIME_AFTER_NUM_ENTRIES, "1.1");
         }
 
         // verify legal configuration
@@ -651,6 +656,8 @@ public void testConfigurableOptionsFromConfig() throws Exception {
             configuration.setString(RocksDBConfigurableOptions.USE_BLOOM_FILTER.key(), "TRUE");
             configuration.setString(
                     RocksDBConfigurableOptions.RESTORE_OVERLAP_FRACTION_THRESHOLD.key(), "0.5");
+            configuration.setString(
+                    RocksDBConfigurableOptions.COMPACT_FILTER_PERIODIC_COMPACTION_TIME.key(), "1h");
 
             try (RocksDBResourceContainer optionsContainer =
                     new RocksDBResourceContainer(
@@ -677,6 +684,7 @@ public void testConfigurableOptionsFromConfig() throws Exception {
                                 CompressionType.SNAPPY_COMPRESSION,
                                 CompressionType.LZ4_COMPRESSION),
                         columnOptions.compressionPerLevel());
+                assertEquals(3600, columnOptions.periodicCompactionSeconds());
 
                 BlockBasedTableConfig tableConfig =
                         (BlockBasedTableConfig) columnOptions.tableFormatConfig();
@@ -948,6 +956,35 @@ public void testConfigureIncrementalRestoreInstanceBufferSize() {
                 notDefault, rocksDBStateBackend.getIncrementalRestoreAsyncCompactAfterRescale());
     }
 
+    @Test
+    public void testConfigurePeriodicCompactionTime() throws Exception {
+        EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true);
+        Configuration configuration = new Configuration();
+        configuration.setString(
+                RocksDBConfigurableOptions.COMPACT_FILTER_PERIODIC_COMPACTION_TIME.key(), "1d");
+        rocksDBStateBackend =
+                rocksDBStateBackend.configure(configuration, getClass().getClassLoader());
+        try (RocksDBResourceContainer resourceContainer =
+                rocksDBStateBackend.createOptionsAndResourceContainer(null)) {
+            assertEquals(Duration.ofDays(1), resourceContainer.getPeriodicCompactionTime());
+        }
+    }
+
+    @Test
+    public void testConfigureQueryTimeAfterNumEntries() throws Exception {
+        EmbeddedRocksDBStateBackend rocksDBStateBackend = new EmbeddedRocksDBStateBackend(true);
+        Configuration configuration = new Configuration();
+        configuration.setString(
+                RocksDBConfigurableOptions.COMPACT_FILTER_QUERY_TIME_AFTER_NUM_ENTRIES.key(),
+                "100");
+        rocksDBStateBackend =
+                rocksDBStateBackend.configure(configuration, getClass().getClassLoader());
+        try (RocksDBResourceContainer resourceContainer =
+                rocksDBStateBackend.createOptionsAndResourceContainer(null)) {
+            assertEquals(100L, resourceContainer.getQueryTimeAfterNumEntries().longValue());
+        }
+    }
+
     private void verifySetParameter(Runnable setter) {
         try {
             setter.run();