[flink] Fix that sink.parallelism doesn't work in compact_database ac…

…tion (apache#2830)
lmagic233 · Jan 31, 2024 · 28eebed · 28eebed
1 parent 0bee22e
commit 28eebed
Showing 1 changed file with 4 additions and 5 deletions.
diff --git a/...imon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java b/...imon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
@@ -46,7 +46,6 @@
 
 import javax.annotation.Nullable;
 
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -142,11 +141,11 @@ private void buildForDividedMode() {
                                                 table.getClass().getName()));
                                 continue;
                             }
+                            Map<String, String> dynamicOptions =
+                                    new HashMap<>(tableOptions.toMap());
+                            dynamicOptions.put(CoreOptions.WRITE_ONLY.key(), "false");
                             FileStoreTable fileStoreTable =
-                                    (FileStoreTable)
-                                            table.copy(
-                                                    Collections.singletonMap(
-                                                            CoreOptions.WRITE_ONLY.key(), "false"));
+                                    (FileStoreTable) table.copy(dynamicOptions);
                             tableMap.put(fullTableName, fileStoreTable);
                         } else {
                             LOG.debug("The table {} is excluded.", fullTableName);