Skip to content

[spark] Introduce reassign_row_id procedure - #9192

Merged
JingsongLi merged 2 commits into
apache:masterfrom
zhuxiangyi:spark/reassign-row-id-procedure
Aug 12, 2026
Merged

[spark] Introduce reassign_row_id procedure#9192
JingsongLi merged 2 commits into
apache:masterfrom
zhuxiangyi:spark/reassign-row-id-procedure

Conversation

@zhuxiangyi

Copy link
Copy Markdown
Contributor

Purpose

Flink provides reassign_row_id both as an action (ReassignRowIdAction) and as a
procedure, but Spark has no counterpart. As a result, users running Paimon on Spark
cannot make row IDs of a data evolution table partition-contiguous.

Row IDs are assigned in commit order, so writes that interleave across partitions
leave each partition with a scattered row-id range, which hurts column-store read
efficiency. Reassignment rewrites the row-id ranges so that each partition owns a
contiguous block.

This PR adds the Spark procedure:

-- reassign all partitions
CALL sys.reassign_row_id(table => 'default.T')

-- reassign selected partitions ("," means AND, ";" means OR)
CALL sys.reassign_row_id(table => 'default.T', partitions => 'dt=2026-05-19')

Notes on the implementation:

  • The actual work is delegated to DataEvolutionRowIdReassigner in paimon-core,
    which is engine-agnostic and already used by the Flink side, so no reassignment
    logic is duplicated here.
  • partitions is parsed with the existing
    SparkProcedureUtils#convertPartitionsToPartitionPredicate, keeping the partition
    spec syntax consistent with other Spark procedures; when omitted, all partitions
    are reassigned.
  • The table must have row-tracking.enabled = true and data-evolution.enabled = true;
    both are validated by the core reassigner.
  • The procedure returns a single result string describing whether the reassignment
    happened, and for a skip, why it was skipped.

Docs for the new procedure are added to docs/docs/spark/procedures.md.

Tests

Added ReassignRowIdProcedureTest in paimon-spark-ut, covering:

  • reassigning a table written with interleaved partitions, asserting the row IDs
    become contiguous per partition and the data itself is unchanged;
  • idempotency: a second call on already-contiguous row IDs is skipped;
  • the partitions filter, for both a spec that matches no partition (skipped) and
    one that matches (reassigned);
  • the two precondition failures, row-tracking.enabled=true and
    data-evolution.enabled=true;
  • a non-partitioned table, which is skipped.

Spark has no counterpart to Flink's reassign_row_id procedure, so row IDs
of a data evolution table cannot be made partition-contiguous from Spark.

Add sys.reassign_row_id(table, partitions), delegating to the engine-agnostic
DataEvolutionRowIdReassigner in paimon-core.
@JingsongLi

Copy link
Copy Markdown
Contributor

+1

@JingsongLi
JingsongLi merged commit 35f5d89 into apache:master Aug 12, 2026
13 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants