zeroshade commented on code in PR #1324:
URL: https://github.com/apache/iceberg-go/pull/1324#discussion_r3521440307
##########
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 still makes every 404 from `FetchPlanningResult` indistinguishable from
an expired plan, which conflicts with the documented follow-up contract at
lines 41-45. A dropped table or namespace will now surface as `ErrPlanExpired`
instead of `catalog.ErrNoSuchTable` / `catalog.ErrNoSuchNamespace`, so the
polling layer cannot make the required distinction. Please decode the REST
error model's `error.type` on 404 and map `NoSuchPlanIdException` to
`ErrPlanExpired`, `NoSuchTableException` to `catalog.ErrNoSuchTable`, and
`NoSuchNamespaceException` to `catalog.ErrNoSuchNamespace`, with httptest
coverage for each.
##########
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)
Review Comment:
This test only covers an empty-body 404 and therefore pins the lossy
`ErrPlanExpired` behavior. Please switch these 404 cases to typed REST error
bodies for plan-id/table/namespace and add corresponding `FetchScanTasks` 404
coverage, including the `NoSuchPlanTaskException` case.
##########
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))
+ if err != nil {
+ return FetchPlanningResultResponse{}, err
+ }
+ switch resp.Status {
+ case PlanStatusFailed:
+ return FetchPlanningResultResponse{}, &PlanFailedError{Detail:
resp.Error}
+ case PlanStatusCancelled:
+ return FetchPlanningResultResponse{}, ErrPlanCancelled
+ }
+
+ return resp, nil
}
// 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:
This 404 override hides `NoSuchPlanTaskException` as
`catalog.ErrNoSuchTable`. For the follow-up task-fetch layer to distinguish a
missing table from an expired/unknown plan-task handle, please add a distinct
`ErrNoSuchPlanTask` sentinel and map the 404 by REST `error.type` instead of
status alone, with matching tests.
--
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]
