laskoviymishka commented on code in PR #1883: URL: https://github.com/apache/iceberg-go/pull/1883#discussion_r3893131435
########## table/incremental_changelog_scan.go: ########## @@ -0,0 +1,322 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "cmp" + "context" + "fmt" + "maps" + "slices" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" +) + +// ChangelogOperation identifies the change represented by a changelog task. +type ChangelogOperation string + +const ( + ChangelogOperationInsert ChangelogOperation = "insert" + ChangelogOperationDelete ChangelogOperation = "delete" +) + +// ChangelogScanTask plans one data file that contributes a change to the +// table's changelog. The embedded FileScanTask can be passed to Scan.ReadTasks +// to read the file contents. +type ChangelogScanTask struct { + FileScanTask + Operation ChangelogOperation + ChangeOrdinal int + CommitSnapshotID int64 +} + +// IncrementalChangelogScan plans data-file changes between snapshots. It +// emits insert and delete tasks for data-manifest entries and skips replace +// snapshots. PlanFiles returns an error if any in-range snapshot's manifest +// list references a delete manifest, including ones carried forward from +// earlier snapshots. +type IncrementalChangelogScan struct { + scan *Scan + fromSnapshotID *int64 + fromInclusive bool + toSnapshotID *int64 +} + +// NewIncrementalChangelogScan creates an incremental changelog planner. +// Projection and row limits are not applied to returned tasks. Auto planning +// falls back to local planning, while remote planning is not supported. +func (t Table) NewIncrementalChangelogScan(opts ...ScanOption) *IncrementalChangelogScan { + return &IncrementalChangelogScan{scan: t.Scan(opts...)} +} + +// FromSnapshotInclusive includes changes committed by the starting snapshot. +func (s *IncrementalChangelogScan) FromSnapshotInclusive(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.fromSnapshotID = &snapshotID + out.fromInclusive = true + + return &out +} + +// FromSnapshotExclusive starts after the given snapshot. The starting +// snapshot must be a parent ancestor of the ending snapshot when planning. +func (s *IncrementalChangelogScan) FromSnapshotExclusive(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.fromSnapshotID = &snapshotID + out.fromInclusive = false + + return &out +} + +// ToSnapshot sets the inclusive ending snapshot. +func (s *IncrementalChangelogScan) ToSnapshot(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.toSnapshotID = &snapshotID + + return &out +} + +// PlanFiles returns one task for each added or deleted data-file entry and +// emits a ScanReport through the configured reporter on successful planning. +func (s *IncrementalChangelogScan) PlanFiles(ctx context.Context) ([]ChangelogScanTask, error) { + switch s.scan.planningMode { + case ScanPlanningLocal, ScanPlanningAuto: + case ScanPlanningRemote: + return nil, fmt.Errorf("%w: incremental changelog scans do not support remote planning", ErrInvalidOperation) + default: + return nil, fmt.Errorf("%w: unknown scan planning mode %q", iceberg.ErrInvalidArgument, s.scan.planningMode) + } + start := time.Now() + + 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 changelog scan from %d", + iceberg.ErrInvalidArgument, *s.fromSnapshotID) + } + + return nil, nil + } + + planningScan := *s.scan + planningScan.identifier = slices.Clone(s.scan.identifier) + planningScan.selectedFields = slices.Clone(s.scan.selectedFields) + planningScan.options = maps.Clone(s.scan.options) + if s.toSnapshotID != nil { + planningScan.snapshotID = &toSnapshot.SnapshotID + planningScan.asOfTimestamp = nil + } + schema, err := planningScan.effectiveSchema() + if err != nil { + return nil, err + } + residual, err := bindTaskFilter(schema, planningScan.rowFilter, planningScan.caseSensitive) + if err != nil { + return nil, fmt.Errorf("bind incremental changelog scan residual: %w", err) + } + var acc scanMetricsAccumulator + finish := func(tasks []ChangelogScanTask) ([]ChangelogScanTask, error) { + acc.resultDataFiles = int64(len(tasks)) + for _, task := range tasks { + acc.totalFileSize += task.File.FileSizeBytes() + } + planningDuration := time.Since(start) + + if rep := planningScan.Reporter(); !metrics.IsNop(rep) { + projected, _ := planningScan.Projection() + safeReport(ctx, rep, planningScan.buildScanReport(&acc, schema, projected, planningDuration)) + } + + return tasks, nil + } + + snapshotRange, err := incrementalSnapshotsBetween( + s.scan.metadata, s.fromSnapshotID, s.fromInclusive, toSnapshot.SnapshotID) + if err != nil { + return nil, err + } + snapshots, err := changelogSnapshots(snapshotRange) + if err != nil { + return nil, err + } + if len(snapshots) == 0 { + return finish(nil) + } + + changelogSnapshotIDs := make(map[int64]struct{}, len(snapshots)) + snapshotOrdinals := make(map[int64]int, len(snapshots)) + for ordinal, snapshot := range snapshots { + changelogSnapshotIDs[snapshot.SnapshotID] = struct{}{} + snapshotOrdinals[snapshot.SnapshotID] = ordinal + } + + 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 + } + + 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.ManifestContentDeletes { + return nil, fmt.Errorf("%w: incremental changelog scans do not support delete manifests from snapshot %d", + ErrInvalidOperation, manifest.SnapshotID()) + } + if manifest.ManifestContent() != iceberg.ManifestContentData { + continue + } + if _, ok := changelogSnapshotIDs[manifest.SnapshotID()]; !ok { + continue + } + manifestsByPath[manifest.FilePath()] = manifest + } + } + + paths := make([]string, 0, len(manifestsByPath)) + for path := range manifestsByPath { + paths = append(paths, path) + } + slices.Sort(paths) + manifestList := make([]iceberg.ManifestFile, 0, len(paths)) + for _, path := range paths { + manifestList = append(manifestList, manifestsByPath[path]) + } + + manifestList = slices.DeleteFunc(manifestList, func(manifest iceberg.ManifestFile) bool { Review Comment: This pre-filter runs before filterManifestsWithSchema increments totalDataManifests, so the ScanReport for a changelog scan counts fewer manifests than a regular scan over the same endpoint (the test asserts 3; a normal scan would see more). That's a defensible choice, but an operator comparing the two reports would read it as an efficiency difference that isn't real. I'd add a comment at the filter site noting totalDataManifests intentionally excludes no-change manifests and diverges from Java's count. tanmayrauth's already got a thread open on SkipsManifestsWithoutChanges not pinning this down; asserting the manifest count there closes both, so I'd fold it into his thread rather than duplicate it. ########## table/incremental_changelog_scan.go: ########## @@ -0,0 +1,322 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "cmp" + "context" + "fmt" + "maps" + "slices" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" +) + +// ChangelogOperation identifies the change represented by a changelog task. +type ChangelogOperation string + +const ( + ChangelogOperationInsert ChangelogOperation = "insert" + ChangelogOperationDelete ChangelogOperation = "delete" +) + +// ChangelogScanTask plans one data file that contributes a change to the +// table's changelog. The embedded FileScanTask can be passed to Scan.ReadTasks +// to read the file contents. +type ChangelogScanTask struct { + FileScanTask + Operation ChangelogOperation + ChangeOrdinal int + CommitSnapshotID int64 +} + +// IncrementalChangelogScan plans data-file changes between snapshots. It +// emits insert and delete tasks for data-manifest entries and skips replace +// snapshots. PlanFiles returns an error if any in-range snapshot's manifest +// list references a delete manifest, including ones carried forward from +// earlier snapshots. +type IncrementalChangelogScan struct { + scan *Scan + fromSnapshotID *int64 + fromInclusive bool + toSnapshotID *int64 +} + +// NewIncrementalChangelogScan creates an incremental changelog planner. +// Projection and row limits are not applied to returned tasks. Auto planning +// falls back to local planning, while remote planning is not supported. +func (t Table) NewIncrementalChangelogScan(opts ...ScanOption) *IncrementalChangelogScan { + return &IncrementalChangelogScan{scan: t.Scan(opts...)} +} + +// FromSnapshotInclusive includes changes committed by the starting snapshot. +func (s *IncrementalChangelogScan) FromSnapshotInclusive(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.fromSnapshotID = &snapshotID + out.fromInclusive = true + + return &out +} + +// FromSnapshotExclusive starts after the given snapshot. The starting +// snapshot must be a parent ancestor of the ending snapshot when planning. +func (s *IncrementalChangelogScan) FromSnapshotExclusive(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.fromSnapshotID = &snapshotID + out.fromInclusive = false + + return &out +} + +// ToSnapshot sets the inclusive ending snapshot. +func (s *IncrementalChangelogScan) ToSnapshot(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.toSnapshotID = &snapshotID + + return &out +} + +// PlanFiles returns one task for each added or deleted data-file entry and +// emits a ScanReport through the configured reporter on successful planning. +func (s *IncrementalChangelogScan) PlanFiles(ctx context.Context) ([]ChangelogScanTask, error) { + switch s.scan.planningMode { + case ScanPlanningLocal, ScanPlanningAuto: + case ScanPlanningRemote: + return nil, fmt.Errorf("%w: incremental changelog scans do not support remote planning", ErrInvalidOperation) + default: + return nil, fmt.Errorf("%w: unknown scan planning mode %q", iceberg.ErrInvalidArgument, s.scan.planningMode) + } + start := time.Now() + + 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 changelog scan from %d", + iceberg.ErrInvalidArgument, *s.fromSnapshotID) + } + + return nil, nil + } + + planningScan := *s.scan + planningScan.identifier = slices.Clone(s.scan.identifier) + planningScan.selectedFields = slices.Clone(s.scan.selectedFields) + planningScan.options = maps.Clone(s.scan.options) + if s.toSnapshotID != nil { + planningScan.snapshotID = &toSnapshot.SnapshotID + planningScan.asOfTimestamp = nil + } + schema, err := planningScan.effectiveSchema() + if err != nil { + return nil, err + } + residual, err := bindTaskFilter(schema, planningScan.rowFilter, planningScan.caseSensitive) + if err != nil { + return nil, fmt.Errorf("bind incremental changelog scan residual: %w", err) + } + var acc scanMetricsAccumulator + finish := func(tasks []ChangelogScanTask) ([]ChangelogScanTask, error) { + acc.resultDataFiles = int64(len(tasks)) + for _, task := range tasks { + acc.totalFileSize += task.File.FileSizeBytes() + } + planningDuration := time.Since(start) + + if rep := planningScan.Reporter(); !metrics.IsNop(rep) { + projected, _ := planningScan.Projection() + safeReport(ctx, rep, planningScan.buildScanReport(&acc, schema, projected, planningDuration)) + } + + return tasks, nil + } + + snapshotRange, err := incrementalSnapshotsBetween( + s.scan.metadata, s.fromSnapshotID, s.fromInclusive, toSnapshot.SnapshotID) + if err != nil { + return nil, err + } + snapshots, err := changelogSnapshots(snapshotRange) + if err != nil { + return nil, err + } + if len(snapshots) == 0 { + return finish(nil) + } + + changelogSnapshotIDs := make(map[int64]struct{}, len(snapshots)) + snapshotOrdinals := make(map[int64]int, len(snapshots)) + for ordinal, snapshot := range snapshots { + changelogSnapshotIDs[snapshot.SnapshotID] = struct{}{} + snapshotOrdinals[snapshot.SnapshotID] = ordinal + } + + 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 + } + + 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.ManifestContentDeletes { + return nil, fmt.Errorf("%w: incremental changelog scans do not support delete manifests from snapshot %d", + ErrInvalidOperation, manifest.SnapshotID()) Review Comment: manifest.SnapshotID() here is the snapshot that wrote the delete manifest, not the one we're iterating. With FromSnapshotExclusive(2) this prints "from snapshot 2" even though 2 is the excluded start, outside the caller's range (the test encodes exactly that), so it reads as if 2 were in range. Small thing: I'd either drop the ID the way Java does, or rephrase to something like "carried into the scan range from snapshot %d" so it's clear the named snapshot is the source, not a member. wdyt? ########## table/incremental_changelog_scan.go: ########## @@ -0,0 +1,322 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "cmp" + "context" + "fmt" + "maps" + "slices" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" +) + +// ChangelogOperation identifies the change represented by a changelog task. +type ChangelogOperation string + +const ( + ChangelogOperationInsert ChangelogOperation = "insert" + ChangelogOperationDelete ChangelogOperation = "delete" +) + +// ChangelogScanTask plans one data file that contributes a change to the +// table's changelog. The embedded FileScanTask can be passed to Scan.ReadTasks +// to read the file contents. +type ChangelogScanTask struct { + FileScanTask + Operation ChangelogOperation + ChangeOrdinal int + CommitSnapshotID int64 +} + +// IncrementalChangelogScan plans data-file changes between snapshots. It +// emits insert and delete tasks for data-manifest entries and skips replace +// snapshots. PlanFiles returns an error if any in-range snapshot's manifest +// list references a delete manifest, including ones carried forward from +// earlier snapshots. +type IncrementalChangelogScan struct { + scan *Scan + fromSnapshotID *int64 + fromInclusive bool + toSnapshotID *int64 +} + +// NewIncrementalChangelogScan creates an incremental changelog planner. +// Projection and row limits are not applied to returned tasks. Auto planning +// falls back to local planning, while remote planning is not supported. +func (t Table) NewIncrementalChangelogScan(opts ...ScanOption) *IncrementalChangelogScan { + return &IncrementalChangelogScan{scan: t.Scan(opts...)} +} + +// FromSnapshotInclusive includes changes committed by the starting snapshot. +func (s *IncrementalChangelogScan) FromSnapshotInclusive(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.fromSnapshotID = &snapshotID + out.fromInclusive = true + + return &out +} + +// FromSnapshotExclusive starts after the given snapshot. The starting +// snapshot must be a parent ancestor of the ending snapshot when planning. +func (s *IncrementalChangelogScan) FromSnapshotExclusive(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.fromSnapshotID = &snapshotID + out.fromInclusive = false + + return &out +} + +// ToSnapshot sets the inclusive ending snapshot. +func (s *IncrementalChangelogScan) ToSnapshot(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.toSnapshotID = &snapshotID + + return &out +} + +// PlanFiles returns one task for each added or deleted data-file entry and +// emits a ScanReport through the configured reporter on successful planning. +func (s *IncrementalChangelogScan) PlanFiles(ctx context.Context) ([]ChangelogScanTask, error) { + switch s.scan.planningMode { + case ScanPlanningLocal, ScanPlanningAuto: + case ScanPlanningRemote: + return nil, fmt.Errorf("%w: incremental changelog scans do not support remote planning", ErrInvalidOperation) + default: + return nil, fmt.Errorf("%w: unknown scan planning mode %q", iceberg.ErrInvalidArgument, s.scan.planningMode) + } + start := time.Now() + + 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 changelog scan from %d", + iceberg.ErrInvalidArgument, *s.fromSnapshotID) + } + + return nil, nil + } + + planningScan := *s.scan + planningScan.identifier = slices.Clone(s.scan.identifier) + planningScan.selectedFields = slices.Clone(s.scan.selectedFields) + planningScan.options = maps.Clone(s.scan.options) + if s.toSnapshotID != nil { + planningScan.snapshotID = &toSnapshot.SnapshotID + planningScan.asOfTimestamp = nil + } + schema, err := planningScan.effectiveSchema() + if err != nil { + return nil, err + } + residual, err := bindTaskFilter(schema, planningScan.rowFilter, planningScan.caseSensitive) + if err != nil { + return nil, fmt.Errorf("bind incremental changelog scan residual: %w", err) + } + var acc scanMetricsAccumulator + finish := func(tasks []ChangelogScanTask) ([]ChangelogScanTask, error) { + acc.resultDataFiles = int64(len(tasks)) + for _, task := range tasks { + acc.totalFileSize += task.File.FileSizeBytes() + } + planningDuration := time.Since(start) + + if rep := planningScan.Reporter(); !metrics.IsNop(rep) { + projected, _ := planningScan.Projection() + safeReport(ctx, rep, planningScan.buildScanReport(&acc, schema, projected, planningDuration)) + } + + return tasks, nil + } + + snapshotRange, err := incrementalSnapshotsBetween( + s.scan.metadata, s.fromSnapshotID, s.fromInclusive, toSnapshot.SnapshotID) + if err != nil { + return nil, err + } + snapshots, err := changelogSnapshots(snapshotRange) + if err != nil { + return nil, err + } + if len(snapshots) == 0 { + return finish(nil) + } + + changelogSnapshotIDs := make(map[int64]struct{}, len(snapshots)) + snapshotOrdinals := make(map[int64]int, len(snapshots)) + for ordinal, snapshot := range snapshots { + changelogSnapshotIDs[snapshot.SnapshotID] = struct{}{} + snapshotOrdinals[snapshot.SnapshotID] = ordinal + } + + 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 + } + + 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.ManifestContentDeletes { + return nil, fmt.Errorf("%w: incremental changelog scans do not support delete manifests from snapshot %d", + ErrInvalidOperation, manifest.SnapshotID()) + } + if manifest.ManifestContent() != iceberg.ManifestContentData { + continue + } + if _, ok := changelogSnapshotIDs[manifest.SnapshotID()]; !ok { + continue + } + manifestsByPath[manifest.FilePath()] = manifest + } + } + + paths := make([]string, 0, len(manifestsByPath)) + for path := range manifestsByPath { + paths = append(paths, path) + } + slices.Sort(paths) + manifestList := make([]iceberg.ManifestFile, 0, len(paths)) + for _, path := range paths { + manifestList = append(manifestList, manifestsByPath[path]) + } + + manifestList = slices.DeleteFunc(manifestList, func(manifest iceberg.ManifestFile) bool { + return !manifestHasChangelogEntries(manifest) + }) + manifestList, err = planningScan.filterManifestsWithSchema(manifestList, schema, &acc) + if err != nil { + return nil, err + } + if len(manifestList) == 0 { + return finish(nil) + } + entries, err := planningScan.collectManifestEntriesWithSchemaOptions(ctx, manifestList, schema, false, true) Review Comment: These two bools differ only by the middle word and are positional, so a future caller could transpose them and silently start including DELETED entries. This site is correct, but nothing guards it. At minimum I'd add named-arg comments here (`/*discardDeleted=*/ false, /*discardExisting=*/ true`). If you'd rather, folding them into a small manifestReadOptions struct on openManifestWithOptions removes the footgun entirely. Non-blocking. ########## table/incremental_changelog_scan.go: ########## @@ -0,0 +1,322 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "cmp" + "context" + "fmt" + "maps" + "slices" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" +) + +// ChangelogOperation identifies the change represented by a changelog task. +type ChangelogOperation string + +const ( + ChangelogOperationInsert ChangelogOperation = "insert" + ChangelogOperationDelete ChangelogOperation = "delete" +) + +// ChangelogScanTask plans one data file that contributes a change to the +// table's changelog. The embedded FileScanTask can be passed to Scan.ReadTasks +// to read the file contents. +type ChangelogScanTask struct { + FileScanTask + Operation ChangelogOperation + ChangeOrdinal int + CommitSnapshotID int64 +} + +// IncrementalChangelogScan plans data-file changes between snapshots. It +// emits insert and delete tasks for data-manifest entries and skips replace +// snapshots. PlanFiles returns an error if any in-range snapshot's manifest +// list references a delete manifest, including ones carried forward from +// earlier snapshots. +type IncrementalChangelogScan struct { + scan *Scan + fromSnapshotID *int64 + fromInclusive bool + toSnapshotID *int64 +} + +// NewIncrementalChangelogScan creates an incremental changelog planner. +// Projection and row limits are not applied to returned tasks. Auto planning +// falls back to local planning, while remote planning is not supported. +func (t Table) NewIncrementalChangelogScan(opts ...ScanOption) *IncrementalChangelogScan { + return &IncrementalChangelogScan{scan: t.Scan(opts...)} +} + +// FromSnapshotInclusive includes changes committed by the starting snapshot. +func (s *IncrementalChangelogScan) FromSnapshotInclusive(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.fromSnapshotID = &snapshotID + out.fromInclusive = true + + return &out +} + +// FromSnapshotExclusive starts after the given snapshot. The starting +// snapshot must be a parent ancestor of the ending snapshot when planning. +func (s *IncrementalChangelogScan) FromSnapshotExclusive(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.fromSnapshotID = &snapshotID + out.fromInclusive = false + + return &out +} + +// ToSnapshot sets the inclusive ending snapshot. +func (s *IncrementalChangelogScan) ToSnapshot(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.toSnapshotID = &snapshotID + + return &out +} + +// PlanFiles returns one task for each added or deleted data-file entry and +// emits a ScanReport through the configured reporter on successful planning. +func (s *IncrementalChangelogScan) PlanFiles(ctx context.Context) ([]ChangelogScanTask, error) { + switch s.scan.planningMode { + case ScanPlanningLocal, ScanPlanningAuto: + case ScanPlanningRemote: + return nil, fmt.Errorf("%w: incremental changelog scans do not support remote planning", ErrInvalidOperation) + default: + return nil, fmt.Errorf("%w: unknown scan planning mode %q", iceberg.ErrInvalidArgument, s.scan.planningMode) + } + start := time.Now() + + 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 changelog scan from %d", + iceberg.ErrInvalidArgument, *s.fromSnapshotID) + } + + return nil, nil + } + + planningScan := *s.scan + planningScan.identifier = slices.Clone(s.scan.identifier) + planningScan.selectedFields = slices.Clone(s.scan.selectedFields) + planningScan.options = maps.Clone(s.scan.options) + if s.toSnapshotID != nil { + planningScan.snapshotID = &toSnapshot.SnapshotID + planningScan.asOfTimestamp = nil + } + schema, err := planningScan.effectiveSchema() + if err != nil { + return nil, err + } + residual, err := bindTaskFilter(schema, planningScan.rowFilter, planningScan.caseSensitive) + if err != nil { + return nil, fmt.Errorf("bind incremental changelog scan residual: %w", err) + } + var acc scanMetricsAccumulator + finish := func(tasks []ChangelogScanTask) ([]ChangelogScanTask, error) { + acc.resultDataFiles = int64(len(tasks)) + for _, task := range tasks { + acc.totalFileSize += task.File.FileSizeBytes() + } + planningDuration := time.Since(start) + + if rep := planningScan.Reporter(); !metrics.IsNop(rep) { + projected, _ := planningScan.Projection() + safeReport(ctx, rep, planningScan.buildScanReport(&acc, schema, projected, planningDuration)) + } + + return tasks, nil + } + + snapshotRange, err := incrementalSnapshotsBetween( + s.scan.metadata, s.fromSnapshotID, s.fromInclusive, toSnapshot.SnapshotID) + if err != nil { + return nil, err + } + snapshots, err := changelogSnapshots(snapshotRange) + if err != nil { + return nil, err + } + if len(snapshots) == 0 { + return finish(nil) + } + + changelogSnapshotIDs := make(map[int64]struct{}, len(snapshots)) + snapshotOrdinals := make(map[int64]int, len(snapshots)) + for ordinal, snapshot := range snapshots { + changelogSnapshotIDs[snapshot.SnapshotID] = struct{}{} + snapshotOrdinals[snapshot.SnapshotID] = ordinal + } + + 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 + } + + 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.ManifestContentDeletes { + return nil, fmt.Errorf("%w: incremental changelog scans do not support delete manifests from snapshot %d", + ErrInvalidOperation, manifest.SnapshotID()) + } + if manifest.ManifestContent() != iceberg.ManifestContentData { + continue + } + if _, ok := changelogSnapshotIDs[manifest.SnapshotID()]; !ok { + continue + } + manifestsByPath[manifest.FilePath()] = manifest + } + } + + paths := make([]string, 0, len(manifestsByPath)) + for path := range manifestsByPath { + paths = append(paths, path) + } + slices.Sort(paths) + manifestList := make([]iceberg.ManifestFile, 0, len(paths)) + for _, path := range paths { + manifestList = append(manifestList, manifestsByPath[path]) + } + + manifestList = slices.DeleteFunc(manifestList, func(manifest iceberg.ManifestFile) bool { + return !manifestHasChangelogEntries(manifest) + }) + manifestList, err = planningScan.filterManifestsWithSchema(manifestList, schema, &acc) + if err != nil { + return nil, err + } + if len(manifestList) == 0 { + return finish(nil) + } + entries, err := planningScan.collectManifestEntriesWithSchemaOptions(ctx, manifestList, schema, false, true) + if err != nil { + return nil, err + } + + tasks := make([]ChangelogScanTask, 0, len(entries.dataEntries)) + for _, entry := range entries.dataEntries { + ordinal, ok := snapshotOrdinals[entry.SnapshotID()] + if !ok { + continue + } + + operation, err := changelogOperation(entry.Status()) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + + file := entry.DataFile() + task := ChangelogScanTask{ + FileScanTask: FileScanTask{ + File: file, + Start: 0, + Length: file.FileSizeBytes(), + Residual: residual, + FirstRowID: file.FirstRowID(), + }, + Operation: operation, + ChangeOrdinal: ordinal, + CommitSnapshotID: entry.SnapshotID(), + } + if sequenceNumber := entry.SequenceNum(); sequenceNumber >= 0 { + task.DataSequenceNumber = &sequenceNumber + } + tasks = append(tasks, task) + } + slices.SortFunc(tasks, func(left, right ChangelogScanTask) int { + if ordinal := cmp.Compare(left.ChangeOrdinal, right.ChangeOrdinal); ordinal != 0 { + return ordinal + } + if operation := cmp.Compare(left.Operation, right.Operation); operation != 0 { Review Comment: This ordering only holds because "delete" sorts before "insert" lexically. Deletes-before-inserts within a change ordinal is a real replay contract, but nothing here states it. If the string value of ChangelogOperationDelete ever changed to something that sorts after "insert", this would silently flip and we'd emit inserts before deletes in the same commit window. I'd make it explicit instead of leaning on the string values: a small changelogOperationOrder(op) that returns 0 for delete and 1 for insert, then compare those. A test catches a regression, but the intent should live in the code. wdyt? ########## table/incremental_changelog_scan.go: ########## @@ -0,0 +1,322 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package table + +import ( + "cmp" + "context" + "fmt" + "maps" + "slices" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/metrics" +) + +// ChangelogOperation identifies the change represented by a changelog task. +type ChangelogOperation string + +const ( + ChangelogOperationInsert ChangelogOperation = "insert" + ChangelogOperationDelete ChangelogOperation = "delete" +) + +// ChangelogScanTask plans one data file that contributes a change to the +// table's changelog. The embedded FileScanTask can be passed to Scan.ReadTasks +// to read the file contents. +type ChangelogScanTask struct { + FileScanTask + Operation ChangelogOperation + ChangeOrdinal int + CommitSnapshotID int64 +} + +// IncrementalChangelogScan plans data-file changes between snapshots. It +// emits insert and delete tasks for data-manifest entries and skips replace +// snapshots. PlanFiles returns an error if any in-range snapshot's manifest +// list references a delete manifest, including ones carried forward from +// earlier snapshots. +type IncrementalChangelogScan struct { + scan *Scan + fromSnapshotID *int64 + fromInclusive bool + toSnapshotID *int64 +} + +// NewIncrementalChangelogScan creates an incremental changelog planner. +// Projection and row limits are not applied to returned tasks. Auto planning +// falls back to local planning, while remote planning is not supported. +func (t Table) NewIncrementalChangelogScan(opts ...ScanOption) *IncrementalChangelogScan { + return &IncrementalChangelogScan{scan: t.Scan(opts...)} +} + +// FromSnapshotInclusive includes changes committed by the starting snapshot. +func (s *IncrementalChangelogScan) FromSnapshotInclusive(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.fromSnapshotID = &snapshotID + out.fromInclusive = true + + return &out +} + +// FromSnapshotExclusive starts after the given snapshot. The starting +// snapshot must be a parent ancestor of the ending snapshot when planning. +func (s *IncrementalChangelogScan) FromSnapshotExclusive(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.fromSnapshotID = &snapshotID + out.fromInclusive = false + + return &out +} + +// ToSnapshot sets the inclusive ending snapshot. +func (s *IncrementalChangelogScan) ToSnapshot(snapshotID int64) *IncrementalChangelogScan { + out := *s + out.toSnapshotID = &snapshotID + + return &out +} + +// PlanFiles returns one task for each added or deleted data-file entry and +// emits a ScanReport through the configured reporter on successful planning. +func (s *IncrementalChangelogScan) PlanFiles(ctx context.Context) ([]ChangelogScanTask, error) { + switch s.scan.planningMode { + case ScanPlanningLocal, ScanPlanningAuto: + case ScanPlanningRemote: + return nil, fmt.Errorf("%w: incremental changelog scans do not support remote planning", ErrInvalidOperation) + default: + return nil, fmt.Errorf("%w: unknown scan planning mode %q", iceberg.ErrInvalidArgument, s.scan.planningMode) + } + start := time.Now() + + 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 changelog scan from %d", + iceberg.ErrInvalidArgument, *s.fromSnapshotID) + } + + return nil, nil + } + + planningScan := *s.scan + planningScan.identifier = slices.Clone(s.scan.identifier) + planningScan.selectedFields = slices.Clone(s.scan.selectedFields) + planningScan.options = maps.Clone(s.scan.options) + if s.toSnapshotID != nil { + planningScan.snapshotID = &toSnapshot.SnapshotID + planningScan.asOfTimestamp = nil + } + schema, err := planningScan.effectiveSchema() + if err != nil { + return nil, err + } + residual, err := bindTaskFilter(schema, planningScan.rowFilter, planningScan.caseSensitive) + if err != nil { + return nil, fmt.Errorf("bind incremental changelog scan residual: %w", err) + } + var acc scanMetricsAccumulator + finish := func(tasks []ChangelogScanTask) ([]ChangelogScanTask, error) { + acc.resultDataFiles = int64(len(tasks)) + for _, task := range tasks { + acc.totalFileSize += task.File.FileSizeBytes() + } + planningDuration := time.Since(start) + + if rep := planningScan.Reporter(); !metrics.IsNop(rep) { + projected, _ := planningScan.Projection() + safeReport(ctx, rep, planningScan.buildScanReport(&acc, schema, projected, planningDuration)) + } + + return tasks, nil + } + + snapshotRange, err := incrementalSnapshotsBetween( + s.scan.metadata, s.fromSnapshotID, s.fromInclusive, toSnapshot.SnapshotID) + if err != nil { + return nil, err + } + snapshots, err := changelogSnapshots(snapshotRange) + if err != nil { + return nil, err + } + if len(snapshots) == 0 { + return finish(nil) + } + + changelogSnapshotIDs := make(map[int64]struct{}, len(snapshots)) + snapshotOrdinals := make(map[int64]int, len(snapshots)) + for ordinal, snapshot := range snapshots { + changelogSnapshotIDs[snapshot.SnapshotID] = struct{}{} + snapshotOrdinals[snapshot.SnapshotID] = ordinal + } + + 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 + } + + 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.ManifestContentDeletes { + return nil, fmt.Errorf("%w: incremental changelog scans do not support delete manifests from snapshot %d", + ErrInvalidOperation, manifest.SnapshotID()) + } + if manifest.ManifestContent() != iceberg.ManifestContentData { + continue + } + if _, ok := changelogSnapshotIDs[manifest.SnapshotID()]; !ok { + continue + } + manifestsByPath[manifest.FilePath()] = manifest + } + } + + paths := make([]string, 0, len(manifestsByPath)) + for path := range manifestsByPath { + paths = append(paths, path) + } + slices.Sort(paths) + manifestList := make([]iceberg.ManifestFile, 0, len(paths)) + for _, path := range paths { + manifestList = append(manifestList, manifestsByPath[path]) + } + + manifestList = slices.DeleteFunc(manifestList, func(manifest iceberg.ManifestFile) bool { + return !manifestHasChangelogEntries(manifest) + }) + manifestList, err = planningScan.filterManifestsWithSchema(manifestList, schema, &acc) + if err != nil { + return nil, err + } + if len(manifestList) == 0 { + return finish(nil) + } + entries, err := planningScan.collectManifestEntriesWithSchemaOptions(ctx, manifestList, schema, false, true) + if err != nil { + return nil, err + } + + tasks := make([]ChangelogScanTask, 0, len(entries.dataEntries)) + for _, entry := range entries.dataEntries { + ordinal, ok := snapshotOrdinals[entry.SnapshotID()] + if !ok { + continue + } + + operation, err := changelogOperation(entry.Status()) + if err != nil { + return nil, fmt.Errorf("incremental changelog scan snapshot %d: %w", entry.SnapshotID(), err) + } + + file := entry.DataFile() + task := ChangelogScanTask{ + FileScanTask: FileScanTask{ + File: file, + Start: 0, + Length: file.FileSizeBytes(), + Residual: residual, + FirstRowID: file.FirstRowID(), + }, + Operation: operation, + ChangeOrdinal: ordinal, + CommitSnapshotID: entry.SnapshotID(), + } + if sequenceNumber := entry.SequenceNum(); sequenceNumber >= 0 { + task.DataSequenceNumber = &sequenceNumber + } + tasks = append(tasks, task) + } + slices.SortFunc(tasks, func(left, right ChangelogScanTask) int { + if ordinal := cmp.Compare(left.ChangeOrdinal, right.ChangeOrdinal); ordinal != 0 { + return ordinal + } + if operation := cmp.Compare(left.Operation, right.Operation); operation != 0 { + return operation + } + + return cmp.Compare(left.File.FilePath(), right.File.FilePath()) + }) + + return finish(tasks) +} + +func manifestHasChangelogEntries(manifest iceberg.ManifestFile) bool { + return manifest.AddedDataFiles() != 0 || manifest.DeletedDataFiles() != 0 +} + +func changelogOperation(status iceberg.ManifestEntryStatus) (ChangelogOperation, error) { + switch status { + case iceberg.EntryStatusADDED: + return ChangelogOperationInsert, nil + case iceberg.EntryStatusDELETED: + return ChangelogOperationDelete, nil + default: + return "", fmt.Errorf("%w: unknown manifest entry status %d", ErrInvalidMetadata, status) + } +} + +func (s *IncrementalChangelogScan) toSnapshot() (*Snapshot, error) { + if s.toSnapshotID != nil { + snapshot := s.scan.metadata.SnapshotByID(*s.toSnapshotID) + if snapshot == nil { + return nil, fmt.Errorf("%w: ending snapshot not found: %d", iceberg.ErrInvalidArgument, *s.toSnapshotID) + } + + return snapshot, nil + } + + return s.scan.ResolveSnapshot() +} + +func changelogSnapshots(snapshots []Snapshot) ([]Snapshot, error) { + result := make([]Snapshot, 0, len(snapshots)) + for _, snapshot := range snapshots { + if snapshot.Summary == nil || snapshot.Summary.Operation == "" { + return nil, fmt.Errorf("%w: cannot determine operation for snapshot %d", + ErrMissingOperation, snapshot.SnapshotID) + } + + switch snapshot.Summary.Operation { + case OpReplace: + continue + case OpAppend, OpOverwrite, OpDelete: + result = append(result, snapshot) + default: Review Comment: Java's orderedChangelogSnapshots only skips REPLACE and includes every other operation. Here the explicit default rejects anything outside {append, overwrite, delete}. If a future spec revision adds a new snapshot operation (the way "delete" was once added), we'd error out on tables that Java and PyIceberg still read. Not a blocker since it's correct for today's spec, but I'd lean toward matching Java (default: continue, with a comment), or keep it strict and add a note that this deliberately assumes the current operation set. Either is fine, I'd just want it to be a choice. wdyt? -- 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]
