laskoviymishka commented on code in PR #1897: URL: https://github.com/apache/iceberg-go/pull/1897#discussion_r3874747555
########## 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: Design-direction question for this first slice: Java has these three implement a common `ChangelogScanTask` interface with `operation()`, `changeOrdinal()`, `commitSnapshotId()`. Here each type carries `ChangeOrdinal`/`CommitSnapshotID` but there's no shared interface and no `Operation()`. Without it the planning follow-up can't return a uniform `[]ChangelogScanTask` and every consumer needs a type switch to tell inserts from deletes. I'd lean toward defining the interface plus a `ChangelogOperation` enum now so the follow-ups have something to build against, but if you'd rather defer until planning lands that's reasonable too. wdyt? ########## 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: This switch has no default, so any file that isn't a DV, eq-delete, or pos-delete (a plain data file, `ContentType` 0) is silently dropped. `scanner.go` classifies into these same three buckets but returns a wrapped `ErrInvalidMetadata` on unknown content (scanner.go:776-791). I'd match that here rather than swallow it, since once the planner is feeding this, a misrouted data file would produce a silently-wrong changelog instead of a loud failure. Given `scanner.go` already does this exact three-way split with the error branch, it's worth extracting one shared helper so the two don't drift. Returning an error means threading it through `fileScanTaskWithDeletes` and the constructors, but I think that's the right trade. ########## 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: Minor, but this nil guard is a little misleading: `f` is an interface, so a typed-nil (a `(*dataFile)(nil)` appended to the slice) passes `f == nil` and then panics on `IsDeletionVector`'s `FileFormat()` call right below. Real callers get `DataFile`s from `Build()` or manifest entries, neither of which produces a typed-nil, so I'd just drop the guard and note that nil elements aren't a supported input rather than half-guarding against a case that can't happen. ########## 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: I'd hold off on modeling `addedDeletes` as a `FileScanTask`. Java keeps it as a plain `List<DeleteFile>`, and here we get a second `.File` pointing at the same data file that no method ever reads, plus a whole `FileScanTask` whose `Start`/`Length`/`FirstRowID`/`DataSequenceNumber` are all zero/nil. That last part is the real trap: once the reader slice lands and passes this into the read path, the zeroed range and nil `FirstRowID`/`DataSequenceNumber` will read as "intentionally absent" and silently suppress row-lineage synthesis (arrow_scanner gates on those being non-nil), so the output looks valid but is wrong. I'd store the classified lists directly, a small internal struct like `classifiedDeletes{pos, eq, dv []iceberg.DataFile}`, and have `AddedDeletes()` flatten it. That makes the incomplete-metadata state impossible to express. wdyt? ########## 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: `nlreturn` wants a blank line before this return, and the same before `return out` in `allDeleteFiles` just below. CI will fail on both until they're added. Quick fix, but it's the thing currently keeping the build red. ########## 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: The assertion a few lines up already checks `AddedDeletes()` returns `[]{added}`; this one reaches into the unexported `addedDeletes.EqualityDeleteFiles` to assert the same fact through internal layout. I'd drop it. If the goal is to prove eq-vs-pos classification, a dedicated `classifyDeleteFiles` unit test reads better and doesn't couple to the struct shape. ########## 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: These three tests only exercise the happy path: every input is a cleanly-typed delete file. There's no case for the branch that matters most, a plain data file (or nil) in the deletes slice, which today is silently dropped. If that path becomes an error (per the `classifyDeleteFiles` comment), I'd want a test asserting the error; if it stays a skip, a test that documents the intent. -- 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]
