voonhous commented on code in PR #18359: URL: https://github.com/apache/hudi/pull/18359#discussion_r3505469813
########## rfc/rfc-100/rfc-100-blob-cleaner-design.md: ########## @@ -0,0 +1,777 @@ +<!-- + 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. +--> + +# RFC-100 Part 2: External Blob Cleanup for Unstructured Data + +## Proposers + +- @voon + +## Approvers + +- @rahil-c +- @vinothchandar +- @yihua + +## Status + +Issue: <Link to GH feature issue> + +> Please keep the status updated in `rfc/README.md`. + +--- + +## Abstract + +When Hudi cleans expired file slices, external out-of-line blob files they reference may become +orphaned -- still consuming storage but unreachable by any query. This RFC extends the existing file +slice cleaner to identify and delete these orphaned blob files safely and efficiently. The design +uses a two-stage pipeline: (1) per-file-group set-difference to find locally-orphaned blobs, and +(2) cross-file-group verification via MDT secondary index lookup. Targeted index lookups scale with +the number of candidates, not the table size. Tables without blob columns pay zero cost. + +This design focuses on **external blobs** -- the Phase 1 use case of RFC-100 where users have +existing blob files in external storage (e.g., `s3://media-bucket/videos/`) and Hudi manages the +*references* via the `BlobReference` schema, not the *storage layout*. + +--- + +## Background + +### Why Blob Cleanup Is Needed + +RFC-100 introduces out-of-line blob storage for unstructured data (images, video, documents). A +record's `BlobReference` field points to an external blob file by `reference.external_path`. When +the cleaner expires old file slices, the blob files they reference may no longer be needed -- but the +existing cleaner has no concept of transitive references. It deletes file slices without considering +the blob files they point to. Without blob cleanup, orphaned blobs accumulate indefinitely. + +### External Blobs + +Users have existing blob files in external storage (e.g., `s3://media-bucket/videos/`). Records +reference these blobs directly by path. Hudi manages the *references*, not the *storage layout*. +Cross-file-group sharing is common -- multiple records across different file groups can point to the +same blob. Key properties: + +| Property | External blobs | +|---------------------------|----------------------------------------------| +| Path uniqueness | Not guaranteed (user controls) | +| Cross-FG sharing | Common (multiple records, same blob) | +| Writer/cleaner race | Can occur (external paths outside MVCC) | +| Per-FG cleanup sufficient | No -- cross-FG verification needed | + +### Constraints and Requirements Reference + +Full descriptions and failure modes in [Problem Statement](rfc-100-blob-cleaner-problem.md). + +| ID | Constraint | Remarks | +|-----|-----------------------------------------------------|----------------------------------| +| C1 | Blob immutability (append-once, read-many) | | +| C2 | Delete-and-re-add same path | Real concern for external blobs | +| C3 | Cross-file-group blob sharing | Common for external blobs | +| C4 | MOR log updates shadow base file blob refs | | +| C5 | Existing cleaner is per-file-group scoped | | +| C6 | OCC is per-file-group | No global contention allowed | +| C7 | Replace commits move blob refs between file groups | Clustering, insert_overwrite | +| C8 | Savepoints freeze file slices and blob refs | | +| C9 | Rollback and restore can invalidate or resurrect | | +| C10 | Archival removes commit metadata | | +| C11 | Cross-FG verification needed at scale | | + +| ID | Requirement | +|-----|------------------------------------------------------------------| +| R1 | No premature deletion (hard invariant) | +| R2 | No permanent orphans (bounded cleanup) | +| R3 | MOR correctness (over-retention acceptable, under-retention not) | +| R4 | Concurrency safety (no global serialization) | +| R5 | Scale proportional to work, not table size | +| R6 | No cost for non-blob tables | +| R7 | All cleaning policies supported | +| R8 | Crash safety and idempotency | +| R9 | Observability (metrics for deleted, retained, reclaimed) | + +--- + +## Design Overview + +### Design Philosophy + +Blob cleanup extends the existing `CleanPlanner` / `CleanActionExecutor` pipeline -- same timeline +instant, same plan-execute-complete lifecycle, same crash recovery and OCC integration. A +`hasBlobColumns()` check gates all blob logic so non-blob tables pay near zero cost (schema scan +cost). + +External blobs require cross-file-group verification because the same blob can be referenced from +multiple file groups (C3, C11). The design uses targeted MDT secondary index lookups that scale +with the number of candidates, not the table size. + +### Two-Stage Pipeline + +| Stage | Scope | Purpose | When it runs | +|-------------|------------------|----------------------------------------------------------------------|------------------------------| +| **Stage 1** | Per-file-group | Collect expired/retained blob refs, compute set difference | Always (for blob tables) | +| **Stage 2** | Cross-file-group | Verify candidates against MDT secondary index or fallback scan | When local orphans exist | + +### Key Decisions + +| Decision | Choice | Rationale | +|---------------------|---------------------------------------------------------|----------------------------------------------------------------| +| Blob identity | `reference.external_path` | Path-based identity for external blobs | +| Cleanup scope | Per-FG candidate identification + cross-FG verification | Aligns with OCC (C6) and existing cleaner (C5); scales for C11 | +| Cross-FG mechanism | MDT secondary index on `reference.external_path` | Short-circuits on first non-cleaned FG ref | +| Blob delete storage | Sidecar Parquet file (`.hoodie/.aux/clean/`) | Avoids plan bloat; durable artifact for writer conflict checks | +| MOR strategy | Over-retain (union of base + log refs) | Safe (C4, R3); cleaned after compaction | + +```mermaid +flowchart LR + subgraph Planning["CleanPlanActionExecutor.requestClean()"] + direction TB + Gate{"hasBlobColumns()?"} + Gate -- No --> Skip["Skip blob cleanup<br/>(zero cost)"] + Gate -- Yes --> CP + + subgraph CP["CleanPlanner (per-partition, per-FG)"] + direction TB + Policy["Policy method<br/>→ FileGroupCleanResult<br/>(expired + retained slices)"] + S1["<b>Stage 1</b><br/>Per-FG blob ref<br/>set difference"] + Policy --> S1 + end + + S1 --> S2["<b>Stage 2</b><br/>Cross-FG verification<br/>(MDT secondary index)"] + S2 --> SC["Write sidecar Parquet<br/>.hoodie/.aux/clean/<instant><br/>.blob_deletes.parquet"] + end + + subgraph Plan["HoodieCleanerPlan"] + FP["filePathsToBeDeleted<br/>(existing)"] + EM["extraMetadata[blobDeletesPath]<br/>(pointer to sidecar)"] + end + + SC --> EM + CP --> FP + + subgraph Execution["CleanActionExecutor.runClean()"] + direction TB + RS["Read sidecar Parquet"] + DF["Delete file slices<br/>(existing, parallel)"] + DB["Delete blob files<br/>(new, parallel)"] + RS --> DB + end + + FP --> DF + EM --> RS +``` + +--- + +## Algorithm + +### Stage 1: Per-File-Group Local Cleanup + +Stage 1 runs after the existing policy logic determines which file slices are expired and retained +for a given file group. It collects blob refs from both sets and computes locally-orphaned blobs by +set difference. All local orphans proceed to Stage 2 for cross-FG verification. + +``` +Input: A file group FG with expired_slices and retained_slices (from policy) +Output: local_orphan_candidates -- external blobs needing cross-FG verification + +for each file_group being cleaned: + + // Collect expired blob refs (base files + log files) + // Must read log files: blob refs introduced and superseded within the log + // chain before compaction would otherwise become permanent orphans. + expired_refs = Set<external_path>() + for slice in expired_slices: + for ref in extractBlobRefs(slice.baseFile): // columnar projection + if ref.type == OUT_OF_LINE and ref.managed == true: + expired_refs.add(ref.external_path) + for ref in extractBlobRefs(slice.logFiles): // full record read + if ref.type == OUT_OF_LINE and ref.managed == true: + expired_refs.add(ref.external_path) + + if expired_refs is empty: + continue // no blob work for this FG + + // Collect retained blob refs (base files only) + // Cleaning is fenced on compaction: retained base files contain the merged + // state. Log reads are unnecessary -- any shadowed base ref causes safe + // over-retention, cleaned after the next compaction cycle. + retained_refs = Set<external_path>() + for slice in retained_slices: + for ref in extractBlobRefs(slice.baseFile): // columnar projection only + if ref.type == OUT_OF_LINE and ref.managed == true: + retained_refs.add(ref.external_path) + + // Compute local orphans by set difference + local_orphans = expired_refs - retained_refs + + // All local orphans proceed to Stage 2 for cross-FG verification + all_local_orphans.addAll(local_orphans) +``` + +**Correctness notes:** + +- **MOR -- expired side reads base + logs:** Blob refs can be introduced and superseded entirely + within the log chain (e.g., `log@t2: row1->blob_B`, then `log@t3: row1->blob_C`). After + compaction, `blob_B` exists only in the expired log. Skipping logs would orphan it permanently. +- **MOR -- retained side reads base only:** Cleaning is fenced on compaction, so retained base + files contain the merged state. Shadowed base refs cause over-retention (safe), cleaned after + the next compaction. +- **Savepoints:** Inherited from existing cleaner -- savepointed slices stay in the retained set. +- **Replaced FGs (replace commits):** `retained_slices` is empty, so all blob refs become + candidates. For external blobs, clustering copies the pointer to the target FG, so Stage 2 + finds the reference in the target FG and retains the blob. + +### Stage 2: Cross-File-Group Verification + +Stage 2 verifies each local orphan candidate against the global state to determine if the blob is +still referenced by any active file slice outside the cleaned file groups. This is necessary because +external blobs can be shared across file groups (C3, C11). + +#### Primary path: MDT secondary index + +When the MDT secondary index on `reference.external_path` is available and fully built: + +``` +Input: all_local_orphans, cleaned_fg_ids +Output: blob_files_to_delete (confirmed globally orphaned) + +candidate_paths = all_local_orphans.distinct() + +// Step 1: Batched prefix scan on secondary index +// Key format: escaped(external_path)$escaped(record_key) +// Returns ALL record keys that reference each candidate path +// Uses engine-context HoodieData (e.g., RDD on Spark) to distribute work +// across executors -- candidate sets can be large (row-level blob refs). +candidate_paths_data = engineContext.parallelize(candidate_paths) +path_to_record_keys = mdtMetadata.readSecondaryIndexDataTableRecordKeysWithKeys( + candidate_paths_data, indexPartitionName) + .groupBy(pair -> pair.getKey()) + +// Step 2: Batch record index lookup -- ONE call for ALL record keys +// Sorts keys internally, single sequential forward-scan through HFile. +all_record_keys = path_to_record_keys.values().flatMap() +all_locations = mdtMetadata.readRecordIndexLocations( + all_record_keys) // -> Map<recordKey, (partition, fileId)> + +// Step 3: In-memory resolution with short-circuit per candidate +for path in candidate_paths: + record_keys = path_to_record_keys.getOrDefault(path, []) + + if record_keys is empty: + blob_files_to_delete.add(path) // globally orphaned + continue + + found_live_reference = false + for rk in record_keys: + location = all_locations.get(rk) + if location != null and location.fileId NOT in cleaned_fg_ids: + found_live_reference = true + break // short-circuit (in-memory) + + if not found_live_reference: + blob_files_to_delete.add(path) // all refs in cleaned FGs +``` + +**Cost model.** Three steps: (1) batched prefix scan on secondary index, (2) batched record index +lookup in a single sorted HFile scan, (3) in-memory resolution with short-circuit. Steps 1 and 2 +are each a single I/O pass; step 3 is pure hash set lookups. + +| Step | I/O | Estimated cost (2K candidates) | +|---------------------------|-----------------------------------------------|--------------------------------| +| 1. Prefix scan (batched) | 1 HFile open + forward scan of N prefix keys | ~2-5s | +| 2. Record index (batched) | 1 HFile open + forward scan of 6K sorted keys | ~1-2s | +| 3. In-memory resolution | Hash set checks (cleaned_fg_ids) | ~0ms | + +*Estimates assume cloud object storage (S3/GCS/ADLS), ~10-100ms per-read latency, ~50-200 MB/s +sequential throughput, 64-256KB HFile blocks. Pending benchmarking.* + +**Index definition.** Uses the existing `HoodieIndexDefinition` mechanism with +`sourceFields = ["<blob_col>", "reference", "external_path"]`. The nested field path is supported +by `HoodieSchemaUtils.projectSchema()` and `SecondaryIndexRecordGenerationUtils`. No new index +infrastructure is needed. + +**Safety check.** The cleaner verifies the index is fully built before using it via +`getMetadataPartitions()` and `getMetadataPartitionsInflight()`. A partially-built index falls +back to the table scan path. + +#### Fallback path: table scan with circuit breaker + +When the MDT secondary index is unavailable, Stage 2 falls back to a parallelized table scan +across all partitions. A circuit breaker (`hoodie.cleaner.blob.external.scan.max.candidates`, +default 1000) defers cleanup if candidates exceed the threshold, preventing the scan from becoming +a bottleneck on large tables. The operator is warned to enable the MDT secondary index. + +#### Decision matrix + +| Condition | Path used | Cost | Suitable for | +|-----------------------------|---------------|-----------------------|---------------------------| +| No local orphan candidates | Skip Stage 2 | Zero | No blob work this cycle | +| MDT secondary index enabled | Index lookup | O(candidates) | Any scale | +| No index, few candidates | Table scan | O(candidates * table) | Small tables | +| No index, many candidates | Circuit break | Zero (deferred) | Large tables need index | + +```mermaid +sequenceDiagram + participant C as Cleaner (Stage 2) + participant SI as MDT Secondary Index + participant RI as MDT Record Index + + Note over C: Step 1: Batch prefix scan + C->>SI: All candidate paths (N paths, single call) + SI-->>C: Map<path, List<recordKey>> + + Note over C: Step 2: Batch record index lookup + C->>C: Collect all record keys from all candidates + C->>RI: readRecordIndexLocations(all record keys) + Note over RI: Sort keys → single sequential<br/>forward-scan through HFile + RI-->>C: Map<recordKey, (partition, fileId)> + + Note over C: Step 3: In-memory resolution + loop For each candidate path + alt No record keys for this path + Note right of C: Globally orphaned → DELETE + else Has record keys + C->>C: Check each location.fileId<br/>against cleaned_fg_ids (in-memory) + alt Any fileId NOT in cleaned_fg_ids + Note right of C: Live reference → RETAIN + else All in cleaned FGs + Note right of C: Globally orphaned → DELETE + end + end + end +``` + +### Execution Flow + +``` +1. CleanPlanActionExecutor.requestClean() + ├── hasBlobColumns(table)? // R6: zero-cost gate + ├── CleanPlanner: for each partition, for each file group: + │ ├── Refactored policy method -> FileGroupCleanResult + │ └── If hasBlobColumns: Stage 1 per FG + ├── CleanPlanner: replaced file groups -> Stage 1 + ├── If local orphan candidates non-empty: Stage 2 + ├── Write sidecar Parquet to .hoodie/.aux/clean/<instant>.blob_deletes.parquet + ├── Build HoodieCleanerPlan with extraMetadata["blobDeletesPath"] + └── Persist plan to timeline (REQUESTED state) + +2. CleanActionExecutor.runClean() + ├── Transition to INFLIGHT + ├── Read sidecar Parquet (blob delete list) + ├── Delete file slices (existing, parallelized) + ├── Delete blob files (new, parallelized) // parallel with file slice deletes + ├── Build HoodieCleanMetadata with blobCleanStats + blobDeletesSidecarPath + └── Transition to COMPLETED +``` + +```mermaid +sequenceDiagram + participant P as CleanPlanActionExecutor + participant TL as Timeline + participant AUX as .hoodie/.aux/clean/ + participant E as CleanActionExecutor + participant S as Storage + + Note over P: requestClean() + + P->>P: Stage 1: per-FG blob ref set difference + P->>P: Stage 2: MDT index lookup (if candidates exist) + P->>AUX: Write sidecar Parquet (blob delete list) + P->>TL: Persist HoodieCleanerPlan<br/>(extraMetadata["blobDeletesPath"]) + + Note over TL: REQUESTED + + rect rgb(255, 245, 230) + Note right of TL: Crash before plan → orphaned sidecar<br/>(harmless, cleaned at next startup) + end + + E->>TL: Transition plan state + Note over TL: INFLIGHT + + E->>AUX: Read sidecar Parquet + + rect rgb(255, 245, 230) + Note right of TL: Crash here → re-read sidecar,<br/>re-delete (FileNotFound = success) + end + + par Parallel deletion + E->>S: Delete file slices (existing) + and + E->>S: Delete blob files (from sidecar) + end + + E->>E: Build HoodieCleanMetadata<br/>(blobCleanStats + sidecarPath) + E->>TL: Transition plan state + + rect rgb(255, 245, 230) + Note right of TL: Crash here → re-execute<br/>(all deletes are no-ops) + end + + Note over TL: COMPLETED + Note over AUX: Sidecar lives until archival<br/>(for writer conflict checks) +``` + +--- + +## Integration with Existing Cleaner + +### CleanPlanner Refactoring + +The existing `CleanPlanner` policy methods produce `CleanFileInfo` objects (file paths to delete) +without exposing the expired/retained slice partition that blob cleanup needs. We introduce a new +return type: + +```java +public class FileGroupCleanResult { + private final List<CleanFileInfo> filePathsToDelete; + private final List<FileSlice> expiredSlices; + private final List<FileSlice> retainedSlices; +} +``` + +The three policy methods (`getFilesToCleanKeepingLatestVersions`, +`getFilesToCleanKeepingLatestCommits`, `getFilesToCleanKeepingLatestHours`) are refactored to +collect both expired and retained slices alongside the existing `CleanFileInfo` production. The +existing behavior is unchanged -- the refactoring adds output without modifying the +expired/retained classification logic. + +### Replaced File Group Handling + +Replaced file groups (from clustering, insert_overwrite, insert_overwrite_table) are cleaned via +`getReplacedFilesEligibleToClean()`. A parallel method `getReplacedFileGroupBlobCleanResults()` +produces `FileGroupCleanResult` objects with `retainedSlices = empty` and +`expiredSlices = all slices`. This feeds into Stage 1 identically to normal file groups. + +### Schema Changes: HoodieCleanerPlan + +**No new fields.** The blob delete list is stored in a sidecar Parquet file, not in the plan. The +plan references the sidecar via the existing `extraMetadata` map: + +- `extraMetadata["blobDeletesPath"]` -- path to the sidecar Parquet file + (e.g., `.hoodie/.aux/clean/20240101120000.blob_deletes.parquet`) + +This avoids plan bloat regardless of the number of blob candidates. The `extraMetadata` map is +already part of the `HoodieCleanerPlan` Avro schema -- no schema change is needed. + +### Sidecar Parquet File: Blob Delete List + +The blob delete list is stored as a sidecar Parquet file at +`.hoodie/.aux/clean/<instant>.blob_deletes.parquet`. This file is the **single source of truth** +for blob delete candidates across all clean states (REQUESTED, INFLIGHT, COMPLETED). + +**Schema:** +``` +message BlobDeleteList { + required binary external_path (STRING); +} +``` + +**Write sequence:** The sidecar is written **before** the plan is persisted to the timeline. +By the time the plan is visible, the sidecar is already durable on storage. This ensures +atomicity: if a writer sees the plan, the sidecar is guaranteed to exist. + +**Lifecycle:** The sidecar lives until the clean instant is **archived**. This covers writer +conflict checks against REQUESTED, INFLIGHT, and COMPLETED clean actions. When the instant is +archived, the sidecar is deleted as part of archival cleanup. + +**Orphan cleanup:** If a crash occurs after writing the sidecar but before persisting the plan, +the sidecar is orphaned. At cleaner startup, a lightweight cleanup routine lists +`.hoodie/.aux/clean/` and deletes any sidecar whose instant has no corresponding plan on the +timeline. + +**Rollback:** When a clean plan is rolled back, rollback logic reads +`extraMetadata["blobDeletesPath"]` and deletes the sidecar. + +**Size:** Dictionary encoding on shared path prefixes (e.g., `s3://media-bucket/videos/...`) +provides good compression. 10K blob paths ≈ a few hundred KB. + +### Schema Changes: HoodieCleanMetadata + +A new nullable field `blobCleanStats` of type `HoodieBlobCleanStats`: Review Comment: Good catch -- archival doesn't touch `.aux/` today, but it does already clean up an archived instant's per-instant side files, so this slots into an existing hook rather than a new service. `TimelineArchiverV2.archiveIfRequired` runs a per-archived-action callback during archival -- today `deleteAnyLeftOverMarkers(context, activeAction)` (`:121`, `:395`), which deletes the instant's marker directory keyed on `activeAction.getInstantTime()`. Blob-sidecar cleanup extends that same callback: for an archived `CLEAN_ACTION`, derive `.hoodie/.aux/clean/<instantTime>.blob_deletes.parquet` by convention from `activeAction.getInstantTime()` and delete it (idempotent -- missing file = success). No plan/metadata read needed since the path is conventional. Added to both archiver versions (V1 + V2). Backstop is the startup orphan sweep the design already specifies: it lists `.aux/clean/` and drops any sidecar whose instant isn't on the active timeline (archived, or never persisted), so a crash between archiving the instant and deleting its sidecar self-heals on the next cleaner run. I also tightened that sweep's predicate to key on instant-presence rather than "has a plan," so it can't reclaim a COMPLETED-but-not-yet-archived sidecar a writer might still need. Updated the "Lifecycle" and "Orphan cleanup" bullets accordingly. -- 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]
