You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "rambleraptor (via GitHub)" <gi...@apache.org> on 2026/07/15 19:43:25 UTC
[PR] ScanPlan requests + responses [iceberg-go]
rambleraptor opened a new pull request, #1477:
URL: https://github.com/apache/iceberg-go/pull/1477
This allows catalog.PlanFiles to send the Plan requests and parse the responses. This brings together the work that @Revanth14 has been doing with my expression parsing work.
I also changed the expression parsing interface, since we already have JSON parsing interfaces for expressions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(catalog/rest): ScanPlan requests + responses [iceberg-go]
Posted by "rambleraptor (via GitHub)" <gi...@apache.org>.
rambleraptor commented on code in PR #1477:
URL: https://github.com/apache/iceberg-go/pull/1477#discussion_r3659740844
##########
catalog/rest/vended_creds.go:
##########
@@ -108,14 +114,20 @@ func (v *vendedCredentialRefresher) loadFS(ctx context.Context) (iceio.IO, error
}
defer v.mu.Release(1)
- if v.cachedIO != nil && !v.now().After(v.expiresAt) {
+ if v.cachedIO != nil && !v.expired() {
return v.cachedIO, nil
}
var config iceberg.Properties
- if v.cachedIO == nil {
+ switch {
+ case v.cachedIO == nil:
config = v.props
Review Comment:
Thanks for the nit @zeroshade. I was OOO last week and just getting around to it. Fixed now.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] ScanPlan requests + responses [iceberg-go]
Posted by "tanmayrauth (via GitHub)" <gi...@apache.org>.
tanmayrauth commented on code in PR #1477:
URL: https://github.com/apache/iceberg-go/pull/1477#discussion_r3605558647
##########
catalog/rest/scan_planning.go:
##########
@@ -187,9 +190,208 @@ func (r *Catalog) SupportsRemoteScanPlanning() bool {
}
// PlanFiles plans a scan server-side and returns tasks (and, optionally, a
-// plan-scoped FileIO) for the table to read.
+// plan-scoped FileIO) for the table to read. It submits the plan, polls a
+// 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.
func (r *Catalog) PlanFiles(ctx context.Context, req table.ScanPlanningRequest) (table.ScanPlanningResult, error) {
- return table.ScanPlanningResult{}, fmt.Errorf("%w: REST scan planning", iceberg.ErrNotImplemented)
+ wire, err := planTableScanRequestFrom(req)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ resp, err := r.PlanTableScan(ctx, req.Identifier, wire)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ // Resolve to a completed plan: use an inline result as-is, or poll a
+ // submitted one to completion. PlanTableScan maps failed to an error and
+ // rejects other statuses, so only completed/submitted reach here.
+ var completed CompletedPlanningResult
+ switch resp.Status {
+ case PlanStatusCompleted:
+ completed = CompletedPlanningResult{
+ ScanTasks: resp.ScanTasks,
+ StorageCredentials: resp.StorageCredentials,
+ }
+ case PlanStatusSubmitted:
+ completed, err = r.WaitForPlan(ctx, req.Identifier, *resp.PlanID, WaitForPlanOptions{})
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+ default:
+ return table.ScanPlanningResult{}, fmt.Errorf(
+ "%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)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ return table.ScanPlanningResult{
+ Tasks: tasks,
+ IO: planIOFromCredentials(completed.StorageCredentials, req.MetadataLocation, r.planIOBaseProps(req.Metadata)),
+ }, nil
+}
+
+// planIOBaseProps rebuilds the props the table's own FileIO was built with, so a
+// plan-scoped IO keeps settings like a custom S3 endpoint or region. loadTable's
+// config block never reaches the planner seam, so it isn't included.
+func (r *Catalog) planIOBaseProps(meta table.ScanPlanningMetadata) iceberg.Properties {
+ props := make(iceberg.Properties, len(r.props))
+ maps.Copy(props, r.props)
+ if meta != nil {
+ maps.Copy(props, meta.Properties())
Review Comment:
This overlay carries catalog props + table metadata props, but the normal table FileIO is built from r.props + Metadata.Properties() + ret.Config + vended creds (rest.go:1256-1260). ret.Config is the loadTable `config` block — merged last-but-one so it overrides catalog/table props — and it's the standard place a REST server returns s3.endpoint, s3.path-style-access, and client.region, especially for non-AWS / vended-cred setups.
Dropping it means those are the settings most likely to go missing: a MinIO-backed table whose server returns s3.endpoint + s3.path-style-access in the config block gets them on the normal FileIO but not on the plan-scoped IO, so vended-cred ReadTasks point at the AWS-default host and fail — unless the plan creds happen to re-carry that config, which isn't guaranteed for endpoint/path-style/region. This is the same class of property drop flagged earlier; carrying r.props + table props doesn't cover it because the override source is ret.Config. It can't be fixed with another maps.Copy here — ScanPlanningRequest/ScanPlanningMetadata never expose ret.Config, so the seam itself needs to carry the effective IO props (config block included) for the overlay to actually preserve endpoint/region/path-style.
##########
catalog/rest/scan_planning.go:
##########
@@ -187,9 +190,208 @@ func (r *Catalog) SupportsRemoteScanPlanning() bool {
}
// PlanFiles plans a scan server-side and returns tasks (and, optionally, a
-// plan-scoped FileIO) for the table to read.
+// plan-scoped FileIO) for the table to read. It submits the plan, polls a
+// 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.
func (r *Catalog) PlanFiles(ctx context.Context, req table.ScanPlanningRequest) (table.ScanPlanningResult, error) {
- return table.ScanPlanningResult{}, fmt.Errorf("%w: REST scan planning", iceberg.ErrNotImplemented)
+ wire, err := planTableScanRequestFrom(req)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ resp, err := r.PlanTableScan(ctx, req.Identifier, wire)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ // Resolve to a completed plan: use an inline result as-is, or poll a
+ // submitted one to completion. PlanTableScan maps failed to an error and
+ // rejects other statuses, so only completed/submitted reach here.
+ var completed CompletedPlanningResult
+ switch resp.Status {
+ case PlanStatusCompleted:
+ completed = CompletedPlanningResult{
+ ScanTasks: resp.ScanTasks,
+ StorageCredentials: resp.StorageCredentials,
+ }
+ case PlanStatusSubmitted:
+ completed, err = r.WaitForPlan(ctx, req.Identifier, *resp.PlanID, WaitForPlanOptions{})
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+ default:
+ return table.ScanPlanningResult{}, fmt.Errorf(
+ "%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)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ return table.ScanPlanningResult{
+ Tasks: tasks,
+ IO: planIOFromCredentials(completed.StorageCredentials, req.MetadataLocation, r.planIOBaseProps(req.Metadata)),
+ }, nil
+}
+
+// planIOBaseProps rebuilds the props the table's own FileIO was built with, so a
+// plan-scoped IO keeps settings like a custom S3 endpoint or region. loadTable's
+// config block never reaches the planner seam, so it isn't included.
+func (r *Catalog) planIOBaseProps(meta table.ScanPlanningMetadata) iceberg.Properties {
+ props := make(iceberg.Properties, len(r.props))
+ maps.Copy(props, r.props)
+ if meta != nil {
+ maps.Copy(props, meta.Properties())
+ }
+
+ 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...)
+
+ queue := append([]string(nil), tasks.PlanTasks...)
+ seen := make(map[string]bool, len(queue))
+ for len(queue) > 0 {
+ handle := queue[0]
+ queue = queue[1:]
+ if seen[handle] {
+ continue
+ }
+ seen[handle] = true
+
+ resp, err := r.FetchScanTasks(ctx, ident, FetchScanTasksRequest{PlanTask: handle})
+ if err != nil {
+ return nil, 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
+ }
+
+ return nil, fmt.Errorf("%w: decoding remote scan tasks", iceberg.ErrNotImplemented)
+}
+
+// planIOFromCredentials wraps a plan's vended storage credentials in a lazy
+// PlanIO, overlaid on the table's base IO props so vended keys win on collision.
+// With none vended it returns a nil PlanIO, and the scan falls back to the
+// table's own FileIO.
+func planIOFromCredentials(creds []StorageCredential, location string, baseProps iceberg.Properties) table.PlanIO {
+ if len(creds) == 0 {
+ return nil
+ }
+
+ // TODO(#1178): plan.storage-credentials is prefix-scoped, but this builds one
+ // IO from the metadata location — wrong cred if data/delete files sit under a
+ // different prefix. The scan-task decoder settles the PlanIO shape.
+ resolved := resolveStorageCredentials(creds, location)
+ props := make(iceberg.Properties, len(baseProps)+len(resolved))
+ maps.Copy(props, baseProps)
+ maps.Copy(props, resolved)
+
+ return &planScopedIO{refresher: &vendedCredentialRefresher{
+ mu: semaphore.NewWeighted(1),
+ location: location,
+ props: props,
+ // fetchCreds stays nil: a plan's creds can't be renewed via the
+ // table-credentials endpoint, so an expiry is fatal, not a re-fetch.
+ }}
+}
+
+// planScopedIO lazily builds and caches a FileIO from a plan's vended creds,
+// reusing vendedCredentialRefresher for expiry handling. Satisfies table.PlanIO.
+type planScopedIO struct {
+ refresher *vendedCredentialRefresher
+}
+
+func (p *planScopedIO) Load(ctx context.Context) (iceio.IO, error) {
+ return p.refresher.loadFS(ctx)
+}
+
+func (p *planScopedIO) Close() error {
+ return p.refresher.close()
+}
+
+// planTableScanRequestFrom builds the planTableScan request body from the
+// planner-level request, serializing the row filter to ExpressionParser JSON. A
+// nil or always-true filter is left off so the server plans without one.
+func planTableScanRequestFrom(req table.ScanPlanningRequest) (PlanTableScanRequest, error) {
+ out := PlanTableScanRequest{
+ SnapshotID: req.SnapshotID,
+ Select: req.SelectedFields,
+ MinRowsRequested: req.MinRowsRequested,
+ CaseSensitive: req.CaseSensitive,
+ UseSnapshotSchema: req.UseSnapshotSchema,
+ StatsFields: req.StatsFields,
+ }
+
+ if req.RowFilter != nil && !req.RowFilter.Equals(iceberg.AlwaysTrue{}) {
+ filter, err := marshalScanFilter(req)
+ if err != nil {
+ return PlanTableScanRequest{}, err
+ }
+ out.Filter = filter
+ }
+
+ return out, nil
+}
+
+// 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)
+ }
+
+ 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)
+ if err != nil {
+ return nil, fmt.Errorf("%w: binding scan filter: %s", iceberg.ErrInvalidArgument, err)
+ }
+
+ filter, err := json.Marshal(bound)
Review Comment:
Binding before marshal fixed the timestamp case but regressed nested-field refs. boundRef.MarshalJSON (expr_json.go:276) emits the leaf name, so EqualTo(Reference("location.lat"), 45.0) now serializes as
{"type":"eq","term":"lat",...} where the old unbound path emitted "location.lat". The server then resolves the wrong column (or a different top-level "lat") → wrong files planned / silent wrong results. Suggest binding only for literal typing and serializing the term from the full path (schema.FindColumnName(field.ID)), plus a nested-field test.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(catalog/rest): ScanPlan requests + responses [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade commented on code in PR #1477:
URL: https://github.com/apache/iceberg-go/pull/1477#discussion_r3616148026
##########
catalog/rest/vended_creds.go:
##########
@@ -108,14 +114,20 @@ func (v *vendedCredentialRefresher) loadFS(ctx context.Context) (iceio.IO, error
}
defer v.mu.Release(1)
- if v.cachedIO != nil && !v.now().After(v.expiresAt) {
+ if v.cachedIO != nil && !v.expired() {
return v.cachedIO, nil
}
var config iceberg.Properties
- if v.cachedIO == nil {
+ switch {
+ case v.cachedIO == nil:
config = v.props
Review Comment:
Non-blocking nit (fine as a follow-up — not gating the approval):
The `ErrVendedCredentialsExpired` fail-loud path and the `expired()` guard above both only fire when `cachedIO != nil`. On the **first** `Load` we take this branch (`cachedIO == nil`), set `config = v.props`, and build the FileIO via `iceio.LoadFS` below **without** checking whether the creds are already past `expiresAt`. So a plan whose creds expire before the first read still gets a working-looking IO whose reads 403 — the exact opaque failure this sentinel was added to prevent.
It's narrow (needs creds already expired at first use) and non-regressive, so it's a nit rather than a blocker. To close it: when `fetchCreds == nil && cachedIO == nil`, compute the expiry from `config` and return `ErrVendedCredentialsExpired` if it's already past, before `LoadFS`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] fix(catalog/rest): ScanPlan requests + responses [iceberg-go]
Posted by "zeroshade (via GitHub)" <gi...@apache.org>.
zeroshade merged PR #1477:
URL: https://github.com/apache/iceberg-go/pull/1477
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] ScanPlan requests + responses [iceberg-go]
Posted by "rambleraptor (via GitHub)" <gi...@apache.org>.
rambleraptor commented on code in PR #1477:
URL: https://github.com/apache/iceberg-go/pull/1477#discussion_r3605187846
##########
catalog/rest/scan_planning.go:
##########
@@ -187,9 +189,180 @@ func (r *Catalog) SupportsRemoteScanPlanning() bool {
}
// PlanFiles plans a scan server-side and returns tasks (and, optionally, a
-// plan-scoped FileIO) for the table to read.
+// plan-scoped FileIO) for the table to read. It submits the plan, polls a
+// 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.
func (r *Catalog) PlanFiles(ctx context.Context, req table.ScanPlanningRequest) (table.ScanPlanningResult, error) {
- return table.ScanPlanningResult{}, fmt.Errorf("%w: REST scan planning", iceberg.ErrNotImplemented)
+ wire, err := planTableScanRequestFrom(req)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ resp, err := r.PlanTableScan(ctx, req.Identifier, wire)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ // Resolve to a completed plan: use an inline result as-is, or poll a
+ // submitted one to completion. PlanTableScan maps failed to an error and
+ // rejects other statuses, so only completed/submitted reach here.
+ var completed CompletedPlanningResult
+ switch resp.Status {
+ case PlanStatusCompleted:
+ completed = CompletedPlanningResult{
+ ScanTasks: resp.ScanTasks,
+ StorageCredentials: resp.StorageCredentials,
+ }
+ case PlanStatusSubmitted:
+ completed, err = r.WaitForPlan(ctx, req.Identifier, *resp.PlanID, WaitForPlanOptions{})
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+ default:
+ return table.ScanPlanningResult{}, fmt.Errorf(
+ "%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)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ return table.ScanPlanningResult{
+ Tasks: tasks,
+ IO: planIOFromCredentials(completed.StorageCredentials, req.MetadataLocation),
Review Comment:
I set a TODO to help determine this in a later phase if that works.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] ScanPlan requests + responses [iceberg-go]
Posted by "rambleraptor (via GitHub)" <gi...@apache.org>.
rambleraptor commented on code in PR #1477:
URL: https://github.com/apache/iceberg-go/pull/1477#discussion_r3605183141
##########
catalog/rest/scan_planning.go:
##########
@@ -187,9 +189,180 @@ func (r *Catalog) SupportsRemoteScanPlanning() bool {
}
// PlanFiles plans a scan server-side and returns tasks (and, optionally, a
-// plan-scoped FileIO) for the table to read.
+// plan-scoped FileIO) for the table to read. It submits the plan, polls a
+// 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.
func (r *Catalog) PlanFiles(ctx context.Context, req table.ScanPlanningRequest) (table.ScanPlanningResult, error) {
- return table.ScanPlanningResult{}, fmt.Errorf("%w: REST scan planning", iceberg.ErrNotImplemented)
+ wire, err := planTableScanRequestFrom(req)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ resp, err := r.PlanTableScan(ctx, req.Identifier, wire)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ // Resolve to a completed plan: use an inline result as-is, or poll a
+ // submitted one to completion. PlanTableScan maps failed to an error and
+ // rejects other statuses, so only completed/submitted reach here.
+ var completed CompletedPlanningResult
+ switch resp.Status {
+ case PlanStatusCompleted:
+ completed = CompletedPlanningResult{
+ ScanTasks: resp.ScanTasks,
+ StorageCredentials: resp.StorageCredentials,
+ }
+ case PlanStatusSubmitted:
+ completed, err = r.WaitForPlan(ctx, req.Identifier, *resp.PlanID, WaitForPlanOptions{})
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+ default:
+ return table.ScanPlanningResult{}, fmt.Errorf(
+ "%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)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ return table.ScanPlanningResult{
+ Tasks: tasks,
+ IO: planIOFromCredentials(completed.StorageCredentials, req.MetadataLocation),
+ }, nil
+}
+
+// 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...)
+
+ queue := append([]string(nil), tasks.PlanTasks...)
+ seen := make(map[string]bool, len(queue))
+ for len(queue) > 0 {
+ handle := queue[0]
+ queue = queue[1:]
+ if seen[handle] {
+ continue
+ }
+ seen[handle] = true
+
+ resp, err := r.FetchScanTasks(ctx, ident, FetchScanTasksRequest{PlanTask: handle})
+ if err != nil {
+ return nil, 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
+ }
+
+ return nil, fmt.Errorf("%w: decoding remote scan tasks", iceberg.ErrNotImplemented)
+}
+
+// planIOFromCredentials wraps a plan's vended storage credentials in a lazy
+// PlanIO. With none vended it returns a nil PlanIO, and the scan falls back to
+// the table's own FileIO.
+func planIOFromCredentials(creds []StorageCredential, location string) table.PlanIO {
+ if len(creds) == 0 {
+ return nil
+ }
+
+ return &planScopedIO{creds: creds, location: location}
+}
+
+// planScopedIO builds a FileIO from a plan's vended storage credentials on first
+// use and caches it. It satisfies table.PlanIO. Per the delivery contract a Scan
+// carrying plan-scoped IO is not read concurrently, but the mutex keeps the lazy
+// build and Close honest.
+type planScopedIO struct {
+ creds []StorageCredential
+ location string
+
+ mu sync.Mutex
+ cached iceio.IO
+}
+
+func (p *planScopedIO) Load(ctx context.Context) (iceio.IO, error) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ if p.cached != nil {
+ return p.cached, nil
+ }
+
+ fs, err := iceio.LoadFS(ctx, resolveStorageCredentials(p.creds, p.location), p.location)
+ if err != nil {
+ return nil, err
+ }
+ p.cached = fs
+
+ return p.cached, nil
+}
+
+func (p *planScopedIO) Close() error {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ closer, ok := p.cached.(interface{ Close() error })
+ p.cached = nil
+ if ok {
+ return closer.Close()
+ }
+
+ return nil
+}
+
+// planTableScanRequestFrom builds the planTableScan request body from the
+// planner-level request, serializing the row filter to ExpressionParser JSON. A
+// nil or always-true filter is left off so the server plans without one.
+func planTableScanRequestFrom(req table.ScanPlanningRequest) (PlanTableScanRequest, error) {
+ out := PlanTableScanRequest{
Review Comment:
I believe that @zeroshade's copy suggestion will fix this.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] ScanPlan requests + responses [iceberg-go]
Posted by "rambleraptor (via GitHub)" <gi...@apache.org>.
rambleraptor commented on PR #1477:
URL: https://github.com/apache/iceberg-go/pull/1477#issuecomment-5006138210
@zeroshade made those changes! Let me know if the JSON marshaling looks wonky. I'm trying to make sure we can use standard json.Marshal / json.Unmarshal whenever possible.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] ScanPlan requests + responses [iceberg-go]
Posted by "tanmayrauth (via GitHub)" <gi...@apache.org>.
tanmayrauth commented on code in PR #1477:
URL: https://github.com/apache/iceberg-go/pull/1477#discussion_r3597488881
##########
catalog/rest/scan_planning.go:
##########
@@ -187,9 +189,180 @@ func (r *Catalog) SupportsRemoteScanPlanning() bool {
}
// PlanFiles plans a scan server-side and returns tasks (and, optionally, a
-// plan-scoped FileIO) for the table to read.
+// plan-scoped FileIO) for the table to read. It submits the plan, polls a
+// 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.
func (r *Catalog) PlanFiles(ctx context.Context, req table.ScanPlanningRequest) (table.ScanPlanningResult, error) {
- return table.ScanPlanningResult{}, fmt.Errorf("%w: REST scan planning", iceberg.ErrNotImplemented)
+ wire, err := planTableScanRequestFrom(req)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ resp, err := r.PlanTableScan(ctx, req.Identifier, wire)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ // Resolve to a completed plan: use an inline result as-is, or poll a
+ // submitted one to completion. PlanTableScan maps failed to an error and
+ // rejects other statuses, so only completed/submitted reach here.
+ var completed CompletedPlanningResult
+ switch resp.Status {
+ case PlanStatusCompleted:
+ completed = CompletedPlanningResult{
+ ScanTasks: resp.ScanTasks,
+ StorageCredentials: resp.StorageCredentials,
+ }
+ case PlanStatusSubmitted:
+ completed, err = r.WaitForPlan(ctx, req.Identifier, *resp.PlanID, WaitForPlanOptions{})
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+ default:
+ return table.ScanPlanningResult{}, fmt.Errorf(
+ "%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)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ return table.ScanPlanningResult{
+ Tasks: tasks,
+ IO: planIOFromCredentials(completed.StorageCredentials, req.MetadataLocation),
Review Comment:
This resolves the plan's credential array against the metadata location and builds a single FileIO for the scan. That matches how the rest of iceberg-go handles vended creds (loadTable resolves against MetadataLoc too), so it's not a regression — but plan.storage-credentials is an array of prefix-scoped creds, and resolving once against the metadata location means a plan whose data/delete files sit under a different prefix picks the wrong or no credential once the decoder routes reads through this IO. Fine for a single-bucket table where data lives under the table root; worth settling the PlanIO shape before the
decoder makes it load-bearing.
##########
catalog/rest/scan_planning.go:
##########
@@ -187,9 +189,180 @@ func (r *Catalog) SupportsRemoteScanPlanning() bool {
}
// PlanFiles plans a scan server-side and returns tasks (and, optionally, a
-// plan-scoped FileIO) for the table to read.
+// plan-scoped FileIO) for the table to read. It submits the plan, polls a
+// 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.
func (r *Catalog) PlanFiles(ctx context.Context, req table.ScanPlanningRequest) (table.ScanPlanningResult, error) {
- return table.ScanPlanningResult{}, fmt.Errorf("%w: REST scan planning", iceberg.ErrNotImplemented)
+ wire, err := planTableScanRequestFrom(req)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ resp, err := r.PlanTableScan(ctx, req.Identifier, wire)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ // Resolve to a completed plan: use an inline result as-is, or poll a
+ // submitted one to completion. PlanTableScan maps failed to an error and
+ // rejects other statuses, so only completed/submitted reach here.
+ var completed CompletedPlanningResult
+ switch resp.Status {
+ case PlanStatusCompleted:
+ completed = CompletedPlanningResult{
+ ScanTasks: resp.ScanTasks,
+ StorageCredentials: resp.StorageCredentials,
+ }
+ case PlanStatusSubmitted:
+ completed, err = r.WaitForPlan(ctx, req.Identifier, *resp.PlanID, WaitForPlanOptions{})
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+ default:
+ return table.ScanPlanningResult{}, fmt.Errorf(
+ "%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)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ return table.ScanPlanningResult{
+ Tasks: tasks,
+ IO: planIOFromCredentials(completed.StorageCredentials, req.MetadataLocation),
+ }, nil
+}
+
+// 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...)
+
+ queue := append([]string(nil), tasks.PlanTasks...)
+ seen := make(map[string]bool, len(queue))
+ for len(queue) > 0 {
+ handle := queue[0]
+ queue = queue[1:]
+ if seen[handle] {
+ continue
+ }
+ seen[handle] = true
+
+ resp, err := r.FetchScanTasks(ctx, ident, FetchScanTasksRequest{PlanTask: handle})
+ if err != nil {
+ return nil, 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
+ }
+
+ return nil, fmt.Errorf("%w: decoding remote scan tasks", iceberg.ErrNotImplemented)
+}
+
+// planIOFromCredentials wraps a plan's vended storage credentials in a lazy
+// PlanIO. With none vended it returns a nil PlanIO, and the scan falls back to
+// the table's own FileIO.
+func planIOFromCredentials(creds []StorageCredential, location string) table.PlanIO {
+ if len(creds) == 0 {
+ return nil
+ }
+
+ return &planScopedIO{creds: creds, location: location}
+}
+
+// planScopedIO builds a FileIO from a plan's vended storage credentials on first
+// use and caches it. It satisfies table.PlanIO. Per the delivery contract a Scan
+// carrying plan-scoped IO is not read concurrently, but the mutex keeps the lazy
+// build and Close honest.
+type planScopedIO struct {
+ creds []StorageCredential
+ location string
+
+ mu sync.Mutex
+ cached iceio.IO
+}
+
+func (p *planScopedIO) Load(ctx context.Context) (iceio.IO, error) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ if p.cached != nil {
+ return p.cached, nil
+ }
+
+ fs, err := iceio.LoadFS(ctx, resolveStorageCredentials(p.creds, p.location), p.location)
+ if err != nil {
+ return nil, err
+ }
+ p.cached = fs
+
+ return p.cached, nil
+}
+
+func (p *planScopedIO) Close() error {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ closer, ok := p.cached.(interface{ Close() error })
+ p.cached = nil
+ if ok {
+ return closer.Close()
+ }
+
+ return nil
+}
+
+// planTableScanRequestFrom builds the planTableScan request body from the
+// planner-level request, serializing the row filter to ExpressionParser JSON. A
+// nil or always-true filter is left off so the server plans without one.
+func planTableScanRequestFrom(req table.ScanPlanningRequest) (PlanTableScanRequest, error) {
+ out := PlanTableScanRequest{
Review Comment:
Neither the initial plan request here nor the poll at line 223 (WaitForPlanOptions{}) sets AccessDelegation, so X-Iceberg-Access-Delegation is never sent. A spec-compliant server vends storage credentials only when asked, so completed.StorageCredentials comes back empty, planIOFromCredentials returns nil, and the scan silently falls back to the table's ambient FileIO — for a table with no ambient creds that's a read failure at scan time. TestPlanFilesSurfacesVendedCredentials passes only because the mock vends unconditionally, and ScanPlanningRequest has no field to carry delegation, so a caller can't thread it either. This is the fix that has to land first — the planScopedIO path below is unreachable against a real server until it does.
##########
catalog/rest/scan_planning.go:
##########
@@ -187,9 +189,180 @@ func (r *Catalog) SupportsRemoteScanPlanning() bool {
}
// PlanFiles plans a scan server-side and returns tasks (and, optionally, a
-// plan-scoped FileIO) for the table to read.
+// plan-scoped FileIO) for the table to read. It submits the plan, polls a
+// 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.
func (r *Catalog) PlanFiles(ctx context.Context, req table.ScanPlanningRequest) (table.ScanPlanningResult, error) {
- return table.ScanPlanningResult{}, fmt.Errorf("%w: REST scan planning", iceberg.ErrNotImplemented)
+ wire, err := planTableScanRequestFrom(req)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ resp, err := r.PlanTableScan(ctx, req.Identifier, wire)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ // Resolve to a completed plan: use an inline result as-is, or poll a
+ // submitted one to completion. PlanTableScan maps failed to an error and
+ // rejects other statuses, so only completed/submitted reach here.
+ var completed CompletedPlanningResult
+ switch resp.Status {
+ case PlanStatusCompleted:
+ completed = CompletedPlanningResult{
+ ScanTasks: resp.ScanTasks,
+ StorageCredentials: resp.StorageCredentials,
+ }
+ case PlanStatusSubmitted:
+ completed, err = r.WaitForPlan(ctx, req.Identifier, *resp.PlanID, WaitForPlanOptions{})
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+ default:
+ return table.ScanPlanningResult{}, fmt.Errorf(
+ "%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)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ return table.ScanPlanningResult{
+ Tasks: tasks,
+ IO: planIOFromCredentials(completed.StorageCredentials, req.MetadataLocation),
+ }, nil
+}
+
+// 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...)
+
+ queue := append([]string(nil), tasks.PlanTasks...)
+ seen := make(map[string]bool, len(queue))
+ for len(queue) > 0 {
+ handle := queue[0]
+ queue = queue[1:]
+ if seen[handle] {
+ continue
+ }
+ seen[handle] = true
+
+ resp, err := r.FetchScanTasks(ctx, ident, FetchScanTasksRequest{PlanTask: handle})
+ if err != nil {
+ return nil, 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
+ }
+
+ return nil, fmt.Errorf("%w: decoding remote scan tasks", iceberg.ErrNotImplemented)
+}
+
+// planIOFromCredentials wraps a plan's vended storage credentials in a lazy
+// PlanIO. With none vended it returns a nil PlanIO, and the scan falls back to
+// the table's own FileIO.
+func planIOFromCredentials(creds []StorageCredential, location string) table.PlanIO {
+ if len(creds) == 0 {
+ return nil
+ }
+
+ return &planScopedIO{creds: creds, location: location}
+}
+
+// planScopedIO builds a FileIO from a plan's vended storage credentials on first
+// use and caches it. It satisfies table.PlanIO. Per the delivery contract a Scan
+// carrying plan-scoped IO is not read concurrently, but the mutex keeps the lazy
+// build and Close honest.
+type planScopedIO struct {
+ creds []StorageCredential
+ location string
+
+ mu sync.Mutex
+ cached iceio.IO
+}
+
+func (p *planScopedIO) Load(ctx context.Context) (iceio.IO, error) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ if p.cached != nil {
+ return p.cached, nil
+ }
+
+ fs, err := iceio.LoadFS(ctx, resolveStorageCredentials(p.creds, p.location), p.location)
+ if err != nil {
+ return nil, err
+ }
+ p.cached = fs
+
+ return p.cached, nil
+}
+
+func (p *planScopedIO) Close() error {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ closer, ok := p.cached.(interface{ Close() error })
+ p.cached = nil
+ if ok {
+ return closer.Close()
+ }
+
+ return nil
+}
+
+// planTableScanRequestFrom builds the planTableScan request body from the
+// planner-level request, serializing the row filter to ExpressionParser JSON. A
+// nil or always-true filter is left off so the server plans without one.
+func planTableScanRequestFrom(req table.ScanPlanningRequest) (PlanTableScanRequest, error) {
+ out := PlanTableScanRequest{
+ SnapshotID: req.SnapshotID,
+ Select: req.SelectedFields,
+ MinRowsRequested: req.MinRowsRequested,
+ CaseSensitive: req.CaseSensitive,
+ UseSnapshotSchema: req.UseSnapshotSchema,
+ StatsFields: req.StatsFields,
+ }
+
+ if req.RowFilter != nil && !req.RowFilter.Equals(iceberg.AlwaysTrue{}) {
+ filter, err := json.Marshal(req.RowFilter)
Review Comment:
Not a bug today — all BooleanExpression types have MarshalJSON and predicateType errors on unknown ops. But moving from a dedicated MarshalExpressionJSON entry point to bare json.Marshal drops the one place that guaranteed an expression is REST-serializable: marshalChildren recurses through json.Marshal on the interface, so a future expression type added without a MarshalJSON won't fail to compile and won't error — it silently emits struct-default JSON and the server gets a malformed filter. Marshaling through a helper that asserts the concrete type is known would keep that a loud error rather than a silent wrong
filter.
##########
catalog/rest/scan_planning.go:
##########
@@ -187,9 +189,180 @@ func (r *Catalog) SupportsRemoteScanPlanning() bool {
}
// PlanFiles plans a scan server-side and returns tasks (and, optionally, a
-// plan-scoped FileIO) for the table to read.
+// plan-scoped FileIO) for the table to read. It submits the plan, polls a
+// 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.
func (r *Catalog) PlanFiles(ctx context.Context, req table.ScanPlanningRequest) (table.ScanPlanningResult, error) {
- return table.ScanPlanningResult{}, fmt.Errorf("%w: REST scan planning", iceberg.ErrNotImplemented)
+ wire, err := planTableScanRequestFrom(req)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ resp, err := r.PlanTableScan(ctx, req.Identifier, wire)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ // Resolve to a completed plan: use an inline result as-is, or poll a
+ // submitted one to completion. PlanTableScan maps failed to an error and
+ // rejects other statuses, so only completed/submitted reach here.
+ var completed CompletedPlanningResult
+ switch resp.Status {
+ case PlanStatusCompleted:
+ completed = CompletedPlanningResult{
+ ScanTasks: resp.ScanTasks,
+ StorageCredentials: resp.StorageCredentials,
+ }
+ case PlanStatusSubmitted:
+ completed, err = r.WaitForPlan(ctx, req.Identifier, *resp.PlanID, WaitForPlanOptions{})
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+ default:
+ return table.ScanPlanningResult{}, fmt.Errorf(
+ "%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)
+ if err != nil {
+ return table.ScanPlanningResult{}, err
+ }
+
+ return table.ScanPlanningResult{
+ Tasks: tasks,
+ IO: planIOFromCredentials(completed.StorageCredentials, req.MetadataLocation),
+ }, nil
+}
+
+// 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...)
+
+ queue := append([]string(nil), tasks.PlanTasks...)
+ seen := make(map[string]bool, len(queue))
+ for len(queue) > 0 {
+ handle := queue[0]
+ queue = queue[1:]
+ if seen[handle] {
+ continue
+ }
+ seen[handle] = true
+
+ resp, err := r.FetchScanTasks(ctx, ident, FetchScanTasksRequest{PlanTask: handle})
+ if err != nil {
+ return nil, 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
+ }
+
+ return nil, fmt.Errorf("%w: decoding remote scan tasks", iceberg.ErrNotImplemented)
+}
+
+// planIOFromCredentials wraps a plan's vended storage credentials in a lazy
+// PlanIO. With none vended it returns a nil PlanIO, and the scan falls back to
+// the table's own FileIO.
+func planIOFromCredentials(creds []StorageCredential, location string) table.PlanIO {
+ if len(creds) == 0 {
+ return nil
+ }
+
+ return &planScopedIO{creds: creds, location: location}
+}
+
+// planScopedIO builds a FileIO from a plan's vended storage credentials on first
+// use and caches it. It satisfies table.PlanIO. Per the delivery contract a Scan
+// carrying plan-scoped IO is not read concurrently, but the mutex keeps the lazy
+// build and Close honest.
+type planScopedIO struct {
+ creds []StorageCredential
+ location string
+
+ mu sync.Mutex
+ cached iceio.IO
+}
+
+func (p *planScopedIO) Load(ctx context.Context) (iceio.IO, error) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ if p.cached != nil {
+ return p.cached, nil
+ }
+
+ fs, err := iceio.LoadFS(ctx, resolveStorageCredentials(p.creds, p.location), p.location)
Review Comment:
This builds the plan FileIO once and caches it until Close with no expiry handling, whereas the normal table path (rest.go:1044) reads vended creds through vendedCredentialRefresher, which tracks the credential expiry keys parseCredentialExpiry already knows about and re-fetches on TTL. A scan that outlives the vended credential's TTL — the long fanout plans that make server-side planning worthwhile — starts getting raw storage 403s on later ReadTasks reads with no diagnosable "expired" error. Plan creds can't be refreshed via the table-cred endpoint, so this needs a deliberate call: at least reuse parseCredentialExpiry to surface an expiry error, and ideally don't stand up a second vended-cred IO path parallel to vendedCredentialRefresher.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org