laskoviymishka commented on code in PR #2170: URL: https://github.com/apache/iceberg-go/pull/2170#discussion_r4241786776
########## table/changelog_reader.go: ########## @@ -0,0 +1,293 @@ +// 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 + +import ( + "context" + "fmt" + "iter" + "maps" + "strconv" + "sync/atomic" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/iceberg-go" +) + +const ( + // ChangelogChangeTypeColumn is the changelog operation: INSERT, DELETE, + // UPDATE_BEFORE, or UPDATE_AFTER. + ChangelogChangeTypeColumn = "_change_type" + // ChangelogChangeOrdinalColumn is the zero-based ordinal of the snapshot + // that produced the change, shared by every row from that snapshot. + ChangelogChangeOrdinalColumn = "_change_ordinal" + // ChangelogCommitSnapshotIDColumn is the snapshot ID that committed the change. + ChangelogCommitSnapshotIDColumn = "_commit_snapshot_id" +) + +// ToArrowRecords plans the changelog and reads it. Added and removed data +// files are read with the deletes already stored on their scan task. A plan +// that includes a row-level delete returns an error instead of dropping those +// rows. +func (s *IncrementalChangelogScan) ToArrowRecords(ctx context.Context) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) { + tasks, err := s.PlanFiles(ctx) + if err != nil { + return nil, nil, err + } + + return s.Read(ctx, tasks) +} + +// Read materializes changelog rows for tasks. Each batch has the scan +// projection followed by _change_type, _change_ordinal, and +// _commit_snapshot_id. AddedRowsScanTask and DeletedDataFileScanTask are read +// with Scan.ReadTasks, so same-snapshot deletes and deletes that already +// applied to a removed file are applied before the changelog columns are +// added. DeletedRowsScanTask returns an error; those rows are produced by a +// later reader. The returned iterator is single-use. The caller releases every +// batch. +func (s *IncrementalChangelogScan) Read(ctx context.Context, tasks []ChangelogScanTask) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) { + if s == nil || s.scan == nil { + return nil, nil, fmt.Errorf("%w: incremental changelog scan is not initialized", ErrInvalidOperation) + } + if err := ctx.Err(); err != nil { + return nil, nil, err + } + for _, task := range tasks { + if err := readableChangelogTask(task); err != nil { + return nil, nil, err + } + } + + schema, useLargeTypes, err := changelogArrowSchema(s.scan) + if err != nil { + return nil, nil, err + } + if s.scan.limit == 0 || len(tasks) == 0 { + return schema, func(yield func(arrow.RecordBatch, error) bool) {}, nil + } + + readScan := s.scan.clone() + limit := s.scan.limit + + return schema, func(yield func(arrow.RecordBatch, error) bool) { + var emitted int64 + for _, task := range tasks { + if err := ctx.Err(); err != nil { + yield(nil, err) + + return + } + if limit > 0 { + readScan.limit = limit - emitted + } + + _, records, err := readScan.ReadTasks(ctx, []FileScanTask{task.ScanTask()}) + if err != nil { + yield(nil, err) + + return + } + + for rec, err := range records { + if err != nil { + yield(nil, err) + + return + } + + stamped, rows, err := stampChangelogBatch(ctx, schema, useLargeTypes, rec, task, limit, emitted) + rec.Release() + if err != nil { + yield(nil, err) + + return + } + if stamped == nil { + continue + } + if !yield(stamped, nil) { + return + } + emitted += rows + if limit > 0 && emitted >= limit { + return + } + } + } + }, nil +} + +func readableChangelogTask(task ChangelogScanTask) error { + switch task.(type) { + case AddedRowsScanTask, DeletedDataFileScanTask: + return nil + case DeletedRowsScanTask: + return fmt.Errorf("%w: reading deleted rows from added delete files is not supported", iceberg.ErrNotImplemented) + default: + return fmt.Errorf("%w: unsupported changelog scan task %T", iceberg.ErrInvalidArgument, task) + } +} + +func changelogArrowSchema(scan *Scan) (*arrow.Schema, bool, error) { + if atomic.LoadUint32(&scan.closed) != 0 { + return nil, false, fmt.Errorf("%w: scan is closed", ErrInvalidOperation) + } + if scan.selectorErr != nil { + return nil, false, scan.selectorErr + } + + projected, err := scan.Projection() + if err != nil { + return nil, false, err + } + + useLargeTypes, err := strconv.ParseBool(scan.options.Get(ScanOptionArrowUseLargeTypes, "false")) + if err != nil { + useLargeTypes = false + } + + props := scan.metadata.Properties() + if scan.arrowBatchSize > 0 { + props = maps.Clone(props) + if props == nil { + props = iceberg.Properties{} + } + props[ParquetBatchSizeKey] = strconv.Itoa(scan.arrowBatchSize) + } + + dataSchema, err := SchemaToArrowSchemaWithOptions(projected, ArrowSchemaOptions{ + UseLargeTypes: useLargeTypes, + TableProperties: props, + }) + if err != nil { + return nil, false, err + } + + schema := dataSchema + for _, field := range changelogArrowFields(useLargeTypes) { + schema, err = schema.AddField(schema.NumFields(), field) Review Comment: `AddField` will happily take a second `_change_type` if the table already projects a column by that name, and we'd hand back an ambiguous schema with no error. I'd check for a collision here and return an error, or at minimum note the reserved names in the doc, rather than letting it through silently. ########## table/changelog_reader.go: ########## @@ -0,0 +1,293 @@ +// 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 + +import ( + "context" + "fmt" + "iter" + "maps" + "strconv" + "sync/atomic" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/iceberg-go" +) + +const ( + // ChangelogChangeTypeColumn is the changelog operation: INSERT, DELETE, + // UPDATE_BEFORE, or UPDATE_AFTER. + ChangelogChangeTypeColumn = "_change_type" + // ChangelogChangeOrdinalColumn is the zero-based ordinal of the snapshot + // that produced the change, shared by every row from that snapshot. + ChangelogChangeOrdinalColumn = "_change_ordinal" + // ChangelogCommitSnapshotIDColumn is the snapshot ID that committed the change. + ChangelogCommitSnapshotIDColumn = "_commit_snapshot_id" +) + +// ToArrowRecords plans the changelog and reads it. Added and removed data Review Comment: I'd spell out here (and on `Read`) that these are raw file-level changes: a CoW overwrite that rewrites a file emits a DELETE and an INSERT for every unchanged row, carry-overs aren't collapsed, and `UPDATE_BEFORE`/`UPDATE_AFTER` are never produced. Java's raw scan behaves the same, but Spark's `create_changelog_view` strips carry-overs and pairs updates by default, so a CDC consumer wiring this straight into a sink will see delete+insert churn it probably doesn't expect. Just want the contract stated plainly. ########## table/changelog_reader_external_test.go: ########## @@ -0,0 +1,196 @@ +// 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" + "iter" + "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" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/require" +) + +func TestIncrementalChangelogReadEmptyTable(t *testing.T) { + schema, records, err := newV2RowLineageTestTable(t).NewIncrementalChangelogScan().ToArrowRecords(t.Context()) + require.NoError(t, err) + require.Equal(t, []string{ + "id", "data", + table.ChangelogChangeTypeColumn, + table.ChangelogChangeOrdinalColumn, + table.ChangelogCommitSnapshotIDColumn, + }, schemaFieldNames(schema)) + + require.Empty(t, collectChangelogRows(t, records)) +} + +func TestIncrementalChangelogReadFileChanges(t *testing.T) { + ctx := t.Context() + tbl := changelogOverwriteTable(t, ctx) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(ctx) + require.NoError(t, err) + require.Len(t, tasks, 3) + + schema, records, err := tbl.NewIncrementalChangelogScan().Read(ctx, tasks) + require.NoError(t, err) + require.Equal(t, []string{ + "id", "data", + table.ChangelogChangeTypeColumn, + table.ChangelogChangeOrdinalColumn, + table.ChangelogCommitSnapshotIDColumn, + }, schemaFieldNames(schema)) + + got := collectChangelogRows(t, records) + require.Equal(t, []changelogRow{ + {id: 1, data: "a", op: string(table.ChangelogOpInsert), ordinal: int64(tasks[0].ChangeOrdinal()), snapshot: tasks[0].CommitSnapshotID()}, + {id: 1, data: "a", op: string(table.ChangelogOpDelete), ordinal: int64(tasks[1].ChangeOrdinal()), snapshot: tasks[1].CommitSnapshotID()}, + {id: 2, data: "b", op: string(table.ChangelogOpInsert), ordinal: int64(tasks[2].ChangeOrdinal()), snapshot: tasks[2].CommitSnapshotID()}, + }, got) + require.NotEqual(t, tasks[0].CommitSnapshotID(), tasks[1].CommitSnapshotID()) + require.Equal(t, tasks[1].CommitSnapshotID(), tasks[2].CommitSnapshotID()) + require.Equal(t, 0, tasks[0].ChangeOrdinal()) + require.Equal(t, 1, tasks[1].ChangeOrdinal()) + require.Equal(t, 1, tasks[2].ChangeOrdinal()) +} + +func TestIncrementalChangelogReadHonorsProjectionFilterAndLimit(t *testing.T) { + ctx := t.Context() + tbl := changelogOverwriteTable(t, ctx) + + schema, records, err := tbl.NewIncrementalChangelogScan( + table.WithSelectedFields("id"), + table.WithRowFilter(iceberg.EqualTo(iceberg.Reference("id"), int64(2))), + table.WithLimit(1), Review Comment: `WithLimit(1)` here lands on a filter that already yields exactly one row (`id==2`), so the slice/limit-budget path (`NewSlice`, the `emitted+rows > limit` branch) never actually runs. I'd add a case where the limit cuts a multi-row batch mid-way. And since this PR is largely about Release correctness, I'd drive at least one read through a `memory.CheckedAllocator` (`compute.WithAllocator`) and assert zero outstanding bytes after draining and after an early break, cheapest way to catch a leak regression. ########## table/changelog_reader.go: ########## @@ -0,0 +1,293 @@ +// 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 + +import ( + "context" + "fmt" + "iter" + "maps" + "strconv" + "sync/atomic" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/iceberg-go" +) + +const ( + // ChangelogChangeTypeColumn is the changelog operation: INSERT, DELETE, + // UPDATE_BEFORE, or UPDATE_AFTER. + ChangelogChangeTypeColumn = "_change_type" + // ChangelogChangeOrdinalColumn is the zero-based ordinal of the snapshot + // that produced the change, shared by every row from that snapshot. + ChangelogChangeOrdinalColumn = "_change_ordinal" + // ChangelogCommitSnapshotIDColumn is the snapshot ID that committed the change. + ChangelogCommitSnapshotIDColumn = "_commit_snapshot_id" +) + +// ToArrowRecords plans the changelog and reads it. Added and removed data +// files are read with the deletes already stored on their scan task. A plan +// that includes a row-level delete returns an error instead of dropping those +// rows. +func (s *IncrementalChangelogScan) ToArrowRecords(ctx context.Context) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) { + tasks, err := s.PlanFiles(ctx) + if err != nil { + return nil, nil, err + } + + return s.Read(ctx, tasks) +} + +// Read materializes changelog rows for tasks. Each batch has the scan +// projection followed by _change_type, _change_ordinal, and +// _commit_snapshot_id. AddedRowsScanTask and DeletedDataFileScanTask are read +// with Scan.ReadTasks, so same-snapshot deletes and deletes that already +// applied to a removed file are applied before the changelog columns are +// added. DeletedRowsScanTask returns an error; those rows are produced by a +// later reader. The returned iterator is single-use. The caller releases every +// batch. +func (s *IncrementalChangelogScan) Read(ctx context.Context, tasks []ChangelogScanTask) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) { + if s == nil || s.scan == nil { + return nil, nil, fmt.Errorf("%w: incremental changelog scan is not initialized", ErrInvalidOperation) + } + if err := ctx.Err(); err != nil { + return nil, nil, err + } + for _, task := range tasks { + if err := readableChangelogTask(task); err != nil { + return nil, nil, err + } + } + + schema, useLargeTypes, err := changelogArrowSchema(s.scan) + if err != nil { + return nil, nil, err + } + if s.scan.limit == 0 || len(tasks) == 0 { + return schema, func(yield func(arrow.RecordBatch, error) bool) {}, nil + } + + readScan := s.scan.clone() + limit := s.scan.limit + + return schema, func(yield func(arrow.RecordBatch, error) bool) { + var emitted int64 + for _, task := range tasks { + if err := ctx.Err(); err != nil { + yield(nil, err) + + return + } + if limit > 0 { + readScan.limit = limit - emitted + } + + _, records, err := readScan.ReadTasks(ctx, []FileScanTask{task.ScanTask()}) + if err != nil { + yield(nil, err) + + return + } + + for rec, err := range records { + if err != nil { + yield(nil, err) + + return + } + + stamped, rows, err := stampChangelogBatch(ctx, schema, useLargeTypes, rec, task, limit, emitted) + rec.Release() + if err != nil { + yield(nil, err) + + return + } + if stamped == nil { + continue + } + if !yield(stamped, nil) { + return + } + emitted += rows + if limit > 0 && emitted >= limit { + return + } + } + } + }, nil +} + +func readableChangelogTask(task ChangelogScanTask) error { + switch task.(type) { + case AddedRowsScanTask, DeletedDataFileScanTask: + return nil + case DeletedRowsScanTask: + return fmt.Errorf("%w: reading deleted rows from added delete files is not supported", iceberg.ErrNotImplemented) + default: + return fmt.Errorf("%w: unsupported changelog scan task %T", iceberg.ErrInvalidArgument, task) + } +} + +func changelogArrowSchema(scan *Scan) (*arrow.Schema, bool, error) { + if atomic.LoadUint32(&scan.closed) != 0 { + return nil, false, fmt.Errorf("%w: scan is closed", ErrInvalidOperation) + } + if scan.selectorErr != nil { + return nil, false, scan.selectorErr + } + + projected, err := scan.Projection() + if err != nil { + return nil, false, err + } + + useLargeTypes, err := strconv.ParseBool(scan.options.Get(ScanOptionArrowUseLargeTypes, "false")) + if err != nil { + useLargeTypes = false + } + + props := scan.metadata.Properties() + if scan.arrowBatchSize > 0 { + props = maps.Clone(props) + if props == nil { + props = iceberg.Properties{} + } + props[ParquetBatchSizeKey] = strconv.Itoa(scan.arrowBatchSize) + } + + dataSchema, err := SchemaToArrowSchemaWithOptions(projected, ArrowSchemaOptions{ + UseLargeTypes: useLargeTypes, + TableProperties: props, + }) + if err != nil { + return nil, false, err + } + + schema := dataSchema + for _, field := range changelogArrowFields(useLargeTypes) { + schema, err = schema.AddField(schema.NumFields(), field) + if err != nil { + return nil, false, err + } + } + + return schema, useLargeTypes, nil +} + +func changelogArrowFields(useLargeTypes bool) []arrow.Field { + changeType := arrow.BinaryTypes.String + if useLargeTypes { + changeType = arrow.BinaryTypes.LargeString + } + + return []arrow.Field{ + {Name: ChangelogChangeTypeColumn, Type: changeType, Nullable: false}, + {Name: ChangelogChangeOrdinalColumn, Type: arrow.PrimitiveTypes.Int64, Nullable: false}, + {Name: ChangelogCommitSnapshotIDColumn, Type: arrow.PrimitiveTypes.Int64, Nullable: false}, + } +} + +// stampChangelogBatch appends the changelog columns and applies any remaining +// row limit. A nil batch means the limit was already exhausted. rec is not +// released here. +func stampChangelogBatch(ctx context.Context, schema *arrow.Schema, useLargeTypes bool, rec arrow.RecordBatch, task ChangelogScanTask, limit, emitted int64) (arrow.RecordBatch, int64, error) { + if limit > 0 && emitted >= limit { + return nil, 0, nil + } + + rows := rec.NumRows() + source := rec + if limit > 0 && emitted+rows > limit { + source = rec.NewSlice(0, limit-emitted) + defer source.Release() + rows = source.NumRows() + } + if rows == 0 { + return nil, 0, nil + } + + stamped, err := appendChangelogColumns(ctx, schema, useLargeTypes, source, task) + if err != nil { + return nil, 0, err + } + + return stamped, rows, nil +} + +func appendChangelogColumns(ctx context.Context, schema *arrow.Schema, useLargeTypes bool, rec arrow.RecordBatch, task ChangelogScanTask) (arrow.RecordBatch, error) { + mem := allocatorFrom(ctx) + n := int(rec.NumRows()) + changeType := constantStringArray(mem, useLargeTypes, string(task.Operation()), n) + defer changeType.Release() + + ordinal := constantInt64Array(mem, int64(task.ChangeOrdinal()), n) + defer ordinal.Release() + snapshotID := constantInt64Array(mem, task.CommitSnapshotID(), n) + defer snapshotID.Release() + + cols := make([]arrow.Array, 0, rec.NumCols()+3) + cols = append(cols, rec.Columns()...) + cols = append(cols, changeType, ordinal, snapshotID) + + return array.NewRecordBatch(schema, cols, rec.NumRows()), nil Review Comment: The output schema is built independently of what `ReadTasks` actually yields: we discard the schema it returns at the `ReadTasks` call and hand this precomputed one to `NewRecordBatch`. If the two ever diverge (large-type options, field metadata, row-lineage columns), this panics or silently mislabels columns. I'd take the data portion from the first yielded record's schema and append the three fields to that, or assert `rec.Schema().Equal(dataSchema)` and return an error instead of panicking. That also gives `appendChangelogColumns` a real error to return: right now it always returns nil, so the error branch in `stampChangelogBatch` is dead code. ########## table/changelog_reader.go: ########## @@ -0,0 +1,293 @@ +// 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 + +import ( + "context" + "fmt" + "iter" + "maps" + "strconv" + "sync/atomic" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/compute" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/apache/iceberg-go" +) + +const ( + // ChangelogChangeTypeColumn is the changelog operation: INSERT, DELETE, + // UPDATE_BEFORE, or UPDATE_AFTER. + ChangelogChangeTypeColumn = "_change_type" + // ChangelogChangeOrdinalColumn is the zero-based ordinal of the snapshot + // that produced the change, shared by every row from that snapshot. + ChangelogChangeOrdinalColumn = "_change_ordinal" + // ChangelogCommitSnapshotIDColumn is the snapshot ID that committed the change. + ChangelogCommitSnapshotIDColumn = "_commit_snapshot_id" +) + +// ToArrowRecords plans the changelog and reads it. Added and removed data +// files are read with the deletes already stored on their scan task. A plan +// that includes a row-level delete returns an error instead of dropping those +// rows. +func (s *IncrementalChangelogScan) ToArrowRecords(ctx context.Context) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) { + tasks, err := s.PlanFiles(ctx) + if err != nil { + return nil, nil, err + } + + return s.Read(ctx, tasks) +} + +// Read materializes changelog rows for tasks. Each batch has the scan +// projection followed by _change_type, _change_ordinal, and +// _commit_snapshot_id. AddedRowsScanTask and DeletedDataFileScanTask are read +// with Scan.ReadTasks, so same-snapshot deletes and deletes that already +// applied to a removed file are applied before the changelog columns are +// added. DeletedRowsScanTask returns an error; those rows are produced by a +// later reader. The returned iterator is single-use. The caller releases every +// batch. +func (s *IncrementalChangelogScan) Read(ctx context.Context, tasks []ChangelogScanTask) (*arrow.Schema, iter.Seq2[arrow.RecordBatch, error], error) { + if s == nil || s.scan == nil { + return nil, nil, fmt.Errorf("%w: incremental changelog scan is not initialized", ErrInvalidOperation) + } + if err := ctx.Err(); err != nil { + return nil, nil, err + } + for _, task := range tasks { + if err := readableChangelogTask(task); err != nil { + return nil, nil, err + } + } + + schema, useLargeTypes, err := changelogArrowSchema(s.scan) + if err != nil { + return nil, nil, err + } + if s.scan.limit == 0 || len(tasks) == 0 { + return schema, func(yield func(arrow.RecordBatch, error) bool) {}, nil + } + + readScan := s.scan.clone() + limit := s.scan.limit + + return schema, func(yield func(arrow.RecordBatch, error) bool) { + var emitted int64 + for _, task := range tasks { + if err := ctx.Err(); err != nil { + yield(nil, err) + + return + } + if limit > 0 { + readScan.limit = limit - emitted + } + + _, records, err := readScan.ReadTasks(ctx, []FileScanTask{task.ScanTask()}) + if err != nil { + yield(nil, err) + + return + } + + for rec, err := range records { + if err != nil { + yield(nil, err) + + return + } + + stamped, rows, err := stampChangelogBatch(ctx, schema, useLargeTypes, rec, task, limit, emitted) + rec.Release() + if err != nil { + yield(nil, err) + + return + } + if stamped == nil { + continue + } + if !yield(stamped, nil) { + return + } + emitted += rows + if limit > 0 && emitted >= limit { + return + } + } + } + }, nil +} + +func readableChangelogTask(task ChangelogScanTask) error { + switch task.(type) { + case AddedRowsScanTask, DeletedDataFileScanTask: + return nil + case DeletedRowsScanTask: + return fmt.Errorf("%w: reading deleted rows from added delete files is not supported", iceberg.ErrNotImplemented) + default: + return fmt.Errorf("%w: unsupported changelog scan task %T", iceberg.ErrInvalidArgument, task) + } +} + +func changelogArrowSchema(scan *Scan) (*arrow.Schema, bool, error) { + if atomic.LoadUint32(&scan.closed) != 0 { + return nil, false, fmt.Errorf("%w: scan is closed", ErrInvalidOperation) + } + if scan.selectorErr != nil { + return nil, false, scan.selectorErr + } + + projected, err := scan.Projection() + if err != nil { + return nil, false, err + } + + useLargeTypes, err := strconv.ParseBool(scan.options.Get(ScanOptionArrowUseLargeTypes, "false")) + if err != nil { + useLargeTypes = false + } + + props := scan.metadata.Properties() + if scan.arrowBatchSize > 0 { + props = maps.Clone(props) + if props == nil { + props = iceberg.Properties{} + } + props[ParquetBatchSizeKey] = strconv.Itoa(scan.arrowBatchSize) + } + + dataSchema, err := SchemaToArrowSchemaWithOptions(projected, ArrowSchemaOptions{ + UseLargeTypes: useLargeTypes, + TableProperties: props, + }) + if err != nil { + return nil, false, err + } + + schema := dataSchema + for _, field := range changelogArrowFields(useLargeTypes) { + schema, err = schema.AddField(schema.NumFields(), field) + if err != nil { + return nil, false, err + } + } + + return schema, useLargeTypes, nil +} + +func changelogArrowFields(useLargeTypes bool) []arrow.Field { + changeType := arrow.BinaryTypes.String + if useLargeTypes { + changeType = arrow.BinaryTypes.LargeString + } + + return []arrow.Field{ + {Name: ChangelogChangeTypeColumn, Type: changeType, Nullable: false}, + {Name: ChangelogChangeOrdinalColumn, Type: arrow.PrimitiveTypes.Int64, Nullable: false}, Review Comment: I'd emit `_change_ordinal` as Int32 (add a `constantInt32Array`) to match Java/Spark, which declare `MetadataColumns.CHANGE_ORDINAL` as a 32-bit int, or document the Int64 as a deliberate divergence. As-is, anyone unioning this with Spark's `changes` output or a later rust/PyIceberg reader hits a schema mismatch. Worth settling before these columns and the exported constants become public contract, since changing the Arrow width after release breaks callers. (`_commit_snapshot_id` as long and `_change_type` as string already match, so it's just the ordinal.) ########## table/incremental_changelog_deletes.go: ########## @@ -0,0 +1,658 @@ +// 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 + +import ( + "cmp" + "context" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/io" +) + +// changelogTaskPlan is the snapshot range and the data-file changes already +// read for one changelog plan. +type changelogTaskPlan struct { + scan *Scan + fs io.IO + snapshots []Snapshot + manifestsBySnapshot map[int64][]iceberg.ManifestFile + deleteManifests map[int64][]iceberg.ManifestFile + dataEntries []iceberg.ManifestEntry + snapshotOrdinals map[int64]int + schema *iceberg.Schema + partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression] + residual iceberg.BooleanExpression + metrics *scanMetricsAccumulator +} + +// changelogDeleteIndex matches delete files to a data file with the same +// sequence rules as a normal scan. +type changelogDeleteIndex struct { + pos *positionalDeleteIndex + dv map[string]iceberg.ManifestEntry + eq *equalityDeleteIndex +} + +// changelogDeleteState tracks delete files added and removed inside the +// changelog range, plus delete files that were already live before it. +type changelogDeleteState struct { + active bool + snapshots []Snapshot + preRange []iceberg.ManifestEntry + added map[int64][]iceberg.ManifestEntry + removed map[int64]map[string]struct{} + specs partitionSpecLookup + schema *iceberg.Schema + + addedCache map[int64]*changelogDeleteIndex + beforeCache map[int64]*changelogDeleteIndex +} + +func (s *IncrementalChangelogScan) planChangelogTasks(ctx context.Context, plan changelogTaskPlan) ([]plannedChangelogTask, error) { + state, err := s.loadChangelogDeleteState(ctx, plan) + if err != nil { + return nil, err + } + + changedPaths := make(map[int64]map[string]struct{}) + tasks := make([]plannedChangelogTask, 0, len(plan.dataEntries)) + for _, entry := range plan.dataEntries { + ordinal, ok := plan.snapshotOrdinals[entry.SnapshotID()] + if !ok { + continue + } + paths := changedPaths[entry.SnapshotID()] + if paths == nil { + paths = make(map[string]struct{}) + changedPaths[entry.SnapshotID()] = paths + } + paths[entry.DataFile().FilePath()] = struct{}{} + + deletes, err := state.deletesFor(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + task, err := newChangelogScanTask(entry, ordinal, plan.residual, deletes) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + tasks = append(tasks, plannedChangelogTask{task: task}) + } + + rowTasks, err := s.planDeletedRowTasks(ctx, plan, state, changedPaths) + if err != nil { + return nil, err + } + tasks = append(tasks, rowTasks...) + sortChangelogTasks(tasks) + + return tasks, nil +} + +func (s *IncrementalChangelogScan) loadChangelogDeleteState(ctx context.Context, plan changelogTaskPlan) (*changelogDeleteState, error) { + state := &changelogDeleteState{ + snapshots: plan.snapshots, + added: make(map[int64][]iceberg.ManifestEntry), + removed: make(map[int64]map[string]struct{}), + specs: plan.scan.metadata, + schema: plan.schema, + } + for _, snapshot := range plan.snapshots { + if err := ctx.Err(); err != nil { + return nil, err + } + manifests := plan.deleteManifests[snapshot.SnapshotID] + if len(manifests) == 0 { + continue + } + filtered, err := plan.scan.filterManifestsWithSchemaOptions( + manifests, plan.schema, plan.metrics, plan.partitionFilters, true) + if err != nil { + return nil, err + } + entries, err := readManifestEntries(ctx, plan.fs, filtered, false, false) + if err != nil { + return nil, err + } + for _, entry := range entries { + if entry.SnapshotID() != snapshot.SnapshotID { + continue + } + if entry.Status() == iceberg.EntryStatusDELETED { + if state.removed[snapshot.SnapshotID] == nil { + state.removed[snapshot.SnapshotID] = make(map[string]struct{}) + } + state.removed[snapshot.SnapshotID][entry.DataFile().FilePath()] = struct{}{} + + continue + } + state.added[snapshot.SnapshotID] = append(state.added[snapshot.SnapshotID], entry) + } + } + + deletedData := false + for _, entry := range plan.dataEntries { + if entry.Status() == iceberg.EntryStatusDELETED { + deletedData = true + + break + } + } + for _, entries := range state.added { + if len(entries) > 0 { + state.active = true + + break + } + } + if len(state.removed) > 0 || deletedData { + state.active = true + } + if !state.active { + return state, nil + } + + preRange, err := s.liveDeletesBeforeRange(ctx, plan.fs, plan.snapshots) + if err != nil { + return nil, err + } + state.preRange = preRange + + return state, nil +} + +// liveDeletesBeforeRange returns delete files that were already live on the +// snapshot immediately before the changelog range. +func (s *IncrementalChangelogScan) liveDeletesBeforeRange(ctx context.Context, fs io.IO, snapshots []Snapshot) ([]iceberg.ManifestEntry, error) { + parent := s.snapshotBeforeRange(snapshots) + if parent == nil { + return nil, nil + } + manifests, err := parent.Manifests(fs) + if err != nil { + return nil, err + } + deleteManifests := make([]iceberg.ManifestFile, 0) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentDeletes { + deleteManifests = append(deleteManifests, manifest) + } + } + entries, err := readManifestEntries(ctx, fs, deleteManifests, false, false) + if err != nil { + return nil, err + } + + return liveManifestEntries(entries), nil +} + +func (s *IncrementalChangelogScan) snapshotBeforeRange(snapshots []Snapshot) *Snapshot { + if s.fromSnapshotID != nil && !s.fromInclusive { + return s.scan.metadata.SnapshotByID(*s.fromSnapshotID) + } + if len(snapshots) == 0 || snapshots[0].ParentSnapshotID == nil { + return nil + } + + return s.scan.metadata.SnapshotByID(*snapshots[0].ParentSnapshotID) +} + +func (s *IncrementalChangelogScan) planDeletedRowTasks( + ctx context.Context, + plan changelogTaskPlan, + state *changelogDeleteState, + changedPaths map[int64]map[string]struct{}, +) ([]plannedChangelogTask, error) { + if state == nil || !state.active { + return nil, nil + } + matcher, err := plan.scan.dataFileMatcher(plan.schema, plan.partitionFilters) + if err != nil { + return nil, err + } + + var tasks []plannedChangelogTask + for _, snapshot := range plan.snapshots { + if err := ctx.Err(); err != nil { + return nil, err + } + if len(state.added[snapshot.SnapshotID]) == 0 { + continue + } + addedIndex, err := state.addedIndex(snapshot.SnapshotID) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + manifests := plan.manifestsBySnapshot[snapshot.SnapshotID] + dataManifests := make([]iceberg.ManifestFile, 0, len(manifests)) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + dataManifests = append(dataManifests, manifest) + } + } + // Prune with the scan partition filters before opening manifests. + // A local accumulator keeps carried-forward data manifests out of the + // changelog scan report, which already counted them. + var pruned scanMetricsAccumulator + dataManifests, err = plan.scan.filterManifestsWithSchemaOptions( + dataManifests, plan.schema, &pruned, plan.partitionFilters, true) + if err != nil { + return nil, err + } + live, err := liveDataEntries(ctx, plan.fs, dataManifests) + if err != nil { + return nil, err + } + ordinal := plan.snapshotOrdinals[snapshot.SnapshotID] + seen := make(map[string]struct{}) + for _, entry := range live { + path := entry.DataFile().FilePath() + if _, changed := changedPaths[snapshot.SnapshotID][path]; changed { + continue + } + if _, ok := seen[path]; ok { + continue + } + seen[path] = struct{}{} + + keep, err := matcher(entry.DataFile()) + if err != nil { + return nil, err + } + if !keep { + continue + } + added, err := addedIndex.forDataFile(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + if len(added) == 0 { + continue + } + var existing []iceberg.DataFile + before, err := state.beforeIndex(snapshot.SnapshotID) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + if before != nil { + existing, err = before.forDataFile(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + } + + task, err := NewDeletedRowsScanTask(entry.DataFile(), added, existing, ordinal, snapshot.SnapshotID) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + configureChangelogFileTask(&task.FileScanTask, entry, plan.residual) + tasks = append(tasks, plannedChangelogTask{task: task}) + } + } + + return tasks, nil +} + +func (st *changelogDeleteState) deletesFor(entry iceberg.ManifestEntry) ([]iceberg.DataFile, error) { + if st == nil || !st.active { + return nil, nil + } + + var ( + idx *changelogDeleteIndex + err error + ) + switch entry.Status() { + case iceberg.EntryStatusADDED: + idx, err = st.addedIndex(entry.SnapshotID()) + case iceberg.EntryStatusDELETED: + idx, err = st.beforeIndex(entry.SnapshotID()) + default: + return nil, fmt.Errorf("%w: unknown manifest entry status %d", ErrInvalidMetadata, entry.Status()) + } + if err != nil || idx == nil { + return nil, err + } + + return idx.forDataFile(entry) +} + +func (st *changelogDeleteState) addedIndex(snapshotID int64) (*changelogDeleteIndex, error) { + if st.addedCache == nil { + st.addedCache = make(map[int64]*changelogDeleteIndex) + } + if idx, ok := st.addedCache[snapshotID]; ok { + return idx, nil + } + idx, err := buildChangelogDeleteIndex(st.added[snapshotID], st.specs, st.schema) + if err != nil { + return nil, err + } + st.addedCache[snapshotID] = idx + + return idx, nil +} + +func (st *changelogDeleteState) beforeIndex(snapshotID int64) (*changelogDeleteIndex, error) { + if st.beforeCache == nil { + st.beforeCache = make(map[int64]*changelogDeleteIndex) + } + if idx, ok := st.beforeCache[snapshotID]; ok { + return idx, nil + } + idx, err := buildChangelogDeleteIndex(st.entriesBefore(snapshotID), st.specs, st.schema) + if err != nil { + return nil, err + } + st.beforeCache[snapshotID] = idx + + return idx, nil +} + +// entriesBefore returns delete-file entries that applied strictly before +// snapshotID. Removals drop a path from the running set. A path removed and +// added again in one snapshot stays live as the added entry. +func (st *changelogDeleteState) entriesBefore(snapshotID int64) []iceberg.ManifestEntry { + removed := make(map[string]struct{}) + var accumulated []iceberg.ManifestEntry + for _, snapshot := range st.snapshots { + if snapshot.SnapshotID == snapshotID { + break + } + currentRemoved := st.removed[snapshot.SnapshotID] + if len(currentRemoved) > 0 { + for path := range currentRemoved { + removed[path] = struct{}{} + } + accumulated = dropManifestEntries(accumulated, currentRemoved) + } + accumulated = append(accumulated, st.added[snapshot.SnapshotID]...) + } + + out := make([]iceberg.ManifestEntry, 0, len(st.preRange)+len(accumulated)) + for _, entry := range st.preRange { + if _, gone := removed[entry.DataFile().FilePath()]; gone { + continue + } + out = append(out, entry) + } + out = append(out, accumulated...) + + return keepNewestDeletionVector(out) +} + +func (idx *changelogDeleteIndex) forDataFile(entry iceberg.ManifestEntry) ([]iceberg.DataFile, error) { + if idx == nil { + return nil, nil + } + task, err := fileScanTaskForDataEntry(entry, idx.pos, idx.dv, idx.eq) + if err != nil { + return nil, err + } + files := allDeleteFiles(task) + if len(files) == 0 { + return nil, nil + } + + return files, nil +} + +func buildChangelogDeleteIndex(entries []iceberg.ManifestEntry, specs partitionSpecLookup, schema *iceberg.Schema) (*changelogDeleteIndex, error) { + if len(entries) == 0 { + return nil, nil + } + entries = keepNewestDeletionVector(entries) + classified, err := classifyManifestEntries(entries) + if err != nil { + return nil, err + } + if len(classified.dataEntries) > 0 { + return nil, fmt.Errorf("%w: expected delete file, got %s", + ErrInvalidMetadata, classified.dataEntries[0].DataFile().FilePath()) + } + + pos, err := buildPositionalDeleteIndex(classified.positionalDeleteEntries) + if err != nil { + return nil, err + } + dv, err := buildDVIndex(classified.dvEntries) + if err != nil { + return nil, err + } + eq, err := buildEqualityDeleteIndex(classified.equalityDeleteEntries, specs, schema) + if err != nil { + return nil, err + } + + return &changelogDeleteIndex{pos: pos, dv: dv, eq: eq}, nil +} + +func (scan *Scan) dataFileMatcher( + schema *iceberg.Schema, + partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression], +) (func(iceberg.DataFile) (bool, error), error) { + metricsEval, err := newInclusiveMetricsEvaluator( + schema, + scan.rowFilter, + scan.caseSensitive, + scan.options["include_empty_files"] == "true", + ) + if err != nil { + return nil, err + } + partitionEvaluators := newKeyDefaultMapWrapErr(func(specID int) (func(iceberg.DataFile) (bool, error), error) { + return buildPartitionEvaluator(specID, scan.metadata, schema, partitionFilters, scan.caseSensitive) + }) + + return func(file iceberg.DataFile) (bool, error) { + partEval, err := partitionEvaluators.Get(int(file.SpecID())) + if err != nil { + return false, fmt.Errorf("failed to build partition evaluator for spec %d: %w", file.SpecID(), err) + } + use, err := partEval(file) + if err != nil || !use { + return use, err + } + + return metricsEval(file) + }, nil +} + +func liveDataEntries(ctx context.Context, fs io.IO, manifests []iceberg.ManifestFile) ([]iceberg.ManifestEntry, error) { + dataManifests := make([]iceberg.ManifestFile, 0, len(manifests)) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + dataManifests = append(dataManifests, manifest) + } + } + entries, err := readManifestEntries(ctx, fs, dataManifests, false, false) + if err != nil { + return nil, err + } + dataEntries := make([]iceberg.ManifestEntry, 0, len(entries)) + for _, entry := range entries { + if entry.DataFile().ContentType() == iceberg.EntryContentData { + dataEntries = append(dataEntries, entry) + } + } + + return liveManifestEntries(dataEntries), nil +} + +// liveManifestEntries keeps the highest-sequence entry for each path and +// drops paths whose winning entry is a deletion. Equal sequence numbers +// keep the later entry. +func liveManifestEntries(entries []iceberg.ManifestEntry) []iceberg.ManifestEntry { + latest := make(map[string]iceberg.ManifestEntry, len(entries)) + order := make([]string, 0, len(entries)) + for _, entry := range entries { + path := entry.DataFile().FilePath() Review Comment: `liveManifestEntries` keys delete entries by the delete file's own `FilePath()`, and `removed` / `dropManifestEntries` / `entriesBefore` do the same, but several deletion vectors for different data files can live in one Puffin file and share that path. So two live DVs in the same Puffin collapse to one here, and a removed entry for one DV drops every accumulated entry sharing that Puffin path, including live DVs for other data files. The missing deletes then come back as spurious DELETE rows. `keepNewestDeletionVector` already keys DVs by referenced data file, I'd key the live/removed sets the same way (referenced data file, plus content offset for DVs) rather than by Puffin path. A test with two DVs in one Puffin where only one is removed would pin it. ########## table/incremental_changelog_deletes.go: ########## @@ -0,0 +1,658 @@ +// 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 + +import ( + "cmp" + "context" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/io" +) + +// changelogTaskPlan is the snapshot range and the data-file changes already +// read for one changelog plan. +type changelogTaskPlan struct { + scan *Scan + fs io.IO + snapshots []Snapshot + manifestsBySnapshot map[int64][]iceberg.ManifestFile + deleteManifests map[int64][]iceberg.ManifestFile + dataEntries []iceberg.ManifestEntry + snapshotOrdinals map[int64]int + schema *iceberg.Schema + partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression] + residual iceberg.BooleanExpression + metrics *scanMetricsAccumulator +} + +// changelogDeleteIndex matches delete files to a data file with the same +// sequence rules as a normal scan. +type changelogDeleteIndex struct { + pos *positionalDeleteIndex + dv map[string]iceberg.ManifestEntry + eq *equalityDeleteIndex +} + +// changelogDeleteState tracks delete files added and removed inside the +// changelog range, plus delete files that were already live before it. +type changelogDeleteState struct { + active bool + snapshots []Snapshot + preRange []iceberg.ManifestEntry + added map[int64][]iceberg.ManifestEntry + removed map[int64]map[string]struct{} + specs partitionSpecLookup + schema *iceberg.Schema + + addedCache map[int64]*changelogDeleteIndex + beforeCache map[int64]*changelogDeleteIndex +} + +func (s *IncrementalChangelogScan) planChangelogTasks(ctx context.Context, plan changelogTaskPlan) ([]plannedChangelogTask, error) { + state, err := s.loadChangelogDeleteState(ctx, plan) + if err != nil { + return nil, err + } + + changedPaths := make(map[int64]map[string]struct{}) + tasks := make([]plannedChangelogTask, 0, len(plan.dataEntries)) + for _, entry := range plan.dataEntries { + ordinal, ok := plan.snapshotOrdinals[entry.SnapshotID()] + if !ok { + continue + } + paths := changedPaths[entry.SnapshotID()] + if paths == nil { + paths = make(map[string]struct{}) + changedPaths[entry.SnapshotID()] = paths + } + paths[entry.DataFile().FilePath()] = struct{}{} + + deletes, err := state.deletesFor(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + task, err := newChangelogScanTask(entry, ordinal, plan.residual, deletes) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + tasks = append(tasks, plannedChangelogTask{task: task}) + } + + rowTasks, err := s.planDeletedRowTasks(ctx, plan, state, changedPaths) + if err != nil { + return nil, err + } + tasks = append(tasks, rowTasks...) + sortChangelogTasks(tasks) + + return tasks, nil +} + +func (s *IncrementalChangelogScan) loadChangelogDeleteState(ctx context.Context, plan changelogTaskPlan) (*changelogDeleteState, error) { + state := &changelogDeleteState{ + snapshots: plan.snapshots, + added: make(map[int64][]iceberg.ManifestEntry), + removed: make(map[int64]map[string]struct{}), + specs: plan.scan.metadata, + schema: plan.schema, + } + for _, snapshot := range plan.snapshots { + if err := ctx.Err(); err != nil { + return nil, err + } + manifests := plan.deleteManifests[snapshot.SnapshotID] + if len(manifests) == 0 { + continue + } + filtered, err := plan.scan.filterManifestsWithSchemaOptions( + manifests, plan.schema, plan.metrics, plan.partitionFilters, true) + if err != nil { + return nil, err + } + entries, err := readManifestEntries(ctx, plan.fs, filtered, false, false) + if err != nil { + return nil, err + } + for _, entry := range entries { + if entry.SnapshotID() != snapshot.SnapshotID { + continue + } + if entry.Status() == iceberg.EntryStatusDELETED { + if state.removed[snapshot.SnapshotID] == nil { + state.removed[snapshot.SnapshotID] = make(map[string]struct{}) + } + state.removed[snapshot.SnapshotID][entry.DataFile().FilePath()] = struct{}{} + + continue + } + state.added[snapshot.SnapshotID] = append(state.added[snapshot.SnapshotID], entry) + } + } + + deletedData := false + for _, entry := range plan.dataEntries { + if entry.Status() == iceberg.EntryStatusDELETED { + deletedData = true + + break + } + } + for _, entries := range state.added { + if len(entries) > 0 { + state.active = true + + break + } + } + if len(state.removed) > 0 || deletedData { + state.active = true + } + if !state.active { + return state, nil + } + + preRange, err := s.liveDeletesBeforeRange(ctx, plan.fs, plan.snapshots) + if err != nil { + return nil, err + } + state.preRange = preRange + + return state, nil +} + +// liveDeletesBeforeRange returns delete files that were already live on the +// snapshot immediately before the changelog range. +func (s *IncrementalChangelogScan) liveDeletesBeforeRange(ctx context.Context, fs io.IO, snapshots []Snapshot) ([]iceberg.ManifestEntry, error) { + parent := s.snapshotBeforeRange(snapshots) + if parent == nil { + return nil, nil + } + manifests, err := parent.Manifests(fs) + if err != nil { + return nil, err + } + deleteManifests := make([]iceberg.ManifestFile, 0) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentDeletes { + deleteManifests = append(deleteManifests, manifest) + } + } + entries, err := readManifestEntries(ctx, fs, deleteManifests, false, false) + if err != nil { + return nil, err + } + + return liveManifestEntries(entries), nil +} + +func (s *IncrementalChangelogScan) snapshotBeforeRange(snapshots []Snapshot) *Snapshot { + if s.fromSnapshotID != nil && !s.fromInclusive { + return s.scan.metadata.SnapshotByID(*s.fromSnapshotID) + } + if len(snapshots) == 0 || snapshots[0].ParentSnapshotID == nil { + return nil + } + + return s.scan.metadata.SnapshotByID(*snapshots[0].ParentSnapshotID) +} + +func (s *IncrementalChangelogScan) planDeletedRowTasks( + ctx context.Context, + plan changelogTaskPlan, + state *changelogDeleteState, + changedPaths map[int64]map[string]struct{}, +) ([]plannedChangelogTask, error) { + if state == nil || !state.active { + return nil, nil + } + matcher, err := plan.scan.dataFileMatcher(plan.schema, plan.partitionFilters) + if err != nil { + return nil, err + } + + var tasks []plannedChangelogTask + for _, snapshot := range plan.snapshots { + if err := ctx.Err(); err != nil { + return nil, err + } + if len(state.added[snapshot.SnapshotID]) == 0 { + continue + } + addedIndex, err := state.addedIndex(snapshot.SnapshotID) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + manifests := plan.manifestsBySnapshot[snapshot.SnapshotID] + dataManifests := make([]iceberg.ManifestFile, 0, len(manifests)) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + dataManifests = append(dataManifests, manifest) + } + } + // Prune with the scan partition filters before opening manifests. + // A local accumulator keeps carried-forward data manifests out of the + // changelog scan report, which already counted them. + var pruned scanMetricsAccumulator + dataManifests, err = plan.scan.filterManifestsWithSchemaOptions( + dataManifests, plan.schema, &pruned, plan.partitionFilters, true) + if err != nil { + return nil, err + } + live, err := liveDataEntries(ctx, plan.fs, dataManifests) + if err != nil { + return nil, err + } + ordinal := plan.snapshotOrdinals[snapshot.SnapshotID] + seen := make(map[string]struct{}) + for _, entry := range live { + path := entry.DataFile().FilePath() + if _, changed := changedPaths[snapshot.SnapshotID][path]; changed { + continue + } + if _, ok := seen[path]; ok { + continue + } + seen[path] = struct{}{} + + keep, err := matcher(entry.DataFile()) + if err != nil { + return nil, err + } + if !keep { + continue + } + added, err := addedIndex.forDataFile(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + if len(added) == 0 { + continue + } + var existing []iceberg.DataFile + before, err := state.beforeIndex(snapshot.SnapshotID) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + if before != nil { + existing, err = before.forDataFile(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + } + + task, err := NewDeletedRowsScanTask(entry.DataFile(), added, existing, ordinal, snapshot.SnapshotID) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + configureChangelogFileTask(&task.FileScanTask, entry, plan.residual) + tasks = append(tasks, plannedChangelogTask{task: task}) + } + } + + return tasks, nil +} + +func (st *changelogDeleteState) deletesFor(entry iceberg.ManifestEntry) ([]iceberg.DataFile, error) { + if st == nil || !st.active { + return nil, nil + } + + var ( + idx *changelogDeleteIndex + err error + ) + switch entry.Status() { + case iceberg.EntryStatusADDED: + idx, err = st.addedIndex(entry.SnapshotID()) + case iceberg.EntryStatusDELETED: + idx, err = st.beforeIndex(entry.SnapshotID()) + default: + return nil, fmt.Errorf("%w: unknown manifest entry status %d", ErrInvalidMetadata, entry.Status()) + } + if err != nil || idx == nil { + return nil, err + } + + return idx.forDataFile(entry) +} + +func (st *changelogDeleteState) addedIndex(snapshotID int64) (*changelogDeleteIndex, error) { + if st.addedCache == nil { + st.addedCache = make(map[int64]*changelogDeleteIndex) + } + if idx, ok := st.addedCache[snapshotID]; ok { + return idx, nil + } + idx, err := buildChangelogDeleteIndex(st.added[snapshotID], st.specs, st.schema) + if err != nil { + return nil, err + } + st.addedCache[snapshotID] = idx + + return idx, nil +} + +func (st *changelogDeleteState) beforeIndex(snapshotID int64) (*changelogDeleteIndex, error) { + if st.beforeCache == nil { + st.beforeCache = make(map[int64]*changelogDeleteIndex) + } + if idx, ok := st.beforeCache[snapshotID]; ok { + return idx, nil + } + idx, err := buildChangelogDeleteIndex(st.entriesBefore(snapshotID), st.specs, st.schema) + if err != nil { + return nil, err + } + st.beforeCache[snapshotID] = idx + + return idx, nil +} + +// entriesBefore returns delete-file entries that applied strictly before +// snapshotID. Removals drop a path from the running set. A path removed and +// added again in one snapshot stays live as the added entry. +func (st *changelogDeleteState) entriesBefore(snapshotID int64) []iceberg.ManifestEntry { + removed := make(map[string]struct{}) + var accumulated []iceberg.ManifestEntry + for _, snapshot := range st.snapshots { Review Comment: `st.snapshots` is the post-filter list that drops `OpReplace` snapshots, so `entriesBefore` (and the `preRange` it builds on) never replays delete-file changes made by a replace. Compaction (rewrite-position-deletes, DV compaction) commits as a replace and adds/removes delete files, so in any table that runs maintenance (the common case) the "before" state for a later snapshot can be wrong, and a removed file's DELETE task re-emits already-deleted rows or drops the wrong ones. I'd replay delete-manifest changes across the skipped replace snapshots for the before-state, or compute "before" directly from the removing snapshot's parent manifest list. Either way I'd add a test with an in-range replace that rewrites delete files, nothing covers that today. ########## table/changelog_reader_external_test.go: ########## @@ -0,0 +1,196 @@ +// 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" + "iter" + "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" + "github.com/apache/iceberg-go/table" + "github.com/stretchr/testify/require" +) + +func TestIncrementalChangelogReadEmptyTable(t *testing.T) { + schema, records, err := newV2RowLineageTestTable(t).NewIncrementalChangelogScan().ToArrowRecords(t.Context()) + require.NoError(t, err) + require.Equal(t, []string{ + "id", "data", + table.ChangelogChangeTypeColumn, + table.ChangelogChangeOrdinalColumn, + table.ChangelogCommitSnapshotIDColumn, + }, schemaFieldNames(schema)) + + require.Empty(t, collectChangelogRows(t, records)) +} + +func TestIncrementalChangelogReadFileChanges(t *testing.T) { Review Comment: The reader tests never actually read a task that carries deletes: `changelogOverwriteTable` only does Append + Overwrite, so every task is a whole-file add/remove with no delete file in play. The PR's headline behavior (same-snapshot deletes applied to an added file, earlier deletes subtracted from a removed file, no double-emitting already-deleted rows) rests entirely on `ReadTasks` honoring `task.ScanTask()`'s deletes, and nothing here exercises that. I'd add a reader test that writes a real position-delete (or DV) and asserts the deleted rows disappear from the INSERT on the added file and from the DELETE on the removed file. That's the one gap I'd most want closed before merge. ########## table/incremental_changelog_deletes.go: ########## @@ -0,0 +1,658 @@ +// 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 + +import ( + "cmp" + "context" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/io" +) + +// changelogTaskPlan is the snapshot range and the data-file changes already +// read for one changelog plan. +type changelogTaskPlan struct { + scan *Scan + fs io.IO + snapshots []Snapshot + manifestsBySnapshot map[int64][]iceberg.ManifestFile + deleteManifests map[int64][]iceberg.ManifestFile + dataEntries []iceberg.ManifestEntry + snapshotOrdinals map[int64]int + schema *iceberg.Schema + partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression] + residual iceberg.BooleanExpression + metrics *scanMetricsAccumulator +} + +// changelogDeleteIndex matches delete files to a data file with the same +// sequence rules as a normal scan. +type changelogDeleteIndex struct { + pos *positionalDeleteIndex + dv map[string]iceberg.ManifestEntry + eq *equalityDeleteIndex +} + +// changelogDeleteState tracks delete files added and removed inside the +// changelog range, plus delete files that were already live before it. +type changelogDeleteState struct { + active bool + snapshots []Snapshot + preRange []iceberg.ManifestEntry + added map[int64][]iceberg.ManifestEntry + removed map[int64]map[string]struct{} + specs partitionSpecLookup + schema *iceberg.Schema + + addedCache map[int64]*changelogDeleteIndex + beforeCache map[int64]*changelogDeleteIndex +} + +func (s *IncrementalChangelogScan) planChangelogTasks(ctx context.Context, plan changelogTaskPlan) ([]plannedChangelogTask, error) { + state, err := s.loadChangelogDeleteState(ctx, plan) + if err != nil { + return nil, err + } + + changedPaths := make(map[int64]map[string]struct{}) + tasks := make([]plannedChangelogTask, 0, len(plan.dataEntries)) + for _, entry := range plan.dataEntries { + ordinal, ok := plan.snapshotOrdinals[entry.SnapshotID()] + if !ok { + continue + } + paths := changedPaths[entry.SnapshotID()] + if paths == nil { + paths = make(map[string]struct{}) + changedPaths[entry.SnapshotID()] = paths + } + paths[entry.DataFile().FilePath()] = struct{}{} + + deletes, err := state.deletesFor(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + task, err := newChangelogScanTask(entry, ordinal, plan.residual, deletes) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + tasks = append(tasks, plannedChangelogTask{task: task}) + } + + rowTasks, err := s.planDeletedRowTasks(ctx, plan, state, changedPaths) + if err != nil { + return nil, err + } + tasks = append(tasks, rowTasks...) + sortChangelogTasks(tasks) + + return tasks, nil +} + +func (s *IncrementalChangelogScan) loadChangelogDeleteState(ctx context.Context, plan changelogTaskPlan) (*changelogDeleteState, error) { + state := &changelogDeleteState{ + snapshots: plan.snapshots, + added: make(map[int64][]iceberg.ManifestEntry), + removed: make(map[int64]map[string]struct{}), + specs: plan.scan.metadata, + schema: plan.schema, + } + for _, snapshot := range plan.snapshots { + if err := ctx.Err(); err != nil { + return nil, err + } + manifests := plan.deleteManifests[snapshot.SnapshotID] + if len(manifests) == 0 { + continue + } + filtered, err := plan.scan.filterManifestsWithSchemaOptions( + manifests, plan.schema, plan.metrics, plan.partitionFilters, true) + if err != nil { + return nil, err + } + entries, err := readManifestEntries(ctx, plan.fs, filtered, false, false) + if err != nil { + return nil, err + } + for _, entry := range entries { + if entry.SnapshotID() != snapshot.SnapshotID { + continue + } + if entry.Status() == iceberg.EntryStatusDELETED { + if state.removed[snapshot.SnapshotID] == nil { + state.removed[snapshot.SnapshotID] = make(map[string]struct{}) + } + state.removed[snapshot.SnapshotID][entry.DataFile().FilePath()] = struct{}{} + + continue + } + state.added[snapshot.SnapshotID] = append(state.added[snapshot.SnapshotID], entry) + } + } + + deletedData := false + for _, entry := range plan.dataEntries { + if entry.Status() == iceberg.EntryStatusDELETED { + deletedData = true + + break + } + } + for _, entries := range state.added { + if len(entries) > 0 { + state.active = true + + break + } + } + if len(state.removed) > 0 || deletedData { + state.active = true + } + if !state.active { + return state, nil + } + + preRange, err := s.liveDeletesBeforeRange(ctx, plan.fs, plan.snapshots) + if err != nil { + return nil, err + } + state.preRange = preRange + + return state, nil +} + +// liveDeletesBeforeRange returns delete files that were already live on the +// snapshot immediately before the changelog range. +func (s *IncrementalChangelogScan) liveDeletesBeforeRange(ctx context.Context, fs io.IO, snapshots []Snapshot) ([]iceberg.ManifestEntry, error) { + parent := s.snapshotBeforeRange(snapshots) + if parent == nil { + return nil, nil + } + manifests, err := parent.Manifests(fs) + if err != nil { + return nil, err + } + deleteManifests := make([]iceberg.ManifestFile, 0) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentDeletes { + deleteManifests = append(deleteManifests, manifest) + } + } + entries, err := readManifestEntries(ctx, fs, deleteManifests, false, false) + if err != nil { + return nil, err + } + + return liveManifestEntries(entries), nil +} + +func (s *IncrementalChangelogScan) snapshotBeforeRange(snapshots []Snapshot) *Snapshot { + if s.fromSnapshotID != nil && !s.fromInclusive { + return s.scan.metadata.SnapshotByID(*s.fromSnapshotID) + } + if len(snapshots) == 0 || snapshots[0].ParentSnapshotID == nil { + return nil + } + + return s.scan.metadata.SnapshotByID(*snapshots[0].ParentSnapshotID) +} + +func (s *IncrementalChangelogScan) planDeletedRowTasks( + ctx context.Context, + plan changelogTaskPlan, + state *changelogDeleteState, + changedPaths map[int64]map[string]struct{}, +) ([]plannedChangelogTask, error) { + if state == nil || !state.active { + return nil, nil + } + matcher, err := plan.scan.dataFileMatcher(plan.schema, plan.partitionFilters) + if err != nil { + return nil, err + } + + var tasks []plannedChangelogTask + for _, snapshot := range plan.snapshots { + if err := ctx.Err(); err != nil { + return nil, err + } + if len(state.added[snapshot.SnapshotID]) == 0 { + continue + } + addedIndex, err := state.addedIndex(snapshot.SnapshotID) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", snapshot.SnapshotID, err) + } + manifests := plan.manifestsBySnapshot[snapshot.SnapshotID] + dataManifests := make([]iceberg.ManifestFile, 0, len(manifests)) + for _, manifest := range manifests { + if manifest.ManifestContent() == iceberg.ManifestContentData { + dataManifests = append(dataManifests, manifest) + } + } + // Prune with the scan partition filters before opening manifests. + // A local accumulator keeps carried-forward data manifests out of the + // changelog scan report, which already counted them. + var pruned scanMetricsAccumulator + dataManifests, err = plan.scan.filterManifestsWithSchemaOptions( + dataManifests, plan.schema, &pruned, plan.partitionFilters, true) + if err != nil { + return nil, err + } + live, err := liveDataEntries(ctx, plan.fs, dataManifests) Review Comment: `Read` rejects every `DeletedRowsScanTask` with `ErrNotImplemented`, yet `planDeletedRowTasks` still does a full live-data-manifest scan per delete-bearing snapshot to build them, so `ToArrowRecords` pays that whole cost just to return an error, and nothing in this PR can actually consume the row-level plan. With the row reader landing next, I'd hold this planning (or gate it so `ToArrowRecords` doesn't trigger it) until there's a consumer. At minimum I'd fix the description: this reads as "adds the reader," but roughly two-thirds of the diff is a planning rewrite that also drops the old hard error on delete manifests, and that's worth flagging for reviewers. ########## table/incremental_changelog_deletes.go: ########## @@ -0,0 +1,658 @@ +// 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 + +import ( + "cmp" + "context" + "fmt" + "slices" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/io" +) + +// changelogTaskPlan is the snapshot range and the data-file changes already +// read for one changelog plan. +type changelogTaskPlan struct { + scan *Scan + fs io.IO + snapshots []Snapshot + manifestsBySnapshot map[int64][]iceberg.ManifestFile + deleteManifests map[int64][]iceberg.ManifestFile + dataEntries []iceberg.ManifestEntry + snapshotOrdinals map[int64]int + schema *iceberg.Schema + partitionFilters *keyDefaultMapErr[int, iceberg.BooleanExpression] + residual iceberg.BooleanExpression + metrics *scanMetricsAccumulator +} + +// changelogDeleteIndex matches delete files to a data file with the same +// sequence rules as a normal scan. +type changelogDeleteIndex struct { + pos *positionalDeleteIndex + dv map[string]iceberg.ManifestEntry + eq *equalityDeleteIndex +} + +// changelogDeleteState tracks delete files added and removed inside the +// changelog range, plus delete files that were already live before it. +type changelogDeleteState struct { + active bool + snapshots []Snapshot + preRange []iceberg.ManifestEntry + added map[int64][]iceberg.ManifestEntry + removed map[int64]map[string]struct{} + specs partitionSpecLookup + schema *iceberg.Schema + + addedCache map[int64]*changelogDeleteIndex + beforeCache map[int64]*changelogDeleteIndex +} + +func (s *IncrementalChangelogScan) planChangelogTasks(ctx context.Context, plan changelogTaskPlan) ([]plannedChangelogTask, error) { + state, err := s.loadChangelogDeleteState(ctx, plan) + if err != nil { + return nil, err + } + + changedPaths := make(map[int64]map[string]struct{}) + tasks := make([]plannedChangelogTask, 0, len(plan.dataEntries)) + for _, entry := range plan.dataEntries { + ordinal, ok := plan.snapshotOrdinals[entry.SnapshotID()] + if !ok { + continue + } + paths := changedPaths[entry.SnapshotID()] + if paths == nil { + paths = make(map[string]struct{}) + changedPaths[entry.SnapshotID()] = paths + } + paths[entry.DataFile().FilePath()] = struct{}{} + + deletes, err := state.deletesFor(entry) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + task, err := newChangelogScanTask(entry, ordinal, plan.residual, deletes) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + tasks = append(tasks, plannedChangelogTask{task: task}) + } + + rowTasks, err := s.planDeletedRowTasks(ctx, plan, state, changedPaths) + if err != nil { + return nil, err + } + tasks = append(tasks, rowTasks...) + sortChangelogTasks(tasks) + + return tasks, nil +} + +func (s *IncrementalChangelogScan) loadChangelogDeleteState(ctx context.Context, plan changelogTaskPlan) (*changelogDeleteState, error) { + state := &changelogDeleteState{ + snapshots: plan.snapshots, + added: make(map[int64][]iceberg.ManifestEntry), + removed: make(map[int64]map[string]struct{}), + specs: plan.scan.metadata, + schema: plan.schema, + } + for _, snapshot := range plan.snapshots { + if err := ctx.Err(); err != nil { + return nil, err + } + manifests := plan.deleteManifests[snapshot.SnapshotID] + if len(manifests) == 0 { + continue + } + filtered, err := plan.scan.filterManifestsWithSchemaOptions( + manifests, plan.schema, plan.metrics, plan.partitionFilters, true) + if err != nil { + return nil, err + } + entries, err := readManifestEntries(ctx, plan.fs, filtered, false, false) + if err != nil { + return nil, err + } + for _, entry := range entries { + if entry.SnapshotID() != snapshot.SnapshotID { + continue + } + if entry.Status() == iceberg.EntryStatusDELETED { + if state.removed[snapshot.SnapshotID] == nil { + state.removed[snapshot.SnapshotID] = make(map[string]struct{}) + } + state.removed[snapshot.SnapshotID][entry.DataFile().FilePath()] = struct{}{} + + continue + } + state.added[snapshot.SnapshotID] = append(state.added[snapshot.SnapshotID], entry) + } + } + + deletedData := false + for _, entry := range plan.dataEntries { + if entry.Status() == iceberg.EntryStatusDELETED { + deletedData = true + + break + } + } + for _, entries := range state.added { + if len(entries) > 0 { + state.active = true + + break + } + } + if len(state.removed) > 0 || deletedData { + state.active = true + } + if !state.active { + return state, nil + } + + preRange, err := s.liveDeletesBeforeRange(ctx, plan.fs, plan.snapshots) + if err != nil { + return nil, err + } + state.preRange = preRange + + return state, nil +} + +// liveDeletesBeforeRange returns delete files that were already live on the +// snapshot immediately before the changelog range. +func (s *IncrementalChangelogScan) liveDeletesBeforeRange(ctx context.Context, fs io.IO, snapshots []Snapshot) ([]iceberg.ManifestEntry, error) { + parent := s.snapshotBeforeRange(snapshots) + if parent == nil { Review Comment: `snapshotBeforeRange` returns nil both when there's genuinely no parent and when `SnapshotByID` can't find the parent because it's been expired, and here we treat nil as "no deletes were live before the range." The expired case is real: `TestIncrementalChangelogScanAllowsExpiredExclusiveParent` exercises exactly that. When the parent is expired, `preRange` comes back empty, the before-state loses the deletes that were already live, and a removed file's DELETE task re-emits rows that were already deleted before the range. I'd distinguish "no parent" from "parent not found" and, for the expired case where pre-range deletes matter, return an error (or document the limitation) rather than silently dropping them. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
