Revanth14 commented on code in PR #1324:
URL: https://github.com/apache/iceberg-go/pull/1324#discussion_r3496552822
##########
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:
I’m deferring the granular `error.type` split here as well.
For this PR, `FetchScanTasks` keeps the low-level status-based mapping. The
plan-task-vs-table distinction should land together with the polling/fanout
layer that consumes it, so we can add the right `ErrNoSuchPlanTask` contract
and test retry/abort behavior against the exact spec/OpenAPI `error.type`
values.
For now this PR stays focused on wiring the low-level client method and
preserving the existing REST status override style.
--
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]
