laskoviymishka commented on code in PR #1857:
URL: https://github.com/apache/iceberg-go/pull/1857#discussion_r3862507113
##########
catalog/rest/vended_creds.go:
##########
@@ -230,3 +271,165 @@ func (v *vendedCredentialRefresher) close() error {
return nil
}
+
+// prefixScopedIO selects a plan credential using the actual object location
+// passed to Open/Remove. A single plan may cover metadata, data, and delete
+// files in different storage prefixes, so resolving credentials once at plan
+// creation would be incorrect.
+type prefixScopedIO struct {
+ ctx context.Context
+ baseProps iceberg.Properties
+ credentials []StorageCredential
+
+ mu sync.Mutex
+ filesystems map[string]iceio.IO
+ closed bool
+ nowFunc func() time.Time
+}
+
+func newPrefixScopedIO(ctx context.Context, baseProps iceberg.Properties,
credentials []StorageCredential) *prefixScopedIO {
+ return &prefixScopedIO{
+ ctx: ctx,
+ baseProps: maps.Clone(baseProps),
+ credentials: slices.Clone(credentials),
+ filesystems: make(map[string]iceio.IO),
+ }
+}
+
+func (p *prefixScopedIO) Open(name string) (iceio.File, error) {
+ fs, err := p.filesystemFor(name)
+ if err != nil {
+ return nil, err
+ }
+
+ return fs.Open(name)
+}
+
+func (p *prefixScopedIO) Remove(name string) error {
+ fs, err := p.filesystemFor(name)
+ if err != nil {
+ return err
+ }
+
+ return fs.Remove(name)
+}
+
+func (p *prefixScopedIO) filesystemFor(name string) (iceio.IO, error) {
+ credentialIndex := matchingStorageCredentialIndex(p.credentials, name)
+ if credentialIndex >= 0 {
+ if expiresAt, ok :=
parseCredentialExpiry(p.credentials[credentialIndex].Config); ok &&
+ p.now().After(expiresAt) {
+ return nil, fmt.Errorf("%w: %s expired at %s",
+ ErrVendedCredentialsExpired, name,
expiresAt.Format(time.RFC3339))
+ }
+ }
+
+ key := scopedFilesystemKey(credentialIndex, name)
+
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ if p.closed {
+ return nil, errors.New("prefix-scoped IO is closed")
+ }
+ if fs, ok := p.filesystems[key]; ok {
+ return fs, nil
+ }
+
+ props := p.propertiesForLocation(name)
+
+ fs, err := iceio.LoadFS(p.ctx, props, name)
Review Comment:
I think this holds `p.mu` across the `iceio.LoadFS` call, and LoadFS can do
real network I/O (S3 instance-metadata, GCS workload-identity, ADLS token
discovery). During a parallel scan every reader goroutine calling
`Open`/`Remove` blocks behind whichever one is initializing the first uncached
prefix, even for prefixes that are already cached.
I'd load the filesystem outside the lock and only take `mu` to write the map
(double-checked):
```go
p.mu.Lock()
if p.closed { p.mu.Unlock(); return nil, errors.New("prefix-scoped IO is
closed") }
if fs, ok := p.filesystems[key]; ok { p.mu.Unlock(); return fs, nil }
p.mu.Unlock()
props := p.propertiesForLocation(name)
fs, err := iceio.LoadFS(p.ctx, props, name)
if err != nil { return nil, err }
p.mu.Lock()
defer p.mu.Unlock()
if p.closed { return nil, errors.New("prefix-scoped IO is closed") }
if existing, ok := p.filesystems[key]; ok { return existing, nil } // lost
the race
p.filesystems[key] = fs
return fs, nil
```
While we're restructuring this, the expiry check above also runs before the
lock, so I'd fold it in here too. wdyt?
##########
catalog/rest/scan_planning.go:
##########
@@ -380,8 +436,18 @@ 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)
+ schema := req.Schema
+ if schema == nil {
+ if req.Metadata == nil {
Review Comment:
Since `marshalScanFilter` already bails on nil `Metadata` up front, by the
time we're inside this `if schema == nil` block `req.Metadata` is guaranteed
non-nil and this inner check is dead. The `if schema == nil` guard just below
already covers the nil-`CurrentSchema` case.
I'd drop the inner check. It's harmless today, but it carries a different
error message than the reachable guard, so a later refactor that reaches it
would surface the wrong one.
##########
catalog/rest/scan_planning.go:
##########
@@ -145,60 +146,47 @@ const headerIdempotencyKey = "Idempotency-Key"
// --- Capability gating
-------------------------------------------------------
//
// Capability is split into two predicates. SupportsPlanTableScan is the narrow
-// "server can plan inline" check (plan endpoint only).
SupportsFullRemoteScanPlanning
-// is the endpoint-level "server advertises all four endpoints" check: an
-// end-to-end plan can come back `submitted` or with `plan-tasks` that need the
-// poll/cancel/fetch endpoints to finish, and auto mode has no second chance to
-// fall back to local once it commits to remote, so a plan-only server must not
-// 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.
-
-// SupportsPlanTableScan reports whether the server advertised the synchronous
-// plan endpoint.
+// "server can plan" check (plan endpoint only). SupportsFullRemoteScanPlanning
+// reports whether every continuation endpoint is also advertised. A plan-only
+// server can still complete a remote scan synchronously with inline file
tasks,
+// so table.Scan routes on the narrow predicate and PlanFiles checks
continuation
+// endpoints only when the response requires polling or task expansion. The
+// cancel endpoint is best-effort cleanup rather than an execution dependency.
+
+// SupportsPlanTableScan reports whether the server advertised the plan
+// submission endpoint.
func (r *Catalog) SupportsPlanTableScan() bool {
return r.endpoints.contains(endpointPlanTableScan)
}
-// SupportsFullRemoteScanPlanning reports whether the server advertised all
four
-// scan-planning endpoints (plan, fetch-result, cancel, fetch-tasks), i.e. it
can
-// drive the async/fanout path, not just sync inline planning.
+// SupportsFullRemoteScanPlanning reports whether the server advertised the
+// execution endpoints (plan, fetch-result, fetch-tasks), i.e. it can drive the
+// async/fanout path, not just sync inline planning. Cancellation is optional
+// cleanup and does not prevent a plan from producing tasks.
func (r *Catalog) SupportsFullRemoteScanPlanning() bool {
return r.SupportsPlanTableScan() &&
r.endpoints.contains(endpointFetchPlanResult) &&
- r.endpoints.contains(endpointCancelPlanning) &&
r.endpoints.contains(endpointFetchScanTasks)
}
// --- table.ScanPlanner implementation ---------------------------------------
-// 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.
-//
-// TODO(#1178): return SupportsFullRemoteScanPlanning() once PlanFiles is wired
-// end-to-end. Until then, callers probing endpoint capability should use
-// SupportsFullRemoteScanPlanning / SupportsPlanTableScan directly.
+// SupportsRemoteScanPlanning reports whether this catalog can submit a remote
+// plan. Any continuation capability is validated against the response returned
+// by the server.
func (r *Catalog) SupportsRemoteScanPlanning() bool {
- return false
+ return r.SupportsPlanTableScan()
Review Comment:
This flips `SupportsRemoteScanPlanning` from a stub to
`SupportsPlanTableScan()`, so a plan-only server now counts as a capable remote
planner. In explicit `ScanPlanningRemote` mode against such a server, a
`submitted` response fails in `WaitForPlan` with `ErrEndpointNotSupported` and
the deferred `abandonPlan` drops the same error, so the plan leaks until
server-side expiry with nothing telling the caller why.
The narrower semantics are fine, but they're invisible from outside: the
`fullRemoteScanPlanner` seam that makes auto mode stricter is unexported, so a
third-party `ScanPlanner` routes through this predicate with no signal it
differs from REST. I'd document the contract here (what auto vs explicit each
gate on) and consider a more actionable error on the submitted-then-unsupported
path. wdyt?
##########
table/scanner.go:
##########
@@ -1231,14 +1348,29 @@ func (scan *Scan) ReadTasks(ctx context.Context, tasks
[]FileScanTask) (*arrow.S
// closePlanIO releases the scoped resources associated with the current
// remote plan. It is safe to call when no remote plan has been installed.
-func (scan *Scan) closePlanIO() {
+func (scan *Scan) closePlanIO() error {
if scan.planIO == nil {
- return
+ return nil
}
planIO := scan.planIO
scan.planIO = nil
- planIO.releaseOwner()
+
+ return planIO.releaseOwner()
+}
+
+// Close releases the plan-scoped resources owned by this scan. It is safe to
+// call more than once. Active ReadTasks iterators retain their reader lease
and
+// can finish; the plan IO closes after the last lease is released. A scan must
+// not be used after Close.
+func (scan *Scan) Close() error {
+ if scan == nil || scan.closed {
+ return nil
+ }
+
+ scan.closed = true
Review Comment:
`Close()` writes `scan.closed = true` without any synchronization, while
`PlanFiles` and `ReadTasks` read it. The documented iterator-after-Close
pattern is safe since the iterator itself doesn't read the flag, but a second
goroutine calling `PlanFiles`/`ReadTasks` after Close, which is a pretty common
defensive pattern, is a data race that `-race` would flag.
I'd make `closed` an `atomic.Bool` and have `Close()` do `if
scan.closed.Swap(true) { return nil }` so the double-close guard and the reads
are race-free.
##########
table/scanner.go:
##########
@@ -1231,14 +1348,29 @@ func (scan *Scan) ReadTasks(ctx context.Context, tasks
[]FileScanTask) (*arrow.S
// closePlanIO releases the scoped resources associated with the current
// remote plan. It is safe to call when no remote plan has been installed.
-func (scan *Scan) closePlanIO() {
+func (scan *Scan) closePlanIO() error {
if scan.planIO == nil {
- return
+ return nil
}
planIO := scan.planIO
scan.planIO = nil
- planIO.releaseOwner()
+
+ return planIO.releaseOwner()
+}
+
+// Close releases the plan-scoped resources owned by this scan. It is safe to
+// call more than once. Active ReadTasks iterators retain their reader lease
and
+// can finish; the plan IO closes after the last lease is released. A scan must
+// not be used after Close.
+func (scan *Scan) Close() error {
Review Comment:
`Scan.Close` is a new exported method but there's no `io.Closer` assertion
on `*Scan`, and a caller who does `PlanFiles`, inspects tasks, and returns
early without draining via `ReadTasks` will leak the plan IO until server-side
expiry.
I'd add `var _ io.Closer = (*Scan)(nil)` and note in the `Scan` godoc that
it implements `io.Closer` and should be closed on early exit. Small thing, but
it makes the ownership contract discoverable.
##########
catalog/rest/fetch_scan_tasks_validation_test.go:
##########
@@ -0,0 +1,56 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package rest
+
+import (
+ "encoding/json"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestFetchScanTasksResponseRejectsInvalidShape(t *testing.T) {
+ t.Parallel()
+
+ for _, payload := range []string{
Review Comment:
These table cases run in a bare `for` loop with `require`, so the first
failing payload aborts the loop and the rest never run. Wrapping each payload
in a `t.Run` (with `t.Parallel()`) keeps them independent and points at exactly
which payload broke.
Same pattern in the rejection/validation loops over in
`scan_planning_test.go`. And while we're here, `AcceptsPresentEmptyTaskField`
asserts `PlanTasks`/`FileScanTasks` empty but not `DeleteFiles`, even though it
feeds `{"delete-files":[]}` as a case.
##########
table/scan_planning_remote.go:
##########
@@ -0,0 +1,114 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package table
+
+import (
+ "slices"
+
+ "github.com/apache/iceberg-go"
+)
+
+type fullRemoteScanPlanner interface {
+ SupportsFullRemoteScanPlanning() bool
+}
+
+// supportsAutomaticRemotePlanning is deliberately more conservative than
+// explicit remote mode when a planner exposes a split capability surface. A
+// REST server that only advertises the initial /plan endpoint may complete an
+// inline plan, but auto mode cannot know that before submitting and has no
safe
+// local fallback once the server responds with a continuation handle.
+func supportsAutomaticRemotePlanning(planner ScanPlanner) bool {
+ if planner == nil {
+ return false
+ }
+ if full, ok := planner.(fullRemoteScanPlanner); ok {
+ return full.SupportsFullRemoteScanPlanning()
+ }
+
+ return planner.SupportsRemoteScanPlanning()
+}
+
+// remotePlanningSelectedFields returns the fully qualified physical field
names
+// sent for a wildcard REST scan projection. It mirrors Java's
+// TypeUtil.getProjectedIds + Schema.findColumnName behavior. Java includes
+// struct field IDs as well as primitive/variant field IDs. List element IDs
are
+// included only for primitive elements, and map key/value IDs are included
only
+// when the value is primitive. Explicit projections keep their user-provided
+// names unchanged.
+func remotePlanningSelectedFields(scan *Scan, schema *iceberg.Schema)
([]string, error) {
+ if schema == nil || !slices.Contains(scan.selectedFields, "*") {
+ return scan.remoteSelectedFields(schema), nil
+ }
+
+ ids := make([]int, 0, len(schema.Fields()))
+ for _, field := range schema.Fields() {
+ appendRemoteProjectedFieldIDs(&ids, field)
+ }
+ slices.Sort(ids)
+
+ selected := make([]string, 0, len(ids))
+ for _, id := range ids {
+ if name, ok := schema.FindColumnName(id); ok {
+ selected = append(selected, name)
+ }
+ }
+
+ return selected, nil
+}
+
+func appendRemoteProjectedFieldIDs(ids *[]int, field iceberg.NestedField) {
+ appendRemoteProjectedTypeIDs(ids, field.Type, field.ID, true)
+}
+
+func appendRemoteProjectedTypeIDs(ids *[]int, typ iceberg.Type, fieldID int,
includeFieldID bool) {
+ switch typ := typ.(type) {
+ case *iceberg.StructType:
+ if includeFieldID {
+ *ids = append(*ids, fieldID)
+ }
+ for _, field := range typ.Fields() {
+ appendRemoteProjectedFieldIDs(ids, field)
+ }
+ case *iceberg.ListType:
+ if remoteProjectedLeaf(typ.Element) {
+ *ids = append(*ids, typ.ElementID)
+ } else {
+ appendRemoteProjectedTypeIDs(ids, typ.Element,
typ.ElementID, false)
+ }
+ case *iceberg.MapType:
+ if remoteProjectedLeaf(typ.ValueType) {
+ *ids = append(*ids, typ.KeyID, typ.ValueID)
+ } else {
+ appendRemoteProjectedTypeIDs(ids, typ.ValueType,
typ.ValueID, false)
Review Comment:
For a map with a non-primitive value we only recurse into the value type
here, so the key type never gets visited. Java's `GetProjectedIds.map()` visits
both key and value unconditionally.
That means for a spec-valid V3 `map<struct<...>, struct<...>>`, a wildcard
projection drops the key struct's sub-field IDs, the server never sees them,
and the returned task carries incomplete key data with no error surfaced. The
tested cases all use a primitive key, so this slips through.
I'd mirror Java by also recursing the key in the else branch:
```go
} else {
appendRemoteProjectedTypeIDs(ids, typ.ValueType, typ.ValueID, false)
appendRemoteProjectedTypeIDs(ids, typ.KeyType, typ.KeyID, false)
}
```
It's a no-op for primitive keys. Worth a `map<struct<...>, ...>` case
alongside it.
##########
table/scanner.go:
##########
@@ -896,7 +927,7 @@ func (scan *Scan) PlanFiles(ctx context.Context)
([]FileScanTask, error) {
func (scan *Scan) planFilesLocal(ctx context.Context, acc
*scanMetricsAccumulator, schema *iceberg.Schema) (results []FileScanTask, err
error) {
defer func() {
if err == nil {
- scan.closePlanIO()
+ _ = scan.closePlanIO()
Review Comment:
`closePlanIO` now returns an error, but this success defer discards it, so
if the plan IO fails to close cleanly the caller still gets a nil error back.
For vended creds the release is best-effort anyway, but the same discard would
hide a genuine `io.IO.Close()` failure.
I'd fold it into the named return:
```go
defer func() {
if err == nil {
err = scan.closePlanIO()
}
}()
```
##########
catalog/rest/fetch_scan_tasks_validation.go:
##########
@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package rest
+
+import (
+ "bytes"
+ "encoding/json"
+ "fmt"
+)
+
+// validatePlanningTaskEnvelope mirrors the Java response validation for the
+// status-discriminated planning responses. Empty arrays are still present on
+// the wire, so inspect the raw JSON instead of relying only on decoded slice
+// lengths when rejecting task fields before planning completes.
+func validatePlanningTaskEnvelope(data []byte, status PlanStatus, tasks
ScanTasks, endpoint string) error {
+ var fields map[string]json.RawMessage
+ if err := json.Unmarshal(data, &fields); err != nil {
+ return err
+ }
+
+ if status != PlanStatusCompleted {
+ for _, name := range []string{"plan-tasks", "file-scan-tasks"} {
+ raw, ok := fields[name]
+ if ok && !isJSONNull(raw) {
+ return fmt.Errorf("%w: %s response includes %s
for status %q", ErrRESTError, endpoint, name, status)
+ }
+ }
+ if len(tasks.DeleteFiles) > 0 {
Review Comment:
Small inconsistency: for `plan-tasks`/`file-scan-tasks` we reject a
present-but-non-null field via the raw JSON, but for delete-files we only check
the decoded `len(tasks.DeleteFiles) > 0`, so
`{"status":"submitted","delete-files":[]}` slips past the pre-completion guard.
I'd add a parallel raw-JSON check so all three are treated the same:
```go
if raw, ok := fields["delete-files"]; ok && !isJSONNull(raw) {
return fmt.Errorf("%w: %s response includes delete-files for status %q",
ErrRESTError, endpoint, status)
}
```
##########
table/scan_planning_test.go:
##########
@@ -279,6 +280,45 @@ func
TestScanPlanningLocalClosesPreviousPlanIOAfterSuccess(t *testing.T) {
assert.Equal(t, 1, pio.closeCalls)
}
+func TestScanCloseReleasesPlanIO(t *testing.T) {
+ t.Parallel()
+
+ pio := &countingPlanIO{}
+ scan := &Scan{
+ planner: &fakeScanPlanner{result: ScanPlanningResult{IO:
pio}, supports: true},
+ planningMode: ScanPlanningRemote,
+ }
+
+ _, err := scan.PlanFiles(context.Background())
+ require.NoError(t, err)
+ require.NoError(t, scan.Close())
+ require.NoError(t, scan.Close())
+
+ assert.Equal(t, 1, pio.closeCalls)
+ assert.Nil(t, scan.planIO)
+
+ _, err = scan.PlanFiles(context.Background())
+ require.ErrorIs(t, err, ErrInvalidOperation)
+}
+
+func TestScanCloseWaitsForActiveReadTasks(t *testing.T) {
Review Comment:
The name says this waits for active `ReadTasks`, but `Close()` is called on
the same goroutine as `ReadTasks` and before the iterator is drained, so it
really only checks that the reader lease keeps the IO open until consumption
finishes. The dangerous case, `Close()` racing an iterator running on a
separate goroutine, isn't exercised, and this would still pass even if the
ref-counting had a race.
I'd add a parallel subtest that calls `Close()` from a background goroutine
while the main goroutine consumes records (with a channel sync point), run
under `-race`. Also this is the one lifecycle test missing `t.Parallel()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
