tanmayrauth commented on code in PR #1883: URL: https://github.com/apache/iceberg-go/pull/1883#discussion_r3865599963
########## table/incremental_changelog_scan.go: ########## @@ -0,0 +1,320 @@ +// 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. Delete manifests are not supported by this scan. +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 in snapshot %d", Review Comment: The message interpolates `snapshot.SnapshotID`, i.e. the snapshot whose manifest list is being iterated — not the snapshot that created the delete manifest. In the carried-forward case (delete manifest from an older out-of-range snapshot, appearing in a later append snapshot's list) this names a pure-append snapshot as the offender, so someone debugging why their append-only range was rejected is pointed at the wrong commit. Either reference `manifest.SnapshotID()` (the origin), or reword to "snapshot %d references a delete manifest" so it's clear it's about the manifest list, not an operation that snapshot performed. ########## table/incremental_changelog_scan_test.go: ########## @@ -0,0 +1,575 @@ +// 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 ( + "bytes" + "context" + "testing" + + "github.com/apache/iceberg-go" + iceio "github.com/apache/iceberg-go/io" + "github.com/apache/iceberg-go/metrics" + "github.com/stretchr/testify/require" +) + +func TestIncrementalChangelogScanPlansAddedAndDeletedEntries(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 4) + + expected := []struct { + path string + operation ChangelogOperation + ordinal int + commitSnapshotID int64 + }{ + {"mem://default/changelog/data-old.parquet", ChangelogOperationInsert, 0, 1}, + {"mem://default/changelog/data-old.parquet", ChangelogOperationDelete, 1, 2}, + {"mem://default/changelog/data-new.parquet", ChangelogOperationInsert, 1, 2}, + {"mem://default/changelog/data-later.parquet", ChangelogOperationInsert, 2, 4}, + } + for i, want := range expected { + require.Equal(t, want.path, tasks[i].File.FilePath()) + require.Equal(t, want.operation, tasks[i].Operation) + require.Equal(t, want.ordinal, tasks[i].ChangeOrdinal) + require.Equal(t, want.commitSnapshotID, tasks[i].CommitSnapshotID) + require.Zero(t, tasks[i].DeleteFiles) + require.Zero(t, tasks[i].EqualityDeleteFiles) + require.Zero(t, tasks[i].DeletionVectorFiles) + require.NotNil(t, tasks[i].DataSequenceNumber) + } +} + +func TestIncrementalChangelogScanSkipsManifestsWithoutChanges(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 4) +} + +func TestOpenManifestWithOptionsCanDiscardExistingEntries(t *testing.T) { + spec := partitionedSpec() + schema := simpleSchema() + _, fs := createTestTransactionWithMemIO(t, spec) + + existingFile := newTestDataFile(t, spec, + "mem://default/changelog/existing.parquet", map[int]any{1000: int32(1)}) + addedFile := newTestDataFile(t, spec, + "mem://default/changelog/added.parquet", map[int]any{1000: int32(2)}) + snapshotID := int64(1) + sequenceNumber := int64(1) + entries := []iceberg.ManifestEntry{ + iceberg.NewManifestEntry(iceberg.EntryStatusEXISTING, &snapshotID, &sequenceNumber, &sequenceNumber, existingFile), + iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapshotID, &sequenceNumber, &sequenceNumber, addedFile), + } + manifestPath := "mem://default/changelog/metadata/mixed-manifest.avro" + var buf bytes.Buffer + manifest, err := iceberg.WriteManifest(manifestPath, &buf, 2, spec, schema, snapshotID, entries) + require.NoError(t, err) + require.NoError(t, fs.WriteFile(manifestPath, buf.Bytes())) + + partitionCalls := 0 + metricsCalls := 0 + got, err := openManifestWithOptions( + fs, + manifest, + func(iceberg.DataFile) (bool, error) { + partitionCalls++ + + return true, nil + }, + func(iceberg.DataFile) (bool, error) { + metricsCalls++ + + return true, nil + }, + false, + true, + ) + require.NoError(t, err) + require.Len(t, got, 1) + require.Equal(t, iceberg.EntryStatusADDED, got[0].Status()) + require.Equal(t, 1, partitionCalls) + require.Equal(t, 1, metricsCalls) +} + +func TestIncrementalChangelogScanHonorsSnapshotBoundaries(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan(). + FromSnapshotExclusive(1). + ToSnapshot(4). + PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 3) + require.Equal(t, "mem://default/changelog/data-old.parquet", tasks[0].File.FilePath()) + require.Equal(t, ChangelogOperationDelete, tasks[0].Operation) + require.Equal(t, 0, tasks[0].ChangeOrdinal) + require.Equal(t, "mem://default/changelog/data-new.parquet", tasks[1].File.FilePath()) + require.Equal(t, 0, tasks[1].ChangeOrdinal) + require.Equal(t, "mem://default/changelog/data-later.parquet", tasks[2].File.FilePath()) + require.Equal(t, 1, tasks[2].ChangeOrdinal) + + tasks, err = tbl.NewIncrementalChangelogScan(). + FromSnapshotInclusive(2). + ToSnapshot(4). + PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 3) + for _, task := range tasks { + require.GreaterOrEqual(t, task.CommitSnapshotID, int64(2)) + } +} + +func TestIncrementalChangelogScanSkipsReplaceSnapshots(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan(). + FromSnapshotInclusive(3). + ToSnapshot(3). + PlanFiles(context.Background()) + require.NoError(t, err) + require.Empty(t, tasks) +} + +func TestIncrementalChangelogScanPreservesChangesAcrossManifestRewrite(t *testing.T) { + tbl := incrementalChangelogManifestRewriteTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan().PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 3) + + require.Equal(t, "mem://default/changelog-rewrite/data-a.parquet", tasks[0].File.FilePath()) + require.Equal(t, ChangelogOperationInsert, tasks[0].Operation) + require.Equal(t, 0, tasks[0].ChangeOrdinal) + require.Equal(t, int64(1), tasks[0].CommitSnapshotID) + + require.Equal(t, "mem://default/changelog-rewrite/data-b.parquet", tasks[1].File.FilePath()) + require.Equal(t, ChangelogOperationInsert, tasks[1].Operation) + require.Equal(t, 1, tasks[1].ChangeOrdinal) + require.Equal(t, int64(2), tasks[1].CommitSnapshotID) + + require.Equal(t, "mem://default/changelog-rewrite/data-c.parquet", tasks[2].File.FilePath()) + require.Equal(t, ChangelogOperationInsert, tasks[2].Operation) + require.Equal(t, 2, tasks[2].ChangeOrdinal) + require.Equal(t, int64(4), tasks[2].CommitSnapshotID) +} + +func TestIncrementalChangelogScanUsesLiveSchemaForImplicitCurrent(t *testing.T) { + tbl := incrementalAppendSchemaEvolutionTable(t) + filter := iceberg.EqualTo(iceberg.Reference("category"), "new") + + normalTasks, err := tbl.Scan(WithRowFilter(filter)).PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, normalTasks, 2) + + incrementalTasks, err := tbl.NewIncrementalChangelogScan(WithRowFilter(filter)).PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, incrementalTasks, 2) +} + +func TestIncrementalChangelogScanUsesSnapshotSchemaForExplicitEnd(t *testing.T) { + tbl := incrementalAppendSchemaEvolutionTable(t) + filter := iceberg.EqualTo(iceberg.Reference("category"), "new") + + scan := tbl.NewIncrementalChangelogScan(WithRowFilter(filter)).ToSnapshot(2) + _, err := scan.PlanFiles(context.Background()) + require.Error(t, err) + require.ErrorContains(t, err, "category") +} + +func TestIncrementalChangelogScanAppliesRowFilters(t *testing.T) { + tbl := incrementalChangelogTestTable(t) + filter := iceberg.EqualTo(iceberg.Reference("id"), int32(2)) + + tasks, err := tbl.NewIncrementalChangelogScan( + WithRowFilter(filter), + ).ToSnapshot(4).PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 1) + require.Equal(t, "mem://default/changelog/data-new.parquet", tasks[0].File.FilePath()) + require.Equal(t, ChangelogOperationInsert, tasks[0].Operation) + require.NotNil(t, tasks[0].Residual) +} + +func TestIncrementalChangelogScanEmitsScanReport(t *testing.T) { + reporter := &metrics.InMemoryReporter{} + tbl := incrementalChangelogTestTable(t) + + tasks, err := tbl.NewIncrementalChangelogScan( + WithSelectedFields("id"), + WithReporter(reporter), + ).PlanFiles(context.Background()) + require.NoError(t, err) + require.Len(t, tasks, 4) + + reports := reporter.Reports() + require.Len(t, reports, 1) + report, ok := reports[0].(metrics.ScanReport) + require.True(t, ok) + require.Equal(t, int64(4), report.SnapshotID) + require.Equal(t, []string{"id"}, report.ProjectedFieldNames) + require.Equal(t, int64(4), report.Metrics.ResultDataFiles.Value) + require.Equal(t, int64(3), report.Metrics.TotalDataManifests.Value) + require.Equal(t, int64(3), report.Metrics.ScannedDataManifests.Value) +} + +func TestIncrementalChangelogScanRejectsDeleteManifests(t *testing.T) { Review Comment: This only covers a delete manifest owned by the single in-range snapshot. It doesn't cover the case that actually depends on the check ordering in incremental_changelog_scan.go:189 — a delete manifest created by an out-of-range snapshot and *carried forward* into an otherwise pure-append range. I confirmed that case currently errors (correct — it matches Java's BaseIncrementalChangelogScan:103-118, where snapshot.deleteManifests() returns carried-forward manifests too), but nothing pins it: a future refactor that moves the `changelogSnapshotIDs[manifest.SnapshotID()]` range filter above the content check would silently start accepting those ranges and diverge from Java, with no test failing. Worth adding a table with snapshots [append S1, MoR-delete S2, append S3] and asserting a scan over (S2, S3] fails with ErrInvalidOperation / "do not support delete manifests". ########## table/incremental_changelog_scan.go: ########## @@ -0,0 +1,320 @@ +// 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. Delete manifests are not supported by this scan. Review Comment: "not supported" reads like delete-file-driven changes are silently omitted, but PlanFiles actually hard-errors whenever an in-range snapshot's manifest list references a delete manifest — including delete manifests carried forward from earlier snapshots. That means a pure-append range on any table that has ever done a MoR delete (and still has live delete files) fails rather than returning the appends. That's the intended parity behavior, but the doc should set the expectation: e.g. "PlanFiles returns an error if any in-range snapshot's manifest list references a delete manifest, including ones carried forward from earlier snapshots." -- 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]
