[RFC-15][HUDI-1325] Merge updates of unsynced instants to metadata table

[rmpifer]

Tips

• Thank you very much for contributing to Apache Hudi.
• Please review https://hudi.apache.org/contributing.html before opening a pull request.

What is the purpose of the pull request

Currently every time an operation (i.e. commit, clean, rollback) completes we attempt to keep the metadata table up to date by creating an upsert on the metadata table with the same instant time. This way we can reference both datasets timelines and directly compare the instants on them to see if they are in sync.

There can be the possibility that the dataset timeline and the metadata table timeline become out of sync. When trying to read from the metadata table while the timeline is out of sync you would get incorrect values for getAllFilesInPartition and getAllPartitionPaths.

This change provides a way to overcome this scenario by reading unsynced timeline instants and merging it with existing metadata table records to get the most up to date state of the file system

JIRA: https://issues.apache.org/jira/browse/HUDI-1325

Brief change log

Timeline Sync

• The logic of converting timeline metadata to metadata table records was directly tied to the commit phase in HoodieBackedMetadataWriter. Refactored this logic to a utility class HoodieTableMetadataTimelineUtil
• Created a scanner HoodieMetadataMergedInstantRecordScanner which handles conversion of timeline instants to metadata records and merges results
• Added final step in AbstractHoodieTableMetadata.getMergedRecordByKey which uses the new scanner mentioned to fetch the HoodieRecord associated with the desired key from the unsynced timeline instants and merge it with the record from the metadata table
• When converting rollback operation to metadata table records there was logic that re-read from the metadata table to ensure any files being deleted as part of roll back existed.

// Rollbacks deletes instants from timeline. The instant being rolled-back may not have been synced to the
// metadata table. Hence, the deleted filed need to be checked against the metadata.

This doesn't make sense since all instants are processed in serial order so there would never be the case where a rollback was being written before an instant earlier on the timeline was already synced. Removed this logic because it created circular dependency when implementing timeline merging

• Changed the validate metadata step in tests to use the metadata reader HoodieBackedTableMetadata. By default when metadata writer HoodieBackedTableMetadataWriter is initialized it syncs all instants to the metadata table. By using the reader we can simulate metadata table being out of sync.
• Modified initMetaClient in test base class to allow table type to be passed in since table type is always set as COPY_ON_WRITE if using this method to initialize the meta client

Refactor

For the following reasons I modified the HoodieTableMetadata interface to be an AbstractHoodieMetadata class which contains the following shared functionality irresepective of how the metadata is stored (as a Hoodie table, in some key/value store):

• Fetching getAllPartitions and getAllFileInPartition from metadata should validate metadata flag is enabled and should default to file listing if any error occurs regardless of storage type
• During fetchAllPartitions and fetchAllFilesInPartition metrics should be published on operations and if validateLookUps is enabled, results returned from metadata should be compared against actual file listing results regardless of what storage type is
• In getMergedRecordByKey regardless of how the key is fetched from storage the result is merged against unsynced timeline instants. This is why I introduced abstract getRecordByKeyFromMetadata which can be implemented by inheriting class which contains logic specific to storage type.
• Moved findInstantsToSync to AbstractHoodieTableMetadata. This method is needed regardless of storage type to find if metadata is in sync with timeline. However since how the last synced instant is stored can be dependent on storage type made this method abstract.

Verify this pull request

(Please pick either of the following options)

This pull request is a trivial rework / code cleanup without any test coverage.

(or)

This pull request is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

Testing

Ran TestHoodieBackedMetadata which contains all tests related to metadata table

Added tests which use an unsynced client after commits, cleans, and restores have been performed to ensure updates not written to the metadata yet are still being reflected when reading from the metadata

Committer checklist

• Has a corresponding JIRA in PR title & commit

• Commit message is descriptive of the change

• CI is green

• Necessary doc changes done or have another open PR

• For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.

[vinothchandar]

@rmpifer could you please rebase this against latest rfc-15 branch. I ll get started with the review in the meantime

[vinothchandar]

Two small structural comments.

I am still mulling if it's alright to just implement this at the scanner level, and not have a wrapper HoodieTableMetadata implementation around it.

[vinothchandar]

I would prefer to still have an interface for HoodieTableMetadata and return that here. We can have the abstract class internal to the actual implementation?

[rmpifer]

I maintained this interface in revision after rebasing and then built AbstractHoodieTableMetadata on top of this.

[vinothchandar]

I think we can simply call this HoodieTableMetadataUtils ?

[rmpifer]

Updated this

[vinothchandar]

@rmpifer On this point specifically,

so there would never be the case where a rollback was being written before an instant earlier on the timeline was already synced. Removed this logic because it created circular dependency when implementing timeline merging

Following situation is technically possible. Timeline has commits c1, c2, c3, c4 and we have synced only upto C2. User decides to restore the table to commit time c2, thus generating rollback metadata for c3, c4 which now have no files in the metadata table. I think we should defensively handle this scenario at the payload merge level i.e if there is a delete for a file that does not exist in the base file or prior instants, we should just do a no-op and continue. What I am saying is : I agree with you that the extra logic need not be present, but we should check the payload once to ensure we can handle the scenario above.

Let me know if we are not talking about the same thing :)

[vinothchandar]

LGTM overall. Comments on structure, corner case, testing

I think we can stick with the scanner approach. its more elegant than what I had in mind i.e the wrapped HoodieTableMetadata impl

[vinothchandar]

I think we can keep the factory method in the interface?

[rmpifer]

Along with Prashant's comment we can just create HoodieBackedTableMetadata since HoodieBackedTableMetadaWriter should only be associated with this implementation

[vinothchandar]

left a comment around this scenario. there is one valid case here. good call out in the summary

[vinothchandar]

just return out of here, instead of reassigning to another important variable. may be easier to read.

[rmpifer]

I created separate variable for this so there is still one place of return

[vinothchandar]

Sorry, my core concern was that we are reassigning to a variable that holds the returned value from a key method call getRecordByKeyFromMetadata

I was suggesting something like.

    Option<HoodieRecord<HoodieMetadataPayload>> metadataHoodieRecord = getRecordByKeyFromMetadata(key);
    // Retrieve record from unsynced timeline instants
    Option<HoodieRecord<HoodieMetadataPayload>> timelineHoodieRecord = timelineRecordScanner.getRecordByKey(key);
    if (timelineHoodieRecord.isPresent()) {
      if (metadataHoodieRecord.isPresent()) {
        HoodieRecordPayload mergedPayload = timelineHoodieRecord.get().getData().preCombine(metadataHoodieRecord.get().getData());
        return Option.of(new HoodieRecord(metadataHoodieRecord.get().getKey(), mergedPayload));
      } else {
        return timelineHoodieRecord;
      }
    }

    return metadataHoodieRecord;

if you feel strongly, we can keep as-is.

[rmpifer]

Sorry my comment was unclear. I created separate mergedRecord variable so I am not overriding metadataHoodieRecord

[rmpifer]

@rmpifer On this point specifically,

so there would never be the case where a rollback was being written before an instant earlier on the timeline was already synced. Removed this logic because it created circular dependency when implementing timeline merging

Following situation is technically possible. Timeline has commits c1, c2, c3, c4 and we have synced only upto C2. User decides to restore the table to commit time c2, thus generating rollback metadata for c3, c4 which now have no files in the metadata table. I think we should defensively handle this scenario at the payload merge level i.e if there is a delete for a file that does not exist in the base file or prior instants, we should just do a no-op and continue. What I am saying is : I agree with you that the extra logic need not be present, but we should check the payload once to ensure we can handle the scenario above.

Let me know if we are not talking about the same thing :)

If this is the case what happens when c3 and c4 are then finally synced? These files would remain in the metadata table even though rollback should have removed them

[prashantwason]

if there is a delete for a file that does not exist in the base file or prior instants, we should just do a no-op and continue.
How do we differentiate this from a bug which lead to files missing from the metadata listing?

The current merge code provides some sort of correctness - deletes are always for files already created. I think we should maintain this.

[prashantwason]

can be merged with the above function. I dont see it being called anywhere else.

[rmpifer]

Updated

[prashantwason]

The reader-write pairs in metadata are related. So this can be simplified to be simply new HoodieBackedTableMetadata.

HoodieBackedTableMetadata cannot work with any other implementaton of HoodieTableMetadata.

[rmpifer]

Agreed. Updated this

[vinothchandar]

If this is the case what happens when c3 and c4 are then finally synced? These files would remain in the metadata table even though rollback should have removed them

So restore in that example, would delete C3, C4 from the timeline. it won't be synced at all.

How do we differentiate this from a bug which lead to files missing from the metadata listing? The current merge code provides some sort of correctness - deletes are always for files already created. I think we should maintain this.

I understand the need for wanting to fail-fast. but the scenario above is legit, right? Do you suggest we sync the rollbacks only if the original instants for which the rollback was done were synced to the metadata table? cc @prashantwason
Let's make the call on this one. Otherwise PR seems ready more or less, after another rebase/set of comments. ?

[rmpifer]

If this is the case what happens when c3 and c4 are then finally synced? These files would remain in the metadata table even though rollback should have removed them

So restore in that example, would delete C3, C4 from the timeline. it won't be synced at all.

How do we differentiate this from a bug which lead to files missing from the metadata listing? The current merge code provides some sort of correctness - deletes are always for files already created. I think we should maintain this.

I understand the need for wanting to fail-fast. but the scenario above is legit, right? Do you suggest we sync the rollbacks only if the original instants for which the rollback was done were synced to the metadata table? cc @prashantwason
Let's make the call on this one. Otherwise PR seems ready more or less, after another rebase/set of comments. ?

If this is the case for restore, that they are removed from the dataset timeline, then having this logic makes sense to me.

If we would like to maintain the requirement that deletes are only for files already created I can add this filtering logic. I like the approach vinoth mentions of rather than filtering at file level, check to see whether instant specified in rollback/restore was synced and skip if not. Let me know your thoughts

[rmpifer]

Along with Prashant's comment we can just create HoodieBackedTableMetadata since HoodieBackedTableMetadaWriter should only be associated with this implementation

[rmpifer]

Agreed. Updated this

[rmpifer]

Sorry my comment was unclear. I created separate mergedRecord variable so I am not overriding metadataHoodieRecord

[rmpifer]

Updated

[vinothchandar]

Do you suggest we sync the rollbacks only if the original instants for which the rollback was done were synced to the metadata table?

Thinking out loud whether there are tricky cases here. Specifically, is there a chance of skipping a rollback instant incorrectly.
Let's say we restored the last 11 commits (12 -> 2) with metadata synced upto C8

data timeline : C1,  rollback C12, rollback C11, ..... rollback C2
metadata timeline : C1, C2, ... C8

data timeline archival does not move past C8, but can technically archive C8. Similarly metadata timeline is free to archive independently and lets say it archives upto C2. If it ends up looking like

data timeline : C1,  rollback C12, rollback C11, ..... rollback C2
metadata timeline: C3, ..., C8 

It's now hard to determine whether or not rollback C2 should be applied. (we could compare the latest instant on metadata timeline C8 and assume its applied?)

I think this complexity is unwarranted for sake of failing-fast. We are better of defensively, ignoring deletes for files that were not logged before.

@prashantwason Hope you can be convinced of this :)

[rmpifer]

data timeline : C1,  rollback C12, rollback C11, ..... rollback C2
metadata timeline: C3, ..., C8 

It's now hard to determine whether or not rollback C2 should be applied. (we could compare the latest instant on metadata timeline C8 and assume its applied?)

Regarding this scenario, my thoughts would be to check rollback metadata only against unsynced instants, i.e. if any instants after C8 were considered in rollback. Is this potentially surfacing a more general issue though. Say metadata timeline has been synced up to C2, but data timeline archival has happened to C8. It seems next time we try to find unsynced instants we won't consider anything between C2...C8

[vinothchandar]

Regarding this scenario, my thoughts would be to check rollback metadata only against unsynced instants, i.e. if any instants after C8 were considered in rollback.

To be clear, we want to skip syncing a rollback instant to metadata table, if the instant it rolled back is not in completed state on the unsynced portion of the data timeline.? Thats what I meant by (we could compare the latest instant on metadata timeline C8 and assume its applied?). I think it will work out fine. Let's implement this?

Say metadata timeline has been synced up to C2, but data timeline archival has happened to C8.

We have code to guard against this already. Data timeline archiving won't move past latest instant on metadata timeline

hudi/hudi-client/src/main/java/org/apache/hudi/table/HoodieTimelineArchiveLog.java

Line 203 in f7f8c33

// If metadata table is enabled, do not archive instants which are more recent that the latest synced

[vinothchandar]

@rmpifer @prashantwason lets just implement the check as discussed above and move on. We can file a follow-on JIRA if needed.
I would like merge the feature branch to master asap, to give sometime for testing

[rmpifer]

Apologies for the delay in pushing changes, was offline over the holidays :). Updated to address feedback and to resolve our discussions to check whether the commit being rolled back is after last sync point on metadata timeline

[vinothchandar]

@rmpifer no worries. will review today and land

[vinothchandar]

LGTM