From ea29eef38637a9975b81ba9f28ef6a37fb4fd012 Mon Sep 17 00:00:00 2001 From: Akshay Thorat Date: Mon, 10 Aug 2026 15:52:31 -0700 Subject: [PATCH] [#11284] feat(iceberg-rest): Add the fetch scan tasks endpoint and its plumbing Wire `POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/tasks`, the second step of the Iceberg REST two-step scan planning protocol, through the Iceberg REST server: the JAX-RS resource, the table dispatcher chain, the `FETCH_SCAN_TASKS` audit operation with its three listener events, and `NoSuchPlanTaskException` mapped to 404. Scan planning still returns every file scan task inline and hands out no `plan-tasks`, so no plan task presented to this endpoint was issued by this server and every request is rejected as unknown. The endpoint is deliberately not advertised in `/v1/config` while that is the case, so a spec-compliant client never reaches it. Batching a plan into plan tasks, and redeeming them here, follows in a later change. Also report a 400 rather than a 500 when either scan planning endpoint is called without a request body: Jersey passes a null entity, which would otherwise surface as a downstream NPE. --- .../org/apache/gravitino/audit/AuditLog.java | 2 + .../audit/v2/CompatibilityUtils.java | 1 + .../listener/api/event/OperationType.java | 1 + .../audit/v2/TestCompatibilityUtils.java | 1 + .../service/CatalogWrapperForREST.java | 31 ++++++++ .../service/IcebergExceptionMapper.java | 2 + .../IcebergTableEventDispatcher.java | 35 ++++++++ .../IcebergTableHookDispatcher.java | 19 +++++ .../IcebergTableOperationDispatcher.java | 16 ++++ .../IcebergTableOperationExecutor.java | 12 +++ .../service/rest/IcebergTableOperations.java | 77 ++++++++++++++++++ .../api/event/IcebergFetchScanTasksEvent.java | 37 +++++++++ .../IcebergFetchScanTasksFailureEvent.java | 37 +++++++++ .../event/IcebergFetchScanTasksPreEvent.java | 37 +++++++++ .../rest/TestIcebergTableOperations.java | 79 +++++++++++++++++++ 15 files changed, 387 insertions(+) create mode 100644 iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFetchScanTasksEvent.java create mode 100644 iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFetchScanTasksFailureEvent.java create mode 100644 iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFetchScanTasksPreEvent.java diff --git a/core/src/main/java/org/apache/gravitino/audit/AuditLog.java b/core/src/main/java/org/apache/gravitino/audit/AuditLog.java index 7ebc32012ec..c3f9ea20552 100644 --- a/core/src/main/java/org/apache/gravitino/audit/AuditLog.java +++ b/core/src/main/java/org/apache/gravitino/audit/AuditLog.java @@ -262,6 +262,8 @@ enum Operation { PLAN_TABLE_SCAN, + FETCH_SCAN_TASKS, + PURGE_TABLE, LIST_TABLE, diff --git a/core/src/main/java/org/apache/gravitino/audit/v2/CompatibilityUtils.java b/core/src/main/java/org/apache/gravitino/audit/v2/CompatibilityUtils.java index 725bb5dba62..912a59495f2 100644 --- a/core/src/main/java/org/apache/gravitino/audit/v2/CompatibilityUtils.java +++ b/core/src/main/java/org/apache/gravitino/audit/v2/CompatibilityUtils.java @@ -55,6 +55,7 @@ public class CompatibilityUtils { .put(OperationType.LOAD_TABLE, Operation.LOAD_TABLE) .put(OperationType.LOAD_TABLE_CREDENTIAL, Operation.LOAD_TABLE_CREDENTIAL) .put(OperationType.PLAN_TABLE_SCAN, Operation.PLAN_TABLE_SCAN) + .put(OperationType.FETCH_SCAN_TASKS, Operation.FETCH_SCAN_TASKS) .put(OperationType.LIST_TABLE, Operation.LIST_TABLE) .put(OperationType.ALTER_TABLE, Operation.ALTER_TABLE) .put(OperationType.RENAME_TABLE, Operation.RENAME_TABLE) diff --git a/core/src/main/java/org/apache/gravitino/listener/api/event/OperationType.java b/core/src/main/java/org/apache/gravitino/listener/api/event/OperationType.java index 5350c18562e..6709fc7917d 100644 --- a/core/src/main/java/org/apache/gravitino/listener/api/event/OperationType.java +++ b/core/src/main/java/org/apache/gravitino/listener/api/event/OperationType.java @@ -27,6 +27,7 @@ public enum OperationType { LOAD_TABLE, LOAD_TABLE_CREDENTIAL, PLAN_TABLE_SCAN, + FETCH_SCAN_TASKS, LIST_TABLE, ALTER_TABLE, RENAME_TABLE, diff --git a/core/src/test/java/org/apache/gravitino/audit/v2/TestCompatibilityUtils.java b/core/src/test/java/org/apache/gravitino/audit/v2/TestCompatibilityUtils.java index 94840fed0f0..e08eeb4cb0d 100644 --- a/core/src/test/java/org/apache/gravitino/audit/v2/TestCompatibilityUtils.java +++ b/core/src/test/java/org/apache/gravitino/audit/v2/TestCompatibilityUtils.java @@ -62,6 +62,7 @@ public void testToAuditLogOperation() { {OperationType.LOAD_TABLE, Operation.LOAD_TABLE}, {OperationType.LOAD_TABLE_CREDENTIAL, Operation.LOAD_TABLE_CREDENTIAL}, {OperationType.PLAN_TABLE_SCAN, Operation.PLAN_TABLE_SCAN}, + {OperationType.FETCH_SCAN_TASKS, Operation.FETCH_SCAN_TASKS}, {OperationType.TABLE_EXISTS, Operation.TABLE_EXISTS}, {OperationType.LIST_TABLE, Operation.LIST_TABLE}, {OperationType.RENAME_TABLE, Operation.RENAME_TABLE}, diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java index 1af51f5295e..9107a022281 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java @@ -59,13 +59,16 @@ import org.apache.iceberg.TableScan; import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.NoSuchPlanTaskException; import org.apache.iceberg.exceptions.ServiceUnavailableException; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.rest.CatalogHandlers; import org.apache.iceberg.rest.PlanStatus; import org.apache.iceberg.rest.requests.CreateTableRequest; +import org.apache.iceberg.rest.requests.FetchScanTasksRequest; import org.apache.iceberg.rest.requests.PlanTableScanRequest; import org.apache.iceberg.rest.requests.RegisterTableRequest; +import org.apache.iceberg.rest.responses.FetchScanTasksResponse; import org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse; import org.apache.iceberg.rest.responses.LoadCredentialsResponse; import org.apache.iceberg.rest.responses.LoadTableResponse; @@ -484,6 +487,34 @@ public PlanTableScanResponse planTableScan( } } + /** + * Fetch the scan tasks covered by a {@code plan-task} previously handed out by {@link + * #planTableScan}, completing the second step of the Iceberg REST two-step scan planning + * protocol. + * + *

{@link #planTableScan} currently returns every file scan task inline and hands out no {@code + * plan-tasks}, so no plan task presented here was issued by this server and every request is + * rejected as unknown. The endpoint is deliberately not advertised in {@code /v1/config} while + * that is the case, so a spec-compliant client never reaches it. Batching a plan into plan tasks, + * and redeeming them here, follows in a later change. + * + * @param tableIdentifier the table the plan task belongs to. + * @param request the request carrying the {@code plan-task}. + * @return the file scan tasks the plan task covers. + * @throws org.apache.iceberg.exceptions.NoSuchTableException if the table doesn't exist. + * @throws NoSuchPlanTaskException if the plan task was not issued for this table. + */ + public FetchScanTasksResponse fetchScanTasks( + TableIdentifier tableIdentifier, FetchScanTasksRequest request) { + // Validate the table exists first, so a bad table reports 404 for the table rather than + // masking it as an unknown plan task. Consistent with planTableScan behavior. + getCatalog().loadTable(tableIdentifier); + + LOG.info("Rejecting unknown plan task '{}' for table {}", request.planTask(), tableIdentifier); + throw new NoSuchPlanTaskException( + "Plan task %s was not issued for table %s", request.planTask(), tableIdentifier); + } + /** * Inject vended credentials into a scan response using the already-loaded table, avoiding a * redundant {@code loadTable} call. Follows the same eligibility logic as {@link diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergExceptionMapper.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergExceptionMapper.java index dd00b9c4b68..7d55a264fb1 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergExceptionMapper.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergExceptionMapper.java @@ -37,6 +37,7 @@ import org.apache.iceberg.exceptions.NamespaceNotEmptyException; import org.apache.iceberg.exceptions.NoSuchIcebergTableException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.exceptions.NoSuchPlanTaskException; import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.exceptions.NoSuchViewException; import org.apache.iceberg.exceptions.NotAuthorizedException; @@ -72,6 +73,7 @@ public class IcebergExceptionMapper implements ExceptionMapper { .put(NoSuchTableException.class, 404) .put(NoSuchIcebergTableException.class, 404) .put(NoSuchCatalogException.class, 404) + .put(NoSuchPlanTaskException.class, 404) .put(UnsupportedOperationException.class, 406) .put(NoSuchViewException.class, 404) .put(AlreadyExistsException.class, 409) diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableEventDispatcher.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableEventDispatcher.java index 9a6cb21ee2f..cbdfd21c1d2 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableEventDispatcher.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableEventDispatcher.java @@ -31,6 +31,9 @@ import org.apache.gravitino.listener.api.event.IcebergDropTableEvent; import org.apache.gravitino.listener.api.event.IcebergDropTableFailureEvent; import org.apache.gravitino.listener.api.event.IcebergDropTablePreEvent; +import org.apache.gravitino.listener.api.event.IcebergFetchScanTasksEvent; +import org.apache.gravitino.listener.api.event.IcebergFetchScanTasksFailureEvent; +import org.apache.gravitino.listener.api.event.IcebergFetchScanTasksPreEvent; import org.apache.gravitino.listener.api.event.IcebergListTableEvent; import org.apache.gravitino.listener.api.event.IcebergListTableFailureEvent; import org.apache.gravitino.listener.api.event.IcebergListTablePreEvent; @@ -56,9 +59,11 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.rest.requests.CreateTableRequest; +import org.apache.iceberg.rest.requests.FetchScanTasksRequest; import org.apache.iceberg.rest.requests.PlanTableScanRequest; import org.apache.iceberg.rest.requests.RenameTableRequest; import org.apache.iceberg.rest.requests.UpdateTableRequest; +import org.apache.iceberg.rest.responses.FetchScanTasksResponse; import org.apache.iceberg.rest.responses.ListTablesResponse; import org.apache.iceberg.rest.responses.LoadCredentialsResponse; import org.apache.iceberg.rest.responses.LoadTableResponse; @@ -305,6 +310,36 @@ public PlanTableScanResponse planTableScan( return planTableScanResponse; } + /** + * Fetch the scan tasks for a {@code plan-task} returned by a prior scan plan. + * + * @param context Iceberg REST request context information. + * @param tableIdentifier The Iceberg table identifier. + * @param request The request carrying the {@code plan-task}. + * @return A FetchScanTasksResponse containing the scan tasks for that plan task + */ + @Override + public FetchScanTasksResponse fetchScanTasks( + IcebergRequestContext context, + TableIdentifier tableIdentifier, + FetchScanTasksRequest request) { + NameIdentifier gravitinoNameIdentifier = + IcebergRESTUtils.getGravitinoNameIdentifier( + metalakeName, context.catalogName(), tableIdentifier); + eventBus.dispatchEvent(new IcebergFetchScanTasksPreEvent(context, gravitinoNameIdentifier)); + FetchScanTasksResponse fetchScanTasksResponse; + try { + fetchScanTasksResponse = + icebergTableOperationDispatcher.fetchScanTasks(context, tableIdentifier, request); + } catch (Exception e) { + eventBus.dispatchEvent( + new IcebergFetchScanTasksFailureEvent(context, gravitinoNameIdentifier, e)); + throw e; + } + eventBus.dispatchEvent(new IcebergFetchScanTasksEvent(context, gravitinoNameIdentifier)); + return fetchScanTasksResponse; + } + @Override public Optional getTableMetadataLocation( IcebergRequestContext context, TableIdentifier tableIdentifier) { diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableHookDispatcher.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableHookDispatcher.java index bf95e0a5a3f..d804c14c14e 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableHookDispatcher.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableHookDispatcher.java @@ -38,9 +38,11 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.rest.requests.CreateTableRequest; +import org.apache.iceberg.rest.requests.FetchScanTasksRequest; import org.apache.iceberg.rest.requests.PlanTableScanRequest; import org.apache.iceberg.rest.requests.RenameTableRequest; import org.apache.iceberg.rest.requests.UpdateTableRequest; +import org.apache.iceberg.rest.responses.FetchScanTasksResponse; import org.apache.iceberg.rest.responses.ListTablesResponse; import org.apache.iceberg.rest.responses.LoadCredentialsResponse; import org.apache.iceberg.rest.responses.LoadTableResponse; @@ -197,6 +199,23 @@ public PlanTableScanResponse planTableScan( return dispatcher.planTableScan(context, tableIdentifier, scanRequest); } + /** + * Fetch the scan tasks for a {@code plan-task} returned by a prior scan plan. Read-only, so no + * hooks are needed and the call is passed straight through. + * + * @param context Iceberg REST request context information. + * @param tableIdentifier The Iceberg table identifier. + * @param request The request carrying the {@code plan-task}. + * @return A FetchScanTasksResponse containing the scan tasks for that plan task. + */ + @Override + public FetchScanTasksResponse fetchScanTasks( + IcebergRequestContext context, + TableIdentifier tableIdentifier, + FetchScanTasksRequest request) { + return dispatcher.fetchScanTasks(context, tableIdentifier, request); + } + @Override public Optional getTableMetadataLocation( IcebergRequestContext context, TableIdentifier tableIdentifier) { diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationDispatcher.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationDispatcher.java index 5a90842dbb9..c05173621ad 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationDispatcher.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationDispatcher.java @@ -24,9 +24,11 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.rest.requests.CreateTableRequest; +import org.apache.iceberg.rest.requests.FetchScanTasksRequest; import org.apache.iceberg.rest.requests.PlanTableScanRequest; import org.apache.iceberg.rest.requests.RenameTableRequest; import org.apache.iceberg.rest.requests.UpdateTableRequest; +import org.apache.iceberg.rest.responses.FetchScanTasksResponse; import org.apache.iceberg.rest.responses.ListTablesResponse; import org.apache.iceberg.rest.responses.LoadCredentialsResponse; import org.apache.iceberg.rest.responses.LoadTableResponse; @@ -135,6 +137,20 @@ PlanTableScanResponse planTableScan( TableIdentifier tableIdentifier, PlanTableScanRequest scanRequest); + /** + * Fetch the scan tasks for a {@code plan-task} returned by a prior {@link #planTableScan} call, + * completing the second step of the Iceberg REST two-step scan planning protocol. + * + * @param context Iceberg REST request context information. + * @param tableIdentifier The Iceberg table identifier. + * @param request The request carrying the {@code plan-task}. + * @return A {@link FetchScanTasksResponse} containing the scan tasks for that plan task. + */ + FetchScanTasksResponse fetchScanTasks( + IcebergRequestContext context, + TableIdentifier tableIdentifier, + FetchScanTasksRequest request); + /** * Retrieves the metadata file location for a table without loading full table metadata. This is * an optional fast path for catalogs that support cheap metadata location retrieval. diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java index 954d2a8c270..5a8a21d71e7 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java @@ -41,9 +41,11 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.rest.requests.CreateTableRequest; +import org.apache.iceberg.rest.requests.FetchScanTasksRequest; import org.apache.iceberg.rest.requests.PlanTableScanRequest; import org.apache.iceberg.rest.requests.RenameTableRequest; import org.apache.iceberg.rest.requests.UpdateTableRequest; +import org.apache.iceberg.rest.responses.FetchScanTasksResponse; import org.apache.iceberg.rest.responses.ListTablesResponse; import org.apache.iceberg.rest.responses.LoadCredentialsResponse; import org.apache.iceberg.rest.responses.LoadTableResponse; @@ -231,6 +233,16 @@ public PlanTableScanResponse planTableScan( .planTableScan(tableIdentifier, scanRequest, context.requestCredentialVending(), privilege); } + @Override + public FetchScanTasksResponse fetchScanTasks( + IcebergRequestContext context, + TableIdentifier tableIdentifier, + FetchScanTasksRequest request) { + return icebergCatalogWrapperManager + .getCatalogWrapper(context.catalogName()) + .fetchScanTasks(tableIdentifier, request); + } + @Override public Optional getTableMetadataLocation( IcebergRequestContext context, TableIdentifier tableIdentifier) { diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java index 4bedb7c41ea..3f624117575 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java @@ -75,9 +75,11 @@ import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.rest.RESTUtil; import org.apache.iceberg.rest.requests.CreateTableRequest; +import org.apache.iceberg.rest.requests.FetchScanTasksRequest; import org.apache.iceberg.rest.requests.PlanTableScanRequest; import org.apache.iceberg.rest.requests.ReportMetricsRequest; import org.apache.iceberg.rest.requests.UpdateTableRequest; +import org.apache.iceberg.rest.responses.FetchScanTasksResponse; import org.apache.iceberg.rest.responses.ListTablesResponse; import org.apache.iceberg.rest.responses.LoadCredentialsResponse; import org.apache.iceberg.rest.responses.LoadTableResponse; @@ -526,6 +528,9 @@ public Response planTableScan( @Encoded() @PathParam("table") @AuthorizationMetadata(type = EntityType.TABLE) String table, PlanTableScanRequest scanRequest, @HeaderParam(X_ICEBERG_ACCESS_DELEGATION) String accessDelegation) { + if (scanRequest == null) { + return missingRequestBody("plan table scan"); + } boolean isCredentialVending = isCredentialVending(accessDelegation); String catalogName = IcebergRESTUtils.getCatalogName(prefix); Namespace icebergNS = @@ -559,6 +564,68 @@ public Response planTableScan( } } + /** + * Fetch scan tasks endpoint. Completes the second step of the Iceberg REST two-step scan planning + * protocol: a client exchanges a {@code plan-task} handed out by {@code POST + * .../tables/{table}/plan} for the scan tasks it covers. + * + * @param prefix The catalog prefix + * @param namespace The namespace + * @param table The table name + * @param request The request containing the {@code plan-task} + * @return Response containing the scan tasks for the given plan task + */ + @POST + @Path("{table}/tasks") + @Produces(MediaType.APPLICATION_JSON) + @Consumes(MediaType.APPLICATION_JSON) + @Timed(name = "fetch-scan-tasks." + MetricNames.HTTP_PROCESS_DURATION, absolute = true) + @ResponseMetered(name = "fetch-scan-tasks", absolute = true) + @AuthorizationExpression( + expression = + "ANY(OWNER, METALAKE, CATALOG) || " + + "SCHEMA_OWNER_WITH_USE_CATALOG || " + + "ANY_USE_CATALOG && ANY_USE_SCHEMA && (TABLE::OWNER || ANY_SELECT_TABLE || ANY_MODIFY_TABLE)", + accessMetadataType = MetadataObject.Type.TABLE) + public Response fetchScanTasks( + @PathParam("prefix") @AuthorizationMetadata(type = EntityType.CATALOG) String prefix, + @Encoded() @PathParam("namespace") @AuthorizationMetadata(type = EntityType.SCHEMA) + String namespace, + @Encoded() @PathParam("table") @AuthorizationMetadata(type = EntityType.TABLE) String table, + FetchScanTasksRequest request) { + if (request == null) { + return missingRequestBody("fetch scan tasks"); + } + String catalogName = IcebergRESTUtils.getCatalogName(prefix); + Namespace icebergNS = + RESTUtil.decodeNamespace(namespace, IcebergRESTUtils.NAMESPACE_SEPARATOR_URLENCODED_UTF_8); + String tableName = RESTUtil.decodeString(table); + LOG.info( + "Fetch scan tasks, catalog: {}, namespace: {}, table: {}, planTask: {}", + catalogName, + icebergNS, + tableName, + request.planTask()); + + try { + return Utils.doAs( + httpRequest, + () -> { + TableIdentifier tableIdentifier = TableIdentifier.of(icebergNS, tableName); + IcebergRequestContext context = + new IcebergRequestContext(httpServletRequest(), catalogName); + + FetchScanTasksResponse response = + tableOperationDispatcher.fetchScanTasks(context, tableIdentifier, request); + + return IcebergRESTUtils.ok(response); + }); + } catch (Exception e) { + LOG.error("Failed to fetch scan tasks: {}", e.getMessage(), e); + return IcebergExceptionMapper.toRESTResponse(e); + } + } + /** * Filters the {@link LoadTableResponse} to include only snapshots that are directly referenced by * the table's refs (branches and tags). This implements the {@code snapshots=refs} query @@ -586,6 +653,16 @@ static LoadTableResponse filterSnapshotsByRefs(LoadTableResponse loadTableRespon .build(); } + /** + * Builds a 400 response for a request that arrived without a body. Jersey passes a {@code null} + * entity when the body is absent, which would otherwise surface as a 500 from a downstream NPE + * even though the Iceberg REST specification expects a 400 for a malformed request. + */ + private static Response missingRequestBody(String operation) { + return IcebergExceptionMapper.toRESTResponse( + new IllegalArgumentException("Missing request body for " + operation)); + } + private static Response buildResponseWithETag(LoadTableResponse loadTableResponse) { return IcebergRESTUtils.buildResponseWithETag(loadTableResponse); } diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFetchScanTasksEvent.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFetchScanTasksEvent.java new file mode 100644 index 00000000000..94a82a6774e --- /dev/null +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFetchScanTasksEvent.java @@ -0,0 +1,37 @@ +/* + * 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 org.apache.gravitino.listener.api.event; + +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.annotation.DeveloperApi; + +/** Represent an event after fetching Iceberg scan tasks for a plan task successfully. */ +@DeveloperApi +public class IcebergFetchScanTasksEvent extends IcebergTableEvent { + public IcebergFetchScanTasksEvent( + IcebergRequestContext icebergRequestContext, NameIdentifier resourceIdentifier) { + super(icebergRequestContext, resourceIdentifier); + } + + @Override + public OperationType operationType() { + return OperationType.FETCH_SCAN_TASKS; + } +} diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFetchScanTasksFailureEvent.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFetchScanTasksFailureEvent.java new file mode 100644 index 00000000000..594554f97d1 --- /dev/null +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFetchScanTasksFailureEvent.java @@ -0,0 +1,37 @@ +/* + * 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 org.apache.gravitino.listener.api.event; + +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.annotation.DeveloperApi; + +/** Represent a failure event when fetching Iceberg scan tasks for a plan task failed. */ +@DeveloperApi +public class IcebergFetchScanTasksFailureEvent extends IcebergTableFailureEvent { + public IcebergFetchScanTasksFailureEvent( + IcebergRequestContext icebergRequestContext, NameIdentifier nameIdentifier, Exception e) { + super(icebergRequestContext, nameIdentifier, e); + } + + @Override + public OperationType operationType() { + return OperationType.FETCH_SCAN_TASKS; + } +} diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFetchScanTasksPreEvent.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFetchScanTasksPreEvent.java new file mode 100644 index 00000000000..5d92bb255e3 --- /dev/null +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/listener/api/event/IcebergFetchScanTasksPreEvent.java @@ -0,0 +1,37 @@ +/* + * 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 org.apache.gravitino.listener.api.event; + +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.annotation.DeveloperApi; + +/** Represent a pre event before fetching Iceberg scan tasks for a plan task. */ +@DeveloperApi +public class IcebergFetchScanTasksPreEvent extends IcebergTablePreEvent { + public IcebergFetchScanTasksPreEvent( + IcebergRequestContext icebergRequestContext, NameIdentifier tableIdentifier) { + super(icebergRequestContext, tableIdentifier); + } + + @Override + public OperationType operationType() { + return OperationType.FETCH_SCAN_TASKS; + } +} diff --git a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java index 0abaa7305c1..82295e30f7c 100644 --- a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java +++ b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java @@ -46,6 +46,8 @@ import org.apache.gravitino.listener.api.event.IcebergDropTableEvent; import org.apache.gravitino.listener.api.event.IcebergDropTableFailureEvent; import org.apache.gravitino.listener.api.event.IcebergDropTablePreEvent; +import org.apache.gravitino.listener.api.event.IcebergFetchScanTasksFailureEvent; +import org.apache.gravitino.listener.api.event.IcebergFetchScanTasksPreEvent; import org.apache.gravitino.listener.api.event.IcebergListTableEvent; import org.apache.gravitino.listener.api.event.IcebergListTableFailureEvent; import org.apache.gravitino.listener.api.event.IcebergListTablePreEvent; @@ -81,10 +83,12 @@ import org.apache.iceberg.rest.PlanStatus; import org.apache.iceberg.rest.RESTUtil; import org.apache.iceberg.rest.requests.CreateTableRequest; +import org.apache.iceberg.rest.requests.FetchScanTasksRequest; import org.apache.iceberg.rest.requests.PlanTableScanRequest; import org.apache.iceberg.rest.requests.RenameTableRequest; import org.apache.iceberg.rest.requests.ReportMetricsRequest; import org.apache.iceberg.rest.requests.UpdateTableRequest; +import org.apache.iceberg.rest.responses.ErrorResponse; import org.apache.iceberg.rest.responses.ListTablesResponse; import org.apache.iceberg.rest.responses.LoadTableResponse; import org.apache.iceberg.types.Types.NestedField; @@ -243,6 +247,76 @@ void testPlanTableScanWithIncrementalAppendScanValidRange(Namespace namespace) { Assertions.assertTrue(dummyEventListener.popPostEvent() instanceof IcebergPlanTableScanEvent); } + @ParameterizedTest + @MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces") + void testFetchScanTasksUnknownPlanTask(Namespace namespace) { + verifyCreateNamespaceSucc(namespace); + verifyCreateTableSucc(namespace, "fetch_tasks_table", true); + + dummyEventListener.clearEvent(); + + // Scan planning hands out no plan tasks yet, so every plan task presented here is one this + // server never issued, reported as 404 per the Iceberg REST spec. + Response response = + doFetchScanTasks( + namespace, "fetch_tasks_table", new FetchScanTasksRequest("not-a-plan-task")); + Assertions.assertEquals(Status.NOT_FOUND.getStatusCode(), response.getStatus()); + + // Assert on the error payload, not just the status: an unregistered route would also yield + // 404, which would let this test pass without the endpoint existing. + ErrorResponse error = response.readEntity(ErrorResponse.class); + Assertions.assertEquals("NoSuchPlanTaskException", error.type()); + Assertions.assertTrue( + error.message().contains("not-a-plan-task"), + "Error message should name the rejected plan task, but was: " + error.message()); + + Assertions.assertTrue( + dummyEventListener.popPreEvent() instanceof IcebergFetchScanTasksPreEvent); + Assertions.assertTrue( + dummyEventListener.popPostEvent() instanceof IcebergFetchScanTasksFailureEvent); + } + + @ParameterizedTest + @MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces") + void testFetchScanTasksTableNotFound(Namespace namespace) { + verifyCreateNamespaceSucc(namespace); + dummyEventListener.clearEvent(); + + // A missing table is reported as a missing table, not masked as an unknown plan task. + Response response = + doFetchScanTasks(namespace, "missing_table", new FetchScanTasksRequest("any-plan-task")); + Assertions.assertEquals(Status.NOT_FOUND.getStatusCode(), response.getStatus()); + + ErrorResponse error = response.readEntity(ErrorResponse.class); + Assertions.assertEquals("NoSuchTableException", error.type()); + + Assertions.assertTrue( + dummyEventListener.popPreEvent() instanceof IcebergFetchScanTasksPreEvent); + Assertions.assertTrue( + dummyEventListener.popPostEvent() instanceof IcebergFetchScanTasksFailureEvent); + } + + @ParameterizedTest + @MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces") + void testScanPlanningEndpointsRejectMissingRequestBody(Namespace namespace) { + verifyCreateNamespaceSucc(namespace); + verifyCreateTableSucc(namespace, "empty_body_table", true); + + // Jersey hands the resource method a null entity when the body is absent. Both scan planning + // endpoints must report that as a 400 rather than letting a downstream NPE become a 500. + for (String endpoint : new String[] {"plan", "tasks"}) { + Response response = + getTableClientBuilder(namespace, Optional.of("empty_body_table/" + endpoint)) + .post(Entity.entity("", MediaType.APPLICATION_JSON_TYPE)); + Assertions.assertEquals( + Status.BAD_REQUEST.getStatusCode(), + response.getStatus(), + "Empty body on /" + endpoint + " should be a 400, not a 500"); + } + // No events are asserted here: the request is rejected at the REST boundary before it reaches + // the dispatcher chain, so no operation event is dispatched. + } + @ParameterizedTest @MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces") void testDropTable(Namespace namespace) { @@ -557,6 +631,11 @@ private Response doPlanTableScan(Namespace ns, String tableName, PlanTableScanRe return builder.post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE)); } + private Response doFetchScanTasks(Namespace ns, String tableName, FetchScanTasksRequest request) { + Invocation.Builder builder = getTableClientBuilder(ns, Optional.of(tableName + "/tasks")); + return builder.post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE)); + } + private Response doUpdateTable(Namespace ns, String name, TableMetadata base) { TableMetadata newMetadata = base.updateSchema(newTableSchema); List metadataUpdates = newMetadata.changes();