[HUDI-5823][RFC-65] RFC for Partition TTL Management

[stream2000]

Change Logs

RFC-65 Partition Lifecycle Management

Impact

None

Risk level (write none, low medium or high below)

Low

Documentation Update

None

Contributor's checklist

• Read through contributor's guide
• Change Logs and Impact were stated clearly
• Adequate tests were added if applicable
• CI passed

[codope]

@stream2000 Thanks for putting up an RFC. Left some comments to consider. Can you also share the PR in the dev list thread?

[codope]

Can you define valid data file? Is it just the latest version of each file id i.e. latest file slice?

[stream2000]

Yes, we calculate only the latest file slice. If we want to calculate all the file slices instead of just the latest file slice, we can add a config to control the behavior or adding another stat field. Here we choose to calculate only the latest file slice because we think it reveals the real data size of the file group.

[codope]

Would be great if you can also discuss how the read path will work? How do we maintain a consistent filesystem view for readers given that delete partition operation can take time?

[stream2000]

delete_partition has been already implemented in current hudi master branch. When we call delete_partition to delete a list partitions, the executor will list all files for the partitions to delete and store them in the replacecommit commit metadata. After the replace commit committed, all the filesystem views that have seen the replace commit will exclude files that were replaced in the replace commit.

[nsivabalan]

can you help me understand how is the orchestration or scheduling of these policies are doe. Is it manual via sql procedure? for eg, how often this clean up job is triggered.

RFC should cover that as well. and definitely, we should avoid adding/storing data in adhoc json files if possible.
If we plan to run this once a day or much lower candence, its not that bad a idea to collect stats on a on-demand basis, clean up and move on rather than storing the stats and ensuring we keep it updated after every commit. It will add some additional overhead to regular writes to ensure the stats are aligned w/ actual data.

[nsivabalan]

can you help us understand the use-case here. I mean, I am trying to get an understanding of the sub-partitions here. in hudi, we have only one partitioning, but if could be multi-leveled. so, trying to see, if we can keep it high level.

[nsivabalan]

I feel, both (2) and (3) is very much catered towards multi-field partitioning like an ProductId/datstr based partitioning. can we layout high level strategies for one level partitioning as well in addition to multi-field partitioning.

[nsivabalan]

is it possible to simplify the strategies where in we can achieve it for both single or multi field partitioning. for eg,
TTL any partition whose last mod time (last time when data was added/updated), is > 1 month for eg. this will work for both a single field partitioning (datestr), or multi-field (productId/datestr).
Open to ideas.

[nsivabalan]

we should also call out that the sub-partitioning might work only for day based or time based sub-partitioning right. for eg, lets say, if partitioning is datestr/productId. how do we know out of 1000 productIds under a given day, which 100 is older or newer (assuming all 1000 was created in same commit).

[nsivabalan]

we should be able to add regex and achieve this.
for eg,
Map<{PartitionRegex/Static Partitions{ -> {TTL policy} >
so, this map can have multiple entries as well.

[nsivabalan]

what is last mod time. is it referring to new inserts, or updates as well ?

[stream2000]

Yes, inserts/updates will be treated as modification to the partition. And we track them by looking the commit/deltacommit write stats.

[danny0405]

How to fetch the info when a commit is archived?

[stream2000]

In the latest version of the RFC, we use the max instant time of the committed file slices in the partition as the partition's last modified time for simplicity. Otherwise, we need some extra mechanism to get the last modified time. In our inner version, we maintain an extra JSON file and update it incrementally as new instants committed to get the real modified time for the partition. Also, we can use metadata table to track the last modify time. What do you think about this?

[nbalajee]

When retiring the old/unused/not-accessed partitions, another approach we are taking internally is:
(a) stash the partitions to be cleaned up in .stashedForDeletion folder (at .hoodie level).
(b) partitions stashed for deletion will wait in the folder for a week (or time dictated by the policy) before actually getting deleted. In cases, where we realize that something has been accidentally deleted (like a bad policy configuration, TTL exclusion not configured etc), we can always move back from the stash to quickly recover from the TTL event.
(c) We shall configure policies for .stashedForDeletion// subfolders to manage for appropriate tiering level (whether to be moved to a warm/cold tier etc)
(d) in addition to the deletePartitions() API, which would stash the folder (instead of deleting) based on the configs, we would need a restore API to move the subfolder/files back to their original location.
(e) Metadata left by the delete operation to be synced with MDT to keep the file listing metadata in sync with the file system. (In cases where replication to a different region is supported, this also would warrant applying the changes on the replicated copies of data).

[stream2000]

Maybe we can add the stash/restore mechanism to replace commit/clean process of hudi instead of dealing with it in TTL management? TTL management should only decide which partitions are outdated and call delete_partition to delete them. If we want to retain the deleted data we can add extra mechanism in the delete_parrtition method.

[SteNicholas]

@stream2000, could we also introduce record ttl management? Partition ttl management and record ttl management both need the ttl policy.

[SteNicholas]

@stream2000, could we introduce HoodieTTLPolicy interface? Then HoodiePartitionTTLPolicy implements the HoodieTTLPolicy. HoodieRecordTTLPolicy could also implement this interface in feature.

[stream2000]

@SteNicholas Nice suggestions! I will take it into consideration when implementing the policy and make sure we can integrate more type of TTL policy in the future

[stream2000]

@codope @nsivabalan @SteNicholas @nbalajee Sorry for the delayed response, I've been busy with other things these days. Will address comments and update the RFC.

[stream2000]

@stream2000, could we also introduce record ttl management? Partition ttl management and record ttl management both need the ttl policy.

@SteNicholas It will be nice to have record-level ttl management~ We can open another RFC for record-level TTL management and concentrate on Partition TTL Management in this RFC. We can also discuss more on how to make it easy to integrate other types of ttl policy in this RFC.

[stream2000]

@nsivabalan @SteNicholas Hi, thanks for your review. I have updated the RFC according to your comments.

1. Simplify the TTL policy, now we will only support KEEP_BY_TIME so we don't need to compare partition values between partitions.
2. Remove the persistent JSON stats mechanism. We can gather partition stats every time we do TTL management.
3. Add more detail about executing TTL management. We will support async table services for TTL management and both spark and flink engine can execute the async service.
4. Consider Record level TTL policy when design Pattition TTL Policy.

And here are the answer to your remaining question.

Why store the policy in hoodie.properties instead of using write config?

I think we need to store the policy in hudi somewhere otherwise user needs to store the policy themselves. Think when the user has 1000 partitions and wants to set TTL policies for 500 of them. It's better to store the policy metadata in hudi.

Hope for your another round of review!

[SteNicholas]

@stream2000, thanks for the update of this RFC. I left the comments for the updates.

[SteNicholas]

Could the partitionSpec support multiple level partition?

[stream2000]

Yes, it supports regex expressions for partitions and static partition value

[SteNicholas]

@stream2000, for example, there is a Hudi table partitioned by date and hour. Meanwhile, the user want to configure ttl with a year. How could user configure this ttl with current policy definition? Sets the policyValue with a year and partitionSpec with */*?

[stream2000]

Thanks for your review!

[SteNicholas]

If there is pending clustering and compaction, the ttl policy execution should clean and delete the expired file groups after clustering and compaction. The user doesn't need do any behavior for pending clustering and compaction case.

[SteNicholas]

@stream2000, do you have any updates?

[xicm]

Partition TTL can cover most scenarios, will bring a lot of convenience.
I have a simple record level TTL idea. We configure a general TTL, update the query api to ensure that expired data can't not be queried, and then delete the expired record when we do compaction or clustering.

[zyclove]

Can this feature be supported in advance? Especially hope for the next version after 0.14.

[stream2000]

Can this feature be supported in advance? Especially hope for the next version after 0.14.

I hope so, too! Will work on this feature in the few weeks and push a POC code to my personal repo ASAP.

[stream2000]

@geserdugarov Thanks very much for your review! I think the most important part of the design to be confirmed is whether we need to provide a feature-rich but complicated TTL policy in the first place or just implement a simple but extensible policy.

@geserdugarov @codope @nsivabalan @vinothchandar Hope for your opinion for this ~

[stream2000]

We can change the word TTL to lifecycle if we want to support other types of data expiration policies. In current version this RFC, we do not plan to support other types of policies like KEEP_BY_SIZE since they rely on the condition that the partition values are comparable which is not always true. However we do have some users who have this kind of requirement so we need to make the policy extensible.

[stream2000]

Thanks for your review~

it's just providing TTL functionality by custom implementation of PartitionTTLManagementStrategy is not user friendly.

For common users that want to do TTL management, they just need to add some write configs to their ingestion job. In the simplest situation they may just add the following config to enable async ttl management:

hoodie.partition.ttl.days.retain=10
hoodie.ttl.management.async.enabled=true

The TTL policy will default to KEEP_BY_TIME and we can automatically detect expired partitions by their last modified time and delete them.

However, this simple policy may not suit for all users, they may need to implement their own partition TTL policy so we provide an abstraction PartitionTTLManagementStrategy that returns expired partitions as users need and hudi will help delete them.

This is a main scope for TTL, and we shouldn't allow to have more flexibility.
Customized implementation of PartitionTTLManagementStrategy will allow to do anything with partitions. It still could be PartitionManagementStrategy, but then we shouldn't named it with TTL part.

Customized implementation of PartitionTTLManagementStrategy will only provide a list of expired partitions, and hudi will help delete them. We can rename the word PartitionTTLManagementStrategy to PartitionLifecycleManagementStrategy since we do not always delete the partitions by time.

[stream2000]

We should always think twice before making format changes. We have considered using HoodiePartitionMetadata to provide the lastModifiedTime, however, the current implementation of partition metadata does not support transactions, and will never be modified since the first creation. Not sure it is a good choice for maintaining the lastModifiedTime

[stream2000]

It will be a write config, and we will only apply TTL policy to the partition specified by this config. If this config is not specified, we will apply ttl policy to all the partitions in hudi table.

[stream2000]

I offer to allow user set only spec and TTL values with support of add, show, and delete operations.

So, if I want to implement 6 partition policies, then I should prepare 6 PartitionTTLManagementStrategys and call them 6 times?

We can implement the advanced TTL management policy in the future to provide the ability to set different policies for different partitions. To achieve this, we can introduce a new hoodie.partition.ttl.management.strategy and implement the logic you mentioned. For most users, providing a simple policy that TTL any partition whose last mod time (last time when data was added/updated), is greater than the TTL time specified will be enough.

@geserdugarov @nsivabalan What do you think about this?

[stream2000]

We can implement two independent ttl policies, one is simple while another is advanced as I commented below.

[stream2000]

Hi community, I have implemeted a poc version of partition lifecycle management in this PR: #9723. This PR provide a simple KEEP_BY_TIME strategy and can do the partition lifecycle management inline after commit. This POC implementation will help understand the design mentioned in this RFC. Hope for your review😄

[danny0405]

Guess you are talking about GDPR?

[danny0405]

To clarify: this is very different, the proposal for this doc only yields target partitons, it does not remove the whole partition.

[danny0405]

hoodie.partition.lifecycle.management.strategy -> hoodie.partition.ttl.strategy ?

[stream2000]

Use the word ttl means that we can't apply time unrelative partition deletion strategy in the future. However it is more suitable for current implementation since we only provide a time based strategy. What do you think? Should we use ttl directly or remain some extensibility?

[danny0405]

SHould we handle the cleaning duty just to the cleaner service? Like we do to the delete partition cmds, you just generate a delete partiton commit metadata on the timeline and the cleaner would take care of it.

[danny0405]

What I really care about is the conflict resolution for the operation, does OCC applies here? Do we have some support for the basic concurrent modification.

[stream2000]

Yes, we will use delete partiton command to delete the expired partitions, this is mentioned in Background section. And conflict resolution for the operation is the same with what we do in delete partiton command.

[zyclove]

Will version 1.0 support this feature? This feature is greatly needed.
@danny0405 @stream2000

[stream2000]

For 1.0.0 and later hudi version which supports efficient completion time queries on the timeline(#9565), we can get partition's lastModifiedTime by scanning the timeline and get the last write commit for the partition. Also for efficiency, we can store the partitions' last modified time and current completion time in the replace commit metadata. The next time we need to calculate the partitions' last modified time, we can build incrementally from the replace commit metadata of the last ttl management.

@danny0405 Added new lastModifiedTime calculation method for 1.0.0 and later hudi version. We plan to implement the file listing based lastModifiedTime at first and implement the timeline-based lastModifiedTime calculation in a separate PR. This will help users with earlier hudi versions easy to pick the function to their code base.

I have addressed all comments according to online/offline discussions. If there is no other concern, we can move on this~

[danny0405]

hoodie.partition.ttl.strategy.class ?

[stream2000]

Yes, sure. Will change it later.

[danny0405]

Do we need to pass around basic info like table path or meta client so that user can customize more easily? For example, how can a customized class knows the table path?

[stream2000]

see poc code: https://github.com/apache/hudi/pull/9723/files#diff-854a305c2cf1bbe8545bbe78866f53e84a489b60390ea1cfc7f64ed61165a0aaR33

Will provide a constructor like:

public PartitionTTLManagementStrategy(HoodieTable hoodieTable) {
    this.writeConfig = hoodieTable.getConfig();
    this.hoodieTable = hoodieTable;
 }

[danny0405]

Then make it clear in the pseudocode code.

[danny0405]

We will only support the feature in 1.0 right? Before 1.0, we do not have good manner to know when a log file is committed in a file slice and that might be a snag when calcurating the lastCommitTime.

[stream2000]

You're right. We should support this feature in 1.0 only to have an accurate lastCommitTime.

[danny0405]

Where does the hoodieTable. got instantiated?

[stream2000]

In the constructor of PartitionTTLManagementStrategy like

  protected final HoodieTable hoodieTable;
  protected final HoodieWriteConfig writeConfig;
  public PartitionTTLManagementStrategy(HoodieTable hoodieTable) {
    this.writeConfig = hoodieTable.getConfig();
    this.hoodieTable = hoodieTable;
  }

[danny0405]

Let's make the pseudocode clear.

[geserdugarov]

@geserdugarov Thanks very much for your review! I think the most important part of the design to be confirmed is whether we need to provide a feature-rich but complicated TTL policy in the first place or just implement a simple but extensible policy.

@geserdugarov @codope @nsivabalan @vinothchandar Hope for your opinion for this ~

Yes, you're right about the main difference in our ideas.

[stream2000]

@danny0405 @geserdugarov Thanks for your review! I have addressed all comments, hope for your opinions.

[stream2000]

In the constructor of PartitionTTLManagementStrategy like

  protected final HoodieTable hoodieTable;
  protected final HoodieWriteConfig writeConfig;
  public PartitionTTLManagementStrategy(HoodieTable hoodieTable) {
    this.writeConfig = hoodieTable.getConfig();
    this.hoodieTable = hoodieTable;
  }

[stream2000]

It's better to provide the hoodie.partition.ttl.strategy abtraction so that users can implement their own strategies and reuse the infrastructures provided by RFC-65 to delete expired partitions. For example, user wants to add a new strategy that delete partitions by their specified logic.

In current design, we will implement KEEP_BY_TIME only. However, it doesn't mean that we don't need any other kind of ttl strategy in the future.

[stream2000]

Yes, sure. Will change it later.

[stream2000]

see poc code: https://github.com/apache/hudi/pull/9723/files#diff-854a305c2cf1bbe8545bbe78866f53e84a489b60390ea1cfc7f64ed61165a0aaR33

Will provide a constructor like:

public PartitionTTLManagementStrategy(HoodieTable hoodieTable) {
    this.writeConfig = hoodieTable.getConfig();
    this.hoodieTable = hoodieTable;
 }

[stream2000]

Unfortunately, I don't see any other strategies except KEEP_BY_TIME for partition level TTL.

size: not applicable, could be trigger for lower levels

We need to make the design extensible and help users with their own expire logic easy to use. I think it's better to have a strategy-level abstraction.

[stream2000]

Again, leverage .hoodie_partition_metadata will bring format change, and it doesn't support any kind of transaction currently. As discussed with @danny0405 , In 1.0.0 and later version which supports efficient completion time queries on the timeline(#9565), we will have a more elegant way to get the lastCommitTime.
You can see the the updated RFC for details.

[stream2000]

You're right. We should support this feature in 1.0 only to have an accurate lastCommitTime.

[stream2000]

You can fire an RFC about your design of the wildcard spec version of TTL strategy, make it clear about how user should use this kind of strategy, the data structure of the strategy, how to store them(and problems about it), etc. We can have a discussion about issues of this kind strategy in that RFC.

And implement of this advanced strategy can be added in the future by implementing a new kind of ttl strategy.

I prefer to provide a simple strategy that we can configure with a few simple and easy-to-understand write configs. @danny0405 What do you think?

[stream2000]

We can remove this hoodie.partition.ttl.management.partition.selected after we implement the advance strategy in the future.

[stream2000]

You can fire an RFC about this, and we can have a discussion about your design in detail. And it's not conflict with the design in RFC-65.

[geserdugarov]

@danny0405 @geserdugarov Thanks for your review! I have addressed all comments, hope for your opinions.

@stream2000 Thank you for your reply. As you mentioned, I've wrote all my ideas in a separate RFC to simplify details discussion: #10248
@danny0405 @stream2000 Do you mind me asking for your comments?

[danny0405]

@geserdugarov I'm gonna merge this one first, can you fire your suggestion chagne PR incrementally based on this one? And we can move the discussions there.

[geserdugarov]

@geserdugarov I'm gonna merge this one first, can you fire your suggestion chagne PR incrementally based on this one? And we can move the discussions there.

@danny0405 Ok, I will work on it.