laskoviymishka commented on code in PR #1324:
URL: https://github.com/apache/iceberg-go/pull/1324#discussion_r3491578665
##########
catalog/rest/scan_planning.go:
##########
@@ -113,6 +189,48 @@ func (r *Catalog) WaitForPlan(ctx context.Context, ident
table.Identifier, planI
return CompletedPlanningResult{}, fmt.Errorf("%w: wait for plan",
iceberg.ErrNotImplemented)
}
+func (r *Catalog) scanPlanningPath(ep endpoint, ident table.Identifier, extra
...string) ([]string, error) {
+ ns, tbl, err := r.splitIdentForPath(ident)
+ if err != nil {
+ return nil, err
+ }
+
+ return ep.reqPath(append([]string{ns, tbl}, extra...)...)
+}
+
+func scanPlanningHeaders(idempotencyKey, accessDelegation *string,
includeIdempotency bool) (map[string]string, error) {
+ headers := make(map[string]string, 2)
+ if includeIdempotency {
+ key, err := idempotencyHeaderValue(idempotencyKey)
+ if err != nil {
+ return nil, err
+ }
+ headers[headerIdempotencyKey] = key
+ }
+ if accessDelegation != nil {
+ headers[headerIcebergAccessDelegation] = *accessDelegation
+ }
+ if len(headers) == 0 {
+ return nil, nil
+ }
+
+ return headers, nil
+}
+
+func idempotencyHeaderValue(idempotencyKey *string) (string, error) {
+ if idempotencyKey == nil {
+ // Plan and task POSTs always send an idempotency key. There is
no
+ // transport retry here, so nil means a fresh key for this call.
+ return uuid.NewString(), nil
Review Comment:
`uuid.NewString()` gives us a UUIDv4, but the spec pins this header to
"UUIDv7 in string form (RFC 9562)" and Java generates it via
`UUIDUtil.generateUuidV7()`. Servers that key the idempotency window off the
embedded timestamp will treat our v4 keys as undefined, so cross-client dedup
against a Java-based server is broken.
`google/uuid` v1.6.0 is already in go.mod and exposes `uuid.NewV7()` — I'd
swap to that and propagate the error it returns.
While we're here: Java only sends idempotency headers when the server
advertises a key lifetime (`config.idempotencyKeyLifetime()`), whereas we send
one on every plan and task POST. The header's optional, so this isn't wrong,
but strict servers could reject the unsolicited header. Worth a follow-up to
gate it on config, even if not in this PR.
##########
catalog/rest/scan_planning.go:
##########
@@ -80,29 +88,97 @@ func (r *Catalog) PlanFiles(ctx context.Context, req
table.ScanPlanningRequest)
// --- Low-level client methods -----------------------------------------------
// PlanTableScan submits a scan plan. The result is either completed inline,
-// submitted (returns a plan-id to poll), or failed.
+// submitted (returns a plan-id to poll), or failed. A failed planning response
+// decodes as (resp, nil); callers must branch on resp.Status.
func (r *Catalog) PlanTableScan(ctx context.Context, ident table.Identifier,
req PlanTableScanRequest) (PlanTableScanResponse, error) {
- return PlanTableScanResponse{}, fmt.Errorf("%w: plan table scan",
iceberg.ErrNotImplemented)
+ if err := r.endpoints.check(endpointPlanTableScan); err != nil {
+ return PlanTableScanResponse{}, err
+ }
+
+ path, err := r.scanPlanningPath(endpointPlanTableScan, ident)
+ if err != nil {
+ return PlanTableScanResponse{}, err
+ }
+
+ headers, err := scanPlanningHeaders(req.IdempotencyKey,
req.AccessDelegation, true)
+ if err != nil {
+ return PlanTableScanResponse{}, err
+ }
+
+ return doPost[PlanTableScanRequest, PlanTableScanResponse](
+ ctx, r.baseURI, path, req, r.cl,
+ map[int]error{http.StatusNotFound: catalog.ErrNoSuchTable},
withHeaders(headers))
}
// FetchPlanningResult polls a previously submitted plan. opts.AccessDelegation
// is sent as the X-Iceberg-Access-Delegation header so an async poll can still
// receive plan-scoped storage credentials: the spec defines data-access on
this
// endpoint, and the completed-async result is where those credentials are
vended.
func (r *Catalog) FetchPlanningResult(ctx context.Context, ident
table.Identifier, planID string, opts FetchPlanningResultOptions)
(FetchPlanningResultResponse, error) {
- return FetchPlanningResultResponse{}, fmt.Errorf("%w: fetch planning
result", iceberg.ErrNotImplemented)
+ if err := r.endpoints.check(endpointFetchPlanResult); err != nil {
+ return FetchPlanningResultResponse{}, err
+ }
+
+ path, err := r.scanPlanningPath(endpointFetchPlanResult, ident, planID)
+ if err != nil {
+ return FetchPlanningResultResponse{}, err
+ }
+
+ headers, err := scanPlanningHeaders(nil, opts.AccessDelegation, false)
+ if err != nil {
+ return FetchPlanningResultResponse{}, err
+ }
+
+ return doGet[FetchPlanningResultResponse](
+ ctx, r.baseURI, path, r.cl,
+ map[int]error{http.StatusNotFound: ErrPlanExpired},
withHeaders(headers))
Review Comment:
This maps every 404 to `ErrPlanExpired`, but the spec's
`fetchPlanningResult` 404 can be `NoSuchPlanIdException`,
`NoSuchTableException`, or `NoSuchNamespaceException`. A dropped table comes
back as "plan expired," which is the one distinction `WaitForPlan` actually
needs — "plan expired, retry with a new plan" vs. "table's gone, abort."
I'd settle this before the polling layer lands on top of it, since fixing it
later means reworking the retry contract. The clean version is inspecting the
body's `error.type` before mapping — either a pre-decode pass or a typed
override hook in `handleNon200` that can split on body content. wdyt?
##########
catalog/rest/scan_planning.go:
##########
@@ -80,29 +88,97 @@ func (r *Catalog) PlanFiles(ctx context.Context, req
table.ScanPlanningRequest)
// --- Low-level client methods -----------------------------------------------
// PlanTableScan submits a scan plan. The result is either completed inline,
-// submitted (returns a plan-id to poll), or failed.
+// submitted (returns a plan-id to poll), or failed. A failed planning response
+// decodes as (resp, nil); callers must branch on resp.Status.
func (r *Catalog) PlanTableScan(ctx context.Context, ident table.Identifier,
req PlanTableScanRequest) (PlanTableScanResponse, error) {
Review Comment:
I think this contract is going to bite callers. Returning `(resp, nil)` for
a `failed` plan means anyone using the `if err != nil` idiom treats a failure
as success, then reads `resp.PlanID` (nil for failed) and panics. The doc
comment is honest about it, but the comment is the only thing standing between
this and a nil-deref downstream.
I'd lean toward returning an error on `failed` — something like
`fmt.Errorf("%w: plan failed: %s", ErrRESTError, resp.Error.Message)`, or a
typed error wrapping the `PlanningError` so callers can still get at the
structured detail. The `completed`/`submitted` split stays in the response.
Either way I'd settle it here rather than after `WaitForPlan` and the
scanner are written against the current shape — and the same note needs to go
on `FetchPlanningResult`, which has the identical `(resp, nil)` contract and is
exactly what the poller will loop on. Thoughts?
##########
catalog/rest/rest.go:
##########
@@ -287,7 +295,48 @@ func (s *sessionTransport) RoundTrip(r *http.Request)
(*http.Response, error) {
return s.RoundTripper.RoundTrip(r)
}
-func do[T any](ctx context.Context, method string, baseURI *url.URL, path
[]string, cl *http.Client, override map[int]error, allowNoContent bool) (ret T,
err error) {
+// reqConfig holds the optional per-request knobs shared by do and doPost:
+// headers to set, session-default headers to suppress, and whether an HTTP 204
+// is a valid empty result.
+type reqConfig struct {
+ headers map[string]string
+ suppressHeaders []string
+ allowNoContent bool
+}
+
+type reqOption func(*reqConfig)
+
+// withHeaders sets per-request headers. They take precedence over session
+// defaults because sessionTransport.RoundTrip skips any default whose key is
+// already present on the request.
+func withHeaders(headers map[string]string) reqOption {
+ return func(c *reqConfig) { c.headers = headers }
+}
+
+// withSuppressedHeaders prevents the named session-default headers from being
+// sent on this request. See suppressRequestHeaders for the mechanism.
+func withSuppressedHeaders(names ...string) reqOption {
Review Comment:
Both options assign rather than merge, so two `withSuppressedHeaders(...)`
on one call silently drops all but the last. Not a bug today since each is
passed once, but it's a trap for the next caller who reasonably assumes options
compose.
I'd make `withSuppressedHeaders` append (`c.suppressHeaders =
append(c.suppressHeaders, names...)`) and merge maps in `withHeaders`.
##########
catalog/rest/rest.go:
##########
@@ -389,6 +440,24 @@ func doPostAllowNoContent[Payload, Result any](ctx
context.Context, baseURI *url
return ret, err
}
+func setRequestHeaders(req *http.Request, headers map[string]string) {
+ for k, v := range headers {
+ req.Header.Set(k, v)
+ }
+}
+
+// suppressRequestHeaders prevents the named session-default headers from being
+// sent on this request. It writes a present-but-empty header entry rather than
+// calling Header.Del: sessionTransport.RoundTrip decides whether to apply a
+// default by testing key *presence* (not value), so a present-but-empty key
+// both blocks the default and emits nothing, whereas deleting the key would
let
+// the default reappear. Keep RoundTrip's presence check in sync with this.
+func suppressRequestHeaders(req *http.Request, headers []string) {
+ for _, h := range headers {
+ req.Header[http.CanonicalHeaderKey(h)] = nil
Review Comment:
This leans on net/http not writing a header whose value slice is nil —
correct today, but it's an internal wire invariant, not a documented contract,
and the comment calls it "present-but-empty" when it's actually nil. If a
runtime change ever altered that, the suppression breaks silently and
`X-Iceberg-Access-Delegation` reappears on cancel and task-fetch.
Two ways to firm this up. The lighter one is `req.Header.Set(h, "")` plus a
comment — present, empty string, still skipped by the presence check, no
reliance on nil-slice behavior. The sturdier one is moving suppression into
`reqConfig` as an explicit suppress-set that `RoundTrip` consults directly,
since `RoundTrip` already owns session-default logic — then there's no
two-sided "keep in sync" invariant at all. I'd lean toward the latter given a
follow-up PR will be poking at this. Thoughts?
##########
catalog/rest/scan_planning.go:
##########
@@ -80,29 +88,97 @@ func (r *Catalog) PlanFiles(ctx context.Context, req
table.ScanPlanningRequest)
// --- Low-level client methods -----------------------------------------------
// PlanTableScan submits a scan plan. The result is either completed inline,
-// submitted (returns a plan-id to poll), or failed.
+// submitted (returns a plan-id to poll), or failed. A failed planning response
+// decodes as (resp, nil); callers must branch on resp.Status.
func (r *Catalog) PlanTableScan(ctx context.Context, ident table.Identifier,
req PlanTableScanRequest) (PlanTableScanResponse, error) {
- return PlanTableScanResponse{}, fmt.Errorf("%w: plan table scan",
iceberg.ErrNotImplemented)
+ if err := r.endpoints.check(endpointPlanTableScan); err != nil {
+ return PlanTableScanResponse{}, err
+ }
+
+ path, err := r.scanPlanningPath(endpointPlanTableScan, ident)
+ if err != nil {
+ return PlanTableScanResponse{}, err
+ }
+
+ headers, err := scanPlanningHeaders(req.IdempotencyKey,
req.AccessDelegation, true)
+ if err != nil {
+ return PlanTableScanResponse{}, err
+ }
+
+ return doPost[PlanTableScanRequest, PlanTableScanResponse](
+ ctx, r.baseURI, path, req, r.cl,
+ map[int]error{http.StatusNotFound: catalog.ErrNoSuchTable},
withHeaders(headers))
}
// FetchPlanningResult polls a previously submitted plan. opts.AccessDelegation
// is sent as the X-Iceberg-Access-Delegation header so an async poll can still
// receive plan-scoped storage credentials: the spec defines data-access on
this
// endpoint, and the completed-async result is where those credentials are
vended.
func (r *Catalog) FetchPlanningResult(ctx context.Context, ident
table.Identifier, planID string, opts FetchPlanningResultOptions)
(FetchPlanningResultResponse, error) {
- return FetchPlanningResultResponse{}, fmt.Errorf("%w: fetch planning
result", iceberg.ErrNotImplemented)
+ if err := r.endpoints.check(endpointFetchPlanResult); err != nil {
+ return FetchPlanningResultResponse{}, err
+ }
+
+ path, err := r.scanPlanningPath(endpointFetchPlanResult, ident, planID)
+ if err != nil {
+ return FetchPlanningResultResponse{}, err
+ }
+
+ headers, err := scanPlanningHeaders(nil, opts.AccessDelegation, false)
+ if err != nil {
+ return FetchPlanningResultResponse{}, err
+ }
+
+ return doGet[FetchPlanningResultResponse](
+ ctx, r.baseURI, path, r.cl,
+ map[int]error{http.StatusNotFound: ErrPlanExpired},
withHeaders(headers))
}
// CancelPlanning cancels a server-side plan. Callers should cancel on context
-// cancellation using a detached context with a short timeout.
+// cancellation using a detached context with a short timeout. The spec
supports
+// idempotency and access-delegation headers on cancel; this low-level method
+// deliberately defers those until a cancel options type is added, and
suppresses
+// the session-default access-delegation header (cancel vends no credentials).
A
+// 404 (already-expired or unknown plan) is not special-cased: cancel is
+// best-effort, so the generic REST error is acceptable.
func (r *Catalog) CancelPlanning(ctx context.Context, ident table.Identifier,
planID string) error {
- return fmt.Errorf("%w: cancel planning", iceberg.ErrNotImplemented)
+ if err := r.endpoints.check(endpointCancelPlanning); err != nil {
+ return err
+ }
+
+ path, err := r.scanPlanningPath(endpointCancelPlanning, ident, planID)
+ if err != nil {
+ return err
+ }
+
+ _, err = doDelete[struct{}](
+ ctx, r.baseURI, path, r.cl, nil,
+ withSuppressedHeaders(headerIcebergAccessDelegation))
+
+ return err
}
// FetchScanTasks fetches the scan tasks for a plan-task handle returned by a
// completed plan.
func (r *Catalog) FetchScanTasks(ctx context.Context, ident table.Identifier,
req FetchScanTasksRequest) (FetchScanTasksResponse, error) {
- return FetchScanTasksResponse{}, fmt.Errorf("%w: fetch scan tasks",
iceberg.ErrNotImplemented)
+ if err := r.endpoints.check(endpointFetchScanTasks); err != nil {
+ return FetchScanTasksResponse{}, err
+ }
+
+ path, err := r.scanPlanningPath(endpointFetchScanTasks, ident)
+ if err != nil {
+ return FetchScanTasksResponse{}, err
+ }
+
+ headers, err := scanPlanningHeaders(req.IdempotencyKey, nil, true)
+ if err != nil {
+ return FetchScanTasksResponse{}, err
+ }
+
+ return doPost[FetchScanTasksRequest, FetchScanTasksResponse](
+ ctx, r.baseURI, path, req, r.cl,
+ map[int]error{http.StatusNotFound: catalog.ErrNoSuchTable},
Review Comment:
Same shape as the `FetchPlanningResult` 404, the other direction: here every
404 becomes `catalog.ErrNoSuchTable`, but the spec's `fetchScanTasks` 404 also
covers `NoSuchPlanTaskException`. A caller fanning out over plan-task handles
will read `ErrNoSuchTable` and conclude the table vanished when really a task
handle expired. If we go the `error.type` inspection route for the
`FetchPlanningResult` 404, the same hook covers this — and we'd want an
`ErrNoSuchPlanTask` sentinel to map onto.
##########
catalog/rest/rest.go:
##########
@@ -223,9 +229,11 @@ const emptyStringHash =
"e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991
func (s *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) {
for k, v := range s.defaultHeaders {
- // Skip Content-Type if it's already set in the request
- // to avoid duplicate headers (e.g., when using PostForm)
- if http.CanonicalHeaderKey(k) == "Content-Type" &&
r.Header.Get("Content-Type") != "" {
+ // Per-request headers override session defaults. The check is
on key
+ // *presence*, not value, so suppressRequestHeaders can opt out
of a
+ // default by setting a present-but-empty entry. Keep this in
sync with
+ // suppressRequestHeaders.
+ if _, ok := r.Header[http.CanonicalHeaderKey(k)]; ok {
Review Comment:
This broadens the old behavior: it used to skip only Content-Type, now it
skips any session default whose key is already present. That's almost certainly
what we want, but the oauth2 Content-Type correctness that the old special-case
guaranteed is now a side-effect of the general rule. I'd just note in the
comment that the rule applies to any per-request header sharing a key with a
session default, so the next person doesn't re-add a Content-Type special case.
##########
catalog/rest/scan_planning.go:
##########
@@ -54,21 +59,24 @@ var ErrPlanExpired = fmt.Errorf("%w: scan plan expired",
ErrRESTError)
// SupportsPlanTableScan reports whether the server advertised the synchronous
// plan endpoint.
func (r *Catalog) SupportsPlanTableScan() bool {
- return false
+ return r.endpoints.contains(endpointPlanTableScan)
}
// SupportsFullRemoteScanPlanning reports whether the server advertised all
four
// scan-planning endpoints (plan, fetch-result, cancel, fetch-tasks).
func (r *Catalog) SupportsFullRemoteScanPlanning() bool {
- return false
+ 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; backed by the split capability checks above.
func (r *Catalog) SupportsRemoteScanPlanning() bool {
- return false
+ return r.SupportsFullRemoteScanPlanning()
Review Comment:
This requires all four endpoints, where Java's minimal check only needs the
submit endpoint for the sync-only case (`RESTSessionCatalog.java:641`). A
server that advertises only `plan` — completed-inline, no polling — passes Java
but gets rejected here and falls back to client-side planning.
The doc comment ("complete a plan end-to-end") is consistent with requiring
all four, so this isn't a contradiction so much as a semantics call: is
`SupportsRemoteScanPlanning` "can I do the full async dance" or "can the server
plan at all"? I'd lean toward the latter to match Java — return
`SupportsPlanTableScan()`, and let callers needing fanout check
`SupportsFullRemoteScanPlanning()` explicitly. wdyt?
##########
catalog/rest/rest_internal_test.go:
##########
@@ -914,22 +914,22 @@ func TestResponseBodyLeak(t *testing.T) {
baseURI, err := url.Parse(srv.URL)
require.NoError(t, err)
- _, err = do[struct{}](context.Background(), http.MethodGet,
baseURI, []string{"test"}, client, nil, false)
+ _, err = do[struct{}](context.Background(), http.MethodGet,
baseURI, []string{"test"}, client, nil)
Review Comment:
The generalized presence-check override in `RoundTrip` is only tested
transitively through the scan-planning end-to-end tests. Since it's the most
consequential behavior change in the PR, I'd add a direct unit test here: build
a `sessionTransport` with a default header, send a request that already sets
that header, call `RoundTrip`, and assert the default wasn't appended. That
pins the invariant independently of the scan-planning callers.
##########
catalog/rest/scan_planning_test.go:
##########
@@ -94,3 +183,298 @@ func TestPlanTableScanResponseRejectsUnknownStatus(t
*testing.T) {
err := json.Unmarshal([]byte(`{"status":"bogus"}`), &resp)
require.ErrorIs(t, err, ErrRESTError)
}
+
+func TestFetchPlanningResultResponseRejectsMalformedResponses(t *testing.T) {
Review Comment:
The `cancelled` arm in `FetchPlanningResultResponse.UnmarshalJSON` is
accepted silently with no test, so a refactor routing `cancelled` into the
`default` error case would slip through — and `cancelled` is specifically valid
for poll results, unlike `PlanTableScanResponse`. A subtest unmarshaling
`{"status":"cancelled"}` asserting NoError + `PlanStatusCancelled` would pin it.
Same gap on the round-trip side: `TestFetchPlanningResultRequest` only
covers `completed`, while `TestPlanTableScanRequest` has a `failed` arm. A
`failed`-status HTTP case here would exercise the validation through the real
client path, not just direct `json.Unmarshal`.
##########
catalog/rest/scan_planning_test.go:
##########
@@ -94,3 +183,298 @@ func TestPlanTableScanResponseRejectsUnknownStatus(t
*testing.T) {
err := json.Unmarshal([]byte(`{"status":"bogus"}`), &resp)
require.ErrorIs(t, err, ErrRESTError)
}
+
+func TestFetchPlanningResultResponseRejectsMalformedResponses(t *testing.T) {
+ t.Parallel()
+
+ t.Run("failed without error", func(t *testing.T) {
+ t.Parallel()
+
+ var resp FetchPlanningResultResponse
+ err := json.Unmarshal([]byte(`{"status":"failed"}`), &resp)
+ require.ErrorIs(t, err, ErrRESTError)
+ })
+
+ t.Run("unknown status", func(t *testing.T) {
+ t.Parallel()
+
+ var resp FetchPlanningResultResponse
+ err := json.Unmarshal([]byte(`{"status":"bogus"}`), &resp)
+ require.ErrorIs(t, err, ErrRESTError)
+ })
+}
+
+func TestPlanTableScanGeneratesIdempotencyKeyAndUsesDefaultAccessDelegation(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) {
+ got := req.Header.Get(headerIdempotencyKey)
+ require.NotEmpty(t, got)
+ _, err := uuid.Parse(got)
+ require.NoError(t, err)
+ assert.Equal(t, []string{defaultAccessDelegation},
req.Header.Values(headerIcebergAccessDelegation))
+
+ _, err =
w.Write([]byte(`{"status":"completed","plan-id":"plan-1"}`))
+ require.NoError(t, err)
+ })
+ })
+
+ _, err := cat.PlanTableScan(context.Background(),
table.Identifier{"db", "tbl"}, PlanTableScanRequest{})
+ require.NoError(t, err)
+}
+
+func TestPlanTableScanRejectsInvalidIdempotencyKey(t *testing.T) {
+ t.Parallel()
+
+ badKey := "not-a-uuid"
+ cat := &Catalog{endpoints:
newEndpointSet([]endpoint{endpointPlanTableScan})}
+
+ _, err := cat.PlanTableScan(context.Background(),
table.Identifier{"db", "tbl"}, PlanTableScanRequest{
+ IdempotencyKey: &badKey,
+ })
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+}
+
+func TestPlanTableScanRequest(t *testing.T) {
+ t.Parallel()
+
+ cases := []struct {
+ name string
+ response string
+ wantStatus PlanStatus
+ wantPlanID *string
+ wantError string
+ }{
+ {
+ name: "completed",
+ response:
`{"status":"completed","plan-id":"plan-1","plan-tasks":["task-1"]}`,
+ wantStatus: PlanStatusCompleted,
+ wantPlanID: stringPtr("plan-1"),
+ },
+ {
+ name: "submitted",
+ response: `{"status":"submitted","plan-id":"plan-2"}`,
+ wantStatus: PlanStatusSubmitted,
+ wantPlanID: stringPtr("plan-2"),
+ },
+ {
+ name: "failed",
+ response:
`{"status":"failed","error":{"message":"boom","type":"ServerError","code":500}}`,
+ wantStatus: PlanStatusFailed,
+ wantError: "boom",
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ t.Parallel()
+
+ idempotencyKey := "11111111-1111-1111-1111-111111111111"
+ accessDelegation := "remote-signing"
+ snapshotID := int64(22)
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointPlanTableScan}, func(mux *http.ServeMux) {
+
mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan", func(w http.ResponseWriter,
req *http.Request) {
+ require.Equal(t, http.MethodPost,
req.Method)
+ assert.Equal(t, idempotencyKey,
req.Header.Get(headerIdempotencyKey))
+ assert.Equal(t,
[]string{accessDelegation}, req.Header.Values(headerIcebergAccessDelegation))
+
+ body, err := io.ReadAll(req.Body)
+ require.NoError(t, err)
+ assert.JSONEq(t, `{
+ "snapshot-id": 22,
+ "select": ["id", "data"],
+ "filter": {"type":
"always-true"}
+ }`, string(body))
+
+ _, err = w.Write([]byte(tc.response))
+ require.NoError(t, err)
+ })
+ })
+
+ resp, err := cat.PlanTableScan(context.Background(),
table.Identifier{"db", "tbl"}, PlanTableScanRequest{
+ IdempotencyKey: &idempotencyKey,
+ AccessDelegation: &accessDelegation,
+ SnapshotID: &snapshotID,
+ Select: []string{"id", "data"},
+ Filter:
json.RawMessage(`{"type":"always-true"}`),
+ })
+ require.NoError(t, err)
+ assert.Equal(t, tc.wantStatus, resp.Status)
+ if tc.wantPlanID != nil {
+ require.NotNil(t, resp.PlanID)
+ assert.Equal(t, *tc.wantPlanID, *resp.PlanID)
+ }
+ if tc.wantError != "" {
+ require.NotNil(t, resp.Error)
+ assert.Equal(t, tc.wantError,
resp.Error.Message)
+ }
+ })
+ }
+}
+
+func TestFetchPlanningResultRequest(t *testing.T) {
+ t.Parallel()
+
+ accessDelegation := "remote-signing"
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointFetchPlanResult}, func(mux *http.ServeMux) {
+ mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan/plan-123",
func(w http.ResponseWriter, req *http.Request) {
+ require.Equal(t, http.MethodGet, req.Method)
+ assert.Equal(t, []string{accessDelegation},
req.Header.Values(headerIcebergAccessDelegation))
+
+ _, err :=
w.Write([]byte(`{"status":"completed","plan-tasks":["task-1"]}`))
+ require.NoError(t, err)
+ })
+ })
+
+ resp, err := cat.FetchPlanningResult(context.Background(),
table.Identifier{"db", "tbl"}, "plan-123", FetchPlanningResultOptions{
+ AccessDelegation: &accessDelegation,
+ })
+ require.NoError(t, err)
+ assert.Equal(t, PlanStatusCompleted, resp.Status)
+ assert.Equal(t, []string{"task-1"}, resp.PlanTasks)
+}
+
+func TestFetchPlanningResultUsesDefaultAccessDelegation(t *testing.T) {
+ t.Parallel()
+
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointFetchPlanResult}, func(mux *http.ServeMux) {
+ mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan/plan-123",
func(w http.ResponseWriter, req *http.Request) {
+ require.Equal(t, http.MethodGet, req.Method)
+ assert.Equal(t, []string{defaultAccessDelegation},
req.Header.Values(headerIcebergAccessDelegation))
+
+ _, err := w.Write([]byte(`{"status":"submitted"}`))
+ require.NoError(t, err)
+ })
+ })
+
+ resp, err := cat.FetchPlanningResult(context.Background(),
table.Identifier{"db", "tbl"}, "plan-123", FetchPlanningResultOptions{})
+ require.NoError(t, err)
+ assert.Equal(t, PlanStatusSubmitted, resp.Status)
+}
+
+func TestFetchPlanningResultMapsNotFoundToPlanExpired(t *testing.T) {
+ t.Parallel()
+
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointFetchPlanResult}, func(mux *http.ServeMux) {
+
mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan/expired-plan", func(w
http.ResponseWriter, req *http.Request) {
+ require.Equal(t, http.MethodGet, req.Method)
+ w.WriteHeader(http.StatusNotFound)
+ })
+ })
+
+ _, err := cat.FetchPlanningResult(context.Background(),
table.Identifier{"db", "tbl"}, "expired-plan", FetchPlanningResultOptions{})
+ require.ErrorIs(t, err, ErrPlanExpired)
+}
+
+func TestCancelPlanningRequest(t *testing.T) {
+ t.Parallel()
+
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointCancelPlanning}, func(mux *http.ServeMux) {
+ mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan/plan-123",
func(w http.ResponseWriter, req *http.Request) {
+ require.Equal(t, http.MethodDelete, req.Method)
+ assert.Empty(t, req.Header.Values(headerIdempotencyKey))
+ assert.Empty(t,
req.Header.Values(headerIcebergAccessDelegation))
+ w.WriteHeader(http.StatusNoContent)
+ })
+ })
+
+ require.NoError(t, cat.CancelPlanning(context.Background(),
table.Identifier{"db", "tbl"}, "plan-123"))
+}
+
+func TestFetchScanTasksRequest(t *testing.T) {
+ t.Parallel()
+
+ idempotencyKey := "22222222-2222-2222-2222-222222222222"
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointFetchScanTasks}, func(mux *http.ServeMux) {
+ mux.HandleFunc("/v1/namespaces/db/tables/tbl/tasks", func(w
http.ResponseWriter, req *http.Request) {
+ require.Equal(t, http.MethodPost, req.Method)
+ assert.Equal(t, idempotencyKey,
req.Header.Get(headerIdempotencyKey))
+ assert.Empty(t,
req.Header.Values(headerIcebergAccessDelegation))
+
+ body, err := io.ReadAll(req.Body)
+ require.NoError(t, err)
+ assert.JSONEq(t, `{"plan-task":"task-1"}`, string(body))
+
+ _, err = w.Write([]byte(`{
+ "plan-tasks": ["child-task"],
+ "file-scan-tasks": [{}],
+ "delete-files": [{}]
+ }`))
+ require.NoError(t, err)
+ })
+ })
+
+ resp, err := cat.FetchScanTasks(context.Background(),
table.Identifier{"db", "tbl"}, FetchScanTasksRequest{
+ IdempotencyKey: &idempotencyKey,
+ PlanTask: "task-1",
+ })
+ require.NoError(t, err)
+ assert.Equal(t, []string{"child-task"}, resp.PlanTasks)
+ assert.Len(t, resp.FileScanTasks, 1)
+ assert.Len(t, resp.DeleteFiles, 1)
+}
+
+func TestScanPlanningEndpointGating(t *testing.T) {
Review Comment:
This runs all four endpoint checks in one body, so a failing `require` on
the first masks the other three. `TestScanPlanningCapabilities` right above it
uses `t.Run` subtests — I'd match that here so each endpoint reports
independently.
--
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]
