Data evolution and row tracking

Row tracking

Row tracking assigns a globally unique row identifier (_ROW_ID) to each row in a Paimon append table. RowIds are generated automatically when data is written and stay with the row throughout its lifecycle, providing the foundation for data evolution.

Data evolution

Table schemas in data lakes change frequently as business requirements evolve and AI models are upgraded. Traditionally, adding a column and backfilling its values requires rewriting entire data files, which causes significant I/O and compute overhead.

Data evolution eliminates this cost by letting you update specific columns in an append table without rewriting existing data files. New column data is stored in separate files and merged automatically by RowId at read time, so queries always see a unified view. Data evolution is also a building block for multimodal features such as BLOB storage.

Key benefits:

• Efficient partial column updates: Writes only the changed columns, eliminating the I/O amplification of full-file rewrites.

• Storage-friendly: Leaves original files untouched and appends new column data to separate files.

• Seamless reads: Merges columns by RowId at read time, with no impact on query performance.

Create a table

CREATE TABLE my_db.target_table (
    id INT NOT NULL,
    name STRING,
    age INT
) TBLPROPERTIES (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true'
);

INSERT INTO my_db.target_table VALUES (1, 'Alice', 30), (2, 'Bob', 25);

Partial column update (MERGE INTO)

Use MERGE INTO to update specific columns. In data evolution mode, only the changed columns are written to new files, and the original data files stay in place.

CREATE TABLE my_db.source_table (
  id INT NOT NULL, 
  age INT
);
INSERT INTO my_db.source_table VALUES (1, 31), (2, 26), (3, 28);

MERGE INTO my_db.target_table AS t
USING my_db.source_table AS s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.age = s.age
WHEN NOT MATCHED THEN INSERT (id, name, age) VALUES (s.id, '', s.age);

SELECT * FROM my_db.target_table;

+----+-------+-----+
| id | name  | age |
+----+-------+-----+
| 1  | Alice | 31  |
| 2  | Bob   | 26  |
| 3  |       | 28  |
+----+-------+-----+

After the operation, only the age column is written to a new file. The original files for id and name are unchanged, and the engine merges them automatically at query time.

Prepare data

import pyarrow as pa
from pypaimon import CatalogFactory, Schema
from pypaimon.schema.data_types import AtomicType, DataField

# Connect to a DLF catalog
catalog = CatalogFactory.create({
    'metastore': 'rest',
    'uri': 'https://${regionId}-vpc.dlf.aliyuncs.com',
    'warehouse': 'my_catalog',
    'token.provider': 'dlf',
    'dlf.access-key-id': 'ak',
    'dlf.access-key-secret': 'sk',
    # Configuration for public network access
   # 'dlf.signing-algorithm': 'openapi',                   # Set to 'openapi' for a public endpoint
   # 'dlf.oss-endpoint': 'https://oss-${regionId}.aliyuncs.com',  # Public OSS endpoint
})

# Create a table
schema = Schema(
    fields=[
        DataField(0, 'id', AtomicType('INT NOT NULL')),
        DataField(1, 'name', AtomicType('STRING')),
        DataField(2, 'age', AtomicType('INT')),
    ],
    options={
        'row-tracking.enabled': 'true',
        'data-evolution.enabled': 'true',
    },
)
catalog.create_table('my_db.target_table', schema, True)
table = catalog.get_table('my_db.target_table')

# Write initial data
write_builder = table.new_batch_write_builder()
write = write_builder.new_write()
commit = write_builder.new_commit()

init_data = pa.RecordBatch.from_pydict({
    'id': [1, 2],
    'name': ['Alice', 'Bob'],
    'age': [30, 25],
}, schema=pa.schema([
    ('id', pa.int32()),
    ('name', pa.string()),
    ('age', pa.int32()),
]))

write.write_arrow_batch(init_data)
cmts = write.prepare_commit()
commit.commit(cmts)
commit.close()

# Verify the write operation
read_builder = table.new_read_builder()
scanner = read_builder.new_scan()
reader = read_builder.new_read()
df = reader.to_arrow(scanner.plan().splits()).to_pandas()
print(df)
#    id   name  age
# 0   1  Alice   30
# 1   2    Bob   25

Update columns by RowId

Read the _ROW_ID along with the data, compute the new value in Python, and write it back.

write_builder = table.new_batch_write_builder()
table_update = write_builder.new_update().with_update_type(['age'])
table_commit = write_builder.new_commit()

update_data = pa.Table.from_pydict({
    '_ROW_ID': [0, 1],
    'age': [31, 26],
}, schema=pa.schema([
    ('_ROW_ID', pa.int64()),
    ('age', pa.int32()),
]))

cmts = table_update.update_by_arrow_with_row_id(update_data)
table_commit.commit(cmts)
table_commit.close()

# Verify the update
reader = table.new_read_builder()
print(reader.new_read().to_pandas(reader.new_scan().plan().splits()))
#    id   name  age
# 0   1  Alice   31
# 1   2    Bob   26

Upsert by business key

Update or insert records by matching on a business key, without managing _ROW_ID yourself.

write_builder = table.new_batch_write_builder()
table_update = write_builder.new_update()
table_commit = write_builder.new_commit()

upsert_data = pa.Table.from_pydict({
    'id': [1, 3],
    'name': ['Alice', 'Charlie'],
    'age': [35, 28],
}, schema=pa.schema([
    ('id', pa.int32()),
    ('name', pa.string()),
    ('age', pa.int32()),
]))

# Match by the id column: update if id=1, insert if id=3
cmts = table_update.upsert_by_arrow_with_key(upsert_data, upsert_keys=['id'])
table_commit.commit(cmts)
table_commit.close()

# Verify the upsert operation
reader = table.new_read_builder()
print(reader.new_read().to_pandas(reader.new_scan().plan().splits()))
#    id    name  age
# 0   1   Alice   35
# 1   2     Bob   26
# 2   3  Charlie   28

Shard update

Shard update is designed for large-scale column computation, such as backfilling AI features. Read data by shard, compute the new column values, and write them back.

from pypaimon.schema.schema_change import SchemaChange
from pypaimon.schema.data_types import ArrayType, AtomicType

# Add the embedding column
catalog.alter_table('my_db.target_table', [
    SchemaChange.add_column('embedding', ArrayType(True, AtomicType('FLOAT')))
])
table = catalog.get_table('my_db.target_table')  # Get the table again (schema has been updated)

write_builder = table.new_batch_write_builder()
update = write_builder.new_update()
update.with_read_projection(['name'])  # Columns to read
update.with_update_type(['embedding'])  # Columns to write

# Single-shard mode (multi-shard parallelism can be configured)
upd = update.new_shard_updator(shard_num=0, total_shard_count=1)
reader = upd.arrow_reader()

for batch in reader:
    names = batch.column('name').to_pylist()
    # Generate embeddings using a model (example only)
    embeddings = [[float(i) * 0.1] * 4 for i in range(len(names))]
    upd.update_by_arrow_batch(
        pa.RecordBatch.from_pydict(
            {'embedding': embeddings},
            schema=pa.schema([('embedding', pa.list_(pa.float32()))])
        )
    )

commit_messages = upd.prepare_commit()
commit = write_builder.new_commit()
commit.commit(commit_messages)
commit.close()

# Verify the shard update
reader = table.new_read_builder()
print(reader.new_read().to_pandas(reader.new_scan().plan().splits()))