lasdf1234 commented on code in PR #11700: URL: https://github.com/apache/gravitino/pull/11700#discussion_r4005451885
########## design-docs/iceberg-remove-orphan-files-maintenance-job.md: ########## @@ -0,0 +1,644 @@ +<!-- + 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 Remove Orphan Files Maintenance Job + +| Field | Value | +| ------- | ------------------------------------------------------------ | +| Status | Draft | +| Authors | @laserninja | +| Created | 2026-06-16 | +| Issue | [#11195](https://github.com/apache/gravitino/issues/11195) | +| Module | `api`, `maintenance/jobs`, `maintenance/optimizer` | + +--- + +## 1. Background + +Orphan files accumulate in Iceberg table storage locations from failed writes, +incomplete transactions, schema evolution, or concurrent operations. These files +are no longer referenced by any table snapshot but remain on disk, wasting +significant storage — especially in high-write-volume environments. + +The existing built-in maintenance jobs (`builtin-iceberg-rewrite-data-files` for +data compaction, `builtin-iceberg-update-stats` for metrics, and +`builtin-iceberg-expire-snapshots` for metadata cleanup) address data file +optimization and snapshot lifecycle but do not cover orphan file removal. + +PR [#10500](https://github.com/apache/gravitino/pull/10500) added Trino-side +delegation for `remove_orphan_files` 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 orphan file +removal: from policy definition through strategy evaluation to Spark job +execution. + +--- + +## 2. Goals + +1. Add a new built-in policy type `system_iceberg_orphan_file_removal` for + declarative orphan file cleanup configuration. +2. Add a strategy handler that evaluates when orphan file removal should run + based on time since last cleanup or table statistics. +3. Add a job adapter that converts strategy evaluation results into job + configurations. +4. Add the Spark job that executes Iceberg's `remove_orphan_files` procedure. +5. Ensure the full flow works end-to-end: policy → strategy → job + submission → Spark execution. + +--- + +## 3. Non-Goals + +- Snapshot expiration (separate Iceberg procedure, separate issue [#11194](https://github.com/apache/gravitino/issues/11194)). +- Automatic policy creation — users must explicitly create and attach + policies. +- Changes to the Optimizer scheduling framework itself. +- Custom file-level filtering beyond what Iceberg's procedure supports. + +--- + +## 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 orphan file removal, following the compaction +pattern. + +### 5.1 Architecture Diagram + +``` +┌──────────────────────────────────────────────────────────────┐ +│ REST API: POST /metalakes/{m}/policies │ +│ type: "system_iceberg_orphan_file_removal" │ +│ content: IcebergOrphanFileRemovalContent │ +│ { olderThanDays, location, dryRun } │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ GravitinoStrategyProvider │ +│ Loads policy → GravitinoStrategy │ +│ strategyType: "iceberg-orphan-file-removal" │ +│ jobTemplateName: "builtin-iceberg-remove-orphan-files" │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ OrphanFileRemovalStrategyHandler │ +│ extends BaseExpressionStrategyHandler │ +│ dataRequirements: {TABLE_METADATA, TABLE_STATISTICS} │ +│ Evaluates: custom-last-orphan-cleanup-time (statistic_meta) │ +│ Returns: StrategyEvaluation with score + context │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ OrphanFileRemovalJobContext → JobAdapter │ +│ Extracts: older_than, location, dry_run │ +│ Builds: job config map for template substitution │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ GravitinoJobSubmitter │ +│ Template: "builtin-iceberg-remove-orphan-files" │ +│ Submits via REST: POST /metalakes/{m}/jobs │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ IcebergRemoveOrphanFilesJob (Spark) │ +│ CALL catalog.system.remove_orphan_files( │ +│ table => '…', older_than => TIMESTAMP '…', │ +│ location => '…', dry_run => bool) │ +└──────────────────────────────────────────────────────────────┘ +``` + +--- + +### 5.2 Layer 1 — Policy Definition (`api/`) + +#### 5.2.1 New Policy Type + +Add `ICEBERG_ORPHAN_FILE_REMOVAL` to `Policy.BuiltInType`: + +```java +// api/src/main/java/org/apache/gravitino/policy/Policy.java +enum BuiltInType { + ICEBERG_COMPACTION("system_iceberg_compaction", + IcebergDataCompactionContent.class), + ICEBERG_ORPHAN_FILE_REMOVAL("system_iceberg_orphan_file_removal", + IcebergOrphanFileRemovalContent.class), // NEW + CUSTOM("custom", CustomContent.class); +} +``` + +#### 5.2.2 New Policy Content Class + +Create `IcebergOrphanFileRemovalContent` following the +`IcebergDataCompactionContent` pattern: + +```java +// NEW: api/src/main/java/org/apache/gravitino/policy/ +// IcebergOrphanFileRemovalContent.java +public class IcebergOrphanFileRemovalContent implements PolicyContent { + + // Strategy metadata + public static final String STRATEGY_TYPE_VALUE = + "iceberg-orphan-file-removal"; + public static final String JOB_TEMPLATE_NAME_VALUE = + "builtin-iceberg-remove-orphan-files"; + + // Configurable fields + private final long olderThanDays; // default: 3 + private final String location; // default: null (table location) + private final boolean dryRun; // default: false + + // Trigger / score expressions. + // The interval threshold comes from the uniform minimum-interval + // mechanism shared by all system built-in policies, not this content. + public static final String TRIGGER_EXPR = + "custom-days-since-last-orphan-cleanup >= minIntervalDays"; + public static final String SCORE_EXPR = + "custom-days-since-last-orphan-cleanup"; + + // Defaults + public static final long DEFAULT_OLDER_THAN_DAYS = 3; + public static final boolean DEFAULT_DRY_RUN = false; +} +``` + +#### 5.2.3 Policy Content Fields + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `olderThanDays` | `long` | 3 | Only remove orphan files older than this many days. See [5.2.4](#524-why-olderthandays-defaults-to-3) for the rationale. | +| `location` | `String` | null | Custom location to scan. When specified, **only** this location is scanned instead of the table's default location. Must be validated against the table's own location — see [Section 6.1](#61-location-validation). If null, the table's registered storage location is used. | +| `dryRun` | `boolean` | false | Preview-only mode — list orphan files without deleting | + +> **Note:** The minimum interval between runs is intentionally **not** a field +> of this policy. A uniform minimum-interval mechanism will be defined across +> all four system built-in policies and is out of scope for this design. + +#### 5.2.4 Why `olderThanDays` Defaults to 3 + +The 3-day default mirrors Iceberg's own default for +`remove_orphan_files` and exists to protect **in-flight writes**. + +Orphan file detection compares files on storage against files referenced by +table metadata. A file written by a transaction that has not yet committed +looks identical to an orphan file. If such a file is deleted, the in-flight +commit fails or, worse, produces a table that references a missing file. +A 3-day window is long enough to cover: + +- Long-running Spark / Flink write jobs that stage files before committing +- Retried or paused jobs that resume hours or days later +- Clock skew between the storage system and the job runtime + +**To remove all orphan files regardless of age**, set `olderThanDays` to `0`. +The adapter then passes the current timestamp as `older_than`, so every +unreferenced file is eligible for deletion. + +**This is unsafe while any writer is active** and should only be used when +all writes to the table are known to be stopped — for example, during a +maintenance window or when reclaiming storage from a decommissioned table. +Run with `dryRun: true` first to review the file list. + +#### 5.2.5 Example Policy Creation + +```json +POST /metalakes/default/policies +{ + "name": "remove_orphans_weekly", + "type": "system_iceberg_orphan_file_removal", + "comment": "Remove orphan files older than 3 days", + "enabled": true, + "content": { + "olderThanDays": 3, + "dryRun": false + } +} +``` + +--- + +### 5.3 Layer 2 — Strategy Handler (`maintenance/optimizer/`) + +#### 5.3.1 Strategy Handler + +```java +// NEW: maintenance/optimizer/src/main/java/…/handler/orphan/ +// OrphanFileRemovalStrategyHandler.java +public class OrphanFileRemovalStrategyHandler + extends BaseExpressionStrategyHandler { + + public static final String NAME = "iceberg-orphan-file-removal"; + + @Override + public String strategyType() { + return NAME; + } + + @Override + public Set<DataRequirement> dataRequirements() { + return ImmutableSet.of( + DataRequirement.TABLE_METADATA, + DataRequirement.TABLE_STATISTICS); + // TABLE_STATISTICS supplies custom-last-orphan-cleanup-time, + // read from statistic_meta the same way compaction reads its metrics + } + + @Override + protected JobExecutionContext buildJobExecutionContext( + NameIdentifier nameIdentifier, + Strategy strategy, + Table table, + List<PartitionPath> partitions, + Map<String, String> jobOptions) { + return new OrphanFileRemovalJobContext( + nameIdentifier, jobOptions, strategy.jobTemplateName()); + } +} +``` + +**Key design decision:** Orphan file removal operates at the **table level**, +not partition level. Unlike compaction, which scores and selects individual +partitions, `remove_orphan_files` scans the entire table's storage location. +Therefore: + +- `dataRequirements()` excludes `PARTITION_STATISTICS`. +- No partition scoring / selection logic is needed. + +**Trigger modes:** Gravitino supports two trigger mechanisms, and this +design does not limit users to one: + +1. **Event trigger** — The strategy handler evaluates table metadata + (e.g., snapshot count changes, write events) and triggers cleanup when + conditions are met. +2. **Time trigger** — The Optimizer's scheduling framework can invoke the + strategy handler periodically, and the handler decides whether cleanup + is needed based on the last cleanup time. + +Both modes use the same strategy handler; the difference is in how +often the handler is invoked. + +#### 5.3.2 Tracking the Last Cleanup Time + +The handler needs to know when orphan file removal last ran for a table. +This is stored as a **table statistic** in `statistic_meta`, matching how +compaction persists its metrics: + +| Statistic Name | Type | Description | +| --- | --- | --- | +| `custom-last-orphan-cleanup-time` | `Long` | Epoch millis of the last successful orphan file removal | +| `custom-days-since-last-orphan-cleanup` | `Long` | Derived value used by the trigger expression | + +The statistic is written back after a successful (non-dry-run) job +completion, following the same statistics update path that +`IcebergUpdateStatsAndMetricsJob` uses. Tables with no recorded cleanup +time are treated as never cleaned and are eligible on the first evaluation. + +#### 5.3.3 Job Execution Context + +```java +// NEW: maintenance/optimizer/src/main/java/…/handler/orphan/ +// OrphanFileRemovalJobContext.java +public class OrphanFileRemovalJobContext implements JobExecutionContext { + private final NameIdentifier nameIdentifier; + private final Map<String, String> jobOptions; + private final String jobTemplateName; + + // jobOptions keys: + // older_than, location, dry_run, + // catalog_name, table_identifier +} +``` + +#### 5.3.4 Handler Registration + +The strategy handler type `"iceberg-orphan-file-removal"` must be registered so +that the `Recommender` can instantiate it when it encounters a policy with that +strategy type. This follows the existing pattern where handler classes are +looked up by strategy type name. + +--- + +### 5.4 Layer 3 — Job Adapter (`maintenance/optimizer/`) + +#### 5.4.1 Job Adapter + +```java +// NEW: maintenance/optimizer/src/main/java/…/job/ +// GravitinoOrphanFileRemovalJobAdapter.java +public class GravitinoOrphanFileRemovalJobAdapter + implements GravitinoJobAdapter { + + @Override + public Map<String, String> jobConfig(JobExecutionContext context) { + OrphanFileRemovalJobContext ctx = + (OrphanFileRemovalJobContext) context; + Map<String, String> config = new HashMap<>(); + config.put("catalog_name", + ctx.nameIdentifier().namespace()[0]); + config.put("table_identifier", + ctx.nameIdentifier().namespace()[1] + + "." + ctx.nameIdentifier().name()); + + // Convert olderThanDays → absolute timestamp + // olderThanDays == 0 means "now", i.e. remove all orphan files + Map<String, String> opts = ctx.jobOptions(); + long days = Long.parseLong( + opts.getOrDefault("olderThanDays", "3")); + String ts = Instant.now() + .minus(Duration.ofDays(days)) + .toString() + .replace("T", " ") + .substring(0, 19); // "yyyy-MM-dd HH:mm:ss" + config.put("older_than", ts); + + // Validated against the table's own location before submission + if (opts.containsKey("location")) { + String location = opts.get("location"); + validateLocation(ctx.nameIdentifier(), location); + config.put("location", location); + } + + config.put("dry_run", + opts.getOrDefault("dryRun", "false")); + return config; + } +} +``` + +#### 5.4.2 Register Adapter + +```java +// UPDATE: GravitinoJobSubmitter.java +private static final Map<String, Class<? extends GravitinoJobAdapter>> + jobAdapters = ImmutableMap.of( + "builtin-iceberg-rewrite-data-files", + GravitinoCompactionJobAdapter.class, + "builtin-iceberg-remove-orphan-files", + GravitinoOrphanFileRemovalJobAdapter.class // NEW +); +``` + +--- + +### 5.5 Layer 4 — Spark Job (`maintenance/jobs/`) + +#### 5.5.1 Job Class + +Create `IcebergRemoveOrphanFilesJob` following the same pattern as +`IcebergRewriteDataFilesJob` and `IcebergExpireSnapshotsJob`: + +- **Template name:** `builtin-iceberg-remove-orphan-files` +- **Version:** `v1` +- **Parameters:** `--catalog`, `--table`, `--older-than`, `--location`, + `--dry-run`, `--spark-conf` + +#### 5.5.2 Procedure Call + +```sql +CALL catalog.system.remove_orphan_files( + table => 'db.table_name', + older_than => TIMESTAMP '2026-06-13 00:00:00', + location => 's3://bucket/path/', + dry_run => true +) +``` + +**Parameters (from Iceberg `remove_orphan_files` procedure):** + +| Parameter | Type | Required | Description | +| ----------- | ----------- | -------- | ------------------------------------------------------------- | +| `table` | `string` | Yes | Fully qualified table name | +| `older_than`| `timestamp` | No | Only remove files older than this timestamp (default: 3 days) | +| `location` | `string` | No | Custom directory to scan for orphans (replaces table location when set; must be within the table's own location) | +| `dry_run` | `boolean` | No | If true, list orphan files without deleting them | + +#### 5.5.3 Output + +The procedure returns a result set with one column: + +| Column | Type | Description | +| ------------------- | -------- | ---------------------------------- | +| `orphan_file_location` | `string` | Path of each orphan file found/removed | + +The job will log the count of orphan files removed (or found in dry-run mode). + +#### 5.5.4 Security + +- SQL injection prevention via `escapeSqlString()` / `escapeSqlIdentifier()` + (same utilities as `IcebergExpireSnapshotsJob`) +- Input validation for `dry_run` (must be `true` or `false`) +- `location` validated against the table's own storage location — see + [Section 6.1](#61-location-validation) + +#### 5.5.5 Job Registration + +```java +// UPDATE: BuiltInJobTemplateProvider.java +private static final List<BuiltInJob> BUILT_IN_JOBS = + ImmutableList.of( + new SparkPiJob(), + new IcebergRewriteDataFilesJob(), + new IcebergUpdateStatsAndMetricsJob(), + new IcebergRemoveOrphanFilesJob()); // NEW +``` + +--- + +## 6. Safety Considerations + +Orphan file removal is inherently more dangerous than snapshot expiration or +compaction because it **permanently deletes files**. Several safety mechanisms +are built into the design: + +| Safety Mechanism | Description | +| --- | --- | +| **`older_than` default** | 3-day default ensures files from in-flight writes are not deleted | +| **`dry_run` mode** | Allows previewing which files would be deleted before actual removal | +| **Location validation** | Rejects any `location` outside the table's own storage location — see [6.1](#61-location-validation) | +| **Policy-gated** | Must be explicitly enabled by an administrator via policy creation | +| **Iceberg built-in safety** | The procedure itself only identifies files not referenced by any snapshot | + +### 6.1 Location Validation + +The `location` parameter is the single most dangerous input in this job. +Iceberg's `remove_orphan_files` deletes **every file under the given +location that the target table does not reference**. If a caller passes the +location of a *different* table, that table's data files are all unreferenced +from the target table's point of view, and the procedure deletes them — +silently destroying another table's data. + +The job must therefore validate `location` **before building the SQL**, not +rely on Iceberg to reject it: + +```java +static void validateLocation(NameIdentifier tableIdent, String location) { + String tableLocation = normalize(loadTableLocation(tableIdent)); + String requested = normalize(location); + + Preconditions.checkArgument( + requested.equals(tableLocation) + || requested.startsWith(tableLocation + "/"), + "location '%s' must be within the table's location '%s'", + location, tableLocation); +} +``` + +Validation rules: + +1. Resolve the target table's own storage location from table metadata. +2. Normalize both paths — canonicalize the scheme and authority, collapse + duplicate slashes, resolve `.` / `..` segments, and strip trailing slashes. +3. Reject the request unless the normalized `location` is the table location + itself or a descendant of it. +4. Reject symlinks or paths that resolve outside the table location after + normalization. + +Normalization must happen before the prefix check; comparing raw strings +would let `s3://bucket/db/table/../other_table` pass a naive +`startsWith` test. + +Validation happens in the job adapter (before submission) and again in the +Spark job (before building the procedure call), so an ad-hoc job submission +that bypasses the policy layer is still checked. + +### 6.2 Setting `older_than` Safely + +The `older_than` threshold should be set conservatively. Files from in-flight +writes or concurrent operations may not yet be referenced by a committed +snapshot. A minimum of 3 days is recommended. + +Setting `olderThanDays` to `0` removes all orphan files regardless of age. +This is only safe when no writer is active against the table — see +[Section 5.2.4](#524-why-olderthandays-defaults-to-3). + +--- + +## 7. File Changes Summary + +### 7.1 New Files + +| File | Layer | Description | +| -------------------------------------------------------------------------------- | -------- | ------------------------------------------ | +| `api/…/policy/IcebergOrphanFileRemovalContent.java` | Policy | Policy content with removal configuration | +| `maintenance/optimizer/…/handler/orphan/OrphanFileRemovalStrategyHandler.java` | Strategy | Trigger / score evaluation | +| `maintenance/optimizer/…/handler/orphan/OrphanFileRemovalJobContext.java` | Strategy | Job execution context | +| `maintenance/optimizer/…/job/GravitinoOrphanFileRemovalJobAdapter.java` | Adapter | Context → job config conversion | +| `maintenance/jobs/…/iceberg/IcebergRemoveOrphanFilesJob.java` | Job | Spark job | + +### 7.2 Modified Files + +| File | Change | +| ------------------------------------------------------- | ------------------------------------------------------------- | +| `api/…/policy/Policy.java` | Add `ICEBERG_ORPHAN_FILE_REMOVAL` to `BuiltInType` enum | +| `maintenance/optimizer/…/job/GravitinoJobSubmitter.java` | Register remove-orphan-files adapter in `jobAdapters` map | +| `maintenance/jobs/…/BuiltInJobTemplateProvider.java` | Register `IcebergRemoveOrphanFilesJob` | +| Handler registry | Register `OrphanFileRemovalStrategyHandler` | + Review Comment: Could u align table? ########## design-docs/iceberg-remove-orphan-files-maintenance-job.md: ########## @@ -0,0 +1,644 @@ +<!-- + 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 Remove Orphan Files Maintenance Job + +| Field | Value | +| ------- | ------------------------------------------------------------ | +| Status | Draft | +| Authors | @laserninja | +| Created | 2026-06-16 | +| Issue | [#11195](https://github.com/apache/gravitino/issues/11195) | +| Module | `api`, `maintenance/jobs`, `maintenance/optimizer` | + +--- + +## 1. Background + +Orphan files accumulate in Iceberg table storage locations from failed writes, +incomplete transactions, schema evolution, or concurrent operations. These files +are no longer referenced by any table snapshot but remain on disk, wasting +significant storage — especially in high-write-volume environments. + +The existing built-in maintenance jobs (`builtin-iceberg-rewrite-data-files` for +data compaction, `builtin-iceberg-update-stats` for metrics, and +`builtin-iceberg-expire-snapshots` for metadata cleanup) address data file +optimization and snapshot lifecycle but do not cover orphan file removal. + +PR [#10500](https://github.com/apache/gravitino/pull/10500) added Trino-side +delegation for `remove_orphan_files` 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 orphan file +removal: from policy definition through strategy evaluation to Spark job +execution. + +--- + +## 2. Goals + +1. Add a new built-in policy type `system_iceberg_orphan_file_removal` for + declarative orphan file cleanup configuration. +2. Add a strategy handler that evaluates when orphan file removal should run + based on time since last cleanup or table statistics. +3. Add a job adapter that converts strategy evaluation results into job + configurations. +4. Add the Spark job that executes Iceberg's `remove_orphan_files` procedure. +5. Ensure the full flow works end-to-end: policy → strategy → job + submission → Spark execution. + +--- + +## 3. Non-Goals + +- Snapshot expiration (separate Iceberg procedure, separate issue [#11194](https://github.com/apache/gravitino/issues/11194)). +- Automatic policy creation — users must explicitly create and attach + policies. +- Changes to the Optimizer scheduling framework itself. +- Custom file-level filtering beyond what Iceberg's procedure supports. + +--- + +## 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 orphan file removal, following the compaction +pattern. + +### 5.1 Architecture Diagram + +``` +┌──────────────────────────────────────────────────────────────┐ +│ REST API: POST /metalakes/{m}/policies │ +│ type: "system_iceberg_orphan_file_removal" │ +│ content: IcebergOrphanFileRemovalContent │ +│ { olderThanDays, location, dryRun } │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ GravitinoStrategyProvider │ +│ Loads policy → GravitinoStrategy │ +│ strategyType: "iceberg-orphan-file-removal" │ +│ jobTemplateName: "builtin-iceberg-remove-orphan-files" │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ OrphanFileRemovalStrategyHandler │ +│ extends BaseExpressionStrategyHandler │ +│ dataRequirements: {TABLE_METADATA, TABLE_STATISTICS} │ +│ Evaluates: custom-last-orphan-cleanup-time (statistic_meta) │ +│ Returns: StrategyEvaluation with score + context │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ OrphanFileRemovalJobContext → JobAdapter │ +│ Extracts: older_than, location, dry_run │ +│ Builds: job config map for template substitution │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ GravitinoJobSubmitter │ +│ Template: "builtin-iceberg-remove-orphan-files" │ +│ Submits via REST: POST /metalakes/{m}/jobs │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ IcebergRemoveOrphanFilesJob (Spark) │ +│ CALL catalog.system.remove_orphan_files( │ +│ table => '…', older_than => TIMESTAMP '…', │ +│ location => '…', dry_run => bool) │ +└──────────────────────────────────────────────────────────────┘ +``` + +--- + +### 5.2 Layer 1 — Policy Definition (`api/`) + +#### 5.2.1 New Policy Type + +Add `ICEBERG_ORPHAN_FILE_REMOVAL` to `Policy.BuiltInType`: + +```java +// api/src/main/java/org/apache/gravitino/policy/Policy.java +enum BuiltInType { + ICEBERG_COMPACTION("system_iceberg_compaction", + IcebergDataCompactionContent.class), + ICEBERG_ORPHAN_FILE_REMOVAL("system_iceberg_orphan_file_removal", + IcebergOrphanFileRemovalContent.class), // NEW + CUSTOM("custom", CustomContent.class); +} +``` + +#### 5.2.2 New Policy Content Class + +Create `IcebergOrphanFileRemovalContent` following the +`IcebergDataCompactionContent` pattern: + +```java +// NEW: api/src/main/java/org/apache/gravitino/policy/ +// IcebergOrphanFileRemovalContent.java +public class IcebergOrphanFileRemovalContent implements PolicyContent { + + // Strategy metadata + public static final String STRATEGY_TYPE_VALUE = + "iceberg-orphan-file-removal"; + public static final String JOB_TEMPLATE_NAME_VALUE = + "builtin-iceberg-remove-orphan-files"; + + // Configurable fields + private final long olderThanDays; // default: 3 + private final String location; // default: null (table location) + private final boolean dryRun; // default: false + + // Trigger / score expressions. + // The interval threshold comes from the uniform minimum-interval + // mechanism shared by all system built-in policies, not this content. + public static final String TRIGGER_EXPR = + "custom-days-since-last-orphan-cleanup >= minIntervalDays"; + public static final String SCORE_EXPR = + "custom-days-since-last-orphan-cleanup"; + + // Defaults + public static final long DEFAULT_OLDER_THAN_DAYS = 3; + public static final boolean DEFAULT_DRY_RUN = false; +} +``` + +#### 5.2.3 Policy Content Fields + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `olderThanDays` | `long` | 3 | Only remove orphan files older than this many days. See [5.2.4](#524-why-olderthandays-defaults-to-3) for the rationale. | +| `location` | `String` | null | Custom location to scan. When specified, **only** this location is scanned instead of the table's default location. Must be validated against the table's own location — see [Section 6.1](#61-location-validation). If null, the table's registered storage location is used. | +| `dryRun` | `boolean` | false | Preview-only mode — list orphan files without deleting | + +> **Note:** The minimum interval between runs is intentionally **not** a field +> of this policy. A uniform minimum-interval mechanism will be defined across +> all four system built-in policies and is out of scope for this design. + +#### 5.2.4 Why `olderThanDays` Defaults to 3 + +The 3-day default mirrors Iceberg's own default for +`remove_orphan_files` and exists to protect **in-flight writes**. + +Orphan file detection compares files on storage against files referenced by +table metadata. A file written by a transaction that has not yet committed +looks identical to an orphan file. If such a file is deleted, the in-flight +commit fails or, worse, produces a table that references a missing file. +A 3-day window is long enough to cover: + +- Long-running Spark / Flink write jobs that stage files before committing +- Retried or paused jobs that resume hours or days later +- Clock skew between the storage system and the job runtime + +**To remove all orphan files regardless of age**, set `olderThanDays` to `0`. +The adapter then passes the current timestamp as `older_than`, so every +unreferenced file is eligible for deletion. + +**This is unsafe while any writer is active** and should only be used when +all writes to the table are known to be stopped — for example, during a +maintenance window or when reclaiming storage from a decommissioned table. +Run with `dryRun: true` first to review the file list. + +#### 5.2.5 Example Policy Creation + +```json +POST /metalakes/default/policies +{ + "name": "remove_orphans_weekly", + "type": "system_iceberg_orphan_file_removal", + "comment": "Remove orphan files older than 3 days", + "enabled": true, + "content": { + "olderThanDays": 3, + "dryRun": false + } +} +``` + +--- + +### 5.3 Layer 2 — Strategy Handler (`maintenance/optimizer/`) + +#### 5.3.1 Strategy Handler + +```java +// NEW: maintenance/optimizer/src/main/java/…/handler/orphan/ +// OrphanFileRemovalStrategyHandler.java +public class OrphanFileRemovalStrategyHandler + extends BaseExpressionStrategyHandler { + + public static final String NAME = "iceberg-orphan-file-removal"; + + @Override + public String strategyType() { + return NAME; + } + + @Override + public Set<DataRequirement> dataRequirements() { + return ImmutableSet.of( + DataRequirement.TABLE_METADATA, + DataRequirement.TABLE_STATISTICS); + // TABLE_STATISTICS supplies custom-last-orphan-cleanup-time, + // read from statistic_meta the same way compaction reads its metrics + } + + @Override + protected JobExecutionContext buildJobExecutionContext( + NameIdentifier nameIdentifier, + Strategy strategy, + Table table, + List<PartitionPath> partitions, + Map<String, String> jobOptions) { + return new OrphanFileRemovalJobContext( + nameIdentifier, jobOptions, strategy.jobTemplateName()); + } +} +``` + +**Key design decision:** Orphan file removal operates at the **table level**, +not partition level. Unlike compaction, which scores and selects individual +partitions, `remove_orphan_files` scans the entire table's storage location. +Therefore: + +- `dataRequirements()` excludes `PARTITION_STATISTICS`. +- No partition scoring / selection logic is needed. + +**Trigger modes:** Gravitino supports two trigger mechanisms, and this +design does not limit users to one: + +1. **Event trigger** — The strategy handler evaluates table metadata + (e.g., snapshot count changes, write events) and triggers cleanup when + conditions are met. +2. **Time trigger** — The Optimizer's scheduling framework can invoke the + strategy handler periodically, and the handler decides whether cleanup + is needed based on the last cleanup time. + +Both modes use the same strategy handler; the difference is in how +often the handler is invoked. + +#### 5.3.2 Tracking the Last Cleanup Time + +The handler needs to know when orphan file removal last ran for a table. +This is stored as a **table statistic** in `statistic_meta`, matching how +compaction persists its metrics: + +| Statistic Name | Type | Description | +| --- | --- | --- | +| `custom-last-orphan-cleanup-time` | `Long` | Epoch millis of the last successful orphan file removal | +| `custom-days-since-last-orphan-cleanup` | `Long` | Derived value used by the trigger expression | + +The statistic is written back after a successful (non-dry-run) job +completion, following the same statistics update path that +`IcebergUpdateStatsAndMetricsJob` uses. Tables with no recorded cleanup +time are treated as never cleaned and are eligible on the first evaluation. + +#### 5.3.3 Job Execution Context + +```java +// NEW: maintenance/optimizer/src/main/java/…/handler/orphan/ +// OrphanFileRemovalJobContext.java +public class OrphanFileRemovalJobContext implements JobExecutionContext { + private final NameIdentifier nameIdentifier; + private final Map<String, String> jobOptions; + private final String jobTemplateName; + + // jobOptions keys: + // older_than, location, dry_run, + // catalog_name, table_identifier +} +``` + +#### 5.3.4 Handler Registration + +The strategy handler type `"iceberg-orphan-file-removal"` must be registered so +that the `Recommender` can instantiate it when it encounters a policy with that +strategy type. This follows the existing pattern where handler classes are +looked up by strategy type name. + +--- + +### 5.4 Layer 3 — Job Adapter (`maintenance/optimizer/`) + +#### 5.4.1 Job Adapter + +```java +// NEW: maintenance/optimizer/src/main/java/…/job/ +// GravitinoOrphanFileRemovalJobAdapter.java +public class GravitinoOrphanFileRemovalJobAdapter + implements GravitinoJobAdapter { + + @Override + public Map<String, String> jobConfig(JobExecutionContext context) { + OrphanFileRemovalJobContext ctx = + (OrphanFileRemovalJobContext) context; + Map<String, String> config = new HashMap<>(); + config.put("catalog_name", + ctx.nameIdentifier().namespace()[0]); + config.put("table_identifier", + ctx.nameIdentifier().namespace()[1] + + "." + ctx.nameIdentifier().name()); + + // Convert olderThanDays → absolute timestamp + // olderThanDays == 0 means "now", i.e. remove all orphan files + Map<String, String> opts = ctx.jobOptions(); + long days = Long.parseLong( + opts.getOrDefault("olderThanDays", "3")); + String ts = Instant.now() + .minus(Duration.ofDays(days)) + .toString() + .replace("T", " ") + .substring(0, 19); // "yyyy-MM-dd HH:mm:ss" + config.put("older_than", ts); + + // Validated against the table's own location before submission + if (opts.containsKey("location")) { + String location = opts.get("location"); + validateLocation(ctx.nameIdentifier(), location); + config.put("location", location); + } + + config.put("dry_run", + opts.getOrDefault("dryRun", "false")); + return config; + } +} +``` + +#### 5.4.2 Register Adapter + +```java +// UPDATE: GravitinoJobSubmitter.java +private static final Map<String, Class<? extends GravitinoJobAdapter>> + jobAdapters = ImmutableMap.of( + "builtin-iceberg-rewrite-data-files", + GravitinoCompactionJobAdapter.class, + "builtin-iceberg-remove-orphan-files", + GravitinoOrphanFileRemovalJobAdapter.class // NEW +); +``` + +--- + +### 5.5 Layer 4 — Spark Job (`maintenance/jobs/`) + +#### 5.5.1 Job Class + +Create `IcebergRemoveOrphanFilesJob` following the same pattern as +`IcebergRewriteDataFilesJob` and `IcebergExpireSnapshotsJob`: + +- **Template name:** `builtin-iceberg-remove-orphan-files` +- **Version:** `v1` +- **Parameters:** `--catalog`, `--table`, `--older-than`, `--location`, + `--dry-run`, `--spark-conf` + +#### 5.5.2 Procedure Call + +```sql +CALL catalog.system.remove_orphan_files( + table => 'db.table_name', + older_than => TIMESTAMP '2026-06-13 00:00:00', + location => 's3://bucket/path/', + dry_run => true +) +``` + +**Parameters (from Iceberg `remove_orphan_files` procedure):** + +| Parameter | Type | Required | Description | +| ----------- | ----------- | -------- | ------------------------------------------------------------- | +| `table` | `string` | Yes | Fully qualified table name | +| `older_than`| `timestamp` | No | Only remove files older than this timestamp (default: 3 days) | +| `location` | `string` | No | Custom directory to scan for orphans (replaces table location when set; must be within the table's own location) | +| `dry_run` | `boolean` | No | If true, list orphan files without deleting them | + +#### 5.5.3 Output + +The procedure returns a result set with one column: + +| Column | Type | Description | +| ------------------- | -------- | ---------------------------------- | +| `orphan_file_location` | `string` | Path of each orphan file found/removed | + +The job will log the count of orphan files removed (or found in dry-run mode). + +#### 5.5.4 Security + +- SQL injection prevention via `escapeSqlString()` / `escapeSqlIdentifier()` + (same utilities as `IcebergExpireSnapshotsJob`) +- Input validation for `dry_run` (must be `true` or `false`) +- `location` validated against the table's own storage location — see + [Section 6.1](#61-location-validation) + +#### 5.5.5 Job Registration + +```java +// UPDATE: BuiltInJobTemplateProvider.java +private static final List<BuiltInJob> BUILT_IN_JOBS = + ImmutableList.of( + new SparkPiJob(), + new IcebergRewriteDataFilesJob(), + new IcebergUpdateStatsAndMetricsJob(), + new IcebergRemoveOrphanFilesJob()); // NEW +``` + +--- + +## 6. Safety Considerations + +Orphan file removal is inherently more dangerous than snapshot expiration or +compaction because it **permanently deletes files**. Several safety mechanisms +are built into the design: + +| Safety Mechanism | Description | +| --- | --- | +| **`older_than` default** | 3-day default ensures files from in-flight writes are not deleted | +| **`dry_run` mode** | Allows previewing which files would be deleted before actual removal | +| **Location validation** | Rejects any `location` outside the table's own storage location — see [6.1](#61-location-validation) | +| **Policy-gated** | Must be explicitly enabled by an administrator via policy creation | +| **Iceberg built-in safety** | The procedure itself only identifies files not referenced by any snapshot | + +### 6.1 Location Validation + +The `location` parameter is the single most dangerous input in this job. +Iceberg's `remove_orphan_files` deletes **every file under the given +location that the target table does not reference**. If a caller passes the +location of a *different* table, that table's data files are all unreferenced +from the target table's point of view, and the procedure deletes them — +silently destroying another table's data. + +The job must therefore validate `location` **before building the SQL**, not +rely on Iceberg to reject it: + +```java +static void validateLocation(NameIdentifier tableIdent, String location) { + String tableLocation = normalize(loadTableLocation(tableIdent)); + String requested = normalize(location); + + Preconditions.checkArgument( + requested.equals(tableLocation) + || requested.startsWith(tableLocation + "/"), + "location '%s' must be within the table's location '%s'", + location, tableLocation); +} +``` + +Validation rules: + +1. Resolve the target table's own storage location from table metadata. +2. Normalize both paths — canonicalize the scheme and authority, collapse + duplicate slashes, resolve `.` / `..` segments, and strip trailing slashes. +3. Reject the request unless the normalized `location` is the table location + itself or a descendant of it. +4. Reject symlinks or paths that resolve outside the table location after + normalization. + +Normalization must happen before the prefix check; comparing raw strings +would let `s3://bucket/db/table/../other_table` pass a naive +`startsWith` test. + +Validation happens in the job adapter (before submission) and again in the +Spark job (before building the procedure call), so an ad-hoc job submission +that bypasses the policy layer is still checked. + +### 6.2 Setting `older_than` Safely + +The `older_than` threshold should be set conservatively. Files from in-flight +writes or concurrent operations may not yet be referenced by a committed +snapshot. A minimum of 3 days is recommended. + +Setting `olderThanDays` to `0` removes all orphan files regardless of age. +This is only safe when no writer is active against the table — see +[Section 5.2.4](#524-why-olderthandays-defaults-to-3). + +--- + +## 7. File Changes Summary + +### 7.1 New Files + +| File | Layer | Description | +| -------------------------------------------------------------------------------- | -------- | ------------------------------------------ | +| `api/…/policy/IcebergOrphanFileRemovalContent.java` | Policy | Policy content with removal configuration | +| `maintenance/optimizer/…/handler/orphan/OrphanFileRemovalStrategyHandler.java` | Strategy | Trigger / score evaluation | +| `maintenance/optimizer/…/handler/orphan/OrphanFileRemovalJobContext.java` | Strategy | Job execution context | +| `maintenance/optimizer/…/job/GravitinoOrphanFileRemovalJobAdapter.java` | Adapter | Context → job config conversion | +| `maintenance/jobs/…/iceberg/IcebergRemoveOrphanFilesJob.java` | Job | Spark job | + +### 7.2 Modified Files + +| File | Change | +| ------------------------------------------------------- | ------------------------------------------------------------- | +| `api/…/policy/Policy.java` | Add `ICEBERG_ORPHAN_FILE_REMOVAL` to `BuiltInType` enum | +| `maintenance/optimizer/…/job/GravitinoJobSubmitter.java` | Register remove-orphan-files adapter in `jobAdapters` map | +| `maintenance/jobs/…/BuiltInJobTemplateProvider.java` | Register `IcebergRemoveOrphanFilesJob` | +| Handler registry | Register `OrphanFileRemovalStrategyHandler` | + +### 7.3 Test Files + +| File | Description | +| ------------------------------------------------------- | ------------------------------------- | +| `TestIcebergOrphanFileRemovalContent.java` | Policy content unit tests | +| `TestOrphanFileRemovalStrategyHandler.java` | Strategy handler unit tests | +| `TestGravitinoOrphanFileRemovalJobAdapter.java` | Job adapter unit tests | +| `TestIcebergRemoveOrphanFilesJob.java` | Spark job unit tests | + Review Comment: Could u align table? ########## design-docs/iceberg-remove-orphan-files-maintenance-job.md: ########## @@ -0,0 +1,644 @@ +<!-- + 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 Remove Orphan Files Maintenance Job + +| Field | Value | +| ------- | ------------------------------------------------------------ | +| Status | Draft | +| Authors | @laserninja | +| Created | 2026-06-16 | +| Issue | [#11195](https://github.com/apache/gravitino/issues/11195) | +| Module | `api`, `maintenance/jobs`, `maintenance/optimizer` | + +--- + +## 1. Background + +Orphan files accumulate in Iceberg table storage locations from failed writes, +incomplete transactions, schema evolution, or concurrent operations. These files +are no longer referenced by any table snapshot but remain on disk, wasting +significant storage — especially in high-write-volume environments. + +The existing built-in maintenance jobs (`builtin-iceberg-rewrite-data-files` for +data compaction, `builtin-iceberg-update-stats` for metrics, and +`builtin-iceberg-expire-snapshots` for metadata cleanup) address data file +optimization and snapshot lifecycle but do not cover orphan file removal. + +PR [#10500](https://github.com/apache/gravitino/pull/10500) added Trino-side +delegation for `remove_orphan_files` 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 orphan file +removal: from policy definition through strategy evaluation to Spark job +execution. + +--- + +## 2. Goals + +1. Add a new built-in policy type `system_iceberg_orphan_file_removal` for + declarative orphan file cleanup configuration. +2. Add a strategy handler that evaluates when orphan file removal should run + based on time since last cleanup or table statistics. +3. Add a job adapter that converts strategy evaluation results into job + configurations. +4. Add the Spark job that executes Iceberg's `remove_orphan_files` procedure. +5. Ensure the full flow works end-to-end: policy → strategy → job + submission → Spark execution. + +--- + +## 3. Non-Goals + +- Snapshot expiration (separate Iceberg procedure, separate issue [#11194](https://github.com/apache/gravitino/issues/11194)). +- Automatic policy creation — users must explicitly create and attach + policies. +- Changes to the Optimizer scheduling framework itself. +- Custom file-level filtering beyond what Iceberg's procedure supports. + +--- + +## 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 orphan file removal, following the compaction +pattern. + +### 5.1 Architecture Diagram + +``` +┌──────────────────────────────────────────────────────────────┐ +│ REST API: POST /metalakes/{m}/policies │ +│ type: "system_iceberg_orphan_file_removal" │ +│ content: IcebergOrphanFileRemovalContent │ +│ { olderThanDays, location, dryRun } │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ GravitinoStrategyProvider │ +│ Loads policy → GravitinoStrategy │ +│ strategyType: "iceberg-orphan-file-removal" │ +│ jobTemplateName: "builtin-iceberg-remove-orphan-files" │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ OrphanFileRemovalStrategyHandler │ +│ extends BaseExpressionStrategyHandler │ +│ dataRequirements: {TABLE_METADATA, TABLE_STATISTICS} │ +│ Evaluates: custom-last-orphan-cleanup-time (statistic_meta) │ +│ Returns: StrategyEvaluation with score + context │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ OrphanFileRemovalJobContext → JobAdapter │ +│ Extracts: older_than, location, dry_run │ +│ Builds: job config map for template substitution │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ GravitinoJobSubmitter │ +│ Template: "builtin-iceberg-remove-orphan-files" │ +│ Submits via REST: POST /metalakes/{m}/jobs │ +└──────────────────────────┬───────────────────────────────────┘ + ↓ +┌──────────────────────────────────────────────────────────────┐ +│ IcebergRemoveOrphanFilesJob (Spark) │ +│ CALL catalog.system.remove_orphan_files( │ +│ table => '…', older_than => TIMESTAMP '…', │ +│ location => '…', dry_run => bool) │ +└──────────────────────────────────────────────────────────────┘ +``` + +--- + +### 5.2 Layer 1 — Policy Definition (`api/`) + +#### 5.2.1 New Policy Type + +Add `ICEBERG_ORPHAN_FILE_REMOVAL` to `Policy.BuiltInType`: + +```java +// api/src/main/java/org/apache/gravitino/policy/Policy.java +enum BuiltInType { + ICEBERG_COMPACTION("system_iceberg_compaction", + IcebergDataCompactionContent.class), + ICEBERG_ORPHAN_FILE_REMOVAL("system_iceberg_orphan_file_removal", + IcebergOrphanFileRemovalContent.class), // NEW + CUSTOM("custom", CustomContent.class); +} +``` + +#### 5.2.2 New Policy Content Class + +Create `IcebergOrphanFileRemovalContent` following the +`IcebergDataCompactionContent` pattern: + +```java +// NEW: api/src/main/java/org/apache/gravitino/policy/ +// IcebergOrphanFileRemovalContent.java +public class IcebergOrphanFileRemovalContent implements PolicyContent { + + // Strategy metadata + public static final String STRATEGY_TYPE_VALUE = + "iceberg-orphan-file-removal"; + public static final String JOB_TEMPLATE_NAME_VALUE = + "builtin-iceberg-remove-orphan-files"; + + // Configurable fields + private final long olderThanDays; // default: 3 + private final String location; // default: null (table location) + private final boolean dryRun; // default: false + + // Trigger / score expressions. + // The interval threshold comes from the uniform minimum-interval + // mechanism shared by all system built-in policies, not this content. + public static final String TRIGGER_EXPR = + "custom-days-since-last-orphan-cleanup >= minIntervalDays"; + public static final String SCORE_EXPR = + "custom-days-since-last-orphan-cleanup"; + + // Defaults + public static final long DEFAULT_OLDER_THAN_DAYS = 3; + public static final boolean DEFAULT_DRY_RUN = false; +} +``` + +#### 5.2.3 Policy Content Fields + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `olderThanDays` | `long` | 3 | Only remove orphan files older than this many days. See [5.2.4](#524-why-olderthandays-defaults-to-3) for the rationale. | +| `location` | `String` | null | Custom location to scan. When specified, **only** this location is scanned instead of the table's default location. Must be validated against the table's own location — see [Section 6.1](#61-location-validation). If null, the table's registered storage location is used. | +| `dryRun` | `boolean` | false | Preview-only mode — list orphan files without deleting | + +> **Note:** The minimum interval between runs is intentionally **not** a field +> of this policy. A uniform minimum-interval mechanism will be defined across +> all four system built-in policies and is out of scope for this design. + +#### 5.2.4 Why `olderThanDays` Defaults to 3 + +The 3-day default mirrors Iceberg's own default for +`remove_orphan_files` and exists to protect **in-flight writes**. + +Orphan file detection compares files on storage against files referenced by +table metadata. A file written by a transaction that has not yet committed +looks identical to an orphan file. If such a file is deleted, the in-flight +commit fails or, worse, produces a table that references a missing file. +A 3-day window is long enough to cover: + +- Long-running Spark / Flink write jobs that stage files before committing +- Retried or paused jobs that resume hours or days later +- Clock skew between the storage system and the job runtime + +**To remove all orphan files regardless of age**, set `olderThanDays` to `0`. +The adapter then passes the current timestamp as `older_than`, so every +unreferenced file is eligible for deletion. + +**This is unsafe while any writer is active** and should only be used when +all writes to the table are known to be stopped — for example, during a +maintenance window or when reclaiming storage from a decommissioned table. +Run with `dryRun: true` first to review the file list. + +#### 5.2.5 Example Policy Creation + +```json +POST /metalakes/default/policies +{ + "name": "remove_orphans_weekly", + "type": "system_iceberg_orphan_file_removal", + "comment": "Remove orphan files older than 3 days", + "enabled": true, + "content": { + "olderThanDays": 3, + "dryRun": false + } +} +``` + +--- + +### 5.3 Layer 2 — Strategy Handler (`maintenance/optimizer/`) + +#### 5.3.1 Strategy Handler + +```java +// NEW: maintenance/optimizer/src/main/java/…/handler/orphan/ +// OrphanFileRemovalStrategyHandler.java +public class OrphanFileRemovalStrategyHandler + extends BaseExpressionStrategyHandler { + + public static final String NAME = "iceberg-orphan-file-removal"; + + @Override + public String strategyType() { + return NAME; + } + + @Override + public Set<DataRequirement> dataRequirements() { + return ImmutableSet.of( + DataRequirement.TABLE_METADATA, + DataRequirement.TABLE_STATISTICS); + // TABLE_STATISTICS supplies custom-last-orphan-cleanup-time, + // read from statistic_meta the same way compaction reads its metrics + } + + @Override + protected JobExecutionContext buildJobExecutionContext( + NameIdentifier nameIdentifier, + Strategy strategy, + Table table, + List<PartitionPath> partitions, + Map<String, String> jobOptions) { + return new OrphanFileRemovalJobContext( + nameIdentifier, jobOptions, strategy.jobTemplateName()); + } +} +``` + +**Key design decision:** Orphan file removal operates at the **table level**, +not partition level. Unlike compaction, which scores and selects individual +partitions, `remove_orphan_files` scans the entire table's storage location. +Therefore: + +- `dataRequirements()` excludes `PARTITION_STATISTICS`. +- No partition scoring / selection logic is needed. + +**Trigger modes:** Gravitino supports two trigger mechanisms, and this +design does not limit users to one: + +1. **Event trigger** — The strategy handler evaluates table metadata + (e.g., snapshot count changes, write events) and triggers cleanup when + conditions are met. +2. **Time trigger** — The Optimizer's scheduling framework can invoke the + strategy handler periodically, and the handler decides whether cleanup + is needed based on the last cleanup time. + +Both modes use the same strategy handler; the difference is in how +often the handler is invoked. + +#### 5.3.2 Tracking the Last Cleanup Time + +The handler needs to know when orphan file removal last ran for a table. +This is stored as a **table statistic** in `statistic_meta`, matching how +compaction persists its metrics: + +| Statistic Name | Type | Description | +| --- | --- | --- | +| `custom-last-orphan-cleanup-time` | `Long` | Epoch millis of the last successful orphan file removal | +| `custom-days-since-last-orphan-cleanup` | `Long` | Derived value used by the trigger expression | + +The statistic is written back after a successful (non-dry-run) job +completion, following the same statistics update path that +`IcebergUpdateStatsAndMetricsJob` uses. Tables with no recorded cleanup +time are treated as never cleaned and are eligible on the first evaluation. + +#### 5.3.3 Job Execution Context + +```java +// NEW: maintenance/optimizer/src/main/java/…/handler/orphan/ +// OrphanFileRemovalJobContext.java +public class OrphanFileRemovalJobContext implements JobExecutionContext { + private final NameIdentifier nameIdentifier; + private final Map<String, String> jobOptions; + private final String jobTemplateName; + + // jobOptions keys: + // older_than, location, dry_run, + // catalog_name, table_identifier +} +``` + +#### 5.3.4 Handler Registration + +The strategy handler type `"iceberg-orphan-file-removal"` must be registered so +that the `Recommender` can instantiate it when it encounters a policy with that +strategy type. This follows the existing pattern where handler classes are +looked up by strategy type name. + +--- + +### 5.4 Layer 3 — Job Adapter (`maintenance/optimizer/`) + +#### 5.4.1 Job Adapter + +```java +// NEW: maintenance/optimizer/src/main/java/…/job/ +// GravitinoOrphanFileRemovalJobAdapter.java +public class GravitinoOrphanFileRemovalJobAdapter + implements GravitinoJobAdapter { + + @Override + public Map<String, String> jobConfig(JobExecutionContext context) { + OrphanFileRemovalJobContext ctx = + (OrphanFileRemovalJobContext) context; + Map<String, String> config = new HashMap<>(); + config.put("catalog_name", + ctx.nameIdentifier().namespace()[0]); + config.put("table_identifier", + ctx.nameIdentifier().namespace()[1] + + "." + ctx.nameIdentifier().name()); + + // Convert olderThanDays → absolute timestamp + // olderThanDays == 0 means "now", i.e. remove all orphan files + Map<String, String> opts = ctx.jobOptions(); + long days = Long.parseLong( + opts.getOrDefault("olderThanDays", "3")); + String ts = Instant.now() + .minus(Duration.ofDays(days)) + .toString() + .replace("T", " ") + .substring(0, 19); // "yyyy-MM-dd HH:mm:ss" + config.put("older_than", ts); + + // Validated against the table's own location before submission + if (opts.containsKey("location")) { + String location = opts.get("location"); + validateLocation(ctx.nameIdentifier(), location); + config.put("location", location); + } + + config.put("dry_run", + opts.getOrDefault("dryRun", "false")); + return config; + } +} +``` + +#### 5.4.2 Register Adapter + +```java +// UPDATE: GravitinoJobSubmitter.java +private static final Map<String, Class<? extends GravitinoJobAdapter>> + jobAdapters = ImmutableMap.of( + "builtin-iceberg-rewrite-data-files", + GravitinoCompactionJobAdapter.class, + "builtin-iceberg-remove-orphan-files", + GravitinoOrphanFileRemovalJobAdapter.class // NEW +); +``` + +--- + +### 5.5 Layer 4 — Spark Job (`maintenance/jobs/`) + +#### 5.5.1 Job Class + +Create `IcebergRemoveOrphanFilesJob` following the same pattern as +`IcebergRewriteDataFilesJob` and `IcebergExpireSnapshotsJob`: + +- **Template name:** `builtin-iceberg-remove-orphan-files` +- **Version:** `v1` +- **Parameters:** `--catalog`, `--table`, `--older-than`, `--location`, + `--dry-run`, `--spark-conf` + +#### 5.5.2 Procedure Call + +```sql +CALL catalog.system.remove_orphan_files( + table => 'db.table_name', + older_than => TIMESTAMP '2026-06-13 00:00:00', + location => 's3://bucket/path/', + dry_run => true +) +``` + +**Parameters (from Iceberg `remove_orphan_files` procedure):** + +| Parameter | Type | Required | Description | +| ----------- | ----------- | -------- | ------------------------------------------------------------- | +| `table` | `string` | Yes | Fully qualified table name | +| `older_than`| `timestamp` | No | Only remove files older than this timestamp (default: 3 days) | +| `location` | `string` | No | Custom directory to scan for orphans (replaces table location when set; must be within the table's own location) | +| `dry_run` | `boolean` | No | If true, list orphan files without deleting them | + +#### 5.5.3 Output + +The procedure returns a result set with one column: + +| Column | Type | Description | +| ------------------- | -------- | ---------------------------------- | +| `orphan_file_location` | `string` | Path of each orphan file found/removed | + +The job will log the count of orphan files removed (or found in dry-run mode). + +#### 5.5.4 Security + +- SQL injection prevention via `escapeSqlString()` / `escapeSqlIdentifier()` + (same utilities as `IcebergExpireSnapshotsJob`) +- Input validation for `dry_run` (must be `true` or `false`) +- `location` validated against the table's own storage location — see + [Section 6.1](#61-location-validation) + +#### 5.5.5 Job Registration + +```java +// UPDATE: BuiltInJobTemplateProvider.java +private static final List<BuiltInJob> BUILT_IN_JOBS = + ImmutableList.of( + new SparkPiJob(), + new IcebergRewriteDataFilesJob(), + new IcebergUpdateStatsAndMetricsJob(), + new IcebergRemoveOrphanFilesJob()); // NEW +``` + +--- + +## 6. Safety Considerations + +Orphan file removal is inherently more dangerous than snapshot expiration or +compaction because it **permanently deletes files**. Several safety mechanisms +are built into the design: + +| Safety Mechanism | Description | +| --- | --- | +| **`older_than` default** | 3-day default ensures files from in-flight writes are not deleted | +| **`dry_run` mode** | Allows previewing which files would be deleted before actual removal | +| **Location validation** | Rejects any `location` outside the table's own storage location — see [6.1](#61-location-validation) | +| **Policy-gated** | Must be explicitly enabled by an administrator via policy creation | +| **Iceberg built-in safety** | The procedure itself only identifies files not referenced by any snapshot | + +### 6.1 Location Validation + +The `location` parameter is the single most dangerous input in this job. +Iceberg's `remove_orphan_files` deletes **every file under the given +location that the target table does not reference**. If a caller passes the +location of a *different* table, that table's data files are all unreferenced +from the target table's point of view, and the procedure deletes them — +silently destroying another table's data. + +The job must therefore validate `location` **before building the SQL**, not +rely on Iceberg to reject it: + +```java +static void validateLocation(NameIdentifier tableIdent, String location) { + String tableLocation = normalize(loadTableLocation(tableIdent)); + String requested = normalize(location); + + Preconditions.checkArgument( + requested.equals(tableLocation) + || requested.startsWith(tableLocation + "/"), + "location '%s' must be within the table's location '%s'", + location, tableLocation); +} +``` + +Validation rules: + +1. Resolve the target table's own storage location from table metadata. +2. Normalize both paths — canonicalize the scheme and authority, collapse + duplicate slashes, resolve `.` / `..` segments, and strip trailing slashes. +3. Reject the request unless the normalized `location` is the table location + itself or a descendant of it. +4. Reject symlinks or paths that resolve outside the table location after + normalization. + +Normalization must happen before the prefix check; comparing raw strings +would let `s3://bucket/db/table/../other_table` pass a naive +`startsWith` test. + +Validation happens in the job adapter (before submission) and again in the +Spark job (before building the procedure call), so an ad-hoc job submission +that bypasses the policy layer is still checked. + +### 6.2 Setting `older_than` Safely + +The `older_than` threshold should be set conservatively. Files from in-flight +writes or concurrent operations may not yet be referenced by a committed +snapshot. A minimum of 3 days is recommended. + +Setting `olderThanDays` to `0` removes all orphan files regardless of age. +This is only safe when no writer is active against the table — see +[Section 5.2.4](#524-why-olderthandays-defaults-to-3). + +--- + +## 7. File Changes Summary + +### 7.1 New Files + +| File | Layer | Description | +| -------------------------------------------------------------------------------- | -------- | ------------------------------------------ | +| `api/…/policy/IcebergOrphanFileRemovalContent.java` | Policy | Policy content with removal configuration | +| `maintenance/optimizer/…/handler/orphan/OrphanFileRemovalStrategyHandler.java` | Strategy | Trigger / score evaluation | +| `maintenance/optimizer/…/handler/orphan/OrphanFileRemovalJobContext.java` | Strategy | Job execution context | +| `maintenance/optimizer/…/job/GravitinoOrphanFileRemovalJobAdapter.java` | Adapter | Context → job config conversion | +| `maintenance/jobs/…/iceberg/IcebergRemoveOrphanFilesJob.java` | Job | Spark job | + +### 7.2 Modified Files + +| File | Change | +| ------------------------------------------------------- | ------------------------------------------------------------- | +| `api/…/policy/Policy.java` | Add `ICEBERG_ORPHAN_FILE_REMOVAL` to `BuiltInType` enum | +| `maintenance/optimizer/…/job/GravitinoJobSubmitter.java` | Register remove-orphan-files adapter in `jobAdapters` map | +| `maintenance/jobs/…/BuiltInJobTemplateProvider.java` | Register `IcebergRemoveOrphanFilesJob` | +| Handler registry | Register `OrphanFileRemovalStrategyHandler` | + +### 7.3 Test Files + +| File | Description | +| ------------------------------------------------------- | ------------------------------------- | +| `TestIcebergOrphanFileRemovalContent.java` | Policy content unit tests | +| `TestOrphanFileRemovalStrategyHandler.java` | Strategy handler unit tests | +| `TestGravitinoOrphanFileRemovalJobAdapter.java` | Job adapter unit tests | +| `TestIcebergRemoveOrphanFilesJob.java` | Spark job unit tests | + +--- + +## 8. Proposed PR Plan + +| PR | Scope | Dependencies | +| --- | --- | --- | +| **PR 1** | Job layer: `IcebergRemoveOrphanFilesJob` + `BuiltInJobTemplateProvider` + tests | None | +| **PR 2** | Policy + Strategy + Adapter: all remaining layers + tests | PR 1 | + +Since the total code size across the policy, strategy, and adapter layers is +expected to be well under 1000 lines, PRs 2 and 3 from the original plan are +combined into a single PR. + +--- + +## 9. Open Questions + +1. ~~**Cleanup interval tracking**~~ — Resolved: the last cleanup time is + stored as a table statistic (`custom-last-orphan-cleanup-time`) in + `statistic_meta`, aligned with how compaction persists its metrics. + See [Section 5.3.2](#532-tracking-the-last-cleanup-time). +2. ~~**Location parameter**~~ — Resolved: when `location` is specified, + only that location is scanned, and it must be within the table's own + storage location. See [Section 6.1](#61-location-validation). +3. **Dry-run result persistence** — Should dry-run results be stored + somewhere (e.g., job output metadata) for review before actual deletion? +4. ~~**PR granularity**~~ — Resolved: single PR for policy + strategy + + adapter layers since total code is expected to be under 1000 lines. +5. ~~**`older_than` minimum**~~ — Resolved: no hard minimum is enforced. + `olderThanDays: 0` is a deliberate escape hatch for reclaiming storage + when no writer is active. See + [Section 5.2.4](#524-why-olderthandays-defaults-to-3). +6. **Minimum run interval** — Out of scope. A uniform minimum-interval + mechanism will be defined across all four system built-in policies. + +--- + +## 10. Comparison with Other Maintenance Flows + +| Aspect | Compaction | Snapshot Expiration | Orphan File Removal | +| --- | --- | --- | --- | +| Policy type | `system_iceberg_compaction` | `system_iceberg_snapshot_expiration` | `system_iceberg_orphan_file_removal` | +| Strategy type | `iceberg-data-compaction` | `iceberg-snapshot-expiration` | `iceberg-orphan-file-removal` | +| Job template | `builtin-iceberg-rewrite-data-files` | `builtin-iceberg-expire-snapshots` | `builtin-iceberg-remove-orphan-files` | +| Scope | Per-partition (scored, top-N selected) | Whole table | Whole table (or custom location) | +| Data requirements | `TABLE_METADATA` + `TABLE_STATISTICS` + `PARTITION_STATISTICS` | `TABLE_METADATA` + `TABLE_STATISTICS` | `TABLE_METADATA` + `TABLE_STATISTICS` | +| Trigger metric | `custom-data-file-mse`, `custom-delete-file-number` | `custom-snapshot-count` | `custom-days-since-last-orphan-cleanup` | +| Iceberg procedure | `rewrite_data_files` | `expire_snapshots` | `remove_orphan_files` | +| Key parameters | strategy, sort-order, where, options | older_than, retain_last, stream_results | older_than, location, dry_run | +| Destructiveness | Rewrites data (recoverable via snapshots) | Removes metadata (irreversible) | Removes data files (irreversible) | +| Safety concern | Low — data is rewritten, not lost | Medium — old snapshots are removed | High — files are permanently deleted | Review Comment: Could u align table? -- 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]
