dgvj-work commented on code in PR #1897: URL: https://github.com/apache/iceberg-go/pull/1897#discussion_r3876251238
########## table/changelog_scan_task.go: ########## @@ -0,0 +1,152 @@ +// 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 "github.com/apache/iceberg-go" + +// AddedRowsScanTask is a changelog insert produced by adding a data file. +// Matching delete files committed in the same snapshot, or from squashed +// snapshots, are applied while reading so deleted rows are not emitted as +// inserts. +type AddedRowsScanTask struct { Review Comment: Good call — I added `ChangelogScanTask` with `Operation()`, `ChangeOrdinal()`, and `CommitSnapshotID()`, plus a `ChangelogOperation` enum matching Java (`INSERT` / `DELETE` / `UPDATE_BEFORE` / `UPDATE_AFTER`). The three task types implement it so the planning follow-up can return `[]ChangelogScanTask` without a type switch just to tell inserts from deletes. ########## table/changelog_scan_task.go: ########## @@ -0,0 +1,152 @@ +// 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 "github.com/apache/iceberg-go" + +// AddedRowsScanTask is a changelog insert produced by adding a data file. +// Matching delete files committed in the same snapshot, or from squashed +// snapshots, are applied while reading so deleted rows are not emitted as +// inserts. +type AddedRowsScanTask struct { + FileScanTask + changeOrdinal int + commitSnapshotID int64 +} + +// NewAddedRowsScanTask constructs an insert task for dataFile. deletes are +// delete files that apply while reading the added file. Position deletes, +// equality deletes, and deletion vectors are stored on the matching +// FileScanTask fields. +func NewAddedRowsScanTask(dataFile iceberg.DataFile, deletes []iceberg.DataFile, changeOrdinal int, commitSnapshotID int64) AddedRowsScanTask { + return AddedRowsScanTask{ + FileScanTask: fileScanTaskWithDeletes(dataFile, deletes), + changeOrdinal: changeOrdinal, + commitSnapshotID: commitSnapshotID, + } +} + +func (t AddedRowsScanTask) ChangeOrdinal() int { return t.changeOrdinal } +func (t AddedRowsScanTask) CommitSnapshotID() int64 { return t.commitSnapshotID } + +// Deletes returns every delete file applied while reading the added data +// file: position deletes, then equality deletes, then deletion vectors. +func (t AddedRowsScanTask) Deletes() []iceberg.DataFile { + return allDeleteFiles(t.FileScanTask) +} + +// DeletedDataFileScanTask is a changelog delete produced by removing a data +// file. ExistingDeletes are delete files that were already present and must +// be applied so only rows that were live when the file was removed appear as +// deletes. +type DeletedDataFileScanTask struct { + FileScanTask + changeOrdinal int + commitSnapshotID int64 +} + +// NewDeletedDataFileScanTask constructs a delete task for a removed data file. +func NewDeletedDataFileScanTask(dataFile iceberg.DataFile, existingDeletes []iceberg.DataFile, changeOrdinal int, commitSnapshotID int64) DeletedDataFileScanTask { + return DeletedDataFileScanTask{ + FileScanTask: fileScanTaskWithDeletes(dataFile, existingDeletes), + changeOrdinal: changeOrdinal, + commitSnapshotID: commitSnapshotID, + } +} + +func (t DeletedDataFileScanTask) ChangeOrdinal() int { return t.changeOrdinal } +func (t DeletedDataFileScanTask) CommitSnapshotID() int64 { return t.commitSnapshotID } + +// ExistingDeletes returns delete files that applied before the data file was +// removed. +func (t DeletedDataFileScanTask) ExistingDeletes() []iceberg.DataFile { + return allDeleteFiles(t.FileScanTask) +} + +// DeletedRowsScanTask is a changelog delete produced by adding delete files +// against a data file that remains in the table. AddedDeletes remove rows +// that should appear in the changelog. ExistingDeletes already applied and +// those rows must not be emitted again. +type DeletedRowsScanTask struct { + FileScanTask + addedDeletes FileScanTask Review Comment: Agreed, thanks for catching that. `addedDeletes` is now a small internal `classifiedDeletes` struct holding the pos/eq/dv lists, and `AddedDeletes()` just flattens it. That way we never carry a second `FileScanTask` whose range and lineage fields would read as intentionally absent. ########## table/changelog_scan_task.go: ########## @@ -0,0 +1,152 @@ +// 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 "github.com/apache/iceberg-go" + +// AddedRowsScanTask is a changelog insert produced by adding a data file. +// Matching delete files committed in the same snapshot, or from squashed +// snapshots, are applied while reading so deleted rows are not emitted as +// inserts. +type AddedRowsScanTask struct { + FileScanTask + changeOrdinal int + commitSnapshotID int64 +} + +// NewAddedRowsScanTask constructs an insert task for dataFile. deletes are +// delete files that apply while reading the added file. Position deletes, +// equality deletes, and deletion vectors are stored on the matching +// FileScanTask fields. +func NewAddedRowsScanTask(dataFile iceberg.DataFile, deletes []iceberg.DataFile, changeOrdinal int, commitSnapshotID int64) AddedRowsScanTask { + return AddedRowsScanTask{ + FileScanTask: fileScanTaskWithDeletes(dataFile, deletes), + changeOrdinal: changeOrdinal, + commitSnapshotID: commitSnapshotID, + } +} + +func (t AddedRowsScanTask) ChangeOrdinal() int { return t.changeOrdinal } +func (t AddedRowsScanTask) CommitSnapshotID() int64 { return t.commitSnapshotID } + +// Deletes returns every delete file applied while reading the added data +// file: position deletes, then equality deletes, then deletion vectors. +func (t AddedRowsScanTask) Deletes() []iceberg.DataFile { + return allDeleteFiles(t.FileScanTask) +} + +// DeletedDataFileScanTask is a changelog delete produced by removing a data +// file. ExistingDeletes are delete files that were already present and must +// be applied so only rows that were live when the file was removed appear as +// deletes. +type DeletedDataFileScanTask struct { + FileScanTask + changeOrdinal int + commitSnapshotID int64 +} + +// NewDeletedDataFileScanTask constructs a delete task for a removed data file. +func NewDeletedDataFileScanTask(dataFile iceberg.DataFile, existingDeletes []iceberg.DataFile, changeOrdinal int, commitSnapshotID int64) DeletedDataFileScanTask { + return DeletedDataFileScanTask{ + FileScanTask: fileScanTaskWithDeletes(dataFile, existingDeletes), + changeOrdinal: changeOrdinal, + commitSnapshotID: commitSnapshotID, + } +} + +func (t DeletedDataFileScanTask) ChangeOrdinal() int { return t.changeOrdinal } +func (t DeletedDataFileScanTask) CommitSnapshotID() int64 { return t.commitSnapshotID } + +// ExistingDeletes returns delete files that applied before the data file was +// removed. +func (t DeletedDataFileScanTask) ExistingDeletes() []iceberg.DataFile { + return allDeleteFiles(t.FileScanTask) +} + +// DeletedRowsScanTask is a changelog delete produced by adding delete files +// against a data file that remains in the table. AddedDeletes remove rows +// that should appear in the changelog. ExistingDeletes already applied and +// those rows must not be emitted again. +type DeletedRowsScanTask struct { + FileScanTask + addedDeletes FileScanTask + changeOrdinal int + commitSnapshotID int64 +} + +// NewDeletedRowsScanTask constructs a row-level delete task. existingDeletes +// are stored on the embedded FileScanTask so later readers can reuse the +// normal scan delete path for the live-row baseline. +func NewDeletedRowsScanTask(dataFile iceberg.DataFile, addedDeletes, existingDeletes []iceberg.DataFile, changeOrdinal int, commitSnapshotID int64) DeletedRowsScanTask { + return DeletedRowsScanTask{ + FileScanTask: fileScanTaskWithDeletes(dataFile, existingDeletes), + addedDeletes: fileScanTaskWithDeletes(dataFile, addedDeletes), + changeOrdinal: changeOrdinal, + commitSnapshotID: commitSnapshotID, + } +} + +func (t DeletedRowsScanTask) ChangeOrdinal() int { return t.changeOrdinal } +func (t DeletedRowsScanTask) CommitSnapshotID() int64 { return t.commitSnapshotID } + +// AddedDeletes returns delete files whose removals should appear in the +// changelog. +func (t DeletedRowsScanTask) AddedDeletes() []iceberg.DataFile { + return allDeleteFiles(t.addedDeletes) +} + +// ExistingDeletes returns delete files that already applied before this +// snapshot's added deletes. +func (t DeletedRowsScanTask) ExistingDeletes() []iceberg.DataFile { + return allDeleteFiles(t.FileScanTask) +} + +func fileScanTaskWithDeletes(dataFile iceberg.DataFile, deletes []iceberg.DataFile) FileScanTask { + pos, eq, dv := classifyDeleteFiles(deletes) + return FileScanTask{ + File: dataFile, + DeleteFiles: pos, + EqualityDeleteFiles: eq, + DeletionVectorFiles: dv, + } +} + +func classifyDeleteFiles(files []iceberg.DataFile) (pos, eq, dv []iceberg.DataFile) { + for _, f := range files { + if f == nil { Review Comment: Dropped the nil guard. Real callers get files from `Build()` or manifest entries, so a half-working interface-nil check wasn't worth keeping. ########## table/changelog_scan_task.go: ########## @@ -0,0 +1,152 @@ +// 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 "github.com/apache/iceberg-go" + +// AddedRowsScanTask is a changelog insert produced by adding a data file. +// Matching delete files committed in the same snapshot, or from squashed +// snapshots, are applied while reading so deleted rows are not emitted as +// inserts. +type AddedRowsScanTask struct { + FileScanTask + changeOrdinal int + commitSnapshotID int64 +} + +// NewAddedRowsScanTask constructs an insert task for dataFile. deletes are +// delete files that apply while reading the added file. Position deletes, +// equality deletes, and deletion vectors are stored on the matching +// FileScanTask fields. +func NewAddedRowsScanTask(dataFile iceberg.DataFile, deletes []iceberg.DataFile, changeOrdinal int, commitSnapshotID int64) AddedRowsScanTask { + return AddedRowsScanTask{ + FileScanTask: fileScanTaskWithDeletes(dataFile, deletes), + changeOrdinal: changeOrdinal, + commitSnapshotID: commitSnapshotID, + } +} + +func (t AddedRowsScanTask) ChangeOrdinal() int { return t.changeOrdinal } +func (t AddedRowsScanTask) CommitSnapshotID() int64 { return t.commitSnapshotID } + +// Deletes returns every delete file applied while reading the added data +// file: position deletes, then equality deletes, then deletion vectors. +func (t AddedRowsScanTask) Deletes() []iceberg.DataFile { + return allDeleteFiles(t.FileScanTask) +} + +// DeletedDataFileScanTask is a changelog delete produced by removing a data +// file. ExistingDeletes are delete files that were already present and must +// be applied so only rows that were live when the file was removed appear as +// deletes. +type DeletedDataFileScanTask struct { + FileScanTask + changeOrdinal int + commitSnapshotID int64 +} + +// NewDeletedDataFileScanTask constructs a delete task for a removed data file. +func NewDeletedDataFileScanTask(dataFile iceberg.DataFile, existingDeletes []iceberg.DataFile, changeOrdinal int, commitSnapshotID int64) DeletedDataFileScanTask { + return DeletedDataFileScanTask{ + FileScanTask: fileScanTaskWithDeletes(dataFile, existingDeletes), + changeOrdinal: changeOrdinal, + commitSnapshotID: commitSnapshotID, + } +} + +func (t DeletedDataFileScanTask) ChangeOrdinal() int { return t.changeOrdinal } +func (t DeletedDataFileScanTask) CommitSnapshotID() int64 { return t.commitSnapshotID } + +// ExistingDeletes returns delete files that applied before the data file was +// removed. +func (t DeletedDataFileScanTask) ExistingDeletes() []iceberg.DataFile { + return allDeleteFiles(t.FileScanTask) +} + +// DeletedRowsScanTask is a changelog delete produced by adding delete files +// against a data file that remains in the table. AddedDeletes remove rows +// that should appear in the changelog. ExistingDeletes already applied and +// those rows must not be emitted again. +type DeletedRowsScanTask struct { + FileScanTask + addedDeletes FileScanTask + changeOrdinal int + commitSnapshotID int64 +} + +// NewDeletedRowsScanTask constructs a row-level delete task. existingDeletes +// are stored on the embedded FileScanTask so later readers can reuse the +// normal scan delete path for the live-row baseline. +func NewDeletedRowsScanTask(dataFile iceberg.DataFile, addedDeletes, existingDeletes []iceberg.DataFile, changeOrdinal int, commitSnapshotID int64) DeletedRowsScanTask { + return DeletedRowsScanTask{ + FileScanTask: fileScanTaskWithDeletes(dataFile, existingDeletes), + addedDeletes: fileScanTaskWithDeletes(dataFile, addedDeletes), + changeOrdinal: changeOrdinal, + commitSnapshotID: commitSnapshotID, + } +} + +func (t DeletedRowsScanTask) ChangeOrdinal() int { return t.changeOrdinal } +func (t DeletedRowsScanTask) CommitSnapshotID() int64 { return t.commitSnapshotID } + +// AddedDeletes returns delete files whose removals should appear in the +// changelog. +func (t DeletedRowsScanTask) AddedDeletes() []iceberg.DataFile { + return allDeleteFiles(t.addedDeletes) +} + +// ExistingDeletes returns delete files that already applied before this +// snapshot's added deletes. +func (t DeletedRowsScanTask) ExistingDeletes() []iceberg.DataFile { + return allDeleteFiles(t.FileScanTask) +} + +func fileScanTaskWithDeletes(dataFile iceberg.DataFile, deletes []iceberg.DataFile) FileScanTask { + pos, eq, dv := classifyDeleteFiles(deletes) + return FileScanTask{ + File: dataFile, + DeleteFiles: pos, + EqualityDeleteFiles: eq, + DeletionVectorFiles: dv, + } +} + +func classifyDeleteFiles(files []iceberg.DataFile) (pos, eq, dv []iceberg.DataFile) { + for _, f := range files { + if f == nil { + continue + } + switch { Review Comment: Makes sense. I extracted a shared `classifyDataFile` helper and wired it through both `manifestEntries.merge` and the changelog constructors. Unknown content — including a plain data file in the deletes slice — now returns `ErrInvalidMetadata` instead of being dropped. ########## table/changelog_scan_task.go: ########## @@ -0,0 +1,152 @@ +// 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 "github.com/apache/iceberg-go" + +// AddedRowsScanTask is a changelog insert produced by adding a data file. +// Matching delete files committed in the same snapshot, or from squashed +// snapshots, are applied while reading so deleted rows are not emitted as +// inserts. +type AddedRowsScanTask struct { + FileScanTask + changeOrdinal int + commitSnapshotID int64 +} + +// NewAddedRowsScanTask constructs an insert task for dataFile. deletes are +// delete files that apply while reading the added file. Position deletes, +// equality deletes, and deletion vectors are stored on the matching +// FileScanTask fields. +func NewAddedRowsScanTask(dataFile iceberg.DataFile, deletes []iceberg.DataFile, changeOrdinal int, commitSnapshotID int64) AddedRowsScanTask { + return AddedRowsScanTask{ + FileScanTask: fileScanTaskWithDeletes(dataFile, deletes), + changeOrdinal: changeOrdinal, + commitSnapshotID: commitSnapshotID, + } +} + +func (t AddedRowsScanTask) ChangeOrdinal() int { return t.changeOrdinal } +func (t AddedRowsScanTask) CommitSnapshotID() int64 { return t.commitSnapshotID } + +// Deletes returns every delete file applied while reading the added data +// file: position deletes, then equality deletes, then deletion vectors. +func (t AddedRowsScanTask) Deletes() []iceberg.DataFile { + return allDeleteFiles(t.FileScanTask) +} + +// DeletedDataFileScanTask is a changelog delete produced by removing a data +// file. ExistingDeletes are delete files that were already present and must +// be applied so only rows that were live when the file was removed appear as +// deletes. +type DeletedDataFileScanTask struct { + FileScanTask + changeOrdinal int + commitSnapshotID int64 +} + +// NewDeletedDataFileScanTask constructs a delete task for a removed data file. +func NewDeletedDataFileScanTask(dataFile iceberg.DataFile, existingDeletes []iceberg.DataFile, changeOrdinal int, commitSnapshotID int64) DeletedDataFileScanTask { + return DeletedDataFileScanTask{ + FileScanTask: fileScanTaskWithDeletes(dataFile, existingDeletes), + changeOrdinal: changeOrdinal, + commitSnapshotID: commitSnapshotID, + } +} + +func (t DeletedDataFileScanTask) ChangeOrdinal() int { return t.changeOrdinal } +func (t DeletedDataFileScanTask) CommitSnapshotID() int64 { return t.commitSnapshotID } + +// ExistingDeletes returns delete files that applied before the data file was +// removed. +func (t DeletedDataFileScanTask) ExistingDeletes() []iceberg.DataFile { + return allDeleteFiles(t.FileScanTask) +} + +// DeletedRowsScanTask is a changelog delete produced by adding delete files +// against a data file that remains in the table. AddedDeletes remove rows +// that should appear in the changelog. ExistingDeletes already applied and +// those rows must not be emitted again. +type DeletedRowsScanTask struct { + FileScanTask + addedDeletes FileScanTask + changeOrdinal int + commitSnapshotID int64 +} + +// NewDeletedRowsScanTask constructs a row-level delete task. existingDeletes +// are stored on the embedded FileScanTask so later readers can reuse the +// normal scan delete path for the live-row baseline. +func NewDeletedRowsScanTask(dataFile iceberg.DataFile, addedDeletes, existingDeletes []iceberg.DataFile, changeOrdinal int, commitSnapshotID int64) DeletedRowsScanTask { + return DeletedRowsScanTask{ + FileScanTask: fileScanTaskWithDeletes(dataFile, existingDeletes), + addedDeletes: fileScanTaskWithDeletes(dataFile, addedDeletes), + changeOrdinal: changeOrdinal, + commitSnapshotID: commitSnapshotID, + } +} + +func (t DeletedRowsScanTask) ChangeOrdinal() int { return t.changeOrdinal } +func (t DeletedRowsScanTask) CommitSnapshotID() int64 { return t.commitSnapshotID } + +// AddedDeletes returns delete files whose removals should appear in the +// changelog. +func (t DeletedRowsScanTask) AddedDeletes() []iceberg.DataFile { + return allDeleteFiles(t.addedDeletes) +} + +// ExistingDeletes returns delete files that already applied before this +// snapshot's added deletes. +func (t DeletedRowsScanTask) ExistingDeletes() []iceberg.DataFile { + return allDeleteFiles(t.FileScanTask) +} + +func fileScanTaskWithDeletes(dataFile iceberg.DataFile, deletes []iceberg.DataFile) FileScanTask { + pos, eq, dv := classifyDeleteFiles(deletes) + return FileScanTask{ + File: dataFile, + DeleteFiles: pos, + EqualityDeleteFiles: eq, + DeletionVectorFiles: dv, + } +} + +func classifyDeleteFiles(files []iceberg.DataFile) (pos, eq, dv []iceberg.DataFile) { + for _, f := range files { + if f == nil { + continue + } + switch { + case IsDeletionVector(f): + dv = append(dv, f) + case f.ContentType() == iceberg.EntryContentEqDeletes: + eq = append(eq, f) + case f.ContentType() == iceberg.EntryContentPosDeletes: + pos = append(pos, f) + } + } + return pos, eq, dv Review Comment: Fixed — the returns now have the blank line `nlreturn` wants. Thanks for flagging it. ########## table/changelog_scan_task_test.go: ########## @@ -0,0 +1,80 @@ +// 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 ( + "testing" + + "github.com/apache/iceberg-go" + "github.com/stretchr/testify/require" +) + +func changelogTestDataFile(t *testing.T, path string, content iceberg.ManifestEntryContent, format iceberg.FileFormat) iceberg.DataFile { + t.Helper() + + b, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec, + content, path, format, nil, nil, nil, 10, 1024) + require.NoError(t, err) + + return b.Build() +} + +func TestAddedRowsScanTaskAppliesSameSnapshotDeletes(t *testing.T) { Review Comment: Added `TestClassifyDeleteFiles` for the pos/eq/dv split and for a data file in the deletes slice, which now errors with `ErrInvalidMetadata`. ########## table/changelog_scan_task_test.go: ########## @@ -0,0 +1,80 @@ +// 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 ( + "testing" + + "github.com/apache/iceberg-go" + "github.com/stretchr/testify/require" +) + +func changelogTestDataFile(t *testing.T, path string, content iceberg.ManifestEntryContent, format iceberg.FileFormat) iceberg.DataFile { + t.Helper() + + b, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec, + content, path, format, nil, nil, nil, 10, 1024) + require.NoError(t, err) + + return b.Build() +} + +func TestAddedRowsScanTaskAppliesSameSnapshotDeletes(t *testing.T) { + data := changelogTestDataFile(t, "data/f1.parquet", iceberg.EntryContentData, iceberg.ParquetFile) + posDel := changelogTestDataFile(t, "deletes/d1.parquet", iceberg.EntryContentPosDeletes, iceberg.ParquetFile) + eqDel := changelogTestDataFile(t, "deletes/d2.parquet", iceberg.EntryContentEqDeletes, iceberg.ParquetFile) + dv := changelogTestDataFile(t, "deletes/d3.puffin", iceberg.EntryContentPosDeletes, iceberg.PuffinFile) + + task := NewAddedRowsScanTask(data, []iceberg.DataFile{eqDel, dv, posDel}, 0, 42) + + require.Equal(t, 0, task.ChangeOrdinal()) + require.Equal(t, int64(42), task.CommitSnapshotID()) + require.Equal(t, data.FilePath(), task.File.FilePath()) + require.Equal(t, []iceberg.DataFile{posDel}, task.DeleteFiles) + require.Equal(t, []iceberg.DataFile{eqDel}, task.EqualityDeleteFiles) + require.Equal(t, []iceberg.DataFile{dv}, task.DeletionVectorFiles) + require.Equal(t, []iceberg.DataFile{posDel, eqDel, dv}, task.Deletes()) +} + +func TestDeletedDataFileScanTaskKeepsExistingDeletes(t *testing.T) { + data := changelogTestDataFile(t, "data/f2.parquet", iceberg.EntryContentData, iceberg.ParquetFile) + existing := changelogTestDataFile(t, "deletes/d1.parquet", iceberg.EntryContentPosDeletes, iceberg.ParquetFile) + + task := NewDeletedDataFileScanTask(data, []iceberg.DataFile{existing}, 1, 43) + + require.Equal(t, 1, task.ChangeOrdinal()) + require.Equal(t, int64(43), task.CommitSnapshotID()) + require.Equal(t, []iceberg.DataFile{existing}, task.ExistingDeletes()) + require.Equal(t, []iceberg.DataFile{existing}, task.DeleteFiles) +} + +func TestDeletedRowsScanTaskSeparatesAddedAndExistingDeletes(t *testing.T) { + data := changelogTestDataFile(t, "data/f2.parquet", iceberg.EntryContentData, iceberg.ParquetFile) + added := changelogTestDataFile(t, "deletes/d2.parquet", iceberg.EntryContentEqDeletes, iceberg.ParquetFile) + existing := changelogTestDataFile(t, "deletes/d1.parquet", iceberg.EntryContentPosDeletes, iceberg.ParquetFile) + + task := NewDeletedRowsScanTask(data, []iceberg.DataFile{added}, []iceberg.DataFile{existing}, 2, 44) + + require.Equal(t, 2, task.ChangeOrdinal()) + require.Equal(t, int64(44), task.CommitSnapshotID()) + require.Equal(t, []iceberg.DataFile{added}, task.AddedDeletes()) + require.Equal(t, []iceberg.DataFile{existing}, task.ExistingDeletes()) + require.Equal(t, existing.FilePath(), task.DeleteFiles[0].FilePath()) + require.Empty(t, task.EqualityDeleteFiles) + require.Equal(t, []iceberg.DataFile{added}, task.addedDeletes.EqualityDeleteFiles) Review Comment: Removed it. `AddedDeletes()` already covers the public result, and the classify test now owns the eq-vs-pos split without reaching into the struct layout. -- 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]
