This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new d2f49758be [#11284] feat(iceberg-rest): Add the fetch scan tasks
endpoint and its plumbing (#12412)
d2f49758be is described below
commit d2f49758be39a62bf6b2e18e5639dbc783296a49
Author: Akshay Thorat <[email protected]>
AuthorDate: Tue Sep 22 23:43:27 2026 -0700
[#11284] feat(iceberg-rest): Add the fetch scan tasks endpoint and its
plumbing (#12412)
### What changes were proposed in this pull request?
This is PR 2 of the stack that splits #12194, as agreed with @nevzheng
([plan](https://github.com/apache/gravitino/pull/12194#issuecomment-5228220890)).
PR 1 is #12411.
It wires up `POST
/v1/{prefix}/namespaces/{namespace}/tables/{table}/tasks`, the second
step of the Iceberg REST two-step scan planning protocol, without yet
implementing what it hands back:
- `IcebergTableOperations.fetchScanTasks` - the JAX-RS resource, with
the same authorization expression as `POST .../plan` since it is the
same read of the same table.
- `fetchScanTasks` through the table dispatcher chain:
`IcebergTableOperationDispatcher` (interface),
`IcebergTableEventDispatcher` (events), `IcebergTableHookDispatcher`
(pass-through, the operation is read-only),
`IcebergTableOperationExecutor` (delegation to the catalog wrapper).
- `OperationType.FETCH_SCAN_TASKS` and
`AuditLog.Operation.FETCH_SCAN_TASKS`, with
`IcebergFetchScanTasksPreEvent`, `IcebergFetchScanTasksEvent` and
`IcebergFetchScanTasksFailureEvent`.
- `NoSuchPlanTaskException` mapped to `404` in `IcebergExceptionMapper`.
`CatalogWrapperForREST.planTableScan` 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
`CatalogWrapperForREST.fetchScanTasks` rejects every one of them as
unknown. Two deliberate consequences:
- **The endpoint is not advertised in `/v1/config`.** Clients gate on
the advertised endpoint set rather than probing - pyiceberg enables
server-side scan planning when it sees `POST .../tasks` there, and fails
the scan if the endpoint then serves nothing. Advertising it before it
can return tasks would be worse than not having it. From a client's
point of view this PR leaves behaviour exactly as it is today.
- **The success path of the endpoint arrives with batching**, in the
last PR of the stack, together with `TestIcebergFetchScanTasksEndpoint`
which exercises it end to end. I checked that `FetchScanTasksResponse`
is already covered by `RESTSerializers.registerAll`, so the response
serialization this PR leaves untested is not an open question.
One change here is not strictly about `/tasks`: both scan planning
endpoints now report `400` instead of `500` when called with no request
body. Jersey passes the resource method a `null` entity, which became an
NPE downstream. `/tasks` needed the guard, and applying the same three
lines to `/plan` seemed better than leaving the neighbouring endpoint
wrong. Happy to drop it into its own PR if you would rather keep this
one single-purpose.
### Why are the changes needed?
#11284 asks for the fetch scan tasks endpoint. Reviewers on #12194 asked
for that work in reviewable pieces; this is the plumbing piece,
separated from the batching logic that gives it something to return.
Fix: #11284
### Does this PR introduce _any_ user-facing change?
A new REST route exists but is not advertised, and answers `404
NoSuchPlanTaskException` for any plan task. No new configuration
properties. The only behaviour change a user can observe is that `POST
.../plan` and `POST .../tasks` with an empty body now return `400`
instead of `500`.
### How was this patch tested?
`./gradlew :iceberg:iceberg-rest-server:check -PskipITs` and `./gradlew
:core:test --tests "*TestCompatibilityUtils*"`, both green.
Three tests added to `TestIcebergTableOperations`, each run against both
a flat and a nested namespace:
- `testFetchScanTasksUnknownPlanTask` - a plan task this server never
issued is a `404`. Asserts on the error payload (`type` is
`NoSuchPlanTaskException`, message names the rejected plan task), not
just the status, so it cannot pass against an unregistered route.
Asserts the pre event and the failure event are dispatched.
- `testFetchScanTasksTableNotFound` - a missing table reports
`NoSuchTableException`, not a masked unknown plan task.
- `testScanPlanningEndpointsRejectMissingRequestBody` - an empty body on
`/plan` and on `/tasks` is a `400`.
---
.../java/org/apache/gravitino/audit/AuditLog.java | 2 +
.../gravitino/audit/v2/CompatibilityUtils.java | 1 +
.../listener/api/event/OperationType.java | 1 +
.../gravitino/audit/v2/TestCompatibilityUtils.java | 1 +
.../iceberg/service/CatalogWrapperForREST.java | 31 +++++++++
.../iceberg/service/IcebergExceptionMapper.java | 2 +
.../dispatcher/IcebergTableEventDispatcher.java | 35 ++++++++++
.../dispatcher/IcebergTableHookDispatcher.java | 19 ++++++
.../IcebergTableOperationDispatcher.java | 16 +++++
.../dispatcher/IcebergTableOperationExecutor.java | 12 ++++
.../service/rest/IcebergTableOperations.java | 77 +++++++++++++++++++++
.../api/event/IcebergFetchScanTasksEvent.java | 37 ++++++++++
.../event/IcebergFetchScanTasksFailureEvent.java | 37 ++++++++++
.../api/event/IcebergFetchScanTasksPreEvent.java | 37 ++++++++++
.../service/rest/TestIcebergTableOperations.java | 79 ++++++++++++++++++++++
15 files changed, 387 insertions(+)
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 070941a750..42168a24e4 100644
--- a/core/src/main/java/org/apache/gravitino/audit/AuditLog.java
+++ b/core/src/main/java/org/apache/gravitino/audit/AuditLog.java
@@ -266,6 +266,8 @@ public interface AuditLog {
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 c968dc3c6e..fea95fa9e5 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 d3bd1aa697..1a2e7dd0ca 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 8cb300de4f..9793591cdd 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 class TestCompatibilityUtils {
{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 c58547e4fc..64d5f090c0 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.TableProperties;
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;
@@ -498,6 +501,34 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
}
}
+ /**
+ * 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.
+ *
+ * <p>{@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 940bf785e7..b22fe5b407 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
@@ -38,6 +38,7 @@ import org.apache.iceberg.exceptions.ForbiddenException;
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;
@@ -73,6 +74,7 @@ public class IcebergExceptionMapper implements
ExceptionMapper<Throwable> {
.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 7482504625..dc05033e59 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
@@ -32,6 +32,9 @@ import
org.apache.gravitino.listener.api.event.IcebergCreateTablePreEvent;
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;
@@ -58,9 +61,11 @@ import org.apache.gravitino.utils.RequestContext;
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;
@@ -320,6 +325,36 @@ public class IcebergTableEventDispatcher implements
IcebergTableOperationDispatc
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<String> 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 bf95e0a5a3..d804c14c14 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.UpdateRequirement;
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 class IcebergTableHookDispatcher implements
IcebergTableOperationDispatch
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<String> 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 5a90842dbb..c05173621a 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.gravitino.listener.api.event.IcebergRequestContext;
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 @@ public interface IcebergTableOperationDispatcher {
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 954d2a8c27..5a8a21d71e 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.TableMetadata;
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 class IcebergTableOperationExecutor implements
IcebergTableOperationDispa
.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<String> 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 6f7fb4d80f..a5d2998b13 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
@@ -79,9 +79,11 @@ import org.apache.iceberg.catalog.Namespace;
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;
@@ -518,6 +520,9 @@ public class IcebergTableOperations {
@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 =
@@ -551,6 +556,68 @@ public class IcebergTableOperations {
}
}
+ /**
+ * 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
@@ -585,6 +652,16 @@ public class IcebergTableOperations {
return builder.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 0000000000..94a82a6774
--- /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 0000000000..594554f97d
--- /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 0000000000..5d92bb255e
--- /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 c27d0a2f2b..db0610e59f 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.IcebergCreateTablePreEvent;
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;
@@ -84,10 +86,12 @@ import org.apache.iceberg.metrics.ImmutableCommitReport;
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;
@@ -246,6 +250,76 @@ public class TestIcebergTableOperations extends
IcebergNamespaceTestBase {
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) {
@@ -570,6 +644,11 @@ public class TestIcebergTableOperations extends
IcebergNamespaceTestBase {
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<MetadataUpdate> metadataUpdates = newMetadata.changes();
