laserninja commented on code in PR #12194:
URL: https://github.com/apache/gravitino/pull/12194#discussion_r3668617116
##########
docs/iceberg-rest-service.md:
##########
@@ -697,13 +697,20 @@ Gravitino provides the build-in
`org.apache.gravitino.iceberg.common.cache.Local
Gravitino caches scan plan results to speed up repeated queries with identical
parameters. The cache uses snapshot ID as part of the cache key, so queries
against different snapshots will not use stale cached data.
-Plan scan responses follow the Iceberg 1.11 REST API: completed plans return
structured `file-scan-tasks` only. Legacy `plan-tasks` JSON strings (used by
some Iceberg 1.9.x–1.10.x clients) are not emitted.
+Plan scan responses follow the Iceberg 1.11 REST API: `file-scan-tasks` are
returned as structured tasks rather than as the JSON strings used by some
Iceberg 1.9.x–1.10.x clients.
-| Configuration item | Description
| Default value | Required | Since
Version |
-|------------------------------------------------------------|----------------------------------------------------------|---------------|----------|---------------|
-| `gravitino.iceberg-rest.scan-plan-cache-impl` | The
implementation of the scan plan cache. | (none) | No
| 1.2.0 |
-| `gravitino.iceberg-rest.scan-plan-cache-capacity` | The capacity of
the scan plan cache. | 200 | No | 1.2.0
|
-| `gravitino.iceberg-rest.scan-plan-cache-expire-minutes` | The expiration
time (in minutes) of the scan plan cache. | 60 | No | 1.2.0
|
+Scan planning is synchronous: `POST
/v1/{prefix}/namespaces/{namespace}/tables/{table}/plan` always returns status
`COMPLETED`, never `SUBMITTED`, so a plan is never left running in the
background.
+
+A plan is handed to the client in batches of at most
`scan-plan-task-batch-size` file scan tasks. The first batch is returned inline
in the plan response; each remaining batch is offered as a `plan-task` token
that the client exchanges for its tasks through `POST
/v1/{prefix}/namespaces/{namespace}/tables/{table}/tasks`, the second step of
the Iceberg REST scan planning protocol. A plan that fits in one batch carries
no `plan-task` tokens, so most scans complete in a single round trip.
+
+`plan-task` tokens are self-describing: each one carries the scan it was
planned from, with the snapshot pinned at planning time, plus the range of
tasks it covers. Nothing is stored server side between the two calls, so a
token stays redeemable after a server restart and on any Gravitino instance
serving the same catalog. Redeeming a token replans the pinned snapshot unless
the plan is still in the scan plan cache, so enabling the cache is recommended
when planning large tables. A token that this server did not issue, was issued
for another table, or refers to a plan that can no longer be reproduced (for
example because its snapshot expired) returns `404` with a
`NoSuchPlanTaskException` error.
+
+| Configuration item | Description
| Default value | Required | Since Version |
+|------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|----------|---------------|
+| `gravitino.iceberg-rest.scan-plan-cache-impl` | The
implementation of the scan plan cache.
| (none) | No | 1.2.0 |
+| `gravitino.iceberg-rest.scan-plan-cache-capacity` | The capacity of
the scan plan cache.
| 200 | No | 1.2.0 |
+| `gravitino.iceberg-rest.scan-plan-cache-expire-minutes` | The expiration
time (in minutes) of the scan plan cache.
| 60 | No | 1.2.0 |
+| `gravitino.iceberg-rest.scan-plan-task-batch-size` | Maximum number
of file scan tasks returned inline by one scan planning response. Tasks beyond
this limit are offered as `plan-task` tokens. Set to 0 to always return every
task inline. | 1000 | No | 1.3.0 |
Review Comment:
Aligned - the default is 100 now, in the config, the docs table and the PR
description. 1000 was my own guess at "big enough that most scans stay single
round trip"; matching the Iceberg side is the better default, and operators who
want the old behaviour can set `scan-plan-task-batch-size=0` to keep every task
inline.
##########
iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/IcebergConfig.java:
##########
@@ -341,6 +341,18 @@ public class IcebergConfig extends Config implements
OverwriteDefaultConfig {
.checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
.createWithDefault(60);
+ public static final ConfigEntry<Integer> SCAN_PLAN_TASK_BATCH_SIZE =
+ new ConfigBuilder(IcebergConstants.SCAN_PLAN_TASK_BATCH_SIZE)
+ .doc(
+ "Maximum number of file scan tasks returned inline by one scan
planning response. "
+ + "Tasks beyond this limit are handed out as plan-task
tokens that clients "
+ + "exchange for the remaining tasks. Set to 0 to disable
batching and always "
+ + "return every task inline.")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .intConf()
+ .checkValue(value -> value >= 0,
ConfigConstants.NON_NEGATIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(1000);
+
Review Comment:
Changed to 100.
##########
iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java:
##########
@@ -484,6 +479,232 @@ public PlanTableScanResponse planTableScan(
}
}
+ /**
+ * Fetch the scan tasks covered by a {@code plan-task} token previously
handed out by {@link
+ * #planTableScan}, completing the second step of the Iceberg REST scan
planning protocol.
+ *
+ * <p>Tokens are self-describing (see {@link PlanTaskToken}): each one
carries the scan request it
+ * was planned from, with the snapshot pinned at planning time, plus the
range of file scan tasks
+ * it stands for. The plan is reproduced here from the {@linkplain
ScanPlanCache scan plan cache}
+ * when it is still cached and re-planned against the pinned snapshot
otherwise, then the token's
+ * range is returned. Because no state is kept between the two calls, a
token remains redeemable
+ * after a server restart and on any Gravitino instance serving the same
catalog.
+ *
+ * @param tableIdentifier the table the plan task belongs to.
+ * @param request the request carrying the {@code plan-task} token.
+ * @return the file scan tasks the token covers.
+ * @throws org.apache.iceberg.exceptions.NoSuchTableException if the table
doesn't exist.
+ * @throws NoSuchPlanTaskException if the token was not issued for this
table, or the plan it
+ * refers to can no longer be reproduced (for example its snapshot has
expired).
+ */
+ @SuppressWarnings("deprecation")
+ public FetchScanTasksResponse fetchScanTasks(
+ TableIdentifier tableIdentifier, FetchScanTasksRequest request) {
+ Optional<PlanTaskToken> decodedToken =
PlanTaskToken.decode(request.planTask());
+
+ // 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.
+ Table table = getCatalog().loadTable(tableIdentifier);
+
+ if (!decodedToken.isPresent() ||
!decodedToken.get().matchesTable(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);
+ }
+
+ PlanTaskToken token = decodedToken.get();
+ PlanTableScanResponse fullPlan;
+ try {
+ fullPlan = planFullScan(tableIdentifier, table, token.scanRequest());
+ } catch (IllegalArgumentException e) {
+ // The pinned snapshot is gone (expired or rolled back), so the plan the
token refers to can
+ // no longer be reproduced. That is a stale plan task, not a bad request.
+ LOG.info(
+ "Plan task '{}' for table {} can no longer be planned: {}",
+ request.planTask(),
+ tableIdentifier,
+ e.getMessage());
+ throw new NoSuchPlanTaskException(
+ "Plan task %s is no longer available for table %s: %s",
+ request.planTask(), tableIdentifier, e.getMessage());
+ }
+
+ List<FileScanTask> allTasks = fullPlan.fileScanTasks();
+ int taskCount = allTasks == null ? 0 : allTasks.size();
+ if (token.offset() >= taskCount) {
+ LOG.info(
+ "Plan task '{}' for table {} covers tasks from offset {}, but the
plan has {} tasks",
+ request.planTask(),
+ tableIdentifier,
+ token.offset(),
+ taskCount);
+ throw new NoSuchPlanTaskException(
+ "Plan task %s is no longer available for table %s",
request.planTask(), tableIdentifier);
+ }
+
+ List<FileScanTask> batch =
+ ImmutableList.copyOf(
+ allTasks.subList(token.offset(), Math.min(token.offset() +
token.limit(), taskCount)));
+ LOG.info(
+ "Returning {} file scan tasks for plan task of table {} at offset {}",
+ batch.size(),
+ tableIdentifier,
+ token.offset());
+
+ FetchScanTasksResponse.Builder builder =
+ FetchScanTasksResponse.builder()
+ .withFileScanTasks(batch)
+ .withSpecsById(fullPlan.specsById());
+ List<DeleteFile> deleteFiles = referencedDeleteFiles(batch);
+ if (!deleteFiles.isEmpty()) {
+ builder.withDeleteFiles(deleteFiles);
+ }
+ return builder.build();
+ }
+
+ /**
+ * Plans the whole scan and returns every file scan task inline, serving the
{@linkplain
+ * ScanPlanCache scan plan cache} when the same plan was computed before.
+ *
+ * <p>Tasks are ordered deterministically so that a plan-task token, which
addresses tasks by
+ * position, resolves to the same tasks on a later re-plan of the same
snapshot.
+ */
+ private PlanTableScanResponse planFullScan(
+ TableIdentifier tableIdentifier, Table table, PlanTableScanRequest
scanRequest) {
+ ScanPlanCacheKey cacheKey = ScanPlanCacheKey.create(tableIdentifier,
table, scanRequest);
+ Optional<PlanTableScanResponse> cachedResponse =
scanPlanCache.get(cacheKey);
+ if (cachedResponse.isPresent()) {
+ LOG.info("Using cached scan plan for table: {}", tableIdentifier);
+ return cachedResponse.get();
+ }
+
+ List<FileScanTask> fileScanTasks = new ArrayList<>();
+ try (CloseableIterable<FileScanTask> scanTasks =
+ createFilePlanScanTasks(table, tableIdentifier, scanRequest)) {
+ for (FileScanTask fileScanTask : scanTasks) {
+ fileScanTasks.add(fileScanTask);
+ }
+ } catch (IOException e) {
+ LOG.error("Failed to close scan task iterator for table: {}",
tableIdentifier, e);
+ throw new RuntimeException("Failed to plan scan tasks: " +
e.getMessage(), e);
+ }
+
+ if (fileScanTasks.isEmpty()) {
+ LOG.info(
+ "Scan planning returned no tasks for table: {}. Table may be empty
or fully filtered.",
+ tableIdentifier);
+ }
+
+ // Iceberg plans manifests in parallel, so the order tasks come back in is
not reproducible.
+ // Sort them so positions stay stable across re-plans of the same snapshot.
+ fileScanTasks.sort(FILE_SCAN_TASK_ORDER);
+
+ PlanTableScanResponse response;
+ try {
+ response = buildCompletedPlanTableScanResponse(table, fileScanTasks);
+ } catch (Exception e) {
+ LOG.error("Failed to build scan plan response for table: {}",
tableIdentifier, e);
+ throw new RuntimeException(
+ String.format(
+ "Failed to build scan plan response for table: %s. Error: %s",
+ tableIdentifier, e.getMessage()),
+ e);
+ }
+
+ scanPlanCache.put(cacheKey, response);
+ return response;
+ }
+
+ /**
+ * Keeps the first {@link IcebergConfig#SCAN_PLAN_TASK_BATCH_SIZE} file scan
tasks of {@code
+ * fullPlan} inline and turns the remaining tasks into {@code plan-tasks}
tokens, so a single
+ * response never carries an unbounded plan.
+ *
+ * <p>Returns {@code fullPlan} unchanged when batching is disabled or the
plan already fits in one
+ * batch, which is the common case and keeps a plan a client can consume
without a second call.
+ */
+ @SuppressWarnings("deprecation")
+ private PlanTableScanResponse splitIntoPlanTasks(
+ TableIdentifier tableIdentifier,
+ PlanTableScanRequest scanRequest,
+ PlanTableScanResponse fullPlan) {
+ List<FileScanTask> allTasks = fullPlan.fileScanTasks();
+ if (scanPlanTaskBatchSize <= 0
+ || allTasks == null
+ || allTasks.size() <= scanPlanTaskBatchSize) {
+ return fullPlan;
+ }
+
+ List<String> planTasks = new ArrayList<>();
+ for (int offset = scanPlanTaskBatchSize;
+ offset < allTasks.size();
+ offset += scanPlanTaskBatchSize) {
+ planTasks.add(
+ PlanTaskToken.encode(tableIdentifier, scanRequest, offset,
scanPlanTaskBatchSize));
+ }
+
+ List<FileScanTask> firstBatch =
+ ImmutableList.copyOf(allTasks.subList(0, scanPlanTaskBatchSize));
+ LOG.info(
+ "Split scan plan of table {} into {} inline file scan tasks and {}
plan tasks",
+ tableIdentifier,
+ firstBatch.size(),
+ planTasks.size());
+
+ PlanTableScanResponse.Builder builder =
+ PlanTableScanResponse.builder()
+ .withPlanStatus(PlanStatus.COMPLETED)
+ .withFileScanTasks(firstBatch)
+ .withPlanTasks(planTasks)
+ .withSpecsById(fullPlan.specsById());
+ List<DeleteFile> deleteFiles = referencedDeleteFiles(firstBatch);
+ if (!deleteFiles.isEmpty()) {
+ builder.withDeleteFiles(deleteFiles);
+ }
+ return builder.build();
+ }
+
+ /**
+ * Pins the snapshot a scan request plans against, so that plan-task tokens
issued for the
+ * resulting plan keep resolving to the same snapshot after the table
changes.
+ *
+ * <p>Requests that already name a snapshot, incremental requests (which pin
a snapshot range) and
+ * scans of a table without a current snapshot are returned unchanged.
+ */
+ private static PlanTableScanRequest pinSnapshot(Table table,
PlanTableScanRequest scanRequest) {
+ boolean isIncremental =
Review Comment:
Done - the snapshot pinning is now inline in `planTableScan`, so
`pinSnapshot` is gone.
##########
iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java:
##########
@@ -82,9 +89,21 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
private final ScanPlanCache scanPlanCache;
+ /** Maximum number of file scan tasks handed out inline by one scan planning
response. */
+ private final int scanPlanTaskBatchSize;
+
private static final String DATA_ACCESS_VENDED_CREDENTIALS =
"vended-credentials";
private static final String DATA_ACCESS_REMOTE_SIGNING = "remote-signing";
+ /**
+ * Total order over file scan tasks, so that plan-task tokens, which address
tasks by position,
+ * keep pointing at the same tasks when a plan is recomputed.
+ */
+ private static final Comparator<FileScanTask> FILE_SCAN_TASK_ORDER =
Review Comment:
Done - the comparator is now built at the `fileScanTasks.sort(...)` call
site and the constant is gone.
##########
iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/PlanTaskToken.java:
##########
@@ -0,0 +1,166 @@
+/*
+ * 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.iceberg.service;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.Optional;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.PlanTableScanRequest;
+import org.apache.iceberg.rest.requests.PlanTableScanRequestParser;
+import org.apache.iceberg.util.JsonUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A {@code plan-task} token handed out by {@code POST
.../tables/{table}/plan} and redeemed at
+ * {@code POST .../tables/{table}/tasks}.
+ *
+ * <p>The token is self-describing: it carries the table, the scan request it
was planned from (with
+ * the snapshot pinned at planning time) and the slice of the planned file
scan tasks it stands for.
+ * Nothing about the token is stored server side, so it stays valid across
server restarts and can
+ * be redeemed by any Gravitino instance that serves the same catalog.
+ *
+ * <p>The token is opaque to clients but is not a capability: it grants no
access on its own. {@code
+ * POST .../tasks} authorizes the table in the request path and rejects a
token minted for a
+ * different table, so a forged token can at most express a scan the caller
could already submit
+ * through {@code POST .../plan}.
+ */
+class PlanTaskToken {
+
Review Comment:
Agreed, dropped `version`. The token is now exactly `table` + `offset` +
`limit` + `scan`. Version was only there for a future format change, which
`decode` already handles: anything that does not decode into those four fields
is treated as a token this server did not issue and reported as `404
NoSuchPlanTaskException`, so an old server meeting a future token behaves the
same way with or without the field.
--
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]
