lasdf1234 commented on code in PR #12942: URL: https://github.com/apache/gravitino/pull/12942#discussion_r4024430329
########## design-docs/iceberg-rewrite-manifests-job.md: ########## @@ -0,0 +1,346 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. +--> + +# 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 the original work by @ibrahimErbilen in [#11216](https://github.com/apache/gravitino/pull/11216). + +## 1. 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. + +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. + +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. + +## 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. +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. + +## 3. Non-Goals + +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. + +## 4. Existing Architecture Overview + +### 4.1 Layer Summary + +| 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 | + +## 5. Proposed Design + +### 5.1 Architecture Diagram + +```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. + +### 5.2 Layer 1 - Policy Definition (`api/`, follow-up) + +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) + +#### 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 | + +Example logical contents of the two rows for one table: + +```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 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) + +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/`) + +#### 5.5.1 Job Class and Registration + +| 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()` | + +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. + +#### 5.5.2 Parameters + +Use `POST /api/metalakes/{metalake}/jobs/runs` with the existing `jobTemplateName` and `jobConf` fields. + +| `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 | + Review Comment: According to the latest regulations, jobConf key and CLI argument should be same. You can see this PR:https://github.com/apache/gravitino/pull/13134 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
