Skip to content

Begin adding scan tasks from SSP #1860

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
rambleraptor wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Begin adding scan tasks from SSP #1860

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
134 changes: 86 additions & 48 deletions catalog/rest/scan_planning.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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
Expand All @@ -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 {
Expand Down Expand Up @@ -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
}
Expand All @@ -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))
Expand All @@ -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
Expand Down Expand Up @@ -366,22 +390,36 @@ 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
if req.CaseSensitive != nil {
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)
}
Expand Down
154 changes: 138 additions & 16 deletions catalog/rest/scan_planning_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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
Expand Down
Loading
Loading