This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 983a55205 [hive] HiveMigrator support set parallelism for procedures
(#4177)
983a55205 is described below
commit 983a55205cf33a69d4b2770024266a698eb914bd
Author: xuzifu666 <[email protected]>
AuthorDate: Fri Sep 13 17:57:10 2024 +0800
[hive] HiveMigrator support set parallelism for procedures (#4177)
---
docs/content/spark/procedures.md | 8 +++--
.../flink/procedure/MigrateDatabaseProcedure.java | 25 ++++++++++++++++
.../flink/procedure/MigrateFileProcedure.java | 31 +++++++++++++++++--
.../flink/procedure/MigrateTableProcedure.java | 29 ++++++++++++++++++
.../paimon/flink/action/MigrateDatabaseAction.java | 11 +++++--
.../flink/action/MigrateDatabaseActionFactory.java | 9 +++++-
.../paimon/flink/action/MigrateFileAction.java | 8 +++--
.../flink/action/MigrateFileActionFactory.java | 5 +++-
.../paimon/flink/action/MigrateTableAction.java | 11 +++++--
.../flink/action/MigrateTableActionFactory.java | 9 +++++-
.../flink/procedure/MigrateDatabaseProcedure.java | 11 +++++--
.../flink/procedure/MigrateFileProcedure.java | 14 +++++++--
.../flink/procedure/MigrateTableProcedure.java | 12 ++++++--
.../paimon/flink/utils/TableMigrationUtils.java | 10 +++++--
.../apache/paimon/hive/migrate/HiveMigrator.java | 16 ++++++----
.../procedure/MigrateDatabaseProcedureITCase.java | 5 ++--
.../hive/procedure/MigrateFileProcedureITCase.java | 3 +-
.../procedure/MigrateTableProcedureITCase.java | 3 +-
.../spark/procedure/MigrateFileProcedure.java | 7 ++++-
.../spark/procedure/MigrateTableProcedure.java | 7 ++++-
.../paimon/spark/utils/TableMigrationUtils.java | 2 ++
.../spark/procedure/MigrateFileProcedureTest.scala | 35 ++++++++++++++++++++++
.../procedure/MigrateTableProcedureTest.scala | 23 ++++++++++++++
23 files changed, 259 insertions(+), 35 deletions(-)
diff --git a/docs/content/spark/procedures.md b/docs/content/spark/procedures.md
index 085e5866f..2fd1bb620 100644
--- a/docs/content/spark/procedures.md
+++ b/docs/content/spark/procedures.md
@@ -139,9 +139,10 @@ This section introduce all available spark procedures
about paimon.
<li>options: the table options of the paimon table to migrate.</li>
<li>target_table: name of the target paimon table to migrate. If
not set would keep the same name with origin table</li>
<li>delete_origin: If had set target_table, can set delete_origin
to decide whether delete the origin table metadata from hms after migrate.
Default is true</li>
- <li>options_map: Options map for adding key-value options which is
a map.</li>
+ <li>options_map: Options map for adding key-value options which is
a map.</li>
+ <li>parallelism: the parallelism for migrate process, default is
core numbers of machine.</li>
</td>
- <td>CALL sys.migrate_table(source_type => 'hive', table => 'default.T',
options => 'file.format=parquet', options_map => map('k1','v1'))</td>
+ <td>CALL sys.migrate_table(source_type => 'hive', table => 'default.T',
options => 'file.format=parquet', options_map => map('k1','v1'), parallelism =>
6)</td>
</tr>
<tr>
<td>migrate_file</td>
@@ -151,8 +152,9 @@ This section introduce all available spark procedures about
paimon.
<li>source_table: name of the origin table to migrate. Cannot be
empty.</li>
<li>target_table: name of the target table to be migrated. Cannot
be empty.</li>
<li>delete_origin: If had set target_table, can set delete_origin
to decide whether delete the origin table metadata from hms after migrate.
Default is true</li>
+ <li>parallelism: the parallelism for migrate process, default is
core numbers of machine.</li>
</td>
- <td>CALL sys.migrate_file(source_type => 'hive', table => 'default.T',
delete_origin => true)</td>
+ <td>CALL sys.migrate_file(source_type => 'hive', table => 'default.T',
delete_origin => true, parallelism => 6)</td>
</tr>
<tr>
<td>remove_orphan_files</td>
diff --git
a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
index 128875a8b..b398b3258 100644
---
a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
+++
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
@@ -51,6 +51,31 @@ public class MigrateDatabaseProcedure extends ProcedureBase {
connector,
catalog,
sourceDatabasePath,
+ Runtime.getRuntime().availableProcessors(),
+
ParameterUtils.parseCommaSeparatedKeyValues(properties));
+
+ for (Migrator migrator : migrators) {
+ migrator.executeMigrate();
+ migrator.renameTable(false);
+ }
+
+ return new String[] {"Success"};
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String connector,
+ String sourceDatabasePath,
+ String properties,
+ Integer parallelism)
+ throws Exception {
+ Integer p = parallelism == null ?
Runtime.getRuntime().availableProcessors() : parallelism;
+ List<Migrator> migrators =
+ TableMigrationUtils.getImporters(
+ connector,
+ catalog,
+ sourceDatabasePath,
+ p,
ParameterUtils.parseCommaSeparatedKeyValues(properties));
for (Migrator migrator : migrators) {
diff --git
a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateFileProcedure.java
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateFileProcedure.java
index 110b4e25f..5d68cc0f5 100644
---
a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateFileProcedure.java
+++
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateFileProcedure.java
@@ -40,7 +40,12 @@ public class MigrateFileProcedure extends ProcedureBase {
String sourceTablePath,
String targetPaimonTablePath)
throws Exception {
- call(procedureContext, connector, sourceTablePath,
targetPaimonTablePath, true);
+ migrateHandle(
+ connector,
+ sourceTablePath,
+ targetPaimonTablePath,
+ true,
+ Runtime.getRuntime().availableProcessors());
return new String[] {"Success"};
}
@@ -51,7 +56,25 @@ public class MigrateFileProcedure extends ProcedureBase {
String targetPaimonTablePath,
boolean deleteOrigin)
throws Exception {
- migrateHandle(connector, sourceTablePath, targetPaimonTablePath,
deleteOrigin);
+ migrateHandle(
+ connector,
+ sourceTablePath,
+ targetPaimonTablePath,
+ deleteOrigin,
+ Runtime.getRuntime().availableProcessors());
+ return new String[] {"Success"};
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String connector,
+ String sourceTablePath,
+ String targetPaimonTablePath,
+ boolean deleteOrigin,
+ Integer parallelism)
+ throws Exception {
+ Integer p = parallelism == null ?
Runtime.getRuntime().availableProcessors() : parallelism;
+ migrateHandle(connector, sourceTablePath, targetPaimonTablePath,
deleteOrigin, p);
return new String[] {"Success"};
}
@@ -59,7 +82,8 @@ public class MigrateFileProcedure extends ProcedureBase {
String connector,
String sourceTablePath,
String targetPaimonTablePath,
- boolean deleteOrigin)
+ boolean deleteOrigin,
+ Integer parallelism)
throws Exception {
Identifier sourceTableId = Identifier.fromString(sourceTablePath);
Identifier targetTableId =
Identifier.fromString(targetPaimonTablePath);
@@ -77,6 +101,7 @@ public class MigrateFileProcedure extends ProcedureBase {
sourceTableId.getObjectName(),
targetTableId.getDatabaseName(),
targetTableId.getObjectName(),
+ parallelism,
Collections.emptyMap());
importer.deleteOriginTable(deleteOrigin);
importer.executeMigrate();
diff --git
a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateTableProcedure.java
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateTableProcedure.java
index 39e6092d8..196528d31 100644
---
a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateTableProcedure.java
+++
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateTableProcedure.java
@@ -62,6 +62,35 @@ public class MigrateTableProcedure extends ProcedureBase {
sourceTableId.getObjectName(),
targetTableId.getDatabaseName(),
targetTableId.getObjectName(),
+ Runtime.getRuntime().availableProcessors(),
+
ParameterUtils.parseCommaSeparatedKeyValues(properties))
+ .executeMigrate();
+
+ LOG.info("Last step: rename " + targetTableId + " to " +
sourceTableId);
+ catalog.renameTable(targetTableId, sourceTableId, false);
+ return new String[] {"Success"};
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext,
+ String connector,
+ String sourceTablePath,
+ String properties,
+ Integer parallelism)
+ throws Exception {
+ String targetPaimonTablePath = sourceTablePath + PAIMON_SUFFIX;
+
+ Identifier sourceTableId = Identifier.fromString(sourceTablePath);
+ Identifier targetTableId =
Identifier.fromString(targetPaimonTablePath);
+
+ TableMigrationUtils.getImporter(
+ connector,
+ catalog,
+ sourceTableId.getDatabaseName(),
+ sourceTableId.getObjectName(),
+ targetTableId.getDatabaseName(),
+ targetTableId.getObjectName(),
+ parallelism,
ParameterUtils.parseCommaSeparatedKeyValues(properties))
.executeMigrate();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateDatabaseAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateDatabaseAction.java
index 40c193f8a..6a8afd206 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateDatabaseAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateDatabaseAction.java
@@ -29,17 +29,20 @@ public class MigrateDatabaseAction extends ActionBase {
private final String connector;
private final String hiveDatabaseName;
private final String tableProperties;
+ private final Integer parallelism;
public MigrateDatabaseAction(
String connector,
String warehouse,
String hiveDatabaseName,
Map<String, String> catalogConfig,
- String tableProperties) {
+ String tableProperties,
+ Integer parallelism) {
super(warehouse, catalogConfig);
this.connector = connector;
this.hiveDatabaseName = hiveDatabaseName;
this.tableProperties = tableProperties;
+ this.parallelism = parallelism;
}
@Override
@@ -47,6 +50,10 @@ public class MigrateDatabaseAction extends ActionBase {
MigrateDatabaseProcedure migrateDatabaseProcedure = new
MigrateDatabaseProcedure();
migrateDatabaseProcedure.withCatalog(catalog);
migrateDatabaseProcedure.call(
- new DefaultProcedureContext(env), connector, hiveDatabaseName,
tableProperties);
+ new DefaultProcedureContext(env),
+ connector,
+ hiveDatabaseName,
+ tableProperties,
+ parallelism);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateDatabaseActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateDatabaseActionFactory.java
index 25bb50e79..8ce33f58e 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateDatabaseActionFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateDatabaseActionFactory.java
@@ -28,6 +28,7 @@ public class MigrateDatabaseActionFactory implements
ActionFactory {
private static final String SOURCE_TYPE = "source_type";
private static final String OPTIONS = "options";
+ private static final String PARALLELISM = "parallelism";
@Override
public String identifier() {
@@ -41,10 +42,16 @@ public class MigrateDatabaseActionFactory implements
ActionFactory {
String sourceHiveDatabase = params.get(DATABASE);
Map<String, String> catalogConfig = optionalConfigMap(params,
CATALOG_CONF);
String tableConf = params.get(OPTIONS);
+ Integer parallelism = Integer.parseInt(params.get(PARALLELISM));
MigrateDatabaseAction migrateDatabaseAction =
new MigrateDatabaseAction(
- connector, warehouse, sourceHiveDatabase,
catalogConfig, tableConf);
+ connector,
+ warehouse,
+ sourceHiveDatabase,
+ catalogConfig,
+ tableConf,
+ parallelism);
return Optional.of(migrateDatabaseAction);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateFileAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateFileAction.java
index 0f2404685..798d1d347 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateFileAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateFileAction.java
@@ -32,6 +32,7 @@ public class MigrateFileAction extends ActionBase {
private final String targetTable;
private final String tableProperties;
private boolean deleteOrigin;
+ private Integer parallelism;
public MigrateFileAction(
String connector,
@@ -40,13 +41,15 @@ public class MigrateFileAction extends ActionBase {
String targetTable,
boolean deleteOrigin,
Map<String, String> catalogConfig,
- String tableProperties) {
+ String tableProperties,
+ Integer parallelism) {
super(warehouse, catalogConfig);
this.connector = connector;
this.sourceTable = sourceTable;
this.targetTable = targetTable;
this.deleteOrigin = deleteOrigin;
this.tableProperties = tableProperties;
+ this.parallelism = parallelism;
}
@Override
@@ -58,6 +61,7 @@ public class MigrateFileAction extends ActionBase {
connector,
sourceTable,
targetTable,
- deleteOrigin);
+ deleteOrigin,
+ parallelism);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateFileActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateFileActionFactory.java
index 3c15b03cf..34df99cfd 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateFileActionFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateFileActionFactory.java
@@ -35,6 +35,7 @@ public class MigrateFileActionFactory implements
ActionFactory {
private static final String DELETE_ORIGIN = "delete_origin";
private static final String OPTIONS = "options";
+ private static final String PARALLELISM = "parallelism";
@Override
public String identifier() {
@@ -50,6 +51,7 @@ public class MigrateFileActionFactory implements
ActionFactory {
boolean deleteOrigin = Boolean.parseBoolean(params.get(DELETE_ORIGIN));
Map<String, String> catalogConfig = optionalConfigMap(params,
CATALOG_CONF);
String tableConf = params.get(OPTIONS);
+ Integer parallelism = Integer.parseInt(params.get(PARALLELISM));
MigrateFileAction migrateFileAction =
new MigrateFileAction(
@@ -59,7 +61,8 @@ public class MigrateFileActionFactory implements
ActionFactory {
targetTable,
deleteOrigin,
catalogConfig,
- tableConf);
+ tableConf,
+ parallelism);
return Optional.of(migrateFileAction);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableAction.java
index ae7ff0216..8a4efdfc7 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableAction.java
@@ -30,17 +30,20 @@ public class MigrateTableAction extends ActionBase {
private final String connector;
private final String hiveTableFullName;
private final String tableProperties;
+ private final Integer parallelism;
public MigrateTableAction(
String connector,
String warehouse,
String hiveTableFullName,
Map<String, String> catalogConfig,
- String tableProperties) {
+ String tableProperties,
+ Integer parallelism) {
super(warehouse, catalogConfig);
this.connector = connector;
this.hiveTableFullName = hiveTableFullName;
this.tableProperties = tableProperties;
+ this.parallelism = parallelism;
}
@Override
@@ -48,6 +51,10 @@ public class MigrateTableAction extends ActionBase {
MigrateTableProcedure migrateTableProcedure = new
MigrateTableProcedure();
migrateTableProcedure.withCatalog(catalog);
migrateTableProcedure.call(
- new DefaultProcedureContext(env), connector,
hiveTableFullName, tableProperties);
+ new DefaultProcedureContext(env),
+ connector,
+ hiveTableFullName,
+ tableProperties,
+ parallelism);
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableActionFactory.java
index ffe2a9b99..a1a93bc91 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableActionFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/MigrateTableActionFactory.java
@@ -28,6 +28,7 @@ public class MigrateTableActionFactory implements
ActionFactory {
private static final String SOURCE_TYPE = "source_type";
private static final String OPTIONS = "options";
+ private static final String PARALLELISM = "parallelism";
@Override
public String identifier() {
@@ -41,10 +42,16 @@ public class MigrateTableActionFactory implements
ActionFactory {
String sourceHiveTable = params.get(TABLE);
Map<String, String> catalogConfig = optionalConfigMap(params,
CATALOG_CONF);
String tableConf = params.get(OPTIONS);
+ Integer parallelism = Integer.parseInt(params.get(PARALLELISM));
MigrateTableAction migrateTableAction =
new MigrateTableAction(
- connector, warehouse, sourceHiveTable, catalogConfig,
tableConf);
+ connector,
+ warehouse,
+ sourceHiveTable,
+ catalogConfig,
+ tableConf,
+ parallelism);
return Optional.of(migrateTableAction);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
index ad4f70470..936d9069e 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
@@ -41,20 +41,27 @@ public class MigrateDatabaseProcedure extends ProcedureBase
{
argument = {
@ArgumentHint(name = "connector", type =
@DataTypeHint("STRING")),
@ArgumentHint(name = "source_database", type =
@DataTypeHint("STRING")),
- @ArgumentHint(name = "options", type =
@DataTypeHint("STRING"), isOptional = true)
+ @ArgumentHint(name = "options", type =
@DataTypeHint("STRING"), isOptional = true),
+ @ArgumentHint(
+ name = "parallelism",
+ type = @DataTypeHint("Integer"),
+ isOptional = true)
})
public String[] call(
ProcedureContext procedureContext,
String connector,
String sourceDatabasePath,
- String properties)
+ String properties,
+ Integer parallelism)
throws Exception {
properties = notnull(properties);
+ Integer p = parallelism == null ?
Runtime.getRuntime().availableProcessors() : parallelism;
List<Migrator> migrators =
TableMigrationUtils.getImporters(
connector,
catalog,
sourceDatabasePath,
+ p,
ParameterUtils.parseCommaSeparatedKeyValues(properties));
for (Migrator migrator : migrators) {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateFileProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateFileProcedure.java
index 2538be288..34b016fe0 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateFileProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateFileProcedure.java
@@ -45,6 +45,10 @@ public class MigrateFileProcedure extends ProcedureBase {
@ArgumentHint(
name = "delete_origin",
type = @DataTypeHint("BOOLEAN"),
+ isOptional = true),
+ @ArgumentHint(
+ name = "parallelism",
+ type = @DataTypeHint("Integer"),
isOptional = true)
})
public String[] call(
@@ -52,12 +56,14 @@ public class MigrateFileProcedure extends ProcedureBase {
String connector,
String sourceTablePath,
String targetPaimonTablePath,
- Boolean deleteOrigin)
+ Boolean deleteOrigin,
+ Integer parallelism)
throws Exception {
if (deleteOrigin == null) {
deleteOrigin = true;
}
- migrateHandle(connector, sourceTablePath, targetPaimonTablePath,
deleteOrigin);
+ Integer p = parallelism == null ?
Runtime.getRuntime().availableProcessors() : parallelism;
+ migrateHandle(connector, sourceTablePath, targetPaimonTablePath,
deleteOrigin, p);
return new String[] {"Success"};
}
@@ -65,7 +71,8 @@ public class MigrateFileProcedure extends ProcedureBase {
String connector,
String sourceTablePath,
String targetPaimonTablePath,
- boolean deleteOrigin)
+ boolean deleteOrigin,
+ Integer parallelism)
throws Exception {
Identifier sourceTableId = Identifier.fromString(sourceTablePath);
Identifier targetTableId =
Identifier.fromString(targetPaimonTablePath);
@@ -83,6 +90,7 @@ public class MigrateFileProcedure extends ProcedureBase {
sourceTableId.getObjectName(),
targetTableId.getDatabaseName(),
targetTableId.getObjectName(),
+ parallelism,
Collections.emptyMap());
importer.deleteOriginTable(deleteOrigin);
importer.executeMigrate();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateTableProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateTableProcedure.java
index 3dfc948c6..fff05a1a8 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateTableProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateTableProcedure.java
@@ -45,13 +45,18 @@ public class MigrateTableProcedure extends ProcedureBase {
argument = {
@ArgumentHint(name = "connector", type =
@DataTypeHint("STRING")),
@ArgumentHint(name = "source_table", type =
@DataTypeHint("STRING")),
- @ArgumentHint(name = "options", type =
@DataTypeHint("STRING"), isOptional = true)
+ @ArgumentHint(name = "options", type =
@DataTypeHint("STRING"), isOptional = true),
+ @ArgumentHint(
+ name = "parallelism",
+ type = @DataTypeHint("Integer"),
+ isOptional = true)
})
public String[] call(
ProcedureContext procedureContext,
String connector,
String sourceTablePath,
- String properties)
+ String properties,
+ Integer parallelism)
throws Exception {
properties = notnull(properties);
@@ -60,6 +65,8 @@ public class MigrateTableProcedure extends ProcedureBase {
Identifier sourceTableId = Identifier.fromString(sourceTablePath);
Identifier targetTableId =
Identifier.fromString(targetPaimonTablePath);
+ Integer p = parallelism == null ?
Runtime.getRuntime().availableProcessors() : parallelism;
+
TableMigrationUtils.getImporter(
connector,
catalog,
@@ -67,6 +74,7 @@ public class MigrateTableProcedure extends ProcedureBase {
sourceTableId.getObjectName(),
targetTableId.getDatabaseName(),
targetTableId.getObjectName(),
+ p,
ParameterUtils.parseCommaSeparatedKeyValues(properties))
.executeMigrate();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableMigrationUtils.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableMigrationUtils.java
index 47655d39a..f493dd244 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableMigrationUtils.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableMigrationUtils.java
@@ -37,6 +37,7 @@ public class TableMigrationUtils {
String souceTableName,
String targetDatabase,
String targetTableName,
+ Integer parallelism,
Map<String, String> options) {
switch (connector) {
case "hive":
@@ -52,6 +53,7 @@ public class TableMigrationUtils {
souceTableName,
targetDatabase,
targetTableName,
+ parallelism,
options);
default:
throw new UnsupportedOperationException("Don't support
connector " + connector);
@@ -59,7 +61,11 @@ public class TableMigrationUtils {
}
public static List<Migrator> getImporters(
- String connector, Catalog catalog, String sourceDatabase,
Map<String, String> options) {
+ String connector,
+ Catalog catalog,
+ String sourceDatabase,
+ Integer parallelism,
+ Map<String, String> options) {
switch (connector) {
case "hive":
if (catalog instanceof CachingCatalog) {
@@ -69,7 +75,7 @@ public class TableMigrationUtils {
throw new IllegalArgumentException("Only support Hive
Catalog");
}
return HiveMigrator.databaseMigrators(
- (HiveCatalog) catalog, sourceDatabase, options);
+ (HiveCatalog) catalog, sourceDatabase, options,
parallelism);
default:
throw new UnsupportedOperationException("Don't support
connector " + connector);
}
diff --git
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
index 379c61f70..b9928ce73 100644
---
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
+++
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
@@ -66,9 +66,7 @@ import static
org.apache.paimon.utils.ThreadPoolUtils.createCachedThreadPool;
public class HiveMigrator implements Migrator {
private static final Logger LOG =
LoggerFactory.getLogger(HiveMigrator.class);
-
- private static final ThreadPoolExecutor EXECUTOR =
- createCachedThreadPool(Runtime.getRuntime().availableProcessors(),
"HIVE_MIGRATOR");
+ private ThreadPoolExecutor executor;
private static final Predicate<FileStatus> HIDDEN_PATH_FILTER =
p -> !p.getPath().getName().startsWith("_") &&
!p.getPath().getName().startsWith(".");
@@ -84,6 +82,7 @@ public class HiveMigrator implements Migrator {
private final String targetTable;
private final CoreOptions coreOptions;
private Boolean delete = true;
+ private Integer parallelism;
public HiveMigrator(
HiveCatalog hiveCatalog,
@@ -91,6 +90,7 @@ public class HiveMigrator implements Migrator {
String sourceTable,
String targetDatabase,
String targetTable,
+ Integer parallelism,
Map<String, String> options) {
this.hiveCatalog = hiveCatalog;
this.fileIO = hiveCatalog.fileIO();
@@ -99,11 +99,16 @@ public class HiveMigrator implements Migrator {
this.sourceTable = sourceTable;
this.targetDatabase = targetDatabase;
this.targetTable = targetTable;
+ this.parallelism = parallelism;
this.coreOptions = new CoreOptions(options);
+ this.executor = createCachedThreadPool(parallelism, "HIVE_MIGRATOR");
}
public static List<Migrator> databaseMigrators(
- HiveCatalog hiveCatalog, String sourceDatabase, Map<String,
String> options) {
+ HiveCatalog hiveCatalog,
+ String sourceDatabase,
+ Map<String, String> options,
+ Integer parallelism) {
IMetaStoreClient client = hiveCatalog.getHmsClient();
try {
return client.getAllTables(sourceDatabase).stream()
@@ -115,6 +120,7 @@ public class HiveMigrator implements Migrator {
sourceTable,
sourceDatabase,
sourceTable + PAIMON_SUFFIX,
+ parallelism,
options))
.collect(Collectors.toList());
} catch (TException e) {
@@ -175,7 +181,7 @@ public class HiveMigrator implements Migrator {
}
List<Future<CommitMessage>> futures =
-
tasks.stream().map(EXECUTOR::submit).collect(Collectors.toList());
+
tasks.stream().map(executor::submit).collect(Collectors.toList());
List<CommitMessage> commitMessages = new ArrayList<>();
try {
for (Future<CommitMessage> future : futures) {
diff --git
a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/MigrateDatabaseProcedureITCase.java
b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/MigrateDatabaseProcedureITCase.java
index 7829a9981..0acb7b839 100644
---
a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/MigrateDatabaseProcedureITCase.java
+++
b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/MigrateDatabaseProcedureITCase.java
@@ -130,7 +130,7 @@ public class MigrateDatabaseProcedureITCase extends
ActionITCaseBase {
tEnv.executeSql(
"CALL sys.migrate_database(connector => 'hive',
source_database => 'my_database', options => 'file.format="
+ format
- + "')")
+ + "', parallelism => 6)")
.await();
} else {
tEnv.executeSql(
@@ -257,7 +257,8 @@ public class MigrateDatabaseProcedureITCase extends
ActionITCaseBase {
System.getProperty(HiveConf.ConfVars.METASTOREWAREHOUSE.varname),
"my_database",
catalogConf,
- "");
+ "",
+ 6);
migrateDatabaseAction.run();
tEnv.executeSql(
diff --git
a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/MigrateFileProcedureITCase.java
b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/MigrateFileProcedureITCase.java
index 3f9aebb04..a50bdca3d 100644
---
a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/MigrateFileProcedureITCase.java
+++
b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/MigrateFileProcedureITCase.java
@@ -169,7 +169,8 @@ public class MigrateFileProcedureITCase extends
ActionITCaseBase {
"default.paimontable02",
false,
catalogConf,
- "");
+ "",
+ 6);
migrateFileAction.run();
tEnv.useCatalog("HIVE");
diff --git
a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/MigrateTableProcedureITCase.java
b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/MigrateTableProcedureITCase.java
index b88e2b95b..ca3b6c82e 100644
---
a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/MigrateTableProcedureITCase.java
+++
b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/procedure/MigrateTableProcedureITCase.java
@@ -191,7 +191,8 @@ public class MigrateTableProcedureITCase extends
ActionITCaseBase {
System.getProperty(HiveConf.ConfVars.METASTOREWAREHOUSE.varname),
"default.hivetable",
catalogConf,
- "");
+ "",
+ 6);
migrateTableAction.run();
tEnv.executeSql(
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateFileProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateFileProcedure.java
index 8f4850fef..32f89d47b 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateFileProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateFileProcedure.java
@@ -33,6 +33,7 @@ import org.apache.spark.sql.types.StructType;
import java.util.Collections;
import static org.apache.spark.sql.types.DataTypes.BooleanType;
+import static org.apache.spark.sql.types.DataTypes.IntegerType;
import static org.apache.spark.sql.types.DataTypes.StringType;
/**
@@ -49,7 +50,8 @@ public class MigrateFileProcedure extends BaseProcedure {
ProcedureParameter.required("source_type", StringType),
ProcedureParameter.required("source_table", StringType),
ProcedureParameter.required("target_table", StringType),
- ProcedureParameter.optional("delete_origin", BooleanType)
+ ProcedureParameter.optional("delete_origin", BooleanType),
+ ProcedureParameter.optional("parallelism", IntegerType)
};
private static final StructType OUTPUT_TYPE =
@@ -78,6 +80,8 @@ public class MigrateFileProcedure extends BaseProcedure {
String sourceTable = args.getString(1);
String targetTable = args.getString(2);
boolean deleteNeed = args.isNullAt(3) ? true : args.getBoolean(3);
+ int parallelism =
+ args.isNullAt(4) ? Runtime.getRuntime().availableProcessors()
: args.getInt(4);
Identifier sourceTableId = Identifier.fromString(sourceTable);
Identifier targetTableId = Identifier.fromString(targetTable);
@@ -98,6 +102,7 @@ public class MigrateFileProcedure extends BaseProcedure {
sourceTableId.getObjectName(),
targetTableId.getDatabaseName(),
targetTableId.getObjectName(),
+ parallelism,
Collections.emptyMap());
migrator.deleteOriginTable(deleteNeed);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateTableProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateTableProcedure.java
index a34bc9dac..b74543df1 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateTableProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateTableProcedure.java
@@ -38,6 +38,7 @@ import java.util.HashMap;
import java.util.Map;
import static org.apache.spark.sql.types.DataTypes.BooleanType;
+import static org.apache.spark.sql.types.DataTypes.IntegerType;
import static org.apache.spark.sql.types.DataTypes.StringType;
/**
@@ -59,7 +60,8 @@ public class MigrateTableProcedure extends BaseProcedure {
ProcedureParameter.optional("delete_origin", BooleanType),
ProcedureParameter.optional("target_table", StringType),
ProcedureParameter.optional(
- "options_map", DataTypes.createMapType(StringType,
StringType))
+ "options_map", DataTypes.createMapType(StringType,
StringType)),
+ ProcedureParameter.optional("parallelism", IntegerType),
};
private static final StructType OUTPUT_TYPE =
@@ -91,6 +93,8 @@ public class MigrateTableProcedure extends BaseProcedure {
String targetTable = args.isNullAt(4) ? null : args.getString(4);
MapData mapData = args.isNullAt(5) ? null : args.getMap(5);
Map<String, String> optionMap = mapDataToHashMap(mapData);
+ int parallelism =
+ args.isNullAt(6) ? Runtime.getRuntime().availableProcessors()
: args.getInt(6);
Identifier sourceTableId = Identifier.fromString(sourceTable);
Identifier tmpTableId =
@@ -112,6 +116,7 @@ public class MigrateTableProcedure extends BaseProcedure {
sourceTableId.getObjectName(),
tmpTableId.getDatabaseName(),
tmpTableId.getObjectName(),
+ parallelism,
options);
migrator.deleteOriginTable(deleteNeed);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/TableMigrationUtils.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/TableMigrationUtils.java
index 992f17253..73702bf35 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/TableMigrationUtils.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/TableMigrationUtils.java
@@ -36,6 +36,7 @@ public class TableMigrationUtils {
String sourceTableName,
String targetDatabase,
String targetTableName,
+ Integer parallelism,
Map<String, String> options) {
switch (connector) {
case "hive":
@@ -51,6 +52,7 @@ public class TableMigrationUtils {
sourceTableName,
targetDatabase,
targetTableName,
+ parallelism,
options);
default:
throw new UnsupportedOperationException("Unsupported connector
" + connector);
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateFileProcedureTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateFileProcedureTest.scala
index ceb0879cb..043223d05 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateFileProcedureTest.scala
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateFileProcedureTest.scala
@@ -57,6 +57,41 @@ class MigrateFileProcedureTest extends PaimonHiveTestBase {
}
})
+ Seq("parquet", "orc", "avro").foreach(
+ format => {
+ test(
+ s"Paimon migrate file procedure: migrate $format non-partitioned table
with parallelism") {
+ withTable("hive_tbl_02", "paimon_tbl_02") {
+ // create hive table
+ spark.sql(s"""
+ |CREATE TABLE hive_tbl_02 (id STRING, name STRING, pt
STRING)
+ |USING $format
+ |""".stripMargin)
+
+ spark.sql(s"INSERT INTO hive_tbl_02 VALUES ('1', 'a', 'p1'), ('2',
'b', 'p2')")
+
+ // create paimon table
+ spark.sql(s"""
+ |CREATE TABLE paimon_tbl_02 (id STRING, name STRING, pt
STRING)
+ |USING PAIMON
+ |TBLPROPERTIES ('file.format'='$format', 'bucket'='-1')
+ |""".stripMargin)
+
+ spark.sql(s"INSERT INTO paimon_tbl_02 VALUES ('3', 'c', 'p1'), ('4',
'd', 'p2')")
+
+ spark.sql(
+ s"CALL sys.migrate_file(source_type => 'hive', source_table =>
'$hiveDbName.hive_tbl_02', target_table => '$hiveDbName.paimon_tbl_02',
parallelism => 6)")
+
+ checkAnswer(
+ spark.sql("SELECT * FROM paimon_tbl_02 ORDER BY id"),
+ Row("1", "a", "p1") :: Row("2", "b", "p2") :: Row("3", "c", "p1")
:: Row(
+ "4",
+ "d",
+ "p2") :: Nil)
+ }
+ }
+ })
+
Seq("parquet", "orc", "avro").foreach(
format => {
test(
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateTableProcedureTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateTableProcedureTest.scala
index 7a97c334f..9bffbce09 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateTableProcedureTest.scala
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateTableProcedureTest.scala
@@ -46,6 +46,29 @@ class MigrateTableProcedureTest extends PaimonHiveTestBase {
}
})
+ Seq("parquet", "orc", "avro").foreach(
+ format => {
+ test(
+ s"Paimon migrate table procedure: migrate $format non-partitioned
table with setting parallelism") {
+ withTable("hive_tbl_01") {
+ // create hive table
+ spark.sql(s"""
+ |CREATE TABLE hive_tbl_01 (id STRING, name STRING, pt
STRING)
+ |USING $format
+ |""".stripMargin)
+
+ spark.sql(s"INSERT INTO hive_tbl_01 VALUES ('1', 'a', 'p1'), ('2',
'b', 'p2')")
+
+ spark.sql(
+ s"CALL sys.migrate_table(source_type => 'hive', table =>
'$hiveDbName.hive_tbl_01', options => 'file.format=$format', parallelism => 6)")
+
+ checkAnswer(
+ spark.sql(s"SELECT * FROM hive_tbl_01 ORDER BY id"),
+ Row("1", "a", "p1") :: Row("2", "b", "p2") :: Nil)
+ }
+ }
+ })
+
Seq("parquet", "orc", "avro").foreach(
format => {
test(s"Paimon migrate table procedure: migrate $format table with
options_map") {
