voonhous commented on code in PR #18359: URL: https://github.com/apache/hudi/pull/18359#discussion_r3504507822
########## 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 Review Comment: Decided against extending OCC for clean actions -- the check lives in `preCommit()` instead. The two questions are what drove that: **What `mutatedFileIds` would a clean report?** None that help. (The field is actually `mutatedPartitionAndFileIds`, `Set<Pair<partition,fileId>>`.) The only fileIds a clean has are the _expired_ slices it deletes, and by MVCC those never overlap a concurrent writer's new slices -- so the `(partition,fileId)` intersection is empty exactly when a blob conflict is real. The real conflict key is the external blob _path_, which that field can't hold. **Does `hasConflict` need a new path?** Yes, a new dimension -- it can't reuse the overlap logic. Blob paths don't map onto `(partition,fileId)`. Routing through OCC would mean: a `CLEAN_ACTION` case in both `ConcurrentOperation.init()` switches (throws today) with a new blob-path field read from the sidecar during metadata construction, a blob-path branch in `hasConflict`, and `CLEAN_ACTION` added to the `getInflightAndRequestedInstants()` / `getCandidateInstants()` whitelists -- four changes on the shared OCC path every writer hits, each gated on `hasBlobColumns` to preserve R6. Too much blast radius for a conflict that's genuinely a different axis. So instead the writer intersects its `HoodieWriteStat.externalBlobPaths` against the clean sidecar directly in `BaseHoodieWriteClient.preCommit()` -- ingestion and `insert_overwrite` only; table services just copy already-visible pointers, so they're covered by the replaced-FG lifecycle / compaction fencing. Zero change to `ConcurrentOperation` / `hasConflict`. Design doc updated under **"Why a targeted preCommit check, not an OCC extension"**, with the full OCC sketch kept as alternatives-considered. Separately: writing the ordering proof out properly surfaced a gap worth your eyes. `scheduleCleaning` snapshots liveness unlocked (`createCleanerPlan`), persists the plan under the lock (`saveToCleanRequested`), and deletes unlocked. A writer committing between the unlocked snapshot and the locked persist is invisible to both sides, producing a dangling reference -- so the preCommit check alone isn't sufficient as scheduled. Fix is a candidate re-validation under the lock, either at plan-persist (a) or at execution (b); I lean (b). Written up as **"Ordering Argument and the Planning-Snapshot Window"** -- would value your read on (a) vs (b). -- 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]
