[flink] make tableConf valid for database compaction job (#5287)

LsomeYeah · web-flow · commit caa2730b4c60 · 2025-03-18T17:31:36.000+08:00
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
@@ -215,6 +215,7 @@ private void buildForCombinedMode() {
                                         .get(CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL)
                                         .toMillis())
                         .withPartitionIdleTime(partitionIdleTime);
+        sourceBuilder.withTableOptions(tableOptions.toMap());
 
         Integer parallelism =
                 tableOptions.get(FlinkConnectorOptions.SINK_PARALLELISM) == null
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiAwareBucketTableScan.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiAwareBucketTableScan.java
@@ -51,8 +51,15 @@ public MultiAwareBucketTableScan(
             Pattern includingPattern,
             Pattern excludingPattern,
             Pattern databasePattern,
-            boolean isStreaming) {
-        super(catalogLoader, includingPattern, excludingPattern, databasePattern, isStreaming);
+            boolean isStreaming,
+            Map<String, String> tableOptions) {
+        super(
+                catalogLoader,
+                includingPattern,
+                excludingPattern,
+                databasePattern,
+                isStreaming,
+                tableOptions);
         tablesMap = new HashMap<>();
         scansMap = new HashMap<>();
     }
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiTableScanBase.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiTableScanBase.java
@@ -32,6 +32,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
+import java.util.Map;
 import java.util.regex.Pattern;
 
 import static org.apache.paimon.flink.utils.MultiTablesCompactorUtil.shouldCompactTable;
@@ -59,18 +60,22 @@ public abstract class MultiTableScanBase<T> implements AutoCloseable {
 
     protected boolean isStreaming;
 
+    protected final Map<String, String> tableOptions;
+
     public MultiTableScanBase(
             CatalogLoader catalogLoader,
             Pattern includingPattern,
             Pattern excludingPattern,
             Pattern databasePattern,
-            boolean isStreaming) {
+            boolean isStreaming,
+            Map<String, String> tableOptions) {
         catalog = catalogLoader.load();
 
         this.includingPattern = includingPattern;
         this.excludingPattern = excludingPattern;
         this.databasePattern = databasePattern;
         this.isStreaming = isStreaming;
+        this.tableOptions = tableOptions;
     }
 
     protected void updateTableMap()
@@ -93,7 +98,7 @@ protected void updateTableMap()
                             continue;
                         }
 
-                        FileStoreTable fileStoreTable = (FileStoreTable) table;
+                        FileStoreTable fileStoreTable = ((FileStoreTable) table).copy(tableOptions);
                         addScanTable(fileStoreTable, identifier);
                     }
                 }
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiUnawareBucketTableScan.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiUnawareBucketTableScan.java
@@ -45,8 +45,15 @@ public MultiUnawareBucketTableScan(
             Pattern includingPattern,
             Pattern excludingPattern,
             Pattern databasePattern,
-            boolean isStreaming) {
-        super(catalogLoader, includingPattern, excludingPattern, databasePattern, isStreaming);
+            boolean isStreaming,
+            Map<String, String> tableOptions) {
+        super(
+                catalogLoader,
+                includingPattern,
+                excludingPattern,
+                databasePattern,
+                isStreaming,
+                tableOptions);
         tablesMap = new HashMap<>();
     }
 
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlyMultiTableCompactionWorkerOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlyMultiTableCompactionWorkerOperator.java
@@ -118,7 +118,7 @@ public void processElement(StreamRecord<MultiTableUnawareAppendCompactionTask> e
     private UnawareBucketCompactor compactor(Identifier tableId) {
         try {
             return new UnawareBucketCompactor(
-                    (FileStoreTable) catalog.getTable(tableId),
+                    (FileStoreTable) catalog.getTable(tableId).copy(options.toMap()),
                     commitUser,
                     this::workerExecutor,
                     getMetricGroup());
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesStoreCompactOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesStoreCompactOperator.java
@@ -226,6 +226,7 @@ private FileStoreTable getTable(Identifier tableId) throws InterruptedException
             while (true) {
                 try {
                     table = (FileStoreTable) catalog.getTable(tableId);
+                    table = table.copy(options.toMap());
                     HashMap<String, String> dynamicOptions =
                             new HashMap<String, String>() {
                                 {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PrepareCommitOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PrepareCommitOperator.java
@@ -53,7 +53,7 @@ public abstract class PrepareCommitOperator<IN, OUT> extends AbstractStreamOpera
 
     @Nullable protected transient MemorySegmentPool memoryPool;
     @Nullable private transient MemorySegmentAllocator memoryAllocator;
-    private final Options options;
+    protected final Options options;
     private boolean endOfInput = false;
 
     public PrepareCommitOperator(StreamOperatorParameters<OUT> parameters, Options options) {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CombinedTableCompactorSourceBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CombinedTableCompactorSourceBuilder.java
@@ -37,6 +37,8 @@
 import javax.annotation.Nullable;
 
 import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.regex.Pattern;
 
 /**
@@ -54,6 +56,7 @@ public class CombinedTableCompactorSourceBuilder {
     private boolean isContinuous = false;
     private StreamExecutionEnvironment env;
     @Nullable private Duration partitionIdleTime = null;
+    private Map<String, String> tableOptions = new HashMap<>();
 
     public CombinedTableCompactorSourceBuilder(
             CatalogLoader catalogLoader,
@@ -84,6 +87,11 @@ public CombinedTableCompactorSourceBuilder withPartitionIdleTime(
         return this;
     }
 
+    public CombinedTableCompactorSourceBuilder withTableOptions(Map<String, String> tableOptions) {
+        this.tableOptions = tableOptions;
+        return this;
+    }
+
     public DataStream<RowData> buildAwareBucketTableSource() {
         Preconditions.checkArgument(env != null, "StreamExecutionEnvironment should not be null.");
         RowType produceType = CompactBucketsTable.getRowType();
@@ -96,6 +104,7 @@ public DataStream<RowData> buildAwareBucketTableSource() {
                     includingPattern,
                     excludingPattern,
                     databasePattern,
+                    tableOptions,
                     monitorInterval);
         } else {
             return CombinedAwareBatchSource.buildSource(
@@ -106,6 +115,7 @@ public DataStream<RowData> buildAwareBucketTableSource() {
                     includingPattern,
                     excludingPattern,
                     databasePattern,
+                    tableOptions,
                     partitionIdleTime);
         }
     }
@@ -120,6 +130,7 @@ public DataStream<MultiTableUnawareAppendCompactionTask> buildForUnawareBucketsT
                     includingPattern,
                     excludingPattern,
                     databasePattern,
+                    tableOptions,
                     monitorInterval);
         } else {
             return CombinedUnawareBatchSource.buildSource(
@@ -129,6 +140,7 @@ public DataStream<MultiTableUnawareAppendCompactionTask> buildForUnawareBucketsT
                     includingPattern,
                     excludingPattern,
                     databasePattern,
+                    tableOptions,
                     partitionIdleTime);
         }
     }
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAwareBatchSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAwareBatchSource.java
@@ -43,6 +43,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.time.Duration;
+import java.util.Map;
 import java.util.regex.Pattern;
 
 import static org.apache.paimon.flink.compact.MultiTableScanBase.ScanResult.FINISHED;
@@ -52,13 +53,16 @@
 public class CombinedAwareBatchSource extends CombinedCompactorSource<Tuple2<Split, String>> {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(CombinedAwareBatchSource.class);
+    private final Map<String, String> tableOptions;
 
     public CombinedAwareBatchSource(
             CatalogLoader catalogLoader,
             Pattern includingPattern,
             Pattern excludingPattern,
-            Pattern databasePattern) {
+            Pattern databasePattern,
+            Map<String, String> tableOptions) {
         super(catalogLoader, includingPattern, excludingPattern, databasePattern, false);
+        this.tableOptions = tableOptions;
     }
 
     @Override
@@ -79,7 +83,8 @@ public void start() {
                             includingPattern,
                             excludingPattern,
                             databasePattern,
-                            isStreaming);
+                            isStreaming,
+                            tableOptions);
         }
 
         @Override
@@ -116,10 +121,15 @@ public static DataStream<RowData> buildSource(
             Pattern includingPattern,
             Pattern excludingPattern,
             Pattern databasePattern,
+            Map<String, String> tableOptions,
             Duration partitionIdleTime) {
         CombinedAwareBatchSource source =
                 new CombinedAwareBatchSource(
-                        catalogLoader, includingPattern, excludingPattern, databasePattern);
+                        catalogLoader,
+                        includingPattern,
+                        excludingPattern,
+                        databasePattern,
+                        tableOptions);
         TupleTypeInfo<Tuple2<Split, String>> tupleTypeInfo =
                 new TupleTypeInfo<>(
                         new JavaTypeInfo<>(Split.class), BasicTypeInfo.STRING_TYPE_INFO);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAwareStreamingSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedAwareStreamingSource.java
@@ -40,6 +40,7 @@
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.data.RowData;
 
+import java.util.Map;
 import java.util.regex.Pattern;
 
 import static org.apache.paimon.flink.compact.MultiTableScanBase.ScanResult.FINISHED;
@@ -49,14 +50,17 @@
 public class CombinedAwareStreamingSource extends CombinedCompactorSource<Tuple2<Split, String>> {
 
     private final long monitorInterval;
+    private final Map<String, String> tableOptions;
 
     public CombinedAwareStreamingSource(
             CatalogLoader catalogLoader,
             Pattern includingPattern,
             Pattern excludingPattern,
             Pattern databasePattern,
+            Map<String, String> tableOptions,
             long monitorInterval) {
         super(catalogLoader, includingPattern, excludingPattern, databasePattern, true);
+        this.tableOptions = tableOptions;
         this.monitorInterval = monitorInterval;
     }
 
@@ -78,7 +82,8 @@ public void start() {
                             includingPattern,
                             excludingPattern,
                             databasePattern,
-                            isStreaming);
+                            isStreaming,
+                            tableOptions);
         }
 
         @Override
@@ -111,6 +116,7 @@ public static DataStream<RowData> buildSource(
             Pattern includingPattern,
             Pattern excludingPattern,
             Pattern databasePattern,
+            Map<String, String> tableOptions,
             long monitorInterval) {
 
         CombinedAwareStreamingSource source =
@@ -119,6 +125,7 @@ public static DataStream<RowData> buildSource(
                         includingPattern,
                         excludingPattern,
                         databasePattern,
+                        tableOptions,
                         monitorInterval);
         TupleTypeInfo<Tuple2<Split, String>> tupleTypeInfo =
                 new TupleTypeInfo<>(
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareBatchSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareBatchSource.java
@@ -62,13 +62,16 @@ public class CombinedUnawareBatchSource
         extends CombinedCompactorSource<MultiTableUnawareAppendCompactionTask> {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(CombinedUnawareBatchSource.class);
+    private final Map<String, String> tableOptions;
 
     public CombinedUnawareBatchSource(
             CatalogLoader catalogLoader,
             Pattern includingPattern,
             Pattern excludingPattern,
-            Pattern databasePattern) {
+            Pattern databasePattern,
+            Map<String, String> tableOptions) {
         super(catalogLoader, includingPattern, excludingPattern, databasePattern, false);
+        this.tableOptions = tableOptions;
     }
 
     @Override
@@ -90,7 +93,8 @@ public void start() {
                             includingPattern,
                             excludingPattern,
                             databasePattern,
-                            isStreaming);
+                            isStreaming,
+                            tableOptions);
         }
 
         @Override
@@ -126,10 +130,15 @@ public static DataStream<MultiTableUnawareAppendCompactionTask> buildSource(
             Pattern includingPattern,
             Pattern excludingPattern,
             Pattern databasePattern,
+            Map<String, String> tableOptions,
             @Nullable Duration partitionIdleTime) {
         CombinedUnawareBatchSource combinedUnawareBatchSource =
                 new CombinedUnawareBatchSource(
-                        catalogLoader, includingPattern, excludingPattern, databasePattern);
+                        catalogLoader,
+                        includingPattern,
+                        excludingPattern,
+                        databasePattern,
+                        tableOptions);
         MultiTableCompactionTaskTypeInfo compactionTaskTypeInfo =
                 new MultiTableCompactionTaskTypeInfo();
 
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareStreamingSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/CombinedUnawareStreamingSource.java
@@ -34,6 +34,7 @@
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
+import java.util.Map;
 import java.util.regex.Pattern;
 
 import static org.apache.paimon.flink.compact.MultiTableScanBase.ScanResult.FINISHED;
@@ -46,14 +47,17 @@ public class CombinedUnawareStreamingSource
         extends CombinedCompactorSource<MultiTableUnawareAppendCompactionTask> {
 
     private final long monitorInterval;
+    private final Map<String, String> tableOptions;
 
     public CombinedUnawareStreamingSource(
             CatalogLoader catalogLoader,
             Pattern includingPattern,
             Pattern excludingPattern,
             Pattern databasePattern,
+            Map<String, String> tableOptions,
             long monitorInterval) {
         super(catalogLoader, includingPattern, excludingPattern, databasePattern, true);
+        this.tableOptions = tableOptions;
         this.monitorInterval = monitorInterval;
     }
 
@@ -76,7 +80,8 @@ public void start() {
                             includingPattern,
                             excludingPattern,
                             databasePattern,
-                            isStreaming);
+                            isStreaming,
+                            tableOptions);
         }
 
         @Override
@@ -108,6 +113,7 @@ public static DataStream<MultiTableUnawareAppendCompactionTask> buildSource(
             Pattern includingPattern,
             Pattern excludingPattern,
             Pattern databasePattern,
+            Map<String, String> tableOptions,
             long monitorInterval) {
 
         CombinedUnawareStreamingSource source =
@@ -116,6 +122,7 @@ public static DataStream<MultiTableUnawareAppendCompactionTask> buildSource(
                         includingPattern,
                         excludingPattern,
                         databasePattern,
+                        tableOptions,
                         monitorInterval);
         MultiTableCompactionTaskTypeInfo compactionTaskTypeInfo =
                 new MultiTableCompactionTaskTypeInfo();
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java

Original file line number	Diff line number	Diff line change
`@@ -226,6 +226,7 @@ private FileStoreTable getTable(Identifier tableId) throws InterruptedException`
`226`	`226`	`while (true) {`
`227`	`227`	`try {`
`228`	`228`	`table = (FileStoreTable) catalog.getTable(tableId);`
	`229`	`+ table = table.copy(options.toMap());`
`229`	`230`	`HashMap<String, String> dynamicOptions =`
`230`	`231`	`new HashMap<String, String>() {`
`231`	`232`	`{`