This is an automated email from the ASF dual-hosted git repository.
huaxingao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 8348decfd5 Spark3.4: Backport ProcedureInput for MigrateTableProcedure
And SnapshotTableProcedure. (#12837)
8348decfd5 is described below
commit 8348decfd5b2ffeb907f5a0f9a4c36e2404f0c65
Author: slfan1989 <[email protected]>
AuthorDate: Sat Apr 19 01:06:58 2025 +0800
Spark3.4: Backport ProcedureInput for MigrateTableProcedure And
SnapshotTableProcedure. (#12837)
---
.../spark/procedures/MigrateTableProcedure.java | 44 ++++++++++---------
.../spark/procedures/SnapshotTableProcedure.java | 49 +++++++++++-----------
2 files changed, 46 insertions(+), 47 deletions(-)
diff --git
a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/procedures/MigrateTableProcedure.java
b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/procedures/MigrateTableProcedure.java
index 7c67a1aced..5b4595b9d6 100644
---
a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/procedures/MigrateTableProcedure.java
+++
b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/procedures/MigrateTableProcedure.java
@@ -21,7 +21,7 @@ package org.apache.iceberg.spark.procedures;
import java.util.Map;
import org.apache.iceberg.actions.MigrateTable;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.spark.SparkTableUtil;
import org.apache.iceberg.spark.actions.MigrateTableSparkAction;
import org.apache.iceberg.spark.actions.SparkActions;
@@ -33,16 +33,23 @@ import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
-import scala.runtime.BoxedUnit;
class MigrateTableProcedure extends BaseProcedure {
+
+ private static final ProcedureParameter TABLE_PARAM =
+ ProcedureParameter.required("table", DataTypes.StringType);
+ private static final ProcedureParameter PROPERTIES_PARAM =
+ ProcedureParameter.optional("properties", STRING_MAP);
+ private static final ProcedureParameter DROP_BACKUP_PARAM =
+ ProcedureParameter.optional("drop_backup", DataTypes.BooleanType);
+ private static final ProcedureParameter BACKUP_TABLE_NAME_PARAM =
+ ProcedureParameter.optional("backup_table_name", DataTypes.StringType);
+ private static final ProcedureParameter PARALLELISM_PARAM =
+ ProcedureParameter.optional("parallelism", DataTypes.IntegerType);
+
private static final ProcedureParameter[] PARAMETERS =
new ProcedureParameter[] {
- ProcedureParameter.required("table", DataTypes.StringType),
- ProcedureParameter.optional("properties", STRING_MAP),
- ProcedureParameter.optional("drop_backup", DataTypes.BooleanType),
- ProcedureParameter.optional("backup_table_name", DataTypes.StringType),
- ProcedureParameter.optional("parallelism", DataTypes.IntegerType)
+ TABLE_PARAM, PROPERTIES_PARAM, DROP_BACKUP_PARAM,
BACKUP_TABLE_NAME_PARAM, PARALLELISM_PARAM
};
private static final StructType OUTPUT_TYPE =
@@ -76,25 +83,16 @@ class MigrateTableProcedure extends BaseProcedure {
@Override
public InternalRow[] call(InternalRow args) {
- String tableName = args.getString(0);
+ ProcedureInput input = new ProcedureInput(spark(), tableCatalog(),
PARAMETERS, args);
+ String tableName = input.asString(TABLE_PARAM, null);
Preconditions.checkArgument(
tableName != null && !tableName.isEmpty(),
"Cannot handle an empty identifier for argument table");
- Map<String, String> properties = Maps.newHashMap();
- if (!args.isNullAt(1)) {
- args.getMap(1)
- .foreach(
- DataTypes.StringType,
- DataTypes.StringType,
- (k, v) -> {
- properties.put(k.toString(), v.toString());
- return BoxedUnit.UNIT;
- });
- }
+ Map<String, String> properties = input.asStringMap(PROPERTIES_PARAM,
ImmutableMap.of());
- boolean dropBackup = args.isNullAt(2) ? false : args.getBoolean(2);
- String backupTableName = args.isNullAt(3) ? null : args.getString(3);
+ boolean dropBackup = input.asBoolean(DROP_BACKUP_PARAM, false);
+ String backupTableName = input.asString(BACKUP_TABLE_NAME_PARAM, null);
MigrateTableSparkAction migrateTableSparkAction =
SparkActions.get().migrateTable(tableName).tableProperties(properties);
@@ -107,8 +105,8 @@ class MigrateTableProcedure extends BaseProcedure {
migrateTableSparkAction =
migrateTableSparkAction.backupTableName(backupTableName);
}
- if (!args.isNullAt(4)) {
- int parallelism = args.getInt(4);
+ if (input.isProvided(PARALLELISM_PARAM)) {
+ int parallelism = input.asInt(PARALLELISM_PARAM);
Preconditions.checkArgument(parallelism > 0, "Parallelism should be
larger than 0");
migrateTableSparkAction =
migrateTableSparkAction.executeWith(SparkTableUtil.migrationService(parallelism));
diff --git
a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/procedures/SnapshotTableProcedure.java
b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/procedures/SnapshotTableProcedure.java
index 37dfde76b7..1ae7b71bc3 100644
---
a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/procedures/SnapshotTableProcedure.java
+++
b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/procedures/SnapshotTableProcedure.java
@@ -21,7 +21,7 @@ package org.apache.iceberg.spark.procedures;
import java.util.Map;
import org.apache.iceberg.actions.SnapshotTable;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.spark.SparkTableUtil;
import org.apache.iceberg.spark.actions.SparkActions;
import org.apache.spark.sql.catalyst.InternalRow;
@@ -31,16 +31,23 @@ import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
-import scala.runtime.BoxedUnit;
class SnapshotTableProcedure extends BaseProcedure {
+
+ private static final ProcedureParameter SOURCE_TABLE_PARAM =
+ ProcedureParameter.required("source_table", DataTypes.StringType);
+ private static final ProcedureParameter TABLE_PARAM =
+ ProcedureParameter.required("table", DataTypes.StringType);
+ private static final ProcedureParameter LOCATION_PARAM =
+ ProcedureParameter.optional("location", DataTypes.StringType);
+ private static final ProcedureParameter PROPERTIES_PARAM =
+ ProcedureParameter.optional("properties", STRING_MAP);
+ private static final ProcedureParameter PARALLELISM_PARAM =
+ ProcedureParameter.optional("parallelism", DataTypes.IntegerType);
+
private static final ProcedureParameter[] PARAMETERS =
new ProcedureParameter[] {
- ProcedureParameter.required("source_table", DataTypes.StringType),
- ProcedureParameter.required("table", DataTypes.StringType),
- ProcedureParameter.optional("location", DataTypes.StringType),
- ProcedureParameter.optional("properties", STRING_MAP),
- ProcedureParameter.optional("parallelism", DataTypes.IntegerType)
+ SOURCE_TABLE_PARAM, TABLE_PARAM, LOCATION_PARAM, PROPERTIES_PARAM,
PARALLELISM_PARAM
};
private static final StructType OUTPUT_TYPE =
@@ -74,26 +81,20 @@ class SnapshotTableProcedure extends BaseProcedure {
@Override
public InternalRow[] call(InternalRow args) {
- String source = args.getString(0);
+ ProcedureInput input = new ProcedureInput(spark(), tableCatalog(),
PARAMETERS, args);
+
+ String source = input.asString(SOURCE_TABLE_PARAM, null);
Preconditions.checkArgument(
source != null && !source.isEmpty(),
"Cannot handle an empty identifier for argument source_table");
- String dest = args.getString(1);
+
+ String dest = input.asString(TABLE_PARAM, null);
Preconditions.checkArgument(
dest != null && !dest.isEmpty(), "Cannot handle an empty identifier
for argument table");
- String snapshotLocation = args.isNullAt(2) ? null : args.getString(2);
-
- Map<String, String> properties = Maps.newHashMap();
- if (!args.isNullAt(3)) {
- args.getMap(3)
- .foreach(
- DataTypes.StringType,
- DataTypes.StringType,
- (k, v) -> {
- properties.put(k.toString(), v.toString());
- return BoxedUnit.UNIT;
- });
- }
+
+ String snapshotLocation = input.asString(LOCATION_PARAM, null);
+
+ Map<String, String> properties = input.asStringMap(PROPERTIES_PARAM,
ImmutableMap.of());
Preconditions.checkArgument(
!source.equals(dest),
@@ -104,8 +105,8 @@ class SnapshotTableProcedure extends BaseProcedure {
action.tableLocation(snapshotLocation);
}
- if (!args.isNullAt(4)) {
- int parallelism = args.getInt(4);
+ if (input.isProvided(PARALLELISM_PARAM)) {
+ int parallelism = input.asInt(PARALLELISM_PARAM);
Preconditions.checkArgument(parallelism > 0, "Parallelism should be
larger than 0");
action =
action.executeWith(SparkTableUtil.migrationService(parallelism));
}
