[core][spark][flink] Enable data evolution on an existing append table in place

[zhuxiangyi]

Purpose

Feature: convert an existing append table to a Data Evolution table in place, without rewriting any data file.

Before converting, stop writers that run an older Paimon version. They do not refuse to commit after the schema change, so files they commit afterwards would have no row id. Writers on this version are handled by the procedure (see "What to expect around the conversion"). Do not restore a job from a checkpoint taken before the conversion if it may hold pending compaction: that commit is refused.

Why

row-tracking.enabled and data-evolution.enabled are immutable (ALTER TABLE ... SET is refused as soon as the table has a snapshot). Today the only way to get Data Evolution on a table that already holds data is to create a new table and INSERT OVERWRITE everything into it: a full rewrite, a new table location, and all history/tags/branches lost. Data Evolution is exactly the feature people want on big tables (partial-column updates, sub-field updates, blob columns), so this is the tables it is most painful for.

Global indexes are the second reason. A global index stores row ids, so CreateGlobalIndexProcedure refuses any table without row-tracking.enabled, in Spark, Flink and PyPaimon alike — a plain append table cannot have one at all, whatever the index type:

	Needs row tracking	Works on a plain append table
Global index: full-text, vector, FM, BTree, bitmap, multivalue	yes	no
File index: bloom filter, bitmap, BSI, range bitmap	no	yes, and rewrite_file_index backfills them

File indexes live next to each data file and are rebuilt whenever a file is rewritten, so file-level pruning has always been available. Row-id addressing is what a global index needs, and it is what unlocks full-text and vector search, LIKE '%x%' through FM, and index-driven top-k. Unlike partial-column updates, that costs nothing on the write side: it is pure read-side gain. Once again the tables people most want to index — a multi-TB log or document table — are the ones they can least afford to rewrite.

Just flipping the two options is not enough, and is why they were made immutable: a Data Evolution table addresses rows by firstRowId + position, and the existing files have no firstRowId. Every code path that assumes the id (nonNullFirstRowId, the split generator, compaction, MERGE INTO on _ROW_ID) would fail or silently skip rows. Enabling on an existing table therefore needs three things: row ids for the existing files, a safe switch of the schema, and reads that tolerate the window in between.

Usage

-- Spark SQL
CALL sys.enable_data_evolution(table => 'default.target_table', dry_run => true);  -- report only
CALL sys.enable_data_evolution(table => 'default.target_table');

-- Flink SQL
CALL sys.enable_data_evolution(`table` => 'default.target_table', dry_run => true);
CALL sys.enable_data_evolution('default.target_table');

# Flink action jar
<FLINK_HOME>/bin/flink run paimon-flink-action-<version>.jar enable_data_evolution \
    --warehouse <warehouse-path> --database default --table target_table [--dry_run true] \
    [--catalog_conf <key>=<value> ...]

The call returns one row:

Result	Meaning
Dry run. Enabling data evolution on table 'db.t' would do: ...	the same report, nothing is changed
Success. Enabled data evolution on table 'db.t': schema 0 -> 1, snapshot 2 -> 3, N file(s) with M row(s) assigned row ids, nextRowId=M.	converted: one new schema, a metadata-only OVERWRITE snapshot that assigns the row ids, and an empty APPEND snapshot that fences off writers which checked the previous schema (plus one more OVERWRITE snapshot if such a writer committed before the fence)
Skipped. Table 'db.t' was not changed: data evolution is already enabled.	idempotent

What the table looks like afterwards

• Exactly two options are added to a new schema, everything else in the schema (fields, partition keys, comment, all other options) is carried over unchanged:

  row-tracking.enabled   = true
  data-evolution.enabled = true
  

  The result is the same schema a table gets from CREATE TABLE ... TBLPROPERTIES ('row-tracking.enabled'='true', 'data-evolution.enabled'='true'). Nothing else is switched on: bucket, deletion vectors and data-evolution.nested-field.enabled stay as they were and can be changed afterwards with ALTER TABLE as on any Data Evolution table.
• Every existing data file keeps its name, path (external paths included), statistics, embedded file index and deletion vector. It gets a firstRowId, and its sequence numbers are set to the id of the snapshot that assigns it, as for any row-tracking commit, so that later column writes take precedence. On a table that already had row tracking, the files keep their row ids and their sequence numbers are reset to 1: a row-tracking-only writer numbers the rows of a file, which would hide later column writes. Row ids are contiguous per partition, in commit order, so sys.reassign_row_id has nothing to do afterwards. New writes continue at nextRowId.
• The two options are written after every existing file has its row id (manifest rewrite first, schema second). The procedure is the only way to switch them on for an existing table: the schema change it uses cannot be created elsewhere, and ALTER TABLE refuses both options even before the first snapshot.
• After the switch the procedure commits a fence and repairs what was committed before it, so a writer that checked the previous schema cannot leave a file without row id behind (details below).

Requirements: append table without primary key, bucket = -1 (the default for an append table), clustering.incremental off. Tables of a REST catalog are rejected in this version. A row-tracking table whose files store the row ids of their rows (written by a copy-on-write UPDATE, DELETE or MERGE INTO) is refused before anything changes, because those ids need not be contiguous and no first row id can describe them; rewrite those rows first, for example with INSERT OVERWRITE.

What to expect around the conversion (also in the docs):

• Writers of an older Paimon version must be stopped first, see the note at the top.
• A writer or compactor that loaded the table before the conversion, typically a running Flink streaming job, is refused on its next commit with a message asking to restart it, also when the table already had row tracking; after the restart it assigns row ids as usual. Run the conversion at a write-traffic low.
• A job restored from a checkpoint taken before the conversion: restored appended files are committed and get their row ids and sequence numbers from that commit; restored compaction output, which has no row ids and would replace files that have them, is refused before anything is published.
• Snapshots and tags from before the conversion stay readable; their files have no row id, so _ROW_ID reads as NULL. Rolling back to them (rollback_to, also by tag, and rollback_to_as_latest) is refused while data evolution is enabled, also for a table that already had row tracking, since that would restore the old sequence numbers. So is a rollback to a snapshot that still holds files the conversion had not repaired yet.
• Streaming reads that were started before the conversion continue past the conversion snapshot without replaying the table.

What

One PR. Commits 1–4 below are the feature, followed by [core] Report the table's row id when a resumed run assigns none (a run that finds the row ids already committed by an earlier, interrupted run reports the table's nextRowId instead of none). The commits after that address the review, see "Review follow-ups" below.

1. [core] Guard commits and rollbacks across a row-tracking switch

Two guards that make the switch safe against concurrent writers and against history:

• FileStoreCommitImpl refuses a commit from a writer that loaded the table without row tracking when the latest schema now has it enabled (checkRowTrackingNotEnabledAfterLoad). Such a writer would commit files without a row id into a row-tracking table. The message asks to restart the writer. Writers that loaded the table with row tracking, and ordinary schema changes (add column, set other options), are unaffected.
• AbstractFileStoreTable.rollbackTo (snapshot and tag) and rollbackToAsLatest refuse to roll back to a snapshot whose schema has no row tracking while the current schema has it: that snapshot's files would become unaddressable.
• replaceManifestList gets an overload that takes the schema id and nextRowId explicitly (used by commit 3).

2. [core] Read data-evolution files that have no first row id as plain files

A Data Evolution table can now hold files without firstRowId (files written before the switch, snapshots and tags from before the conversion, or a table where only the schema was flipped). They are treated as complete-row files:

• DataEvolutionUtils.splitByRowIdPresence separates them; DataEvolutionSplitGenerator bin-packs them into raw-convertible splits after the row-id-range groups, so a file with an id and a file without never share a split.
• DataEvolutionSplitReadProvider does not match such splits; AppendOnlyFileStoreTable.newRead always registers the plain append raw-file provider after the Data Evolution one, so they are read as ordinary append files with _ROW_ID = NULL.
• DataEvolutionFileStoreScan groups them as singletons and applies file statistics to them individually (predicate pushdown keeps working on them).
• DataFileMeta.nonNullFirstRowId and DataEvolutionCompactRangePlanner name the file in their message and point at sys.enable_data_evolution, instead of a bare firstRowId must not be null.

3. [core] Add DataEvolutionEnabler to convert an append table in place

• A dedicated schema change that sets both options in a new schema. SchemaManagerUtils lets it through the immutable-option check; validateTableSchema still enforces every other Data Evolution constraint (no primary key, bucket = -1, no clustering.incremental). After the review it became DataEvolutionEnabler.EnableDataEvolution with a private constructor, so only the procedure can issue it; it is not part of the public SchemaChange API or the REST protocol.
• DataEvolutionEnabler(catalog, identifier).run(dryRun):
  1. validate: append table, candidate schema passes validateTableSchema; REST catalog tables are rejected in this version (the manifest rewrite is a direct replaceManifestList on the file store, which the REST protocol does not expose yet).
  2. assign row ids: metadata only. Live ADD entries without id are ordered per partition in manifest/commit order and given contiguous ids starting at nextRowId (or 0); only the manifests that hold converted files are rewritten, and the snapshot is replaced with replaceManifestList carrying the new nextRowId. On a conflict (another commit landed) it re-plans from the new latest snapshot, bounded by the table's commit retry settings.
  3. switch the schema: catalog.alterTable with the change above, which keeps catalog locks and metastore sync. Skipped when a concurrent run switched it already.
  4. fence and repair: an empty snapshot through the normal commit path, then row ids for every file committed before it without one.
  • Dry run reports the plan without changing anything. Idempotent: a converted table reports Skipped.
  • Row ids are contiguous per partition in commit order, so a converted table needs no reassign_row_id afterwards.

4. [spark][flink] Add the enable_data_evolution procedure and action

• Spark CALL sys.enable_data_evolution(table => 't' [, dry_run => true]), Flink CALL sys.enable_data_evolution('db.t' [, dry_run]) and the enable_data_evolution action jar.
• Spark MERGE INTO on a Data Evolution table refuses a target that still has files without row id, with a message pointing at the procedure, instead of silently skipping those rows.
• Spark BaseScan reads _ROW_ID / _SEQUENCE_NUMBER as nullable so such files read as NULL (a non-nullable read type turned it into 0).
• Two copy(...) → copyWithoutTimeTravel(...) changes in MergeIntoPaimonDataEvolutionTable (Spark 3 and the Spark 4.0 copy) and DataEvolutionPaimonWriter. This fixes a pre-existing bug on native Data Evolution tables too: the merge pins its scan with scan.snapshot-id, and copy re-applies that time travel and adopts the schema the snapshot was committed with. After a schema-only change (ALTER TABLE creates no snapshot: enabling data-evolution.nested-field.enabled, adding a column) the files written by the merge were stamped with the old schema id, and reading them back returned the stale sub-field value or failed in TableSchema.project. copyWithoutTimeTravel keeps the current schema while still pinning the snapshot.

Review follow-ups (each finding was first reproduced by a deterministic regression test):

• [core] Check the latest schema in the rollback guards: rollbackTo(snapshot), rollbackTo(tag) and rollbackToAsLatest decide by the latest persisted schema, not by the options of the table object, so a table loaded before the conversion cannot roll back across it.
• [api][core] Keep enabling data evolution out of the REST protocol: removed from the SchemaChange JSON subtypes and the OpenAPI spec (which is back to master).
• [core] Fence writers and verify row ids when enabling data evolution: the fence and repair above; every row id commit reads the latest schema right before the replacement (a concurrent ADD COLUMN does not move the snapshot).
• [core] Stamp converted files with the sequence number of the row id commit: found while checking the tests. A plain append writer numbers its rows, so a file written in one commit of many rows has sequence numbers far above the snapshot ids of later commits; a data-evolution read takes each column from the file with the highest one, so after the conversion a partial-column write or MERGE INTO on such a file was silently ignored.
• [core] Test expiring snapshots and tags from before the data evolution conversion: the data files, which the conversion re-references through new manifests, are kept.
• [core] Only the enable_data_evolution procedure can switch data evolution on: the public SchemaChange.enableDataEvolution() is removed; a direct catalog.alterTable could switch the options without the fence. The schema-manager precondition and snapshot mark of the earlier follow-up are removed with it.
• [core] Refuse switching row tracking on with ALTER TABLE before the first snapshot: the same race existed on master, because the options are only immutable once a table has snapshots. Behaviour change: ALTER TABLE SET ('row-tracking.enabled' | 'data-evolution.enabled' = 'true') is now refused on an empty table too (set it at CREATE TABLE or call the procedure); switching them off is unaffected.
• [core] Fix row-tracking conversion sequences and resumed writer fences: a table that already had row tracking keeps its row ids, and its files get sequence number 1, below every data-evolution update, including one committed while the repair retries. A writer or compactor loaded before data evolution was enabled is refused even if row tracking was on already. An interrupted conversion reports Skipped. only once a snapshot on the data-evolution schema has fenced old writers; otherwise the next run completes the fence and repair, also on an empty table.
• [core] Refuse rollbacks and restored compactions across the data evolution conversion (second review): the rollback guards also check the data-evolution boundary, in one helper shared by rollbackTo and rollbackToAsLatest. A data-evolution commit refuses an added file that would be left without row ids (not APPEND, no first row id, no stored row ids), which is what compaction output restored from a pre-conversion checkpoint is.
• [core] Do not depend on the listing order of schema files in DataEvolutionEnablerTest: test only.
• [core] Cover restored appends and unrepaired fence snapshots of the data evolution conversion: appended files restored from a pre-conversion writer are stamped with the commit's snapshot id instead of keeping row-count sequence numbers. A rollback of a converted table also checks the files the target makes live again, so a rollback to the fence snapshot before the repair finished is refused.
• [core] Refuse enabling data evolution on files that store their row ids: see Requirements; a rollback no longer counts such files as unconverted.

Benefit

• A table of any size gets Data Evolution with one metadata-only call: no data rewrite, same location, history/tags/branches kept, old snapshots stay readable.
• The switch is safe: stale writers are refused instead of corrupting row ids, rollbacks across the boundary are refused, files that slipped in are repaired, and every row-id-dependent path reports the offending file and the fix.
• Sub-field / added-column merges after ALTER TABLE on existing Data Evolution tables now write files with the right schema id.
• What the table can use afterwards: partial-column and sub-field updates that rewrite only the touched columns, blob columns, and global indexes through CALL sys.create_global_index(...) — which is itself incremental, so the first call indexes the existing data and later calls only cover what has been appended since.

Tests

paimon-core

• RowTrackingEnableGuardTest (7): stale writer refused after the switch, stale writer still commits after an ordinary schema change, writer on a row-tracking table never refused, rollback across the boundary refused (snapshot, tag and as-latest), also through a table object loaded before the switch, plain append table unaffected, replaceManifestList with schema id.
• DataEvolutionFilesWithoutRowIdTest (5): full read / projection / predicate on files without id (_ROW_ID NULL, statistics prune), mixed files with and without id never share a split, time travel to a pre-conversion snapshot, compaction refuses with a message naming the file, nonNullFirstRowId message.
• DataEvolutionEnablerTest (42): empty table switches the schema and commits the fence; contiguous ids in sequence order; per-partition ranges; only manifests holding converted files are rewritten; every other file attribute (external path, embedded file index, stats, ...) is kept; dry run changes nothing; partial-column write and Data Evolution compaction on a converted table; a column write after converting a 20-row commit wins over the converted file; expiring pre-conversion snapshots and deleting a pre-conversion tag keep the data; converts one branch only; a row-tracking-only table keeps its row ids; rejects primary-key / bucketed / incremental-clustering / REST catalog tables; concurrent append and compaction before the row id commit are assigned on retry; a stale write between the row id commit and the switch is repaired; the row id commit references a schema changed concurrently; two concurrent runs switch the schema once; a writer paused right after its schema check (a FileIO that blocks its first manifest write) is refused when released after the procedure, repaired when released before the fence, refused on an empty table; the schema change cannot be created outside the enabler; ALTER TABLE cannot enable row tracking, with or without snapshots, also with a writer in flight (the reviewer's probe). Every conversion test asserts that no live file lacks a row id or has a sequence number above the latest snapshot id. Added in the later follow-ups: a column write after converting a row-tracking-only table wins over its old sequence numbers, also when committed while the repair retries; row-tracking-only writers and compactors loaded before the conversion are refused; row-tracking-only compaction committed before the fence is fully repaired; restored compaction output of a row-tracking-only table is refused after the conversion (a real AppendCompactTask message serialized and restored through filterAndCommitMultiple); a restored append of a pre-conversion writer is stamped with its snapshot; rollback to a snapshot, a tag or as-latest from before converting a row-tracking-only table is refused, and so is a rollback to the fence before the repair; a retry after an interrupted schema switch still fences a paused writer, also on an empty table; files that store their row ids are refused.
• SchemaManagerTest#testSetOptionCannotEnableRowTrackingWithoutSnapshots; RESTApiJsonTest asserts that a server cannot parse an enableDataEvolution action.
• All existing DataEvolution* suites, AppendOnlySimpleTableTest, BlobTableTest pass.

paimon-spark (Spark 3.5)

• EnableDataEvolutionProcedureTest (9): UPDATE and MERGE INTO after converting a 100-row commit are applied; basic conversion + dry run + idempotence + MERGE INTO / UPDATE / INSERT afterwards (partial-column files, row ids continue after the assigned range); partition-contiguous ids and reassign_row_id reports Skipped.; deletion vectors kept and DELETE / merge-delete keep working; sub-field data evolution on a converted table; pre-conversion snapshot and tag readable with _ROW_ID NULL; refusals (primary key, bucketed, ALTER TABLE with and without data); MERGE INTO refuses files without id, then the procedure repairs the table.
• RowTrackingTestBase: new merge after a schema-only change writes files with the current schema (fails on master: the written file carries schema id 0). RowTrackingTest, BlobUpdateTest, DataEvolutionDeletionTest, UpdateTableTest, ReassignRowIdProcedureTest, NestedSubfieldMergeIntoTest, DataEvolutionUpdateSnapshotTest pass.

paimon-flink

• EnableDataEvolutionProcedureITCase (6): procedure (ids read through t$row_tracking), refusal of a primary-key table, action jar, writer loaded before the conversion is refused, streaming reads started before the conversion continue past the conversion snapshots without replaying the table (default and streaming-read-append-overwrite), ALTER TABLE cannot enable row tracking.

spotless:check and checkstyle:check pass on paimon-api, paimon-core, paimon-spark-common, paimon-spark-4.0, paimon-spark-ut and paimon-flink-common.

API and Format

• No new public SchemaChange, no REST protocol change (the OpenAPI spec is unchanged). The REST catalog does not support the conversion yet; the procedure reports this.
• ALTER TABLE can no longer switch row-tracking.enabled or data-evolution.enabled on for a table without snapshots (it already could not once the table had snapshots).
• A commit to a data-evolution table refuses an added file that would be left without row ids, and a writer loaded before data evolution was enabled is refused even if row tracking was on already.
• Data Evolution tables may now contain data files with firstRowId = null; they are read as complete-row files with _ROW_ID = NULL. The file format itself is unchanged.
• New Spark procedure sys.enable_data_evolution, Flink procedure sys.enable_data_evolution and action enable_data_evolution.

Documentation

• multimodal-table/data-evolution.mdx: new section "Enable on an Existing Append Table" (what the procedure does, that it is the only way to switch the options on, what to expect around the conversion: older-version writers must be stopped, stale writers, jobs restored from a checkpoint, tables whose files store row ids, old snapshots, rollback, nested-field option).
• append-table/row-tracking.md, spark/procedures.md, spark/procedures/indexes.md, flink/procedures.md, flink/procedures/compaction.md, flink/action-jars.md: procedure and action entries.

[Akash3121]

This needs to check the latest persisted schema rather than this table object's cached coreOptions() . A  FileStoreTable  loaded before  enable_data_evolution  continues to report  rowTrackingEnabled() == false ; after another handle completes conversion, that stale object can therefore call  rollbackTo(1)  /  rollbackTo("before")  and bypass this guard, making a pre-conversion snapshot current while the latest schema still has row tracking enabled. The stale-writer commit guard does not cover these direct rollback paths, and  rollbackToAsLatest  has the same issue. Please derive the source-side state from  schemaManager().latest()  in every rollback path, and add a regression test that retains the pre-conversion table handle, enables row tracking through another handle, then verifies snapshot, tag, and as-latest rollback are all refused and leave the latest snapshot unchanged.

fix: Base both rollback guards on the latest persisted schema, not the table object's cached options.

[zhuxiangyi]

Fixed in 35110ec. Both guards now go by the latest persisted schema instead of the options of the table object:

• AbstractFileStoreTable.checkRollbackKeepsRowTracking, used by rollbackTo(long) and rollbackTo(String), reads schemaManager().latest().
• FileStoreCommitImpl.rollbackToAsLatest reads schemaManager.latestOrThrow(...).

Regression test: RowTrackingEnableGuardTest#testRollbackThroughTableLoadedBeforeRowTrackingIsRefused. It keeps the table object loaded before row tracking was enabled, enables row tracking through another handle, and checks that snapshot, tag and as-latest rollback are all refused, that the latest snapshot and the tag are unchanged, and that a target committed with row tracking still rolls back through the same object. Reverting either guard makes the test fail.

[Akash3121]

Please do not stamp this snapshot with the schema cached when the enabler initially loaded the table. Schema commits do not move the latest snapshot, so an  ADD COLUMN  can complete between  plan()  and this CAS without making  replaceManifestList  fail; this line will then create a newer conversion snapshot referencing the older schema, and time travel to it loses a schema change that committed before the snapshot. The problem persists across retries because the same  table  object is reused. Fetch the latest schema ID for each commit attempt (after  beforeRowIdCommit , consistent with the normal commit path), and add a hook-based regression test that commits a schema change before the replacement snapshot and asserts that snapshot references the new schema and exposes the new column under time travel.

fix: Resolve the latest schema ID inside each assignment commit attempt, after the test/concurrency hook and immediately before replacement.

[zhuxiangyi]

Fixed in 37ff2bd. commitAssignment now reads the schema id with table.schemaManager().latestOrThrow(...) inside every commit attempt, after the beforeRowIdCommit hook and right before replaceManifestList, the same way the normal commit path does. The cached table.schema() is no longer used for the snapshot.

Regression test: DataEvolutionEnablerTest#testRowIdCommitReferencesTheLatestSchema. It commits ADD COLUMN c from the hook, between planning and the replacement, and asserts that the row id snapshot references the new schema, that time travel to it exposes c, and that the final schema has c with data evolution enabled. Reverting the change makes the test fail.

[Akash3121]

This conversion-only action should not be exposed through generic REST schema alteration while  DataEvolutionEnabler  explicitly rejects REST catalogs. As added here and in the OpenAPI schema, a client can POST  enableDataEvolution  through  alterTable ;  SchemaManagerUtils  then flips both options without assigning IDs to existing files. The table is left in the transient mixed state, but REST has no supported conversion procedure to repair it, so operations such as MERGE remain unusable. Please either keep this action out of public REST deserialization/OpenAPI until an atomic server-side conversion endpoint exists, or make the REST server reject it. Add a REST regression test on a nonempty append table verifying that direct alteration cannot change either option.

applies for the line 85 in this same file; also applies to docs/static/rest-catalogn-open-api.yaml and SchemaManagerUtils.java files

[zhuxiangyi]

Fixed in 7eb1237 (REST) and 37ff2bd (the other catalogs).

REST: the action is no longer part of the protocol.

• It is removed from the SchemaChange JSON subtypes and from rest-catalog-open-api.yaml, which is back to master's version, so a REST server cannot parse it.
• RESTCatalog.alterTable rejects it before sending, with the same message the procedure gives.
• Tests: RESTApiJsonTest asserts that {"action":"enableDataEvolution"} does not parse. DataEvolutionEnablerTest#testRejectsRESTCatalog now also sends a direct alterTable on a REST table that has data: it is refused and the options stay unchanged.

Other catalogs: FileSystemSchemaManager.commitChanges accepts EnableDataEvolution on a table without row tracking only while every live data file has a row id. That means the table has no snapshot yet, or its latest snapshot is a row id commit of the procedure.

• Such a commit is marked with the snapshot property data-evolution.row-ids-assigned-snapshot-id, whose value is the snapshot's own id. A later snapshot that copies its base's properties therefore does not carry the mark, the same pattern as the reassign plan in [core] Persist row ID reassignment plans and mark snapshots #10008.
• SchemaManagerUtils still skips the immutable-option check for this change, but only after this precondition has been checked.
• Tests:
  • testSchemaChangeAloneIsRefusedWhileFilesHaveNoRowId: a table with data is refused and keeps its options.
  • testSchemaChangeAloneIsAcceptedOnTableWithoutSnapshot
  • testSchemaChangeNeedsTheLatestSnapshotToBeAMarkedOne: a snapshot that only copied the mark is refused.

[JingsongLi]

The use case has strong end-to-end value: converting an existing large append table in place avoids a full data rewrite and preserves its history, while enabling Data Evolution and global indexes. The normal read/split fallback and Spark/Flink procedure paths look coherent. I cannot recommend this conversion for production yet because its safety conditions are not enforced at all interleavings:

1. P1: An in-flight stale writer can commit after the procedure reports success. FileStoreCommitImpl.tryCommitOnce reads and checks the latest schema at lines 1047-1050, then does manifest work before the snapshot CAS at line 1315. If that writer pauses after the check, DataEvolutionEnabler can change the schema and finish its last repair scan (DataEvolutionEnabler.java:142-171); the writer then commits an old-schema file without firstRowId. Snapshot CAS checks the snapshot UUID, not the schema, and no subsequent repair occurs. This leaves an ID-less live file in a supposedly converted table. Coordinate the schema switch and snapshot commit or otherwise fence stale commits atomically; add a deterministic paused-writer test that releases the writer after the procedure's final scan.
2. P1: Rollback guards use cached options. A table/commit handle opened before conversion still has rowTrackingEnabled() == false, so AbstractFileStoreTable.checkRollbackKeepsRowTracking and FileStoreCommitImpl.rollbackToAsLatest can restore a pre-conversion snapshot after the latest persisted schema enabled row tracking. Akash3121 already reported the table rollback paths. Check the latest persisted schema in all rollback entry points and test snapshot, tag, and as-latest rollback through a retained old handle.
3. P2: The assignment snapshot can carry a stale schema ID. DataEvolutionEnabler.commitAssignment uses table.schema().id() at line 351. That schema is cached on the loaded table. A concurrent schema-only change does not move the snapshot CAS, so this commit can create a newer conversion snapshot referring to the older schema. Akash3121 already reported this; resolve the latest schema for each commit attempt and test an ADD COLUMN between planning and replacement.
4. P1: Public schema alteration bypasses assignment. SchemaChange.enableDataEvolution() is exposed through generic catalog/REST alteration and SchemaManagerUtils applies it without proving existing files have IDs. A client can enable both options on a populated table without the manifest rewrite, although the procedure itself rejects REST catalogs. Akash3121 already raised this. Restrict the action to a verified conversion path or reject direct alteration on nonempty tables; test direct REST/catalog alteration on populated data.

Production rollout also needs an explicit requirement to stop older Paimon writer binaries before conversion: the stale-writer guard exists only in this PR's runtime, so an older running binary will not enforce it after the finite repair loop.

Verification: RowTrackingEnableGuardTest (6/6), DataEvolutionFilesWithoutRowIdTest (5/5), RESTApiJsonTest (36/36), and the new SchemaManagerTest case (1/1) passed locally. The local enabler suite was blocked mainly by the existing CodeGenerator service-loading failure; one REST test hit this sandbox's bind restriction. The Flink SQL procedure IT failed at its initial write because of a local Avro DataFileWriter.setEncoder linkage error. The PR head's JDK 8/11, Flink 1, Spark, and E2E CI checks pass; Flink 2 CI is still running. These CI results do not exercise the unsafe interleavings above.

[zhuxiangyi]

@JingsongLi thanks for the review. All four points were real. I reproduced each one with a deterministic test before changing anything, and they are fixed in new commits on top of the branch. Akash's three inline threads are answered individually with the code references.

1. In-flight stale writer (P1), 37ff2bd
After the schema change, the procedure now commits a fence: an empty snapshot through the normal commit path. It then assigns row ids to anything that was committed before the fence.

A writer reads its base snapshot before it checks the schema. So a writer that passed the check on the old schema has only two outcomes:

• it committed before the fence, and the repair after the fence assigns ids to its files;
• its snapshot CAS loses to the fence, and its retry is refused by the schema check.

Tests pause a real writer right after its schema check, with a FileIO that blocks its first manifest write, and release it at three points:

• after the procedure returned: refused, and no file without row id is left;
• between the schema change and the fence: committed, then repaired;
• on an empty table, where the fence is the first snapshot: refused.

2. Rollback guards on cached options (P1), 35110ec
rollbackTo(snapshot), rollbackTo(tag) and rollbackToAsLatest now decide by the latest persisted schema. Tested through a table object that was loaded before the conversion.

3. Stale schema id on the row id snapshot (P2), 37ff2bd
Every row id commit attempt reads the latest schema right before the replacement.

4. Direct schema change (P1), 7eb1237 and 37ff2bd

• The action is out of the REST protocol, both the JSON subtypes and the OpenAPI spec, and RESTCatalog.alterTable rejects it.
• For the other catalogs, the schema manager accepts it only on a table without snapshot, or when the latest snapshot is a row id commit of the procedure. Such a commit is marked with its own snapshot id, so a copied mark does not count.
• If a writer commits between the row id commit and the schema change, the procedure assigns ids again and retries, bounded by the commit retry settings.

Older writer binaries: agreed, this cannot be enforced in code. The conversion section of the docs now requires stopping writers of older Paimon versions before converting.

One more bug, found while checking the tests (not raised in the review), ff3eeba
A plain append writer numbers its rows, so a file written in one commit of many rows has sequence numbers far above the snapshot ids of later commits. A data-evolution read takes each column from the file with the highest sequence number. After the conversion, a partial-column write or a MERGE INTO on such a file was therefore silently ignored; the small test tables hid it. Converted files now get the sequence number of the row id commit, like any row-tracking commit. New tests write a 20-row commit (core) and a 100-row commit (Spark UPDATE and MERGE INTO); both fail without the fix.

Also

• A run that finds the schema already switched by a concurrent run does not switch it again.
• 79606fa covers expiring the pre-conversion snapshots and deleting a pre-conversion tag: the data files that the conversion re-references through new manifests are kept.
• A conversion now commits two snapshots: the OVERWRITE snapshot that assigns the row ids, and the empty APPEND fence.

Verification

• Every fix was checked by mutation: reverting it makes at least one of the new tests fail.
• Locally on the final head:
  • 29 core suites (1006 tests: all DataEvolution*, RowTracking* and rollback suites, SchemaManagerTest, MockRESTCatalogTest and more);
  • EnableDataEvolutionProcedureITCase (Flink);
  • EnableDataEvolutionProcedureTest and RowTrackingTest (Spark 3.5).
• CI is running for the other Spark versions.

[JingsongLi]

Re-reviewed head 79606fa for production. The in-place conversion has strong end-to-end value. The new procedure fence, latest-schema rollback guard, schema-ID refresh, and REST protocol restriction address most of the earlier findings. Local JDK 8 conversion/rollback/read suites passed 42/42; the Spark, Flink and E2E CI jobs pass. Both Core CI jobs are red because S3FileIOTest could not pull its MinIO Docker image, so those jobs still need a rerun. git diff --check passed.

P1 — the public direct catalog action can still enable Data Evolution without a fence. FileSystemSchemaManager.checkEnableDataEvolution() expressly accepts SchemaChange.enableDataEvolution() when an empty table has no snapshot; DataEvolutionEnablerTest#testSchemaChangeAloneIsAcceptedOnTableWithoutSnapshot pins that path. Unlike DataEvolutionEnabler.run(), this direct catalog.alterTable() call does not commit the empty fence or repair a writer already in flight. I reproduced the race on this exact head with a temporary test using the existing PausedWriter hook: create an empty table; start an old-schema writer and pause it just after FileStoreCommitImpl.tryCommitOnce() checks the schema; call catalog.alterTable(TABLE, SchemaChange.enableDataEvolution(), false); release the writer. The alter succeeds, the writer commit succeeds, and the sole live file has firstRowId == null even though Data Evolution is enabled. The probe passed and was removed from the review checkout. The same gap exists if a direct alter is allowed on a marked assignment snapshot while an old writer is in flight.

Please make the generic action unavailable to external callers or ensure every allowed entry point performs the same fence-and-repair protocol before reporting success. Add the direct-alter in-flight-writer regression. The documented rollout must also require older writer binaries to stop, since they cannot enforce the new stale-writer check. This is a production merge blocker; keep the PR open for the fix.

[zhuxiangyi]

@JingsongLi thanks, confirmed with the same probe. Fixed by leaving the procedure as the only way to switch the options on:

• 08f1e1c: SchemaChange.enableDataEvolution() is removed from the public API. The change now has a private constructor inside DataEvolutionEnabler, so only the procedure (which always fences and repairs) can issue it. It still goes through catalog.alterTable, so catalog locks and metastore sync are kept.
• 1566bc1: the same race existed on master via ALTER TABLE SET ('row-tracking.enabled' = 'true') on a table without snapshots. That is now refused too (set it at CREATE TABLE or use the procedure). It is a separate commit because it changes behaviour.

Your probe is added as a regression test (testAlterTableWithWriterInFlightOnTableWithoutSnapshot). The requirement to stop older-version writers is in the docs and will be at the top of the PR description. The Core CI failures were the MinIO image pull in S3FileIOTest.

[zhuxiangyi]

Fixed three additional issues in 9f001773e:

• Sequence numbers: normalize row-tracking-only files without changing their row IDs or overriding DE updates committed before recovery.
• Stale writers/compactors: reject commits from writers loaded before DE was enabled, even if row tracking was already on.
• Interrupted conversion: only return Skipped after confirming a committed DE-schema snapshot has fenced old writers; otherwise finish the fence and repair, including for empty tables.

Added seven regression tests. 215 core tests passed, along with Checkstyle and Spotless. Spark/Flink suites were not rerun locally.

[JingsongLi]

Reviewed exact head 9f001773e373ddf40559c5f58671e4b146788308. The in-place conversion has strong end-to-end value, and the latest stale-handle, sequence-normalization and interrupted-conversion fixes pass their regressions. Two additional historical/recovery paths still break the converted table.

1. [P1] Guard the Data Evolution boundary when rolling back a row-tracking-only table. Both AbstractFileStoreTable.checkRollbackKeepsRowTracking and the FileStoreCommitImpl.rollbackToAsLatest guard only compare row-tracking.enabled. A snapshot from before DE conversion can already have row tracking enabled while retaining the incompatible row-count sequences that this PR now normalizes. In a real 20-row, two-file table, conversion changed the old files' max sequences from [1,19] to [1,1], and all 10 subsequent column updates were visible. rollbackTo(oldSnapshot) and rollbackTo("before") both succeeded under the current DE schema, restored [1,19], and a successful subsequent column-update commit applied 0/10 updates. rollbackToAsLatest also reproduced 0/10 after a real DE compaction replaced the original files; the same-file case was a passing negative control because it kept the normalized current metadata. Refuse the DE-disabled -> DE-enabled rollback boundary as well, or safely normalize the target metadata before allowing it. Please cover snapshot, tag and as-latest paths on a row-tracking-only source table.

2. [P1] Validate restored compaction files, not only the options of the committer handle. The new commit guard rejects a committer loaded before conversion, but a newly loaded DE committer can restore a checkpoint's compaction messages produced by the previous RT-only writer. I ran a real AppendCompactTask, serialized its pending ManifestCommittable, converted the table, applied a DE column update, then restored through filterAndCommitMultiple(..., true). The old COMPACT output has schema 0 and no firstRowId; restoration committed it and replaced the converted source files. Read results changed from (1,new-a,0), (2,b,1) to (NULL,new-a,0), (1,a,NULL), (2,b,NULL). Without the column update, both original rows simply lost their row IDs. Flink's RestoreCommittableStateManager.recover -> StoreCommitter.filterAndCommit uses this same path when a job is resubmitted with the converted schema; RestoreAndFailCommittableStateManager intentionally fails only after the restored commit, so that later failure does not undo the corruption. Reject incompatible restored compaction output before publication, or convert it with proven preservation of row identity. Please add a serialized-message recovery regression using a freshly loaded table/committer. An isolated guard rejecting the incompatible missing-row-ID COMPACT output preserved both updated rows in the same probe.

Validation: 139 focused Core tests passed with normal JDK 8 Maven checks; all 24 selected conversion/row-tracking/sub-field Spark tests passed on each of Spark 3.5.8 and Spark 4.1.2, including actual SQL conversion, update/merge/delete, historical reads and incremental BTree index use. The 6 procedure/action/streaming tests also passed on each of Flink 1.20.1 and Flink 2.2.0 after aligning the local reactor test classpath from Avro 1.11.3 to the Paimon compiler's 1.11.4; the initial mismatch failed before conversion. The additional actual-table probes above reproduce both P1s on this head. Exact-head CI is green. Older-version writers still must be stopped as documented; these findings also occur with this version's supported rollback/recovery APIs.

[zhuxiangyi]

@JingsongLi

Thanks for the detailed probes. Both issues reproduced as described and are fixed on de-enable-data-evolution (now 4ef4e2028, merged with the latest master).

1. Rollback across the DE boundary (ca833162c): rollbackTo(snapshot/tag) and rollbackToAsLatest now also refuse a target from before data evolution was enabled, not only one from before row tracking was enabled. Regression tests cover the snapshot, tag and as-latest paths (the as-latest one after a real DE compaction) on a row-tracking-only source table.

2. Restored compaction output (ca833162c): a data-evolution commit now refuses, before anything is published, any added file that would be left without row ids, so a pre-conversion COMPACT output restored by a freshly loaded committer is rejected. The regression test serializes a real AppendCompactTask message and restores it through filterAndCommitMultiple(..., true); rows and row ids stay intact.

While verifying I also fixed three related issues:

• Restored pre-conversion APPEND files are now stamped with the commit's snapshot id, so they no longer hide later column updates.
• Rollback to the conversion's fence snapshot is refused if it still holds files that were not repaired yet.
• Conversion is refused for tables whose files physically store _ROW_ID (left by a copy-on-write UPDATE/DELETE/MERGE INTO).

Flink and Spark procedure tests and the relevant core suites pass.

[JingsongLi]

Requirement fit: SUPPORTED. Metadata-only conversion has substantial value for existing append tables and global indexing. Implementation: FINDINGS: the latest rollback and restored-compaction fixes pass, but a current-version concurrent copy-on-write update can leave an unsupported converted table and the retry then reports Skipped. Validation: 148 formal core, 9 Spark SQL procedure and 6 Flink SQL procedure cases pass. A source-free actual file commit/conversion/read-planning probe exposes the missing interleaving (with a serial rejection control). Current code/E2E CI passes; licensing failed before build on Maven Central connection reset. Actual Spark copy-file controls also reproduce overlapping row-ID allocation, including silent loss of copied rows after a successful conversion.

[JingsongLi]

[P1] Reserve existing row-ID ranges before assigning missing IDs

A legal copy_files from a row-tracking source into an existing ordinary append target retains file.firstRowId, while that target still has nextRowId=0. The enabler preserves these existing IDs but assigns ordinary appended files starting at zero too. I reproduced this through the actual public Spark copy workflow (CopySchemaOperator/ListDataFilesOperator/CopyDataFilesOperator/CopyFilesCommitOperator, physical Parquet), then appended rows normally. Copy two tracked rows + append two ordinary rows reads four rows before conversion; conversion reports Success, but reads only the two appended rows afterwards because both files now own [0,2) and the newer sequence wins. Copy two + append one instead makes ordinary reads fail with overlapping-range error. The same public flow with a plain source is a passing control (all rows retained). Derive and validate allocation bounds from the live ranges, or reassign/reject retained IDs when converting a non-row-tracking table; do not assume its nextRowId accounted for copied file metadata. Include copies with unequal/equal row counts and copied-only tables.

[zhuxiangyi]

Fixed in bbadbc6. Missing row ids now start after the largest row id of the live files instead of the snapshot's nextRowId. A table whose files from before the conversion already have overlapping row ids is refused. Tests:

• DataEvolutionEnablerTest: copied rows only, copy 2 + append 2, and copy 2 + append 1 into an ordinary table. The copied rows keep row ids 0 and 1, the others continue after them, and so do rows written later. Files of two tables with the same row ids are refused.
• EnableDataEvolutionProcedureTest: your flow with the real sys.copy into an ordinary table, followed by a MERGE INTO by row id.

The root cause is in sys.copy: it commits the copied files with their row ids without advancing nextRowId, which leaves duplicate row ids in row-tracking targets regardless of this PR. That is fixed separately in #10391, which also shifts copied row ids that collide with rows a partial overwrite keeps, and orders sequence numbers on data-evolution targets. The two PRs are independent: since 64aa7ce the tests here build their state without relying on how sys.copy behaves, and they pass with and without #10391.

[JingsongLi]

[P1] Validate physical row-ID files before reporting conversion complete

A current-version row-tracking-only copy-on-write UPDATE/MERGE may commit after the initial check but before the schema switch. Its file stores physical _ROW_ID and has no firstRowId. The enabler switches DE on and commits its fence, then rejects this file, leaving the table on the DE schema. On retry, plan records it only in storingRowIds (then continues); hasChanges is false and this early Skipped return bypasses checkNoFileStoresRowIds. I reproduced this with the same real public BatchTableWrite/write-type and CommitMessage path as testRefusesFilesThatStoreTheirRowIds, injected between row-ID assignment and schema change: first run throws with DE=true/schema=1; second run returns Skipped; real DataEvolutionCompactCoordinator.plan still throws missing first row id; normal rollback to the pre-conversion snapshot is refused. The serial control correctly refuses before changing schema. This input is also produced by the normal Spark copy-on-write path, and docs only require stopping older-version writers. Coordinate these writes before the schema switch (or explicitly enforce a full writer pause), and always validate storingRowIds before the idempotent success check. Cover this interleaving and recovery, not just a file that exists at the initial scan.

[zhuxiangyi]

Fixed in bbadbc6. The check runs before the Skipped return, on every re-plan, and right before the schema switch. A copy-on-write commit that lands before the switch therefore fails the conversion and leaves the table unchanged (testCopyOnWriteDuringConversionIsRefusedBeforeTheSchemaSwitch).

The window that remains is a copy-on-write writer that passed its schema check before the switch and commits before the fence; closing it would take coordinating the writers. In that case every later run reports the files instead of Skipped until those rows are rewritten, after which the table is complete (testCopyOnWriteBeforeTheFenceIsReportedUntilTheRowsAreRewritten). The docs say not to run such statements during the conversion.

[JingsongLi]

Requirement fit: SUPPORTED. Implementation: FINDINGS. Rechecked bbadbc6: the previous copied-row-ID allocation and false Skipped-after-COW findings are addressed. 152 formal core tests and all 11 Spark SQL procedure tests pass; the previous 6 Flink procedure tests passed. The actual public Spark copy/conversion/append controls retain all rows for both equal and unequal file sizes. However, an additional actual public-copy/column-update probe finds a sequence normalization gap for a row-tracking source copied into an existing ordinary append target. Production migration still requires stopping old-version writers and COW UPDATE/DELETE/MERGE, and discarding pre-conversion pending compaction state as documented.

[JingsongLi]

[P1] Normalize copied tracked files rebound to a plain target schema

sys.copy_files permits a row-tracking source to be copied into an existing ordinary append target. It preserves firstRowId and the source sequence numbers, while rebinding file.schemaId to the target plain schema. This condition therefore excludes those files from resetSequences, and they already have a row ID so the missing-ID rewrite does not normalize them either. I reproduced this through the actual CopySchemaOperator/ListDataFilesOperator/CopyDataFilesOperator/CopyFilesCommitOperator with physical Parquet: after 10 real source overwrites, the one copied live row has firstRowId=9 and sequence=10. The plain target reads it correctly; conversion reports Success with nextRowId=10 at snapshot 3, but retains sequence=10. A column-v update to that row then commits successfully at snapshot 4 (sequence=4), yet actual reads still return old9 instead of NEW because the copied full-row file wins the sequence comparison. The same workflow into an existing row-tracking target is a passing control: conversion resets sequence to 1 and NEW is visible. Normalize retained complete-row files from every pre-DE schema, including these legal copied files, and keep the conversion-detection/fence helpers consistent; add a high-source-sequence plain-target copy followed by a real column-update regression test.

[zhuxiangyi]

Fixed in 8673cc1. Every file from a schema without data evolution now gets the baseline sequence number 1, whatever wrote it. That covers row-tracking-only writers, plain writers, and files copied in with the sequence numbers of another table. The same rule is used in all three places:

• the conversion plan;
• DataEvolutionUtils.needsDataEvolutionConversion, which the Skipped check and the rollback guard use;
• the stamping of restored pre-conversion appends in FileStoreCommitImpl.

Tests:

• testCopiedFilesWithHighSequenceNumbersGetTheBaselineSequence: a source overwritten ten times (first row id 9, sequence 10) is copied into an ordinary table and converted. A column update committed afterwards is visible, and a second run reports Skipped.
• The same flow in Spark with the real sys.copy, followed by UPDATE and MERGE INTO.

[zhuxiangyi]

@JingsongLi thanks for re-checking. All three points are answered in their threads:

• Copied row-id ranges and physical row-id files: fixed in bbadbc6, as you confirmed. 64aa7ce rebuilds the copied-file tests the way you asked: an ordinary target with copied rows only, copy 2 + append 2, and copy 2 + append 1.
• Sequence numbers of copied files on a plain target: fixed in 8673cc1. Every file from a schema without data evolution gets the baseline sequence 1. Your copy + column-update flow is a regression test in core and in Spark with the real sys.copy.

Separate PR for sys.copy: #10391. Two of your findings come from sys.copy itself, not from the conversion. It commits the copied files with the firstRowId and sequence numbers of the source. That breaks row-tracking and data-evolution targets even without converting anything:

1. it does not advance nextRowId, so rows written after a copy get the copied rows' row ids again: duplicate _ROW_IDs;
2. a partial overwrite (dynamic partition overwrite, or a where filter) brings copied row ids that collide with the rows of the partitions it keeps;
3. on a data-evolution target, the copied sequence numbers can exceed the target's snapshot ids, so later UPDATE / MERGE INTO / column updates are silently ignored.

#10391 fixes these in sys.copy: it advances nextRowId, shifts colliding copied row ids, maps the sequence numbers in order, and refuses files that store their row ids on row-tracking targets.

This PR still has to handle copied files at conversion time, for two reasons:

• your probes copy into an ordinary table, which keeps no row ids or nextRowId, so sys.copy has nothing to fix there;
• tables copied before [core][spark] Fix row ids and sequence numbers of files copied by sys.copy #10391 already hold such files.

The two PRs are independent. The tests here build their state without relying on how sys.copy behaves, and they pass with and without #10391.

[JingsongLi]

This is a reasonable requirement, but it is the wrong approach; the content of Data Evolution should not be modified, as doing so makes the code within it difficult to understand.

[JingsongLi]

Additionally, please split the PR. This is only a draft. And please share your production env, I need more information not just from AI.

[zhuxiangyi]

@JingsongLi thanks, agreed. The conversion should not make the Data Evolution code harder to understand, and this PR changes too much of the existing Data Evolution core logic for that.

Our use case: in production we have a very large wide append table that needed to become a Data Evolution table. The launch was urgent and there was no way to convert it in place, so we had to rewrite all of its data into a new Data Evolution table. A metadata-only conversion would have avoided that rewrite.

I'll redesign it so that the conversion stays outside the existing Data Evolution read/write/commit paths as much as possible, and split it into small PRs that are easy to review. I'll keep this PR as a draft until then.