dgvj-work commented on code in PR #2170: URL: https://github.com/apache/iceberg-go/pull/2170#discussion_r4244573665
########## 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: ToArrowRecords and Read now say this is a raw file-level changelog. A rewritten file emits DELETE and INSERT for every row in that file, including rows that did not change. UPDATE_BEFORE and UPDATE_AFTER are not produced. ########## 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: rejectChangelogColumnCollision runs before the changelog columns are added. TestChangelogArrowSchemaRejectsReservedName projects _change_type and expects ErrInvalidSchema. ########## 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: _change_ordinal is int32, via constantInt32Array, matching MetadataColumns.CHANGE_ORDINAL. The three changelog columns stay non-null, which is stricter than Java's optional ordinal and snapshot id. ########## 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: appendChangelogColumns calls changelogBatchMatches before building the batch. A read batch whose column count or fields do not match the changelog schema returns ErrInvalidSchema instead of a panic. ########## 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: TestIncrementalChangelogReadAppliesFileDeletes commits a data file and a same-snapshot position delete of position 1, then overwrites the file. The first insert omits the deleted row, and the overwrite emits that row as DELETE and the replacement as INSERT. ########## 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: TestIncrementalChangelogReadLimitCutsABatch appends three rows and reads with WithLimit(2). The first batch is cut to the first two inserts. ########## 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: A missing parent is no longer an empty pre-range. The deletes that already applied are rebuilt from the first timeline snapshot's delete manifests: a non-deleted entry from another snapshot, or a deleted entry written by this snapshot. The overwrite and carried-delete tests on #2169 cover that path. ########## 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: This PR still returns ErrNotImplemented for DeletedRowsScanTask. Emitting those rows is #2171. The manifest reads for that planning live on #2169, which now caches decoded entries by manifest path. ########## 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: plan.timeline includes replace snapshots, and their delete adds and removes are replayed. Tasks are still taken from the list that skips OpReplace. TestIncrementalChangelogScanFoldsReplaceDeleteRewrites is on #2169. -- 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]
