Revanth14 commented on code in PR #1494: URL: https://github.com/apache/iceberg-go/pull/1494#discussion_r3609053970
########## catalog/rest/scan_planning_fake_test.go: ########## @@ -0,0 +1,831 @@ +// 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_test + +import ( + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "testing" + "time" + + "github.com/apache/iceberg-go/catalog/rest" + "github.com/apache/iceberg-go/catalog/rest/internal/planfake" + "github.com/apache/iceberg-go/table" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestPlanFakeCapabilityDiscovery(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + endpoints []string + wantPlan bool + wantFullPlanning bool + }{ + { + name: "all planning endpoints", + endpoints: []string{ + planfake.PlanTableScanEndpoint, + planfake.FetchPlanningResultEndpoint, + planfake.CancelPlanningEndpoint, + planfake.FetchScanTasksEndpoint, + }, + wantPlan: true, + wantFullPlanning: true, + }, + { + name: "plan only", + endpoints: []string{planfake.PlanTableScanEndpoint}, + wantPlan: true, + wantFullPlanning: false, + }, + { + name: "missing plan submission", + endpoints: []string{ + planfake.FetchPlanningResultEndpoint, + planfake.CancelPlanningEndpoint, + planfake.FetchScanTasksEndpoint, + }, + wantPlan: false, + wantFullPlanning: false, + }, + { + name: "missing result fetch", + endpoints: []string{ + planfake.PlanTableScanEndpoint, + planfake.CancelPlanningEndpoint, + planfake.FetchScanTasksEndpoint, + }, + wantPlan: true, + wantFullPlanning: false, + }, + { + name: "missing cancellation", + endpoints: []string{ + planfake.PlanTableScanEndpoint, + planfake.FetchPlanningResultEndpoint, + planfake.FetchScanTasksEndpoint, + }, + wantPlan: true, + wantFullPlanning: false, + }, + { + name: "missing task fetch", + endpoints: []string{ + planfake.PlanTableScanEndpoint, + planfake.FetchPlanningResultEndpoint, + planfake.CancelPlanningEndpoint, + }, + wantPlan: true, + wantFullPlanning: false, + }, + { + name: "no planning endpoints", + endpoints: []string{"GET /v1/{prefix}/namespaces"}, + wantPlan: false, + wantFullPlanning: false, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + srv := planfake.New(t, planfake.Scenario{ + ConfigResponse: planfake.ConfigResponse(test.endpoints...), + }) + catalog, err := rest.NewCatalog(t.Context(), "rest", srv.URL()) + require.NoError(t, err) + + assert.Equal(t, test.wantPlan, catalog.SupportsPlanTableScan()) + assert.Equal(t, test.wantFullPlanning, catalog.SupportsFullRemoteScanPlanning()) + + requests := srv.Requests() + require.Len(t, requests, 1) + assert.Equal(t, http.MethodGet, requests[0].Method) + assert.Equal(t, "/v1/config", requests[0].Path) + }) + } +} + +func TestPlanFakeSynchronousPlanningCapturesRequest(t *testing.T) { + t.Parallel() + + srv := planfake.New(t, planfake.Scenario{ + ConfigResponse: fullPlanningConfigResponse(), + ExpectedTarget: &planfake.ExpectedTarget{ + Prefix: "catalog", Namespace: "org\x1fanalytics", Table: "events", + }, + PlanResponse: planFakeJSONResponse(canonicalCompletedPlanJSON), + }) + catalog, err := rest.NewCatalog(t.Context(), "rest", srv.URL(), rest.WithPrefix("catalog")) + require.NoError(t, err) + + idempotencyKey := "0190b6c5-1c3d-7000-8000-000000000011" + accessDelegation := "remote-signing" + snapshotID := int64(123) + minimumRows := int64(50) + caseSensitive := false + useSnapshotSchema := true + response, err := catalog.PlanTableScan(t.Context(), table.Identifier{"org", "analytics", "events"}, rest.PlanTableScanRequest{ + IdempotencyKey: &idempotencyKey, + AccessDelegation: &accessDelegation, + SnapshotID: &snapshotID, + Select: []string{"id", "payload"}, + Filter: json.RawMessage(`{"type":"eq","term":"id","value":34}`), + MinRowsRequested: &minimumRows, + CaseSensitive: &caseSensitive, + UseSnapshotSchema: &useSnapshotSchema, + StatsFields: []string{"id", "payload"}, + }) + require.NoError(t, err) + + assert.Equal(t, rest.PlanStatusCompleted, response.Status) + require.NotNil(t, response.PlanID) + assert.Equal(t, "plan-sync", *response.PlanID) + assert.Equal(t, []string{"inline-child"}, response.PlanTasks) + assert.Len(t, response.FileScanTasks, 1) + assert.Len(t, response.DeleteFiles, 1) + require.Len(t, response.StorageCredentials, 1) + assert.Equal(t, "s3://warehouse/table/", response.StorageCredentials[0].Prefix) + assert.Equal(t, "plan-key", response.StorageCredentials[0].Config["s3.access-key-id"]) + + requests := srv.Requests() + require.Len(t, requests, 2) + request := requests[1] + assert.Equal(t, http.MethodPost, request.Method) + assert.Equal(t, "/v1/catalog/namespaces/org%1Fanalytics/tables/events/plan", request.Path) + assert.Equal(t, "catalog", request.Prefix) + assert.Equal(t, "org\x1fanalytics", request.Namespace) + assert.Equal(t, "events", request.Table) + assert.Equal(t, idempotencyKey, request.Header.Get("Idempotency-Key")) + assert.Equal(t, accessDelegation, request.Header.Get("X-Iceberg-Access-Delegation")) + assert.JSONEq(t, `{ + "snapshot-id": 123, + "select": ["id", "payload"], + "filter": {"type": "eq", "term": "id", "value": 34}, + "min-rows-requested": 50, + "case-sensitive": false, + "use-snapshot-schema": true, + "stats-fields": ["id", "payload"] + }`, string(request.Body)) +} + +func TestPlanFakeAsynchronousPlanningWaitsThroughRetry(t *testing.T) { + t.Parallel() + + srv := planfake.New(t, planfake.Scenario{ + ConfigResponse: fullPlanningConfigResponse(), + ExpectedTarget: standardPlanFakeTarget(), + PlanResponse: planFakeJSONResponse(`{"status":"submitted","plan-id":"plan-async"}`), + PollResponses: map[string]planfake.ResponseSequence{ + "plan-async": { + Responses: []planfake.Response{ + { + Status: http.StatusServiceUnavailable, + Body: json.RawMessage(`{ + "error":{"message":"busy","type":"ServiceUnavailableException","code":503} + }`), + }, + planFakeJSONResponse(`{"status":"submitted"}`), + planFakeJSONResponse(`{ + "status":"completed", + "plan-tasks":["root-task"], + "storage-credentials":[{ + "prefix":"s3://warehouse/table/", + "config":{"token":"async-token"} + }] + }`), + }, + }, + }, + }) + catalog, err := rest.NewCatalog(t.Context(), "rest", srv.URL(), rest.WithPrefix("warehouse")) + require.NoError(t, err) + + accessDelegation := "remote-signing" + initial, err := catalog.PlanTableScan(t.Context(), standardPlanFakeIdentifier(), rest.PlanTableScanRequest{ + AccessDelegation: &accessDelegation, + }) + require.NoError(t, err) + require.NotNil(t, initial.PlanID) + assert.Equal(t, rest.PlanStatusSubmitted, initial.Status) + + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + completed, err := catalog.WaitForPlan(ctx, standardPlanFakeIdentifier(), *initial.PlanID, rest.WaitForPlanOptions{ + MinDelay: time.Millisecond, + MaxDelay: 2 * time.Millisecond, + CancelGracePeriod: time.Second, + MaxRetries: 10, + AccessDelegation: &accessDelegation, + }) + require.NoError(t, err) + + assert.Equal(t, rest.PlanStatusCompleted, completed.Status) + assert.Equal(t, []string{"root-task"}, completed.PlanTasks) + require.Len(t, completed.StorageCredentials, 1) + assert.Equal(t, "async-token", completed.StorageCredentials[0].Config["token"]) + assert.Equal(t, 3, srv.PollCount("plan-async")) + assert.Zero(t, srv.CancelCount("plan-async")) + + requests := srv.Requests() + assertPlanFakeUUIDv7Header(t, requests[1]) + pollRequests := requestsForPlanFakeOperation(requests, http.MethodGet, "plan-async", "") + require.Len(t, pollRequests, 3) + for _, request := range pollRequests { + assert.Equal(t, accessDelegation, request.Header.Get("X-Iceberg-Access-Delegation")) + } +} + +func TestPlanFakeWaitCancellationCleansUpServerPlan(t *testing.T) { + t.Parallel() + + srv := planfake.New(t, planfake.Scenario{ + ConfigResponse: fullPlanningConfigResponse(), + ExpectedTarget: standardPlanFakeTarget(), + PlanResponse: planFakeJSONResponse(`{"status":"submitted","plan-id":"plan-cancel"}`), + PollResponses: map[string]planfake.ResponseSequence{ + "plan-cancel": { + Responses: []planfake.Response{planFakeJSONResponse(`{"status":"submitted"}`)}, + RepeatLast: true, + }, + }, + CancelResponses: map[string]planfake.ResponseSequence{ + "plan-cancel": { + Responses: []planfake.Response{{Status: http.StatusNoContent}}, + }, + }, + }) + catalog, err := rest.NewCatalog(t.Context(), "rest", srv.URL(), rest.WithPrefix("warehouse")) + require.NoError(t, err) + + initial, err := catalog.PlanTableScan(t.Context(), standardPlanFakeIdentifier(), rest.PlanTableScanRequest{}) + require.NoError(t, err) + require.NotNil(t, initial.PlanID) + + waitContext, cancelWait := context.WithCancel(t.Context()) + defer cancelWait() + waitDone := make(chan error, 1) + go func() { + _, waitErr := catalog.WaitForPlan(waitContext, standardPlanFakeIdentifier(), *initial.PlanID, rest.WaitForPlanOptions{ + MinDelay: time.Hour, + MaxDelay: time.Hour, + CancelGracePeriod: time.Second, + }) + waitDone <- waitErr + }() + + observeContext, cancelObserve := context.WithTimeout(t.Context(), 5*time.Second) + defer cancelObserve() + require.NoError(t, srv.WaitForPollCount(observeContext, "plan-cancel", 1)) + cancelWait() + err = waitForPlanFakeResult(t, observeContext, waitDone) + require.ErrorIs(t, err, context.Canceled) + assert.GreaterOrEqual(t, srv.PollCount("plan-cancel"), 1) + assert.Equal(t, 1, srv.CancelCount("plan-cancel")) + assert.Empty(t, srv.ScenarioErrors()) + + requests := requestsForPlanFakeOperation(srv.Requests(), http.MethodDelete, "plan-cancel", "") + require.Len(t, requests, 1) + assert.Empty(t, requests[0].Header.Get("Idempotency-Key")) + assert.Empty(t, requests[0].Header.Get("X-Iceberg-Access-Delegation")) +} + +func TestPlanFakeWaitCancellationBoundsStalledCleanup(t *testing.T) { + t.Parallel() + + stalledCancel := make(chan struct{}) + srv := planfake.New(t, planfake.Scenario{ + ConfigResponse: fullPlanningConfigResponse(), + ExpectedTarget: standardPlanFakeTarget(), + PlanResponse: planFakeJSONResponse(`{"status":"submitted","plan-id":"plan-stalled-cancel"}`), + PollResponses: map[string]planfake.ResponseSequence{ + "plan-stalled-cancel": { + Responses: []planfake.Response{planFakeJSONResponse(`{"status":"submitted"}`)}, + RepeatLast: true, + }, + }, + CancelResponses: map[string]planfake.ResponseSequence{ + "plan-stalled-cancel": { + Responses: []planfake.Response{{Status: http.StatusNoContent, Gate: stalledCancel}}, + }, + }, + }) + catalog, err := rest.NewCatalog(t.Context(), "rest", srv.URL(), rest.WithPrefix("warehouse")) + require.NoError(t, err) + + initial, err := catalog.PlanTableScan(t.Context(), standardPlanFakeIdentifier(), rest.PlanTableScanRequest{}) + require.NoError(t, err) + require.NotNil(t, initial.PlanID) + + waitContext, cancelWait := context.WithCancel(t.Context()) + defer cancelWait() + waitDone := make(chan error, 1) + go func() { + _, waitErr := catalog.WaitForPlan(waitContext, standardPlanFakeIdentifier(), *initial.PlanID, rest.WaitForPlanOptions{ + MinDelay: time.Hour, + MaxDelay: time.Hour, + CancelGracePeriod: 100 * time.Millisecond, + }) + waitDone <- waitErr + }() + + observeContext, cancelObserve := context.WithTimeout(t.Context(), 5*time.Second) + defer cancelObserve() + require.NoError(t, srv.WaitForPollCount(observeContext, "plan-stalled-cancel", 1)) + cancelWait() + err = waitForPlanFakeResult(t, observeContext, waitDone) + require.ErrorIs(t, err, context.Canceled) + assert.GreaterOrEqual(t, srv.PollCount("plan-stalled-cancel"), 1) + assert.Equal(t, 1, srv.CancelCount("plan-stalled-cancel")) + assert.Empty(t, srv.ScenarioErrors()) +} + +func TestPlanFakeSupportsNestedAndReissuedTaskHandles(t *testing.T) { + t.Parallel() + + srv := planfake.New(t, planfake.Scenario{ + ConfigResponse: fullPlanningConfigResponse(), + ExpectedTarget: standardPlanFakeTarget(), + TaskResponses: map[string]planfake.ResponseSequence{ + "root-task": { + Responses: []planfake.Response{planFakeJSONResponse(`{ + "plan-tasks":["reissued-task","reissued-task"], + "file-scan-tasks":[` + rootFileScanTaskJSON + `] + }`)}, + }, + "reissued-task": { + Responses: []planfake.Response{ + planFakeJSONResponse(`{"plan-tasks":["leaf-a"]}`), + planFakeJSONResponse(`{"plan-tasks":["leaf-b"]}`), + }, + }, + "leaf-a": { + Responses: []planfake.Response{planFakeJSONResponse(envelopeAScanTasksJSON)}, + RepeatLast: true, + }, + "leaf-b": { + Responses: []planfake.Response{planFakeJSONResponse(envelopeBScanTasksJSON)}, + RepeatLast: true, + }, + "grandchild": { + Responses: []planfake.Response{ + planFakeJSONResponse(`{"file-scan-tasks":[` + grandchildFileScanTaskJSON + `]}`), + }, + }, + }, + }) + catalog, err := rest.NewCatalog(t.Context(), "rest", srv.URL(), rest.WithPrefix("warehouse")) + require.NoError(t, err) + + queue := []string{"root-task"} + responsesByHandle := make(map[string][]rest.FetchScanTasksResponse) + for len(queue) > 0 { + handle := queue[0] + queue = queue[1:] + + response, err := catalog.FetchScanTasks(t.Context(), standardPlanFakeIdentifier(), rest.FetchScanTasksRequest{ + PlanTask: handle, + }) + require.NoError(t, err) + responsesByHandle[handle] = append(responsesByHandle[handle], response) + queue = append(queue, response.PlanTasks...) + } + + assert.Equal(t, 1, srv.TaskCount("root-task")) + assert.Equal(t, 2, srv.TaskCount("reissued-task")) + assert.Equal(t, 1, srv.TaskCount("leaf-a")) + assert.Equal(t, 1, srv.TaskCount("leaf-b")) + assert.Equal(t, 1, srv.TaskCount("grandchild")) + require.Len(t, responsesByHandle["reissued-task"], 2) + assert.Equal(t, []string{"leaf-a"}, responsesByHandle["reissued-task"][0].PlanTasks) + assert.Equal(t, []string{"leaf-b"}, responsesByHandle["reissued-task"][1].PlanTasks) + + // Both independently fetched envelopes use delete reference 0 and bind it to + // a different delete path. Keeping these raw envelopes separate is required + // before the decoder turns them into domain tasks. + require.Len(t, responsesByHandle["leaf-a"], 1) + assert.Len(t, responsesByHandle["leaf-a"][0].FileScanTasks, 1) + assert.Len(t, responsesByHandle["leaf-a"][0].DeleteFiles, 1) + require.Len(t, responsesByHandle["leaf-b"], 1) + assert.Len(t, responsesByHandle["leaf-b"][0].FileScanTasks, 1) + assert.Len(t, responsesByHandle["leaf-b"][0].DeleteFiles, 1) + + // Fetch each leaf again as raw JSON through an independent HTTP client. This + // keeps the fixture assertion independent from production wire types and binds + // each handle to its own envelope-local delete reference and child handles. + leafA := fetchRawPlanTaskFixture(t, srv.URL(), "leaf-a") + assertEnvelopeLocalFixture(t, leafA, nil, + "s3://warehouse/table/envelope-a.parquet", + "s3://warehouse/table/envelope-a-delete.parquet") + leafB := fetchRawPlanTaskFixture(t, srv.URL(), "leaf-b") + assertEnvelopeLocalFixture(t, leafB, []string{"grandchild"}, + "s3://warehouse/table/envelope-b.parquet", + "s3://warehouse/table/envelope-b-delete.parquet") + assert.Equal(t, 2, srv.TaskCount("leaf-a")) + assert.Equal(t, 2, srv.TaskCount("leaf-b")) + + for _, request := range srv.Requests() { + if request.PlanTask == "" { + continue + } + assertPlanFakeUUIDv7Header(t, request) + assert.Empty(t, request.Header.Get("X-Iceberg-Access-Delegation")) + } +} + +func TestPlanFakeModelsExpiredPlansAndTasks(t *testing.T) { Review Comment: Addressed. Added tests for: - Failed plan submission preserving `PlanFailedError.Detail`. - Cancelled polling returning `ErrPlanCancelled`. - POST `/plan` 404 mappings for both `ErrNoSuchTable` and `ErrNoSuchNamespace`. The focused tests and race suite pass. Thanks for catching these gaps. -- 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]
