roryqi commented on code in PR #11206: URL: https://github.com/apache/gravitino/pull/11206#discussion_r3429965922
########## design-docs/iceberg-expire-snapshots-maintenance-job.md: ########## @@ -0,0 +1,476 @@ +<!-- + 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 Expire Snapshots Maintenance Job + +| Field | Value | +| ------- | -------------------------------------------------------------------- | +| Status | Draft | +| Authors | @laserninja | +| Created | 2026-05-26 | +| Issue | [#11194](https://github.com/apache/gravitino/issues/11194) | +| Module | `api`, `maintenance/jobs`, `maintenance/optimizer` | + +--- + +## 1. Background + +Without periodic snapshot expiration, Iceberg table metadata grows +indefinitely — accumulating snapshot JSON files and manifest lists that +slow down table operations and waste storage. The existing built-in +maintenance jobs (`builtin-iceberg-rewrite-data-files` for data compaction +and `builtin-iceberg-update-stats` for metrics) address data file +optimization but do not cover metadata cleanup. + +PR [#10500](https://github.com/apache/gravitino/pull/10500) added +Trino-side delegation for `expire_snapshots` as a procedure, but there is +no server-side built-in job that can be triggered automatically via the +Table Maintenance Service (Optimizer) policies. + +This design proposes adding full end-to-end support for Iceberg snapshot +expiration: from policy definition through strategy evaluation to Spark +job execution. + +--- + +## 2. Goals + +1. Add a new built-in policy type `system_iceberg_snapshot_expiration` for + declarative snapshot expiration configuration. +2. Add a strategy handler that evaluates when snapshot expiration should + run based on table statistics (snapshot count, age). +3. Add a job adapter that converts strategy evaluation results into job + configurations. +4. Add the Spark job that executes Iceberg's `expire_snapshots` procedure. +5. Ensure the full flow works end-to-end: policy → strategy → job + submission → Spark execution. + +--- + +## 3. Non-Goals + +- Orphan file removal (separate Iceberg procedure, separate issue). +- Automatic policy creation — users must explicitly create and attach + policies. +- Changes to the Optimizer scheduling framework itself. + +--- + +## 4. Existing Architecture Overview + +The Gravitino maintenance module follows a layered architecture for +automated table maintenance. The existing Iceberg compaction flow +establishes the pattern: + +``` +Policy Creation (REST API) + ↓ +GravitinoStrategyProvider (loads policies as strategies) + ↓ +CompactionStrategyHandler (evaluates trigger / score expressions) + ↓ +CompactionJobContext → GravitinoCompactionJobAdapter (converts to job config) + ↓ +GravitinoJobSubmitter (submits job via REST) + ↓ +IcebergRewriteDataFilesJob (Spark execution) +``` + +### 4.1 Layer Summary + +| Layer | Compaction Components | Purpose | +| ------------ | ---------------------------------------------------------------------------------- | -------------------------------------------------- | +| **Policy** | `Policy.BuiltInType.ICEBERG_COMPACTION`, `IcebergDataCompactionContent` | Define configuration, thresholds, expressions | +| **Strategy** | `CompactionStrategyHandler` extends `BaseExpressionStrategyHandler` | Evaluate trigger conditions, score partitions | +| **Adapter** | `GravitinoCompactionJobAdapter`, `CompactionJobContext` | Convert evaluation result to job configuration | +| **Job** | `IcebergRewriteDataFilesJob`, registered in `BuiltInJobTemplateProvider` | Execute Spark procedure | + +--- + +## 5. Proposed Design + +We add the same four layers for snapshot expiration, following the +compaction pattern. + +### 5.1 Architecture Diagram + +``` +┌──────────────────────────────────────────────────────────────┐ +│ REST API: POST /metalakes/{m}/policies │ +│ type: "system_iceberg_snapshot_expiration" │ +│ content: IcebergSnapshotExpirationContent │ +│ { olderThanDays, retainLast, streamResults } │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ GravitinoStrategyProvider │ +│ Loads policy → GravitinoStrategy │ +│ strategyType: "iceberg-snapshot-expiration" │ +│ jobTemplateName: "builtin-iceberg-expire-snapshots" │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ SnapshotExpirationStrategyHandler │ +│ extends BaseExpressionStrategyHandler │ +│ dataRequirements: {TABLE_METADATA, TABLE_STATISTICS} │ +│ Evaluates: snapshot count ≥ threshold │ +│ Returns: StrategyEvaluation with score + context │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ SnapshotExpirationJobContext → JobAdapter │ +│ Extracts: older_than, retain_last, stream_results │ +│ Builds: job config map for template substitution │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ GravitinoJobSubmitter │ +│ Template: "builtin-iceberg-expire-snapshots" │ +│ Submits via REST: POST /metalakes/{m}/jobs │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ IcebergExpireSnapshotsJob (Spark) │ +│ CALL catalog.system.expire_snapshots( │ +│ table => '…', older_than => TIMESTAMP '…', │ +│ retain_last => N, stream_results => bool) │ +└──────────────────────────────────────────────────────────────┘ +``` + +--- + +### 5.2 Layer 1 — Policy Definition (`api/`) + +#### 5.2.1 New Policy Type + +Add `ICEBERG_SNAPSHOT_EXPIRATION` to `Policy.BuiltInType`: + +```java +// api/src/main/java/org/apache/gravitino/policy/Policy.java +enum BuiltInType { + ICEBERG_COMPACTION("system_iceberg_compaction", + IcebergDataCompactionContent.class), + ICEBERG_SNAPSHOT_EXPIRATION("system_iceberg_snapshot_expiration", + IcebergSnapshotExpirationContent.class), // NEW + CUSTOM("custom", CustomContent.class); +} +``` + +#### 5.2.2 New Policy Content Class + +Create `IcebergSnapshotExpirationContent` following the +`IcebergDataCompactionContent` pattern: + +```java +// NEW: api/src/main/java/org/apache/gravitino/policy/ +// IcebergSnapshotExpirationContent.java +public class IcebergSnapshotExpirationContent implements PolicyContent { + + // Strategy metadata + public static final String STRATEGY_TYPE_VALUE = + "iceberg-snapshot-expiration"; + public static final String JOB_TEMPLATE_NAME_VALUE = + "builtin-iceberg-expire-snapshots"; + + // Configurable fields + private final long olderThanDays; // default: 5 + private final long retainLast; // default: 1 + private final boolean streamResults; // default: false + private final long minSnapshotCount; // trigger threshold, default: 10 + + // Trigger / score expressions + public static final String TRIGGER_EXPR = + "custom-snapshot-count >= minSnapshotCount"; + public static final String SCORE_EXPR = + "custom-snapshot-count"; + + // Defaults + public static final long DEFAULT_OLDER_THAN_DAYS = 5; + public static final long DEFAULT_RETAIN_LAST = 1; + public static final boolean DEFAULT_STREAM_RESULTS = false; + public static final long DEFAULT_MIN_SNAPSHOT_COUNT = 10; +} +``` + +#### 5.2.3 Policy Content Fields + +| Field | Type | Default | Description | +| ----------------- | --------- | ------- | ----------------------------------------------------------- | +| `olderThanDays` | `long` | 5 | Expire snapshots older than this many days | +| `retainLast` | `long` | 1 | Minimum number of snapshots to always retain | Review Comment: Maybe the user wanna to specifiy `olderThanDays` or `retainLast` only. -- 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]
