voonhous commented on code in PR #18359: URL: https://github.com/apache/hudi/pull/18359#discussion_r3505550997
########## 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`: + +- `totalBlobFilesDeleted`, `totalBlobFilesRetained` +- `totalBlobStorageReclaimed` +- `blobDeletesSidecarPath` -- pointer to the sidecar (for COMPLETED-state conflict checks) +- `failedBlobFilePaths` -- only failures (expected to be empty or near-empty) + +### hasBlobColumns() Gate + +An in-memory schema check (`TableSchemaResolver.getTableSchema().containsBlobType()`) gates all +blob cleanup logic. Requires making `containsBlobType()` public (one-line visibility change). + +--- + +## Concurrency & Safety + +### Writer-Cleaner Race: Conflict Check + +Under Hudi's MVCC design, the cleaner and writers operate on non-overlapping file slices -- the +cleaner never conflicts with writers on file slice operations. However, external blob files are +**not** covered by MVCC: a writer's new file slice may reference an external blob that the cleaner +is simultaneously evaluating for deletion. + +**Writer-side conflict check in `preCommit()`.** The gap between the cleaner's planning-time +snapshot and its actual file deletion is closed by a commit-time conflict check: + +1. Writers track external managed blob paths in `HoodieWriteStat.externalBlobPaths` (in-memory + collection, no additional I/O). +2. At commit time (in `preCommit()`, under the existing transaction lock), the writer checks all + three clean states -- COMPLETED, INFLIGHT, and REQUESTED -- because a REQUESTED plan can begin + executing at any moment (the REQUESTED->INFLIGHT transition doesn't acquire the transaction + lock). +3. For each clean instant, the writer locates the sidecar Parquet file: + - REQUESTED/INFLIGHT: reads `extraMetadata["blobDeletesPath"]` from the plan + - COMPLETED: reads `blobDeletesSidecarPath` from `HoodieCleanMetadata` +4. The writer reads the sidecar and checks intersection with its external blob paths. +5. If any overlap is found, the commit is rejected with `HoodieWriteConflictException` and the + writer retries. + +The sidecar is the single source of truth across all three states -- no blob paths are stored +inline in the plan or completed metadata. + +**Note:** `ConcurrentOperation` and `TransactionUtils.getInflightAndRequestedInstants()` currently +exclude `CLEAN_ACTION`. Adding external blob cleanup requires extending conflict resolution to +include clean actions when blob columns exist. + +Cost is zero for non-blob tables. For external blobs: one timeline scan + 1-3 sidecar Parquet +reads (~50-200ms each on cloud storage), gated on the writer having external blob paths. + +```mermaid +sequenceDiagram + participant W as Writer + participant TL as Timeline + participant AUX as Sidecar (.aux) + participant CL as Cleaner + + Note over W,CL: Scenario A: Writer commits BEFORE cleaner plans + + W->>TL: Commit (references blob_X) + CL->>AUX: Write sidecar (blob_X in delete list) + CL->>TL: Plan cleanup (extraMetadata -> sidecar path) + Note right of CL: Sees blob_X in retained slice → not deleted + Note over W,CL: ✓ Safe + + Note over W,CL: Scenario B: Writer commits AFTER cleaner plans, BEFORE delete + + CL->>AUX: Write sidecar (blob_X in delete list) + CL->>TL: Plan cleanup + Note over TL: REQUESTED / INFLIGHT + W->>TL: preCommit() -- reads plan, finds sidecar path + W->>AUX: Read sidecar Parquet + Note left of W: blob_X in sidecar<br/>→ HoodieWriteConflictException<br/>→ Writer retries + Note over W,CL: ✓ Safe -- conflict detected + + Note over W,CL: Scenario C: Writer commits AFTER cleaner deletes, BEFORE COMPLETED + + CL->>CL: Delete blob_X from storage + Note over TL: Still INFLIGHT + W->>TL: preCommit() -- reads INFLIGHT plan + W->>AUX: Read sidecar Parquet + Note left of W: blob_X in sidecar → Rejection + Note over W,CL: ✓ Safe -- same as B + + Note over W,CL: Scenario D: Cleaner completes, THEN writer acquires lock + + CL->>TL: Transition to COMPLETED (metadata has sidecar path) + W->>TL: preCommit() -- reads COMPLETED metadata + W->>AUX: Read sidecar Parquet + Note left of W: blob_X in sidecar → Rejection + Note over W,CL: ✓ Safe +``` + +### Concurrency Matrix + +| Operation | Concurrent with Blob Cleaner | Safety Mechanism | +|--------------------------------|------------------------------|-----------------------------------------------------------| +| Regular write (INSERT/UPSERT) | Safe | Writer-side conflict check in preCommit() | +| Compaction | Safe | `isFileSliceNeededForPendingMajorOrMinorCompaction` | +| Clustering / insert_overwrite | Safe | Replaced FG lifecycle; Stage 2 finds refs in target FG | +| Rollback | Safe | MOR over-retention; clean operates on post-rollback state | +| Restore | Safe | Clean operates on post-restore state | +| Savepoint create/delete | Safe | Savepointed slices excluded from cleaning | +| Archival | No interaction | Blob cleaner reads file slices, not commit metadata | +| Another cleaner instance | Safe | `TransactionManager`; `checkIfOtherWriterCommitted` | +| MDT writes (index maintenance) | Safe | MDT commit atomicity | + +### Crash Recovery + +Crash recovery is idempotent by construction, using the same mechanisms as existing file slice +cleaning. The sidecar Parquet file ensures the blob delete list survives crashes: + +| Crash point | Recovery | +|-----------------------------------------------|--------------------------------------------------------------------------------------------------------| +| After sidecar written, before plan persisted | No REQUESTED instant on timeline. Cleaner starts fresh. Orphaned sidecar cleaned at next startup. | +| After plan persisted, before execution | REQUESTED instant found; plan re-read, sidecar re-read, executed. | +| During execution (partial deletes) | INFLIGHT instant re-executed. Sidecar re-read. Already-deleted files return FileNotFoundException -> success. | +| After execution, before COMPLETED | INFLIGHT re-executed. All deletes are no-ops. Metadata written, instant transitions to COMPLETED. | +| Plan rolled back | Rollback reads `extraMetadata["blobDeletesPath"]` and deletes the sidecar. | + +--- + +## Performance + +### Cost Summary + +| Workload | Stage 1 cost | Stage 2 cost | Total per cleanup cycle | +|------------------------------|---------------------------------|----------------------------|--------------------------------| +| Non-blob table | Zero (`hasBlobColumns` gate) | N/A | **Zero** | +| External blobs (index) | ~6 Parquet reads per cleaned FG | O(C * R_avg) | O(cleaned_FGs + C * R_avg) | +| External blobs (scan) | ~6 Parquet reads per cleaned FG | O(candidates * table_size) | Circuit breaker limits this | + +### Back-of-Envelope: Example 6 (50K FGs, 2K External Candidates) + +| Parameter | Value | Notes | +|-------------------------------------|-----------|------------------------------------------------------| +| FGs cleaned this cycle | 500 | 1% of table | +| Stage 1: reads per FG | ~6 | 3 retained + 3 expired slices | +| Stage 1: total reads | 3,000 | Parallelized across executors, ~20s | +| External blob candidates | 2,000 | Locally orphaned in cleaned FGs | +| Avg refs per candidate | 3 | Random assumption | +| Total record keys | 6,000 | 2,000 * 3 | +| **Stage 2 cost (estimated)** | | | +| Step 1: batched prefix scan | 1 call | Returns 6K record keys, ~2-5s estimated | +| Step 2: batched record index lookup | 1 call | 6K keys sorted, single HFile scan, ~1-2s estimated | +| Step 3: in-memory resolution | 6K checks | Hash set lookups against cleaned_fg_ids, ~0ms | +| **Total Stage 2** | **~3-7s** | Estimated; see I/O assumptions in Stage 2 cost model | +| Comparison: naive full-table scan | 12.5TB | 50K FGs * 5 slices * 50MB = prohibitive | + +### Memory Budget + +Per-FG blob ref sets: ~100MB peak (500K records * 100 bytes/ref for expired + retained). FGs are +processed sequentially within each partition batch -- per-FG sets are computed and discarded, not +accumulated. Only the output list (`all_local_orphans`) grows, containing +only orphaned refs (much smaller). Peak heap for Stage 1: ~100MB * `cleanerParallelism` = 400MB-1.6GB. + +Stage 2 output lists (`candidate_paths`, `all_record_keys`) can be large -- each cleaned FG may +contribute row-level blob refs as candidates. These are backed by engine-context `HoodieData` +(e.g., Spark RDD) and distributed across executors, avoiding driver memory pressure. + +--- + +## Configuration + +| Property | Default | Description | +|----------------------------------------------------|---------|-----------------------------------------------------------------------------------| +| `hoodie.cleaner.blob.enabled` | `true` | Enable blob cleanup during clean action | +| `hoodie.cleaner.blob.dry.run` | `false` | Compute blob cleanup plan and log results but do not execute | +| `hoodie.cleaner.blob.external.scan.parallelism` | `10` | Parallelism for Stage 2 fallback table scan | +| `hoodie.cleaner.blob.external.scan.max.candidates` | `1000` | Circuit breaker for Stage 2 fallback scan; exceeding defers blob cleanup | +| `hoodie.metadata.index.secondary.column` | (none) | Set to `<blob_col>.reference.external_path` for cross-FG verification | + +--- + +## Rollout / Adoption Plan + +**Foundation (shared prerequisite).** `CleanPlanner` refactoring (policy methods return +`FileGroupCleanResult`), sidecar Parquet write/read infrastructure, and the `hasBlobColumns` +zero-cost gate. + +**Stage 1 (per-FG cleanup).** Set-difference logic. Produces local orphan candidates for Stage 2. + +**Stage 2 (cross-FG verification) -- priority.** External blobs are the primary initial use case -- +cross-FG verification prevents premature deletion of shared blobs. Requires MDT + record index + +secondary index on `reference.external_path`. Includes fallback table scan with circuit breaker. + +**Sidecar lifecycle.** Write sidecar at planning time, read at execution time, clean up at archival. +Orphan cleanup at cleaner startup. Rollback cleanup via `extraMetadata`. + +**Writer-side conflict check.** `preCommit()` conflict check reads sidecar for concurrency safety. +Requires extending `ConcurrentOperation` / `TransactionUtils` to include clean actions when blob +columns exist. Review Comment: Dug into the MDT internals for this. The concern is real in general -- an index is a cache, and a false negative (missing a live reference) is the R1-dangerous direction; a false positive just over-retains -- but MDT's design prevents it for steady state, with one real caveat that the existing "fully built" check already gates. Why steady-state staleness can't produce a false negative -- three mechanisms compose: 1. **Same-commit maintenance:** SI records are written in the *same* MDT delta-commit as the data commit, at the same instant time (`HoodieBackedTableMetadataWriter.update` -> `processAndCommit`, SI in the same `partitionToRecordMap`). Not a lazily-maintained side cache. 2. **Failure coupling:** the MDT commit runs before `saveAsComplete` (`BaseHoodieWriteClient.commit`); if the MDT update throws, the data instant never completes. So a completed data commit implies its SI entries exist. 3. **Reader watermark:** every SI read filters to completed data instants via `getValidInstantTimestamps` (`InstantRange` EXACT_MATCH in `readSliceWithFilter`), so a crash-after-MDT-commit-before-`saveAsComplete` entry is hidden, and every completed commit is exposed. Net: the index can't miss a completed commit's reference or surface a failed one. (A *new* reference committed by a concurrent writer after the fence is the writer-race, handled by the preCommit check + the planning-snapshot window, not staleness.) The one genuinely unsafe window is **bootstrap/backfill**: while the SI partition is inflight, the watermark doesn't help -- a completed historical instant can be in the valid set but not yet reconciled by the catch-up task, a real false negative. So the check must be precise: trust the index only when `isMetadataPartitionAvailable(SECONDARY_INDEX)` is true **and** it's absent from `getMetadataPartitionsInflight()`; otherwise fall back to the ground-truth table scan. MDT disable/rebuild/restore flip that flag or are reconciled into the watermark, so the same gate covers them. I also made a one-sided-trust backstop explicit: the index may authorize retention freely, but deletion only under that availability guarantee -- any uncertainty falls back to the scan or defers, never deletes. So even if a future async-SI-maintenance mode weakened the steady-state coupling, R1 holds. Written up under "Index staleness and consistency." -- 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]
