This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new a4b4027a1d [flink] Clone Paimon action supports add additional table
options (#9042)
a4b4027a1d is described below
commit a4b4027a1d03a38bbebb6fb30294c46b65e45322
Author: yuzelin <[email protected]>
AuthorDate: Wed Aug 5 17:11:24 2026 +0800
[flink] Clone Paimon action supports add additional table options (#9042)
---
.../apache/paimon/flink/action/CloneAction.java | 9 +++++++++
.../paimon/flink/action/CloneActionFactory.java | 7 +++++++
.../paimon/flink/clone/ClonePaimonTableUtils.java | 2 ++
.../clone/schema/ClonePaimonSchemaFunction.java | 23 ++++++++++++----------
.../paimon/flink/procedure/CloneProcedure.java | 10 +++++++++-
.../paimon/hive/procedure/CloneActionITCase.java | 14 ++++++++++++-
6 files changed, 53 insertions(+), 12 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java
index e3ed1e01d4..4ee87710e6 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneAction.java
@@ -39,6 +39,7 @@ public class CloneAction extends ActionBase {
private final String sourceTableName;
private final Map<String, String> targetCatalogConfig;
+ private final Map<String, String> targetTableConfig;
private final String targetDatabase;
private final String targetTableName;
@@ -58,6 +59,7 @@ public class CloneAction extends ActionBase {
String targetDatabase,
String targetTableName,
Map<String, String> targetCatalogConfig,
+ Map<String, String> targetTableConfig,
@Nullable Integer parallelism,
@Nullable String whereSql,
@Nullable List<String> includedTables,
@@ -69,6 +71,11 @@ public class CloneAction extends ActionBase {
super(sourceCatalogConfig);
if (cloneFrom.equalsIgnoreCase("hive")) {
+ if (!targetTableConfig.isEmpty()) {
+ throw new UnsupportedOperationException(
+ "Parameter 'target_table_conf' is only supported when
clone_from is paimon.");
+ }
+
Catalog sourceCatalog = catalog;
if (sourceCatalog instanceof CachingCatalog) {
sourceCatalog = ((CachingCatalog) sourceCatalog).wrapped();
@@ -87,6 +94,7 @@ public class CloneAction extends ActionBase {
this.targetDatabase = targetDatabase;
this.targetTableName = targetTableName;
this.targetCatalogConfig = targetCatalogConfig;
+ this.targetTableConfig = targetTableConfig;
this.parallelism = parallelism == null ? env.getParallelism() :
parallelism;
this.whereSql = whereSql;
@@ -133,6 +141,7 @@ public class CloneAction extends ActionBase {
targetDatabase,
targetTableName,
targetCatalogConfig,
+ targetTableConfig,
parallelism,
whereSql,
includedTables,
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneActionFactory.java
index db0680cf3e..6fc83a8dcc 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneActionFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CloneActionFactory.java
@@ -34,6 +34,7 @@ public class CloneActionFactory implements ActionFactory {
private static final String TARGET_DATABASE = "target_database";
private static final String TARGET_TABLE = "target_table";
private static final String TARGET_CATALOG_CONF = "target_catalog_conf";
+ private static final String TARGET_TABLE_CONF = "target_table_conf";
private static final String PARALLELISM = "parallelism";
private static final String WHERE = "where";
private static final String INCLUDED_TABLES = "included_tables";
@@ -54,6 +55,8 @@ public class CloneActionFactory implements ActionFactory {
Map<String, String> targetCatalogConfig =
new HashMap<>(optionalConfigMap(params, TARGET_CATALOG_CONF));
+ Map<String, String> targetTableConfig =
+ new HashMap<>(optionalConfigMap(params, TARGET_TABLE_CONF));
String targetWarehouse = params.get(TARGET_WAREHOUSE);
if (targetWarehouse != null &&
!targetCatalogConfig.containsKey(WAREHOUSE)) {
targetCatalogConfig.put(WAREHOUSE, targetWarehouse);
@@ -97,6 +100,7 @@ public class CloneActionFactory implements ActionFactory {
params.get(TARGET_DATABASE),
params.get(TARGET_TABLE),
targetCatalogConfig,
+ targetTableConfig,
parallelism == null ? null :
Integer.parseInt(parallelism),
params.get(WHERE),
includedTables,
@@ -114,5 +118,8 @@ public class CloneActionFactory implements ActionFactory {
System.out.println(
"Action \"clone\" clones the source files and migrate them to
paimon table.");
System.out.println();
+ System.out.println(
+ "Use repeated '--target_table_conf <key=value>' options to
override options "
+ + "when creating target tables.");
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/ClonePaimonTableUtils.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/ClonePaimonTableUtils.java
index 9cf1473b0f..1286a49132 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/ClonePaimonTableUtils.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/ClonePaimonTableUtils.java
@@ -134,6 +134,7 @@ public class ClonePaimonTableUtils {
String targetDatabase,
String targetTableName,
Map<String, String> targetCatalogConfig,
+ Map<String, String> targetTableConfig,
int parallelism,
@Nullable String whereSql,
@Nullable List<String> includedTables,
@@ -165,6 +166,7 @@ public class ClonePaimonTableUtils {
new ClonePaimonSchemaFunction(
sourceCatalogConfig,
targetCatalogConfig,
+ targetTableConfig,
preferFileFormat,
cloneIfExists))
.name("Clone Schema")
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/schema/ClonePaimonSchemaFunction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/schema/ClonePaimonSchemaFunction.java
index 5ad46beb06..7c1b190e07 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/schema/ClonePaimonSchemaFunction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/schema/ClonePaimonSchemaFunction.java
@@ -34,6 +34,7 @@ import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.HashMap;
import java.util.Map;
import static org.apache.paimon.CoreOptions.BUCKET;
@@ -50,6 +51,7 @@ public class ClonePaimonSchemaFunction
private final Map<String, String> sourceCatalogConfig;
private final Map<String, String> targetCatalogConfig;
+ private final Map<String, String> targetTableConfig;
private final String preferFileFormat;
private final boolean cloneIfExists;
@@ -59,10 +61,12 @@ public class ClonePaimonSchemaFunction
public ClonePaimonSchemaFunction(
Map<String, String> sourceCatalogConfig,
Map<String, String> targetCatalogConfig,
+ Map<String, String> targetTableConfig,
String preferFileFormat,
boolean cloneIfExists) {
this.sourceCatalogConfig = sourceCatalogConfig;
this.targetCatalogConfig = targetCatalogConfig;
+ this.targetTableConfig = targetTableConfig;
this.preferFileFormat = preferFileFormat;
this.cloneIfExists = cloneIfExists;
}
@@ -101,16 +105,15 @@ public class ClonePaimonSchemaFunction
f -> builder.column(f.name(), f.type(),
f.description(), f.defaultValue()));
builder.partitionKeys(sourceTable.partitionKeys());
builder.primaryKey(sourceTable.primaryKeys());
- sourceTable
- .options()
- .forEach(
- (k, v) -> {
- if (k.equalsIgnoreCase(BUCKET.key())
- || k.equalsIgnoreCase(PATH.key())) {
- return;
- }
- builder.option(k, v);
- });
+ Map<String, String> tableOptions = new
HashMap<>(sourceTable.options());
+ tableOptions.putAll(targetTableConfig);
+ tableOptions.forEach(
+ (k, v) -> {
+ if (k.equalsIgnoreCase(BUCKET.key()) ||
k.equalsIgnoreCase(PATH.key())) {
+ return;
+ }
+ builder.option(k, v);
+ });
if (sourceTable.primaryKeys().isEmpty()) {
// for append table with bucket
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CloneProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CloneProcedure.java
index 74a70be4fa..01538d6df5 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CloneProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CloneProcedure.java
@@ -81,6 +81,10 @@ public class CloneProcedure extends ProcedureBase {
@ArgumentHint(
name = "clone_if_exists",
type = @DataTypeHint("BOOLEAN"),
+ isOptional = true),
+ @ArgumentHint(
+ name = "target_table_conf",
+ type = @DataTypeHint("STRING"),
isOptional = true)
})
public String[] call(
@@ -98,13 +102,16 @@ public class CloneProcedure extends ProcedureBase {
String preferFileFormat,
String cloneFrom,
Boolean metaOnly,
- Boolean cloneIfExists)
+ Boolean cloneIfExists,
+ String targetTableConfigStr)
throws Exception {
Map<String, String> sourceCatalogConfig =
new HashMap<>(optionalConfigMap(sourceCatalogConfigStr));
Map<String, String> targetCatalogConfig =
new HashMap<>(optionalConfigMap(targetCatalogConfigStr));
+ Map<String, String> targetTableConfig =
+ new HashMap<>(optionalConfigMap(targetTableConfigStr));
List<String> includedTables =
StringUtils.isNullOrWhitespaceOnly(includedTablesStr)
@@ -123,6 +130,7 @@ public class CloneProcedure extends ProcedureBase {
targetDatabase,
targetTableName,
targetCatalogConfig,
+ targetTableConfig,
parallelism,
where,
includedTables,
diff --git
a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/CloneActionITCase.java
b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/CloneActionITCase.java
index 92ca7804bc..817ca5eda1 100644
---
a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/CloneActionITCase.java
+++
b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/CloneActionITCase.java
@@ -107,6 +107,12 @@ public class CloneActionITCase extends ActionITCaseBase {
"target",
"--target_catalog_conf",
"warehouse=" + warehouse2,
+ "--target_table_conf",
+ "deletion-vectors.enabled=true",
+ "--target_table_conf",
+ "file.format.per.level=0:avro",
+ "--target_table_conf",
+ "metadata.stats-mode.per.level=0:none",
"--clone_from",
"paimon")
.run();
@@ -115,7 +121,13 @@ public class CloneActionITCase extends ActionITCaseBase {
List<Row> result = sql(tEnv, "SELECT * FROM
catalog2.`default`.target");
assertThat(result).containsExactlyInAnyOrder(Row.of(1, 1), Row.of(2,
2));
List<Row> show = sql(tEnv, "SHOW CREATE TABLE
catalog2.`default`.target");
- assertThat(show.toString()).contains("PRIMARY KEY");
+ assertThat(show.toString())
+ .contains(
+ "PRIMARY KEY",
+ "'bucket' = '-2'",
+ "'deletion-vectors.enabled' = 'true'",
+ "'file.format.per.level' = '0:avro'",
+ "'metadata.stats-mode.per.level' = '0:none'");
}
@Test
