laskoviymishka commented on code in PR #1702:
URL: https://github.com/apache/iceberg-go/pull/1702#discussion_r3749977466
##########
table/arrow_scanner.go:
##########
@@ -1091,14 +1146,23 @@ func (as *arrowScan) recordsFromTask(ctx
context.Context, task tblutils.Enumerat
}()
var (
+ rowFilter iceberg.BooleanExpression
rdr tblutils.FileReader
iceSchema *iceberg.Schema
colIndices []int
filterFunc recProcessFn
dropFile bool
)
- iceSchema, colIndices, rdr, err = as.prepareToRead(ctx, task.Value.File)
+ rowFilter, err = bindTaskFilter(as.metadata.CurrentSchema(),
task.Value.Residual, as.caseSensitive)
Review Comment:
I think there's a subtle trap here. `recordsFromTask` binds the residual
against `as.metadata.CurrentSchema()`, but `ReadTasks` already pre-binds every
residual against `effectiveSchema` (which can be an older snapshot's schema)
before it ever reaches here. Today that's fine — `bindTaskFilter`
short-circuits already-bound expressions, so the current-schema bind is a no-op
on the production path.
But `arrowScan` is internal and the tests instantiate it directly. The
moment a caller hands `recordsFromTask` a task with an unbound residual
referencing a column that lives in a historical schema but not the current one,
it binds against the wrong schema silently — wrong field ID, or a spurious
error. I'd thread `effectiveSchema` into `arrowScan` and bind against that
here, so the invariant is enforced in one place rather than assumed.
While we're here, the assign-then-override-then-check order reads a little
backwards — the `err` from `bindTaskFilter` is checked after the `Residual ==
nil` branch that throws its result away. I'd flip it to `if task.Value.Residual
!= nil { rowFilter, err = ...; if err != nil { return err } } else { rowFilter
= as.boundRowFilter }`. wdyt?
##########
table/incremental_append_scan.go:
##########
@@ -0,0 +1,266 @@
+// 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"
+ "slices"
+ "sort"
+
+ "github.com/apache/iceberg-go"
+)
+
+// IncrementalAppendScan plans data files added by append snapshots between a
+// starting snapshot and an ending snapshot. It follows one snapshot ancestry
+// chain and never returns files inherited from an earlier snapshot.
+type IncrementalAppendScan struct {
+ scan *Scan
+ fromSnapshotID *int64
+ fromInclusive bool
+ toSnapshotID *int64
+}
+
+// NewIncrementalAppendScan creates an incremental append scan. Scan options
+// configure the underlying table scan and are retained for callers that pass
+// snapshot, projection, filter, or concurrency options before planning.
+// Auto planning falls back to local planning. Remote planning returns
+// ErrInvalidOperation until incremental remote planning is implemented.
+func (t Table) NewIncrementalAppendScan(opts ...ScanOption)
*IncrementalAppendScan {
+ return &IncrementalAppendScan{scan: t.Scan(opts...)}
+}
+
+// FromSnapshotInclusive includes files added by the starting snapshot.
+func (s *IncrementalAppendScan) FromSnapshotInclusive(snapshotID int64)
(*IncrementalAppendScan, error) {
+ if s.scan.metadata.SnapshotByID(snapshotID) == nil {
+ return nil, fmt.Errorf("%w: starting snapshot not found: %d",
iceberg.ErrInvalidArgument, snapshotID)
+ }
+ out := *s
+ out.fromSnapshotID = &snapshotID
+ out.fromInclusive = true
+
+ return &out, nil
+}
+
+// FromSnapshotExclusive starts after the given snapshot. The starting
+// snapshot must be an ancestor of the ending snapshot when planning.
+func (s *IncrementalAppendScan) FromSnapshotExclusive(snapshotID int64)
*IncrementalAppendScan {
+ out := *s
+ out.fromSnapshotID = &snapshotID
+ out.fromInclusive = false
+
+ return &out
+}
+
+// ToSnapshot sets the inclusive ending snapshot.
+func (s *IncrementalAppendScan) ToSnapshot(snapshotID int64)
(*IncrementalAppendScan, error) {
+ if s.scan.metadata.SnapshotByID(snapshotID) == nil {
+ return nil, fmt.Errorf("%w: ending snapshot not found: %d",
iceberg.ErrInvalidArgument, snapshotID)
+ }
+ out := *s
+ out.toSnapshotID = &snapshotID
+
+ return &out, nil
+}
+
+// PlanFiles returns one task per newly added data file. Delete files are not
+// applied because appended files are not present before the append snapshot.
+func (s *IncrementalAppendScan) PlanFiles(ctx context.Context)
([]FileScanTask, error) {
+ switch s.scan.planningMode {
+ case ScanPlanningLocal, ScanPlanningAuto:
+ case ScanPlanningRemote:
+ return nil, fmt.Errorf("%w: incremental append scans do not
support remote planning", ErrInvalidOperation)
+ default:
+ return nil, fmt.Errorf("%w: unknown scan planning mode %q",
iceberg.ErrInvalidArgument, s.scan.planningMode)
+ }
+
+ toSnapshot, err := s.toSnapshot()
+ if err != nil {
+ return nil, err
+ }
+ if toSnapshot == nil {
+ if s.fromSnapshotID != nil {
+ return nil, fmt.Errorf("%w: no ending snapshot found
for incremental append scan from %d",
+ iceberg.ErrInvalidArgument, *s.fromSnapshotID)
+ }
+
+ return nil, nil
+ }
+
+ snapshots, err := s.snapshotsBetween(toSnapshot.SnapshotID)
+ if err != nil {
+ return nil, err
+ }
+ if len(snapshots) == 0 {
+ return nil, nil
+ }
+ appendSnapshots := make(map[int64]struct{}, len(snapshots))
+ for _, snapshot := range snapshots {
+ if snapshot.Summary != nil && snapshot.Summary.Operation ==
OpAppend {
+ appendSnapshots[snapshot.SnapshotID] = struct{}{}
+ }
+ }
+ if len(appendSnapshots) == 0 {
+ return nil, nil
+ }
+
+ if s.scan.ioF == nil {
+ return nil, fmt.Errorf("%w: table file IO is not configured",
ErrInvalidOperation)
+ }
+ fs, err := s.scan.ioF(ctx)
+ if err != nil {
+ return nil, err
+ }
+
+ // An inherited manifest can occur in every later snapshot's manifest
list.
+ // Read each manifest path once, just as the Java incremental append
scan
+ // collects the selected manifests into a set before opening them.
+ manifestsByPath := make(map[string]iceberg.ManifestFile)
+ for _, snapshot := range snapshots {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ manifests, err := snapshot.Manifests(fs)
+ if err != nil {
+ return nil, err
+ }
+ for _, manifest := range manifests {
+ if manifest.ManifestContent() !=
iceberg.ManifestContentData {
+ continue
+ }
+ if _, ok := appendSnapshots[manifest.SnapshotID()]; !ok
{
+ continue
+ }
+ manifestsByPath[manifest.FilePath()] = manifest
+ }
+ }
+
+ paths := make([]string, 0, len(manifestsByPath))
+ for path := range manifestsByPath {
+ paths = append(paths, path)
+ }
+ sort.Strings(paths)
+ manifestList := make([]iceberg.ManifestFile, 0, len(paths))
+ for _, path := range paths {
+ manifestList = append(manifestList, manifestsByPath[path])
+ }
+
+ planningScan := *s.scan
+ if s.toSnapshotID != nil {
+ // An explicit end snapshot is a historical scan and must use
that
+ // snapshot's schema. An implicit current end remains a live
scan so a
+ // schema-only metadata update is visible during pruning.
+ planningScan.snapshotID = &toSnapshot.SnapshotID
+ planningScan.asOfTimestamp = nil
+ }
+ schema, err := planningScan.effectiveSchema()
+ if err != nil {
+ return nil, err
+ }
+ manifestList, err =
planningScan.filterManifestsWithSchema(manifestList, schema,
&scanMetricsAccumulator{})
+ if err != nil {
+ return nil, err
+ }
+ if len(manifestList) == 0 {
+ return nil, nil
+ }
+ entries, err := planningScan.collectManifestEntriesWithSchema(ctx,
manifestList, schema)
+ if err != nil {
+ return nil, err
+ }
+
+ tasks := make([]FileScanTask, 0, len(entries.dataEntries))
+ for _, entry := range entries.dataEntries {
+ if entry.Status() != iceberg.EntryStatusADDED {
+ continue
+ }
+ if _, ok := appendSnapshots[entry.SnapshotID()]; !ok {
+ continue
+ }
+ file := entry.DataFile()
+ task := FileScanTask{File: file, Start: 0, Length:
file.FileSizeBytes()}
+ task.Residual = s.scan.rowFilter
+ task.FirstRowID = file.FirstRowID()
+ if sequenceNumber := entry.SequenceNum(); sequenceNumber >= 0 {
+ task.DataSequenceNumber = &sequenceNumber
+ }
+ tasks = append(tasks, task)
+ }
+ sort.Slice(tasks, func(left, right int) bool {
+ return tasks[left].File.FilePath() <
tasks[right].File.FilePath()
+ })
+
+ return tasks, nil
+}
+
+func (s *IncrementalAppendScan) toSnapshot() (*Snapshot, error) {
+ if s.toSnapshotID != nil {
+ return s.scan.metadata.SnapshotByID(*s.toSnapshotID), nil
+ }
+
+ return s.scan.ResolveSnapshot()
+}
+
+func (s *IncrementalAppendScan) snapshotsBetween(toSnapshotID int64)
([]Snapshot, error) {
+ ancestors := AncestorsOf(toSnapshotID, s.scan.metadata.SnapshotByID)
+ if len(ancestors) == 0 {
+ return nil, fmt.Errorf("%w: ending snapshot not found: %d",
iceberg.ErrInvalidArgument, toSnapshotID)
+ }
+
+ if s.fromSnapshotID == nil {
+ slices.Reverse(ancestors)
+
+ return appendOnlySnapshots(ancestors), nil
+ }
+
+ fromID := *s.fromSnapshotID
+ if !s.fromInclusive {
+ between, found := AncestorsBetween(toSnapshotID, fromID,
s.scan.metadata.SnapshotByID)
Review Comment:
When `fromSnapshotID == toSnapshotID` on the exclusive path,
`AncestorsBetween` short-circuits and returns `(nil, true)`, so `found` is
true, we get an empty snapshot list, and `PlanFiles` returns zero tasks with no
error. Java's `BaseIncrementalScan` rejects this — `from == to` exclusive can't
have `from` as a parent-ancestor of `to`, so it throws.
Silently returning empty for what's really an invalid range feels risky: a
caller polling `from=X exclusive, to=X` sees "no new data" rather than a clear
error. I'd guard `from == to` explicitly and return `ErrInvalidArgument`.
There's also no test pinning either from==to case. I'd add one for exclusive
(expect the error) and one for inclusive
`FromSnapshotInclusive(X).ToSnapshot(X)` (expect just X's files) — the
inclusive path leans on `IsAncestorOf(X, X)` returning true, and that's
currently an untested assumption. wdyt?
##########
table/incremental_append_scan.go:
##########
@@ -0,0 +1,266 @@
+// 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"
+ "slices"
+ "sort"
+
+ "github.com/apache/iceberg-go"
+)
+
+// IncrementalAppendScan plans data files added by append snapshots between a
+// starting snapshot and an ending snapshot. It follows one snapshot ancestry
+// chain and never returns files inherited from an earlier snapshot.
+type IncrementalAppendScan struct {
+ scan *Scan
+ fromSnapshotID *int64
+ fromInclusive bool
+ toSnapshotID *int64
+}
+
+// NewIncrementalAppendScan creates an incremental append scan. Scan options
+// configure the underlying table scan and are retained for callers that pass
+// snapshot, projection, filter, or concurrency options before planning.
+// Auto planning falls back to local planning. Remote planning returns
+// ErrInvalidOperation until incremental remote planning is implemented.
+func (t Table) NewIncrementalAppendScan(opts ...ScanOption)
*IncrementalAppendScan {
+ return &IncrementalAppendScan{scan: t.Scan(opts...)}
+}
+
+// FromSnapshotInclusive includes files added by the starting snapshot.
+func (s *IncrementalAppendScan) FromSnapshotInclusive(snapshotID int64)
(*IncrementalAppendScan, error) {
+ if s.scan.metadata.SnapshotByID(snapshotID) == nil {
+ return nil, fmt.Errorf("%w: starting snapshot not found: %d",
iceberg.ErrInvalidArgument, snapshotID)
+ }
+ out := *s
+ out.fromSnapshotID = &snapshotID
+ out.fromInclusive = true
+
+ return &out, nil
+}
+
+// FromSnapshotExclusive starts after the given snapshot. The starting
+// snapshot must be an ancestor of the ending snapshot when planning.
+func (s *IncrementalAppendScan) FromSnapshotExclusive(snapshotID int64)
*IncrementalAppendScan {
+ out := *s
+ out.fromSnapshotID = &snapshotID
+ out.fromInclusive = false
+
+ return &out
+}
+
+// ToSnapshot sets the inclusive ending snapshot.
+func (s *IncrementalAppendScan) ToSnapshot(snapshotID int64)
(*IncrementalAppendScan, error) {
+ if s.scan.metadata.SnapshotByID(snapshotID) == nil {
+ return nil, fmt.Errorf("%w: ending snapshot not found: %d",
iceberg.ErrInvalidArgument, snapshotID)
+ }
+ out := *s
+ out.toSnapshotID = &snapshotID
+
+ return &out, nil
+}
+
+// PlanFiles returns one task per newly added data file. Delete files are not
+// applied because appended files are not present before the append snapshot.
+func (s *IncrementalAppendScan) PlanFiles(ctx context.Context)
([]FileScanTask, error) {
+ switch s.scan.planningMode {
+ case ScanPlanningLocal, ScanPlanningAuto:
+ case ScanPlanningRemote:
+ return nil, fmt.Errorf("%w: incremental append scans do not
support remote planning", ErrInvalidOperation)
+ default:
+ return nil, fmt.Errorf("%w: unknown scan planning mode %q",
iceberg.ErrInvalidArgument, s.scan.planningMode)
+ }
+
+ toSnapshot, err := s.toSnapshot()
+ if err != nil {
+ return nil, err
+ }
+ if toSnapshot == nil {
+ if s.fromSnapshotID != nil {
+ return nil, fmt.Errorf("%w: no ending snapshot found
for incremental append scan from %d",
+ iceberg.ErrInvalidArgument, *s.fromSnapshotID)
+ }
+
+ return nil, nil
+ }
+
+ snapshots, err := s.snapshotsBetween(toSnapshot.SnapshotID)
+ if err != nil {
+ return nil, err
+ }
+ if len(snapshots) == 0 {
+ return nil, nil
+ }
+ appendSnapshots := make(map[int64]struct{}, len(snapshots))
+ for _, snapshot := range snapshots {
+ if snapshot.Summary != nil && snapshot.Summary.Operation ==
OpAppend {
+ appendSnapshots[snapshot.SnapshotID] = struct{}{}
+ }
+ }
+ if len(appendSnapshots) == 0 {
+ return nil, nil
+ }
+
+ if s.scan.ioF == nil {
+ return nil, fmt.Errorf("%w: table file IO is not configured",
ErrInvalidOperation)
+ }
+ fs, err := s.scan.ioF(ctx)
+ if err != nil {
+ return nil, err
+ }
+
+ // An inherited manifest can occur in every later snapshot's manifest
list.
+ // Read each manifest path once, just as the Java incremental append
scan
+ // collects the selected manifests into a set before opening them.
+ manifestsByPath := make(map[string]iceberg.ManifestFile)
+ for _, snapshot := range snapshots {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ manifests, err := snapshot.Manifests(fs)
+ if err != nil {
+ return nil, err
+ }
+ for _, manifest := range manifests {
+ if manifest.ManifestContent() !=
iceberg.ManifestContentData {
+ continue
+ }
+ if _, ok := appendSnapshots[manifest.SnapshotID()]; !ok
{
+ continue
+ }
+ manifestsByPath[manifest.FilePath()] = manifest
+ }
+ }
+
+ paths := make([]string, 0, len(manifestsByPath))
+ for path := range manifestsByPath {
+ paths = append(paths, path)
+ }
+ sort.Strings(paths)
+ manifestList := make([]iceberg.ManifestFile, 0, len(paths))
+ for _, path := range paths {
+ manifestList = append(manifestList, manifestsByPath[path])
+ }
+
+ planningScan := *s.scan
Review Comment:
Not a blocker, and it's consistent with the existing `Scan` builder
shallow-copy pattern, but `planningScan := *s.scan` aliases the reference-typed
fields — `options` (map), `selectedFields` and `identifier` (slices) still
point at the caller's underlying data. Nothing mutates through `planningScan`
today so it's safe, but it's the kind of thing that bites later: a future
change to `filterManifestsWithSchema` that appends to `selectedFields` would
corrupt the caller's scan.
If it's cheap, I'd `maps.Clone` the options and `slices.Clone` the slices on
copy (or add a `Scan.clone()` helper) to make the isolation explicit.
##########
table/incremental_append_scan_test.go:
##########
@@ -0,0 +1,437 @@
+// 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
+
+package table
+
+import (
+ "bytes"
+ "context"
+ "testing"
+
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/stretchr/testify/require"
+)
+
+func TestIncrementalAppendScanSnapshotBoundaries(t *testing.T) {
+ scan := snapshotsTestTable().NewIncrementalAppendScan()
+ inclusive, err := scan.FromSnapshotInclusive(101)
+ require.NoError(t, err)
+ inclusive, err = inclusive.ToSnapshot(102)
+ require.NoError(t, err)
+
+ snapshots, err := inclusive.snapshotsBetween(102)
+ require.NoError(t, err)
+ require.Len(t, snapshots, 1)
+ require.EqualValues(t, 101, snapshots[0].SnapshotID)
+
+ exclusive := scan.FromSnapshotExclusive(101)
+ exclusive, err = exclusive.ToSnapshot(102)
+ require.NoError(t, err)
+ snapshots, err = exclusive.snapshotsBetween(102)
+ require.NoError(t, err)
+ require.Empty(t, snapshots, "the only snapshot after 101 is not an
append")
+}
+
+func TestIncrementalAppendScanRejectsUnknownStart(t *testing.T) {
+ _, err :=
snapshotsTestTable().NewIncrementalAppendScan().FromSnapshotInclusive(999)
+ require.Error(t, err)
Review Comment:
Every other error assertion in this file uses `require.ErrorIs(t, err,
iceberg.ErrInvalidArgument)` — I'd match that here instead of bare
`require.Error`, plus a `require.ErrorContains(t, err, "starting snapshot not
found")` so we're actually pinning the not-found path. Small thing while here:
`999` could be a named `const nonExistentSnapshotID = int64(999)` to mirror the
`expiredSnapshotID` const the neighboring test uses.
##########
table/incremental_append_scan_test.go:
##########
@@ -0,0 +1,437 @@
+// 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
Review Comment:
This header stops at the LICENSE-2.0 URL and is missing the closing
boilerplate ("Unless required by applicable law..." through "...limitations
under the License.") that every other file in the tree carries — compare
`task_residual_test.go` in this same PR. I'd paste the full 16-line header for
consistency.
##########
table/incremental_append_scan.go:
##########
@@ -0,0 +1,266 @@
+// 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"
+ "slices"
+ "sort"
+
+ "github.com/apache/iceberg-go"
+)
+
+// IncrementalAppendScan plans data files added by append snapshots between a
+// starting snapshot and an ending snapshot. It follows one snapshot ancestry
+// chain and never returns files inherited from an earlier snapshot.
+type IncrementalAppendScan struct {
+ scan *Scan
+ fromSnapshotID *int64
+ fromInclusive bool
+ toSnapshotID *int64
+}
+
+// NewIncrementalAppendScan creates an incremental append scan. Scan options
+// configure the underlying table scan and are retained for callers that pass
+// snapshot, projection, filter, or concurrency options before planning.
+// Auto planning falls back to local planning. Remote planning returns
+// ErrInvalidOperation until incremental remote planning is implemented.
+func (t Table) NewIncrementalAppendScan(opts ...ScanOption)
*IncrementalAppendScan {
+ return &IncrementalAppendScan{scan: t.Scan(opts...)}
+}
+
+// FromSnapshotInclusive includes files added by the starting snapshot.
+func (s *IncrementalAppendScan) FromSnapshotInclusive(snapshotID int64)
(*IncrementalAppendScan, error) {
+ if s.scan.metadata.SnapshotByID(snapshotID) == nil {
+ return nil, fmt.Errorf("%w: starting snapshot not found: %d",
iceberg.ErrInvalidArgument, snapshotID)
+ }
+ out := *s
+ out.fromSnapshotID = &snapshotID
+ out.fromInclusive = true
+
+ return &out, nil
+}
+
+// FromSnapshotExclusive starts after the given snapshot. The starting
+// snapshot must be an ancestor of the ending snapshot when planning.
+func (s *IncrementalAppendScan) FromSnapshotExclusive(snapshotID int64)
*IncrementalAppendScan {
+ out := *s
+ out.fromSnapshotID = &snapshotID
+ out.fromInclusive = false
+
+ return &out
+}
+
+// ToSnapshot sets the inclusive ending snapshot.
+func (s *IncrementalAppendScan) ToSnapshot(snapshotID int64)
(*IncrementalAppendScan, error) {
+ if s.scan.metadata.SnapshotByID(snapshotID) == nil {
+ return nil, fmt.Errorf("%w: ending snapshot not found: %d",
iceberg.ErrInvalidArgument, snapshotID)
+ }
+ out := *s
+ out.toSnapshotID = &snapshotID
+
+ return &out, nil
+}
+
+// PlanFiles returns one task per newly added data file. Delete files are not
+// applied because appended files are not present before the append snapshot.
+func (s *IncrementalAppendScan) PlanFiles(ctx context.Context)
([]FileScanTask, error) {
+ switch s.scan.planningMode {
+ case ScanPlanningLocal, ScanPlanningAuto:
+ case ScanPlanningRemote:
+ return nil, fmt.Errorf("%w: incremental append scans do not
support remote planning", ErrInvalidOperation)
+ default:
+ return nil, fmt.Errorf("%w: unknown scan planning mode %q",
iceberg.ErrInvalidArgument, s.scan.planningMode)
+ }
+
+ toSnapshot, err := s.toSnapshot()
+ if err != nil {
+ return nil, err
+ }
+ if toSnapshot == nil {
+ if s.fromSnapshotID != nil {
+ return nil, fmt.Errorf("%w: no ending snapshot found
for incremental append scan from %d",
+ iceberg.ErrInvalidArgument, *s.fromSnapshotID)
+ }
+
+ return nil, nil
+ }
+
+ snapshots, err := s.snapshotsBetween(toSnapshot.SnapshotID)
+ if err != nil {
+ return nil, err
+ }
+ if len(snapshots) == 0 {
+ return nil, nil
+ }
+ appendSnapshots := make(map[int64]struct{}, len(snapshots))
+ for _, snapshot := range snapshots {
+ if snapshot.Summary != nil && snapshot.Summary.Operation ==
OpAppend {
Review Comment:
This append-only filter is dead code — `snapshotsBetween` already runs
everything through `appendOnlySnapshots`, so every snapshot reaching here is
guaranteed `OpAppend` and this `if` is always true (and the
`len(appendSnapshots) == 0` check just below is unreachable, given the
`len(snapshots) == 0` guard above it).
Harmless today, but it makes it look like there are two independent
append-only fences when there's really one. The actual invariant is the
`appendSnapshots` membership checks at the manifest level and entry level below
— if someone later widens `snapshotsBetween` to return all snapshots,
`appendSnapshots` silently fills with overwrite/delete IDs and this filter
won't save us. I'd pick a single authority: keep `snapshotsBetween` append-only
and drop this guard with a comment, or have it return all snapshots and let
`PlanFiles` do the filtering.
Related — I don't see an end-to-end test that drives a range containing a
non-append snapshot (append->overwrite->append) through `PlanFiles` and asserts
only the two appends' files show up.
`TestIncrementalAppendScanSnapshotBoundaries` checks `snapshotsBetween`
directly, but the manifest/entry-level fence, the thing that actually protects
correctness, isn't exercised in the full planning flow. I'd add that fixture.
##########
table/incremental_append_scan.go:
##########
@@ -0,0 +1,266 @@
+// 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"
+ "slices"
+ "sort"
+
+ "github.com/apache/iceberg-go"
+)
+
+// IncrementalAppendScan plans data files added by append snapshots between a
+// starting snapshot and an ending snapshot. It follows one snapshot ancestry
+// chain and never returns files inherited from an earlier snapshot.
+type IncrementalAppendScan struct {
+ scan *Scan
+ fromSnapshotID *int64
+ fromInclusive bool
+ toSnapshotID *int64
+}
+
+// NewIncrementalAppendScan creates an incremental append scan. Scan options
+// configure the underlying table scan and are retained for callers that pass
+// snapshot, projection, filter, or concurrency options before planning.
+// Auto planning falls back to local planning. Remote planning returns
+// ErrInvalidOperation until incremental remote planning is implemented.
+func (t Table) NewIncrementalAppendScan(opts ...ScanOption)
*IncrementalAppendScan {
+ return &IncrementalAppendScan{scan: t.Scan(opts...)}
+}
+
+// FromSnapshotInclusive includes files added by the starting snapshot.
+func (s *IncrementalAppendScan) FromSnapshotInclusive(snapshotID int64)
(*IncrementalAppendScan, error) {
+ if s.scan.metadata.SnapshotByID(snapshotID) == nil {
+ return nil, fmt.Errorf("%w: starting snapshot not found: %d",
iceberg.ErrInvalidArgument, snapshotID)
+ }
+ out := *s
+ out.fromSnapshotID = &snapshotID
+ out.fromInclusive = true
+
+ return &out, nil
+}
+
+// FromSnapshotExclusive starts after the given snapshot. The starting
+// snapshot must be an ancestor of the ending snapshot when planning.
+func (s *IncrementalAppendScan) FromSnapshotExclusive(snapshotID int64)
*IncrementalAppendScan {
Review Comment:
The three builder methods return different shapes — `FromSnapshotInclusive`
and `ToSnapshot` return `(*IncrementalAppendScan, error)`, but
`FromSnapshotExclusive` returns just `*IncrementalAppendScan`. That breaks
uniform chaining, and since this is a brand-new public API, I'd settle it
before it ships.
The asymmetry also means a bogus exclusive from-ID gets no feedback until
`PlanFiles`, and the error there ("starting snapshot X is not an ancestor of
ending snapshot Y") is misleading when the snapshot simply doesn't exist. I get
that exclusive intentionally tolerates an expired/pruned parent (the
expired-exclusive test), so eager existence-checking isn't quite right either.
Either way is fine, but it should be deliberate: move all validation into
`PlanFiles` and have every builder return `*IncrementalAppendScan` only
(matches how `Scan` defers errors to `PlanFiles`, most idiomatic here), or make
all three return `(*..., error)` and document why exclusive still allows a
missing/expired from-snapshot. wdyt?
##########
table/incremental_append_scan.go:
##########
@@ -0,0 +1,266 @@
+// 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"
+ "slices"
+ "sort"
+
+ "github.com/apache/iceberg-go"
+)
+
+// IncrementalAppendScan plans data files added by append snapshots between a
+// starting snapshot and an ending snapshot. It follows one snapshot ancestry
+// chain and never returns files inherited from an earlier snapshot.
+type IncrementalAppendScan struct {
+ scan *Scan
+ fromSnapshotID *int64
+ fromInclusive bool
+ toSnapshotID *int64
+}
+
+// NewIncrementalAppendScan creates an incremental append scan. Scan options
+// configure the underlying table scan and are retained for callers that pass
+// snapshot, projection, filter, or concurrency options before planning.
+// Auto planning falls back to local planning. Remote planning returns
+// ErrInvalidOperation until incremental remote planning is implemented.
+func (t Table) NewIncrementalAppendScan(opts ...ScanOption)
*IncrementalAppendScan {
+ return &IncrementalAppendScan{scan: t.Scan(opts...)}
+}
+
+// FromSnapshotInclusive includes files added by the starting snapshot.
+func (s *IncrementalAppendScan) FromSnapshotInclusive(snapshotID int64)
(*IncrementalAppendScan, error) {
+ if s.scan.metadata.SnapshotByID(snapshotID) == nil {
+ return nil, fmt.Errorf("%w: starting snapshot not found: %d",
iceberg.ErrInvalidArgument, snapshotID)
+ }
+ out := *s
+ out.fromSnapshotID = &snapshotID
+ out.fromInclusive = true
+
+ return &out, nil
+}
+
+// FromSnapshotExclusive starts after the given snapshot. The starting
+// snapshot must be an ancestor of the ending snapshot when planning.
+func (s *IncrementalAppendScan) FromSnapshotExclusive(snapshotID int64)
*IncrementalAppendScan {
+ out := *s
+ out.fromSnapshotID = &snapshotID
+ out.fromInclusive = false
+
+ return &out
+}
+
+// ToSnapshot sets the inclusive ending snapshot.
+func (s *IncrementalAppendScan) ToSnapshot(snapshotID int64)
(*IncrementalAppendScan, error) {
+ if s.scan.metadata.SnapshotByID(snapshotID) == nil {
+ return nil, fmt.Errorf("%w: ending snapshot not found: %d",
iceberg.ErrInvalidArgument, snapshotID)
+ }
+ out := *s
+ out.toSnapshotID = &snapshotID
+
+ return &out, nil
+}
+
+// PlanFiles returns one task per newly added data file. Delete files are not
+// applied because appended files are not present before the append snapshot.
+func (s *IncrementalAppendScan) PlanFiles(ctx context.Context)
([]FileScanTask, error) {
+ switch s.scan.planningMode {
+ case ScanPlanningLocal, ScanPlanningAuto:
+ case ScanPlanningRemote:
+ return nil, fmt.Errorf("%w: incremental append scans do not
support remote planning", ErrInvalidOperation)
+ default:
+ return nil, fmt.Errorf("%w: unknown scan planning mode %q",
iceberg.ErrInvalidArgument, s.scan.planningMode)
+ }
+
+ toSnapshot, err := s.toSnapshot()
+ if err != nil {
+ return nil, err
+ }
+ if toSnapshot == nil {
+ if s.fromSnapshotID != nil {
+ return nil, fmt.Errorf("%w: no ending snapshot found
for incremental append scan from %d",
+ iceberg.ErrInvalidArgument, *s.fromSnapshotID)
+ }
+
+ return nil, nil
+ }
+
+ snapshots, err := s.snapshotsBetween(toSnapshot.SnapshotID)
+ if err != nil {
+ return nil, err
+ }
+ if len(snapshots) == 0 {
+ return nil, nil
+ }
+ appendSnapshots := make(map[int64]struct{}, len(snapshots))
+ for _, snapshot := range snapshots {
+ if snapshot.Summary != nil && snapshot.Summary.Operation ==
OpAppend {
+ appendSnapshots[snapshot.SnapshotID] = struct{}{}
+ }
+ }
+ if len(appendSnapshots) == 0 {
+ return nil, nil
+ }
+
+ if s.scan.ioF == nil {
+ return nil, fmt.Errorf("%w: table file IO is not configured",
ErrInvalidOperation)
+ }
+ fs, err := s.scan.ioF(ctx)
+ if err != nil {
+ return nil, err
+ }
+
+ // An inherited manifest can occur in every later snapshot's manifest
list.
+ // Read each manifest path once, just as the Java incremental append
scan
+ // collects the selected manifests into a set before opening them.
+ manifestsByPath := make(map[string]iceberg.ManifestFile)
+ for _, snapshot := range snapshots {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ manifests, err := snapshot.Manifests(fs)
+ if err != nil {
+ return nil, err
+ }
+ for _, manifest := range manifests {
+ if manifest.ManifestContent() !=
iceberg.ManifestContentData {
+ continue
+ }
+ if _, ok := appendSnapshots[manifest.SnapshotID()]; !ok
{
+ continue
+ }
+ manifestsByPath[manifest.FilePath()] = manifest
+ }
+ }
+
+ paths := make([]string, 0, len(manifestsByPath))
+ for path := range manifestsByPath {
+ paths = append(paths, path)
+ }
+ sort.Strings(paths)
Review Comment:
The surrounding code uses `slices` consistently, so I'd swap
`sort.Strings(paths)` for `slices.Sort(paths)` here and the `sort.Slice(tasks,
...)` below for `slices.SortFunc(tasks, func(a, b FileScanTask) int { return
cmp.Compare(a.File.FilePath(), b.File.FilePath()) })`, then drop the `sort`
import.
--
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]
