Lakehouse – Phase 6 Schema Evolution
In a traditional database, changing a column’s name is fast because the engine controls the storage format. In a lakehouse, data lives in immutable Parquet files. You cannot edit a Parquet file — it is a sealed, self-describing unit. Rewriting petabytes of files just to add a column would take hours. Schema evolution solves this: change the schema in metadata without touching the data files. Old files still have the old column set; new files have the new columns. The reader reconciles the difference using field IDs — immutable integer identifiers assigned to each column at creation time.
Here is the roadmap for the phases to come:
- Phase 6: Schema evolution
- Phase 7: Deletes and updates
- Phase 8: Storage and access
Full Source Code
The code referenced in this post can be found in https://gitlab.com/kimserey.lam/lake-learn.
Field IDs: The Immutable Identity
Every column gets a unique integer field ID (field_id in ducklake_column) at creation time. This ID never changes for the lifetime of the column. Field IDs are written into Parquet files as schema metadata, and they are what the reader uses to map file columns to catalog columns.
1
2
3
4
5
6
7
8
9
CREATE TABLE orders (id INT, name VARCHAR, amount FLOAT);
ducklake_column:
field_id=1 name="id" type=INT begin=0 end=NULL
field_id=2 name="name" type=VARCHAR begin=0 end=NULL
field_id=3 name="amount" type=FLOAT begin=0 end=NULL
Parquet file written at snapshot 1:
Schema: { field_id=1: INT, field_id=2: VARCHAR, field_id=3: FLOAT }
The Parquet file does not store column names for matching purposes — it stores field IDs. This decouples the file’s schema from the catalog’s current schema.
In DuckLake’s C++, each column is represented by a DuckLakeFieldId object in src/include/storage/ducklake_field_data.hpp. ALTER TABLE operations are dispatched through DuckLakeTableEntry::Alter (src/storage/ducklake_table_entry.cpp).
ADD COLUMN (lake/alter.go)
Adding a column allocates a new field ID and inserts a new row in ducklake_column. Existing Parquet files are untouched.
1
2
3
4
5
6
7
8
9
10
11
12
Before (snapshot 3):
field_id=1 name="id" begin=0 end=NULL
field_id=2 name="name" begin=0 end=NULL
field_id=3 name="amount" begin=0 end=NULL
ALTER TABLE orders ADD COLUMN status VARCHAR;
After (snapshot 4):
field_id=1 name="id" begin=0 end=NULL
field_id=2 name="name" begin=0 end=NULL
field_id=3 name="amount" begin=0 end=NULL
field_id=4 name="status" begin=4 end=NULL ← new row
When reading old files:
1
2
3
4
5
6
7
8
9
Old file (written at snapshot 1):
Has field IDs: {1, 2, 3}
Does NOT have field ID 4
Reader sees current schema: {1: id, 2: name, 3: amount, 4: status}
field_id=1 → found in file → read normally
field_id=2 → found in file → read normally
field_id=3 → found in file → read normally
field_id=4 → NOT in file → fill with NULL
A missing field ID means the column did not exist when the file was written, so every row gets NULL for that column. No data rewrite needed.
The Go implementation:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
func (l *Lake) AddColumn(ctx context.Context, schemaName, tableName string, col catalog.ColumnDef) error {
txn, err := l.snaps.Begin(ctx)
if err != nil {
return fmt.Errorf("lake: begin transaction: %w", err)
}
defer txn.Rollback(ctx)
snapID := txn.SnapshotID()
_, table, err := l.resolveTableInTx(ctx, txn, schemaName, tableName)
if err != nil {
return err
}
catalogID := txn.NextCatalogID()
if err := l.store.AddColumn(ctx, txn.Tx(), snapID, table.TableID, col, catalogID); err != nil {
return err
}
txn.SetNextCatalogID(catalogID + 1)
txn.SetSchemaVersion(txn.SchemaVersion() + 1)
txn.RecordChange(fmt.Sprintf("add column %q to %s.%s", col.Name, schemaName, tableName))
_, err = txn.Commit(ctx)
return err
}
The metadata operation is a single INSERT:
1
2
3
4
5
6
7
8
9
func (s *MetadataStore) AddColumn(ctx context.Context, tx pgx.Tx, snap int64, tableID int64, col ColumnDef, catalogID int64) error {
order, _ := s.nextColumnOrder(ctx, tx, tableID)
_, err = tx.Exec(ctx, `
INSERT INTO ducklake_column
(column_id, table_id, field_id, column_order, name, type, nullable, begin_snapshot)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
`, catalogID, tableID, catalogID, order, col.Name, col.Type.String(), col.Nullable, snap)
return err
}
DROP COLUMN
Dropping a column sets end_snapshot on the column’s catalog row. The field ID is preserved in history for time travel.
1
2
3
4
5
6
7
Before (snapshot 5):
field_id=4 name="status" begin=4 end=NULL
ALTER TABLE orders DROP COLUMN status;
After (snapshot 6):
field_id=4 name="status" begin=4 end=6 ← end_snapshot set
At snapshot 6 and beyond, queries with the visibility filter will not see field_id=4. The data is still in Parquet files written between snapshots 4 and 6, but the scan will not request field ID 4.
Time travel still works: querying at snapshot 5 applies begin_snapshot <= 5 AND (end_snapshot IS NULL OR end_snapshot > 5), which includes field_id=4 (since end_snapshot=6 > 5).
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
func (l *Lake) DropColumn(ctx context.Context, schemaName, tableName, columnName string) error {
txn, err := l.snaps.Begin(ctx)
if err != nil {
return fmt.Errorf("lake: begin transaction: %w", err)
}
defer txn.Rollback(ctx)
snapID := txn.SnapshotID()
_, table, err := l.resolveTableInTx(ctx, txn, schemaName, tableName)
if err != nil {
return err
}
if err := l.store.DropColumn(ctx, txn.Tx(), snapID, table.TableID, columnName); err != nil {
return err
}
txn.SetSchemaVersion(txn.SchemaVersion() + 1)
txn.RecordChange(fmt.Sprintf("drop column %q from %s.%s", columnName, schemaName, tableName))
_, err = txn.Commit(ctx)
return err
}
The metadata operation is an UPDATE:
1
2
3
4
5
6
7
8
9
10
11
func (s *MetadataStore) DropColumn(ctx context.Context, tx pgx.Tx, snap int64, tableID int64, columnName string) error {
tag, err := tx.Exec(ctx, `
UPDATE ducklake_column
SET end_snapshot = $1
WHERE table_id = $2 AND name = $3 AND end_snapshot IS NULL
`, snap, tableID, columnName)
if tag.RowsAffected() == 0 {
return fmt.Errorf("column %q not found or already dropped", columnName)
}
return err
}
RENAME COLUMN
Renaming a column does not change the field ID. The catalog row is updated in place:
1
2
3
4
5
6
7
Before (snapshot 6):
field_id=3 name="amount" begin=0 end=NULL
ALTER TABLE orders RENAME COLUMN amount TO total;
After (snapshot 7):
field_id=3 name="total" begin=0 end=NULL ← same field ID, new name
Since Parquet files reference columns by field ID (3), every existing file works with the new name automatically:
1
2
3
4
5
Parquet file header: field_id=3
At snapshot 6: field_id=3 → "amount"
At snapshot 7: field_id=3 → "total"
Same file, same bytes, different column name in the result set.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
func (l *Lake) RenameColumn(ctx context.Context, schemaName, tableName, oldName, newName string) error {
txn, err := l.snaps.Begin(ctx)
if err != nil {
return fmt.Errorf("lake: begin transaction: %w", err)
}
defer txn.Rollback(ctx)
_, table, err := l.resolveTableInTx(ctx, txn, schemaName, tableName)
if err != nil {
return err
}
if err := l.store.RenameColumn(ctx, txn.Tx(), table.TableID, oldName, newName); err != nil {
return err
}
txn.SetSchemaVersion(txn.SchemaVersion() + 1)
txn.RecordChange(fmt.Sprintf("rename column %q to %q in %s.%s", oldName, newName, schemaName, tableName))
_, err = txn.Commit(ctx)
return err
}
Column Mapping During Reads
When reading a Parquet file written under an older schema, the reader must map the file’s field IDs to the current schema:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
Current schema (snapshot 7):
Position 0: field_id=1 "id" INT
Position 1: field_id=2 "name" VARCHAR
Position 2: field_id=3 "total" FLOAT (was "amount")
Position 3: field_id=4 "status" VARCHAR
Old file (written at snapshot 1):
Position 0: field_id=1 INT
Position 1: field_id=2 VARCHAR
Position 2: field_id=3 FLOAT
Mapping:
Current pos 0 (field_id=1) → File pos 0 ✓ read from file
Current pos 1 (field_id=2) → File pos 1 ✓ read from file
Current pos 2 (field_id=3) → File pos 2 ✓ read from file (name changed)
Current pos 3 (field_id=4) → not in file → fill with NULL
In DuckLake’s C++, this reconciliation happens in DuckLakeMultiFileReader. In our Go code, the fillMissingColumns function in the scan path handles it by inserting nil for any column not present in the row.
Schema Version Tracking
The ducklake_snapshot table has a schema_version field that is incremented by schema-altering operations but not by data operations:
1
2
3
4
5
Snapshot 1: CREATE TABLE orders → schema_version = 1
Snapshot 2: INSERT INTO orders → schema_version = 1 (unchanged)
Snapshot 3: ADD COLUMN status → schema_version = 2
Snapshot 4: RENAME COLUMN → schema_version = 3
Snapshot 5: INSERT INTO orders → schema_version = 3 (unchanged)
Schema version is used for compaction safety: files written under the same schema version can be merged without column reconciliation.