laskoviymishka commented on code in PR #1324:
URL: https://github.com/apache/iceberg-go/pull/1324#discussion_r3539853304
##########
catalog/rest/scan_planning.go:
##########
@@ -79,30 +125,125 @@ 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.
+// PlanTableScan submits a scan plan. A completed or submitted plan returns
+// (resp, nil); callers branch on resp.Status for the plan-id (completed) vs
+// poll (submitted) distinction. A failed plan returns a zero response and a
+// non-nil *PlanFailedError (errors.Is(err, ErrPlanFailed)) so the if-err idiom
+// does not mistake a failure for success; the server detail rides on the
error.
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
+ }
+
+ resp, err := doPost[PlanTableScanRequest, PlanTableScanResponse](
+ ctx, r.baseURI, path, req, r.cl,
+ map[int]error{http.StatusNotFound: catalog.ErrNoSuchTable},
withHeaders(headers))
+ if err != nil {
+ return PlanTableScanResponse{}, err
+ }
+ if resp.Status == PlanStatusFailed {
+ return PlanTableScanResponse{}, &PlanFailedError{Detail:
resp.Error}
+ }
+
+ return resp, nil
}
// 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.
+//
+// completed and submitted return (resp, nil) and callers branch on resp.Status
+// (done vs poll again). The terminal failure states surface as errors so an
+// if-err poll loop cannot mistake them for an empty scan: failed returns a
+// *PlanFailedError (errors.Is(err, ErrPlanFailed)) and cancelled returns
+// ErrPlanCancelled. A 404 maps to ErrPlanExpired (the plan-id the server
forgot).
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
+ }
+
+ resp, err := doGet[FetchPlanningResultResponse](
+ ctx, r.baseURI, path, r.cl,
+ map[int]error{http.StatusNotFound: ErrPlanExpired},
withHeaders(headers))
Review Comment:
This is the 404 split from last round, still here — and zeroshade re-flagged
it too, so I'd treat it as the merge blocker for this pass.
The problem with mapping every 404 to `ErrPlanExpired` (and
`PlanTableScan`/`FetchScanTasks` mapping theirs to `ErrNoSuchTable`) is that
`error.type` never gets consulted, so a table-gone 404 is indistinguishable
from an expired plan-id. `WaitForPlan` is the consumer, and once it's real it
needs `errors.Is(err, ErrPlanExpired)` to mean "retry" and a table-gone to mean
"abort" — with the collapse baked in here, the poll loop spins forever where
Java and PyIceberg terminate. The comment defers this to the polling layer, but
this method is the only place with the `error.type`, so the split has to happen
here, not downstream.
zeroshade already laid out the full type enumeration on the other thread
(`NoSuchPlanIdException` vs `NoSuchTableException`/`NoSuchNamespaceException`,
plus the distinct `NoSuchPlanTask` sentinel for `/tasks`) — I'd follow that
rather than restate it. Concretely, read `errorResponse.Type` in `handleNon200`
and key the override on it: `NoSuchTableException`/`NoSuchNamespaceException` →
`catalog.ErrNoSuchTable`, and keep `ErrPlanExpired` for
`NoSuchPlanIdException`. Once that's in for all three call sites I'm happy to
approve.
##########
catalog/rest/scan_planning.go:
##########
@@ -180,8 +378,8 @@ type CompletedPlanningResult struct {
// wire type and the seam stay in agreement.
type PlanTableScanRequest struct {
// IdempotencyKey is sent as the Idempotency-Key header, not in the
JSON body.
- // If set, it must be a UUID string; nil lets the implementation choose
a
- // safe retry key.
+ // If set, it must be a UUIDv7 string (RFC 9562); nil lets the
implementation
+ // generate a safe UUIDv7 retry key.
Review Comment:
Calling this a "retry key" is a bit of a footgun given the implementation
generates a fresh UUIDv7 on every invocation — so a caller who passes nil and
retries on a transient error sends a different key each attempt, which silently
defeats the server's dedup. And since the generated key isn't returned, they
can't reuse it either.
I'd reword to something like "a new key is generated per call; pass an
explicit key to get idempotency across retries." wdyt about also returning the
generated key so callers can reuse it?
##########
catalog/rest/rest.go:
##########
@@ -222,10 +228,17 @@ type sessionTransport struct {
const emptyStringHash =
"e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
func (s *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) {
+ // A session default is applied unless the request already carries that
+ // header (a per-request override of any default, not just Content-Type
+ // wins) or explicitly opted out of it via withSuppressedHeaders
(carried on
+ // the context as an explicit set, never inferred from header values).
+ suppressed := suppressedHeadersFrom(r.Context())
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") != "" {
+ ck := http.CanonicalHeaderKey(k)
+ if _, ok := r.Header[ck]; ok {
Review Comment:
This is the right general semantics for the suppression rework, but widening
the skip from Content-Type to all default headers quietly changes how
Authorization behaves — a future `withHeaders` carrying an Authorization value
would now suppress the session default. It's safe today only because
`authManager` runs after this loop, which is a pretty fragile ordering
dependency to leave implicit.
I'd add a one-line comment noting that ordering is load-bearing, plus a
small regression test asserting `authManager.AuthHeader()` still wins over a
`withHeaders`-supplied Authorization.
##########
catalog/rest/scan_planning_test.go:
##########
@@ -94,3 +186,385 @@ 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)
+ parsed, err := uuid.Parse(got)
+ require.NoError(t, err)
+ // The spec pins the generated key to UUIDv7.
+ assert.Equal(t, 7, int(parsed.Version()))
+ 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()
+
+ for _, tc := range []struct {
+ name string
+ key string
+ }{
+ {"not a uuid", "not-a-uuid"},
+ // Valid UUID, but v1 not v7 — the spec pins the header to
UUIDv7.
+ {"valid uuid wrong version",
"11111111-1111-1111-1111-111111111111"},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ t.Parallel()
+
+ key := tc.key
+ cat := &Catalog{endpoints:
newEndpointSet([]endpoint{endpointPlanTableScan})}
+
+ _, err := cat.PlanTableScan(context.Background(),
table.Identifier{"db", "tbl"}, PlanTableScanRequest{
+ IdempotencyKey: &key,
+ })
+ 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 := "0190b6c5-1c3d-7000-8000-000000000001"
+ 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"}`),
+ })
+ if tc.wantError != "" {
+ // A failed plan returns a *PlanFailedError and
a zero resp; the
+ // detail rides on the error.
+ require.ErrorIs(t, err, ErrPlanFailed)
+ var pfe *PlanFailedError
+ require.ErrorAs(t, err, &pfe)
+ require.NotNil(t, pfe.Detail)
+ assert.Equal(t, tc.wantError,
pfe.Detail.Message)
+ assert.Equal(t, PlanTableScanResponse{}, resp)
+
+ return
+ }
+
+ 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)
+ }
+ })
+ }
+}
+
+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 TestFetchPlanningResultResponseAcceptsCancelled(t *testing.T) {
+ t.Parallel()
+
+ // cancelled is a valid poll result (unlike PlanTableScanResponse, which
+ // rejects it). Pin it so a refactor routing it into the default error
case
+ // is caught.
+ var resp FetchPlanningResultResponse
+ require.NoError(t, json.Unmarshal([]byte(`{"status":"cancelled"}`),
&resp))
+ assert.Equal(t, PlanStatusCancelled, resp.Status)
+}
+
+func TestFetchPlanningResultStatusArms(t *testing.T) {
+ t.Parallel()
+
+ t.Run("cancelled returns ErrPlanCancelled", func(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) {
+ _, err :=
w.Write([]byte(`{"status":"cancelled"}`))
+ require.NoError(t, err)
+ })
+ })
+
+ _, err := cat.FetchPlanningResult(context.Background(),
table.Identifier{"db", "tbl"}, "plan-123", FetchPlanningResultOptions{})
+ require.ErrorIs(t, err, ErrPlanCancelled)
+ })
+
+ t.Run("failed returns PlanFailedError", func(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) {
+ _, err :=
w.Write([]byte(`{"status":"failed","error":{"message":"boom","type":"ServerError","code":500}}`))
+ require.NoError(t, err)
+ })
+ })
+
+ _, err := cat.FetchPlanningResult(context.Background(),
table.Identifier{"db", "tbl"}, "plan-123", FetchPlanningResultOptions{})
+ require.ErrorIs(t, err, ErrPlanFailed)
+ var pfe *PlanFailedError
+ require.ErrorAs(t, err, &pfe)
+ assert.Equal(t, "boom", pfe.Detail.Message)
Review Comment:
`pfe.Detail` is dereferenced here without a nil guard — if `Detail` were
ever nil this panics and takes down the test process rather than failing
cleanly. The `TestPlanTableScanRequest` sibling guards it with
`require.NotNil(t, pfe.Detail)`; worth mirroring that here.
--
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]
