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 35f5d892cf [spark] Introduce reassign_row_id procedure (#9192)
35f5d892cf is described below
commit 35f5d892cf6854e4c3e2f69bf85d04d45536c942
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Wed Aug 12 18:54:12 2026 +0800
[spark] Introduce reassign_row_id procedure (#9192)
---
docs/docs/spark/procedures.md | 12 ++
.../org/apache/paimon/spark/SparkProcedures.java | 2 +
.../spark/procedure/ReassignRowIdProcedure.java | 138 +++++++++++++++++++++
.../procedure/ReassignRowIdProcedureTest.scala | 134 ++++++++++++++++++++
4 files changed, 286 insertions(+)
diff --git a/docs/docs/spark/procedures.md b/docs/docs/spark/procedures.md
index 87a5d5826a..4e2a01c51c 100644
--- a/docs/docs/spark/procedures.md
+++ b/docs/docs/spark/procedures.md
@@ -574,6 +574,18 @@ This section introduce all available spark procedures
about paimon.
CALL sys.drop_global_index(table => 'default.T', index_column =>
'name', index_type => 'btree', dry_run => true)
</td>
</tr>
+ <tr>
+ <td>reassign_row_id</td>
+ <td>
+ To reassign row IDs for a data evolution table when partition row-id
ranges overlap. The table must have <code>row-tracking.enabled=true</code> and
<code>data-evolution.enabled=true</code>. Arguments:
+ <li>table: the target table identifier. Cannot be empty.</li>
+ <li>partitions: partition filter to limit the partitions to
reassign. The comma (",") represents "AND", the semicolon (";") represents
"OR". Left empty for all partitions.</li>
+ </td>
+ <td>
+ CALL sys.reassign_row_id(table => 'default.T')<br/><br/>
+ CALL sys.reassign_row_id(table => 'default.T', partitions =>
'dt=2026-05-19')
+ </td>
+ </tr>
<tr>
<td>copy</td>
<td>
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
index 3b12cf0936..60b5747e3d 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
@@ -47,6 +47,7 @@ import
org.apache.paimon.spark.procedure.MigrateTableProcedure;
import org.apache.paimon.spark.procedure.Procedure;
import org.apache.paimon.spark.procedure.ProcedureBuilder;
import org.apache.paimon.spark.procedure.PurgeFilesProcedure;
+import org.apache.paimon.spark.procedure.ReassignRowIdProcedure;
import org.apache.paimon.spark.procedure.RemoveOrphanFilesProcedure;
import org.apache.paimon.spark.procedure.RemoveUnexistingFilesProcedure;
import org.apache.paimon.spark.procedure.RenameBranchProcedure;
@@ -132,6 +133,7 @@ public class SparkProcedures {
"trigger_tag_automatic_creation",
TriggerTagAutomaticCreationProcedure::builder);
procedureBuilders.put("rewrite_file_index",
RewriteFileIndexProcedure::builder);
procedureBuilders.put("copy", CopyFilesProcedure::builder);
+ procedureBuilders.put("reassign_row_id",
ReassignRowIdProcedure::builder);
return procedureBuilders.build();
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ReassignRowIdProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ReassignRowIdProcedure.java
new file mode 100644
index 0000000000..8863ab56fe
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ReassignRowIdProcedure.java
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.procedure;
+
+import org.apache.paimon.append.dataevolution.DataEvolutionRowIdReassigner;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.spark.utils.SparkProcedureUtils;
+import org.apache.paimon.table.FileStoreTable;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/**
+ * Reassign row id procedure. Reassigns row IDs for a data evolution table
when partition row-id
+ * ranges overlap. Usage:
+ *
+ * <pre><code>
+ * CALL sys.reassign_row_id(table => 'default.T')
+ * CALL sys.reassign_row_id(table => 'default.T', partitions =>
'dt=2026-05-19')
+ * </code></pre>
+ */
+public class ReassignRowIdProcedure extends BaseProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", StringType),
+ ProcedureParameter.optional("partitions", StringType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("result", StringType, true,
Metadata.empty())
+ });
+
+ protected ReassignRowIdProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ Identifier tableIdent = toIdentifier(args.getString(0),
PARAMETERS[0].name());
+ String partitions = args.isNullAt(1) ? null : args.getString(1);
+
+ return modifyPaimonTable(
+ tableIdent,
+ table -> {
+ checkArgument(
+ table instanceof FileStoreTable,
+ "Only FileStoreTable supports reassign_row_id
procedure. The table type is '%s'.",
+ table.getClass().getName());
+
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ PartitionPredicate partitionPredicate =
+
SparkProcedureUtils.convertPartitionsToPartitionPredicate(
+ partitions, fileStoreTable, spark());
+
+ DataEvolutionRowIdReassigner.Result result =
+ new DataEvolutionRowIdReassigner(fileStoreTable,
partitionPredicate)
+ .reassign();
+ String message = formatResult(tableIdent.toString(),
result);
+ return new InternalRow[]
{newInternalRow(UTF8String.fromString(message))};
+ });
+ }
+
+ private static String formatResult(
+ String tableName, DataEvolutionRowIdReassigner.Result result) {
+ if (!result.reassigned) {
+ String reason =
+ result.skipReason == null
+ ? "row IDs are already partition-contiguous"
+ : result.skipReason;
+ return String.format(
+ "Skipped. Row IDs for table '%s' were not reassigned
because %s: "
+ + "snapshot %d unchanged, nextRowId=%d.",
+ tableName, reason, result.previousSnapshotId,
result.nextRowId);
+ }
+ return String.format(
+ "Success. Reassigned row IDs for table '%s': snapshot %d ->
%d, "
+ + "nextRowId %d -> %d, files=%d, rows=%d,
indexFiles=%d.",
+ tableName,
+ result.previousSnapshotId,
+ result.newSnapshotId,
+ result.firstAssignedRowId,
+ result.nextRowId,
+ result.fileCount,
+ result.rowCount,
+ result.indexFileCount);
+ }
+
+ public static ProcedureBuilder builder() {
+ return new Builder<ReassignRowIdProcedure>() {
+ @Override
+ public ReassignRowIdProcedure doBuild() {
+ return new ReassignRowIdProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "ReassignRowIdProcedure";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ReassignRowIdProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ReassignRowIdProcedureTest.scala
new file mode 100644
index 0000000000..f6daa63fbc
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ReassignRowIdProcedureTest.scala
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.procedure
+
+import org.apache.paimon.spark.PaimonSparkTestBase
+
+import org.apache.spark.sql.Row
+import org.assertj.core.api.Assertions.assertThatThrownBy
+
+/** IT Case for [[ReassignRowIdProcedure]]. */
+class ReassignRowIdProcedureTest extends PaimonSparkTestBase {
+
+ test("Paimon Procedure: reassign row ids for interleaved partitions") {
+ withTable("t") {
+ sql(s"""
+ |CREATE TABLE t (id INT, pt STRING)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |PARTITIONED BY (pt)
+ |""".stripMargin)
+ // each INSERT is its own commit, so row ids are assigned in commit
order and
+ // interleave across the 'a' and 'b' partitions
+ sql("INSERT INTO t VALUES (0, 'a')")
+ sql("INSERT INTO t VALUES (1, 'b')")
+ sql("INSERT INTO t VALUES (2, 'a')")
+ sql("INSERT INTO t VALUES (3, 'b')")
+ sql("INSERT INTO t VALUES (4, 'a')")
+
+ checkAnswer(
+ sql("SELECT id, pt, _ROW_ID FROM t ORDER BY id"),
+ Seq(Row(0, "a", 0), Row(1, "b", 1), Row(2, "a", 2), Row(3, "b", 3),
Row(4, "a", 4)))
+
+ val result = sql("CALL sys.reassign_row_id(table => 't')").collect()
+ assert(result.length == 1)
+ assert(
+ result(0).getString(0).startsWith("Success."),
+ s"Unexpected result: ${result(0).getString(0)}")
+
+ // row ids are now contiguous within each partition, data is unaffected
+ checkAnswer(
+ sql("SELECT id, pt, _ROW_ID FROM t ORDER BY id"),
+ Seq(Row(0, "a", 5), Row(1, "b", 8), Row(2, "a", 6), Row(3, "b", 9),
Row(4, "a", 7)))
+
+ // calling again on already-contiguous row ids is a no-op
+ val second = sql("CALL sys.reassign_row_id(table => 't')").collect()
+ assert(
+ second(0).getString(0).startsWith("Skipped."),
+ s"Unexpected result: ${second(0).getString(0)}")
+ }
+ }
+
+ test("Paimon Procedure: reassign row id with partitions filter") {
+ withTable("t") {
+ sql(s"""
+ |CREATE TABLE t (id INT, pt STRING)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |PARTITIONED BY (pt)
+ |""".stripMargin)
+ sql("INSERT INTO t VALUES (0, 'a')")
+ sql("INSERT INTO t VALUES (1, 'b')")
+ sql("INSERT INTO t VALUES (2, 'a')")
+
+ // no partition matches this filter, so nothing needs to be reassigned
+ val skipped =
+ sql("CALL sys.reassign_row_id(table => 't', partitions =>
'pt=c')").collect()
+ assert(
+ skipped(0).getString(0).startsWith("Skipped."),
+ s"Unexpected result: ${skipped(0).getString(0)}")
+
+ val result =
+ sql("CALL sys.reassign_row_id(table => 't', partitions =>
'pt=a')").collect()
+ assert(
+ result(0).getString(0).startsWith("Success."),
+ s"Unexpected result: ${result(0).getString(0)}")
+
+ checkAnswer(
+ sql("SELECT id, pt FROM t ORDER BY id"),
+ Seq(Row(0, "a"), Row(1, "b"), Row(2, "a")))
+ }
+ }
+
+ test("Paimon Procedure: reassign row id requires row tracking enabled") {
+ withTable("t") {
+ sql("CREATE TABLE t (id INT, pt STRING) PARTITIONED BY (pt)")
+ sql("INSERT INTO t VALUES (0, 'a')")
+
+ assertThatThrownBy(() => sql("CALL sys.reassign_row_id(table => 't')"))
+ .hasMessageContaining("row-tracking.enabled=true")
+ }
+ }
+
+ test("Paimon Procedure: reassign row id requires data evolution enabled") {
+ withTable("t") {
+ sql(
+ "CREATE TABLE t (id INT, pt STRING) TBLPROPERTIES
('row-tracking.enabled' = 'true') PARTITIONED BY (pt)")
+ sql("INSERT INTO t VALUES (0, 'a')")
+
+ assertThatThrownBy(() => sql("CALL sys.reassign_row_id(table => 't')"))
+ .hasMessageContaining("data-evolution.enabled=true")
+ }
+ }
+
+ test("Paimon Procedure: reassign row id skips non-partitioned table") {
+ withTable("t") {
+ sql(
+ "CREATE TABLE t (id INT) TBLPROPERTIES ('row-tracking.enabled' =
'true', 'data-evolution.enabled' = 'true')")
+ sql("INSERT INTO t VALUES (1)")
+
+ val result = sql("CALL sys.reassign_row_id(table => 't')").collect()
+ assert(
+ result(0).getString(0).contains("table is not partitioned"),
+ s"Unexpected result: ${result(0).getString(0)}")
+ }
+ }
+}
