Skip to main content

Data Evolution

Overview​

Paimon supports schema evolution, allowing you to add, modify, or delete column schema. Data Evolution mode extends this for append tables by allowing partial column updates without rewriting entire data files. Updated column data is written to new files and merged with the original data during reads.

Data Evolution mode offers the following advantages:

  • Efficient partial-column updates: update a subset of columns and avoid the I/O cost of rewriting untouched columns.
  • Reduced file rewrites: append new column data to dedicated files when backfilling or evolving data.
  • Delete support: record row deletions with deletion vectors without rewriting existing column files.
  • Optimized reads: merge original and updated column files at read time.

To enable Data Evolution, create an append table with both row-tracking.enabled and data-evolution.enabled set to true.

CREATE TABLE target_table (id INT, b INT, c INT) TBLPROPERTIES (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true'
);

INSERT INTO target_table VALUES (1, 1, 1), (2, 2, 2);

Partial Updates​

You can update selected columns with Spark SQL UPDATE or MERGE INTO, the Flink data_evolution_merge_into procedure, or the PyPaimon table update API. Only the updated column files are written; untouched columns remain in their original files.

UPDATE target_table SET b = b + 10 WHERE id = 1;

CREATE TABLE source_table (id INT, b INT);
INSERT INTO source_table VALUES (1, 11), (2, 22);

MERGE INTO target_table AS t
USING source_table AS s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.b = s.b;

SELECT * FROM target_table;
+----+----+----+
| id | b  | c  |
+----+----+----+
| 1  | 11 | 1  |
| 2  | 22 | 2  |
+----+----+----+

Notes:

  • Spark SQL supports standalone UPDATE statements for Data Evolution tables.
  • Concurrent Spark SQL UPDATE statements that update the same data file and columns may be retried automatically. Configure retry attempts with spark.paimon.write.data-evolution.update-conflict-retry.max-attempts and retry wait time with spark.paimon.write.data-evolution.update-conflict-retry.wait-ms.
  • If concurrent compaction changes row-ID file boundaries after Spark SQL MERGE INTO stages regular partial-column files, Spark rebases those staged files onto the latest boundaries before committing instead of rerunning the MERGE source and join. Spark performs this rewrite with distributed DataFrame processing, so the PyPaimon-only data-evolution.row-id-conflict-rewrite.max-size limit does not apply. Recovery does not apply when deletion vectors are enabled or when existing-row BLOB or VECTOR files are staged, and it does not hide logical concurrent-update conflicts.
  • In Spark SQL, MERGE INTO supports WHEN NOT MATCHED BY SOURCE for delete actions on Data Evolution tables.
  • The Flink data_evolution_merge_into procedure currently supports updating or inserting columns, but not inserting new rows.

Deletes​

Data Evolution tables can use deletion vectors to record deleted rows without rewriting existing column files. To write deletes, enable deletion vectors together with row tracking and Data Evolution:

warning

Deleting rows with deletion vectors does not physically remove the original data immediately. Data Evolution compaction preserves row IDs and logical deletions, so it does not materialize deleted rows. The legacy data-evolution.compaction.rewrite-row-ids option is no longer supported. Use the materialize_deletion_vectors procedure to apply deletion vectors to the latest table state and replace the affected files. Historical snapshots and tags may still reference the replaced files; reclaiming their storage requires those references to expire and snapshot expiration to remove the files.

Both Flink and Spark expose the same procedure:

CALL sys.materialize_deletion_vectors(`table` => 'default.target_table');

The procedure can limit the rewrite to selected partitions. partitions and where cannot be used together:

CALL sys.materialize_deletion_vectors(
    `table` => 'default.target_table',
    partitions => 'dt=2026-08-12');

CALL sys.materialize_deletion_vectors(
    `table` => 'default.target_table',
    `where` => 'dt >= 20260801');

Materialization rewrites the affected data, assigns new row IDs to surviving rows, removes the applied deletion vectors, and drops affected global indexes. Consumers which persist _ROW_ID values must therefore coordinate with this operation. Concurrent changes to affected row-ID ranges cause the procedure to fail instead of committing stale results. Materialization of vector-store files is not currently supported.

Spark processes bounded batches until all matching deletion vectors have been materialized. To keep a Flink job bounded, one Flink procedure invocation processes one batch with a soft target of 100,000 deletion vectors. An overlapping row-ID component is never split and can exceed the target. Invoke the Flink procedure repeatedly until an invocation makes no changes.

CREATE TABLE target_table (id INT, b INT, c INT) TBLPROPERTIES (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true',
    'deletion-vectors.enabled' = 'true'
);

Spark SQL supports DELETE FROM for Data Evolution tables:

DELETE FROM target_table WHERE id = 1;

Spark SQL also supports delete actions in MERGE INTO:

CREATE TABLE source_table (id INT, op STRING);
INSERT INTO source_table VALUES (2, 'delete'), (3, 'update');

MERGE INTO target_table AS t
USING source_table AS s
ON t.id = s.id
WHEN MATCHED AND s.op = 'delete' THEN DELETE
WHEN MATCHED AND s.op = 'update' THEN UPDATE SET t.b = t.b + 10
WHEN NOT MATCHED BY SOURCE AND t.id > 10 THEN DELETE;

The WHEN NOT MATCHED BY SOURCE clause requires Spark 3.4 or later.

With Flink 1.17 or later, SQL supports DELETE FROM for Data Evolution tables in batch mode. For row-level predicates, matching rows are recorded in deletion vectors, so data files are not rewritten:

DELETE FROM target_table WHERE id = 1;

Flink users can also submit the delete action.

Self Updates​

Self updates transform existing column values in place. In Flink SQL, create a temporary source view from the $row_tracking system table and join by _ROW_ID. In PyPaimon, use the shard scan + rewrite workflow to read existing values and write the derived columns back.

CREATE TEMPORARY VIEW source_view AS
SELECT _ROW_ID, b + c AS b
FROM default.target_table$row_tracking;

CALL sys.data_evolution_merge_into(
    'default.target_table',
    'TempT',
    '',
    'source_view',
    'TempT._ROW_ID=source_view._ROW_ID',
    'b=source_view.b',
    2
);

Self-update notes:

  • The source and target table name cannot be the same in the Flink procedure. Create a temporary view as the source.
  • Use view._ROW_ID = source._ROW_ID to identify the self-merge pattern in Flink.
  • _ROW_ID is only available via the $row_tracking system table in SQL.
  • Self-merge only supports WHEN MATCHED THEN UPDATE semantics.

File Group Spec​

Through the row-id metadata, files are organized into file groups.

When writing, the Data Evolution update path writes only the specified updated columns to new files. The original data files remain unchanged.

When reading, Paimon reads both the original data files and the new files containing updated column data, then merges files with the same first row id to present a unified view of the table.

After writing, files in target_table are organized as below:

When reading, files with the same first row id are merged:

The advantages of this mode are:

  • Avoid rewriting the whole file when updating partial columns, reducing I/O cost.
  • Keep read performance efficient through optimized merge processing.
  • Use disk space more efficiently because only updated columns are written to new files.