zhuxiangyi opened a new issue, #10592: URL: https://github.com/apache/paimon/issues/10592
### Search before asking - [X] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar. ### Motivation In production we have a very large wide append table that needed to become a Data Evolution table. There is no way to convert an existing append table 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, and the same applies to any existing append table that wants Data Evolution or global indexes. #10098 implemented such a conversion, but it changed the existing Data Evolution read, scan, commit and rollback code to support converting while writers are running. The review concluded that the requirement is reasonable but the approach is wrong: the content of Data Evolution should not be modified, and the work should be split into small PRs. This issue proposes a redesign for agreement before writing code. ### Solution A converted table is indistinguishable from a table created with Data Evolution, so the Data Evolution read, write, scan, commit and compaction code does not change. **Scope of the first version** - Plain append tables only (no row tracking yet). Row-tracking-only tables are refused and left for later. - Offline: all writers are stopped first (any version, including copy-on-write UPDATE/DELETE/MERGE and compaction jobs). Flink jobs are stopped with a savepoint. - Not for REST catalogs. **How the conversion works** All logic lives in a new `DataEvolutionEnabler`, written like the existing `reassign_row_id` (metadata-only, committed with `FileStoreCommitImpl.replaceManifestList`). 1. **Validate**: append table, Data Evolution off, the target schema passes `SchemaValidation`, no file stores `_ROW_ID`. 2. **Plan on the latest snapshot N**: give every live file a contiguous `firstRowId`, set every file's sequence number to N+1, compute `nextRowId`. 3. **Commit snapshot N+1** with the rewritten base manifests and an empty delta. The schema is still plain, so plain reads and writes ignore `firstRowId`. If the latest snapshot is no longer N, fail and ask to stop the writers. 4. **Switch the schema** to `row-tracking.enabled` + `data-evolution.enabled` through an internal schema-level API (not `SchemaChange`, not REST; `ALTER TABLE` still refuses). Inside the catalog's table lock it first checks that the latest snapshot is still N+1, so with a catalog lock the switch cannot interleave with another snapshot commit. 5. **Re-check**: files without a row id (a writer that slipped in between) go through steps 2–3 again. | State | Schema | Files | Undo | | --- | --- | --- | --- | | Before | plain | no row ids | — | | After step 3 | plain | row ids, sequence N+1, `nextRowId` | re-run to continue, or abort: roll back to N | | After step 4 | row tracking + Data Evolution | same | abort while nothing was written since: write a schema with both options off | The procedure is re-runnable: it only touches files written under a schema without Data Evolution and returns `Skipped` when there is nothing to do. `mode => 'abort'` undoes the conversion as long as nothing has been written after it. `dry_run => true` reports what would happen without converting. **What moves from code to documented preconditions** The guards #10098 added to the commit, scan and rollback paths are dropped. Violating a precondition makes reads fail loudly in `nonNullFirstRowId` instead of returning wrong rows, and each case has a recovery. | Earlier finding in #10098 | New handling | | --- | --- | | In-flight writer commits after the switch | Precondition: stop all writers. If violated, reads fail loudly; re-running the procedure repairs it. No commit-path change. | | Rollback across the conversion | Plain tables: reads fail loudly after such a rollback; re-running repairs it. Row-tracking-only tables, where this would be silent, are out of scope. | | Restored pre-conversion compaction | Precondition, plus a small generic guard: before committing restored committables, `TableCommitImpl.filterAndCommitMultiple` (next to `checkFilesExistence`) rejects files whose schema differs from the table's in row tracking / Data Evolution. It contains nothing conversion- or DE-specific, and it also catches restored pre-conversion appends, whose high sequence numbers would otherwise hide later column updates silently. | | Copied row ids and sequence numbers | A plain table's `firstRowId`s carry no meaning, so all row ids and sequence numbers are reassigned. | | Stale schema id on the row-id snapshot | Kept: the latest schema id is read on every commit attempt of the enabler. | | Direct schema change bypassing the conversion | Kept: the switch is an internal schema-level API only, not in the public `SchemaChange` API or REST. | **Keeping snapshots and tags from before the conversion readable** Many tables create tags automatically, so historical tags must stay readable. Today `AbstractFileStoreTable.tryTimeTravel` takes the columns from the snapshot's schema but mostly the current options; only column storage declarations such as `blob-field` and `vector-field` are restored from the historical schema in `excludeCurrentSchemaFieldOptions`. After the conversion, reading snapshot N or earlier would go through the Data Evolution read and fail on files without row ids. Proposal: time travel also restores `row-tracking.enabled` and `data-evolution.enabled` from the historical schema (unless set explicitly as dynamic options), so snapshots and tags from before the conversion are read as the plain append table they were. Both options were immutable until now, so historical and current values are always equal on existing tables and nothing changes for them. | | Rewrite tags | Fallback in the DE read | Time travel keeps historical settings (proposed) | | --- | --- | --- | --- | | How | Give the manifests of every affected tag row ids and overwrite the tag files | The DE read treats files without row ids as plain files | Both switches take the values of the snapshot's own schema | | Where | Enabler | DE scan, split generation, split read | `AbstractFileStoreTable.tryTimeTravel` | | Tags from before | readable, with row ids | readable, `_ROW_ID` NULL | readable as a plain table | | Plain snapshots from before | not readable (snapshot files are immutable) | readable | readable | | Effect on existing tables | none | changes the DE read | none | | Worst case | a manifest deleted while still used: `TagManager.doClean` only checks the two neighbours of a deleted tag | affects every DE table's read path | reading an old snapshot fails | Open points: streaming reads starting before the conversion (`tryTimeTravel` also serves them; either restore the historical settings for batch reads only, or refuse such streams), incremental reads across the conversion, and consumers whose progress is before it (listed by `dry_run`). **PR plan** Each PR is reviewable on its own; they merge in this order. | Order | PR | Content | Depends on | | --- | --- | --- | --- | | 1 | Time travel keeps historical settings | Restore the row tracking / Data Evolution switches from the historical schema in `tryTimeTravel`, and handle streaming reads | — | | 1 | Schema API | Internal schema-level API to switch on row tracking + Data Evolution | — | | 2 | Enabler | `DataEvolutionEnabler` for plain append tables, with re-run, abort and dry run | Schema API | | 3 | Restore guard | Generic check in `TableCommitImpl.filterAndCommitMultiple`, with a serialized-message regression test | none in code | | 4 | Spark | `sys.enable_data_evolution` procedure + docs | Enabler; time travel and restore guard merged | | 4 | Flink | Procedure / action + docs | same | | any | Separate fix | Race of `ALTER TABLE SET ('row-tracking.enabled' = 'true')` on a table without snapshots, which exists on master today | — | ### Anything else? Questions for the reviewers: 1. Does this direction work, in particular relying on the documented offline precondition instead of guards in the commit path? 2. Is it acceptable that time travel reads snapshots and tags from before the conversion with their historical row tracking / Data Evolution settings? Previous attempt and review: #10098. ### Are you willing to submit a PR? - [X] I'm willing to submit a PR! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
