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 abac270b35 [#11196] feat(maintenance): Collect manifest statistics in
single-writer table cycles (#13717)
abac270b35 is described below
commit abac270b35f5231895f6ff28b1ca0d60267d628b
Author: Akshay Thorat <[email protected]>
AuthorDate: Thu Oct 15 00:12:46 2026 -0700
[#11196] feat(maintenance): Collect manifest statistics in single-writer
table cycles (#13717)
### What changes were proposed in this pull request?
Rebase the collector directly onto current main and remove the
dependency on #13532. Collect every spec from one pinned table snapshot
with one `allManifests` traversal. Publish both complete maps and
`custom-manifest-snapshot-id` in one existing PUT/update call.
Default to last-write-wins publication without a coordination directory.
Consumers requiring fresh statistics must compare the published snapshot
ID with the table's current snapshot and skip/recollect on mismatch. An
optional shared coordination directory retains per-table locking through
publication and explicit recovery after ambiguous writes. Collection
failures are always rerunnable. The collector forces
`gravitino.client.retryTimes=0`; other clients retain the default of one
retry.
### Why are the changes needed?
Follow the single-writer direction agreed in #13532. Avoid scanning the
same manifest list once per spec and introducing a public merge API
solely for this collector. Related to #11196 and design #12942.
### Does this PR introduce _any_ user-facing change?
Adds `builtin-iceberg-update-manifest-stats` with optional
`coordination_directory`. Omitted or blank uses last-write-wins; opting
in requires a shared filesystem supporting durable files and cross-host
Java/POSIX locks. Collection is table-wide with no `spec_id` option.
Defined specs without manifests publish zeroes; each update replaces the
complete maps, removing stale entries. Snapshot ID is -1 when no
snapshot exists. Automatic policy triggering remains separate.
### How was this patch tested?
- 222 jobs tests, 19 optimizer API tests, and 30 focused Java client
tests passed.
- Coverage includes partition evolution, empty tables, one
snapshot/manifest pass, independent processes contending for the lock,
uncoordinated publication and reruns, safe reruns after collection
failures, ambiguous PUT recovery, auth alias propagation, and HTTP retry
counts of zero, one, and two.
- Formatting, jobs Javadocs, `:docs:build`, `rat`, and `git diff
--check` passed.
- Cross-host filesystem failure behavior was not exercised; deployments
must verify shared lock/durability semantics.
---
.../client/GravitinoClientConfiguration.java | 15 +
.../org/apache/gravitino/client/HTTPClient.java | 5 +
.../client/TestGravitinoClientConfiguration.java | 23 ++
.../apache/gravitino/client/TestHTTPClient.java | 29 ++
docs/how-to-use-gravitino-client.md | 1 +
.../optimizer-cli-reference.md | 108 +++++-
maintenance/jobs/build.gradle.kts | 2 +-
.../jobs/BuiltInJobTemplateProvider.java | 2 +
.../iceberg/IcebergUpdateManifestStatsJob.java | 306 ++++++++++++++++
.../jobs/TestBuiltInJobTemplateProvider.java | 8 +-
.../iceberg/TestIcebergUpdateManifestStatsJob.java | 198 +++++++++++
.../TestIcebergUpdateManifestStatsJobMain.java | 91 +++++
...TestIcebergUpdateManifestStatsJobWithSpark.java | 383 +++++++++++++++++++++
.../common/IcebergManifestStatistics.java | 129 +++++++
.../common/util/GravitinoClientUtils.java | 8 +
.../common/TestIcebergManifestStatistics.java | 86 +++++
.../common/util/TestGravitinoAuthSettings.java | 17 +
17 files changed, 1407 insertions(+), 4 deletions(-)
diff --git
a/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClientConfiguration.java
b/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClientConfiguration.java
index 03326ed912..edad75e0f2 100644
---
a/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClientConfiguration.java
+++
b/clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClientConfiguration.java
@@ -66,8 +66,12 @@ public class GravitinoClientConfiguration {
public static final String CLIENT_SOCKET_TIMEOUT_MS =
GRAVITINO_CLIENT_CONFIG_PREFIX + "socketTimeoutMs";
+ /** Maximum automatic HTTP retries after the initial request; zero disables
retries. */
+ public static final String CLIENT_RETRY_TIMES =
GRAVITINO_CLIENT_CONFIG_PREFIX + "retryTimes";
+
private static final Set<String> SUPPORT_CLIENT_CONFIG_KEYS =
ImmutableSet.of(
+ CLIENT_RETRY_TIMES,
CLIENT_CONNECTION_TIMEOUT_MS,
CLIENT_SOCKET_TIMEOUT_MS,
CLIENT_MAX_CONNECTIONS,
@@ -96,6 +100,17 @@ public class GravitinoClientConfiguration {
return config;
}
+ /**
+ * Returns the maximum automatic HTTP retries after the initial request.
+ *
+ * @return nonnegative retry count, defaulting to one as in Apache HttpClient
+ */
+ public int getClientRetryTimes() {
+ int retryTimes = MapUtils.propertyAsInt(properties, CLIENT_RETRY_TIMES, 1);
+ checkValue(value -> value >= 0, CLIENT_RETRY_TIMES, retryTimes,
POSITIVE_NUMBER_ERROR_MSG);
+ return retryTimes;
+ }
+
/**
* Extract connection timeout from the properties map
*
diff --git
a/clients/client-java/src/main/java/org/apache/gravitino/client/HTTPClient.java
b/clients/client-java/src/main/java/org/apache/gravitino/client/HTTPClient.java
index f7ff051433..daf3cc50dd 100644
---
a/clients/client-java/src/main/java/org/apache/gravitino/client/HTTPClient.java
+++
b/clients/client-java/src/main/java/org/apache/gravitino/client/HTTPClient.java
@@ -48,6 +48,7 @@ import org.apache.gravitino.rest.RESTUtils;
import org.apache.hc.client5.http.classic.methods.HttpUriRequest;
import org.apache.hc.client5.http.classic.methods.HttpUriRequestBase;
import org.apache.hc.client5.http.config.ConnectionConfig;
+import org.apache.hc.client5.http.impl.DefaultHttpRequestRetryStrategy;
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
import org.apache.hc.client5.http.impl.classic.HttpClientBuilder;
@@ -68,6 +69,7 @@ import org.apache.hc.core5.http.message.BasicHeader;
import org.apache.hc.core5.io.CloseMode;
import org.apache.hc.core5.net.URIBuilder;
import org.apache.hc.core5.reactor.ssl.SSLBufferMode;
+import org.apache.hc.core5.util.TimeValue;
/**
* An HttpClient for usage with the REST catalog.
@@ -148,6 +150,9 @@ public class HTTPClient implements RESTClient {
GravitinoClientConfiguration.buildFromProperties(properties);
HttpClientBuilder clientBuilder = HttpClients.custom();
+ clientBuilder.setRetryStrategy(
+ new DefaultHttpRequestRetryStrategy(
+ clientConfiguration.getClientRetryTimes(),
TimeValue.ofSeconds(1)));
clientBuilder.setConnectionManager(
configureConnectionManager(clientConfiguration, tlsConfigurer));
diff --git
a/clients/client-java/src/test/java/org/apache/gravitino/client/TestGravitinoClientConfiguration.java
b/clients/client-java/src/test/java/org/apache/gravitino/client/TestGravitinoClientConfiguration.java
index 20e10ab133..fb70b28e11 100644
---
a/clients/client-java/src/test/java/org/apache/gravitino/client/TestGravitinoClientConfiguration.java
+++
b/clients/client-java/src/test/java/org/apache/gravitino/client/TestGravitinoClientConfiguration.java
@@ -24,6 +24,29 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
public class TestGravitinoClientConfiguration {
+ @Test
+ void testAutomaticRetryConfiguration() {
+ Assertions.assertEquals(
+ 1,
+
GravitinoClientConfiguration.buildFromProperties(ImmutableMap.of()).getClientRetryTimes());
+ for (int count : new int[] {0, 1, 3}) {
+ Assertions.assertEquals(
+ count,
+ GravitinoClientConfiguration.buildFromProperties(
+ ImmutableMap.of(
+ GravitinoClientConfiguration.CLIENT_RETRY_TIMES,
Integer.toString(count)))
+ .getClientRetryTimes());
+ }
+ for (String invalid : new String[] {"-1", "invalid", "1.5", "2147483648"})
{
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ GravitinoClientConfiguration.buildFromProperties(
+
ImmutableMap.of(GravitinoClientConfiguration.CLIENT_RETRY_TIMES, invalid))
+ .getClientRetryTimes());
+ }
+ }
+
@Test
void testValidConfig() {
Map<String, String> properties =
diff --git
a/clients/client-java/src/test/java/org/apache/gravitino/client/TestHTTPClient.java
b/clients/client-java/src/test/java/org/apache/gravitino/client/TestHTTPClient.java
index 3d8515be9b..dc5db152f5 100644
---
a/clients/client-java/src/test/java/org/apache/gravitino/client/TestHTTPClient.java
+++
b/clients/client-java/src/test/java/org/apache/gravitino/client/TestHTTPClient.java
@@ -54,6 +54,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.mockserver.integration.ClientAndServer;
+import org.mockserver.model.HttpError;
import org.mockserver.model.HttpRequest;
import org.mockserver.model.HttpResponse;
@@ -100,6 +101,34 @@ public class TestHTTPClient {
testHttpMethodOnSuccess(Method.PUT, true, true);
}
+ /** Verifies ambiguous PUT connection failures can be surfaced without an
automatic resend. */
+ @Test
+ public void testPutRetryTimes() throws Exception {
+ for (int retryTimes : new int[] {0, 1, 2}) {
+ String path = "put-dropped-connection-" + retryTimes;
+ HttpRequest expected = request("/" + path).withMethod("PUT");
+
mockServer.when(expected).error(HttpError.error().withDropConnection(true));
+ try (RESTClient client =
+ HTTPClient.builder(
+ ImmutableMap.of(
+ GravitinoClientConfiguration.CLIENT_RETRY_TIMES,
+ Integer.toString(retryTimes)))
+ .uri("http://127.0.0.1:"; + mockServer.getPort())
+ .build()) {
+ Assertions.assertThrows(
+ RESTException.class,
+ () ->
+ client.put(
+ path,
+ new Item(1L, "stats"),
+ Item.class,
+ ImmutableMap.of(),
+ mock(ErrorHandler.class)));
+ }
+ Assertions.assertEquals(retryTimes + 1,
mockServer.retrieveRecordedRequests(expected).length);
+ }
+ }
+
@Test
public void testPutFailure() throws Exception {
testHttpMethodOnFailure(Method.PUT, true, true);
diff --git a/docs/how-to-use-gravitino-client.md
b/docs/how-to-use-gravitino-client.md
index 04bc0f2a6b..d0d1135f90 100644
--- a/docs/how-to-use-gravitino-client.md
+++ b/docs/how-to-use-gravitino-client.md
@@ -42,6 +42,7 @@ GravitinoAdminClient gravitinoAdminClient =
GravitinoAdminClient.builder("http:/
| Configuration item | Description
| Default value | Required |
|----------------------------------------|------------------------------------------------------|---------------------|----------|
| `gravitino.client.connectionTimeoutMs` | An optional http connection timeout
in milliseconds. | `180000`(3 minutes) | No |
+| `gravitino.client.retryTimes` | Maximum automatic HTTP retries after the
initial request. Must be a nonnegative integer; `0` disables retries. | `1` |
No |
| `gravitino.client.socketTimeoutMs` | An optional http socket timeout in
milliseconds. | `180000`(3 minutes) | No |
**Note:** Invalid configuration properties will result in exceptions.
diff --git a/docs/table-maintenance-service/optimizer-cli-reference.md
b/docs/table-maintenance-service/optimizer-cli-reference.md
index 26c989d799..90614f575b 100644
--- a/docs/table-maintenance-service/optimizer-cli-reference.md
+++ b/docs/table-maintenance-service/optimizer-cli-reference.md
@@ -237,7 +237,7 @@ EvaluationResult{scopeType=TABLE,
identifier=rest_catalog.db.t1, partitionPath=<
## Built-in Job Templates
-Five job templates ship with the service, and they are complementary rather
than alternatives. A full maintenance pass collects statistics, compacts data
files, expires the snapshot history that compaction just created, consolidates
manifests, and removes old orphan files.
+The following Iceberg job templates ship with the service. A full maintenance
pass collects statistics, compacts data files, expires the snapshot history
that compaction just created, consolidates manifests, and removes old orphan
files.
| Job template | What it does
|
|---------------------------------------|-------------------------------------------|
@@ -246,6 +246,7 @@ Five job templates ship with the service, and they are
complementary rather than
| `builtin-iceberg-expire-snapshots` | Removes old snapshot metadata
|
| `builtin-iceberg-remove-orphan-files` | Removes unreferenced files from
storage |
| `builtin-iceberg-rewrite-manifests` | Consolidates small manifest files
|
+| `builtin-iceberg-update-manifest-stats` | Collects complete per-spec
manifest maps |
Each can be submitted directly over REST, and the first two are also what the
policy-driven workflow submits on your behalf. See [Quick
Start](./optimizer.md#walkthrough) for the policy-driven path.
@@ -282,6 +283,111 @@ table, so it must be asked for explicitly.
Its `jobConf` is documented in
[Configuration](./optimizer-configuration.md#job-submission-configuration).
+## Update Manifest Statistics
+
+`builtin-iceberg-update-manifest-stats` is a separate Spark job that collects
two table-level
+statistics, including for partitioned tables:
+
+| Statistic | Object entry for each decimal spec ID |
+| --- | --- |
+| `custom-manifest-number-by-spec` | Long manifest count |
+| `custom-avg-manifest-size-by-spec` | Double average manifest size in bytes |
+
+The job reads manifest metadata without scanning file statistics. It works
after partition
+evolution, including for specs whose partition fields are absent from older
files.
+`builtin-iceberg-update-stats` and the `submit-update-stats-job` CLI retain
their existing behavior;
+they do not collect manifest statistics. Submit the new template directly
through the jobs API.
+
+Each table cycle pins one current snapshot, calls `allManifests` once, and
groups data and
+delete manifests by `partitionSpecId`. Every defined spec is included. A spec
with no manifests,
+including a table without a snapshot, produces count `0` and average `0.0`.
+The job also publishes `custom-manifest-snapshot-id` in the same update (`-1`
without a snapshot).
+Snapshot IDs identify measurements; they are not monotonically increasing
version numbers.
+
+The template accepts `catalog_name`, `table_identifier`, `updater_options`,
and optional
+`coordination_directory` and `spark_conf`, alongside the standard Spark
template settings. There is no `spec_id` filter:
+collection always replaces the complete maps for the table.
+`updater_options` is a JSON string containing `gravitino_uri` and `metalake`;
it can also select
+a `statistics_updater` implementing the existing table statistics update
operation.
+For example:
+
+```bash
+curl -X POST http://localhost:8090/api/metalakes/test/jobs/runs \
+ -H 'Content-Type: application/json' \
+ -d '{
+ "jobTemplateName": "builtin-iceberg-update-manifest-stats",
+ "jobConf": {
+ "catalog_name": "rest_catalog",
+ "table_identifier": "db.t1",
+ "updater_options":
"{\"gravitino_uri\":\"http://localhost:8090\",\"metalake\":\"test\"}";,
+ "spark_master": "local[2]",
+ "spark_executor_instances": "1",
+ "spark_executor_cores": "1",
+ "spark_executor_memory": "1g",
+ "spark_driver_memory": "1g",
+ "catalog_type": "rest",
+ "catalog_uri": "http://localhost:9001/iceberg";,
+ "warehouse_location": ""
+ }
+ }'
+```
+
+For example, one collection can produce complete maps `{"0": 620, "1": 120}`
and
+`{"0": 10485760.0, "1": 4194304.0}`. The collector owns both statistic names
and publishes
+them together with the snapshot ID in one existing PUT/update call. Old map
entries are
+replaced, not merged. Other statistic names are unchanged. No PATCH or merge
API is needed.
+
+By default, `coordination_directory` is omitted or blank and publications use
last-write-wins
+semantics. Each update contains the complete maps and snapshot ID; a
concurrent older collection
+may overwrite newer measurements. Failed jobs can simply be rerun, with no
lock files or recovery.
+Consumers requiring fresh statistics must compare
`custom-manifest-snapshot-id` with the table's
+current snapshot ID (`-1` if none) and skip/recollect on a mismatch. This is a
consumer requirement,
+not a freshness check enforced by this collector. Snapshot IDs must be
compared for equality,
+not numerical ordering.
+
+### Optional shared coordination and recovery
+
+For stricter serialization, provision `coordination_directory`
(`--coordination-dir` for direct Spark invocation) as an
+existing absolute directory on a shared filesystem that supports durable files
and cross-host
+Java/POSIX file locks. Every runner for a destination table must use the same
shared directory,
+metalake, and canonical catalog/schema/table names. A host-local directory,
object-store mount,
+or filesystem without reliable cross-host locking does not satisfy this
requirement.
+All writers of these statistic names must opt into this mode and use the same
directory;
+mixing coordinated and uncoordinated writers does not provide serialization.
+
+The job locks a persistent file keyed by metalake and table before refreshing
the Iceberg table
+or choosing a snapshot, and holds it through the PUT. Concurrent runs for that
table fail;
+other tables can proceed. Snapshot IDs are recorded for provenance, not
compared numerically.
+The job forces `gravitino.client.retryTimes=0` for its Gravitino client.
+Custom statistics updaters and any intervening proxies must likewise avoid
retrying PUT requests;
+an automatic resend could hide an original request that is still executing.
There are no expiring
+leases, automatic retries, scheduling, or cooldowns.
+
+Collection failures can be rerun normally: the lock is released without
leaving a recovery marker.
+After collection and immediately before calling the updater, the job durably
writes an in-progress
+marker. It clears the marker only after a successful update. A crash after the
marker is written or
+an update failure leaves the marker in place, even after the operating system
releases the lock.
+The updater interface cannot distinguish a failed request from a request still
executing on the
+server, so these failures require recovery. The error reports the recovery
file path.
+
+For example, run A sends a PUT and loses its connection while the server is
still processing it.
+An immediate rerun B could publish newer measurements before A's delayed PUT
overwrites them.
+The file lock prevents concurrent collection across hosts; the marker prevents
this overlap after
+an ambiguous failure. Snapshot IDs provide provenance, but the existing PUT
does not enforce
+snapshot ordering or fencing. Automatic reruns after ambiguous writes would
require server-side
+conditional publication or an equivalent guarantee that the earlier request
has ended.
+
+For recovery, stop all runners for that table and verify the previous process
has terminated
+and any outstanding Gravitino PUT has completed or been definitively
cancelled. While holding
+an exclusive lock on the reported file, truncate it to zero bytes and flush it
to durable storage,
+then resume collection. Never delete or replace the file: a new inode can
permit two independent
+locks. If the previous PUT outcome cannot be established, leave the marker
intact.
+
+Consumers must read both objects from one table-statistics response and use
the same chosen
+spec key in each. A missing object or key means collection is required, not
zero manifests.
+`IcebergManifestStatistics.fromStatistics` returns an empty result for an
incomplete pair.
+Automatic manifest policy triggering is separate follow-up work.
+
## Rewrite Data Files
`builtin-iceberg-rewrite-data-files` performs the compaction itself, merging
small data files into larger ones. It is what a compaction policy submits when
its thresholds are crossed.
diff --git a/maintenance/jobs/build.gradle.kts
b/maintenance/jobs/build.gradle.kts
index 49615a7d97..ae5055bda8 100644
--- a/maintenance/jobs/build.gradle.kts
+++ b/maintenance/jobs/build.gradle.kts
@@ -36,6 +36,7 @@ val sparkMajorVersion = "3.5"
dependencies {
implementation(project(":api"))
+ implementation(project(":clients:client-java"))
implementation(project(":maintenance:optimizer-api"))
implementation(project(":maintenance:updaters"))
@@ -56,7 +57,6 @@ dependencies {
testImplementation(project(":common"))
// Used to check that the placeholders of the built-in templates resolve the
way the server does.
testImplementation(project(":core"))
- testImplementation(project(":clients:client-java"))
testImplementation(libs.bundles.log4j)
testImplementation(libs.hadoop3.common) {
exclude("org.slf4j")
diff --git
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/BuiltInJobTemplateProvider.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/BuiltInJobTemplateProvider.java
index df7cae0888..6b26dce55e 100644
---
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/BuiltInJobTemplateProvider.java
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/BuiltInJobTemplateProvider.java
@@ -29,6 +29,7 @@ import
org.apache.gravitino.maintenance.jobs.iceberg.IcebergExpireSnapshotsJob;
import
org.apache.gravitino.maintenance.jobs.iceberg.IcebergRemoveOrphanFilesJob;
import
org.apache.gravitino.maintenance.jobs.iceberg.IcebergRewriteDataFilesJob;
import
org.apache.gravitino.maintenance.jobs.iceberg.IcebergRewriteManifestsJob;
+import
org.apache.gravitino.maintenance.jobs.iceberg.IcebergUpdateManifestStatsJob;
import
org.apache.gravitino.maintenance.jobs.iceberg.IcebergUpdateStatsAndMetricsJob;
import org.apache.gravitino.maintenance.jobs.spark.SparkPiJob;
import org.slf4j.Logger;
@@ -50,6 +51,7 @@ public class BuiltInJobTemplateProvider implements
JobTemplateProvider {
new IcebergRewriteDataFilesJob(),
new IcebergRewriteManifestsJob(),
new IcebergUpdateStatsAndMetricsJob(),
+ new IcebergUpdateManifestStatsJob(),
new IcebergExpireSnapshotsJob(),
new IcebergRemoveOrphanFilesJob());
diff --git
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateManifestStatsJob.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateManifestStatsJob.java
new file mode 100644
index 0000000000..2f398aa7d4
--- /dev/null
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateManifestStatsJob.java
@@ -0,0 +1,306 @@
+/*
+ * 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.maintenance.jobs.iceberg;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.hash.Hashing;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.client.GravitinoClientConfiguration;
+import org.apache.gravitino.job.JobTemplateProvider;
+import org.apache.gravitino.job.SparkJobTemplate;
+import org.apache.gravitino.maintenance.jobs.BuiltInJob;
+import org.apache.gravitino.maintenance.optimizer.api.common.StatisticEntry;
+import
org.apache.gravitino.maintenance.optimizer.api.updater.StatisticsUpdater;
+import
org.apache.gravitino.maintenance.optimizer.common.IcebergManifestStatistics;
+import org.apache.gravitino.maintenance.optimizer.common.OptimizerEnv;
+import org.apache.gravitino.maintenance.optimizer.common.StatisticEntryImpl;
+import org.apache.gravitino.maintenance.optimizer.common.conf.OptimizerConfig;
+import
org.apache.gravitino.maintenance.optimizer.common.util.GravitinoAuthSettings;
+import
org.apache.gravitino.maintenance.optimizer.common.util.IcebergSparkConfigUtils;
+import org.apache.gravitino.maintenance.optimizer.common.util.ProviderUtils;
+import org.apache.gravitino.stats.StatisticValue;
+import org.apache.gravitino.stats.StatisticValues;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Built-in job for collecting all Iceberg manifest statistics in one table
cycle. */
+public class IcebergUpdateManifestStatsJob implements BuiltInJob {
+ private static final Logger LOG =
LoggerFactory.getLogger(IcebergUpdateManifestStatsJob.class);
+ private static final String NAME =
+ JobTemplateProvider.BUILTIN_NAME_PREFIX +
"iceberg-update-manifest-stats";
+ private static final String DEFAULT_STATISTICS_UPDATER =
"gravitino-statistics-updater";
+
+ @Override
+ public SparkJobTemplate jobTemplate() {
+ return SparkJobTemplate.builder()
+ .withName(NAME)
+ .withComment("Collect manifest count and average size for every
Iceberg partition spec")
+ .withExecutable(resolveExecutable(IcebergUpdateManifestStatsJob.class))
+ .withClassName(IcebergUpdateManifestStatsJob.class.getName())
+ .withArguments(
+ Arrays.asList(
+ "--catalog",
+ "{{catalog_name}}",
+ "--table",
+ "{{table_identifier}}",
+ "--coordination-dir",
+ "{{coordination_directory:-}}",
+ "--updater-options",
+ "{{updater_options}}",
+ "--spark-conf",
+ "{{spark_conf:-}}"))
+ .withConfigs(IcebergSparkConfigUtils.buildTemplateSparkConfigs())
+
.withCustomFields(Collections.singletonMap(JobTemplateProvider.PROPERTY_VERSION_KEY,
"v1"))
+ .build();
+ }
+
+ /**
+ * Collects and persists manifest statistics using the configured Spark
catalog and updater.
+ *
+ * @param args catalog, table, optional shared coordination directory,
updater options, and Spark
+ * configuration
+ * @throws Exception if collection, persistence, or resource cleanup fails
+ */
+ public static void main(String[] args) throws Exception {
+ Map<String, String> arguments =
+ IcebergJobUtils.parseArguments(
+ args,
+ ImmutableSet.of(
+ "catalog", "table", "updater-options", "spark-conf",
"coordination-dir"),
+ ImmutableSet.of("catalog", "table", "updater-options"));
+ String coordinationDirectory = arguments.get("coordination-dir");
+ Path coordinationPath =
+ coordinationDirectory == null || coordinationDirectory.trim().isEmpty()
+ ? null
+ : Paths.get(coordinationDirectory);
+ String catalog = arguments.get("catalog");
+ String table = arguments.get("table");
+ tableIdentifier(catalog, table);
+ Map<String, String> options =
+ IcebergSparkConfigUtils.parseFlatJsonMap(
+ arguments.get("updater-options"), "updater-options");
+ Map<String, String> properties = buildOptimizerProperties(options);
+ Map<String, String> sparkConfigs =
+ IcebergSparkConfigUtils.parseFlatJsonMap(arguments.get("spark-conf"),
"spark-conf");
+ try (StatisticsUpdater updater =
+ ProviderUtils.createStatisticsUpdaterInstance(
+ options.getOrDefault("statistics_updater",
DEFAULT_STATISTICS_UPDATER).trim())) {
+ updater.initialize(new OptimizerEnv(new OptimizerConfig(properties)));
+ SparkSession.Builder builder =
+ SparkSession.builder().appName("Gravitino Iceberg Manifest
Statistics");
+ sparkConfigs.forEach(builder::config);
+ SparkSession spark = builder.getOrCreate();
+ try {
+ IcebergJobUtils.requireIcebergSparkRuntime();
+ updateStatistics(
+ spark,
+ updater,
+ catalog,
+ table,
+ properties.get(OptimizerConfig.GRAVITINO_METALAKE),
+ coordinationPath);
+ } finally {
+ spark.stop();
+ }
+ }
+ }
+
+ /**
+ * Collects complete per-spec measurements from one pinned table snapshot.
+ *
+ * @param spark Spark session with the Iceberg catalog configured
+ * @param catalogName Spark catalog name
+ * @param tableIdentifier schema.table identifier
+ * @return both complete maps and the snapshot ID (-1 for a table without a
snapshot)
+ */
+ public static List<StatisticEntry<?>> collectManifestStatistics(
+ SparkSession spark, String catalogName, String tableIdentifier) {
+ String identifier = buildTableIdentifier(catalogName, tableIdentifier);
+ final Table table;
+ try {
+ table = Spark3Util.loadIcebergTable(spark, identifier);
+ table.refresh();
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Cannot load Iceberg table " +
identifier, e);
+ }
+ return collectManifestStatistics(table);
+ }
+
+ @VisibleForTesting
+ static List<StatisticEntry<?>> collectManifestStatistics(Table table) {
+ Snapshot snapshot = table.currentSnapshot();
+ Map<Integer, Long> counts = new HashMap<>();
+ Map<Integer, Double> sizes = new HashMap<>();
+ table.specs().keySet().forEach(spec -> counts.put(spec, 0L));
+ if (snapshot != null) {
+ for (ManifestFile manifest : snapshot.allManifests(table.io())) {
+ int spec = manifest.partitionSpecId();
+ counts.merge(spec, 1L, Long::sum);
+ sizes.merge(spec, (double) manifest.length(), Double::sum);
+ }
+ }
+ Map<String, StatisticValue<?>> countValues = new HashMap<>();
+ Map<String, StatisticValue<?>> averageValues = new HashMap<>();
+ counts.forEach(
+ (spec, count) -> {
+ String key = Integer.toString(spec);
+ countValues.put(key, StatisticValues.longValue(count));
+ averageValues.put(
+ key, StatisticValues.doubleValue(count == 0 ? 0D :
sizes.get(spec) / count));
+ });
+ return Arrays.asList(
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.MANIFEST_NUMBER,
StatisticValues.objectValue(countValues)),
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.AVG_MANIFEST_SIZE,
+ StatisticValues.objectValue(averageValues)),
+ new StatisticEntryImpl<>(
+ IcebergManifestStatistics.SNAPSHOT_ID,
+ StatisticValues.longValue(snapshot == null ? -1L :
snapshot.snapshotId())));
+ }
+
+ @VisibleForTesting
+ static void updateStatistics(
+ SparkSession spark,
+ StatisticsUpdater updater,
+ String catalogName,
+ String tableIdentifier,
+ String metalake,
+ @Nullable Path coordinationDirectory)
+ throws IOException {
+ NameIdentifier identifier = tableIdentifier(catalogName, tableIdentifier);
+ if (coordinationDirectory == null) {
+ updater.updateTableStatistics(
+ identifier, collectManifestStatistics(spark, catalogName,
tableIdentifier));
+ LOG.info("Updated complete manifest statistics for {}", identifier);
+ return;
+ }
+ if (!coordinationDirectory.isAbsolute() ||
!Files.isDirectory(coordinationDirectory)) {
+ throw new IllegalArgumentException(
+ "coordination-dir must be an existing absolute shared directory");
+ }
+ String key =
+ Hashing.sha256()
+ .hashString(
+ metalake + "\0" + catalogName + "\0" + tableIdentifier,
StandardCharsets.UTF_8)
+ .toString();
+ Path lockPath = coordinationDirectory.resolve(key + ".lock");
+ // Never delete lock files: replacing the inode lets two hosts lock
different files.
+ try (FileChannel channel =
+ FileChannel.open(
+ lockPath,
+ StandardOpenOption.CREATE,
+ StandardOpenOption.READ,
+ StandardOpenOption.WRITE);
+ FileLock lock = channel.tryLock()) {
+ if (lock == null) {
+ throw new IllegalStateException("Manifest collection is already
running for " + identifier);
+ }
+ if (channel.size() != 0) {
+ throw new IllegalStateException(
+ "Unfinished manifest collection requires recovery: " + lockPath);
+ }
+ List<StatisticEntry<?>> measurements =
+ collectManifestStatistics(spark, catalogName, tableIdentifier);
+ byte[] marker =
+ ("In-progress manifest collection for "
+ + metalake
+ + "."
+ + identifier
+ + "; verify the previous process and any outstanding PUT
have finished before recovery.\n")
+ .getBytes(StandardCharsets.UTF_8);
+ ByteBuffer buffer = ByteBuffer.wrap(marker);
+ while (buffer.hasRemaining()) {
+ channel.write(buffer);
+ }
+ channel.force(true);
+ updater.updateTableStatistics(identifier, measurements);
+ // A failed or ambiguous PUT leaves the marker intact, preventing an
unsafe automatic retry.
+ channel.truncate(0);
+ channel.force(true);
+ LOG.info("Updated complete manifest statistics for {}", identifier);
+ }
+ }
+
+ @VisibleForTesting
+ static Map<String, String> buildOptimizerProperties(Map<String, String>
options) {
+ Map<String, String> properties = new HashMap<>(options);
+ if (options.containsKey("gravitino_uri")) {
+ properties.put(OptimizerConfig.GRAVITINO_URI,
options.get("gravitino_uri"));
+ }
+ if (options.containsKey("metalake")) {
+ properties.put(OptimizerConfig.GRAVITINO_METALAKE,
options.get("metalake"));
+ }
+ for (String key :
+ Arrays.asList(OptimizerConfig.GRAVITINO_URI,
OptimizerConfig.GRAVITINO_METALAKE)) {
+ String value = properties.get(key);
+ if (value == null || value.trim().isEmpty()) {
+ throw new IllegalArgumentException("updater_options must configure " +
key);
+ }
+ properties.put(key, value.trim());
+ }
+ // A retried PUT can hide an older request that is still executing after a
connection failure.
+ properties.put(GravitinoClientConfiguration.CLIENT_RETRY_TIMES, "0");
+ GravitinoAuthSettings.copyAliases(properties);
+ return properties;
+ }
+
+ @VisibleForTesting
+ static String buildTableIdentifier(String catalogName, String tableName) {
+ NameIdentifier identifier = tableIdentifier(catalogName, tableName);
+ return IcebergJobUtils.escapeSqlIdentifier(identifier.namespace().level(0))
+ + "."
+ + IcebergJobUtils.escapeSqlIdentifier(identifier.namespace().level(1))
+ + "."
+ + IcebergJobUtils.escapeSqlIdentifier(identifier.name());
+ }
+
+ private static NameIdentifier tableIdentifier(
+ @Nullable String catalogName, @Nullable String tableName) {
+ if (catalogName == null || catalogName.trim().isEmpty() || tableName ==
null) {
+ throw new IllegalArgumentException("--catalog and --table are required");
+ }
+ String[] levels = tableName.split("\\.", -1);
+ if (levels.length != 2 || levels[0].trim().isEmpty() ||
levels[1].trim().isEmpty()) {
+ throw new IllegalArgumentException("--table must use schema.table
format: " + tableName);
+ }
+ return NameIdentifier.of(catalogName, levels[0], levels[1]);
+ }
+}
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/TestBuiltInJobTemplateProvider.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/TestBuiltInJobTemplateProvider.java
index 7f38c34966..4ed6c75093 100644
---
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/TestBuiltInJobTemplateProvider.java
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/TestBuiltInJobTemplateProvider.java
@@ -89,8 +89,9 @@ public class TestBuiltInJobTemplateProvider {
@Test
public void testTemplatesOnlyRequireTheParametersTheyHaveTo() {
// A built-in template must let a caller run it with the catalog and the
table alone, so every
- // other parameter needs a default value. The catalog connection has no
default that fits every
- // catalog, and an empty where clause rewrites a whole table, so those
stay required.
+ // other parameter needs a default value. Catalog and updater connections
have no defaults that
+ // fit every deployment, and an empty where clause rewrites a whole table,
so those stay
+ // required.
Map<String, Set<String>> expected = new HashMap<>();
Set<String> icebergCatalog =
ImmutableSet.of(
@@ -101,6 +102,9 @@ public class TestBuiltInJobTemplateProvider {
"warehouse_location");
expected.put("builtin-sparkpi", ImmutableSet.of());
expected.put("builtin-iceberg-update-stats", icebergCatalog);
+ expected.put(
+ "builtin-iceberg-update-manifest-stats",
+
ImmutableSet.<String>builder().addAll(icebergCatalog).add("updater_options").build());
expected.put("builtin-iceberg-expire-snapshots", icebergCatalog);
expected.put("builtin-iceberg-remove-orphan-files", icebergCatalog);
expected.put("builtin-iceberg-rewrite-manifests", icebergCatalog);
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateManifestStatsJob.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateManifestStatsJob.java
new file mode 100644
index 0000000000..24b1efe30e
--- /dev/null
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateManifestStatsJob.java
@@ -0,0 +1,198 @@
+/*
+ * 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.maintenance.jobs.iceberg;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.sun.net.httpserver.HttpServer;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.client.GravitinoClientConfiguration;
+import org.apache.gravitino.job.JobTemplate;
+import org.apache.gravitino.job.SparkJobTemplate;
+import org.apache.gravitino.maintenance.jobs.BuiltInJobTemplateProvider;
+import org.apache.gravitino.maintenance.optimizer.common.OptimizerEnv;
+import org.apache.gravitino.maintenance.optimizer.common.StatisticEntryImpl;
+import org.apache.gravitino.maintenance.optimizer.common.conf.OptimizerConfig;
+import
org.apache.gravitino.maintenance.optimizer.updater.statistics.GravitinoStatisticsUpdater;
+import org.apache.gravitino.stats.StatisticValues;
+import org.junit.jupiter.api.Test;
+
+/** Tests the standalone manifest statistics template and argument validation.
*/
+public class TestIcebergUpdateManifestStatsJob {
+ /** Verifies manifest statistics use an independent built-in template. */
+ @Test
+ public void testIndependentTemplateRegistration() {
+ Map<String, JobTemplate> templates =
+ new BuiltInJobTemplateProvider()
+ .jobTemplates().stream()
+ .collect(Collectors.toMap(JobTemplate::name, template ->
template));
+ SparkJobTemplate template =
+ (SparkJobTemplate)
templates.get("builtin-iceberg-update-manifest-stats");
+ assertNotNull(template);
+ assertEquals(IcebergUpdateManifestStatsJob.class.getName(),
template.className());
+ assertEquals(10, template.arguments().size());
+ assertFalse(template.arguments().contains("--spec-id"));
+ assertTrue(template.arguments().contains("{{spark_conf:-}}"));
+ assertTrue(template.arguments().contains("{{coordination_directory:-}}"));
+ assertFalse(template.arguments().contains("--update-mode"));
+ assertEquals("v1", template.customFields().get("version"));
+ assertTrue(template.configs().containsKey("spark.sql.extensions"));
+ SparkJobTemplate existing = (SparkJobTemplate)
templates.get("builtin-iceberg-update-stats");
+ assertNotNull(existing);
+ assertFalse(existing.arguments().contains("--spec-id"));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> IcebergUpdateStatsAndMetricsJob.parseUpdateMode("manifests"));
+ }
+
+ /** Verifies catalog and table identifiers are validated and escaped. */
+ @Test
+ public void testTableIdentifierValidationAndEscaping() {
+ assertEquals(
+ "`cat``alog`.`db`.`tbl``name`",
+ IcebergUpdateManifestStatsJob.buildTableIdentifier("cat`alog",
"db.tbl`name"));
+ for (String invalid : Arrays.asList("table", "db.", ".table",
"db.table.extra", " ")) {
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> IcebergUpdateManifestStatsJob.buildTableIdentifier("catalog",
invalid));
+ }
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> IcebergUpdateManifestStatsJob.main(new String[] {"--catalog",
"catalog"}));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> IcebergUpdateManifestStatsJob.main(new String[] {"--table",
"db.table"}));
+ }
+
+ /** Verifies short auth options reach HTTP requests made by the real
statistics updater. */
+ @Test
+ public void testUpdaterSendsBasicAuthFromAliases() throws Exception {
+ AtomicReference<String> authorization = new AtomicReference<>();
+ HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1",
0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+
authorization.set(exchange.getRequestHeaders().getFirst("Authorization"));
+ exchange.sendResponseHeaders(401, -1);
+ exchange.close();
+ });
+ server.start();
+ try (GravitinoStatisticsUpdater updater = new
GravitinoStatisticsUpdater()) {
+ Map<String, String> options = new HashMap<>();
+ options.put("gravitino_uri", "http://127.0.0.1:"; +
server.getAddress().getPort());
+ options.put("metalake", "test");
+ options.put("auth_type", "basic");
+ options.put("username", "alice");
+ options.put("password", "test-password");
+ updater.initialize(
+ new OptimizerEnv(
+ new OptimizerConfig(
+
IcebergUpdateManifestStatsJob.buildOptimizerProperties(options))));
+ // Stop at the HTTP boundary after capturing the credentials sent by the
updater.
+ assertThrows(
+ RuntimeException.class,
+ () ->
+ updater.updateTableStatistics(
+ NameIdentifier.of("catalog", "db", "table"),
+ Collections.singletonList(
+ new StatisticEntryImpl<>("custom-count",
StatisticValues.longValue(1)))));
+ assertEquals(
+ "Basic "
+ + Base64.getEncoder()
+
.encodeToString("alice:test-password".getBytes(StandardCharsets.UTF_8)),
+ authorization.get());
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ /** Verifies auth aliases are promoted without overriding canonical
settings. */
+ @Test
+ public void testAuthAliases() {
+ Map<String, String> options = new HashMap<>();
+ options.put("gravitino_uri", "http://localhost:8090";);
+ options.put("metalake", "test");
+ options.put("auth_type", "oauth");
+ options.put("oauth_server_uri", "http://localhost:9000";);
+ options.put("oauth_path", "token");
+ options.put("oauth_credential", "client:secret");
+ options.put("oauth_scope", "catalog");
+ Map<String, String> properties =
+ IcebergUpdateManifestStatsJob.buildOptimizerProperties(options);
+ assertEquals("oauth", properties.get(OptimizerConfig.AUTH_TYPE));
+ assertEquals("http://localhost:9000";,
properties.get(OptimizerConfig.AUTH_OAUTH_SERVER_URI));
+ assertEquals("token", properties.get(OptimizerConfig.AUTH_OAUTH_PATH));
+ assertEquals("client:secret",
properties.get(OptimizerConfig.AUTH_OAUTH_CREDENTIAL));
+ assertEquals("catalog", properties.get(OptimizerConfig.AUTH_OAUTH_SCOPE));
+ assertFalse(options.containsKey(OptimizerConfig.AUTH_TYPE));
+ options.put(OptimizerConfig.AUTH_TYPE, "none");
+ assertEquals(
+ "none",
+ IcebergUpdateManifestStatsJob.buildOptimizerProperties(options)
+ .get(OptimizerConfig.AUTH_TYPE));
+ }
+
+ /** Verifies updater configuration aliases and required properties. */
+ @Test
+ public void testUpdaterConfiguration() {
+ Map<String, String> options = new HashMap<>();
+ options.put("gravitino_uri", " http://localhost:8090 ");
+ options.put("metalake", " test ");
+ options.put(GravitinoClientConfiguration.CLIENT_RETRY_TIMES, "2");
+ Map<String, String> properties =
+ IcebergUpdateManifestStatsJob.buildOptimizerProperties(options);
+ assertEquals("0",
properties.get(GravitinoClientConfiguration.CLIENT_RETRY_TIMES));
+ assertEquals("http://localhost:8090";,
properties.get(OptimizerConfig.GRAVITINO_URI));
+ assertEquals("test", properties.get(OptimizerConfig.GRAVITINO_METALAKE));
+ assertEquals(" test ", options.get("metalake"));
+ properties.remove("metalake");
+ properties.remove("gravitino_uri");
+ assertEquals(properties,
IcebergUpdateManifestStatsJob.buildOptimizerProperties(properties));
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
IcebergUpdateManifestStatsJob.buildOptimizerProperties(Collections.emptyMap()));
+ options.put("metalake", " ");
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> IcebergUpdateManifestStatsJob.buildOptimizerProperties(options));
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ IcebergUpdateManifestStatsJob.main(
+ new String[] {"--catalog", "catalog", "--table", "db.table",
"--spec-id", "-1"}));
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ IcebergUpdateManifestStatsJob.main(
+ new String[] {"--catalog", "catalog", "--table", "db.table"}));
+ }
+}
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateManifestStatsJobMain.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateManifestStatsJobMain.java
new file mode 100644
index 0000000000..27a4c9ef95
--- /dev/null
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateManifestStatsJobMain.java
@@ -0,0 +1,91 @@
+/*
+ * 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.maintenance.jobs.iceberg;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/** Tests manifest statistics job startup in a Spark environment without
Iceberg. */
+public class TestIcebergUpdateManifestStatsJobMain {
+ @TempDir Path tempDir;
+
+ /** Verifies a missing Iceberg runtime fails with an actionable classpath
error. */
+ @Test
+ public void testMissingRuntimeHasActionableError() throws Exception {
+ checkMissingRuntime(false);
+ }
+
+ /** Verifies an explicitly blank coordination directory reaches normal job
startup. */
+ @Test
+ public void testBlankCoordinationDirectory() throws Exception {
+ checkMissingRuntime(true);
+ }
+
+ private void checkMissingRuntime(boolean blankCoordination) throws Exception
{
+ String classpath =
+
Arrays.stream(System.getProperty("java.class.path").split(File.pathSeparator))
+ .filter(entry -> !entry.contains("iceberg-spark-runtime"))
+ .collect(Collectors.joining(File.pathSeparator));
+ List<String> command = new ArrayList<>();
+ command.add(new File(System.getProperty("java.home"),
"bin/java").toString());
+ command.add("--add-opens=java.base/sun.nio.ch=ALL-UNNAMED");
+ command.add("-cp");
+ command.add(classpath);
+ command.add(IcebergUpdateManifestStatsJob.class.getName());
+ command.addAll(
+ Arrays.asList(
+ "--catalog",
+ "cli",
+ "--table",
+ "db.cli",
+ "--updater-options",
+
"{\"gravitino_uri\":\"http://localhost:8090\",\"metalake\":\"test\"}";,
+ "--spark-conf",
+ "{\"spark.master\":\"local[1]\",\"spark.ui.enabled\":\"false\","
+ + "\"spark.sql.extensions\":"
+ +
"\"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions\"}"));
+ if (blankCoordination) {
+ command.addAll(Arrays.asList("--coordination-dir", " "));
+ }
+ Path log = tempDir.resolve("missing-runtime.log");
+ Process process =
+ new
ProcessBuilder(command).redirectErrorStream(true).redirectOutput(log.toFile()).start();
+ try {
+ assertTrue(process.waitFor(45, TimeUnit.SECONDS), "Manifest statistics
job did not exit");
+ String output = new String(Files.readAllBytes(log),
StandardCharsets.UTF_8);
+ assertEquals(1, process.exitValue(), output);
+ assertTrue(output.contains("Missing Iceberg Spark session extensions"),
output);
+ assertTrue(output.contains("iceberg-spark-runtime"), output);
+ } finally {
+ process.destroyForcibly();
+ }
+ }
+}
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateManifestStatsJobWithSpark.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateManifestStatsJobWithSpark.java
new file mode 100644
index 0000000000..ff05e76b59
--- /dev/null
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateManifestStatsJobWithSpark.java
@@ -0,0 +1,383 @@
+/*
+ * 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.maintenance.jobs.iceberg;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.File;
+import java.lang.reflect.Proxy;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.maintenance.optimizer.api.common.PartitionPath;
+import org.apache.gravitino.maintenance.optimizer.api.common.StatisticEntry;
+import
org.apache.gravitino.maintenance.optimizer.api.updater.StatisticsUpdater;
+import
org.apache.gravitino.maintenance.optimizer.common.IcebergManifestStatistics;
+import org.apache.gravitino.maintenance.optimizer.common.OptimizerEnv;
+import org.apache.gravitino.stats.StatisticValue;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/** Tests manifest statistics collection against real Spark and Iceberg
tables. */
+public class TestIcebergUpdateManifestStatsJobWithSpark {
+ private static final String CATALOG_NAME = "manifest_catalog";
+ @TempDir static File tempDir;
+ private static SparkSession spark;
+
+ /** Starts Spark with a local Iceberg catalog for the tests. */
+ @BeforeAll
+ public static void setUp() {
+ spark =
+ SparkSession.builder()
+ .master("local[2]")
+ .appName("TestIcebergUpdateManifestStatsJob")
+ .config(
+ "spark.sql.extensions",
+
"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
+ .config("spark.sql.catalog." + CATALOG_NAME,
"org.apache.iceberg.spark.SparkCatalog")
+ .config("spark.sql.catalog." + CATALOG_NAME + ".type", "hadoop")
+ .config(
+ "spark.sql.catalog." + CATALOG_NAME + ".warehouse",
+ new File(tempDir, "warehouse").getAbsolutePath())
+ .getOrCreate();
+ spark.sql("CREATE NAMESPACE " + CATALOG_NAME + ".db");
+ }
+
+ /** Stops the shared Spark session. */
+ @AfterAll
+ public static void tearDown() {
+ if (spark != null) {
+ spark.stop();
+ }
+ }
+
+ /** Verifies all specs are measured together and published in one
replacement update. */
+ @Test
+ public void testManifestStatisticsAcrossPartitionEvolution() throws
Exception {
+ String name = CATALOG_NAME + ".db.manifest_evolution";
+ spark.sql("CREATE TABLE " + name + " (id INT, ds STRING) USING iceberg
PARTITIONED BY (ds)");
+ try {
+ spark.sql("INSERT INTO " + name + " VALUES (1, 'a'), (2, 'b')");
+ spark.sql("ALTER TABLE " + name + " ADD PARTITION FIELD bucket(4, id)");
+ spark.sql("INSERT INTO " + name + " VALUES (3, 'c')");
+ spark.sql("ALTER TABLE " + name + " ADD PARTITION FIELD truncate(2,
ds)");
+ Table table = Spark3Util.loadIcebergTable(spark, name);
+ Snapshot pinned = table.currentSnapshot();
+ AtomicInteger snapshotReads = new AtomicInteger();
+ AtomicInteger manifestReads = new AtomicInteger();
+ Snapshot snapshot =
+ (Snapshot)
+ Proxy.newProxyInstance(
+ Snapshot.class.getClassLoader(),
+ new Class<?>[] {Snapshot.class},
+ (proxy, method, args) -> {
+ if (method.getName().equals("allManifests")) {
+ manifestReads.incrementAndGet();
+ }
+ return method.invoke(pinned, args);
+ });
+ Table countedTable =
+ (Table)
+ Proxy.newProxyInstance(
+ Table.class.getClassLoader(),
+ new Class<?>[] {Table.class},
+ (proxy, method, args) -> {
+ if (method.getName().equals("currentSnapshot")) {
+ snapshotReads.incrementAndGet();
+ return snapshot;
+ }
+ return method.invoke(table, args);
+ });
+ IcebergUpdateManifestStatsJob.collectManifestStatistics(countedTable);
+ assertEquals(1, snapshotReads.get());
+ assertEquals(1, manifestReads.get());
+ RecordingStatisticsUpdater updater = new RecordingStatisticsUpdater();
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark, updater, CATALOG_NAME, "db.manifest_evolution", "test",
tempDir.toPath());
+ assertEquals(1, updater.updateCalls);
+ assertEquals(3, updater.manifestStatistics.size());
+ Map<String, StatisticValue<?>> values = new HashMap<>();
+ updater.manifestStatistics.forEach(stat -> values.put(stat.name(),
stat.value()));
+ for (Row expected :
+ spark
+ .sql(
+ "SELECT partition_spec_id, COUNT(*), AVG(length) FROM "
+ + name
+ + ".manifests GROUP BY partition_spec_id")
+ .collectAsList()) {
+ IcebergManifestStatistics actual =
+ IcebergManifestStatistics.fromStatistics(values,
expected.getInt(0)).get();
+ assertEquals(expected.getLong(1), actual.count());
+ assertEquals(expected.getDouble(2), actual.averageSize());
+ }
+ IcebergManifestStatistics empty =
IcebergManifestStatistics.fromStatistics(values, 2).get();
+ assertEquals(0L, empty.count());
+ assertEquals(0D, empty.averageSize());
+ assertFalse(IcebergManifestStatistics.fromStatistics(values,
Integer.MAX_VALUE).isPresent());
+ long snapshotId =
+ spark
+ .sql("SELECT snapshot_id FROM " + name + ".snapshots ORDER BY
committed_at DESC")
+ .first()
+ .getLong(0);
+ assertEquals(snapshotId,
values.get(IcebergManifestStatistics.SNAPSHOT_ID).value());
+ } finally {
+ spark.sql("DROP TABLE " + name);
+ }
+ }
+
+ /** Verifies an empty table publishes a complete zero-valued measurement. */
+ @Test
+ public void testPublishesEmptyUnpartitionedTable() throws Exception {
+ String name = CATALOG_NAME + ".db.empty";
+ spark.sql("CREATE TABLE " + name + " (id INT) USING iceberg");
+ try {
+ RecordingStatisticsUpdater updater = new RecordingStatisticsUpdater();
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark, updater, CATALOG_NAME, "db.empty", "test", tempDir.toPath());
+ assertEquals(NameIdentifier.of(CATALOG_NAME, "db", "empty"),
updater.identifier);
+ assertEquals(1, updater.updateCalls);
+ Map<String, StatisticValue<?>> values = new HashMap<>();
+ updater.manifestStatistics.forEach(stat -> values.put(stat.name(),
stat.value()));
+ IcebergManifestStatistics result =
IcebergManifestStatistics.fromStatistics(values, 0).get();
+ assertEquals(0L, result.count());
+ assertEquals(0D, result.averageSize());
+ assertEquals(-1L,
values.get(IcebergManifestStatistics.SNAPSHOT_ID).value());
+ } finally {
+ spark.sql("DROP TABLE " + name);
+ }
+ }
+
+ /** Verifies uncoordinated publications replace complete measurements and
can be rerun. */
+ @Test
+ public void testPublicationWithoutCoordination() throws Exception {
+ String name = CATALOG_NAME + ".db.uncoordinated";
+ spark.sql("CREATE TABLE " + name + " (id INT) USING iceberg");
+ RecordingStatisticsUpdater updater = new RecordingStatisticsUpdater();
+ try {
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark, updater, CATALOG_NAME, "db.uncoordinated", "test", null);
+ assertEquals(1, updater.updateCalls);
+ assertEquals(3, updater.manifestStatistics.size());
+ spark.sql("INSERT INTO " + name + " VALUES (1)");
+ updater.fail = true;
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark, updater, CATALOG_NAME, "db.uncoordinated", "test",
null));
+ updater.fail = false;
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark, updater, CATALOG_NAME, "db.uncoordinated", "test", null);
+ assertEquals(3, updater.updateCalls);
+ Map<String, StatisticValue<?>> values = new HashMap<>();
+ updater.manifestStatistics.forEach(stat -> values.put(stat.name(),
stat.value()));
+ assertEquals(3, values.size());
+ assertEquals(
+ Spark3Util.loadIcebergTable(spark,
name).currentSnapshot().snapshotId(),
+ values.get(IcebergManifestStatistics.SNAPSHOT_ID).value());
+ assertEquals(1L, IcebergManifestStatistics.fromStatistics(values,
0).get().count());
+ } finally {
+ spark.sql("DROP TABLE " + name);
+ }
+ }
+
+ /** Verifies a collection failure can be rerun without manually recovering
the lock file. */
+ @Test
+ public void testCollectionFailureCanBeRerun() throws Exception {
+ Path directory =
Files.createDirectory(tempDir.toPath().resolve("collection-failure"));
+ RecordingStatisticsUpdater updater = new RecordingStatisticsUpdater();
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark, updater, CATALOG_NAME, "db.retry_collection", "test",
directory));
+ assertEquals(0, updater.updateCalls);
+ spark.sql("CREATE TABLE " + CATALOG_NAME + ".db.retry_collection (id INT)
USING iceberg");
+ try {
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark, updater, CATALOG_NAME, "db.retry_collection", "test",
directory);
+ assertEquals(1, updater.updateCalls);
+ } finally {
+ spark.sql("DROP TABLE " + CATALOG_NAME + ".db.retry_collection");
+ }
+ }
+
+ /** Verifies separate job processes share the lock and failed writes block
later cycles. */
+ @Test
+ public void testSharedCoordinationAndFailedWriteRecovery() throws Exception {
+ String name = CATALOG_NAME + ".db.coordinated";
+ spark.sql("CREATE TABLE " + name + " (id INT) USING iceberg");
+ Path directory =
Files.createDirectory(tempDir.toPath().resolve("coordination"));
+ RecordingStatisticsUpdater updater = new RecordingStatisticsUpdater();
+ try {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark, updater, CATALOG_NAME, "db.coordinated", "test",
Paths.get("relative")));
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark,
+ updater,
+ CATALOG_NAME,
+ "db.coordinated",
+ "test",
+ directory.resolve("missing")));
+ assertEquals(0, updater.updateCalls);
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark, updater, CATALOG_NAME, "db.coordinated", "test", directory);
+ Path lockPath;
+ try (Stream<Path> files = Files.list(directory)) {
+ lockPath = files.findFirst().get();
+ }
+ Path ready = directory.resolve("ready");
+ Process otherHost =
+ new ProcessBuilder(
+ new File(System.getProperty("java.home"),
"bin/java").toString(), "-cp",
+ System.getProperty("java.class.path"),
LockHolder.class.getName(),
+ lockPath.toString(), ready.toString())
+ .redirectErrorStream(true)
+ .redirectOutput(directory.resolve("lock-holder.log").toFile())
+ .start();
+ try {
+ Awaitility.await().atMost(15, TimeUnit.SECONDS).until(() ->
Files.exists(ready));
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark, updater, CATALOG_NAME, "db.coordinated", "test",
directory));
+ assertEquals(1, updater.updateCalls);
+ // Independent table targets do not share a coordination lock.
+ spark.sql("CREATE TABLE " + CATALOG_NAME + ".db.independent (id INT)
USING iceberg");
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark, updater, CATALOG_NAME, "db.independent", "test", directory);
+ assertEquals(2, updater.updateCalls);
+ } finally {
+ otherHost.destroyForcibly();
+ assertTrue(otherHost.waitFor(10, TimeUnit.SECONDS));
+ }
+ updater.fail = true;
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark, updater, CATALOG_NAME, "db.coordinated", "test",
directory));
+ assertEquals(3, updater.updateCalls);
+ updater.fail = false;
+ IllegalStateException unfinished =
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark, updater, CATALOG_NAME, "db.coordinated", "test",
directory));
+ assertTrue(unfinished.getMessage().contains("requires recovery"));
+ assertEquals(3, updater.updateCalls);
+ // Recovery is explicit and preserves the shared inode, after the prior
write has ended.
+ try (FileChannel channel = FileChannel.open(lockPath,
StandardOpenOption.WRITE);
+ FileLock lock = channel.lock()) {
+ channel.truncate(0);
+ channel.force(true);
+ }
+ IcebergUpdateManifestStatsJob.updateStatistics(
+ spark, updater, CATALOG_NAME, "db.coordinated", "test", directory);
+ assertEquals(4, updater.updateCalls);
+ assertEquals(0, Files.size(lockPath));
+ } finally {
+ spark.sql("DROP TABLE " + name);
+ spark.sql("DROP TABLE IF EXISTS " + CATALOG_NAME + ".db.independent");
+ }
+ }
+
+ /** Holds the shared filesystem lock from a separate JVM to model another
job host. */
+ public static class LockHolder {
+ /**
+ * Holds the lock until this test subprocess is terminated.
+ *
+ * @param args lock path and readiness path
+ * @throws Exception if coordination fails
+ */
+ public static void main(String[] args) throws Exception {
+ try (FileChannel channel = FileChannel.open(Paths.get(args[0]),
StandardOpenOption.WRITE);
+ FileLock lock = channel.lock()) {
+ Files.write(Paths.get(args[1]), new byte[] {1});
+ new CountDownLatch(1).await();
+ }
+ }
+ }
+
+ private static final class RecordingStatisticsUpdater implements
StatisticsUpdater {
+ private NameIdentifier identifier;
+ private int updateCalls;
+ private boolean fail;
+ private List<StatisticEntry<?>> manifestStatistics =
Collections.emptyList();
+
+ @Override
+ public String name() {
+ return "recording-updater";
+ }
+
+ @Override
+ public void initialize(OptimizerEnv env) {}
+
+ @Override
+ public void updateTableStatistics(
+ NameIdentifier identifier, List<StatisticEntry<?>> statistics) {
+ this.identifier = identifier;
+ this.manifestStatistics = statistics;
+ updateCalls++;
+ if (fail) {
+ throw new IllegalStateException("Ambiguous PUT failure");
+ }
+ }
+
+ @Override
+ public void updatePartitionStatistics(
+ NameIdentifier identifier, Map<PartitionPath, List<StatisticEntry<?>>>
statistics) {
+ throw new AssertionError("Manifest statistics must be table-level
statistics");
+ }
+
+ @Override
+ public void close() {}
+ }
+}
diff --git
a/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/IcebergManifestStatistics.java
b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/IcebergManifestStatistics.java
new file mode 100644
index 0000000000..11f7851bb7
--- /dev/null
+++
b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/IcebergManifestStatistics.java
@@ -0,0 +1,129 @@
+/*
+ * 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.maintenance.optimizer.common;
+
+import java.util.Map;
+import java.util.Optional;
+import javax.annotation.Nullable;
+import org.apache.gravitino.stats.StatisticValue;
+import org.apache.gravitino.stats.StatisticValues;
+
+/** A complete pair of manifest measurements for one resolved Iceberg
partition spec. */
+public final class IcebergManifestStatistics {
+ /** Object-valued manifest counts keyed by decimal spec ID. */
+ public static final String MANIFEST_NUMBER =
"custom-manifest-number-by-spec";
+ /** Object-valued average manifest sizes in bytes keyed by decimal spec ID.
*/
+ public static final String AVG_MANIFEST_SIZE =
"custom-avg-manifest-size-by-spec";
+
+ /** Snapshot ID shared by both complete maps, or -1 for a table without a
snapshot. */
+ public static final String SNAPSHOT_ID = "custom-manifest-snapshot-id";
+
+ private final int specId;
+ private final long count;
+ private final double averageSize;
+
+ /**
+ * Creates a complete measurement for a resolved spec.
+ *
+ * @param specId non-negative resolved spec ID
+ * @param count non-negative manifest count
+ * @param averageSize finite non-negative average size in bytes, zero for no
manifests
+ */
+ public IcebergManifestStatistics(int specId, long count, double averageSize)
{
+ if (specId < 0
+ || count < 0
+ || !Double.isFinite(averageSize)
+ || averageSize < 0
+ || (count == 0 && averageSize != 0)) {
+ throw new IllegalArgumentException("Invalid manifest measurements");
+ }
+ this.specId = specId;
+ this.count = count;
+ this.averageSize = averageSize;
+ }
+
+ /**
+ * Returns the partition spec used for collection.
+ *
+ * @return the resolved spec ID to retain through evaluation and job
submission
+ */
+ public int specId() {
+ return specId;
+ }
+
+ /**
+ * Returns the collected manifest count.
+ *
+ * @return manifest count for this spec
+ */
+ public long count() {
+ return count;
+ }
+
+ /**
+ * Returns the collected average manifest size.
+ *
+ * @return average manifest size in bytes for this spec
+ */
+ public double averageSize() {
+ return averageSize;
+ }
+
+ /**
+ * Reads a complete pair from one table-statistics response. Callers must
not combine responses
+ * from separate reads. An absent measurement means collection is required,
not zero manifests.
+ *
+ * @param statistics values from one atomic table-statistics read
+ * @param specId the previously resolved spec ID
+ * @return a complete measurement, or empty when either spec entry is absent
+ */
+ public static Optional<IcebergManifestStatistics> fromStatistics(
+ Map<String, StatisticValue<?>> statistics, int specId) {
+ if (specId < 0) {
+ throw new IllegalArgumentException("Spec ID must be non-negative");
+ }
+ StatisticValue<?> count = entry(statistics.get(MANIFEST_NUMBER), specId);
+ StatisticValue<?> average = entry(statistics.get(AVG_MANIFEST_SIZE),
specId);
+ if (count == null || average == null) {
+ return Optional.empty();
+ }
+ if (!(count instanceof StatisticValues.LongValue)
+ || !(average instanceof StatisticValues.DoubleValue)) {
+ throw new IllegalArgumentException(
+ "Manifest measurements must be a long count and double size");
+ }
+ return Optional.of(
+ new IcebergManifestStatistics(
+ specId,
+ ((StatisticValues.LongValue) count).value(),
+ ((StatisticValues.DoubleValue) average).value()));
+ }
+
+ @Nullable
+ private static StatisticValue<?> entry(@Nullable StatisticValue<?> value,
int specId) {
+ if (value == null) {
+ return null;
+ }
+ if (!(value instanceof StatisticValues.ObjectValue)) {
+ throw new IllegalArgumentException("Manifest statistics must be object
values");
+ }
+ return ((StatisticValues.ObjectValue)
value).value().get(Integer.toString(specId));
+ }
+}
diff --git
a/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/GravitinoClientUtils.java
b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/GravitinoClientUtils.java
index 9154f8fc80..c9e014dd47 100644
---
a/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/GravitinoClientUtils.java
+++
b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/GravitinoClientUtils.java
@@ -20,10 +20,12 @@
package org.apache.gravitino.maintenance.optimizer.common.util;
import com.google.common.base.Preconditions;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import javax.annotation.Nullable;
import org.apache.gravitino.client.GravitinoClient;
+import org.apache.gravitino.client.GravitinoClientConfiguration;
import org.apache.gravitino.maintenance.optimizer.common.OptimizerEnv;
import org.apache.gravitino.maintenance.optimizer.common.conf.OptimizerConfig;
@@ -64,6 +66,12 @@ public final class GravitinoClientUtils {
String metalake = config.get(OptimizerConfig.GRAVITINO_METALAKE_CONFIG);
GravitinoClient.ClientBuilder builder =
GravitinoClient.builder(uri).withMetalake(metalake);
+ String retryTimes =
config.getRawString(GravitinoClientConfiguration.CLIENT_RETRY_TIMES);
+ if (retryTimes != null) {
+ builder.withClientConfig(
+
Collections.singletonMap(GravitinoClientConfiguration.CLIENT_RETRY_TIMES,
retryTimes));
+ }
+
OptimizerConfig authConfig = config;
if (authFromUpdaterOptions != null && !authFromUpdaterOptions.isEmpty()) {
Map<String, String> authProperties = new
HashMap<>(authFromUpdaterOptions);
diff --git
a/maintenance/optimizer-api/src/test/java/org/apache/gravitino/maintenance/optimizer/common/TestIcebergManifestStatistics.java
b/maintenance/optimizer-api/src/test/java/org/apache/gravitino/maintenance/optimizer/common/TestIcebergManifestStatistics.java
new file mode 100644
index 0000000000..3ce7379c09
--- /dev/null
+++
b/maintenance/optimizer-api/src/test/java/org/apache/gravitino/maintenance/optimizer/common/TestIcebergManifestStatistics.java
@@ -0,0 +1,86 @@
+/*
+ * 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.maintenance.optimizer.common;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.gravitino.stats.StatisticValue;
+import org.apache.gravitino.stats.StatisticValues;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Tests validation and per-spec lookup of manifest statistics. */
+public class TestIcebergManifestStatistics {
+ @Test
+ void testMissingAndEmptyMeasurements() {
+ Map<String, StatisticValue<?>> values = new HashMap<>();
+ Assertions.assertFalse(IcebergManifestStatistics.fromStatistics(values,
2).isPresent());
+ values.putAll(measurement(2, 0L, 0D));
+ IcebergManifestStatistics empty =
IcebergManifestStatistics.fromStatistics(values, 2).get();
+ Assertions.assertEquals(2, empty.specId());
+ Assertions.assertEquals(0L, empty.count());
+ Assertions.assertEquals(0D, empty.averageSize());
+ Assertions.assertFalse(IcebergManifestStatistics.fromStatistics(values,
1).isPresent());
+ values.remove(IcebergManifestStatistics.AVG_MANIFEST_SIZE);
+ Assertions.assertFalse(IcebergManifestStatistics.fromStatistics(values,
2).isPresent());
+ values.put(
+ IcebergManifestStatistics.AVG_MANIFEST_SIZE,
+ StatisticValues.objectValue(
+ Collections.singletonMap("1", StatisticValues.doubleValue(10D))));
+ Assertions.assertFalse(IcebergManifestStatistics.fromStatistics(values,
2).isPresent());
+ }
+
+ @Test
+ void testInvalidMeasurements() {
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> new
IcebergManifestStatistics(-1, 1, 1));
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> new IcebergManifestStatistics(1,
-1, 1));
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> new IcebergManifestStatistics(1,
0, 1));
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> new IcebergManifestStatistics(1,
1, Double.NaN));
+ Map<String, StatisticValue<?>> values = new HashMap<>();
+ values.putAll(measurement(1, 3L, 20D));
+ Assertions.assertEquals(3L,
IcebergManifestStatistics.fromStatistics(values, 1).get().count());
+ values.put(
+ IcebergManifestStatistics.MANIFEST_NUMBER,
+ StatisticValues.objectValue(
+ Collections.singletonMap("1", StatisticValues.doubleValue(3D))));
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
IcebergManifestStatistics.fromStatistics(values, 1));
+ values.put(IcebergManifestStatistics.MANIFEST_NUMBER,
StatisticValues.longValue(3));
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
IcebergManifestStatistics.fromStatistics(values, 1));
+ }
+
+ private static Map<String, StatisticValue<?>> measurement(int spec, long
count, double size) {
+ Map<String, StatisticValue<?>> values = new HashMap<>();
+ values.put(
+ IcebergManifestStatistics.MANIFEST_NUMBER,
+ StatisticValues.objectValue(
+ Collections.singletonMap(Integer.toString(spec),
StatisticValues.longValue(count))));
+ values.put(
+ IcebergManifestStatistics.AVG_MANIFEST_SIZE,
+ StatisticValues.objectValue(
+ Collections.singletonMap(Integer.toString(spec),
StatisticValues.doubleValue(size))));
+ return values;
+ }
+}
diff --git
a/maintenance/optimizer-api/src/test/java/org/apache/gravitino/maintenance/optimizer/common/util/TestGravitinoAuthSettings.java
b/maintenance/optimizer-api/src/test/java/org/apache/gravitino/maintenance/optimizer/common/util/TestGravitinoAuthSettings.java
index b6cc121923..bf2ef0ddd2 100644
---
a/maintenance/optimizer-api/src/test/java/org/apache/gravitino/maintenance/optimizer/common/util/TestGravitinoAuthSettings.java
+++
b/maintenance/optimizer-api/src/test/java/org/apache/gravitino/maintenance/optimizer/common/util/TestGravitinoAuthSettings.java
@@ -22,6 +22,8 @@ package
org.apache.gravitino.maintenance.optimizer.common.util;
import java.util.HashMap;
import java.util.Map;
import org.apache.gravitino.client.GravitinoClient;
+import org.apache.gravitino.client.GravitinoClientConfiguration;
+import org.apache.gravitino.maintenance.optimizer.common.OptimizerEnv;
import org.apache.gravitino.maintenance.optimizer.common.conf.OptimizerConfig;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -29,6 +31,21 @@ import org.mockito.Mockito;
class TestGravitinoAuthSettings {
+ @Test
+ void testClientReceivesRetryConfiguration() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put(OptimizerConfig.GRAVITINO_URI, "http://localhost:1";);
+ properties.put(OptimizerConfig.GRAVITINO_METALAKE, "test");
+ properties.put(GravitinoClientConfiguration.CLIENT_RETRY_TIMES, "invalid");
+ IllegalArgumentException error =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ GravitinoClientUtils.createClient(
+ new OptimizerEnv(new OptimizerConfig(properties))));
+ Assertions.assertTrue(error.getMessage().contains("retryTimes"));
+ }
+
@Test
void fromReturnsNoneWhenUnset() {
GravitinoAuthSettings settings = GravitinoAuthSettings.from(new
OptimizerConfig());
