[flink] introduce 'sink.writer-refresh-detect-options' to allow more configs to be refreshed

Author: LsomeYeah

docs/layouts/shortcodes/generated/core_configuration.html

@@ -434,12 +434,6 @@
             <td>Boolean</td>
             <td>Optional endInput check partition expire used in case of batch mode or bounded stream.</td>
         </tr>
-        <tr>
-            <td><h5>external-paths.detect-config.enabled</h5></td>
-            <td style="word-wrap: break-word;">false</td>
-            <td>Boolean</td>
-            <td>Whether to auto detected the change of data-file.external-paths when streaming writing.</td>
-        </tr>
         <tr>
             <td><h5>fields.default-aggregate-function</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

docs/layouts/shortcodes/generated/flink_connector_configuration.html

@@ -326,6 +326,12 @@
             <td>MemorySize</td>
             <td>Sink writer memory to control heap memory of writer.</td>
         </tr>
+        <tr>
+            <td><h5>sink.writer-refresh-detect-options</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>the options which are expected to be refreshed when streaming writing, multiple options separated by commas.</td>
+        </tr>
         <tr>
             <td><h5>source.checkpoint-align.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -206,13 +206,6 @@ public InlineElement getDescription() {
                                     + ExternalPathStrategy.SPECIFIC_FS
                                     + ", should be the prefix scheme of the external path, now supported are s3 and oss.");
 
-    public static final ConfigOption<Boolean> AUTO_DETECT_DATA_FILE_EXTERNAL_PATHS =
-            key("external-paths.detect-config.enabled")
-                    .booleanType()
-                    .defaultValue(false)
-                    .withDescription(
-                            "Whether to auto detected the change of data-file.external-paths when streaming writing.");
-
     public static final ConfigOption<Boolean> COMPACTION_FORCE_REWRITE_ALL_FILES =
             key("compaction.force-rewrite-all-files")
                     .booleanType()
@@ -1973,6 +1966,18 @@ public Map<String, String> toMap() {
         return options.toMap();
     }
 
+    public Map<String, String> getSpecificOptions(String... keys) {
+        Map<String, String> result = new HashMap<>();
+        for (String key : keys) {
+            String value = options.get(key);
+            // if value is null, it means the key is not set explicitly
+            if (value != null) {
+                result.put(key, value);
+            }
+        }
+        return result;
+    }
+
     public int bucket() {
         return options.get(BUCKET);
     }
@@ -2682,19 +2687,6 @@ public String dataFileExternalPaths() {
         return options.get(DATA_FILE_EXTERNAL_PATHS);
     }
 
-    public boolean autoDetectDataFileExternalPaths() {
-        return options.get(AUTO_DETECT_DATA_FILE_EXTERNAL_PATHS);
-    }
-
-    public Map<String, String> dataFileExternalPathConfig() {
-        Map<String, String> externalPathsConfig = new HashMap<>();
-        externalPathsConfig.put(DATA_FILE_EXTERNAL_PATHS.key(), dataFileExternalPaths());
-        externalPathsConfig.put(
-                DATA_FILE_EXTERNAL_PATHS_STRATEGY.key(), externalPathStrategy().toString());
-        externalPathsConfig.put(DATA_FILE_EXTERNAL_PATHS_SPECIFIC_FS.key(), externalSpecificFS());
-        return externalPathsConfig;
-    }
-
     public ExternalPathStrategy externalPathStrategy() {
         return options.get(DATA_FILE_EXTERNAL_PATHS_STRATEGY);
     }

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java

@@ -530,6 +530,14 @@ public class FlinkConnectorOptions {
                     .defaultValue(true)
                     .withDescription("Enable pass job level filesystem settings to table file IO.");
 
+    public static final ConfigOption<String> SINK_WRITER_REFRESH_DETECT_OPTIONS =
+            key("sink.writer-refresh-detect-options")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "the options which are expected to be refreshed when streaming writing, "
+                                    + "multiple options separated by commas.");
+
     public static List<ConfigOption<?>> getOptions() {
         final Field[] fields = FlinkConnectorOptions.class.getFields();
         final List<ConfigOption<?>> list = new ArrayList<>(fields.length);

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/AppendTableCompactor.java

@@ -22,16 +22,19 @@
 import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.append.AppendCompactTask;
 import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.flink.FlinkConnectorOptions;
 import org.apache.paimon.flink.metrics.FlinkMetricRegistry;
 import org.apache.paimon.flink.sink.Committable;
 import org.apache.paimon.operation.BaseAppendFileStoreWrite;
 import org.apache.paimon.operation.FileStoreWrite;
 import org.apache.paimon.operation.metrics.CompactionMetrics;
 import org.apache.paimon.operation.metrics.MetricUtils;
+import org.apache.paimon.options.Options;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.sink.CommitMessage;
 import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.utils.StringUtils;
 
 import org.apache.flink.metrics.MetricGroup;
 import org.slf4j.Logger;
@@ -220,7 +223,12 @@ private void replace(FileStoreTable newTable) throws Exception {
     }
 
     protected void updateWriteWithNewSchema() {
-        if (table.coreOptions().autoDetectDataFileExternalPaths()) {
+        // the configs to be refreshed
+        String refreshedConfigs =
+                Options.fromMap(table.options())
+                        .get(FlinkConnectorOptions.SINK_WRITER_REFRESH_DETECT_OPTIONS);
+
+        if (!StringUtils.isNullOrWhitespaceOnly(refreshedConfigs)) {
             Optional<TableSchema> lastestSchema = table.schemaManager().latest();
             if (lastestSchema.isPresent() && lastestSchema.get().id() > table.schema().id()) {
                 LOG.info(
@@ -229,7 +237,9 @@ protected void updateWriteWithNewSchema() {
                         lastestSchema.get().id());
                 try {
                     CoreOptions newCoreOptions = new CoreOptions(lastestSchema.get().options());
-                    table = table.copy(newCoreOptions.dataFileExternalPathConfig());
+                    table =
+                            table.copy(
+                                    newCoreOptions.getSpecificOptions(refreshedConfigs.split(",")));
                     replace(table);
                 } catch (Exception e) {
                     throw new RuntimeException("update write failed.", e);

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PrepareCommitOperator.java

@@ -19,12 +19,14 @@
 package org.apache.paimon.flink.sink;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.flink.FlinkConnectorOptions;
 import org.apache.paimon.flink.memory.FlinkMemorySegmentPool;
 import org.apache.paimon.flink.memory.MemorySegmentAllocator;
 import org.apache.paimon.memory.MemorySegmentPool;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.StringUtils;
 
 import org.apache.flink.runtime.memory.MemoryManager;
 import org.apache.flink.streaming.api.graph.StreamConfig;
@@ -111,7 +113,12 @@ private void emitCommittables(boolean waitCompaction, long checkpointId) throws
 
     protected void updateWriteWithNewSchema(
             FileStoreTable table, StoreSinkWrite write, int taskId) {
-        if (table.coreOptions().autoDetectDataFileExternalPaths()) {
+        // the configs to be refreshed
+        String refreshedConfigs =
+                Options.fromMap(table.options())
+                        .get(FlinkConnectorOptions.SINK_WRITER_REFRESH_DETECT_OPTIONS);
+
+        if (!StringUtils.isNullOrWhitespaceOnly(refreshedConfigs)) {
             Optional<TableSchema> lastestSchema = table.schemaManager().latest();
             if (lastestSchema.isPresent() && lastestSchema.get().id() > table.schema().id()) {
                 LOG.info(
@@ -121,7 +128,9 @@ protected void updateWriteWithNewSchema(
                         lastestSchema.get().id());
                 try {
                     CoreOptions newCoreOptions = new CoreOptions(lastestSchema.get().options());
-                    table = table.copy(newCoreOptions.dataFileExternalPathConfig());
+                    table =
+                            table.copy(
+                                    newCoreOptions.getSpecificOptions(refreshedConfigs.split(",")));
                     write.replace(table);
                 } catch (Exception e) {
                     throw new RuntimeException("update write failed.", e);

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendOnlyTableITCase.java

@@ -275,7 +275,8 @@ public void testTableReadWriteWithChangedExternalPath() throws Exception {
                         + bucket
                         + "',"
                         + bucketKey
-                        + "'external-paths.detect-config.enabled' = 'true',"
+                        + "'sink.writer-refresh-detect-options' = 'data-file.external-paths,"
+                        + "data-file.external-paths.strategy,data-file.external-paths.specific-fs',"
                         + "'data-file.external-paths' = '"
                         + externalPaths1
                         + "',"

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java

@@ -499,7 +499,8 @@ public void testTableReadWriteWithChangedExternalPath() throws Exception {
                         + "'bucket' = '"
                         + bucket
                         + "',"
-                        + "'external-paths.detect-config.enabled' = 'true',"
+                        + "'sink.writer-refresh-detect-options' = 'data-file.external-paths,"
+                        + "data-file.external-paths.strategy,data-file.external-paths.specific-fs',"
                         + "'data-file.external-paths' = '"
                         + externalPaths1
                         + "',"

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java

@@ -300,7 +300,9 @@ public void testStreamingCompactWithChangedExternalPath() throws Exception {
                 CoreOptions.DATA_FILE_EXTERNAL_PATHS.key(),
                 TraceableFileIO.SCHEME + "://" + externalPath1);
         tableOptions.put(CoreOptions.DATA_FILE_EXTERNAL_PATHS_STRATEGY.key(), "round-robin");
-        tableOptions.put(CoreOptions.AUTO_DETECT_DATA_FILE_EXTERNAL_PATHS.key(), "true");
+        tableOptions.put(
+                FlinkConnectorOptions.SINK_WRITER_REFRESH_DETECT_OPTIONS.key(),
+                "data-file.external-paths,data-file.external-paths.strategy,data-file.external-paths.specific-fs");
         tableOptions.put(CoreOptions.WRITE_ONLY.key(), "true");
 
         FileStoreTable table =
@@ -420,7 +422,9 @@ public void testUnawareBucketStreamingCompactWithChangedExternalPath() throws Ex
                 CoreOptions.DATA_FILE_EXTERNAL_PATHS.key(),
                 TraceableFileIO.SCHEME + "://" + externalPath1);
         tableOptions.put(CoreOptions.DATA_FILE_EXTERNAL_PATHS_STRATEGY.key(), "round-robin");
-        tableOptions.put(CoreOptions.AUTO_DETECT_DATA_FILE_EXTERNAL_PATHS.key(), "true");
+        tableOptions.put(
+                FlinkConnectorOptions.SINK_WRITER_REFRESH_DETECT_OPTIONS.key(),
+                "data-file.external-paths,data-file.external-paths.strategy,data-file.external-paths.specific-fs");
 
         FileStoreTable table =
                 prepareTable(