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

Author: masteryhx

flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStConfigurableOptions.java

@@ -397,7 +397,9 @@ public class ForStConfigurableOptions implements Serializable {
                 BLOOM_FILTER_BLOCK_BASED_MODE,
                 RESTORE_OVERLAP_FRACTION_THRESHOLD,
                 USE_INGEST_DB_RESTORE_MODE,
-                USE_DELETE_FILES_IN_RANGE_DURING_RESCALING
+                USE_DELETE_FILES_IN_RANGE_DURING_RESCALING,
+                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-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java

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

flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java

@@ -65,6 +65,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -496,6 +497,10 @@ public void testConfigurableOptionsFromConfig() throws Exception {
             verifyIllegalArgument(ForStConfigurableOptions.COMPRESSION_PER_LEVEL, "SNAP");
             verifyIllegalArgument(ForStConfigurableOptions.USE_BLOOM_FILTER, "NO");
             verifyIllegalArgument(ForStConfigurableOptions.BLOOM_FILTER_BLOCK_BASED_MODE, "YES");
+            verifyIllegalArgument(
+                    ForStConfigurableOptions.COMPACT_FILTER_PERIODIC_COMPACTION_TIME, "-1s");
+            verifyIllegalArgument(
+                    ForStConfigurableOptions.COMPACT_FILTER_QUERY_TIME_AFTER_NUM_ENTRIES, "1.1");
         }
 
         // verify legal configuration
@@ -520,6 +525,8 @@ public void testConfigurableOptionsFromConfig() throws Exception {
             configuration.setString(ForStConfigurableOptions.METADATA_BLOCK_SIZE.key(), "8 kb");
             configuration.setString(ForStConfigurableOptions.BLOCK_CACHE_SIZE.key(), "512 mb");
             configuration.setString(ForStConfigurableOptions.USE_BLOOM_FILTER.key(), "TRUE");
+            configuration.setString(
+                    ForStConfigurableOptions.COMPACT_FILTER_PERIODIC_COMPACTION_TIME.key(), "1h");
 
             try (ForStResourceContainer optionsContainer =
                     new ForStResourceContainer(
@@ -546,6 +553,7 @@ public void testConfigurableOptionsFromConfig() throws Exception {
                                 CompressionType.SNAPPY_COMPRESSION,
                                 CompressionType.LZ4_COMPRESSION),
                         columnOptions.compressionPerLevel());
+                assertEquals(3600, columnOptions.periodicCompactionSeconds());
 
                 BlockBasedTableConfig tableConfig =
                         (BlockBasedTableConfig) columnOptions.tableFormatConfig();
@@ -779,6 +787,32 @@ public void testSupportSavepoint() {
         assertTrue(forStStateBackend.supportsSavepointFormat(SavepointFormatType.NATIVE));
     }
 
+    @Test
+    public void testConfigurePeriodicCompactionTime() throws Exception {
+        ForStStateBackend forStStateBackend = new ForStStateBackend();
+        Configuration configuration = new Configuration();
+        configuration.setString(
+                ForStConfigurableOptions.COMPACT_FILTER_PERIODIC_COMPACTION_TIME.key(), "1d");
+        forStStateBackend = forStStateBackend.configure(configuration, getClass().getClassLoader());
+        try (ForStResourceContainer resourceContainer =
+                forStStateBackend.createOptionsAndResourceContainer(null)) {
+            assertEquals(Duration.ofDays(1), resourceContainer.getPeriodicCompactionTime());
+        }
+    }
+
+    @Test
+    public void testConfigureQueryTimeAfterNumEntries() throws Exception {
+        ForStStateBackend forStStateBackend = new ForStStateBackend();
+        Configuration configuration = new Configuration();
+        configuration.setString(
+                ForStConfigurableOptions.COMPACT_FILTER_QUERY_TIME_AFTER_NUM_ENTRIES.key(), "100");
+        forStStateBackend = forStStateBackend.configure(configuration, getClass().getClassLoader());
+        try (ForStResourceContainer resourceContainer =
+                forStStateBackend.createOptionsAndResourceContainer(null)) {
+            assertEquals(100L, resourceContainer.getQueryTimeAfterNumEntries().longValue());
+        }
+    }
+
     private void verifySetParameter(Runnable setter) {
         try {
             setter.run();