From 84ad91432bb2256559ef1a1eaf7b633023fc1108 Mon Sep 17 00:00:00 2001 From: Akshay Thorat Date: Sun, 6 Sep 2026 20:00:28 -0700 Subject: [PATCH 1/5] docs(design-docs): add Iceberg rewrite manifests job design MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Documents the design of builtin-iceberg-rewrite-manifests: why manifest bloat is a distinct problem from data file bloat, why the Spark SQL procedure was chosen over the Iceberg Java action, the parameter surface and its validation, the unresolved-placeholder hazard that shapes the job's input handling, and the metrics-then-policy sequencing needed before the job can be triggered automatically. Requested during review of #12937. Co-authored-by: İbrahim ERBİLEN --- design-docs/iceberg-rewrite-manifests-job.md | 387 +++++++++++++++++++ 1 file changed, 387 insertions(+) create mode 100644 design-docs/iceberg-rewrite-manifests-job.md diff --git a/design-docs/iceberg-rewrite-manifests-job.md b/design-docs/iceberg-rewrite-manifests-job.md new file mode 100644 index 00000000000..55ce8403e5b --- /dev/null +++ b/design-docs/iceberg-rewrite-manifests-job.md @@ -0,0 +1,387 @@ + + +# Design of the Iceberg Rewrite Manifests Job in Gravitino + +Tracking issue: [#11196](https://github.com/apache/gravitino/issues/11196). Umbrella: [#8864](https://github.com/apache/gravitino/issues/8864). Implementation: [#12937](https://github.com/apache/gravitino/pull/12937), continuing [#11216](https://github.com/apache/gravitino/pull/11216). + +--- + +## Background + +### How Iceberg scan planning reads metadata + +An Iceberg table stores its file inventory in a three-level tree. Planning a scan walks the tree from the top: + +``` +table metadata (vN.metadata.json) + | + +-- snapshot --> manifest list (snap-.avro) + | + +-- manifest A (.avro) --> data file, data file, ... + +-- manifest B (.avro) --> data file, data file, ... + +-- manifest C (.avro) --> data file, data file, ... +``` + +The manifest list holds per-manifest partition summaries, so a filter can skip whole manifests. Every manifest that survives that skip must then be opened and read to find matching data files. Planning cost therefore tracks the **number of manifests the filter cannot exclude**, not the size of the table. + +### Why manifest count grows + +Each commit writes at least one new manifest for the files it adds. Manifest count grows with **commit frequency**, independent of how much data the table holds: + +| Workload | Commits/day | Manifests after 30 days | +|----------|-------------|-------------------------| +| Hourly batch load | 24 | ~720 | +| Streaming, 1-minute checkpoints | 1,440 | ~43,200 | + +A streaming table of modest size can accumulate tens of thousands of manifests, each covering a handful of data files. Filters exclude few of them because a small manifest written per commit spans whatever partitions that commit touched, so partition summaries end up wide and non-selective. + +Compaction makes this worse rather than better. `rewrite_data_files` commits its results, and that commit writes new manifests of its own. + +### What Gravitino ships today + +The table maintenance service separates **deciding** that work is needed from **executing** it. Built-in job templates are the execution half; policies, strategies, and the recommender are the decision half. + +| Built-in job template | Purpose | Triggered by a built-in policy? | +|-----------------------|---------|---------------------------------| +| `builtin-iceberg-update-stats` | Writes the statistics and metrics policies read | No, submitted directly or by the CLI | +| `builtin-iceberg-rewrite-data-files` | Merges small **data** files | Yes, `system_iceberg_compaction` | +| `builtin-iceberg-expire-snapshots` | Drops old snapshots and their metadata | No, submitted directly | + +### The problem + +Nothing in that list consolidates the manifests of the **current** snapshot. + +- `rewrite_data_files` rewrites data files. The manifests it writes are a side effect of its commit, not a consolidation of what was there before. +- `expire_snapshots` deletes manifests reachable only from expired snapshots. Manifests belonging to the live snapshot are exactly the ones scan planning reads, and expiration never touches them. It is not a substitute. + +Iceberg has an action for this, exposed as the `rewrite_manifests` Spark procedure, and Gravitino already delegates it on the **engine** side: [#10500](https://github.com/apache/gravitino/pull/10500) lets a Trino user issue `CALL ... rewrite_manifests`. That closes the manual path and leaves two gaps: + +1. It requires a human at a Trino session. There is no server-side job the maintenance service can submit, and therefore no path to policy-driven manifest optimization. +2. It only helps Trino users. Gravitino deployments driving Spark or Flink have no equivalent. + +--- + +## Goals + +1. **Built-in template**: A template named `builtin-iceberg-rewrite-manifests` appears in `GET /api/metalakes/{metalake}/jobs/templates` once `gravitino-jobs` is on the server classpath, and is submittable through the existing jobs REST API with no new endpoint. +2. **Full procedure surface**: Every parameter Iceberg's `rewrite_manifests` procedure accepts - `table`, `use_caching`, `spec_id` - is reachable through `jobConf`. +3. **Absent means default**: Omitting an optional parameter leaves Iceberg's own default in force. The generated SQL names only the parameters the caller actually supplied. +4. **Invalid input fails before Spark starts**: A malformed `use_caching` or `spec_id` exits non-zero with a message naming the offending value, rather than being coerced into a valid-looking SQL literal. +5. **Runtime version tolerance**: The job runs against any Iceberg version exposing the procedure, without assuming the integer width of its output columns. Iceberg is a `compileOnly` dependency supplied by the cluster, so the compile-time and runtime versions differ in practice. +6. **Injection safety**: A catalog name or table identifier containing quotes, backticks, or statement separators cannot change the shape of the generated statement. +7. **Consistency with siblings**: The job's class layout, argument convention, template registration, and staging log output match `IcebergExpireSnapshotsJob` closely enough that an operator familiar with one can read the other. + +--- + +## Non-Goals + +1. **A built-in policy that triggers this job automatically**: `system_iceberg_compaction` is the only built-in policy type, and it evaluates `custom-data-file-mse` and `custom-delete-file-number` - both data-file metrics. Automatic triggering needs manifest-level metrics to exist first, so it is sequenced as follow-up work below rather than bundled here. +2. **Replacing `expire_snapshots` or `rewrite_data_files`**: The three address different layers - snapshot history, data files, and the manifests indexing them. Running this job does not reduce the need for the other two. +3. **Partition-scoped or predicate-scoped rewrites**: Iceberg's procedure takes no `where` clause; it rewrites all manifests of the current snapshot. Offering a filter would mean bypassing the procedure for the Java action API, which Solution Investigations rejects below. +4. **Engine-side delegation**: Issuing `rewrite_manifests` from a query engine session is already covered by [#10500](https://github.com/apache/gravitino/pull/10500). This job is the server-side counterpart, not a replacement. +5. **Scheduling**: The maintenance service is driven by explicit submission or by the CLI workflow, with no internal scheduler. This job inherits that and does not introduce one. + +--- + +## Solution Investigations + +| Approach | Pros | Cons | Decision | +|----------|------|------|----------| +| A. Engine-side delegation only (status quo, #10500) | Already merged; zero new server code | Needs a human in a Trino session; not policy-drivable; useless to non-Trino deployments | Rejected - leaves the server-side gap that #11196 is about | +| B. Iceberg Java `RewriteManifests` action, called directly | Type-safe; exposes `rewriteIf`, `stagingLocation`, `specId`; no SQL text to escape | Requires catalog-loading plumbing no other job has; binds us to `iceberg-core` internals across the `compileOnly` boundary; diverges from all three sibling jobs | Rejected - see below | +| C. Spark SQL `CALL .system.rewrite_manifests` | Identical shape to all three sibling jobs; Spark resolves the catalog from template configs; procedure API is stable across Iceberg versions | Parameters passed as SQL text, so escaping and validation are ours to get right | **Chosen** | +| D. A flag on `builtin-iceberg-rewrite-data-files` | One template instead of two; one Spark submission for both rewrites | Ties two operations with different cost profiles and cadences; a template's `version` field could no longer describe its behavior; callers wanting only manifests would pay for a data rewrite | Rejected - conflates independent maintenance operations | + +### Why B was rejected in detail + +The Java action is the better API in isolation. Three concrete constraints outweigh that: + +- **The `compileOnly` boundary.** `maintenance/jobs` compiles against Iceberg but ships without it; the cluster's `iceberg-spark-runtime` provides it at runtime. The `Procedure` contract behind `CALL` is a stable extension point across Iceberg versions. `RewriteManifests`, `Table`, and the catalog-loading classes are internals we would be pinning ourselves to across a boundary where the version can change without us. +- **Catalog loading.** Under `CALL`, Spark resolves the catalog from `spark.sql.catalog.{{catalog_name}}.*`, which `IcebergSparkConfigUtils.buildTemplateSparkConfigs()` already emits for every Iceberg job template. The Java action needs a loaded `Table`, which means reimplementing that resolution per catalog backend - REST, Hive, JDBC, Glue. +- **Consistency.** All three sibling jobs use `CALL`. A fourth that loads catalogs itself would need every operator and reviewer to learn a second model for no user-visible gain. + +If a future requirement needs `rewriteIf` or `stagingLocation`, which the procedure does not expose, the right move is to migrate all Iceberg jobs to the action API together, not to fork this one. + +--- + +## Proposal + +### Job template + +| Property | Value | +|----------|-------| +| Name | `builtin-iceberg-rewrite-manifests` | +| Type | Spark (`SparkJobTemplate`) | +| Version | `v1` (in `customFields`, under `JobTemplateProvider.PROPERTY_VERSION_KEY`) | +| Main class | `org.apache.gravitino.maintenance.jobs.iceberg.IcebergRewriteManifestsJob` | +| Registered in | `BuiltInJobTemplateProvider.BUILT_IN_JOBS` | +| Spark configs | `IcebergSparkConfigUtils.buildTemplateSparkConfigs()`, shared with every Iceberg job template | + +Registration is a one-line addition to an `ImmutableList`. `BuiltInJobTemplateProvider` then validates the name against `BUILTIN_NAME_PATTERN` and the version against `VERSION_VALUE_PATTERN`, and logs and skips any template failing either - so a malformed template degrades to absence rather than to a server that will not start. + +### Parameters + +No REST API changes. The job is reached through the existing `POST /api/metalakes/{metalake}/jobs/runs` endpoint, whose `jobConf` map is opaque to the server and interpreted per template. + +| `jobConf` key | CLI argument | Type | Required | Default | Description | +|---------------|--------------|------|----------|---------|-------------| +| `catalog_name` | `--catalog` | string | Yes | - | Iceberg catalog name as registered in Spark | +| `table_identifier` | `--table` | string | Yes | - | Fully qualified table name, such as `db.sample` | +| `use_caching` | `--use-caching` | boolean | No | Iceberg's default (`true`) | Cache table metadata in Spark for the duration of the rewrite | +| `spec_id` | `--spec-id` | int | No | The table's current spec | Rewrite manifests to this partition spec ID | +| `spark_conf` | `--spark-conf` | JSON object | No | None | Extra Spark configuration applied to the session | + +These are exactly the three in-parameters of Iceberg's `RewriteManifestsProcedure` (`table` required, `use_caching` and `spec_id` optional), plus Gravitino's standard `spark_conf`. `spec_id` is what makes the "align manifests with a better partition spec" half of the motivation reachable; without it only consolidation is available. + +The standard Spark template keys - `spark_master`, `spark_executor_instances`, `spark_executor_cores`, `spark_executor_memory`, `spark_driver_memory`, `catalog_type`, `catalog_uri`, `warehouse_location` - behave as they do for every other Iceberg job template. + +### Generated SQL + +The job emits a `CALL` naming only the parameters the caller supplied: + +```sql +-- catalog_name and table_identifier only +CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample') + +-- with both optional parameters +CALL `iceberg_prod`.system.rewrite_manifests( + table => 'db.sample', use_caching => false, spec_id => 2) +``` + +Two escaping rules, both from `IcebergJobUtils` and shared with the sibling jobs: + +| Value | Function | Rule | Example | +|-------|----------|------|---------| +| Catalog name | `escapeSqlIdentifier` | Double internal backticks, wrap in backticks | ``catalog`; DROP TABLE t; --`` becomes ``` `catalog``; DROP TABLE t; --` ``` | +| Table identifier | `escapeSqlString` | Double single quotes | `db.t' OR '1'='1` becomes `db.t'' OR ''1''=''1` | + +`use_caching` and `spec_id` never reach the statement as caller text. They are validated, then re-rendered from the parsed `boolean` and `int`, so the only characters that can appear are those `Boolean.toString` and `Integer.toString` produce. + +### Placeholder resolution, and why absent parameters need care + +This is the one piece of behavior that is not obvious from the sibling jobs, and it dictated most of the job's input handling. + +A template's arguments are literal placeholder text. `JobManager.replacePlaceholder` substitutes `jobConf` values, and when a key is **absent it keeps the placeholder verbatim**: + +```java +// core/src/main/java/org/apache/gravitino/job/JobManager.java +} else { + // If no replacement is found, keep the placeholder as is + matcher.appendReplacement(result, matcher.group(0)); +} +``` + +So a caller who omits `use_caching` does not produce a missing argument. They produce the literal string `{{use_caching}}` in `argv`, which parses as a value like any other: + +| Caller supplies | `argv` reaching the job | Naive handling | Result | +|-----------------|--------------------------|----------------|--------| +| `use_caching: "false"` | `--use-caching false` | `Boolean.parseBoolean` gives `false` | Correct | +| *nothing* | `--use-caching {{use_caching}}` | `Boolean.parseBoolean` gives `false` | **Wrong** - silently disables caching instead of leaving Iceberg's default | +| *nothing* | `--spec-id {{spec_id}}` | `Integer.parseInt` throws | Job fails, though the caller asked for nothing | + +The `false` row is the dangerous one: no error, no log line, and a parameter the caller never set now materially changes how the rewrite runs. The two-argument case is worse for `spec_id`, where the same mechanism turns an omitted parameter into a hard failure. + +The job therefore passes every optional value through `IcebergJobUtils.nullIfUnresolvedPlaceholder`, which returns `null` for a value that is entirely an unresolved placeholder: + +```java +String useCaching = IcebergJobUtils.nullIfUnresolvedPlaceholder(argMap.get("use-caching")); +String specId = IcebergJobUtils.nullIfUnresolvedPlaceholder(argMap.get("spec-id")); +``` + +A `null` is then omitted from the generated SQL, which is what "absent means Iceberg's default" requires. Only a value that is *entirely* a placeholder is dropped - `{{a}}b` is a real value and survives - so a legitimate value that happens to contain braces is unaffected. + +The same hazard exists in `IcebergExpireSnapshotsJob` (`older_than`, `retain_last`) and in `IcebergRewriteDataFilesJob`. Fixing those is out of scope here and listed as follow-up work. + +### Validation + +| Parameter | Accepted | Rejected | Why validate at all | +|-----------|----------|----------|---------------------| +| `use_caching` | `true`, `false`, any case | Anything else, for example `yes` | `Boolean.parseBoolean` maps every non-`true` string to `false`, so a typo would quietly invert the caller's intent | +| `spec_id` | Non-negative integer | Negative, non-numeric | Keeps a malformed value from reaching Spark as an unparseable SQL literal, and reports it against the argument name rather than as a SQL syntax error | + +Both run before the `SparkSession` is created, so a bad argument costs no cluster resources. Both exit non-zero with the offending value in the message, then print usage. + +### Reading the procedure output + +`rewrite_manifests` returns one row, `(rewritten_manifests_count, added_manifests_count)`. Both are `IntegerType` in Iceberg 1.11.0, the version this repo compiles against, but the column width is an Iceberg implementation detail and Iceberg is `compileOnly` - the cluster supplies it, and a `getInt` against a future `LongType` would throw `ClassCastException` at the end of an otherwise successful rewrite. + +The job reads the columns as `Number`: + +```java +List results = spark.sql(sql).collectAsList(); +if (!results.isEmpty()) { + Row result = results.get(0); + System.out.printf( + "Rewrite Manifests Results:%n Rewritten manifests: %d%n Added manifests: %d%n", + ((Number) result.get(0)).longValue(), ((Number) result.get(1)).longValue()); +} +``` + +The empty-result guard matters because a procedure returning no rows is a legitimate no-op. + +### User process + +1. Confirm the template is registered. `gravitino-jobs` must be in the server's `auxlib`: + + ```bash + curl -sS "http://localhost:8090/api/metalakes/test/jobs/templates?details=true" \ + | jq '.jobTemplates[].name' + ``` + + The list must contain `builtin-iceberg-rewrite-manifests`. + +2. Submit a run. Only `catalog_name` and `table_identifier` are needed beyond the standard Spark keys: + + ```bash + job_id=$(curl -sS -X POST -H "Accept: application/vnd.gravitino.v1+json" \ + -H "Content-Type: application/json" \ + -d '{ + "jobTemplateName": "builtin-iceberg-rewrite-manifests", + "jobConf": { + "catalog_name": "rest_catalog", + "table_identifier": "db.t1", + "spark_master": "local[2]", + "catalog_type": "rest", + "catalog_uri": "http://localhost:9001/iceberg", + "warehouse_location": "" + } + }' \ + http://localhost:8090/api/metalakes/test/jobs/runs | jq -r '.job.jobId') + ``` + +3. Poll for completion. Status is polled rather than pushed, so it lags the Spark process by up to `gravitino.job.statusPullIntervalInMs`: + + ```bash + curl -sS "http://localhost:8090/api/metalakes/test/jobs/runs/${job_id}" | jq '.job.status' + ``` + +4. Read the counts from the staging log, rooted at `gravitino.job.stagingDir`: + + ```bash + grep -E "Rewritten manifests|Added manifests" \ + "/tmp/gravitino/jobs/staging/test/builtin-iceberg-rewrite-manifests/${job_id}/output.log" + ``` + +5. Confirm against the table. `added_manifests_count` well below the previous manifest count is the outcome being sought: + + ```sql + SELECT count(*) FROM rest_catalog.db.t1.manifests; + ``` + +Pair the job with `builtin-iceberg-expire-snapshots`. The rewrite creates a snapshot and leaves the pre-rewrite manifests reachable from the previous one, so storage is only reclaimed once those snapshots expire. + +### Implementation process + +``` +Client + | POST /api/metalakes/{metalake}/jobs/runs { jobTemplateName, jobConf } + v +JobOperations (server/) + | + v +JobManager (core/) + | 1. look up the registered template by name + | 2. replacePlaceholder over arguments and configs, using jobConf + | (absent key -> placeholder kept verbatim) + | 3. persist the resolved template on the JobEntity + v +JobExecutor --spark-submit--> IcebergRewriteManifestsJob.main(argv) + | 4. IcebergJobUtils.parseArguments + | 5. nullIfUnresolvedPlaceholder on optionals + | 6. validateUseCaching / validateSpecId -- fail here, before Spark + | 7. build SparkSession (+ spark_conf) + | 8. buildProcedureCall -> escaped CALL text + v + Spark --resolves spark.sql.catalog..*--> Iceberg + | | + | 9. one row: rewritten/added counts <--------+ + v + stdout -> staging output.log +``` + +Steps 4 through 6 all run before step 7, so every input error is reported without acquiring a Spark session. + +### Backward compatibility + +Additive in every respect: + +- **New template name.** No existing name, argument, or `jobConf` key changes meaning. Callers of the other three templates are unaffected. +- **No API change.** No new or modified REST endpoint, DTO, client method, or entity, so no OpenAPI or client update is needed. +- **No schema change.** Job runs persist through the existing `JobEntity`. +- **Servers without `gravitino-jobs`** simply do not list the template, exactly as with the other built-ins today. +- **Versioned template.** `version` is `v1`. A later change to arguments or semantics bumps it, following the convention the sibling templates set. + +### Follow-up: metrics and a policy for automatic triggering + +Automatic triggering is a Non-Goal above; this sketches the shape so the follow-up issues have a starting point rather than leaving the path undefined. + +**Metrics.** `IcebergUpdateStatsAndMetricsJob` aggregates the `.files` metadata table and writes `custom-`-prefixed statistics such as `custom-data-file-mse` and `custom-small-file-number`. The manifest analogue reads the `.manifests` metadata table: + +| Proposed statistic | Source | Purpose | +|--------------------|--------|---------| +| `custom-manifest-number` | `COUNT(*)` | The primary signal; planning cost tracks it | +| `custom-avg-manifest-size` | `AVG(length)` | Distinguishes many-small from few-large | +| `custom-small-manifest-number` | `COUNT(*)` below a threshold | Directly comparable to `custom-small-file-number` | +| `custom-manifest-spec-count` | `COUNT(DISTINCT partition_spec_id)` | More than one live spec is what makes `spec_id` useful | + +**Policy.** A `system_iceberg_rewrite_manifests` policy type with a `iceberg-manifest-rewrite` strategy type, a `StrategyHandler` registered under `gravitino.optimizer.strategyHandler.*`, and thresholds over the metrics above. It reuses the existing recommender path, so no new framework is required - `CompactionStrategyHandler` is the template to follow. + +To stay aligned with table compaction, the handler declares `DataRequirement.TABLE_STATISTICS` in `dataRequirements()` and reads its thresholds from table statistics rather than from a side channel, exactly as `CompactionStrategyHandler` does. Any "when did this last run" signal belongs in `statistic_meta` as a `custom-`-prefixed statistic written back after a successful run - the same path `IcebergUpdateStatsAndMetricsJob` already uses - rather than in new bespoke state. + +Sequencing matters: the metrics must land, and be collected for long enough to calibrate thresholds against real tables, before a policy that fires on them is worth shipping. + +--- + +## Task Breakdown + +### Phase 1: The job (this design, PR [#12937](https://github.com/apache/gravitino/pull/12937)) + +- [x] Implement `IcebergRewriteManifestsJob` with `table` and `use_caching`, calling `rewrite_manifests` through Spark SQL +- [x] Register the template in `BuiltInJobTemplateProvider` +- [x] Move shared SQL escaping, argument parsing, and Spark config parsing to `IcebergJobUtils` rather than reaching into `IcebergRewriteDataFilesJob` +- [x] Add `IcebergJobUtils.nullIfUnresolvedPlaceholder` and drop unresolved placeholders for optional arguments +- [x] Validate `use_caching`, rejecting values that are neither `true` nor `false` +- [x] Read procedure output as `Number` so the job tolerates the runtime Iceberg version's column width +- [x] Support `spec_id`, with non-negative-integer validation +- [x] Unit tests: template shape, argument parsing, validation, placeholder filtering, generated SQL, injection escaping +- [x] Document the template and its parameters in `docs/table-maintenance-service/optimizer-cli-reference.md` +- [ ] Add `TestIcebergRewriteManifestsJobWithSpark`, following the two existing `*WithSpark` tests, asserting manifest count falls across a real rewrite + +### Phase 2: Manifest metrics (separate issue) + +- [ ] Add manifest aggregation over the `.manifests` metadata table to `IcebergUpdateStatsAndMetricsJob` +- [ ] Emit `custom-manifest-number`, `custom-avg-manifest-size`, `custom-small-manifest-number`, `custom-manifest-spec-count` +- [ ] Unit tests for the aggregation SQL and the statistic mapping +- [ ] Document the new statistics alongside the existing ones + +### Phase 3: Policy-driven triggering (separate issue, depends on Phase 2) + +- [ ] Define the `system_iceberg_rewrite_manifests` policy type and its typed content in `api/` +- [ ] Implement a `StrategyHandler` for strategy type `iceberg-manifest-rewrite` +- [ ] Wire the handler into job submission so the recommender can submit `builtin-iceberg-rewrite-manifests` +- [ ] Unit tests for threshold evaluation and job context construction +- [ ] Integration test for the full collect / evaluate / submit path +- [ ] Document the policy, mirroring `docs/iceberg-compaction-policy.md` + +### Phase 4: Placeholder hardening across the existing jobs (separate issue) + +- [ ] Apply `nullIfUnresolvedPlaceholder` to `older_than` and `retain_last` in `IcebergExpireSnapshotsJob` +- [ ] Apply it to the optional arguments of `IcebergRewriteDataFilesJob` and `IcebergUpdateStatsAndMetricsJob` +- [ ] Regression tests per job proving an omitted optional parameter leaves the engine default in force From 02e0463c2bdf469e83c7ac35285b8ebb98eaad14 Mon Sep 17 00:00:00 2001 From: Akshay Thorat Date: Mon, 14 Sep 2026 10:48:25 -0700 Subject: [PATCH 2/5] docs: address rewrite manifests design review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Align the design with the orphan-files proposal, simplify metrics, and document verified spec selection and discovery. Co-authored-by: İbrahim ERBİLEN --- design-docs/iceberg-rewrite-manifests-job.md | 400 +++++++------------ 1 file changed, 155 insertions(+), 245 deletions(-) diff --git a/design-docs/iceberg-rewrite-manifests-job.md b/design-docs/iceberg-rewrite-manifests-job.md index 55ce8403e5b..29a7ee4176e 100644 --- a/design-docs/iceberg-rewrite-manifests-job.md +++ b/design-docs/iceberg-rewrite-manifests-job.md @@ -17,226 +17,162 @@ under the License. --> -# Design of the Iceberg Rewrite Manifests Job in Gravitino +# Design: Built-in Iceberg Rewrite Manifests Maintenance Job -Tracking issue: [#11196](https://github.com/apache/gravitino/issues/11196). Umbrella: [#8864](https://github.com/apache/gravitino/issues/8864). Implementation: [#12937](https://github.com/apache/gravitino/pull/12937), continuing [#11216](https://github.com/apache/gravitino/pull/11216). +Tracking issue: [#11196](https://github.com/apache/gravitino/issues/11196). Umbrella: [#8864](https://github.com/apache/gravitino/issues/8864). Implementation: [#12937](https://github.com/apache/gravitino/pull/12937), continuing the original work by @ibrahimErbilen in [#11216](https://github.com/apache/gravitino/pull/11216). ---- +## 1. Background -## Background +Iceberg scan planning reads a snapshot's manifest list, uses partition summaries to prune manifests, and opens the remaining manifests to find matching files. Frequent small writes can leave many small manifests and increase planning work. Growth depends on write patterns and Iceberg's automatic manifest merging; it is not a fixed number of manifests per commit. -### How Iceberg scan planning reads metadata +Data-file compaction and snapshot expiration serve different purposes. Compaction changes data-file layout and can also affect manifests through its commit. Expiration removes obsolete snapshots and unreferenced files. Neither provides an explicit job dedicated to reorganizing the current snapshot's manifests. -An Iceberg table stores its file inventory in a three-level tree. Planning a scan walks the tree from the top: +Gravitino needs a built-in job that operators can submit through the existing jobs API to invoke Iceberg's `rewrite_manifests` procedure. It consolidates and clusters manifest entries within a selected partition spec without rewriting data files. -``` -table metadata (vN.metadata.json) - | - +-- snapshot --> manifest list (snap-.avro) - | - +-- manifest A (.avro) --> data file, data file, ... - +-- manifest B (.avro) --> data file, data file, ... - +-- manifest C (.avro) --> data file, data file, ... -``` - -The manifest list holds per-manifest partition summaries, so a filter can skip whole manifests. Every manifest that survives that skip must then be opened and read to find matching data files. Planning cost therefore tracks the **number of manifests the filter cannot exclude**, not the size of the table. - -### Why manifest count grows - -Each commit writes at least one new manifest for the files it adds. Manifest count grows with **commit frequency**, independent of how much data the table holds: - -| Workload | Commits/day | Manifests after 30 days | -|----------|-------------|-------------------------| -| Hourly batch load | 24 | ~720 | -| Streaming, 1-minute checkpoints | 1,440 | ~43,200 | - -A streaming table of modest size can accumulate tens of thousands of manifests, each covering a handful of data files. Filters exclude few of them because a small manifest written per commit spans whatever partitions that commit touched, so partition summaries end up wide and non-selective. - -Compaction makes this worse rather than better. `rewrite_data_files` commits its results, and that commit writes new manifests of its own. +## 2. Goals -### What Gravitino ships today +1. **Job submission**: Register `builtin-iceberg-rewrite-manifests` and submit it through the existing jobs REST API. +2. **Procedure parameters**: Expose `table`, `use_caching`, and `spec_id`, with omitted optional values delegated to Iceberg's defaults. +3. **Operator guidance**: Explain how to discover spec IDs and which manifests a run can rewrite. +4. **Validation and results**: Reject malformed arguments, escape SQL inputs, and report rewritten and added manifest counts. -The table maintenance service separates **deciding** that work is needed from **executing** it. Built-in job templates are the execution half; policies, strategies, and the recommender are the decision half. +## 3. Non-Goals -| Built-in job template | Purpose | Triggered by a built-in policy? | -|-----------------------|---------|---------------------------------| -| `builtin-iceberg-update-stats` | Writes the statistics and metrics policies read | No, submitted directly or by the CLI | -| `builtin-iceberg-rewrite-data-files` | Merges small **data** files | Yes, `system_iceberg_compaction` | -| `builtin-iceberg-expire-snapshots` | Drops old snapshots and their metadata | No, submitted directly | +1. **Data repartitioning**: Rewriting manifests does not migrate existing files between partition specs. +2. **Row or partition predicates**: This job exposes no `where` filter. Selection is by existing partition spec. +3. **Automatic triggering in the initial PR**: Manifest statistics and a policy are follow-up work; the existing compaction policy does not trigger this job. +4. **Scheduling or framework changes**: Reuse existing job submission and status handling without changing JobManager or introducing a scheduler. -### The problem +## 4. Existing Architecture Overview -Nothing in that list consolidates the manifests of the **current** snapshot. +### 4.1 Layer Summary -- `rewrite_data_files` rewrites data files. The manifests it writes are a side effect of its commit, not a consolidation of what was there before. -- `expire_snapshots` deletes manifests reachable only from expired snapshots. Manifests belonging to the live snapshot are exactly the ones scan planning reads, and expiration never touches them. It is not a substitute. +| Layer | Existing role | Impact of this proposal | +| ---------------------------------------------------- | ------------------------------------------------ | ------------------------------------------------- | +| Policy (`api/`) | Defines typed maintenance policy content | No initial change; separate manifest policy later | +| Strategy handler (`maintenance/optimizer/`) | Evaluates collected statistics | No initial change | +| Job adapter and submitter (`maintenance/optimizer/`) | Converts a strategy result into a job submission | Follow-up wiring for automatic triggering | +| Spark job (`maintenance/jobs/`) | Executes registered built-in job templates | Add the rewrite-manifests job | +| Jobs REST API and core | Registers templates, submits runs, tracks status | Reuse existing endpoints and persistence | -Iceberg has an action for this, exposed as the `rewrite_manifests` Spark procedure, and Gravitino already delegates it on the **engine** side: [#10500](https://github.com/apache/gravitino/pull/10500) lets a Trino user issue `CALL ... rewrite_manifests`. That closes the manual path and leaves two gaps: +## 5. Proposed Design -1. It requires a human at a Trino session. There is no server-side job the maintenance service can submit, and therefore no path to policy-driven manifest optimization. -2. It only helps Trino users. Gravitino deployments driving Spark or Flink have no equivalent. +### 5.1 Architecture Diagram ---- - -## Goals - -1. **Built-in template**: A template named `builtin-iceberg-rewrite-manifests` appears in `GET /api/metalakes/{metalake}/jobs/templates` once `gravitino-jobs` is on the server classpath, and is submittable through the existing jobs REST API with no new endpoint. -2. **Full procedure surface**: Every parameter Iceberg's `rewrite_manifests` procedure accepts - `table`, `use_caching`, `spec_id` - is reachable through `jobConf`. -3. **Absent means default**: Omitting an optional parameter leaves Iceberg's own default in force. The generated SQL names only the parameters the caller actually supplied. -4. **Invalid input fails before Spark starts**: A malformed `use_caching` or `spec_id` exits non-zero with a message naming the offending value, rather than being coerced into a valid-looking SQL literal. -5. **Runtime version tolerance**: The job runs against any Iceberg version exposing the procedure, without assuming the integer width of its output columns. Iceberg is a `compileOnly` dependency supplied by the cluster, so the compile-time and runtime versions differ in practice. -6. **Injection safety**: A catalog name or table identifier containing quotes, backticks, or statement separators cannot change the shape of the generated statement. -7. **Consistency with siblings**: The job's class layout, argument convention, template registration, and staging log output match `IcebergExpireSnapshotsJob` closely enough that an operator familiar with one can read the other. +```text +Operator -> Jobs REST API -> JobManager -> JobExecutor -> spark-submit + | + v + IcebergRewriteManifestsJob + | + v + Spark SQL rewrite_manifests + | + v + Iceberg metadata commit + | + v + Counts and job status +``` ---- +The initial implementation enters at the jobs API. A future policy-driven flow will reach the same job through the existing recommender and submitter. -## Non-Goals +### 5.2 Layer 1 - Policy Definition (`api/`, follow-up) -1. **A built-in policy that triggers this job automatically**: `system_iceberg_compaction` is the only built-in policy type, and it evaluates `custom-data-file-mse` and `custom-delete-file-number` - both data-file metrics. Automatic triggering needs manifest-level metrics to exist first, so it is sequenced as follow-up work below rather than bundled here. -2. **Replacing `expire_snapshots` or `rewrite_data_files`**: The three address different layers - snapshot history, data files, and the manifests indexing them. Running this job does not reduce the need for the other two. -3. **Partition-scoped or predicate-scoped rewrites**: Iceberg's procedure takes no `where` clause; it rewrites all manifests of the current snapshot. Offering a filter would mean bypassing the procedure for the Java action API, which Solution Investigations rejects below. -4. **Engine-side delegation**: Issuing `rewrite_manifests` from a query engine session is already covered by [#10500](https://github.com/apache/gravitino/pull/10500). This job is the server-side counterpart, not a replacement. -5. **Scheduling**: The maintenance service is driven by explicit submission or by the CLI workflow, with no internal scheduler. This job inherits that and does not introduce one. +A future `system_iceberg_rewrite_manifests` policy can express thresholds over manifest statistics. Its typed fields and defaults need a separate proposal informed by collected measurements. No policy type is added by #12937. ---- +### 5.3 Layer 2 - Strategy Handler (`maintenance/optimizer/`, follow-up) -## Solution Investigations +Start by collecting only these two table-level statistics through `IcebergUpdateStatsAndMetricsJob`: -| Approach | Pros | Cons | Decision | -|----------|------|------|----------| -| A. Engine-side delegation only (status quo, #10500) | Already merged; zero new server code | Needs a human in a Trino session; not policy-drivable; useless to non-Trino deployments | Rejected - leaves the server-side gap that #11196 is about | -| B. Iceberg Java `RewriteManifests` action, called directly | Type-safe; exposes `rewriteIf`, `stagingLocation`, `specId`; no SQL text to escape | Requires catalog-loading plumbing no other job has; binds us to `iceberg-core` internals across the `compileOnly` boundary; diverges from all three sibling jobs | Rejected - see below | -| C. Spark SQL `CALL .system.rewrite_manifests` | Identical shape to all three sibling jobs; Spark resolves the catalog from template configs; procedure API is stable across Iceberg versions | Parameters passed as SQL text, so escaping and validation are ours to get right | **Chosen** | -| D. A flag on `builtin-iceberg-rewrite-data-files` | One template instead of two; one Spark submission for both rewrites | Ties two operations with different cost profiles and cadences; a template's `version` field could no longer describe its behavior; callers wanting only manifests would pay for a data rewrite | Rejected - conflates independent maintenance operations | +| Proposed statistic | Source from `.manifests` | Purpose | +| -------------------------- | ------------------------ | --------------------------------------------------- | +| `custom-manifest-number` | `COUNT(*)` | Measures current snapshot manifest count | +| `custom-avg-manifest-size` | `AVG(length)` | Distinguishes many small manifests from larger ones | -### Why B was rejected in detail +Do not add small-manifest counts or spec-count statistics without a demonstrated policy use case. A future handler declares `DataRequirement.TABLE_STATISTICS` and reads measured values through the existing statistics path. Policy thresholds come from policy content. Any last-success time needed for a cooldown belongs in `statistic_meta` as a `custom-` statistic written after successful completion. -The Java action is the better API in isolation. Three concrete constraints outweigh that: +These aggregates span all specs in the current snapshot. Since a rewrite targets one spec, an automatic policy must account for this mismatch before shipping, for example by evaluating per-spec measurements. A high total count alone does not prove the default-spec rewrite will reduce it. -- **The `compileOnly` boundary.** `maintenance/jobs` compiles against Iceberg but ships without it; the cluster's `iceberg-spark-runtime` provides it at runtime. The `Procedure` contract behind `CALL` is a stable extension point across Iceberg versions. `RewriteManifests`, `Table`, and the catalog-loading classes are internals we would be pinning ourselves to across a boundary where the version can change without us. -- **Catalog loading.** Under `CALL`, Spark resolves the catalog from `spark.sql.catalog.{{catalog_name}}.*`, which `IcebergSparkConfigUtils.buildTemplateSparkConfigs()` already emits for every Iceberg job template. The Java action needs a loaded `Table`, which means reimplementing that resolution per catalog backend - REST, Hive, JDBC, Glue. -- **Consistency.** All three sibling jobs use `CALL`. A fourth that loads catalogs itself would need every operator and reviewer to learn a second model for no user-visible gain. +### 5.4 Layer 3 - Job Adapter (`maintenance/optimizer/`, follow-up) -If a future requirement needs `rewriteIf` or `stagingLocation`, which the procedure does not expose, the right move is to migrate all Iceberg jobs to the action API together, not to fork this one. +A future adapter maps a strategy decision to `builtin-iceberg-rewrite-manifests`, supplies the catalog and table identifiers, and forwards the selected spec and caching option. It reuses existing job submission and tracking. The initial PR requires no adapter because operators submit the template directly. ---- +### 5.5 Layer 4 - Spark Job (`maintenance/jobs/`) -## Proposal +#### 5.5.1 Job Class and Registration -### Job template +| Property | Value | +| ------------------- | -------------------------------------------------------------------------- | +| Template name | `builtin-iceberg-rewrite-manifests` | +| Type | `SparkJobTemplate` | +| Version | `v1` | +| Main class | `org.apache.gravitino.maintenance.jobs.iceberg.IcebergRewriteManifestsJob` | +| Registration | `BuiltInJobTemplateProvider.BUILT_IN_JOBS` | +| Spark configuration | `IcebergSparkConfigUtils.buildTemplateSparkConfigs()` | -| Property | Value | -|----------|-------| -| Name | `builtin-iceberg-rewrite-manifests` | -| Type | Spark (`SparkJobTemplate`) | -| Version | `v1` (in `customFields`, under `JobTemplateProvider.PROPERTY_VERSION_KEY`) | -| Main class | `org.apache.gravitino.maintenance.jobs.iceberg.IcebergRewriteManifestsJob` | -| Registered in | `BuiltInJobTemplateProvider.BUILT_IN_JOBS` | -| Spark configs | `IcebergSparkConfigUtils.buildTemplateSparkConfigs()`, shared with every Iceberg job template | +The job parses and validates its inputs, builds a Spark session, calls the procedure, reports counts, and stops the session. Execution failures produce a non-zero process exit status. -Registration is a one-line addition to an `ImmutableList`. `BuiltInJobTemplateProvider` then validates the name against `BUILTIN_NAME_PATTERN` and the version against `VERSION_VALUE_PATTERN`, and logs and skips any template failing either - so a malformed template degrades to absence rather than to a server that will not start. +#### 5.5.2 Parameters -### Parameters +Use `POST /api/metalakes/{metalake}/jobs/runs` with the existing `jobTemplateName` and `jobConf` fields. -No REST API changes. The job is reached through the existing `POST /api/metalakes/{metalake}/jobs/runs` endpoint, whose `jobConf` map is opaque to the server and interpreted per template. +| `jobConf` key | CLI argument | Type | Required | Default | +| ------------------ | --------------- | --------------------------- | -------- | ----------------------------------- | +| `catalog_name` | `--catalog` | String | Yes | None | +| `table_identifier` | `--table` | String, such as `db.sample` | Yes | None | +| `use_caching` | `--use-caching` | Boolean | No | Installed Iceberg version's default | +| `spec_id` | `--spec-id` | Non-negative integer | No | Table's current spec ID | +| `spark_conf` | `--spark-conf` | JSON object | No | No additional overrides | -| `jobConf` key | CLI argument | Type | Required | Default | Description | -|---------------|--------------|------|----------|---------|-------------| -| `catalog_name` | `--catalog` | string | Yes | - | Iceberg catalog name as registered in Spark | -| `table_identifier` | `--table` | string | Yes | - | Fully qualified table name, such as `db.sample` | -| `use_caching` | `--use-caching` | boolean | No | Iceberg's default (`true`) | Cache table metadata in Spark for the duration of the rewrite | -| `spec_id` | `--spec-id` | int | No | The table's current spec | Rewrite manifests to this partition spec ID | -| `spark_conf` | `--spark-conf` | JSON object | No | None | Extra Spark configuration applied to the session | +Supply the standard Spark template configuration keys as for the other Iceberg jobs: `spark_master`, `spark_executor_instances`, `spark_executor_cores`, `spark_executor_memory`, `spark_driver_memory`, `catalog_type`, `catalog_uri`, and `warehouse_location`. -These are exactly the three in-parameters of Iceberg's `RewriteManifestsProcedure` (`table` required, `use_caching` and `spec_id` optional), plus Gravitino's standard `spark_conf`. `spec_id` is what makes the "align manifests with a better partition spec" half of the motivation reachable; without it only consolidation is available. +`use_caching` controls caching during the rewrite. Omission delegates to the runtime; Iceberg 1.11.0's action defaults to `false`. Set it explicitly when consistent behavior across runtime versions is required. -The standard Spark template keys - `spark_master`, `spark_executor_instances`, `spark_executor_cores`, `spark_executor_memory`, `spark_driver_memory`, `catalog_type`, `catalog_uri`, `warehouse_location` - behave as they do for every other Iceberg job template. +##### How to find `spec_id` -### Generated SQL +Omit `spec_id` for routine maintenance of the current partition spec. Do not guess IDs such as `0` or `1`. Read `default-spec-id` and the `partition-specs` array from the table's current Iceberg metadata JSON. The array maps each `spec-id` to its partition fields and transforms; `default-spec-id` identifies the current spec. These are Iceberg table metadata fields, not Gravitino catalog properties. -The job emits a `CALL` naming only the parameters the caller supplied: +To discover which specs have manifests in the current snapshot, run: ```sql --- catalog_name and table_identifier only -CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample') - --- with both optional parameters -CALL `iceberg_prod`.system.rewrite_manifests( - table => 'db.sample', use_caching => false, spec_id => 2) +SELECT DISTINCT partition_spec_id +FROM rest_catalog.db.t1.manifests; ``` -Two escaping rules, both from `IcebergJobUtils` and shared with the sibling jobs: - -| Value | Function | Rule | Example | -|-------|----------|------|---------| -| Catalog name | `escapeSqlIdentifier` | Double internal backticks, wrap in backticks | ``catalog`; DROP TABLE t; --`` becomes ``` `catalog``; DROP TABLE t; --` ``` | -| Table identifier | `escapeSqlString` | Double single quotes | `db.t' OR '1'='1` becomes `db.t'' OR ''1''=''1` | - -`use_caching` and `spec_id` never reach the statement as caller text. They are validated, then re-rendered from the parsed `boolean` and `int`, so the only characters that can appear are those `Boolean.toString` and `Integer.toString` produce. +This query lists represented specs, not which one is current. A defined spec with no manifests may be absent; use the metadata JSON to identify the default and interpret the transforms. -### Placeholder resolution, and why absent parameters need care +For example, suppose metadata shows spec `0` uses `day(event_time)` and the current spec `1` uses `hour(event_time)`. Omitting `spec_id` rewrites eligible manifests for spec `1`. Passing `"spec_id": "0"` consolidates the old day-spec manifests. Neither run converts day-partitioned data files to hour partitioning or rewrites manifests belonging to the other spec. -This is the one piece of behavior that is not obvious from the sibling jobs, and it dictated most of the job's input handling. +`spec_id` selects the existing spec whose manifests are eligible for rewriting, and replacement manifests use that same spec. Iceberg validates that the ID exists. This follows the [Iceberg 1.11.0 action implementation](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java), where `findMatchingManifests` compares each manifest's `partitionSpecId()` to the selected spec. It is not a partition-evolution operation. -A template's arguments are literal placeholder text. `JobManager.replacePlaceholder` substitutes `jobConf` values, and when a key is **absent it keeps the placeholder verbatim**: +#### 5.5.3 Procedure Call and Alternatives -```java -// core/src/main/java/org/apache/gravitino/job/JobManager.java -} else { - // If no replacement is found, keep the placeholder as is - matcher.appendReplacement(result, matcher.group(0)); -} -``` - -So a caller who omits `use_caching` does not produce a missing argument. They produce the literal string `{{use_caching}}` in `argv`, which parses as a value like any other: - -| Caller supplies | `argv` reaching the job | Naive handling | Result | -|-----------------|--------------------------|----------------|--------| -| `use_caching: "false"` | `--use-caching false` | `Boolean.parseBoolean` gives `false` | Correct | -| *nothing* | `--use-caching {{use_caching}}` | `Boolean.parseBoolean` gives `false` | **Wrong** - silently disables caching instead of leaving Iceberg's default | -| *nothing* | `--spec-id {{spec_id}}` | `Integer.parseInt` throws | Job fails, though the caller asked for nothing | - -The `false` row is the dangerous one: no error, no log line, and a parameter the caller never set now materially changes how the rewrite runs. The two-argument case is worse for `spec_id`, where the same mechanism turns an omitted parameter into a hard failure. - -The job therefore passes every optional value through `IcebergJobUtils.nullIfUnresolvedPlaceholder`, which returns `null` for a value that is entirely an unresolved placeholder: +```sql +-- Rewrite eligible manifests for the current spec. +CALL `iceberg_prod`.system.rewrite_manifests(table => 'db.sample'); -```java -String useCaching = IcebergJobUtils.nullIfUnresolvedPlaceholder(argMap.get("use-caching")); -String specId = IcebergJobUtils.nullIfUnresolvedPlaceholder(argMap.get("spec-id")); +-- Rewrite eligible manifests for a known existing spec. +CALL `iceberg_prod`.system.rewrite_manifests( + table => 'db.sample', use_caching => false, spec_id => 0); ``` -A `null` is then omitted from the generated SQL, which is what "absent means Iceberg's default" requires. Only a value that is *entirely* a placeholder is dropped - `{{a}}b` is a real value and survives - so a legitimate value that happens to contain braces is unaffected. - -The same hazard exists in `IcebergExpireSnapshotsJob` (`older_than`, `retain_last`) and in `IcebergRewriteDataFilesJob`. Fixing those is out of scope here and listed as follow-up work. +| Approach | Trade-off | Decision | +| --------------------------------- | ----------------------------------------------------------------------------------- | --------------------------------------------- | +| Spark SQL procedure | Reuses Spark catalog configuration and existing maintenance-job conventions | Chosen | +| Spark Java action | Offers finer control, but requires loading a table and invoking the action directly | Reserve for requirements beyond the procedure | +| Combine with data-file compaction | Couples operations that can run at different frequencies | Keep separate jobs | -### Validation +The Java action is a public API, not an Iceberg internal. The SQL procedure is chosen for consistency with the rewrite-data-files and expire-snapshots jobs, not because the Java API requires a separate implementation for every catalog backend. The deployed Spark and Iceberg runtime must support the supplied parameters; the job does not promise compatibility with every Iceberg release. -| Parameter | Accepted | Rejected | Why validate at all | -|-----------|----------|----------|---------------------| -| `use_caching` | `true`, `false`, any case | Anything else, for example `yes` | `Boolean.parseBoolean` maps every non-`true` string to `false`, so a typo would quietly invert the caller's intent | -| `spec_id` | Non-negative integer | Negative, non-numeric | Keeps a malformed value from reaching Spark as an unparseable SQL literal, and reports it against the argument name rather than as a SQL syntax error | +#### 5.5.4 Output -Both run before the `SparkSession` is created, so a bad argument costs no cluster resources. Both exit non-zero with the offending value in the message, then print usage. +The procedure returns `rewritten_manifests_count` and `added_manifests_count`. Iceberg 1.11.0 returns integer columns; the job reads them as `Number` and logs their values. Zero counts indicate a successful no-op, such as no eligible manifests needing rewriting. An empty result is guarded defensively, not assumed to be the normal no-op representation. -### Reading the procedure output +Counts concern the selected spec. A successful rewrite does not guarantee fewer manifests or improved latency for every workload; compare count, size, and planning behavior before and after. -`rewrite_manifests` returns one row, `(rewritten_manifests_count, added_manifests_count)`. Both are `IntegerType` in Iceberg 1.11.0, the version this repo compiles against, but the column width is an Iceberg implementation detail and Iceberg is `compileOnly` - the cluster supplies it, and a `getInt` against a future `LongType` would throw `ClassCastException` at the end of an otherwise successful rewrite. +#### 5.5.5 User Process -The job reads the columns as `Number`: - -```java -List results = spark.sql(sql).collectAsList(); -if (!results.isEmpty()) { - Row result = results.get(0); - System.out.printf( - "Rewrite Manifests Results:%n Rewritten manifests: %d%n Added manifests: %d%n", - ((Number) result.get(0)).longValue(), ((Number) result.get(1)).longValue()); -} -``` - -The empty-result guard matters because a procedure returning no rows is a legitimate no-op. - -### User process 1. Confirm the template is registered. `gravitino-jobs` must be in the server's `auxlib`: @@ -258,6 +194,10 @@ The empty-result guard matters because a procedure returning no rows is a legiti "catalog_name": "rest_catalog", "table_identifier": "db.t1", "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": "" @@ -279,109 +219,79 @@ The empty-result guard matters because a procedure returning no rows is a legiti "/tmp/gravitino/jobs/staging/test/builtin-iceberg-rewrite-manifests/${job_id}/output.log" ``` -5. Confirm against the table. `added_manifests_count` well below the previous manifest count is the outcome being sought: +5. Confirm against the table. Compare manifest count and average size before and after the run, grouped by spec. Counts reported by the procedure cover only the selected spec: ```sql - SELECT count(*) FROM rest_catalog.db.t1.manifests; + SELECT partition_spec_id, COUNT(*) AS manifest_count, AVG(length) AS avg_manifest_size + FROM rest_catalog.db.t1.manifests + GROUP BY partition_spec_id; ``` Pair the job with `builtin-iceberg-expire-snapshots`. The rewrite creates a snapshot and leaves the pre-rewrite manifests reachable from the previous one, so storage is only reclaimed once those snapshots expire. -### Implementation process - -``` -Client - | POST /api/metalakes/{metalake}/jobs/runs { jobTemplateName, jobConf } - v -JobOperations (server/) - | - v -JobManager (core/) - | 1. look up the registered template by name - | 2. replacePlaceholder over arguments and configs, using jobConf - | (absent key -> placeholder kept verbatim) - | 3. persist the resolved template on the JobEntity - v -JobExecutor --spark-submit--> IcebergRewriteManifestsJob.main(argv) - | 4. IcebergJobUtils.parseArguments - | 5. nullIfUnresolvedPlaceholder on optionals - | 6. validateUseCaching / validateSpecId -- fail here, before Spark - | 7. build SparkSession (+ spark_conf) - | 8. buildProcedureCall -> escaped CALL text - v - Spark --resolves spark.sql.catalog..*--> Iceberg - | | - | 9. one row: rewritten/added counts <--------+ - v - stdout -> staging output.log -``` -Steps 4 through 6 all run before step 7, so every input error is reported without acquiring a Spark session. +## 6. Safety Considerations -### Backward compatibility +### 6.1 Input Validation -Additive in every respect: +| Input | Behavior | +| ----------------------- | --------------------------------------------------------------------------- | +| `use_caching` | Accept `true` or `false`, case-insensitively; reject other supplied values | +| `spec_id` | Parse a non-negative 32-bit integer; reject malformed or overflowing values | +| Unknown spec ID | Iceberg rejects IDs absent from table metadata | +| Omitted optional values | Omit them from the SQL call and preserve runtime defaults | +| Catalog and table | Use shared identifier and string escaping when building SQL | -- **New template name.** No existing name, argument, or `jobConf` key changes meaning. Callers of the other three templates are unaffected. -- **No API change.** No new or modified REST endpoint, DTO, client method, or entity, so no OpenAPI or client update is needed. -- **No schema change.** Job runs persist through the existing `JobEntity`. -- **Servers without `gravitino-jobs`** simply do not list the template, exactly as with the other built-ins today. -- **Versioned template.** `version` is `v1`. A later change to arguments or semantics bumps it, following the convention the sibling templates set. +Local validation occurs before creating the Spark session. This avoids starting a session for malformed values, but does not imply that `spark-submit` has allocated no resources. Catalog access and spec existence are validated by Iceberg during execution. -### Follow-up: metrics and a policy for automatic triggering +### 6.2 Metadata and Concurrent Writers -Automatic triggering is a Non-Goal above; this sketches the shape so the follow-up issues have a starting point rather than leaving the path undefined. +Iceberg owns the metadata commit and conflict handling. The job does not edit metadata JSON or delete files directly. A failed or uncertain commit must be checked through the table and job status before resubmission. Snapshot expiration remains a separate operation with its own retention requirements. -**Metrics.** `IcebergUpdateStatsAndMetricsJob` aggregates the `.files` metadata table and writes `custom-`-prefixed statistics such as `custom-data-file-mse` and `custom-small-file-number`. The manifest analogue reads the `.manifests` metadata table: +### 6.3 Backward Compatibility -| Proposed statistic | Source | Purpose | -|--------------------|--------|---------| -| `custom-manifest-number` | `COUNT(*)` | The primary signal; planning cost tracks it | -| `custom-avg-manifest-size` | `AVG(length)` | Distinguishes many-small from few-large | -| `custom-small-manifest-number` | `COUNT(*)` below a threshold | Directly comparable to `custom-small-file-number` | -| `custom-manifest-spec-count` | `COUNT(DISTINCT partition_spec_id)` | More than one live spec is what makes `spec_id` useful | +The template is additive. No existing template name, REST endpoint, client API, or database schema changes. Existing job persistence and status tracking are reused. Installing a runtime that supports the procedure and its supplied parameters remains an operator prerequisite. -**Policy.** A `system_iceberg_rewrite_manifests` policy type with a `iceberg-manifest-rewrite` strategy type, a `StrategyHandler` registered under `gravitino.optimizer.strategyHandler.*`, and thresholds over the metrics above. It reuses the existing recommender path, so no new framework is required - `CompactionStrategyHandler` is the template to follow. +## 7. File Changes Summary -To stay aligned with table compaction, the handler declares `DataRequirement.TABLE_STATISTICS` in `dataRequirements()` and reads its thresholds from table statistics rather than from a side channel, exactly as `CompactionStrategyHandler` does. Any "when did this last run" signal belongs in `statistic_meta` as a `custom-`-prefixed statistic written back after a successful run - the same path `IcebergUpdateStatsAndMetricsJob` already uses - rather than in new bespoke state. +### 7.1 New Files -Sequencing matters: the metrics must land, and be collected for long enough to calibrate thresholds against real tables, before a policy that fires on them is worth shipping. +| File under `maintenance/jobs/src/main/java/` | Purpose | +| ------------------------------------------------------------------------------- | ------------------------------ | +| `org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java` | Template and Spark entry point | ---- +### 7.2 Modified Files -## Task Breakdown +| File | Purpose | +| ----------------------------------------------------------- | -------------------------------------------------------------------- | +| `BuiltInJobTemplateProvider.java` | Register the template | +| `IcebergJobUtils.java` | Reuse shared parsing and escaping; normalize omitted optional values | +| `docs/table-maintenance-service/optimizer-cli-reference.md` | Parameters, spec discovery, and submission examples | -### Phase 1: The job (this design, PR [#12937](https://github.com/apache/gravitino/pull/12937)) +### 7.3 Test Files -- [x] Implement `IcebergRewriteManifestsJob` with `table` and `use_caching`, calling `rewrite_manifests` through Spark SQL -- [x] Register the template in `BuiltInJobTemplateProvider` -- [x] Move shared SQL escaping, argument parsing, and Spark config parsing to `IcebergJobUtils` rather than reaching into `IcebergRewriteDataFilesJob` -- [x] Add `IcebergJobUtils.nullIfUnresolvedPlaceholder` and drop unresolved placeholders for optional arguments -- [x] Validate `use_caching`, rejecting values that are neither `true` nor `false` -- [x] Read procedure output as `Number` so the job tolerates the runtime Iceberg version's column width -- [x] Support `spec_id`, with non-negative-integer validation -- [x] Unit tests: template shape, argument parsing, validation, placeholder filtering, generated SQL, injection escaping -- [x] Document the template and its parameters in `docs/table-maintenance-service/optimizer-cli-reference.md` -- [ ] Add `TestIcebergRewriteManifestsJobWithSpark`, following the two existing `*WithSpark` tests, asserting manifest count falls across a real rewrite +| Test | Coverage | +| --------------------------------------------------- | ---------------------------------------------------------------------- | +| `TestIcebergRewriteManifestsJob` | Template, parsing, validation, and generated SQL | +| `TestIcebergRewriteManifestsJobWithSpark` (planned) | Real procedure execution, preserved table contents, and spec selection | -### Phase 2: Manifest metrics (separate issue) +## 8. Proposed PR Plan -- [ ] Add manifest aggregation over the `.manifests` metadata table to `IcebergUpdateStatsAndMetricsJob` -- [ ] Emit `custom-manifest-number`, `custom-avg-manifest-size`, `custom-small-manifest-number`, `custom-manifest-spec-count` -- [ ] Unit tests for the aggregation SQL and the statistic mapping -- [ ] Document the new statistics alongside the existing ones +1. **Job implementation (#12937)**: Register the template, implement procedure execution and validation, and document submission and spec discovery. Add a Spark-backed test with manifests under two specs: verify that a run affects only the selected spec, preserves records, and handles a no-op. +2. **Manifest statistics**: Collect `custom-manifest-number` and `custom-avg-manifest-size`, with aggregation tests and user documentation. +3. **Automatic triggering**: Once measurements inform thresholds and spec selection, propose typed policy content, a handler, and an adapter. Test the complete collect, evaluate, and submit flow. -### Phase 3: Policy-driven triggering (separate issue, depends on Phase 2) +## 9. Open Questions -- [ ] Define the `system_iceberg_rewrite_manifests` policy type and its typed content in `api/` -- [ ] Implement a `StrategyHandler` for strategy type `iceberg-manifest-rewrite` -- [ ] Wire the handler into job submission so the recommender can submit `builtin-iceberg-rewrite-manifests` -- [ ] Unit tests for threshold evaluation and job context construction -- [ ] Integration test for the full collect / evaluate / submit path -- [ ] Document the policy, mirroring `docs/iceberg-compaction-policy.md` +- Which measured thresholds justify a rewrite without repeatedly submitting no-op jobs? +- How should automatic triggering choose specs after partition evolution? Table-wide aggregates alone are insufficient. +- Which Spark and Iceberg versions should be covered by the job's integration-test matrix? -### Phase 4: Placeholder hardening across the existing jobs (separate issue) +## 10. Comparison with Other Maintenance Flows -- [ ] Apply `nullIfUnresolvedPlaceholder` to `older_than` and `retain_last` in `IcebergExpireSnapshotsJob` -- [ ] Apply it to the optional arguments of `IcebergRewriteDataFilesJob` and `IcebergUpdateStatsAndMetricsJob` -- [ ] Regression tests per job proving an omitted optional parameter leaves the engine default in force +| Job | Primary target | Changes data-file layout? | Partition-spec behavior | +| ------------------- | --------------------------------------- | ------------------------- | ----------------------------------------------------- | +| Rewrite data files | Data files | Yes | Governed by the data rewrite operation | +| Expire snapshots | Snapshot history and unreferenced files | No | Governed by snapshot retention | +| Remove orphan files | Unreferenced storage objects | No | Not a spec rewrite | +| Rewrite manifests | Manifest entries for an existing spec | No | Selects one spec; does not migrate data between specs | From 17f87e3f207cd6ace51e72f7cf05437540a6093e Mon Sep 17 00:00:00 2001 From: Akshay Thorat Date: Tue, 15 Sep 2026 18:30:04 -0700 Subject: [PATCH 3/5] docs: resolve manifest maintenance policy design questions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Define trigger defaults, per-spec object statistics, consistent spec selection, and the initial Spark/Iceberg compatibility target. Co-authored-by: İbrahim ERBİLEN --- design-docs/iceberg-rewrite-manifests-job.md | 79 ++++++++++++++++---- 1 file changed, 64 insertions(+), 15 deletions(-) diff --git a/design-docs/iceberg-rewrite-manifests-job.md b/design-docs/iceberg-rewrite-manifests-job.md index 29a7ee4176e..b8ae5c7c2af 100644 --- a/design-docs/iceberg-rewrite-manifests-job.md +++ b/design-docs/iceberg-rewrite-manifests-job.md @@ -79,24 +79,71 @@ The initial implementation enters at the jobs API. A future policy-driven flow w ### 5.2 Layer 1 - Policy Definition (`api/`, follow-up) -A future `system_iceberg_rewrite_manifests` policy can express thresholds over manifest statistics. Its typed fields and defaults need a separate proposal informed by collected measurements. No policy type is added by #12937. +The follow-up `system_iceberg_rewrite_manifests` policy uses the following configurable thresholds. No policy type is added by #12937. + +| Policy setting | Default | Meaning | +| ----------------------------------- | ----------------- | ----------------------------------------------------------------- | +| `manifest_count_critical` | `500` | Trigger at or above this count, regardless of average size | +| `manifest_count_warning` | `100` | Minimum count for the size-based trigger | +| `avg_manifest_size_threshold_bytes` | `8388608` (8 MiB) | Trigger below this average size when the warning count is reached | + +Evaluate the following expression using statistics for the resolved target spec only: + +```text +IF manifest_count >= manifest_count_critical: + trigger +ELSE IF manifest_count >= manifest_count_warning + AND avg_manifest_size_bytes < avg_manifest_size_threshold_bytes: + trigger +ELSE: + do not trigger +``` + +Count comparisons are inclusive; the size comparison is strict. For example, 500 manifests trigger regardless of size, 100 manifests averaging less than 8 MiB trigger, and 100 manifests averaging exactly 8 MiB do not. Counts below 100 do not trigger under these defaults. ### 5.3 Layer 2 - Strategy Handler (`maintenance/optimizer/`, follow-up) -Start by collecting only these two table-level statistics through `IcebergUpdateStatsAndMetricsJob`: +#### 5.3.1 Resolve the Target Spec + +At the start of a collection/evaluation cycle, resolve `spec_id` from the requested ID or, if omitted, from the table's `default-spec-id`. Validate that the resolved spec exists and carry that ID through collection, evaluation, and job submission. Do not resolve the default again between these steps: partition evolution could otherwise make the job target a different spec from the one evaluated. + +Collect only manifests in the current snapshot whose `partition_spec_id` equals that resolved ID: + +```sql +-- Example: the resolved spec ID is 1. +SELECT COUNT(*) AS manifest_count, AVG(length) AS avg_manifest_size_bytes +FROM rest_catalog.db.t1.manifests +WHERE partition_spec_id = 1; +``` + +The SQL ID is rendered from a validated integer. With no matching manifests, record a count of zero and normalize the null average to zero; this cannot trigger under the default count thresholds. A missing map entry means statistics have not been collected for that spec, not zero manifests: collect it before evaluating. + +#### 5.3.2 Statistics Storage + +Store one row per table and statistic name in `statistic_meta`, not one row per spec. Each statistic value is an object keyed by the decimal spec ID. Use the existing `StatisticValues.objectValue` representation with numeric values, retaining two statistics only: + +| Statistic name | Value for each spec key | Purpose | +| ---------------------------------- | ---------------------------------- | ----------------------------------- | +| `custom-manifest-number-by-spec` | Long from `COUNT(*)` | Manifest count for that spec | +| `custom-avg-manifest-size-by-spec` | Double from `AVG(length)` in bytes | Average manifest size for that spec | -| Proposed statistic | Source from `.manifests` | Purpose | -| -------------------------- | ------------------------ | --------------------------------------------------- | -| `custom-manifest-number` | `COUNT(*)` | Measures current snapshot manifest count | -| `custom-avg-manifest-size` | `AVG(length)` | Distinguishes many small manifests from larger ones | +Example logical contents of the two rows for one table: -Do not add small-manifest counts or spec-count statistics without a demonstrated policy use case. A future handler declares `DataRequirement.TABLE_STATISTICS` and reads measured values through the existing statistics path. Policy thresholds come from policy content. Any last-success time needed for a cooldown belongs in `statistic_meta` as a `custom-` statistic written after successful completion. +```text +statistic_name: custom-manifest-number-by-spec +statistic_value: {"0": 620, "1": 120} + +statistic_name: custom-avg-manifest-size-by-spec +statistic_value: {"0": 10485760.0, "1": 4194304.0} +``` -These aggregates span all specs in the current snapshot. Since a rewrite targets one spec, an automatic policy must account for this mismatch before shipping, for example by evaluating per-spec measurements. A high total count alone does not prove the default-spec rewrite will reduce it. +These illustrate the object values, not a new REST serialization format. A collection run for spec `1` replaces only key `"1"` in each map and preserves key `"0"` and every other spec entry. Read and merge the existing maps before writing them through the table-statistics API. Coordinate concurrent collectors for the same table so one read/merge/write does not overwrite another spec's update. Publish both measurements from the same collection and do not evaluate a partially updated pair; the follow-up implementation must test these update guarantees. + +The handler declares `DataRequirement.TABLE_STATISTICS` and looks up the same resolved spec key in both objects. The threshold expression uses those two numeric values, never table-wide totals or values from another spec. Do not add small-manifest counts or spec-count statistics. Any last-success time required for a future cooldown belongs in `statistic_meta` as a `custom-` statistic written after successful completion. ### 5.4 Layer 3 - Job Adapter (`maintenance/optimizer/`, follow-up) -A future adapter maps a strategy decision to `builtin-iceberg-rewrite-manifests`, supplies the catalog and table identifiers, and forwards the selected spec and caching option. It reuses existing job submission and tracking. The initial PR requires no adapter because operators submit the template directly. +The adapter maps a positive strategy decision to `builtin-iceberg-rewrite-manifests`, supplying the catalog, table, caching option, and the exact resolved `spec_id` used by the collector and trigger expression. Always include that resolved ID in the submitted `jobConf`, even when the original request omitted it. This preserves the target if the table default changes after collection. Reuse existing job submission and tracking. The initial PR requires no adapter because operators submit the template directly. ### 5.5 Layer 4 - Spark Job (`maintenance/jobs/`) @@ -278,14 +325,16 @@ The template is additive. No existing template name, REST endpoint, client API, ## 8. Proposed PR Plan 1. **Job implementation (#12937)**: Register the template, implement procedure execution and validation, and document submission and spec discovery. Add a Spark-backed test with manifests under two specs: verify that a run affects only the selected spec, preserves records, and handles a no-op. -2. **Manifest statistics**: Collect `custom-manifest-number` and `custom-avg-manifest-size`, with aggregation tests and user documentation. -3. **Automatic triggering**: Once measurements inform thresholds and spec selection, propose typed policy content, a handler, and an adapter. Test the complete collect, evaluate, and submit flow. +2. **Manifest statistics**: Collect `custom-manifest-number-by-spec` and `custom-avg-manifest-size-by-spec`. Test spec-filtered aggregation, empty results, missing entries, merge updates preserving other specs, and concurrent collection updates. Document the two object-valued statistics. +3. **Automatic triggering**: Implement the policy defaults and expression in Section 5.2, the handler, and the adapter. Test counts at 99, 100, 499, and 500; sizes below, at, and above 8 MiB; and a default-spec change between collection and submission. Verify the submitted ID matches the collected and evaluated key. + +## 9. Review Decisions and Initial Compatibility -## 9. Open Questions +The three open questions are resolved: -- Which measured thresholds justify a rewrite without repeatedly submitting no-op jobs? -- How should automatic triggering choose specs after partition evolution? Table-wide aggregates alone are insufficient. -- Which Spark and Iceberg versions should be covered by the job's integration-test matrix? +1. **Trigger thresholds**: Use the critical-count or warning-count-plus-size expression in Section 5.2, with defaults of 500, 100, and 8 MiB respectively. +2. **Spec consistency**: Collect, store, evaluate, and submit against the same resolved spec. Represent multiple specs in one object-valued row per statistic name, as specified in Sections 5.3 and 5.4. +3. **Initial runtime support**: Commit to and run CI against **Spark 3.5 and Iceberg 1.11.0** for the initial release, aligned with the other built-in Iceberg jobs. Other version combinations are outside the initial compatibility commitment. The Spark-backed tests in the PR plan must run on this combination. ## 10. Comparison with Other Maintenance Flows From cfa176efdab11fbdfb0255aea6ddcc75eeec3fb7 Mon Sep 17 00:00:00 2001 From: Akshay Thorat Date: Wed, 16 Sep 2026 10:17:21 -0700 Subject: [PATCH 4/5] docs: align manifest job configuration and CLI names MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: İbrahim ERBİLEN --- design-docs/iceberg-rewrite-manifests-job.md | 38 ++++++++++---------- 1 file changed, 19 insertions(+), 19 deletions(-) diff --git a/design-docs/iceberg-rewrite-manifests-job.md b/design-docs/iceberg-rewrite-manifests-job.md index b8ae5c7c2af..938121087cb 100644 --- a/design-docs/iceberg-rewrite-manifests-job.md +++ b/design-docs/iceberg-rewrite-manifests-job.md @@ -32,7 +32,7 @@ Gravitino needs a built-in job that operators can submit through the existing jo ## 2. Goals 1. **Job submission**: Register `builtin-iceberg-rewrite-manifests` and submit it through the existing jobs REST API. -2. **Procedure parameters**: Expose `table`, `use_caching`, and `spec_id`, with omitted optional values delegated to Iceberg's defaults. +2. **Procedure parameters**: Expose Iceberg parameters `table`, `use_caching`, and `spec_id`, with omitted optional values delegated to Iceberg's defaults. 3. **Operator guidance**: Explain how to discover spec IDs and which manifests a run can rewrite. 4. **Validation and results**: Reject malformed arguments, escape SQL inputs, and report rewritten and added manifest counts. @@ -105,7 +105,7 @@ Count comparisons are inclusive; the size comparison is strict. For example, 500 #### 5.3.1 Resolve the Target Spec -At the start of a collection/evaluation cycle, resolve `spec_id` from the requested ID or, if omitted, from the table's `default-spec-id`. Validate that the resolved spec exists and carry that ID through collection, evaluation, and job submission. Do not resolve the default again between these steps: partition evolution could otherwise make the job target a different spec from the one evaluated. +At the start of a collection/evaluation cycle, resolve `spec-id` from the requested ID or, if omitted, from the table's `default-spec-id`. Validate that the resolved spec exists and carry that ID through collection, evaluation, and job submission. Do not resolve the default again between these steps: partition evolution could otherwise make the job target a different spec from the one evaluated. Collect only manifests in the current snapshot whose `partition_spec_id` equals that resolved ID: @@ -143,7 +143,7 @@ The handler declares `DataRequirement.TABLE_STATISTICS` and looks up the same re ### 5.4 Layer 3 - Job Adapter (`maintenance/optimizer/`, follow-up) -The adapter maps a positive strategy decision to `builtin-iceberg-rewrite-manifests`, supplying the catalog, table, caching option, and the exact resolved `spec_id` used by the collector and trigger expression. Always include that resolved ID in the submitted `jobConf`, even when the original request omitted it. This preserves the target if the table default changes after collection. Reuse existing job submission and tracking. The initial PR requires no adapter because operators submit the template directly. +The adapter maps a positive strategy decision to `builtin-iceberg-rewrite-manifests`, supplying the catalog, table, caching option, and the exact resolved `spec-id` used by the collector and trigger expression. Always include that resolved ID in the submitted `jobConf`, even when the original request omitted it. This preserves the target if the table default changes after collection. Reuse existing job submission and tracking. The initial PR requires no adapter because operators submit the template directly. ### 5.5 Layer 4 - Spark Job (`maintenance/jobs/`) @@ -162,23 +162,23 @@ The job parses and validates its inputs, builds a Spark session, calls the proce #### 5.5.2 Parameters -Use `POST /api/metalakes/{metalake}/jobs/runs` with the existing `jobTemplateName` and `jobConf` fields. +Use `POST /api/metalakes/{metalake}/jobs/runs` with the existing `jobTemplateName` and `jobConf` fields. Each job argument key matches its CLI flag without the leading `--`. Translate `use-caching` and `spec-id` to Iceberg SQL parameters `use_caching` and `spec_id` when building the procedure call. Standard Spark configuration placeholders retain their existing names. | `jobConf` key | CLI argument | Type | Required | Default | | ------------------ | --------------- | --------------------------- | -------- | ----------------------------------- | -| `catalog_name` | `--catalog` | String | Yes | None | -| `table_identifier` | `--table` | String, such as `db.sample` | Yes | None | -| `use_caching` | `--use-caching` | Boolean | No | Installed Iceberg version's default | -| `spec_id` | `--spec-id` | Non-negative integer | No | Table's current spec ID | -| `spark_conf` | `--spark-conf` | JSON object | No | No additional overrides | +| `catalog` | `--catalog` | String | Yes | None | +| `table` | `--table` | String, such as `db.sample` | Yes | None | +| `use-caching` | `--use-caching` | Boolean | No | Installed Iceberg version's default | +| `spec-id` | `--spec-id` | Non-negative integer | No | Table's current spec ID | +| `spark-conf` | `--spark-conf` | JSON object | No | No additional overrides | Supply the standard Spark template configuration keys as for the other Iceberg jobs: `spark_master`, `spark_executor_instances`, `spark_executor_cores`, `spark_executor_memory`, `spark_driver_memory`, `catalog_type`, `catalog_uri`, and `warehouse_location`. -`use_caching` controls caching during the rewrite. Omission delegates to the runtime; Iceberg 1.11.0's action defaults to `false`. Set it explicitly when consistent behavior across runtime versions is required. +`use-caching` controls caching during the rewrite. Omission delegates to the runtime; Iceberg 1.11.0's action defaults to `false`. Set it explicitly when consistent behavior across runtime versions is required. -##### How to find `spec_id` +##### How to find `spec-id` -Omit `spec_id` for routine maintenance of the current partition spec. Do not guess IDs such as `0` or `1`. Read `default-spec-id` and the `partition-specs` array from the table's current Iceberg metadata JSON. The array maps each `spec-id` to its partition fields and transforms; `default-spec-id` identifies the current spec. These are Iceberg table metadata fields, not Gravitino catalog properties. +Omit `spec-id` for routine maintenance of the current partition spec. Do not guess IDs such as `0` or `1`. Read `default-spec-id` and the `partition-specs` array from the table's current Iceberg metadata JSON. The array maps each `spec-id` to its partition fields and transforms; `default-spec-id` identifies the current spec. These are Iceberg table metadata fields, not Gravitino catalog properties. To discover which specs have manifests in the current snapshot, run: @@ -189,9 +189,9 @@ FROM rest_catalog.db.t1.manifests; This query lists represented specs, not which one is current. A defined spec with no manifests may be absent; use the metadata JSON to identify the default and interpret the transforms. -For example, suppose metadata shows spec `0` uses `day(event_time)` and the current spec `1` uses `hour(event_time)`. Omitting `spec_id` rewrites eligible manifests for spec `1`. Passing `"spec_id": "0"` consolidates the old day-spec manifests. Neither run converts day-partitioned data files to hour partitioning or rewrites manifests belonging to the other spec. +For example, suppose metadata shows spec `0` uses `day(event_time)` and the current spec `1` uses `hour(event_time)`. Omitting `spec-id` rewrites eligible manifests for spec `1`. Passing `"spec-id": "0"` consolidates the old day-spec manifests. Neither run converts day-partitioned data files to hour partitioning or rewrites manifests belonging to the other spec. -`spec_id` selects the existing spec whose manifests are eligible for rewriting, and replacement manifests use that same spec. Iceberg validates that the ID exists. This follows the [Iceberg 1.11.0 action implementation](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java), where `findMatchingManifests` compares each manifest's `partitionSpecId()` to the selected spec. It is not a partition-evolution operation. +`spec-id` selects the existing spec whose manifests are eligible for rewriting, and replacement manifests use that same spec. Iceberg validates that the ID exists. This follows the [Iceberg 1.11.0 action implementation](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java), where `findMatchingManifests` compares each manifest's `partitionSpecId()` to the selected spec. It is not a partition-evolution operation. #### 5.5.3 Procedure Call and Alternatives @@ -230,7 +230,7 @@ Counts concern the selected spec. A successful rewrite does not guarantee fewer The list must contain `builtin-iceberg-rewrite-manifests`. -2. Submit a run. Only `catalog_name` and `table_identifier` are needed beyond the standard Spark keys: +2. Submit a run. Only `catalog` and `table` are needed beyond the standard Spark keys: ```bash job_id=$(curl -sS -X POST -H "Accept: application/vnd.gravitino.v1+json" \ @@ -238,8 +238,8 @@ Counts concern the selected spec. A successful rewrite does not guarantee fewer -d '{ "jobTemplateName": "builtin-iceberg-rewrite-manifests", "jobConf": { - "catalog_name": "rest_catalog", - "table_identifier": "db.t1", + "catalog": "rest_catalog", + "table": "db.t1", "spark_master": "local[2]", "spark_executor_instances": "1", "spark_executor_cores": "1", @@ -283,8 +283,8 @@ Pair the job with `builtin-iceberg-expire-snapshots`. The rewrite creates a snap | Input | Behavior | | ----------------------- | --------------------------------------------------------------------------- | -| `use_caching` | Accept `true` or `false`, case-insensitively; reject other supplied values | -| `spec_id` | Parse a non-negative 32-bit integer; reject malformed or overflowing values | +| `use-caching` | Accept `true` or `false`, case-insensitively; reject other supplied values | +| `spec-id` | Parse a non-negative 32-bit integer; reject malformed or overflowing values | | Unknown spec ID | Iceberg rejects IDs absent from table metadata | | Omitted optional values | Omit them from the SQL call and preserve runtime defaults | | Catalog and table | Use shared identifier and string escaping when building SQL | From 164422690df98e763b0174d78554dfc0ac0fed29 Mon Sep 17 00:00:00 2001 From: Akshay Thorat Date: Thu, 17 Sep 2026 11:03:25 -0700 Subject: [PATCH 5/5] docs: restore original manifest job configuration naming MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Keep the built-in job naming style after the CLI/key equality restriction was withdrawn in review. Co-authored-by: İbrahim ERBİLEN --- design-docs/iceberg-rewrite-manifests-job.md | 38 ++++++++++---------- 1 file changed, 19 insertions(+), 19 deletions(-) diff --git a/design-docs/iceberg-rewrite-manifests-job.md b/design-docs/iceberg-rewrite-manifests-job.md index 938121087cb..b8ae5c7c2af 100644 --- a/design-docs/iceberg-rewrite-manifests-job.md +++ b/design-docs/iceberg-rewrite-manifests-job.md @@ -32,7 +32,7 @@ Gravitino needs a built-in job that operators can submit through the existing jo ## 2. Goals 1. **Job submission**: Register `builtin-iceberg-rewrite-manifests` and submit it through the existing jobs REST API. -2. **Procedure parameters**: Expose Iceberg parameters `table`, `use_caching`, and `spec_id`, with omitted optional values delegated to Iceberg's defaults. +2. **Procedure parameters**: Expose `table`, `use_caching`, and `spec_id`, with omitted optional values delegated to Iceberg's defaults. 3. **Operator guidance**: Explain how to discover spec IDs and which manifests a run can rewrite. 4. **Validation and results**: Reject malformed arguments, escape SQL inputs, and report rewritten and added manifest counts. @@ -105,7 +105,7 @@ Count comparisons are inclusive; the size comparison is strict. For example, 500 #### 5.3.1 Resolve the Target Spec -At the start of a collection/evaluation cycle, resolve `spec-id` from the requested ID or, if omitted, from the table's `default-spec-id`. Validate that the resolved spec exists and carry that ID through collection, evaluation, and job submission. Do not resolve the default again between these steps: partition evolution could otherwise make the job target a different spec from the one evaluated. +At the start of a collection/evaluation cycle, resolve `spec_id` from the requested ID or, if omitted, from the table's `default-spec-id`. Validate that the resolved spec exists and carry that ID through collection, evaluation, and job submission. Do not resolve the default again between these steps: partition evolution could otherwise make the job target a different spec from the one evaluated. Collect only manifests in the current snapshot whose `partition_spec_id` equals that resolved ID: @@ -143,7 +143,7 @@ The handler declares `DataRequirement.TABLE_STATISTICS` and looks up the same re ### 5.4 Layer 3 - Job Adapter (`maintenance/optimizer/`, follow-up) -The adapter maps a positive strategy decision to `builtin-iceberg-rewrite-manifests`, supplying the catalog, table, caching option, and the exact resolved `spec-id` used by the collector and trigger expression. Always include that resolved ID in the submitted `jobConf`, even when the original request omitted it. This preserves the target if the table default changes after collection. Reuse existing job submission and tracking. The initial PR requires no adapter because operators submit the template directly. +The adapter maps a positive strategy decision to `builtin-iceberg-rewrite-manifests`, supplying the catalog, table, caching option, and the exact resolved `spec_id` used by the collector and trigger expression. Always include that resolved ID in the submitted `jobConf`, even when the original request omitted it. This preserves the target if the table default changes after collection. Reuse existing job submission and tracking. The initial PR requires no adapter because operators submit the template directly. ### 5.5 Layer 4 - Spark Job (`maintenance/jobs/`) @@ -162,23 +162,23 @@ The job parses and validates its inputs, builds a Spark session, calls the proce #### 5.5.2 Parameters -Use `POST /api/metalakes/{metalake}/jobs/runs` with the existing `jobTemplateName` and `jobConf` fields. Each job argument key matches its CLI flag without the leading `--`. Translate `use-caching` and `spec-id` to Iceberg SQL parameters `use_caching` and `spec_id` when building the procedure call. Standard Spark configuration placeholders retain their existing names. +Use `POST /api/metalakes/{metalake}/jobs/runs` with the existing `jobTemplateName` and `jobConf` fields. | `jobConf` key | CLI argument | Type | Required | Default | | ------------------ | --------------- | --------------------------- | -------- | ----------------------------------- | -| `catalog` | `--catalog` | String | Yes | None | -| `table` | `--table` | String, such as `db.sample` | Yes | None | -| `use-caching` | `--use-caching` | Boolean | No | Installed Iceberg version's default | -| `spec-id` | `--spec-id` | Non-negative integer | No | Table's current spec ID | -| `spark-conf` | `--spark-conf` | JSON object | No | No additional overrides | +| `catalog_name` | `--catalog` | String | Yes | None | +| `table_identifier` | `--table` | String, such as `db.sample` | Yes | None | +| `use_caching` | `--use-caching` | Boolean | No | Installed Iceberg version's default | +| `spec_id` | `--spec-id` | Non-negative integer | No | Table's current spec ID | +| `spark_conf` | `--spark-conf` | JSON object | No | No additional overrides | Supply the standard Spark template configuration keys as for the other Iceberg jobs: `spark_master`, `spark_executor_instances`, `spark_executor_cores`, `spark_executor_memory`, `spark_driver_memory`, `catalog_type`, `catalog_uri`, and `warehouse_location`. -`use-caching` controls caching during the rewrite. Omission delegates to the runtime; Iceberg 1.11.0's action defaults to `false`. Set it explicitly when consistent behavior across runtime versions is required. +`use_caching` controls caching during the rewrite. Omission delegates to the runtime; Iceberg 1.11.0's action defaults to `false`. Set it explicitly when consistent behavior across runtime versions is required. -##### How to find `spec-id` +##### How to find `spec_id` -Omit `spec-id` for routine maintenance of the current partition spec. Do not guess IDs such as `0` or `1`. Read `default-spec-id` and the `partition-specs` array from the table's current Iceberg metadata JSON. The array maps each `spec-id` to its partition fields and transforms; `default-spec-id` identifies the current spec. These are Iceberg table metadata fields, not Gravitino catalog properties. +Omit `spec_id` for routine maintenance of the current partition spec. Do not guess IDs such as `0` or `1`. Read `default-spec-id` and the `partition-specs` array from the table's current Iceberg metadata JSON. The array maps each `spec-id` to its partition fields and transforms; `default-spec-id` identifies the current spec. These are Iceberg table metadata fields, not Gravitino catalog properties. To discover which specs have manifests in the current snapshot, run: @@ -189,9 +189,9 @@ FROM rest_catalog.db.t1.manifests; This query lists represented specs, not which one is current. A defined spec with no manifests may be absent; use the metadata JSON to identify the default and interpret the transforms. -For example, suppose metadata shows spec `0` uses `day(event_time)` and the current spec `1` uses `hour(event_time)`. Omitting `spec-id` rewrites eligible manifests for spec `1`. Passing `"spec-id": "0"` consolidates the old day-spec manifests. Neither run converts day-partitioned data files to hour partitioning or rewrites manifests belonging to the other spec. +For example, suppose metadata shows spec `0` uses `day(event_time)` and the current spec `1` uses `hour(event_time)`. Omitting `spec_id` rewrites eligible manifests for spec `1`. Passing `"spec_id": "0"` consolidates the old day-spec manifests. Neither run converts day-partitioned data files to hour partitioning or rewrites manifests belonging to the other spec. -`spec-id` selects the existing spec whose manifests are eligible for rewriting, and replacement manifests use that same spec. Iceberg validates that the ID exists. This follows the [Iceberg 1.11.0 action implementation](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java), where `findMatchingManifests` compares each manifest's `partitionSpecId()` to the selected spec. It is not a partition-evolution operation. +`spec_id` selects the existing spec whose manifests are eligible for rewriting, and replacement manifests use that same spec. Iceberg validates that the ID exists. This follows the [Iceberg 1.11.0 action implementation](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteManifestsSparkAction.java), where `findMatchingManifests` compares each manifest's `partitionSpecId()` to the selected spec. It is not a partition-evolution operation. #### 5.5.3 Procedure Call and Alternatives @@ -230,7 +230,7 @@ Counts concern the selected spec. A successful rewrite does not guarantee fewer The list must contain `builtin-iceberg-rewrite-manifests`. -2. Submit a run. Only `catalog` and `table` are needed beyond the standard Spark keys: +2. Submit a run. Only `catalog_name` and `table_identifier` are needed beyond the standard Spark keys: ```bash job_id=$(curl -sS -X POST -H "Accept: application/vnd.gravitino.v1+json" \ @@ -238,8 +238,8 @@ Counts concern the selected spec. A successful rewrite does not guarantee fewer -d '{ "jobTemplateName": "builtin-iceberg-rewrite-manifests", "jobConf": { - "catalog": "rest_catalog", - "table": "db.t1", + "catalog_name": "rest_catalog", + "table_identifier": "db.t1", "spark_master": "local[2]", "spark_executor_instances": "1", "spark_executor_cores": "1", @@ -283,8 +283,8 @@ Pair the job with `builtin-iceberg-expire-snapshots`. The rewrite creates a snap | Input | Behavior | | ----------------------- | --------------------------------------------------------------------------- | -| `use-caching` | Accept `true` or `false`, case-insensitively; reject other supplied values | -| `spec-id` | Parse a non-negative 32-bit integer; reject malformed or overflowing values | +| `use_caching` | Accept `true` or `false`, case-insensitively; reject other supplied values | +| `spec_id` | Parse a non-negative 32-bit integer; reject malformed or overflowing values | | Unknown spec ID | Iceberg rejects IDs absent from table metadata | | Omitted optional values | Omit them from the SQL call and preserve runtime defaults | | Catalog and table | Use shared identifier and string escaping when building SQL |