You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "tanmayrauth (via GitHub)" <gi...@apache.org> on 2026/05/20 08:21:49 UTC
[PR] feat(table): preserve row lineage through copy-on-write rewrites [iceberg-go]
tanmayrauth opened a new pull request, #1099:
URL: https://github.com/apache/iceberg-go/pull/1099
Parent: #999
Preserves `_row_id` through CoW rewrites and compaction on v3 tables. The scanner synthesizes row IDs from file metadata, the filter is applied post-synthesis to preserve correct positions, and the writer stores `_row_id` explicitly in output Parquet. `_last_updated_sequence_number` is left null to inherit from the new file's `data_sequence_number`.
Also documents that `next-row-id` accounting intentionally overcounts (matching Java's `ManifestListWriter.V4Writer`) and removes the old `Projection()` synthesis helpers replaced by the explicit `WithRowLineage()` scan option.
## Test plan
- `TestCoWRewritePreservesRowID` : verifies surviving rows keep original `_row_id` after a CoW delete
- `TestCoWRewriteRowIDNextRowIDAccounting` : validates next-row-id advances correctly
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): preserve row lineage through copy-on-write rewrites [iceberg-go]
Posted by "tanmayrauth (via GitHub)" <gi...@apache.org>.
tanmayrauth commented on code in PR #1099:
URL: https://github.com/apache/iceberg-go/pull/1099#discussion_r3291868638
##########
table/row_lineage_rewrite_test.go:
##########
@@ -0,0 +1,291 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table_test
+
+import (
+ "context"
+ "path/filepath"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func newV3RowLineageTestTable(t *testing.T) *table.Table {
+ t.Helper()
+
+ location := filepath.ToSlash(t.TempDir())
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "data", Type: iceberg.PrimitiveTypes.String, Required: false},
+ )
+ meta, err := table.NewMetadata(schema, iceberg.UnpartitionedSpec, table.UnsortedSortOrder, location,
+ iceberg.Properties{table.PropertyFormatVersion: "3"})
+ require.NoError(t, err)
+
+ metaLoc := location + "/metadata/v1.metadata.json"
+ fsF := func(context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil }
+ cat := &concurrentTestCatalog{metadata: meta, location: metaLoc, fsF: fsF}
+
+ return table.New(table.Identifier{"db", "row_lineage_test"}, meta, metaLoc, fsF, cat)
+}
+
+// TestCoWRewritePreservesRowID verifies that a copy-on-write overwrite with a
+// row filter preserves the original _row_id and _last_updated_sequence_number
+// values in the rewritten file. Surviving rows must keep both values from the
+// pre-rewrite snapshot — the rewrite is "physically rewritten", not "logically
+// updated", per the v3 spec.
+func TestCoWRewritePreservesRowID(t *testing.T) {
+ ctx := context.Background()
+ mem := memory.DefaultAllocator
+
+ tbl := newV3RowLineageTestTable(t)
+
+ // Append 3 rows: id=1,2,3
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
+ {Name: "data", Type: arrow.BinaryTypes.String, Nullable: true},
+ }, nil)
+
+ initialData, err := array.TableFromJSON(mem, arrowSchema, []string{
+ `[{"id": 1, "data": "a"}, {"id": 2, "data": "b"}, {"id": 3, "data": "c"}]`,
+ })
+ require.NoError(t, err)
+ defer initialData.Release()
+
+ tbl, err = tbl.Append(ctx, array.NewTableReader(initialData, -1), nil)
+ require.NoError(t, err)
+
+ // Verify the append created a valid v3 snapshot with row lineage.
+ snap := tbl.CurrentSnapshot()
+ require.NotNil(t, snap)
+ require.NotNil(t, snap.FirstRowID, "v3 snapshot must have first-row-id")
+ require.NotNil(t, snap.AddedRows, "v3 snapshot must have added-rows")
+ assert.Equal(t, int64(0), *snap.FirstRowID)
+ assert.Equal(t, int64(3), *snap.AddedRows)
+
+ // Capture the snapshot's sequence number so we can assert preservation
+ // after the rewrite. After the append, every row's effective
+ // _last_updated_sequence_number should be this value.
+ createSeq := snap.SequenceNumber
+
+ // Scan with row lineage to see synthesized _row_id values before the rewrite.
+ lineageScan := tbl.Scan(table.WithRowLineage())
+ schema, itr, err := lineageScan.ToArrowRecords(ctx)
+ require.NoError(t, err)
+
+ rowIDIdx := -1
+ for i, f := range schema.Fields() {
+ if f.Name == iceberg.RowIDColumnName {
+ rowIDIdx = i
+
+ break
+ }
+ }
+ require.GreaterOrEqual(t, rowIDIdx, 0, "_row_id should be in scan projection")
+
+ var originalRowIDs []int64
+ for rec, err := range itr {
+ require.NoError(t, err)
+ col := rec.Column(rowIDIdx).(*array.Int64)
+ for i := 0; i < col.Len(); i++ {
+ originalRowIDs = append(originalRowIDs, col.Value(i))
+ }
+ rec.Release()
+ }
+ require.Equal(t, []int64{0, 1, 2}, originalRowIDs, "initial _row_id should be 0,1,2")
+
+ // CoW overwrite: delete the row where id=2, preserving id=1 and id=3.
+ filter := iceberg.EqualTo(iceberg.Reference("id"), int64(2))
+ tbl, err = tbl.Delete(ctx, filter, nil)
+ require.NoError(t, err)
+
+ snap = tbl.CurrentSnapshot()
+ require.NotNil(t, snap)
+ require.Greater(t, snap.SequenceNumber, createSeq,
+ "sanity: rewrite snapshot must have a higher sequence number than the create snapshot")
+
+ // Scan the result with row lineage. The surviving rows should preserve their
+ // original _row_id values: 0 and 2. Their _last_updated_sequence_number must
+ // also still report the create snapshot's seq, NOT the rewrite snapshot's
+ // seq — the rewrite is physical only, not a logical update.
+ lineageScan = tbl.Scan(table.WithRowLineage())
+ _, itr, err = lineageScan.ToArrowRecords(ctx)
+ require.NoError(t, err)
+
+ var afterRowIDs []int64
+ var afterIDs []int64
+ var afterSeq []int64
+ for rec, err := range itr {
+ require.NoError(t, err)
+ idIdx := rec.Schema().FieldIndices("id")
+ require.NotEmpty(t, idIdx)
+ rowIDIndices := rec.Schema().FieldIndices(iceberg.RowIDColumnName)
+ require.NotEmpty(t, rowIDIndices)
+ seqIndices := rec.Schema().FieldIndices(iceberg.LastUpdatedSequenceNumberColumnName)
+ require.NotEmpty(t, seqIndices, "_last_updated_sequence_number must be in projection")
+
+ idCol := rec.Column(idIdx[0]).(*array.Int64)
+ rowIDCol := rec.Column(rowIDIndices[0]).(*array.Int64)
+ seqCol := rec.Column(seqIndices[0]).(*array.Int64)
+ for i := 0; i < int(rec.NumRows()); i++ {
+ afterIDs = append(afterIDs, idCol.Value(i))
+ afterRowIDs = append(afterRowIDs, rowIDCol.Value(i))
+ require.False(t, seqCol.IsNull(i),
+ "row %d must have a non-null _last_updated_sequence_number after CoW rewrite", i)
+ afterSeq = append(afterSeq, seqCol.Value(i))
+ }
+ rec.Release()
+ }
+
+ assert.Equal(t, []int64{1, 3}, afterIDs, "remaining rows should be id=1,3")
+ assert.Equal(t, []int64{0, 2}, afterRowIDs,
+ "_row_id must be preserved through CoW rewrite: row with id=1 keeps _row_id=0, row with id=3 keeps _row_id=2")
+ assert.Equal(t, []int64{createSeq, createSeq}, afterSeq,
+ "_last_updated_sequence_number must report the original creation snapshot's sequence number, not the rewrite's")
+}
+
+// TestCoWRewriteRowIDNextRowIDAccounting verifies that row-id accounting remains
+// correct after a CoW rewrite. The overcounting (where next-row-id advances by
+// the full manifest row count including preserved survivors) is intentional and
+// matches Java's ManifestListWriter.V3Writer behavior.
+func TestCoWRewriteRowIDNextRowIDAccounting(t *testing.T) {
+ ctx := context.Background()
+ mem := memory.DefaultAllocator
+
+ tbl := newV3RowLineageTestTable(t)
+
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
+ {Name: "data", Type: arrow.BinaryTypes.String, Nullable: true},
+ }, nil)
+
+ data, err := array.TableFromJSON(mem, arrowSchema, []string{
+ `[{"id": 10, "data": "x"}, {"id": 20, "data": "y"}, {"id": 30, "data": "z"}]`,
+ })
+ require.NoError(t, err)
+ defer data.Release()
+
+ tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil)
+ require.NoError(t, err)
+
+ // next-row-id should be 3 after appending 3 rows.
+ assert.Equal(t, int64(3), tbl.Metadata().NextRowID())
+
+ // Delete one row via CoW.
+ filter := iceberg.EqualTo(iceberg.Reference("id"), int64(20))
+ tbl, err = tbl.Delete(ctx, filter, nil)
+ require.NoError(t, err)
+
+ // next-row-id advances by 3 (original) + 2 (rewritten survivors) = 5,
+ // even though the surviving rows preserve their old IDs. This "wastes"
+ // ID space but doesn't violate uniqueness — actual row IDs come from
+ // the explicit Parquet column, not the global counter.
+ assert.Equal(t, int64(5), tbl.Metadata().NextRowID(),
+ "next-row-id should advance by original (3) + rewritten (2) = 5")
Review Comment:
On snapshot.AddedRows: the existing comment in snapshot_producers.go:920 explicitly says it matches Java's ManifestListWriter.V3Writer accounting (counts existing + added in new manifests). Worth a separate look since if Java's snapshot summary added-rows field actually means "newly added rows only," that's a pre-existing bug in the v3 writer rather than something this PR introduced. Want me to file that as a separate issue?
##########
metadata_columns.go:
##########
@@ -59,3 +61,38 @@ func LastUpdatedSequenceNumber() NestedField {
func IsMetadataColumn(fieldID int) bool {
return fieldID == RowIDFieldID || fieldID == LastUpdatedSequenceNumberFieldID
}
+
+// SchemaWithRowLineage returns a new schema with the row-lineage metadata columns
+// (_row_id, _last_updated_sequence_number) appended to the given schema's fields.
+// Used when reading source files during a CoW rewrite or compaction so that row
+// identity and per-row update sequence are preserved in the output.
+//
+// Idempotent: if a row-lineage column is already present (by reserved field ID),
+// it is not appended again. The returned schema always allocates a fresh field
+// slice so it cannot alias the input schema's backing array.
+func SchemaWithRowLineage(s *Schema) *Schema {
Review Comment:
On unexporting: SchemaWithRowLineage is already used by callers outside the table package (compaction's ExecuteCompactionGroup builds the projected schema this way before passing it to WithPreserveRowLineage), so unexporting it now means duplicating the helper or carving out a different API. Worth doing as a follow-up if v4 lineage actually changes the shape, but I'd rather not preemptively redesign for a hypothetical. Same reasoning on WithPreserveRowLineage(schema) — having the caller construct the schema once keeps scan-side and write-side projections in lockstep, which matters for the compaction path where both have to match exactly.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): preserve row lineage through copy-on-write rewrites [iceberg-go]
Posted by "tanmayrauth (via GitHub)" <gi...@apache.org>.
tanmayrauth commented on code in PR #1099:
URL: https://github.com/apache/iceberg-go/pull/1099#discussion_r3291870263
##########
table/transaction.go:
##########
@@ -1798,3 +1905,53 @@ func (s *StagedTable) Refresh(ctx context.Context) (*Table, error) {
func (s *StagedTable) Scan(opts ...ScanOption) *Scan {
panic(fmt.Errorf("%w: cannot scan a staged table", ErrInvalidOperation))
}
+
+// prepareBatchFilter binds the given Iceberg filter against schema and converts
+// it to substrait once, returning a per-batch filter function that can be
+// reused across every record batch. The setup work (BindExpr, ConvertExpr) is
+// independent of the batch and is the most expensive part of filter-eval, so
+// hoisting it out of the iterator loop is a measurable win on rewrites that
+// produce many batches.
+//
+// The returned function takes ownership of the input batch (it releases it on
+// the AlwaysFalse fast-path) and returns a possibly-new batch the caller is
+// responsible for releasing.
+func prepareBatchFilter(ctx context.Context, filter iceberg.BooleanExpression, schema *iceberg.Schema, caseSensitive bool) (func(arrow.RecordBatch) (arrow.RecordBatch, error), error) {
Review Comment:
I think the current contract is consistent: input ownership transfers to output, caller releases once. Adding Retain() on the AlwaysTrue path would leak the input (caller would have to release input AND output, but the contract today is "after postFilter(rec), only release the result").
##########
table/transaction.go:
##########
@@ -1530,14 +1565,50 @@ func (t *Transaction) rewriteFilesWithFilter(ctx context.Context, fs io.IO, upda
return nil
}
-// rewriteSingleFile reads a single data file, applies the filter, and writes new files with filtered data
-func (t *Transaction) rewriteSingleFile(ctx context.Context, fs io.IO, originalFile iceberg.DataFile, filter iceberg.BooleanExpression, caseSensitive bool, commitUUID uuid.UUID, concurrency int) ([]iceberg.DataFile, error) {
+// rewriteSingleFile reads a single data file, applies the filter, and writes new files with filtered data.
+// fileSeqNum is the source file's file_sequence_number from its manifest entry; required to synthesize
+// _last_updated_sequence_number for rows whose value is null in the source file.
+func (t *Transaction) rewriteSingleFile(ctx context.Context, fs io.IO, originalFile iceberg.DataFile, fileSeqNum *int64, filter iceberg.BooleanExpression, postFilter func(arrow.RecordBatch) (arrow.RecordBatch, error), caseSensitive bool, commitUUID uuid.UUID, concurrency int) ([]iceberg.DataFile, error) {
Review Comment:
Bundled the params into rewriteSingleFileArgs. On the wasted bind: hoisted BindExpr into the !preserveRowLineage branch so the per-file bind is no longer wasted on the lineage path. Left the prepareBatchFilter bind alone — it runs once per rewriteFilesWithFilter call rather than per file, so the cost is amortized across all rewritten files in the operation.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): preserve row lineage through copy-on-write rewrites [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on PR #1099:
URL: https://github.com/apache/iceberg-go/pull/1099#issuecomment-4512456386
Rebasing should fix the integration tests
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): preserve row lineage through copy-on-write rewrites [iceberg-go]
Posted by "tanmayrauth (via GitHub)" <gi...@apache.org>.
tanmayrauth commented on PR #1099:
URL: https://github.com/apache/iceberg-go/pull/1099#issuecomment-4505129784
Thanks for the thorough pass, addressed all 12 inline comments. Going through them in order:
**1. `metadata_columns.go` : `SchemaWithRowID` panics on duplicate + aliases the input slice.**
Renamed to `SchemaWithRowLineage`. `slices.Clone`'d the field slice up front so the returned schema cannot share a backing array with the input, and the function skips appending when a reserved field ID (`_row_id` or `_last_updated_sequence_number`) is already present. New test `TestProjectionV3SchemaAlreadyHasRowID` locks in the no-panic / no-duplicate-id behavior.
**2. `table/transaction.go` : `_last_updated_sequence_number` left null breaks CDC semantics.**
Agreed, fixed. `rewriteSingleFile` now projects `_last_updated_sequence_number` via `SchemaWithRowLineage`, and `scanTask.DataSequenceNumber` is wired in from the source manifest entry's `FileSequenceNum` (threaded through a new `fileSeqByPath` map returned from `classifyFilesForFilteredDeletions`). The arrow scanner's `synthesizeRowLineageColumns` keeps any explicit per-row value from the source file and only synthesizes from the file's `data_sequence_number` for null source rows — so a pure CoW rewrite preserves the original "last logically updated" sequence per row instead of stamping the new commit's sequence onto every survivor. `TestCoWRewritePreservesRowID` was updated to assert this directly.
**3. `table/rewrite_data_files.go` : `hasRowLineage` permits mixed groups.**
Tightened to `allTasksHaveRowLineage` (renamed function in the other thread too). When even one task in the group lacks `FirstRowID`, the entire group skips the lineage path — avoids the "some explicit `_row_id`s, some nulls" output you flagged. Splitting mixed groups so each side rewrites with its own semantics is the cleaner long-term fix but requires changing `ExecuteCompactionGroup`'s output shape ,left as follow-up; called it out in the gating comment so we don't lose the thread.
**4. `table/table.go` : `WithRowLineage` doc says "ignored on non-v3" but impl projects regardless.**
`Projection()` now returns an explicit error when `includeRowLineage` is set on a v1/v2 table, and the `WithRowLineage` doc was rewritten to match. New test `TestProjectionWithRowLineageRequiresV3` covers it.
**5. `table/table.go` : `WithRowLineage` name suggests whole-feature control.**
Went with the "extend it" branch you suggested: `SchemaWithRowLineage` (the helper `WithRowLineage` calls) now projects both `_row_id` and `_last_updated_sequence_number`. So one switch ↔ both lineage columns ↔ preservation through rewrite. Kept the name `WithRowLineage` since the semantics now match it.
**6. `table/scanner.go` : losing the `Select("_row_id")` escape hatch.**
Restored. `Projection()` now intercepts row-lineage column names (case-aware) in the user's `selectedFields`, strips them before `curSchema.Select`, and appends the corresponding `RowID()` / `LastUpdatedSequenceNumber()` fields back onto the projected schema — but only on v3 tables. On v1/v2 the names fall through to `curSchema.Select`, which fails as before with `ErrInvalidSchema`. Restored `TestProjectionV3SelectRowLineageColumns` and `TestProjectionRowLineageRejectedOnV1V2` to lock both behaviors in.
**7. `table/transaction.go` : `filterRecordBatch` rebinds per batch + `AlwaysFalse` returns schemaless record.**
once for the whole rewrite (the complement filter doesn't change between files) and passes the resulting closure into every `rewriteSingleFile` call. The `AlwaysFalse` fast-path now returns `rec.NewSlice(0, 0)` so downstream code that calls `rec.Column(i)` doesn't panic on the empty result.
**8. `table/transaction.go` : `AlwaysTrue` disables pushdown for the rewrite scan.**
Left a `TODO(perf)` comment on `rewriteSingleFile` calling out the IO cost of disabling pushdown when preserving lineage, and noted that file-level skipping (against the bound filter's residual on file stats) could be re-enabled here without breaking `_row_id` synthesis since synthesis only depends on within-file row positions. Punting the actual implementation to a follow-up — agreed it's worth fixing but didn't want to expand this PR into a metrics-eval refactor.
**9. `table/rewrite_data_files.go` : `hasRowLineage` comment claims a table property but checks per-task.**
Renamed to `allTasksHaveRowLineage` and rewrote the comment. The function now also returns `false` on an empty task list (defensive), and the call-site comment explains why we gate on "all" rather than "any" (mixed groups → per-file invariant violation, see thread 3).
**10. `table/write_records.go` : `WithPreserveRowLineage` accepts anything.**
Added validation in `WriteRecords`: errors when the table format version is below 3, and errors when the input arrow schema doesn't include the `_row_id` column. Docstring rewritten to call this out. Implementation-wise this required a small reshuffle: `cfg` is now built before `checkArrowSchemaCompat` runs so the compat check can be against the projected (with-lineage) schema rather than `tbl.Schema()` — otherwise the compat check itself would reject `_row_id` as unknown.
**11. `table/row_lineage_rewrite_test.go` : `GreaterOrEqual(5)` is vacuous.**
Tightened to `assert.Equal(t, int64(5), tbl.Metadata().NextRowID())`. The comment above the assertion already had the exact expected value (3 + 2), so the loose form was just hiding regressions in either direction.
**12. `table/snapshot_producers.go` : `ManifestListWriter.V4Writer` doesn't exist.**
Typo, fixed to `V3Writer`. Same fix applied to the matching comment in the test that referenced V4Writer for context.
---
**Newly added tests:**
- `TestProjectionV3SelectRowLineageColumns` : `Select("_row_id")` works on v3
- `TestProjectionRowLineageRejectedOnV1V2` : same select on v1/v2 errors with `ErrInvalidSchema`
- `TestProjectionWithRowLineageRequiresV3` : `WithRowLineage` errors on v1/v2
- `TestProjectionV3SchemaAlreadyHasRowID` : idempotency under reserved-id collision
- `TestExecuteCompactionGroupPreservesRowID` : end-to-end compaction on a v3 table preserves `_row_id` for every row across two source files
PTAL when you have a moment.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): preserve row lineage through copy-on-write rewrites [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #1099:
URL: https://github.com/apache/iceberg-go/pull/1099#discussion_r3277166972
##########
metadata_columns.go:
##########
@@ -59,3 +59,12 @@ func LastUpdatedSequenceNumber() NestedField {
func IsMetadataColumn(fieldID int) bool {
return fieldID == RowIDFieldID || fieldID == LastUpdatedSequenceNumberFieldID
}
+
+// SchemaWithRowID returns a new schema that appends the _row_id metadata column
+// to the given schema's fields. Used when reading source files during a CoW rewrite
+// so that row identity is preserved in the output.
+func SchemaWithRowID(s *Schema) *Schema {
+ fields := append(s.Fields(), RowID())
Review Comment:
Two issues here that bite in practice.
First, if the input schema already has a field with ID 1 (e.g. a user column named `id`), `NewSchemaWithIdentifiers` will reject the duplicate and we panic at runtime. We can't assume user schemas avoid 1, 2, 3.
Second, `append(schema.Fields(), ...)` aliases the original schema's backing array when there's spare cap. A later mutation of either schema can corrupt the other. I'd `slices.Clone(schema.Fields())` first, and either skip-or-error when one of the row-lineage IDs collides instead of panicking. wdyt?
##########
table/rewrite_data_files.go:
##########
@@ -289,6 +289,11 @@ func ExecuteCompactionGroup(ctx context.Context, tbl *Table, group CompactionTas
scanOpts = append(scanOpts, WitMaxConcurrency(cfg.scanConcurrency))
}
+ preserveLineage := tbl.metadata.Version() >= 3 && hasRowLineage(group.Tasks)
Review Comment:
Mixing row-lineage and non-row-lineage source files in a single rewrite group is going to produce a file where some rows have `_row_id` / `_last_updated_sequence_number` and others don't, which violates the per-file invariant. `hasRowLineage` returning true if *any* task has lineage means the lineage path runs for the whole group regardless of the mix.
I'd partition the input files by file-level `has_row_lineage` and rewrite each partition separately, or skip non-lineage files when the table has row lineage enabled. Silently merging them feels like a correctness bug waiting to surface.
##########
table/transaction.go:
##########
@@ -1532,6 +1532,26 @@ func (t *Transaction) rewriteSingleFile(ctx context.Context, fs io.IO, originalF
Length: originalFile.FileSizeBytes(),
}
+ // Preserve row lineage for v3 tables: include _row_id in the scan
+ // projection so it is synthesized (or read from existing explicit values)
+ // and written into the output file. _last_updated_sequence_number is left
+ // out — it will be null in the output and inherited from the new file's
+ // data_sequence_number at read time, which satisfies the spec requirement
Review Comment:
The spec says `last-updated-sequence-number` on a data file is the sequence number of the snapshot in which the row was last updated. Leaving it null on rewrite and inheriting from the new file's `data_sequence_number` conflates "physically rewritten" with "logically updated" and breaks downstream readers that use this for CDC / incremental scans.
For a pure compaction (no row changes), we should preserve the original per-row `_last_updated_sequence_number` from the source files. Otherwise every compaction looks like a full table update to consumers.
##########
table/scanner.go:
##########
@@ -261,26 +260,22 @@ func (scan *Scan) Projection() (*iceberg.Schema, error) {
}
}
+ var schema *iceberg.Schema
if slices.Contains(scan.selectedFields, "*") {
- return curSchema, nil
- }
-
- selectedFieldsMeta := metaFieldsFromSelectedFields(scan.selectedFields, caseSensitive)
- schemaMeta := metaFieldsFromSchema(curSchema)
- synthesisMeta := synthesizeMeta(selectedFieldsMeta, schemaMeta)
- if len(synthesisMeta) > 0 && curVersion >= minFormatVersionRowLineage {
-
- // synthesis path
- removedMetaSlice, missingMetaFields := removeMetadataFromSelectedFields(scan.selectedFields, synthesisMeta)
- sch, err := curSchema.Select(scan.caseSensitive, removedMetaSlice...)
+ schema = curSchema
+ } else {
+ var err error
+ schema, err = curSchema.Select(scan.caseSensitive, scan.selectedFields...)
Review Comment:
This loses the metadata-column escape hatch that the old synthesis path supported. `Select("_row_id")` (or `Select("_last_updated_sequence_number")`) now goes straight to `curSchema.Select` and fails because the user schema doesn't contain those names — the only way to get those columns is `WithRowLineage()`, which is a separate API.
If that's intentional, worth a CHANGELOG note. If not, we should intercept the known metadata column names here before the schema lookup, the way the removed `synthesizeMeta`/`removeMetadataFromSelectedFields` did.
##########
table/write_records.go:
##########
@@ -94,6 +95,16 @@ func WithClusteredWrite() WriteRecordOption {
}
}
+// WithPreserveRowLineage sets the output file schema to include _row_id so that
+// row identity is preserved through rewrites and compactions on v3 tables. The
+// input records must already contain the _row_id column (e.g. from a scan that
+// projected lineage columns).
+func WithPreserveRowLineage(schema *iceberg.Schema) WriteRecordOption {
Review Comment:
`WithPreserveRowLineage` is accepted on any table — there's no check that the table actually has v3 row lineage enabled, or that the input records contain `_row_id`. A caller who sets it on a v2 table or feeds records without the lineage column gets silently-wrong output rather than an error.
I'd validate at option-application time (or inside `WriteRecords` once we have the `tbl`): if the table format is < 3, error. If the input Arrow schema is missing `_row_id`, error.
##########
table/table.go:
##########
@@ -816,6 +816,14 @@ func WithOptions(opts iceberg.Properties) ScanOption {
}
}
+// WithRowLineage adds _row_id to the scan projection so that row identity is
Review Comment:
The doc says this is "ignored" on non-v3 tables, but the implementation sets `scan.includeRowLineage = true` unconditionally and `SchemaWithRowID` then runs in `Projection()` regardless of format version. On a v1/v2 table that ends up projecting a metadata column the table doesn't support.
Either the doc is wrong or the guard is missing. I'd add an explicit format-version check (either error or true no-op) and make the doc match the behavior.
##########
table/transaction.go:
##########
@@ -1788,3 +1843,40 @@ func (s *StagedTable) Refresh(ctx context.Context) (*Table, error) {
func (s *StagedTable) Scan(opts ...ScanOption) *Scan {
panic(fmt.Errorf("%w: cannot scan a staged table", ErrInvalidOperation))
}
+
+// filterRecordBatch applies an Iceberg filter to a record batch, returning
+// only matching rows. The filter is provided unbound and is bound against the
+// given schema internally. Used for post-synthesis row filtering in CoW rewrites
+// where row lineage must be synthesized before filtering.
+func filterRecordBatch(ctx context.Context, rec arrow.RecordBatch, filter iceberg.BooleanExpression, schema *iceberg.Schema, caseSensitive bool) (arrow.RecordBatch, error) {
Review Comment:
Two things in `filterRecordBatch`:
This is called once per batch from the iterator loop, but `BindExpr` and `substrait.ConvertExpr` depend only on the (schema, filter) pair, not the batch. We're paying that cost N times for no reason — hoist the bind + convert out and pass them in, or memoize on first call.
Also, the `AlwaysFalse` branch returns `array.NewRecordBatch(emptySchema, nil, 0)` with no columns at all — downstream code that does `rec.Column(i)` will panic. The empty record should have the right schema *and* empty arrays for each field.
##########
table/rewrite_data_files.go:
##########
@@ -328,6 +336,18 @@ func ExecuteCompactionGroup(ctx context.Context, tbl *Table, group CompactionTas
}, nil
}
+// hasRowLineage returns true if any task in the group has a non-nil FirstRowID,
Review Comment:
The comment says "indicating the table has v3 row lineage data to preserve", but the body checks per-task `FirstRowID` — a per-file property, not a table property. A v3 table can legitimately have a mix of tasks where some have `FirstRowID` and some don't (legacy files), so "any task has it" doesn't really mean "the table has lineage".
I'd either rename to `anyTaskHasRowLineage` (and fix the comment to match), or — better — gate on `tbl.metadata.Version() >= 3` at the call site and rewrite this to mean "all tasks have lineage" so we don't trip the mixed-lineage bug above.
##########
table/table.go:
##########
@@ -816,6 +816,14 @@ func WithOptions(opts iceberg.Properties) ScanOption {
}
}
+// WithRowLineage adds _row_id to the scan projection so that row identity is
+// preserved through rewrites. Only meaningful for v3 tables; ignored otherwise.
+func WithRowLineage() ScanOption {
Review Comment:
The name `WithRowLineage` suggests it controls the whole row-lineage feature, but it only adds `_row_id` to the projection — `_last_updated_sequence_number` is governed elsewhere. That's confusing for callers who'd expect one switch to mean one thing.
Could we either rename to scope it (e.g. `WithRowID`) or extend it to also project `_last_updated_sequence_number`? wdyt?
##########
table/transaction.go:
##########
@@ -1542,11 +1562,19 @@ func (t *Transaction) rewriteSingleFile(ctx context.Context, fs io.IO, originalF
return nil, fmt.Errorf("failed to build metadata: %w", err)
}
+ // When preserving row lineage, omit the row filter from the scanner so that
+ // _row_id is synthesized using the true file-level row positions. The filter
+ // is applied below in the record iterator.
+ scanFilter := iceberg.BooleanExpression(iceberg.AlwaysTrue{})
Review Comment:
Passing `AlwaysTrue` here disables both manifest-level and file-level pushdown for the rewrite scan when row lineage is on. For a selective delete on a wide partition that's a lot of unnecessary IO — we read every file in scope just to filter post-synthesis.
Worst case this is unavoidable because `_row_id` synthesis needs original positions, but the comment treats it as obviously fine. At minimum I'd leave a TODO calling out the cost, ideally we'd still apply file-level (not row-level) skipping based on `boundFilter`'s residuals.
##########
table/row_lineage_rewrite_test.go:
##########
@@ -0,0 +1,188 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table_test
+
+import (
+ "context"
+ "path/filepath"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func newV3RowLineageTestTable(t *testing.T) *table.Table {
+ t.Helper()
+
+ location := filepath.ToSlash(t.TempDir())
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "data", Type: iceberg.PrimitiveTypes.String, Required: false},
+ )
+ meta, err := table.NewMetadata(schema, iceberg.UnpartitionedSpec, table.UnsortedSortOrder, location,
+ iceberg.Properties{table.PropertyFormatVersion: "3"})
+ require.NoError(t, err)
+
+ metaLoc := location + "/metadata/v1.metadata.json"
+ fsF := func(context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil }
+ cat := &concurrentTestCatalog{metadata: meta, location: metaLoc, fsF: fsF}
+
+ return table.New(table.Identifier{"db", "row_lineage_test"}, meta, metaLoc, fsF, cat)
+}
+
+// TestCoWRewritePreservesRowID verifies that a copy-on-write overwrite with a
+// row filter preserves the original _row_id values in the rewritten file. The
+// _last_updated_sequence_number should be null (inherited from data_sequence_number).
+func TestCoWRewritePreservesRowID(t *testing.T) {
+ ctx := context.Background()
+ mem := memory.DefaultAllocator
+
+ tbl := newV3RowLineageTestTable(t)
+
+ // Append 3 rows: id=1,2,3
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
+ {Name: "data", Type: arrow.BinaryTypes.String, Nullable: true},
+ }, nil)
+
+ initialData, err := array.TableFromJSON(mem, arrowSchema, []string{
+ `[{"id": 1, "data": "a"}, {"id": 2, "data": "b"}, {"id": 3, "data": "c"}]`,
+ })
+ require.NoError(t, err)
+ defer initialData.Release()
+
+ tbl, err = tbl.Append(ctx, array.NewTableReader(initialData, -1), nil)
+ require.NoError(t, err)
+
+ // Verify the append created a valid v3 snapshot with row lineage.
+ snap := tbl.CurrentSnapshot()
+ require.NotNil(t, snap)
+ require.NotNil(t, snap.FirstRowID, "v3 snapshot must have first-row-id")
+ require.NotNil(t, snap.AddedRows, "v3 snapshot must have added-rows")
+ assert.Equal(t, int64(0), *snap.FirstRowID)
+ assert.Equal(t, int64(3), *snap.AddedRows)
+
+ // Scan with row lineage to see synthesized _row_id values before the rewrite.
+ lineageScan := tbl.Scan(table.WithRowLineage())
+ schema, itr, err := lineageScan.ToArrowRecords(ctx)
+ require.NoError(t, err)
+
+ rowIDIdx := -1
+ for i, f := range schema.Fields() {
+ if f.Name == iceberg.RowIDColumnName {
+ rowIDIdx = i
+
+ break
+ }
+ }
+ require.GreaterOrEqual(t, rowIDIdx, 0, "_row_id should be in scan projection")
+
+ var originalRowIDs []int64
+ for rec, err := range itr {
+ require.NoError(t, err)
+ col := rec.Column(rowIDIdx).(*array.Int64)
+ for i := 0; i < col.Len(); i++ {
+ originalRowIDs = append(originalRowIDs, col.Value(i))
+ }
+ rec.Release()
+ }
+ require.Equal(t, []int64{0, 1, 2}, originalRowIDs, "initial _row_id should be 0,1,2")
+
+ // CoW overwrite: delete the row where id=2, preserving id=1 and id=3.
+ filter := iceberg.EqualTo(iceberg.Reference("id"), int64(2))
+ tbl, err = tbl.Delete(ctx, filter, nil)
+ require.NoError(t, err)
+
+ snap = tbl.CurrentSnapshot()
+ require.NotNil(t, snap)
+
+ // Scan the result with row lineage. The surviving rows should preserve their
+ // original _row_id values: 0 and 2.
+ lineageScan = tbl.Scan(table.WithRowLineage())
+ _, itr, err = lineageScan.ToArrowRecords(ctx)
+ require.NoError(t, err)
+
+ var afterRowIDs []int64
+ var afterIDs []int64
+ for rec, err := range itr {
+ require.NoError(t, err)
+ idIdx := rec.Schema().FieldIndices("id")
+ require.NotEmpty(t, idIdx)
+ rowIDIndices := rec.Schema().FieldIndices(iceberg.RowIDColumnName)
+ require.NotEmpty(t, rowIDIndices)
+
+ idCol := rec.Column(idIdx[0]).(*array.Int64)
+ rowIDCol := rec.Column(rowIDIndices[0]).(*array.Int64)
+ for i := 0; i < int(rec.NumRows()); i++ {
+ afterIDs = append(afterIDs, idCol.Value(i))
+ afterRowIDs = append(afterRowIDs, rowIDCol.Value(i))
+ }
+ rec.Release()
+ }
+
+ assert.Equal(t, []int64{1, 3}, afterIDs, "remaining rows should be id=1,3")
+ assert.Equal(t, []int64{0, 2}, afterRowIDs,
+ "_row_id must be preserved through CoW rewrite: row with id=1 keeps _row_id=0, row with id=3 keeps _row_id=2")
+}
+
+// TestCoWRewriteRowIDNextRowIDAccounting verifies that row-id accounting remains
+// correct after a CoW rewrite. The overcounting (where next-row-id advances by
+// the full manifest row count including preserved survivors) is intentional and
+// matches Java's ManifestListWriter.V4Writer behavior.
+func TestCoWRewriteRowIDNextRowIDAccounting(t *testing.T) {
+ ctx := context.Background()
+ mem := memory.DefaultAllocator
+
+ tbl := newV3RowLineageTestTable(t)
+
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
+ {Name: "data", Type: arrow.BinaryTypes.String, Nullable: true},
+ }, nil)
+
+ data, err := array.TableFromJSON(mem, arrowSchema, []string{
+ `[{"id": 10, "data": "x"}, {"id": 20, "data": "y"}, {"id": 30, "data": "z"}]`,
+ })
+ require.NoError(t, err)
+ defer data.Release()
+
+ tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil)
+ require.NoError(t, err)
+
+ // next-row-id should be 3 after appending 3 rows.
+ assert.Equal(t, int64(3), tbl.Metadata().NextRowID())
+
+ // Delete one row via CoW.
+ filter := iceberg.EqualTo(iceberg.Reference("id"), int64(20))
+ tbl, err = tbl.Delete(ctx, filter, nil)
+ require.NoError(t, err)
+
+ // next-row-id advances by the rewritten manifest's total row count (2 surviving
+ // rows), even though those rows preserve old IDs. This "wastes" ID space but
+ // doesn't violate uniqueness — actual row IDs come from the explicit Parquet
+ // column, not the global counter.
+ nextRowID := tbl.Metadata().NextRowID()
+ assert.GreaterOrEqual(t, nextRowID, int64(5),
Review Comment:
`GreaterOrEqual(..., int64(5))` makes the assertion vacuous in the loose direction — it passes even if next-row-id leapt to some huge value because of a bug elsewhere. And the comment above says the expected accounting is "original + rewritten = 3 + 2 = 5", so we actually know the expected value.
I'd assert `Equal(t, int64(5), nextRowID)` (or whatever the exact post-rewrite expectation is). As written, this test would still pass if next-row-id accounting drifted in either direction.
##########
table/snapshot_producers.go:
##########
@@ -917,6 +917,11 @@ func (sp *snapshotProducer) commit(ctx context.Context) (_ []Update, _ []Require
return nil, nil, err
}
if writer.NextRowID() != nil {
+ // addedRows counts ALL rows in new manifests (existing + added), even
+ // for rewrites where survivors preserve old _row_id values. This
+ // "wastes" ID space but doesn't violate uniqueness: actual row IDs come
+ // from the explicit Parquet column, not the global counter. Java's
+ // ManifestListWriter.V4Writer uses the same accounting.
Review Comment:
`ManifestListWriter.V4Writer` looks like a typo — Java's manifest-list writer for row lineage is `V3Writer` (the v3 format-version writer); there is no V4 manifest. Unless there's a v4 draft I'm missing, the comment should read `V3Writer`.
Misleading provenance comments are extra costly because the next reader trying to verify parity will hunt for a class that doesn't exist.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): preserve row lineage through copy-on-write rewrites [iceberg-go]
Posted by "tanmayrauth (via GitHub)" <gi...@apache.org>.
tanmayrauth commented on PR #1099:
URL: https://github.com/apache/iceberg-go/pull/1099#issuecomment-4524780157
@laskoviymishka PTAL when you have a moment.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): preserve row lineage through copy-on-write rewrites [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka merged PR #1099:
URL: https://github.com/apache/iceberg-go/pull/1099
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): preserve row lineage through copy-on-write rewrites [iceberg-go]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #1099:
URL: https://github.com/apache/iceberg-go/pull/1099#discussion_r3289810518
##########
table/transaction.go:
##########
@@ -1488,13 +1501,23 @@ func (t *Transaction) classifyFilesForFilteredDeletions(ctx context.Context, fs
localDelete = append(localDelete, df)
} else {
localRewrite = append(localRewrite, df)
+ // Capture the file's sequence number from the
+ // manifest entry so the rewrite path can synthesize
+ // _last_updated_sequence_number for source rows
+ // that have a null value (or no column) in the file.
+ if fseq := entry.FileSequenceNum(); fseq != nil {
Review Comment:
I think this is reading the wrong manifest field. Spec §Row Lineage says when `_last_updated_sequence_number` is null on read, it gets assigned the manifest entry's `sequence_number` (field id 3, the data sequence number) — not `file_sequence_number` (field id 4, the snapshot that added the file). For back-dated files where they diverge, a Java or iceberg-rust reader would compute a different value for the same row.
I'd switch both this site and the matching one in `scanner.go` to `entry.SequenceNum()`. `SequenceNum()` returns `int64` not `*int64`, so the nil guard collapses — drop the conditional or guard on `>= 0`. While we're here, I'd rename `FileScanTask.DataSequenceNumber` if that's where this plumbs through, so the field name matches what it actually holds.
##########
table/rewrite_data_files.go:
##########
@@ -289,6 +289,20 @@ func ExecuteCompactionGroup(ctx context.Context, tbl *Table, group CompactionTas
scanOpts = append(scanOpts, WitMaxConcurrency(cfg.scanConcurrency))
}
+ // Preserve row lineage only when every source file in the group carries
+ // it. A mixed group (some files with FirstRowID, some without — e.g.
+ // legacy files on a v3 table) would otherwise produce one output where
+ // post-lineage rows have explicit _row_id values and pre-lineage rows
+ // have nulls, which violates the per-file uniqueness/coverage
+ // invariant the v3 spec requires. Splitting mixed groups into separate
+ // outputs is a larger refactor and is left as a follow-up; for now we
+ // degrade gracefully (the rewrite still succeeds, but lineage is not
+ // preserved for the surviving rows).
+ preserveLineage := tbl.metadata.Version() >= 3 && allTasksHaveRowLineage(group.Tasks)
Review Comment:
Mixed-lineage groups silently fall through to the non-lineage path. Compaction succeeds, the output is valid, but `_row_id` is gone — no log, no metric, no flag in `CompactionGroupResult`. For tables migrating to v3, every compaction pass over a group that contains any legacy file will quietly drop lineage on the surviving rows. That's the common case during migration, not the edge case.
I'd at minimum add a `slog.Warn` matching the v3 pos-delete pattern, and ideally a `RowLineagePreserved bool` on `CompactionGroupResult` so callers can observe it.
##########
table/scanner_internal_test.go:
##########
@@ -238,98 +236,71 @@ func TestBuildPartitionEvaluatorWithInvalidSpecID(t *testing.T) {
assert.ErrorContains(t, err, "id 999")
}
-// TestProjectionV3PreLineageFile verifies that Projection() succeeds and returns
-// _row_id and _last_updated_sequence_number as nullable (all-null-capable) fields when
-// the table is v3 with next-row-id set but the data file predates row lineage (those
-// columns are absent from the schema).
-func TestProjectionV3PreLineageFile(t *testing.T) {
- schema := iceberg.NewSchema(
- 1,
- iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
- iceberg.NestedField{ID: 2, Name: "payload", Type: iceberg.PrimitiveTypes.String, Required: false},
- )
+// TestSynthesizeRowLineageColumns verifies that _row_id and _last_updated_sequence_number
+// are filled from task constants when those columns are present and null.
+func TestSynthesizeRowLineageColumns(t *testing.T) {
Review Comment:
The test only constructs all-null inputs, so it doesn't actually prove the property the feature depends on. If `synthesizeRowLineageColumns` had a bug where it overwrote pre-existing non-null values, this test would still pass — and so would `TestCoWRewritePreservesRowID`, because fresh appends produce all-null source columns too.
The case that matters is the mixed one: some rows have explicit `_row_id` / `_last_updated_sequence_number` values from a prior write, the rest are null. That only arises after multiple rewrites of files that already carry explicit values, which is exactly the scenario the spec's null-coalescing rule exists for.
I'd add a fixture that builds a record with three rows — one explicit non-null, one null, one explicit non-null with a different value — and asserts the non-null ones survive untouched while the null one picks up `firstRowID + position`. Same shape for `_last_updated_sequence_number`.
##########
metadata_columns.go:
##########
@@ -59,3 +61,38 @@ func LastUpdatedSequenceNumber() NestedField {
func IsMetadataColumn(fieldID int) bool {
return fieldID == RowIDFieldID || fieldID == LastUpdatedSequenceNumberFieldID
}
+
+// SchemaWithRowLineage returns a new schema with the row-lineage metadata columns
+// (_row_id, _last_updated_sequence_number) appended to the given schema's fields.
+// Used when reading source files during a CoW rewrite or compaction so that row
+// identity and per-row update sequence are preserved in the output.
+//
+// Idempotent: if a row-lineage column is already present (by reserved field ID),
+// it is not appended again. The returned schema always allocates a fresh field
+// slice so it cannot alias the input schema's backing array.
+func SchemaWithRowLineage(s *Schema) *Schema {
Review Comment:
Two things on the new helper.
The comment a few lines up reads as if the second reserved ID is literally 108. Should be "Integer.MAX_VALUE - 107 and Integer.MAX_VALUE - 108" — `2147483540` and `2147483539` respectively.
More substantively, `SchemaWithRowLineage` lives in the top-level `iceberg` package, which makes it part of the v1 public API surface. Java keeps the equivalent logic internal. I'd consider an unexported `schemaWithRowLineage` in `table`, or a method on `*Schema`, so we're not locked into the helper's exact shape if v4 lineage semantics shift. Same for `WithPreserveRowLineage(schema)` — making callers construct the schema feels backwards; could it take no arg and derive from the table internally?
Also worth a nil guard on `s` at the top here — it's exported, a nil receiver will panic on `s.Fields()`.
##########
table/transaction.go:
##########
@@ -1530,14 +1565,50 @@ func (t *Transaction) rewriteFilesWithFilter(ctx context.Context, fs io.IO, upda
return nil
}
-// rewriteSingleFile reads a single data file, applies the filter, and writes new files with filtered data
-func (t *Transaction) rewriteSingleFile(ctx context.Context, fs io.IO, originalFile iceberg.DataFile, filter iceberg.BooleanExpression, caseSensitive bool, commitUUID uuid.UUID, concurrency int) ([]iceberg.DataFile, error) {
+// rewriteSingleFile reads a single data file, applies the filter, and writes new files with filtered data.
+// fileSeqNum is the source file's file_sequence_number from its manifest entry; required to synthesize
+// _last_updated_sequence_number for rows whose value is null in the source file.
+func (t *Transaction) rewriteSingleFile(ctx context.Context, fs io.IO, originalFile iceberg.DataFile, fileSeqNum *int64, filter iceberg.BooleanExpression, postFilter func(arrow.RecordBatch) (arrow.RecordBatch, error), caseSensitive bool, commitUUID uuid.UUID, concurrency int) ([]iceberg.DataFile, error) {
Review Comment:
Nine positional parameters with several same-typed adjacents (`commitUUID, concurrency`, and now two same-shaped filter closures) — no compile-time order protection if anyone swaps a pair. I'd pull these into a `rewriteSingleFileArgs` struct.
Separate but related: `BindExpr` runs unconditionally in `rewriteSingleFile`, but when `preserveRowLineage == true` the bound filter is unused (scanFilter becomes `AlwaysTrue{}`) and the same expression is bound again inside `prepareBatchFilter`. With `preserveRowLineage == false`, `prepareBatchFilter` runs `BindExpr` + `ConvertExpr` but `postFilter` is never called. One of those two binds is always wasted.
##########
table/scanner.go:
##########
@@ -261,26 +262,96 @@ func (scan *Scan) Projection() (*iceberg.Schema, error) {
}
}
- if slices.Contains(scan.selectedFields, "*") {
- return curSchema, nil
+ if scan.includeRowLineage && curVersion < minFormatVersionRowLineage {
+ return nil, fmt.Errorf("%w: row lineage requires format version %d, table is v%d",
+ ErrInvalidOperation, minFormatVersionRowLineage, curVersion)
}
- selectedFieldsMeta := metaFieldsFromSelectedFields(scan.selectedFields, caseSensitive)
- schemaMeta := metaFieldsFromSchema(curSchema)
- synthesisMeta := synthesizeMeta(selectedFieldsMeta, schemaMeta)
- if len(synthesisMeta) > 0 && curVersion >= minFormatVersionRowLineage {
+ var schema *iceberg.Schema
+ if slices.Contains(scan.selectedFields, "*") {
+ schema = curSchema
+ } else {
+ // Intercept row-lineage metadata column names (_row_id,
+ // _last_updated_sequence_number) before calling Select: they are
+ // reserved and never appear in the user schema's fields, so
+ // Select would fail with "could not find column" on v3 tables
+ // where they are otherwise legal to project. The scanner reads
+ // them from file metadata (or synthesizes them) at scan time;
+ // here we just need to ensure they survive into the projection.
+ //
+ // On v1/v2 tables, Select is left to fail naturally — those
+ // versions don't have row lineage, so requesting these columns
+ // is an error.
+ userFields, lineageFields := splitLineageMetadataFields(scan.selectedFields, scan.caseSensitive)
+ if len(lineageFields) > 0 && curVersion < minFormatVersionRowLineage {
Review Comment:
The rejection here works only because `Select` happens to fail when `_row_id` isn't in the v1/v2 schema — the user sees `ErrInvalidSchema` rather than a clear "row lineage requires v3". And it's fragile: if a future v2 schema ever carries a column literally named `_row_id` for unrelated reasons, this short-circuits in the wrong direction.
I'd return an explicit error here for the `curVersion < minFormatVersionRowLineage && len(lineageFields) > 0` case, naming the format version requirement. Same logic, but the contract is in the code rather than emergent from `Select`'s error path.
##########
table/transaction.go:
##########
@@ -1798,3 +1905,53 @@ func (s *StagedTable) Refresh(ctx context.Context) (*Table, error) {
func (s *StagedTable) Scan(opts ...ScanOption) *Scan {
panic(fmt.Errorf("%w: cannot scan a staged table", ErrInvalidOperation))
}
+
+// prepareBatchFilter binds the given Iceberg filter against schema and converts
+// it to substrait once, returning a per-batch filter function that can be
+// reused across every record batch. The setup work (BindExpr, ConvertExpr) is
+// independent of the batch and is the most expensive part of filter-eval, so
+// hoisting it out of the iterator loop is a measurable win on rewrites that
+// produce many batches.
+//
+// The returned function takes ownership of the input batch (it releases it on
+// the AlwaysFalse fast-path) and returns a possibly-new batch the caller is
+// responsible for releasing.
+func prepareBatchFilter(ctx context.Context, filter iceberg.BooleanExpression, schema *iceberg.Schema, caseSensitive bool) (func(arrow.RecordBatch) (arrow.RecordBatch, error), error) {
+ if filter == nil || filter.Equals(iceberg.AlwaysTrue{}) {
+ return func(rec arrow.RecordBatch) (arrow.RecordBatch, error) {
+ return rec, nil
+ }, nil
+ }
+
+ bound, err := iceberg.BindExpr(schema, filter, caseSensitive)
+ if err != nil {
+ return nil, fmt.Errorf("prepareBatchFilter: bind expression: %w", err)
+ }
+
+ if bound == nil {
+ return func(rec arrow.RecordBatch) (arrow.RecordBatch, error) {
+ return rec, nil
+ }, nil
+ }
+
+ if bound.Equals(iceberg.AlwaysFalse{}) {
+ // Return a record with the same schema and zero rows. NewSlice(0, 0)
+ // preserves the per-field arrays so downstream code that calls
+ // rec.Column(i) does not panic on the empty result.
+ return func(rec arrow.RecordBatch) (arrow.RecordBatch, error) {
+ defer rec.Release()
+
+ return rec.NewSlice(0, 0), nil
+ }, nil
+ }
+
+ extSet, substraitFilter, err := substrait.ConvertExpr(schema, bound, caseSensitive)
+ if err != nil {
+ return nil, fmt.Errorf("prepareBatchFilter: convert expression: %w", err)
+ }
+
+ ctx = exprs.WithExtensionIDSet(ctx, exprs.NewExtensionSetDefault(*extSet))
Review Comment:
The closure captures `ctx` at construction time with the extension-ID set baked in, but per-batch allocator and deadline values from the call-site ctx are then lost. If `filterRecords` resolves the allocator from ctx, intermediate arrays end up on the construction-time allocator and `memory.NewCheckedAllocator` accounting goes sideways under test.
I'd take the call-site ctx as a parameter on the returned closure, derive the extension-set context inside, and let allocator/deadline propagate.
While we're in here: the positional-binding assumption (lineage fields at indices N..N+1 in the record batch, filter bound against indices 0..N-1) is correct but undocumented. Worth a one-line comment so a future change to lineage column placement doesn't silently break filter evaluation.
##########
table/scanner.go:
##########
@@ -261,26 +262,96 @@ func (scan *Scan) Projection() (*iceberg.Schema, error) {
}
}
- if slices.Contains(scan.selectedFields, "*") {
- return curSchema, nil
+ if scan.includeRowLineage && curVersion < minFormatVersionRowLineage {
+ return nil, fmt.Errorf("%w: row lineage requires format version %d, table is v%d",
+ ErrInvalidOperation, minFormatVersionRowLineage, curVersion)
}
- selectedFieldsMeta := metaFieldsFromSelectedFields(scan.selectedFields, caseSensitive)
- schemaMeta := metaFieldsFromSchema(curSchema)
- synthesisMeta := synthesizeMeta(selectedFieldsMeta, schemaMeta)
- if len(synthesisMeta) > 0 && curVersion >= minFormatVersionRowLineage {
+ var schema *iceberg.Schema
+ if slices.Contains(scan.selectedFields, "*") {
+ schema = curSchema
+ } else {
+ // Intercept row-lineage metadata column names (_row_id,
+ // _last_updated_sequence_number) before calling Select: they are
+ // reserved and never appear in the user schema's fields, so
+ // Select would fail with "could not find column" on v3 tables
+ // where they are otherwise legal to project. The scanner reads
+ // them from file metadata (or synthesizes them) at scan time;
+ // here we just need to ensure they survive into the projection.
+ //
+ // On v1/v2 tables, Select is left to fail naturally — those
+ // versions don't have row lineage, so requesting these columns
+ // is an error.
+ userFields, lineageFields := splitLineageMetadataFields(scan.selectedFields, scan.caseSensitive)
+ if len(lineageFields) > 0 && curVersion < minFormatVersionRowLineage {
+ userFields = scan.selectedFields
+ lineageFields = nil
+ }
- // synthesis path
- removedMetaSlice, missingMetaFields := removeMetadataFromSelectedFields(scan.selectedFields, synthesisMeta)
- sch, err := curSchema.Select(scan.caseSensitive, removedMetaSlice...)
+ var err error
+ schema, err = curSchema.Select(scan.caseSensitive, userFields...)
if err != nil {
return nil, err
}
+ if len(lineageFields) > 0 {
+ schema = appendMissingLineageFields(schema, lineageFields)
+ }
+ }
+
+ if scan.includeRowLineage {
Review Comment:
There are now two code paths that select lineage columns into the projection: the `includeRowLineage` flag path (calls `SchemaWithRowLineage`) and the explicit-field-name path (calls `appendMissingLineageFields`). Both run when the user supplied lineage field names *and* set `WithRowLineage`, producing a redundant pass.
More concerning, they have subtly different error behavior — one rejects on v1/v2 with a clear error, the other relies on `Select` to fail. Anyone adding a third lineage column down the road has to update both paths and both test sets.
I'd canonicalize on one path: detect lineage fields in `selectedFields` up front, set `scan.includeRowLineage = true` if any are present, drop them from `selectedFields`, and from there only the `includeRowLineage` branch runs. wdyt?
##########
table/transaction.go:
##########
@@ -1798,3 +1905,53 @@ func (s *StagedTable) Refresh(ctx context.Context) (*Table, error) {
func (s *StagedTable) Scan(opts ...ScanOption) *Scan {
panic(fmt.Errorf("%w: cannot scan a staged table", ErrInvalidOperation))
}
+
+// prepareBatchFilter binds the given Iceberg filter against schema and converts
+// it to substrait once, returning a per-batch filter function that can be
+// reused across every record batch. The setup work (BindExpr, ConvertExpr) is
+// independent of the batch and is the most expensive part of filter-eval, so
+// hoisting it out of the iterator loop is a measurable win on rewrites that
+// produce many batches.
+//
+// The returned function takes ownership of the input batch (it releases it on
+// the AlwaysFalse fast-path) and returns a possibly-new batch the caller is
+// responsible for releasing.
+func prepareBatchFilter(ctx context.Context, filter iceberg.BooleanExpression, schema *iceberg.Schema, caseSensitive bool) (func(arrow.RecordBatch) (arrow.RecordBatch, error), error) {
Review Comment:
The `AlwaysFalse` fast-path does `defer rec.Release()`; the `AlwaysTrue` / bound-nil paths return `rec` unchanged with no Retain. So depending on which fast-path triggers, the caller either does or doesn't own a reference to the returned record. That's two contracts for one return type, and the next caller to add a `Release()` will double-free on the AlwaysFalse path.
I'd have every fast-path do a `rec.Retain()` so the caller always owns exactly one reference on the result, mirroring the bound-filter path's semantics.
##########
table/row_lineage_rewrite_test.go:
##########
@@ -0,0 +1,291 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table_test
+
+import (
+ "context"
+ "path/filepath"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func newV3RowLineageTestTable(t *testing.T) *table.Table {
+ t.Helper()
+
+ location := filepath.ToSlash(t.TempDir())
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name: "data", Type: iceberg.PrimitiveTypes.String, Required: false},
+ )
+ meta, err := table.NewMetadata(schema, iceberg.UnpartitionedSpec, table.UnsortedSortOrder, location,
+ iceberg.Properties{table.PropertyFormatVersion: "3"})
+ require.NoError(t, err)
+
+ metaLoc := location + "/metadata/v1.metadata.json"
+ fsF := func(context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil }
+ cat := &concurrentTestCatalog{metadata: meta, location: metaLoc, fsF: fsF}
+
+ return table.New(table.Identifier{"db", "row_lineage_test"}, meta, metaLoc, fsF, cat)
+}
+
+// TestCoWRewritePreservesRowID verifies that a copy-on-write overwrite with a
+// row filter preserves the original _row_id and _last_updated_sequence_number
+// values in the rewritten file. Surviving rows must keep both values from the
+// pre-rewrite snapshot — the rewrite is "physically rewritten", not "logically
+// updated", per the v3 spec.
+func TestCoWRewritePreservesRowID(t *testing.T) {
+ ctx := context.Background()
+ mem := memory.DefaultAllocator
+
+ tbl := newV3RowLineageTestTable(t)
+
+ // Append 3 rows: id=1,2,3
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
+ {Name: "data", Type: arrow.BinaryTypes.String, Nullable: true},
+ }, nil)
+
+ initialData, err := array.TableFromJSON(mem, arrowSchema, []string{
+ `[{"id": 1, "data": "a"}, {"id": 2, "data": "b"}, {"id": 3, "data": "c"}]`,
+ })
+ require.NoError(t, err)
+ defer initialData.Release()
+
+ tbl, err = tbl.Append(ctx, array.NewTableReader(initialData, -1), nil)
+ require.NoError(t, err)
+
+ // Verify the append created a valid v3 snapshot with row lineage.
+ snap := tbl.CurrentSnapshot()
+ require.NotNil(t, snap)
+ require.NotNil(t, snap.FirstRowID, "v3 snapshot must have first-row-id")
+ require.NotNil(t, snap.AddedRows, "v3 snapshot must have added-rows")
+ assert.Equal(t, int64(0), *snap.FirstRowID)
+ assert.Equal(t, int64(3), *snap.AddedRows)
+
+ // Capture the snapshot's sequence number so we can assert preservation
+ // after the rewrite. After the append, every row's effective
+ // _last_updated_sequence_number should be this value.
+ createSeq := snap.SequenceNumber
+
+ // Scan with row lineage to see synthesized _row_id values before the rewrite.
+ lineageScan := tbl.Scan(table.WithRowLineage())
+ schema, itr, err := lineageScan.ToArrowRecords(ctx)
+ require.NoError(t, err)
+
+ rowIDIdx := -1
+ for i, f := range schema.Fields() {
+ if f.Name == iceberg.RowIDColumnName {
+ rowIDIdx = i
+
+ break
+ }
+ }
+ require.GreaterOrEqual(t, rowIDIdx, 0, "_row_id should be in scan projection")
+
+ var originalRowIDs []int64
+ for rec, err := range itr {
+ require.NoError(t, err)
+ col := rec.Column(rowIDIdx).(*array.Int64)
+ for i := 0; i < col.Len(); i++ {
+ originalRowIDs = append(originalRowIDs, col.Value(i))
+ }
+ rec.Release()
+ }
+ require.Equal(t, []int64{0, 1, 2}, originalRowIDs, "initial _row_id should be 0,1,2")
+
+ // CoW overwrite: delete the row where id=2, preserving id=1 and id=3.
+ filter := iceberg.EqualTo(iceberg.Reference("id"), int64(2))
+ tbl, err = tbl.Delete(ctx, filter, nil)
+ require.NoError(t, err)
+
+ snap = tbl.CurrentSnapshot()
+ require.NotNil(t, snap)
+ require.Greater(t, snap.SequenceNumber, createSeq,
+ "sanity: rewrite snapshot must have a higher sequence number than the create snapshot")
+
+ // Scan the result with row lineage. The surviving rows should preserve their
+ // original _row_id values: 0 and 2. Their _last_updated_sequence_number must
+ // also still report the create snapshot's seq, NOT the rewrite snapshot's
+ // seq — the rewrite is physical only, not a logical update.
+ lineageScan = tbl.Scan(table.WithRowLineage())
+ _, itr, err = lineageScan.ToArrowRecords(ctx)
+ require.NoError(t, err)
+
+ var afterRowIDs []int64
+ var afterIDs []int64
+ var afterSeq []int64
+ for rec, err := range itr {
+ require.NoError(t, err)
+ idIdx := rec.Schema().FieldIndices("id")
+ require.NotEmpty(t, idIdx)
+ rowIDIndices := rec.Schema().FieldIndices(iceberg.RowIDColumnName)
+ require.NotEmpty(t, rowIDIndices)
+ seqIndices := rec.Schema().FieldIndices(iceberg.LastUpdatedSequenceNumberColumnName)
+ require.NotEmpty(t, seqIndices, "_last_updated_sequence_number must be in projection")
+
+ idCol := rec.Column(idIdx[0]).(*array.Int64)
+ rowIDCol := rec.Column(rowIDIndices[0]).(*array.Int64)
+ seqCol := rec.Column(seqIndices[0]).(*array.Int64)
+ for i := 0; i < int(rec.NumRows()); i++ {
+ afterIDs = append(afterIDs, idCol.Value(i))
+ afterRowIDs = append(afterRowIDs, rowIDCol.Value(i))
+ require.False(t, seqCol.IsNull(i),
+ "row %d must have a non-null _last_updated_sequence_number after CoW rewrite", i)
+ afterSeq = append(afterSeq, seqCol.Value(i))
+ }
+ rec.Release()
+ }
+
+ assert.Equal(t, []int64{1, 3}, afterIDs, "remaining rows should be id=1,3")
+ assert.Equal(t, []int64{0, 2}, afterRowIDs,
+ "_row_id must be preserved through CoW rewrite: row with id=1 keeps _row_id=0, row with id=3 keeps _row_id=2")
+ assert.Equal(t, []int64{createSeq, createSeq}, afterSeq,
+ "_last_updated_sequence_number must report the original creation snapshot's sequence number, not the rewrite's")
+}
+
+// TestCoWRewriteRowIDNextRowIDAccounting verifies that row-id accounting remains
+// correct after a CoW rewrite. The overcounting (where next-row-id advances by
+// the full manifest row count including preserved survivors) is intentional and
+// matches Java's ManifestListWriter.V3Writer behavior.
+func TestCoWRewriteRowIDNextRowIDAccounting(t *testing.T) {
+ ctx := context.Background()
+ mem := memory.DefaultAllocator
+
+ tbl := newV3RowLineageTestTable(t)
+
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
+ {Name: "data", Type: arrow.BinaryTypes.String, Nullable: true},
+ }, nil)
+
+ data, err := array.TableFromJSON(mem, arrowSchema, []string{
+ `[{"id": 10, "data": "x"}, {"id": 20, "data": "y"}, {"id": 30, "data": "z"}]`,
+ })
+ require.NoError(t, err)
+ defer data.Release()
+
+ tbl, err = tbl.Append(ctx, array.NewTableReader(data, -1), nil)
+ require.NoError(t, err)
+
+ // next-row-id should be 3 after appending 3 rows.
+ assert.Equal(t, int64(3), tbl.Metadata().NextRowID())
+
+ // Delete one row via CoW.
+ filter := iceberg.EqualTo(iceberg.Reference("id"), int64(20))
+ tbl, err = tbl.Delete(ctx, filter, nil)
+ require.NoError(t, err)
+
+ // next-row-id advances by 3 (original) + 2 (rewritten survivors) = 5,
+ // even though the surviving rows preserve their old IDs. This "wastes"
+ // ID space but doesn't violate uniqueness — actual row IDs come from
+ // the explicit Parquet column, not the global counter.
+ assert.Equal(t, int64(5), tbl.Metadata().NextRowID(),
+ "next-row-id should advance by original (3) + rewritten (2) = 5")
Review Comment:
The comment is arithmetically misleading — the advance is `2` (the new manifest's added rows), and `5 = firstRowID(3) + 2`. The "3" is the starting value, not an addend. Worth rewriting the comment to match the actual computation, and linking to the `snapshot_producers.go` site that mirrors `ManifestListWriter.V3Writer.prepare()` in Java.
Separately, I'd double-check `snapshot.AddedRows` here — Java's `added-rows` summary field is the actual added row count, not `existing + added`. If we're stamping the Java-equivalent of `existing + added` into the snapshot summary, downstream tools reading that field will see inflated numbers.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): preserve row lineage through copy-on-write rewrites [iceberg-go]
Posted by "tanmayrauth (via GitHub)" <gi...@apache.org>.
tanmayrauth commented on PR #1099:
URL: https://github.com/apache/iceberg-go/pull/1099#issuecomment-4499983531
@laskoviymishka can you please review it?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): preserve row lineage through copy-on-write rewrites [iceberg-go]
Posted by "tanmayrauth (via GitHub)" <gi...@apache.org>.
tanmayrauth commented on code in PR #1099:
URL: https://github.com/apache/iceberg-go/pull/1099#discussion_r3291870459
##########
table/transaction.go:
##########
@@ -1798,3 +1905,53 @@ func (s *StagedTable) Refresh(ctx context.Context) (*Table, error) {
func (s *StagedTable) Scan(opts ...ScanOption) *Scan {
panic(fmt.Errorf("%w: cannot scan a staged table", ErrInvalidOperation))
}
+
+// prepareBatchFilter binds the given Iceberg filter against schema and converts
+// it to substrait once, returning a per-batch filter function that can be
+// reused across every record batch. The setup work (BindExpr, ConvertExpr) is
+// independent of the batch and is the most expensive part of filter-eval, so
+// hoisting it out of the iterator loop is a measurable win on rewrites that
+// produce many batches.
+//
+// The returned function takes ownership of the input batch (it releases it on
+// the AlwaysFalse fast-path) and returns a possibly-new batch the caller is
+// responsible for releasing.
+func prepareBatchFilter(ctx context.Context, filter iceberg.BooleanExpression, schema *iceberg.Schema, caseSensitive bool) (func(arrow.RecordBatch) (arrow.RecordBatch, error), error) {
+ if filter == nil || filter.Equals(iceberg.AlwaysTrue{}) {
+ return func(rec arrow.RecordBatch) (arrow.RecordBatch, error) {
+ return rec, nil
+ }, nil
+ }
+
+ bound, err := iceberg.BindExpr(schema, filter, caseSensitive)
+ if err != nil {
+ return nil, fmt.Errorf("prepareBatchFilter: bind expression: %w", err)
+ }
+
+ if bound == nil {
+ return func(rec arrow.RecordBatch) (arrow.RecordBatch, error) {
+ return rec, nil
+ }, nil
+ }
+
+ if bound.Equals(iceberg.AlwaysFalse{}) {
+ // Return a record with the same schema and zero rows. NewSlice(0, 0)
+ // preserves the per-field arrays so downstream code that calls
+ // rec.Column(i) does not panic on the empty result.
+ return func(rec arrow.RecordBatch) (arrow.RecordBatch, error) {
+ defer rec.Release()
+
+ return rec.NewSlice(0, 0), nil
+ }, nil
+ }
+
+ extSet, substraitFilter, err := substrait.ConvertExpr(schema, bound, caseSensitive)
+ if err != nil {
+ return nil, fmt.Errorf("prepareBatchFilter: convert expression: %w", err)
+ }
+
+ ctx = exprs.WithExtensionIDSet(ctx, exprs.NewExtensionSetDefault(*extSet))
Review Comment:
Changed the closure signature to func(context.Context, arrow.RecordBatch) — extSet is pre-computed once and reused, but the per-call ctx (with allocator/deadline) is attached inside the closure on each call. Also added a one-line comment on the positional-binding assumption (lineage fields at indices N..N+1, filter bound against 0..N-1).
##########
table/scanner.go:
##########
@@ -261,26 +262,96 @@ func (scan *Scan) Projection() (*iceberg.Schema, error) {
}
}
- if slices.Contains(scan.selectedFields, "*") {
- return curSchema, nil
+ if scan.includeRowLineage && curVersion < minFormatVersionRowLineage {
+ return nil, fmt.Errorf("%w: row lineage requires format version %d, table is v%d",
+ ErrInvalidOperation, minFormatVersionRowLineage, curVersion)
}
- selectedFieldsMeta := metaFieldsFromSelectedFields(scan.selectedFields, caseSensitive)
- schemaMeta := metaFieldsFromSchema(curSchema)
- synthesisMeta := synthesizeMeta(selectedFieldsMeta, schemaMeta)
- if len(synthesisMeta) > 0 && curVersion >= minFormatVersionRowLineage {
+ var schema *iceberg.Schema
+ if slices.Contains(scan.selectedFields, "*") {
+ schema = curSchema
+ } else {
+ // Intercept row-lineage metadata column names (_row_id,
+ // _last_updated_sequence_number) before calling Select: they are
+ // reserved and never appear in the user schema's fields, so
+ // Select would fail with "could not find column" on v3 tables
+ // where they are otherwise legal to project. The scanner reads
+ // them from file metadata (or synthesizes them) at scan time;
+ // here we just need to ensure they survive into the projection.
+ //
+ // On v1/v2 tables, Select is left to fail naturally — those
+ // versions don't have row lineage, so requesting these columns
+ // is an error.
+ userFields, lineageFields := splitLineageMetadataFields(scan.selectedFields, scan.caseSensitive)
+ if len(lineageFields) > 0 && curVersion < minFormatVersionRowLineage {
+ userFields = scan.selectedFields
+ lineageFields = nil
+ }
- // synthesis path
- removedMetaSlice, missingMetaFields := removeMetadataFromSelectedFields(scan.selectedFields, synthesisMeta)
- sch, err := curSchema.Select(scan.caseSensitive, removedMetaSlice...)
+ var err error
+ schema, err = curSchema.Select(scan.caseSensitive, userFields...)
if err != nil {
return nil, err
}
+ if len(lineageFields) > 0 {
+ schema = appendMissingLineageFields(schema, lineageFields)
+ }
+ }
+
+ if scan.includeRowLineage {
Review Comment:
Did the smaller piece — skipping appendMissingLineageFields when scan.includeRowLineage is set so the redundant pass goes away. Pushed back on the full canonicalization (auto-setting includeRowLineage=true whenever any lineage name appears in selectedFields): selecting only _row_id today gives you only _row_id, but with the rewrite it would also pull in _last_updated_sequence_number. That's a behavior change for the legacy Select(_row_id) escape hatch. The maintainability concern (third column down the road) is mostly handled by splitLineageMetadataFields + SchemaWithRowLineage already centralizing the column list.
##########
table/scanner.go:
##########
@@ -261,26 +262,96 @@ func (scan *Scan) Projection() (*iceberg.Schema, error) {
}
}
- if slices.Contains(scan.selectedFields, "*") {
- return curSchema, nil
+ if scan.includeRowLineage && curVersion < minFormatVersionRowLineage {
+ return nil, fmt.Errorf("%w: row lineage requires format version %d, table is v%d",
+ ErrInvalidOperation, minFormatVersionRowLineage, curVersion)
}
- selectedFieldsMeta := metaFieldsFromSelectedFields(scan.selectedFields, caseSensitive)
- schemaMeta := metaFieldsFromSchema(curSchema)
- synthesisMeta := synthesizeMeta(selectedFieldsMeta, schemaMeta)
- if len(synthesisMeta) > 0 && curVersion >= minFormatVersionRowLineage {
+ var schema *iceberg.Schema
+ if slices.Contains(scan.selectedFields, "*") {
+ schema = curSchema
+ } else {
+ // Intercept row-lineage metadata column names (_row_id,
+ // _last_updated_sequence_number) before calling Select: they are
+ // reserved and never appear in the user schema's fields, so
+ // Select would fail with "could not find column" on v3 tables
+ // where they are otherwise legal to project. The scanner reads
+ // them from file metadata (or synthesizes them) at scan time;
+ // here we just need to ensure they survive into the projection.
+ //
+ // On v1/v2 tables, Select is left to fail naturally — those
+ // versions don't have row lineage, so requesting these columns
+ // is an error.
+ userFields, lineageFields := splitLineageMetadataFields(scan.selectedFields, scan.caseSensitive)
+ if len(lineageFields) > 0 && curVersion < minFormatVersionRowLineage {
Review Comment:
Done, explicit ErrInvalidOperation with the format version named, rather than relying on Select's ErrInvalidSchema. Updated TestProjectionRowLineageRejectedOnV1V2 to match.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(table): preserve row lineage through copy-on-write rewrites [iceberg-go]
Posted by "tanmayrauth (via GitHub)" <gi...@apache.org>.
tanmayrauth commented on code in PR #1099:
URL: https://github.com/apache/iceberg-go/pull/1099#discussion_r3291870842
##########
table/rewrite_data_files.go:
##########
@@ -289,6 +289,20 @@ func ExecuteCompactionGroup(ctx context.Context, tbl *Table, group CompactionTas
scanOpts = append(scanOpts, WitMaxConcurrency(cfg.scanConcurrency))
}
+ // Preserve row lineage only when every source file in the group carries
+ // it. A mixed group (some files with FirstRowID, some without — e.g.
+ // legacy files on a v3 table) would otherwise produce one output where
+ // post-lineage rows have explicit _row_id values and pre-lineage rows
+ // have nulls, which violates the per-file uniqueness/coverage
+ // invariant the v3 spec requires. Splitting mixed groups into separate
+ // outputs is a larger refactor and is left as a follow-up; for now we
+ // degrade gracefully (the rewrite still succeeds, but lineage is not
+ // preserved for the surviving rows).
+ preserveLineage := tbl.metadata.Version() >= 3 && allTasksHaveRowLineage(group.Tasks)
Review Comment:
Added a slog.Warn matching the v3 pos-delete pattern in ExecuteCompactionGroup when a v3 table sees a mixed group with at least one lineage file. Holding off on the RowLineagePreserved bool on CompactionGroupResult for now — the log surfaces the silent loss, and adding to the result struct locks in API surface for what should be a transitional migration concern. Happy to add it as a follow-up if it shows up in practice.
##########
table/scanner_internal_test.go:
##########
@@ -238,98 +236,71 @@ func TestBuildPartitionEvaluatorWithInvalidSpecID(t *testing.T) {
assert.ErrorContains(t, err, "id 999")
}
-// TestProjectionV3PreLineageFile verifies that Projection() succeeds and returns
-// _row_id and _last_updated_sequence_number as nullable (all-null-capable) fields when
-// the table is v3 with next-row-id set but the data file predates row lineage (those
-// columns are absent from the schema).
-func TestProjectionV3PreLineageFile(t *testing.T) {
- schema := iceberg.NewSchema(
- 1,
- iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
- iceberg.NestedField{ID: 2, Name: "payload", Type: iceberg.PrimitiveTypes.String, Required: false},
- )
+// TestSynthesizeRowLineageColumns verifies that _row_id and _last_updated_sequence_number
+// are filled from task constants when those columns are present and null.
+func TestSynthesizeRowLineageColumns(t *testing.T) {
Review Comment:
Added TestSynthesizeRowLineageColumnsPreservesExplicit with the 3-row mixed fixture (explicit / null / explicit-different) for both _row_id and _last_updated_sequence_number. Asserts the non-null rows survive untouched and the null row picks up firstRowID + position / task.DataSequenceNumber.
##########
table/transaction.go:
##########
@@ -1488,13 +1501,23 @@ func (t *Transaction) classifyFilesForFilteredDeletions(ctx context.Context, fs
localDelete = append(localDelete, df)
} else {
localRewrite = append(localRewrite, df)
+ // Capture the file's sequence number from the
+ // manifest entry so the rewrite path can synthesize
+ // _last_updated_sequence_number for source rows
+ // that have a null value (or no column) in the file.
+ if fseq := entry.FileSequenceNum(); fseq != nil {
Review Comment:
Good catch, switched both sites to entry.SequenceNum() and added a guard on >= 0 (since SequenceNum() returns -1 for unset). Left FileScanTask.DataSequenceNumber named as-is since the field name already matches what we're now putting in it (data sequence number); the bug was the source, not the destination.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org