Skip to main content

Data Evolution

Overview​

Data Evolution lets an append table update selected columns while keeping the files for untouched columns. A logical row can therefore be assembled from several physical files: its scalar fields from normal data files, its payload from BLOB files, and its embedding from dedicated vector files. Paimon aligns these pieces by row ID, and uses file versions to select the current values.

This is useful when a pipeline ingests data once and enriches it repeatedly, for example by adding captions, labels, or embeddings to images. Schema evolution changes which columns exist; Data Evolution writes or replaces their values for existing rows. Adding a column alone does not backfill it.

The unit of a normal partial-column write is the selected columns over a complete file row-ID range. Updating one row can still rewrite those columns for every row in its file group. The saving comes from leaving the other columns, especially large payloads, untouched. Repeated updates add files and read work until compaction consolidates them.

Create a Table​

Set both row-tracking.enabled and data-evolution.enabled when creating the table. These options are immutable. Data Evolution requires an append table with no primary key and bucket = -1 (the default for an append table). clustering.incremental cannot be enabled with Data Evolution.

The following examples assume a configured Paimon catalog and the default database. Deletion vectors are enabled so the same table can also run the delete examples; they are optional if row-level deletes are not needed.

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

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

A business column such as id is not a primary key. Repeated INSERT statements append new rows, even when business-key values repeat. UPDATE and MERGE INTO use their predicates to find rows and use Paimon's _ROW_ID internally to write the changes.

For multimodal columns, see BLOB and Vector for their type declarations and storage options. Dedicated BLOB/vector storage needs normal columns alongside it. A partial-column file containing ordinary columns is still a normal file; “dedicated” refers to the storage format, not to whether the file was created by an update.

File Group Spec​

Normal files for the same group cover the same complete row-ID range. A dedicated BLOB or vector file must fit inside one normal range. Readers select the newest provider for each requested column, then combine values at matching row positions.

See File Layout and Reads for range diagrams, column-version selection, and the distinction between an omitted column, explicit NULL, and a BLOB placeholder.

Partial Updates​

For a normal-column update, the write path preserves the file contract through the following steps:

  1. Pin a target snapshot, find the matching row IDs, and map them to normal file ranges in that snapshot.
  2. For each affected range, produce the selected columns in row-ID order: new values for matched rows and existing values for unmatched rows. Keep deleted positions aligned as well.
  3. Write a normal partial file with the original range's firstRowId and rowCount. Normal-file rolling is disabled for this write, because a smaller output would violate the equal-range rule.
  4. Commit the new files atomically with conflict checks. Readers of the new snapshot see the updated column versions; older snapshots keep their previous view.

The engine and update APIs perform this alignment. Applications specify the matching rows and values; they do not need to construct file groups themselves.

Updating b for one matched row writes a new b-only file for all three row positions, preserving b for unmatched rows and leaving the original file unchanged.

Here id = 20 selects only row ID 1, but N1 still contains three values of b. The row-ID labels show alignment; they are not extra business columns.

Spark SQL​

Spark supports UPDATE expressions and MERGE INTO. For example:

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

CREATE TABLE source_table (id INT, b INT, c INT);
INSERT INTO source_table VALUES (1, 100, 1000), (3, 300, 3000);

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
WHEN NOT MATCHED THEN INSERT (id, b, c) VALUES (s.id, s.b, s.c);

SELECT * FROM target_table ORDER BY id;
-- (1, 100, 10), (2, 2, 20), (3, 300, 3000)

The matched update writes b over the affected existing normal ranges and keeps their row IDs. The unmatched insert writes a new row and receives a new row ID. Assign only the columns you intend to change; assigning every column can remove the I/O benefit of a partial update. Partition columns cannot be changed by a matched partial update.

Use the data_evolution_merge_into procedure for matched updates. It runs in batch mode and can populate an existing column, including a newly added column, but does not append unmatched source rows or perform deletes. The target must already contain data.

CREATE TABLE source_table (id INT, b INT);
INSERT INTO source_table VALUES (1, 100), (2, 200);

-- Wait for the source insert to finish before calling the procedure.
CALL sys.data_evolution_merge_into(
    'default.target_table', -- target table
    't',                    -- target alias
    '',                     -- optional source SQL statements
    'source_table',          -- source table or view
    't.id=source_table.id',  -- match condition
    'b=source_table.b',      -- columns and expressions to write
    2                       -- sink parallelism
);

Use '' for unused optional positional arguments. Updating partition columns is not supported. For BLOB updates, this Flink procedure supports descriptor-only and BLOB view fields; it rejects raw-data BLOB fields, including BLOB arrays and maps.

Python API​

PyPaimon returns commit messages from its update APIs. Commit them to make the change visible:

table = catalog.get_table("default.target_table")
builder = table.new_batch_write_builder()
update = builder.new_update()
predicate = update.new_predicate_builder().is_in("id", [1, 2])
messages = update.update_by_predicate(predicate, {"b": 100})

commit = builder.new_commit()
try:
    commit.commit(messages)
finally:
    commit.close()

For expressions, row-ID updates, business-key upserts, merge_into, and stream writer lifecycles, see PyPaimon Data Evolution. These APIs handle file alignment; writing an arbitrary small normal file and assigning it an existing row ID does not satisfy the file contract.

Self Updates​

A self update derives new values from the table's current rows. It is also the usual backfill workflow: first add a nullable column, then compute its values. The examples in this section are alternatives to run after creating the table.

ALTER TABLE target_table ADD COLUMN total INT;

UPDATE target_table SET total = b + c;

SELECT id, total FROM target_table ORDER BY id;

This writes total for the existing ranges. It leaves id, b, and c in their existing column files.

For explicit row-ID inspection in SQL, query the system table:

SELECT _ROW_ID, id, b, c FROM default.`target_table$row_tracking`;

Obtain row IDs from the table instead of deriving them from a business key or from result order. Ordinary updates and compaction preserve them, but maintenance that reassigns row IDs does not.

Nested Fields​

By default, the physical update unit is a top-level column, including a whole ROW/STRUCT column. To write only direct struct subfields, persist data-evolution.nested-field.enabled = true on the table. Written-column metadata can then contain paths such as payload.caption, and readers select the latest provider for the corresponding subfields. Row alignment still covers the full normal-file range.

The current implementation supports composing one level of struct subfields across files. It does not support splitting a deeper sub-struct across multiple files. Engine support for assignment syntax also matters; for example, the Flink procedure requires the option for a subfield assignment.

warning

Upgrade every reader, writer, compactor, and maintenance job before enabling nested-field evolution. Enable it through a persisted table-option change; dynamic overrides and later disabling or removing the option are unsupported. Older components cannot reconstruct files containing nested write paths, so a binary downgrade is unsafe once these files have been committed.

Deletes​

Row-level deletes require deletion-vectors.enabled = true. A deletion vector records deleted positions against a normal anchor file in the group (the oldest normal file, ordered by version and then file name). Readers apply these deletions to all column providers for the same logical rows. Deleting a row therefore hides its scalar, BLOB, and vector values together without rewriting those data files.

Spark SQL supports:

DELETE FROM target_table WHERE id = 1;

Spark can also combine update and delete clauses in MERGE INTO:

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

MERGE INTO target_table AS t
USING delete_source 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;

WHEN NOT MATCHED BY SOURCE requires Spark 3.4 or later.

Flink 1.17 or later supports DELETE FROM for Data Evolution tables in batch mode. Flink also provides the delete action. PyPaimon provides delete_by_predicate and delete_by_row_id; see its delete examples. Partition-only deletes can use a partition-removal path instead of row-level deletion vectors.

Deleting a row leaves its position in the file range occupied but invisible. For example, deleting row ID 2 from [0, 5] leaves the metadata range [0, 5], with five visible rows. An update must not close that gap or shift rows 3–5.

Concurrent Writes​

Updates are planned against a snapshot and checked when committing. Paimon checks both the layout contract and whether intervening writes conflict with the staged changes.

  • A new normal partial file must exactly match a current normal range; an existing-row dedicated file must be contained within one current normal range. A removed row range cannot be updated using stale metadata.
  • Logical update conflicts consider both overlapping row ranges and written field IDs. Updating different columns can avoid this conflict. Updating the same column in the same normal range can conflict even when the predicates selected different rows, because each staged file covers the full range.
  • A concurrent compaction can change boundaries while preserving the row IDs. This is a layout conflict and is distinct from another writer changing the same column values.

Spark UPDATE can retry update conflicts. Configure it with spark.paimon.write.data-evolution.update-conflict-retry.max-attempts and spark.paimon.write.data-evolution.update-conflict-retry.wait-ms.

Spark MERGE INTO can rebase staged normal partial-column files onto the latest boundaries after concurrent compaction, using distributed processing without rerunning the source and join. PyPaimon has similar recovery bounded by data-evolution.row-id-conflict-rewrite.max-size (default 256 MB, 0 B disables it), measured over the affected current data files. This size limit does not apply to Spark.

This boundary recovery is not attempted when deletion vectors are enabled or when the commit contains existing-row BLOB or VECTOR staged files. It does not suppress logical concurrent-update conflicts. If recovery is unavailable, re-plan the operation against the latest snapshot.

Compaction and Row-ID Lifetime​

Repeated updates add column files. Ordinary compaction consolidates them while preserving row IDs and logical deletions. Deletion-vector materialization removes deleted positions and assigns new row IDs to surviving rows in rewritten ranges. Persisted row-ID references must be coordinated with this maintenance.

See Data Evolution Maintenance for commands, large-file splitting, limitations, and the row-ID lifetime table.

File Sizing and Metadata Compatibility​

Choose file sizes with future partial-update costs in mind. See the metadata reference before enabling compact write-column encoding, and use the $files query to inspect alignment.