diff --git a/catalog/rest/scan_planning.go b/catalog/rest/scan_planning.go index 306da4346..484f43747 100644 --- a/catalog/rest/scan_planning.go +++ b/catalog/rest/scan_planning.go @@ -153,10 +153,10 @@ const headerIdempotencyKey = "Idempotency-Key" // count as end-to-end capable. // // SupportsRemoteScanPlanning is the table.ScanPlanner-facing predicate that -// table.Scan's auto mode routes on. It is deliberately gated to false while -// PlanFiles is an unimplemented stub: routing on endpoint capability alone would -// send an auto-mode scan into PlanFiles and surface ErrNotImplemented instead of -// falling back to local planning. It flips on with the PlanFiles phase. +// table.Scan's auto mode routes on. It is still gated to false: PlanFiles now +// plans and decodes end-to-end, but a plan whose data files sit outside the +// metadata prefix can still get the wrong vended credential, and the remote path +// emits no ScanReport. // SupportsPlanTableScan reports whether the server advertised the synchronous // plan endpoint. @@ -178,12 +178,12 @@ func (r *Catalog) SupportsFullRemoteScanPlanning() bool { // SupportsRemoteScanPlanning reports whether this catalog can complete a remote // plan end-to-end. table.Scan's auto mode routes on it, calling PlanFiles when it -// is true, so it must stay false until PlanFiles is implemented — otherwise an -// auto-mode scan against a server advertising all four endpoints would fail with -// ErrNotImplemented instead of falling back to local planning. +// is true. Callers that want to drive remote planning explicitly today can pass +// table.ScanPlanningRemote, which bypasses this predicate. // -// TODO(#1178): return SupportsFullRemoteScanPlanning() once PlanFiles is wired -// end-to-end. Until then, callers probing endpoint capability should use +// TODO(#1178): return SupportsFullRemoteScanPlanning() once plan-scoped +// credentials are prefix-routed and the remote path emits a ScanReport. Until +// then, callers probing endpoint capability should use // SupportsFullRemoteScanPlanning / SupportsPlanTableScan directly. func (r *Catalog) SupportsRemoteScanPlanning() bool { return false @@ -194,11 +194,9 @@ func (r *Catalog) SupportsRemoteScanPlanning() bool { // submitted plan to completion, and expands any plan-task handles into their // tasks before returning. // -// Decoding the returned tasks into table.FileScanTask is the one piece still -// stubbed: RESTFileScanTask/RESTDeleteFile are empty pending the scan-task -// decoder phase, so a plan that yields any task surfaces ErrNotImplemented (see -// remoteScanTasks). SupportsRemoteScanPlanning therefore stays false until that -// lands, so auto-mode scans keep planning locally rather than routing here. +// Each envelope — the inline plan response and every fetchScanTasks response — +// is decoded on its own: delete-file-references index into that envelope's own +// delete-files array. func (r *Catalog) PlanFiles(ctx context.Context, req table.ScanPlanningRequest) (table.ScanPlanningResult, error) { wire, err := planTableScanRequestFrom(req) if err != nil { @@ -230,12 +228,11 @@ func (r *Catalog) PlanFiles(ctx context.Context, req table.ScanPlanningRequest) "%w: unexpected plan status %q from planTableScan", ErrRESTError, resp.Status) } - files, deletes, err := r.collectScanTasks(ctx, req.Identifier, completed.ScanTasks) - if err != nil { - return table.ScanPlanningResult{}, err - } - - tasks, err := remoteScanTasks(files, deletes) + tasks, err := r.collectScanTasks(ctx, req.Identifier, completed.ScanTasks, scanTaskDecoder{ + metadata: req.Metadata, + schema: planningSchema(req), + fallbackResidual: req.RowFilter, + }) if err != nil { return table.ScanPlanningResult{}, err } @@ -259,14 +256,49 @@ func (r *Catalog) planIOBaseProps(meta table.ScanPlanningMetadata) iceberg.Prope return props } -// collectScanTasks expands plan-task handles into their tasks, walking the -// fanout: a fetchScanTasks response can itself return more plan-tasks. It -// accumulates the file-scan-tasks and delete-files reachable from the initial -// set. A handle is fetched at most once; a server that re-issues one would -// otherwise loop forever. -func (r *Catalog) collectScanTasks(ctx context.Context, ident table.Identifier, tasks ScanTasks) ([]RESTFileScanTask, []RESTDeleteFile, error) { - files := append([]RESTFileScanTask(nil), tasks.FileScanTasks...) - deletes := append([]RESTDeleteFile(nil), tasks.DeleteFiles...) +// scanTaskDecoder carries the per-plan context DecodeScanTasks needs. Built +// once per plan, applied to each envelope separately: merging envelopes before +// decoding would silently repoint every delete-file-reference past the first. +type scanTaskDecoder struct { + metadata table.ScanPlanningMetadata + schema *iceberg.Schema + fallbackResidual iceberg.BooleanExpression +} + +func (d scanTaskDecoder) decode(wire ScanTasks) ([]table.FileScanTask, error) { + tasks, err := DecodeScanTasks(wire, d.metadata, d.schema, d.fallbackResidual) + if err != nil { + return nil, err + } + + for i := range tasks { + suppressDeletesCoveredByDV(&tasks[i]) + } + + return tasks, nil +} + +// suppressDeletesCoveredByDV drops positional-delete files from a task that also +// carries a deletion vector. Per the v3 spec a DV encodes every prior positional +// delete for its data file, so applying both is wasted I/O. (*Scan).planFilesLocal +// applies the same rule; a remote plan must not diverge from it. +func suppressDeletesCoveredByDV(task *table.FileScanTask) { + if len(task.DeletionVectorFiles) > 0 { + task.DeleteFiles = nil + } +} + +// collectScanTasks decodes the initial envelope, then expands plan-task handles +// into theirs, walking the fanout: a fetchScanTasks response can itself return +// more plan-tasks. A handle is fetched at most once; a server that re-issues one +// would otherwise loop forever. +func (r *Catalog) collectScanTasks( + ctx context.Context, ident table.Identifier, tasks ScanTasks, dec scanTaskDecoder, +) ([]table.FileScanTask, error) { + out, err := dec.decode(tasks) + if err != nil { + return nil, err + } queue := append([]string(nil), tasks.PlanTasks...) seen := make(map[string]bool, len(queue)) @@ -280,26 +312,18 @@ func (r *Catalog) collectScanTasks(ctx context.Context, ident table.Identifier, resp, err := r.FetchScanTasks(ctx, ident, FetchScanTasksRequest{PlanTask: handle}) if err != nil { - return nil, nil, err + return nil, err } - files = append(files, resp.FileScanTasks...) - deletes = append(deletes, resp.DeleteFiles...) - queue = append(queue, resp.PlanTasks...) - } - - return files, deletes, nil -} -// remoteScanTasks decodes the server's task payload into domain FileScanTasks. -// Blocked for now: RESTFileScanTask/RESTDeleteFile are empty pending the -// scan-task decoder phase, so an empty plan decodes to no tasks while any actual -// task surfaces ErrNotImplemented. The decoder phase replaces this body. -func remoteScanTasks(files []RESTFileScanTask, deletes []RESTDeleteFile) ([]table.FileScanTask, error) { - if len(files) == 0 && len(deletes) == 0 { - return nil, nil + decoded, err := dec.decode(resp.ScanTasks) + if err != nil { + return nil, err + } + out = append(out, decoded...) + queue = append(queue, resp.PlanTasks...) } - return nil, fmt.Errorf("%w: decoding remote scan tasks", iceberg.ErrNotImplemented) + return out, nil } // planIOFromCredentials wraps a plan's vended storage credentials in a lazy @@ -366,13 +390,28 @@ func planTableScanRequestFrom(req table.ScanPlanningRequest) (PlanTableScanReque return out, nil } +// planningSchema resolves the schema a plan binds against: the one the scan +// itself resolved, else the table's current schema. table.Scan always supplies +// it; a hand-built request need not. +func planningSchema(req table.ScanPlanningRequest) *iceberg.Schema { + if req.Schema != nil { + return req.Schema + } + if req.Metadata == nil { + return nil + } + + return req.Metadata.CurrentSchema() +} + // marshalScanFilter binds the row filter before encoding it: a bound term gives // the encoder the field type it needs for ambiguous literals (a bare timestamp // can't pick timestamp vs timestamptz and won't serialize unbound), and binding // errors loudly on an expression type it doesn't know. func marshalScanFilter(req table.ScanPlanningRequest) ([]byte, error) { - if req.Metadata == nil { - return nil, fmt.Errorf("%w: cannot encode scan filter without table metadata", iceberg.ErrInvalidArgument) + schema := planningSchema(req) + if schema == nil { + return nil, fmt.Errorf("%w: cannot encode scan filter without a schema", iceberg.ErrInvalidArgument) } caseSensitive := true @@ -380,8 +419,7 @@ func marshalScanFilter(req table.ScanPlanningRequest) ([]byte, error) { caseSensitive = *req.CaseSensitive } - // Snapshot-schema selection (UseSnapshotSchema) is OQ4, deferred; bind current. - bound, err := iceberg.BindExpr(req.Metadata.CurrentSchema(), req.RowFilter, caseSensitive) + bound, err := iceberg.BindExpr(schema, req.RowFilter, caseSensitive) if err != nil { return nil, fmt.Errorf("%w: binding scan filter: %s", iceberg.ErrInvalidArgument, err) } diff --git a/catalog/rest/scan_planning_test.go b/catalog/rest/scan_planning_test.go index 357256253..94589e1e5 100644 --- a/catalog/rest/scan_planning_test.go +++ b/catalog/rest/scan_planning_test.go @@ -933,17 +933,42 @@ func scanFilterSchema() *iceberg.Schema { ) } -// scanTestMetadata is a ScanPlanningMetadata carrying only a schema — all filter -// encoding needs. +// scanTestMetadata is a ScanPlanningMetadata carrying a schema and the +// unpartitioned spec — what filter encoding and task decoding need. type scanTestMetadata struct{ schema *iceberg.Schema } -func (m scanTestMetadata) CurrentSchema() *iceberg.Schema { return m.schema } -func (m scanTestMetadata) Schemas() []*iceberg.Schema { return []*iceberg.Schema{m.schema} } -func (m scanTestMetadata) PartitionSpec() iceberg.PartitionSpec { return iceberg.PartitionSpec{} } -func (m scanTestMetadata) PartitionSpecByID(int) *iceberg.PartitionSpec { return nil } -func (m scanTestMetadata) CurrentSnapshot() *table.Snapshot { return nil } -func (m scanTestMetadata) SnapshotByID(int64) *table.Snapshot { return nil } -func (m scanTestMetadata) Properties() iceberg.Properties { return nil } +func (m scanTestMetadata) CurrentSchema() *iceberg.Schema { return m.schema } +func (m scanTestMetadata) Schemas() []*iceberg.Schema { return []*iceberg.Schema{m.schema} } +func (m scanTestMetadata) PartitionSpec() iceberg.PartitionSpec { return *iceberg.UnpartitionedSpec } +func (m scanTestMetadata) PartitionSpecByID(id int) *iceberg.PartitionSpec { + if id != 0 { + return nil + } + + return iceberg.UnpartitionedSpec +} +func (m scanTestMetadata) CurrentSnapshot() *table.Snapshot { return nil } +func (m scanTestMetadata) SnapshotByID(int64) *table.Snapshot { return nil } +func (m scanTestMetadata) Properties() iceberg.Properties { return nil } + +// scanTaskJSON is one unpartitioned data-file scan task, optionally referencing +// delete files by index into its own envelope's delete-files array. +func scanTaskJSON(path, refsJSON string) string { + return `{"data-file":{"spec-id":0,"partition":[],"content":"data","file-path":"` + path + + `","file-format":"parquet","file-size-in-bytes":4096,"record-count":100}` + refsJSON + `}` +} + +// deleteFileJSON is one delete file: parquet for a positional delete, puffin for +// a deletion vector. +func deleteFileJSON(path, format string) string { + extra := "" + if format == "puffin" { + extra = `,"content-offset":0,"content-size-in-bytes":16` + } + + return `{"spec-id":0,"partition":[],"content":"position-deletes","file-path":"` + path + + `","file-format":"` + format + `","file-size-in-bytes":512,"record-count":5` + extra + `}` +} // planFilesReq is a minimal planner request naming the test table. func planFilesReq() table.ScanPlanningRequest { @@ -997,21 +1022,118 @@ func TestPlanFilesEncodesFilter(t *testing.T) { require.NoError(t, err) } -// TestPlanFilesTasksNotYetDecodable documents the one stubbed boundary: a plan -// that returns actual file-scan-tasks surfaces ErrNotImplemented until the -// scan-task decoder phase fills in RESTFileScanTask. -func TestPlanFilesTasksNotYetDecodable(t *testing.T) { +// TestPlanFilesDecodesTasks covers the ordinary case: an inline-completed plan +// carrying a data file comes back as a FileScanTask, with the scan's own filter +// standing in as the residual the server omitted. +func TestPlanFilesDecodesTasks(t *testing.T) { t.Parallel() cat := newScanPlanningTestCatalog(t, []endpoint{endpointPlanTableScan}, func(mux *http.ServeMux) { mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan", func(w http.ResponseWriter, req *http.Request) { - _, err := w.Write([]byte(`{"status":"completed","plan-id":"plan-1","file-scan-tasks":[{}]}`)) + _, err := w.Write([]byte(`{"status":"completed","plan-id":"plan-1","file-scan-tasks":[` + + scanTaskJSON("s3://bucket/tbl/data.parquet", "") + `]}`)) require.NoError(t, err) }) }) - _, err := cat.PlanFiles(context.Background(), planFilesReq()) - require.ErrorIs(t, err, iceberg.ErrNotImplemented) + planReq := planFilesReq() + planReq.RowFilter = iceberg.EqualTo(iceberg.Reference("i"), int32(25)) + + result, err := cat.PlanFiles(context.Background(), planReq) + require.NoError(t, err) + require.Len(t, result.Tasks, 1) + assert.Equal(t, "s3://bucket/tbl/data.parquet", result.Tasks[0].File.FilePath()) + assert.Equal(t, int64(4096), result.Tasks[0].Length) + assert.True(t, result.Tasks[0].Residual.Equals(planReq.RowFilter)) +} + +// TestPlanFilesDecodesEachEnvelopeSeparately is the regression guard for +// envelope-local delete references. Both envelopes here say +// delete-file-references [0], each meaning their own delete file. Accumulating +// the raw payloads before decoding would repoint the second task at the first +// envelope's delete — the wrong deletes applied, with no error anywhere. +func TestPlanFilesDecodesEachEnvelopeSeparately(t *testing.T) { + t.Parallel() + + cat := newScanPlanningTestCatalog(t, []endpoint{endpointPlanTableScan, endpointFetchScanTasks}, func(mux *http.ServeMux) { + mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan", func(w http.ResponseWriter, req *http.Request) { + _, err := w.Write([]byte(`{"status":"completed","plan-id":"plan-1","plan-tasks":["h1"],"file-scan-tasks":[` + + scanTaskJSON("s3://bucket/tbl/inline.parquet", `,"delete-file-references":[0]`) + + `],"delete-files":[` + deleteFileJSON("s3://bucket/tbl/inline-delete.parquet", "parquet") + `]}`)) + require.NoError(t, err) + }) + mux.HandleFunc("/v1/namespaces/db/tables/tbl/tasks", func(w http.ResponseWriter, req *http.Request) { + _, err := w.Write([]byte(`{"file-scan-tasks":[` + + scanTaskJSON("s3://bucket/tbl/fanout.parquet", `,"delete-file-references":[0]`) + + `],"delete-files":[` + deleteFileJSON("s3://bucket/tbl/fanout-delete.parquet", "parquet") + `]}`)) + require.NoError(t, err) + }) + }) + + result, err := cat.PlanFiles(context.Background(), planFilesReq()) + require.NoError(t, err) + require.Len(t, result.Tasks, 2) + + // Queue order: the inline envelope first, then each plan-task it fans out to. + assert.Equal(t, "s3://bucket/tbl/inline.parquet", result.Tasks[0].File.FilePath()) + require.Len(t, result.Tasks[0].DeleteFiles, 1) + assert.Equal(t, "s3://bucket/tbl/inline-delete.parquet", result.Tasks[0].DeleteFiles[0].FilePath()) + + assert.Equal(t, "s3://bucket/tbl/fanout.parquet", result.Tasks[1].File.FilePath()) + require.Len(t, result.Tasks[1].DeleteFiles, 1) + assert.Equal(t, "s3://bucket/tbl/fanout-delete.parquet", result.Tasks[1].DeleteFiles[0].FilePath()) +} + +// TestPlanFilesSuppressesPositionalDeletesCoveredByDV checks that a server +// referencing both a deletion vector and a positional delete for one data file +// yields only the DV, matching (*Scan).planFilesLocal. +func TestPlanFilesSuppressesPositionalDeletesCoveredByDV(t *testing.T) { + t.Parallel() + + cat := newScanPlanningTestCatalog(t, []endpoint{endpointPlanTableScan}, func(mux *http.ServeMux) { + mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan", func(w http.ResponseWriter, req *http.Request) { + _, err := w.Write([]byte(`{"status":"completed","plan-id":"plan-1","file-scan-tasks":[` + + scanTaskJSON("s3://bucket/tbl/data.parquet", `,"delete-file-references":[0,1]`) + + `],"delete-files":[` + + deleteFileJSON("s3://bucket/tbl/deletes.parquet", "parquet") + `,` + + deleteFileJSON("s3://bucket/tbl/deletes.puffin", "puffin") + `]}`)) + require.NoError(t, err) + }) + }) + + result, err := cat.PlanFiles(context.Background(), planFilesReq()) + require.NoError(t, err) + require.Len(t, result.Tasks, 1) + assert.Empty(t, result.Tasks[0].DeleteFiles, "a DV supersedes positional deletes for its data file") + require.Len(t, result.Tasks[0].DeletionVectorFiles, 1) + assert.Equal(t, "s3://bucket/tbl/deletes.puffin", result.Tasks[0].DeletionVectorFiles[0].FilePath()) +} + +// TestPlanFilesBindsFilterToRequestSchema checks the filter binds against the +// schema the scan resolved, not the table's current one — the snapshot-pinned +// case, where the two differ. +func TestPlanFilesBindsFilterToRequestSchema(t *testing.T) { + t.Parallel() + + cat := newScanPlanningTestCatalog(t, []endpoint{endpointPlanTableScan}, func(mux *http.ServeMux) { + mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan", func(w http.ResponseWriter, req *http.Request) { + var got PlanTableScanRequest + require.NoError(t, json.NewDecoder(req.Body).Decode(&got)) + assert.JSONEq(t, `{"type":"eq","term":"renamed","value":25}`, string(got.Filter)) + + _, err := w.Write([]byte(`{"status":"completed","plan-id":"plan-1"}`)) + require.NoError(t, err) + }) + }) + + planReq := planFilesReq() + planReq.Schema = iceberg.NewSchema(1, + iceberg.NestedField{ID: 1, Name: "renamed", Type: iceberg.PrimitiveTypes.Int32}, + ) + planReq.RowFilter = iceberg.EqualTo(iceberg.Reference("renamed"), int32(25)) + + _, err := cat.PlanFiles(context.Background(), planReq) + require.NoError(t, err) } // TestPlanFilesPollsSubmittedPlan covers the async arm: a submitted plan is diff --git a/table/scan_metrics_test.go b/table/scan_metrics_test.go index 9a566fee7..d48135b69 100644 --- a/table/scan_metrics_test.go +++ b/table/scan_metrics_test.go @@ -30,7 +30,9 @@ import ( "github.com/stretchr/testify/require" ) -func metricsTestMetadata(t *testing.T) Metadata { +// scanTestMetadata is a minimal two-column unpartitioned table, shared by the +// scan tests in this package. +func scanTestMetadata(t *testing.T) Metadata { t.Helper() schema := iceberg.NewSchema(7, iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, @@ -44,7 +46,7 @@ func metricsTestMetadata(t *testing.T) Metadata { } func TestBuildScanReport(t *testing.T) { - meta := metricsTestMetadata(t) + meta := scanTestMetadata(t) scan := &Scan{ metadata: meta, identifier: Identifier{"db", "tbl"}, @@ -103,7 +105,7 @@ func TestBuildScanReport(t *testing.T) { } func TestBuildScanReportIncludesEnvironmentContext(t *testing.T) { - meta := metricsTestMetadata(t) + meta := scanTestMetadata(t) key := iceberg.EnvironmentEngineNameKey preserveEnvironmentProperties(t, key) @@ -130,7 +132,7 @@ func TestBuildScanReportIncludesEnvironmentContext(t *testing.T) { } func TestProjectedFieldsSelectedSubset(t *testing.T) { - meta := metricsTestMetadata(t) + meta := scanTestMetadata(t) scan := &Scan{metadata: meta, selectedFields: []string{"data"}, caseSensitive: true} projected, err := scan.Projection() @@ -238,7 +240,7 @@ func TestApplyResultDeleteMetricsDVsShareOnePuffin(t *testing.T) { } func TestBuildScanReportSetsFilter(t *testing.T) { - meta := metricsTestMetadata(t) + meta := scanTestMetadata(t) scan := &Scan{ metadata: meta, identifier: Identifier{"db", "tbl"}, @@ -254,7 +256,7 @@ func TestBuildScanReportSetsFilter(t *testing.T) { func TestBuildScanReportSanitizesFilter(t *testing.T) { // Predicate literals can be user data, so the emitted filter must be // sanitized (Java's ExpressionUtil.sanitize) before it reaches a reporter. - meta := metricsTestMetadata(t) + meta := scanTestMetadata(t) scan := &Scan{ metadata: meta, identifier: Identifier{"db", "tbl"}, @@ -274,7 +276,7 @@ func TestPlanFilesNoSnapshotEmitsNoReport(t *testing.T) { // A table with no snapshot plans zero files and has no real snapshot id, so // no ScanReport is emitted (matching Java, which skips the report entirely). rep := &metrics.InMemoryReporter{} - meta := metricsTestMetadata(t) + meta := scanTestMetadata(t) tbl := New(Identifier{"db", "tbl"}, meta, "metadata.json", nil, nil, WithMetricsReporter(rep)) @@ -353,7 +355,7 @@ func TestPlanFilesEmitsReportForRealSnapshot(t *testing.T) { func TestPlanFilesNopReporterDoesNotPanic(t *testing.T) { // Default (no reporter configured) must plan without panicking. - tbl := New(Identifier{"db", "tbl"}, metricsTestMetadata(t), "metadata.json", nil, nil) + tbl := New(Identifier{"db", "tbl"}, scanTestMetadata(t), "metadata.json", nil, nil) assert.NotPanics(t, func() { _, _ = tbl.Scan().PlanFiles(context.Background()) }) @@ -364,6 +366,7 @@ func TestPlanFilesRemoteDoesNotEmitScanReport(t *testing.T) { // emit a local ScanReport (which would carry only zeroed counters). rep := &metrics.InMemoryReporter{} scan := &Scan{ + metadata: scanTestMetadata(t), planner: &fakeScanPlanner{result: ScanPlanningResult{Tasks: []FileScanTask{{}}}, supports: true}, planningMode: ScanPlanningRemote, reporter: rep, diff --git a/table/scan_planning.go b/table/scan_planning.go index bff9cbe04..e8aadf8d0 100644 --- a/table/scan_planning.go +++ b/table/scan_planning.go @@ -84,17 +84,21 @@ var _ ScanPlanningMetadata = (Metadata)(nil) // ScanPlanningRequest is the input a Scan hands to a ScanPlanner. It carries // the resolved scan state a planner needs without depending on catalog/rest. // -// Open question (epic OQ4): when the table has evolved, UseSnapshotSchema must -// pin which schema binds a returned residual and the partition decode: the -// snapshot's schema (via schema-id), kept separate from each file's partition -// spec-id. Incremental scans (start/end snapshot) are deferred to a later -// phase; point-in-time SnapshotID lands first. +// When the table has evolved, Schema pins which schema binds the row filter and +// a returned residual, kept separate from each file's partition spec-id; +// UseSnapshotSchema tells the server the same thing. Incremental scans +// (start/end snapshot) are deferred to a later phase; point-in-time SnapshotID +// lands first. type ScanPlanningRequest struct { Identifier Identifier // Metadata is the narrowed planner view of table metadata (see // ScanPlanningMetadata); MetadataLocation is kept separate. Metadata ScanPlanningMetadata MetadataLocation string + // Schema is the schema the scan is bound to: the table's current schema for + // a live scan, the snapshot's schema for a pinned or as-of one. Nil falls + // back to Metadata.CurrentSchema. + Schema *iceberg.Schema SnapshotID *int64 SelectedFields []string RowFilter iceberg.BooleanExpression @@ -105,7 +109,8 @@ type ScanPlanningRequest struct { // Nil means use the scan default. CaseSensitive *bool // UseSnapshotSchema is a pointer to distinguish the spec default from an - // explicit false when the scanner-delegation phase binds it to table config. + // explicit false. It is the wire-side counterpart of Schema: true when Schema + // came from the scan's snapshot rather than the table's current schema. UseSnapshotSchema *bool } diff --git a/table/scan_planning_test.go b/table/scan_planning_test.go index 4f81f2f97..80776c53a 100644 --- a/table/scan_planning_test.go +++ b/table/scan_planning_test.go @@ -31,7 +31,7 @@ import ( func TestScanPlanningRemoteRequiresPlanner(t *testing.T) { t.Parallel() - scan := &Scan{planningMode: ScanPlanningRemote} + scan := &Scan{metadata: scanTestMetadata(t), planningMode: ScanPlanningRemote} _, err := scan.PlanFiles(context.Background()) require.ErrorIs(t, err, ErrInvalidOperation) @@ -46,6 +46,7 @@ func TestScanPlanningRemoteStoresPlanIO(t *testing.T) { supports: true, } scan := &Scan{ + metadata: scanTestMetadata(t), planner: planner, planningMode: ScanPlanningRemote, } @@ -65,6 +66,7 @@ func TestScanPlanningRemoteClosesPreviousPlanIO(t *testing.T) { results: []ScanPlanningResult{{IO: first}, {IO: second}}, } scan := &Scan{ + metadata: scanTestMetadata(t), planner: planner, planningMode: ScanPlanningRemote, } @@ -99,6 +101,7 @@ func TestRefinedScanRetainsPlanIOOwnership(t *testing.T) { results: []ScanPlanningResult{{IO: first}, {IO: second}}, } scan := &Scan{ + metadata: scanTestMetadata(t), planner: planner, planningMode: ScanPlanningRemote, } @@ -134,6 +137,7 @@ func TestScanPlanningRemoteFailurePreservesPreviousPlanIO(t *testing.T) { errors: []error{nil, want}, } scan := &Scan{ + metadata: scanTestMetadata(t), planner: planner, planningMode: ScanPlanningRemote, } @@ -154,7 +158,7 @@ func TestScanPlanningRemoteKeepsSamePlanIO(t *testing.T) { planner := &sequenceScanPlanner{ results: []ScanPlanningResult{{IO: pio}, {IO: pio}}, } - scan := &Scan{planner: planner, planningMode: ScanPlanningRemote} + scan := &Scan{metadata: scanTestMetadata(t), planner: planner, planningMode: ScanPlanningRemote} _, err := scan.PlanFiles(context.Background()) require.NoError(t, err) @@ -173,7 +177,7 @@ func TestScanPlanningRemoteRejectsNonComparablePlanIO(t *testing.T) { planner := &sequenceScanPlanner{ results: []ScanPlanningResult{{IO: pio}, {IO: slicePlanIO{1, 2, 3}}}, } - scan := &Scan{planner: planner, planningMode: ScanPlanningRemote} + scan := &Scan{metadata: scanTestMetadata(t), planner: planner, planningMode: ScanPlanningRemote} _, err := scan.PlanFiles(context.Background()) require.NoError(t, err) @@ -283,6 +287,7 @@ func TestScanPlanningRemoteRejectsIncapablePlanner(t *testing.T) { t.Parallel() scan := &Scan{ + metadata: scanTestMetadata(t), planner: &fakeScanPlanner{supports: false}, planningMode: ScanPlanningRemote, } @@ -296,6 +301,7 @@ func TestScanPlanningRemotePropagatesPlannerError(t *testing.T) { want := errors.New("planner boom") scan := &Scan{ + metadata: scanTestMetadata(t), planner: &fakeScanPlanner{supports: true, err: want}, planningMode: ScanPlanningRemote, } @@ -309,6 +315,7 @@ func TestScanPlanningRemoteRejectsConflictingSnapshotSelectors(t *testing.T) { planner := &fakeScanPlanner{supports: true} scan := &Scan{ + metadata: scanTestMetadata(t), planner: planner, planningMode: ScanPlanningRemote, } @@ -324,6 +331,7 @@ func TestScanPlanningAutoUsesCapablePlanner(t *testing.T) { t.Parallel() scan := &Scan{ + metadata: scanTestMetadata(t), planner: &fakeScanPlanner{result: ScanPlanningResult{Tasks: []FileScanTask{{}}}, supports: true}, planningMode: ScanPlanningAuto, } @@ -337,6 +345,7 @@ func TestScanPlanningPassesIdentifierCopy(t *testing.T) { t.Parallel() scan := &Scan{ + metadata: scanTestMetadata(t), planner: &fakeScanPlanner{ result: ScanPlanningResult{Tasks: []FileScanTask{{}}}, supports: true, @@ -351,7 +360,7 @@ func TestScanPlanningPassesIdentifierCopy(t *testing.T) { assert.Len(t, tasks, 1) planReq := scan.planner.(*fakeScanPlanner) - planReq.receivedIdentifier[0] = "corrupt" + planReq.receivedRequest.Identifier[0] = "corrupt" assert.Equal(t, Identifier{"db", "scan-copy-test"}, scan.identifier) } @@ -379,6 +388,133 @@ func TestTransactionScanRejectsConflictingSnapshotSelectors(t *testing.T) { assert.Nil(t, scan) } +func TestScanPlanningRemoteSendsCurrentSchema(t *testing.T) { + t.Parallel() + + meta := scanTestMetadata(t) + planner := &fakeScanPlanner{supports: true} + scan := &Scan{ + metadata: meta, + planner: planner, + planningMode: ScanPlanningRemote, + } + + _, err := scan.PlanFiles(context.Background()) + require.NoError(t, err) + + got := planner.receivedRequest + require.NotNil(t, got.Schema) + assert.Equal(t, meta.CurrentSchema().ID, got.Schema.ID) + require.NotNil(t, got.UseSnapshotSchema) + assert.False(t, *got.UseSnapshotSchema) +} + +func TestScanPlanningRemoteSendsSnapshotSchema(t *testing.T) { + t.Parallel() + + meta := scanTestMetadata(t) + old := iceberg.NewSchema(9, + iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true}, + ) + snapshotID := int64(42) + schemaID := 9 + pinned := &planningSnapshotMetadata{ + Metadata: meta, + extra: old, + snapshot: &Snapshot{SnapshotID: snapshotID, SchemaID: &schemaID}, + } + + planner := &fakeScanPlanner{supports: true} + scan := &Scan{ + metadata: pinned, + planner: planner, + planningMode: ScanPlanningRemote, + snapshotID: &snapshotID, + } + + _, err := scan.PlanFiles(context.Background()) + require.NoError(t, err) + + got := planner.receivedRequest + require.NotNil(t, got.Schema) + assert.Equal(t, schemaID, got.Schema.ID) + require.NotNil(t, got.UseSnapshotSchema) + assert.True(t, *got.UseSnapshotSchema) +} + +// planningSnapshotMetadata pins one snapshot to an older schema, so a scan can +// resolve a schema other than the current one. +type planningSnapshotMetadata struct { + Metadata + extra *iceberg.Schema + snapshot *Snapshot +} + +func (m *planningSnapshotMetadata) Schemas() []*iceberg.Schema { + return append(m.Metadata.Schemas(), m.extra) +} + +func (m *planningSnapshotMetadata) SnapshotByID(id int64) *Snapshot { + if m.snapshot != nil && m.snapshot.SnapshotID == id { + return m.snapshot + } + + return m.Metadata.SnapshotByID(id) +} + +func TestScanPlanningRemoteRejectsLineageSequenceNumber(t *testing.T) { + t.Parallel() + + // The REST FileScanTask schema carries no data sequence number, so a remote + // plan cannot supply _last_updated_sequence_number. Rejecting beats handing + // back nulls where a local scan returns values. + + for name, opt := range map[string]ScanOption{ + "explicit column": func(scan *Scan) { + scan.selectedFields = []string{"id", iceberg.LastUpdatedSequenceNumberColumnName} + }, + "row lineage option": WithRowLineage(), + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + + planner := &fakeScanPlanner{supports: true} + scan := &Scan{ + metadata: scanTestMetadata(t), + planner: planner, + planningMode: ScanPlanningRemote, + caseSensitive: true, + } + opt(scan) + + _, err := scan.PlanFiles(context.Background()) + require.ErrorIs(t, err, ErrInvalidOperation) + assert.Contains(t, err.Error(), iceberg.LastUpdatedSequenceNumberColumnName) + assert.False(t, planner.called, "must fail before reaching the planner") + }) + } +} + +func TestScanPlanningRemoteAllowsRowID(t *testing.T) { + t.Parallel() + + // _row_id is unaffected: first_row_id rides on the data file, so it survives + // the wire. + + planner := &fakeScanPlanner{supports: true} + scan := &Scan{ + metadata: scanTestMetadata(t), + planner: planner, + planningMode: ScanPlanningRemote, + caseSensitive: true, + selectedFields: []string{"id", iceberg.RowIDColumnName}, + } + + _, err := scan.PlanFiles(context.Background()) + require.NoError(t, err) + assert.True(t, planner.called) +} + func TestScanPlanningUnknownModeErrors(t *testing.T) { t.Parallel() @@ -394,14 +530,14 @@ type fakeScanPlanner struct { err error called bool // captured after PlanFiles receives it - receivedIdentifier Identifier + receivedRequest ScanPlanningRequest } func (f *fakeScanPlanner) SupportsRemoteScanPlanning() bool { return f.supports } func (f *fakeScanPlanner) PlanFiles(_ context.Context, req ScanPlanningRequest) (ScanPlanningResult, error) { f.called = true - f.receivedIdentifier = req.Identifier + f.receivedRequest = req return f.result, f.err } diff --git a/table/scanner.go b/table/scanner.go index 9201da281..625c247b5 100644 --- a/table/scanner.go +++ b/table/scanner.go @@ -970,15 +970,26 @@ func (scan *Scan) planFilesRemote(ctx context.Context) ([]FileScanTask, error) { return nil, fmt.Errorf("%w: remote scan planning is unavailable", ErrInvalidOperation) } + if err := scan.rejectRemoteRowLineage(); err != nil { + return nil, err + } + + schema, useSnapshotSchema, err := scan.remotePlanSchema() + if err != nil { + return nil, err + } + caseSensitive := scan.caseSensitive result, err := scan.planner.PlanFiles(ctx, ScanPlanningRequest{ - Identifier: slices.Clone(scan.identifier), - Metadata: scan.metadata, - MetadataLocation: scan.metadataLocation, - SnapshotID: scan.snapshotID, - SelectedFields: scan.selectedFields, - RowFilter: scan.rowFilter, - CaseSensitive: &caseSensitive, + Identifier: slices.Clone(scan.identifier), + Metadata: scan.metadata, + MetadataLocation: scan.metadataLocation, + Schema: schema, + SnapshotID: scan.snapshotID, + SelectedFields: scan.selectedFields, + RowFilter: scan.rowFilter, + CaseSensitive: &caseSensitive, + UseSnapshotSchema: &useSnapshotSchema, }) if err != nil { return nil, err @@ -1004,6 +1015,37 @@ func (scan *Scan) planFilesRemote(ctx context.Context) ([]FileScanTask, error) { return result.Tasks, nil } +// remotePlanSchema resolves the schema a remote plan binds against and reports +// whether it came from the scan's snapshot rather than the table's current +// schema, which rides the wire as use-snapshot-schema. Compares by schema id, +// not pointer: CurrentSchema returns a clone. +func (scan *Scan) remotePlanSchema() (*iceberg.Schema, bool, error) { + schema, err := scan.effectiveSchema() + if err != nil { + return nil, false, err + } + + return schema, schema.ID != scan.metadata.CurrentSchema().ID, nil +} + +// rejectRemoteRowLineage fails a remote scan that projects +// _last_updated_sequence_number: the REST FileScanTask schema carries no +// manifest data sequence number, so the reader would emit nulls where a local +// scan emits real values. _row_id is unaffected, its first_row_id lives on the +// data file and survives the wire. +func (scan *Scan) rejectRemoteRowLineage() error { + _, lineage := splitLineageMetadataFields(scan.selectedFields, scan.caseSensitive) + selectsSeqNum := slices.ContainsFunc(lineage, func(f iceberg.NestedField) bool { + return f.ID == iceberg.LastUpdatedSequenceNumberFieldID + }) + if !scan.includeRowLineage && !selectsSeqNum { + return nil + } + + return fmt.Errorf("%w: remote scan planning cannot project %s: the REST FileScanTask schema carries no data sequence number", + ErrInvalidOperation, iceberg.LastUpdatedSequenceNumberColumnName) +} + type planIOState struct { io PlanIO