[WIP][to be discussed] Add HUDI APIs for "stashing" and "restoring" partitions
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
### Task Description
**What needs to be done:**
- Add a "stashPartitions" operation in spark engine `stashPartitions(List partitionPaths, Path backupFolder)`. It will attempt to create a folder in `backupFolder` for each partitions path in `partitionPaths`, where each folder contains all the latest committted data files (without any replaced/compacted/uncleaned files) from the corresponding partition in the basepath. And then all files in the dataset basepath partition will be removed. For example, `stashPartitions(['datestr=2023-01-01'], "/backup/folder")` will create a folder in `/backup/folder/datestr=2023-01-01` containing all the latest data files from that partition in the dataset. If a given partition folder in the dataset was already stashed, it will be skipped. It will return a `HoodieWriteMetadata` that lists each partition and wether it "succeeded" stashing or was "skipped".
- Add a "restorePartitions" operation in spark engine `restorePartitions(List partitionPaths, Path backupFolder)`. It works as the "reverse" of stashPartitions; it will attempt to create a partition in the dataset basepath for each partitions path in `partitionPaths`, where each partition contains all the data files from the corresponding partition folder in the `backupFolder`. For example, `restorePartitions(['datestr=2023-01-01'], "/backup/folder")` will create`//datestr=2023-01-01` containing all the data files from `//datestr=2023-01-01`. If a given partition folder in the dataset was already restored, it will be skipped. It will return a `HoodieWriteMetadata` that lists each partition and wether it "succeeded" in restore or was "skipped".
Assumptions:
- Once the user attempts to stash a partition, it must never be written to again (except via a restore partition operation).
- Once a partition folder in the `backupFolder` has been part of a successful `restorePartitions`, that backup partition folder can be deleted from DFS.
- While a partition is targeted by an inflight stash/restore, it is expected that queries might fail (due to trying to read a DFS file that was just deleted) or return missing/partial data
Requirements:
- **data consistency** Data files from deleted partitions should have their references removed from internal metadata like MDT/indexes (similar to to other HUDI operations) and queries against the partition should return no data. In addition, if a partition is stashed while there are inflight/replaced files, then these files should not re-appear when the partition is restored. Since otherwise there may be duplicate/inconsistent data
- **usability** Stashing should recursively created folders in `backupFolder` as needed for partitions with nested folders, like `hour=x/minute=y`
- **immediate "moving"** When a stash operation commits, all files from the dataset partition (include HUDI internal files like .hoodie_partition_metadata) should be present in desired stashed location
- **immediate cleanup** When a specific operation commits (wether the stash or some other operation like clean), all files from the dataset partition (include HUDI internal files like .hoodie_partition_metadata) should have been deleted. This should not be dependent on wether the stash is in the cleaner commits/hour window.
- **TTL not set at time of stashing** When the stash operation is attempted, the user might not know how long they need to keep the stashed data around for (to restore). This means that we cannot set an explicit partition TTL at time of stashing
- **safe "file moving" (and fast if applicable)** For DFS like HDFS which support atomic rename of folders without copies, we should utilize DFS `rename` APIs to avoid having to wait for each file to be copied. For DFS without this functionality, like GCS, we can do a `[copy all files to dest-> delete all files from src ]`.
- **failures and rollbacks** If the operation fails after creating a plan, then it should be eventually rolled back by a rollback call (as part of clean’s rollback of failed writes). The rollback implementation should consist of “undoing” all the DFS operations: after rollback is completed, any partitions that were attempted to be stashed should still have their (latest) data files and any partitions attempted to be restored should still remain empty. Alternatively, we can make a rollback "re-attempt" the operation. In either case, we need to handle scenarios on GCS where the "move" failed in the middle
- **retries and concurrent writes** Before scheduling the plan, `stashPartitions` should start a heartbeat and check if there are any inflight writes targeting the same partition. It should attempt a rollback of any other `stashPartitions` plans with expired heartbeats. If any inflight instants still remain after that, then it should raise an exception. The same behavior should apply to `restorePartitions`, except it in its case it should only attempt rollback of other `restorePartition` instants
- **handling attempted writes to a stashed partition** Once a partition is stashed, any writes to that partition should emit a metric. We can create a user-provided strategy class where the use can implement what should be done if an ingestion write attempts to write to a stashed partiiton
**Why this task is needed:**
For our use case, when we apply TTL to older partitions of datasets we need to
- Stash them to a separate location for a grace period, in case users request them to be added back. Note that we cannot explicitly set a TTL, since the user will not know how long they need the data for at the time they attempt stashing.
- Ensure that we have an API that will synchronously remove all DFS "objects" in the TTL-ed partition folder. Since even disregarding store space, we need to "clean up" objects/inodes
### Task Type
Other
### Related Issues
**Parent feature issue:** (if applicable )
**Related issues:**
NOTE: Use `Relationships` button to add parent/blocking issues after issue is created.
Contributor guide
No contributing guide indexed for this repository
Research direction
Start by tracing the Spark engine entry points for stashPartitions and restorePartitions, then review how HoodieWriteMetadata, internal metadata, and cleaner rollback handle partition files. Define how recursive backup paths, atomic or copy-delete moves, retries, concurrency, and rollback interact. Done means both operations report per-partition results, remove stale metadata references, preserve consistency, and satisfy the stated cleanup and restore requirements.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, spark
- Domain
- backend-api-design, data-engineering, distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100