[HUDI-1575][RFC-56] Early Conflict Detection For Multi-writer

[zhangyue19921010]

Tips

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

What is the purpose of the pull request

(For example: This pull request adds quick-start document.)

Brief change log

(for example:)

• Modify AnnotationLocation checkstyle rule in checkstyle.xml

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:

(example:)

• Added integration tests for end-to-end.
• Added HoodieClientWriteTest to verify the change.
• Manually verified the change by running a job locally.

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.

[zhangyue19921010]

@hudi-bot run azure

[hudi-bot]

CI report:

• 6069fdc Azure: FAILURE

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build

[yihua]

At high-level, the approach LGTM. Let’s finalize the details in a few spots.

[yihua]

nit: remove redundant empty lines

[zhangyue19921010]

Sure, removed.

[yihua]

Let's spell out OCC: optimized occ mechanism -> OCC (Optimistic Concurrency Control)

[zhangyue19921010]

changed.

[yihua]

This new mechanism will do conflict checking before the writers creating markers and before starting to write corresponding data file.

This only applies to the direct markers in the new design. To be accurate, for the timeline-server based markers, based on the design, the conflicts are periodically checked and both writers may still write the data files of the same file slice, until the conflict is detected in the next round of checking.

[zhangyue19921010]

nice catch! changed.

[yihua]

Similar here for inaccurate description.

[zhangyue19921010]

Changed.

[yihua]

You can give an example of how the direct marker path is constructed for checking whether it exists, instead of listing all markers to save FS calls.

[zhangyue19921010]

Added.

[yihua]

Does the timeline server batch process the check-marker-conflict as well? Why not using the existing marker creation requests so that the timeline server returns false instead of true as the response? In that case, no new endpoint is introduced.

[zhangyue19921010]

Nice catch Ethan. I'd like more discuss here.
A separate api is added here for better abstraction. Because the follow-up may not only support async checking, but also support loop polling based on each writer's time line server. And coupling checking and create together may lose some scalability.

[yihua]

+1. This logic can be generalized for early conflict detection and pre-commit conflict detection.

[yihua]

Great to have the benchmarking numbers here! Let’s also add the setup, e.g., how many concurrent writers, the file system type, etc. and discuss the impact on the latency of ongoing requests.

[zhangyue19921010]

Added.

[yihua]

More details should be added here around how the transaction is ended. Does the driver or executor throw the exception? How to gracefully end the current write operation and possibly retry it in the same Spark job with DeltaStreamer?

[zhangyue19921010]

Added

[pratyakshsharma]

nit: "If this detection was failed" -> "if a conflict is detected"

I guess this makes it more clear.

[zhangyue19921010]

Changed. Thanks a lot for your review and attention :)

[pratyakshsharma]

minor comments. Thank you for writing this.

[pratyakshsharma]

Let us wait to see if @yihua has any more concerns.

[nbalajee]

What are the assumptions about the workload (batch of records being ingested), when the early conflict detection strategy is employed on two are more jobs? In other words, if two jobs A and B concurrently working on the hudi dataset, are they expected to perform "upsert/update" operations only?

Early conflict detection may not be able to catch conflicts in some scenarios:
a) Both jobs A and B are performing an "insert" operation (seeing the record for the first time) and same record key is present on both batches. A might chose fileId F1 and B might choose fileid F2 and no conflict will be flagged. (If all are inserts, with no dupes, there is no need for conflict resolution).
b) Job A received an insert for a record R1 and Job B received an update for record R1. Both A and B might map R1 to different fileIds and conflict won't be detected.
c) If both jobs A and B are not "ingestion" jobs (say A is ingesting to partition P1, B is clustering on partition P1), then early conflict detection strategy will work.

[zhangyue19921010]

Thanks @nbalajee for your attention.

What are the assumptions about the workload (batch of records being ingested), when the early conflict detection strategy is employed on two are more jobs? In other words, if two jobs A and B concurrently working on the hudi dataset, are they expected to perform "upsert/update" operations only?

I had to say, the key point for early conflict detection is to detect if multi-writers modify the same file group or not.

Current occ-based detection can only detect this conflict(which is file group level not record level as you mentioned a), b) and c)) at the end of ingestion.

This new workflow is trying to detected this kind of conflict and fail jobs as early as possible.

[yihua]

@nbalajee Thanks for bringing this up. We can add the scope of conflict detection, i.e., at file group level, to the RFC for clarification.

[yihua]

@nbalajee Thanks for bringing this up. We can add the scope of conflict detection, i.e., at file group level, to the RFC for clarification.

[yihua]

LGTM. @zhangyue19921010 Thanks for putting this up and taking care of all design details! The early conflict detection is going to be a good optimization and resource saver for multi-writer scenarios.

I helped you address a few minor writing issues in this RFC.

[yihua]

We should think about how to abort the Hudi write transaction as a whole if one executor fails to create a marker due to early conflict detection, instead of relying on Spark's failure retries.

[zhangyue19921010]

Sure, Ethan! For now when hudi detected a conflict, it will throw a HoodieEarlyConflictDetectionException on executor ==> current task failed directly

If we want to control this failed ourselves, there is the way i am thinking:

we implement a SparkListener for SparkListenerTaskEnd and register to spark context. If there is any task failed due to HoodieEarlyConflictDetectionException then failed current write transaction

[yihua]

We can do sth like this and take it as a follow-up. Have you measured how much additional time this can save, i.e., how much time do retries take?

[zhangyue19921010]

If there is a conflict and users use AsyncTimelineMarkerEarlyConflictDetectionStrategy, it will take about 150 ms to detect this conflict and fail current task.

spark.task.maxFailures with default value 4. It may takes 4* 150 = 600ms to fail current write transaction

[yihua]

Sg. Let's follow up with proper benchmarking on this. We should show how this contributes to the overall write latency of a transaction. Spark retries the whole job/stage which contains multiple tasks. Considering Spark's overhead, it can take some time for all tasks to fail and retry.

[yihua]

@prasannarajaperumal could you take another look at this RFC before we merge it? I'm going to move on to reviewing the implementation in #6133 .

[prasannarajaperumal]

I feel this is a critical improvement for multiple writers. Thanks for writing this up.
Looks good overall. Few comments.

[prasannarajaperumal]

Might make sense to put a probabilistic data structure (bloom or cuckoo) to frontend the check and speed up checking marker conflicts with all active writers especially for File based markers.

[zhangyue19921010]

Nice Idea! If I understand correctly, we can implement another strategy:

1. When write handler create marker it will create or refresh a bloom index file for current writer.
2. When early conflict detect works it will check this bloom index file first to speed up this check.

[prasannarajaperumal]

Yeah - I think lets bring this as a default into the design. This should speed up early OCC check when there are no conflicts which is really the case we want to optimize for.

[zhangyue19921010]

Add new strategies named BloomDirectMarkerBasedEarlyConflictDetectionStrategy and BloomAsyncTimelineMarkerEarlyConflictDetectionStrategy which design to read pre-created marker bloom files firstly to pick out potentially conflicting marker files as quick as possible.

[yihua]

Given that the timeline-server-based markers are the default now, should we rely more on the timeline-server-based early conflict detection for better performance and make the Bloom filter based marker conflict detection as a nice-to-have in the first cut?

[yihua]

For the Bloom filter based marker conflict detection, given we're writing additional metadata beyond just the markers, we need to design it properly. That's why I'm thinking we can have it in the second cut.

[zhangyue19921010]

Given that the timeline-server-based markers are the default now, should we rely more on the timeline-server-based early conflict detection for better performance and make the Bloom filter based marker conflict detection as a nice-to-have in the first cut?

+1

Because of hoodie using timeline-server-based markers as default. And there only 20(MARKER0~MARKER19) marker files under .temp, so maybe this bloom filters are not a very urgent blocker which need more careful design on both marker creator and marker reader :)

[prasannarajaperumal]

It is not clear what this means to me. For every single write - Do we not have to check with all the existing markers by all the active writers to detect a conflict?.

It makes sense to have some ordering to break ties between concurrent writes trying to create markers - but that is not what we are doing here though if I understand this right.

[zhangyue19921010]

The design here is to prevent two writers from detecting each other and all failing both.
For example writer1 start at 9:00 and writer2 start at 10:00 which conflict each other.

Without this limit, writer1 detect conflict with writer2 and failed. On the other hand writer2 also detect this conflict and failed either.

With this limit, only writer2 failed.

[zhangyue19921010]

Also we can design a priority strategy as follow up :)

[yihua]

@prasannarajaperumal I think what @zhangyue19921010 means here is that if two writers have a conflict, one of the writers can succeed and the other fails. This is consistent with the current pre-commit conflict detection, i.e., only the commits in the Hudi timeline snapshot captured at the beginning of the write transaction are going to be checked for conflict. If writer2 starts in the middle of the writer1's transaction, at pre-commit time, writer1 does not see and check the writer2's inflight commit even though it is in the Hudi timeline (because at the beginning of writer1's transaction, the snapshot of the timeline does not have the writer2's commit).

[prasannarajaperumal]

Is this what you mean at time t4?
"at time t4 - writer2 tries to create marker for a file already updated by writer1"

[zhangyue19921010]

Yeap => at time t4 - writer2 tries to create marker for a file fileA which already updated by writer1 to fileAa(updated)

[prasannarajaperumal]

Can you update the RFC to word it this way for more clarity? Thanks

[zhangyue19921010]

added!

[yihua]

+1

[prasannarajaperumal]

Naming is a bit confusing as set1 used here and set1 used in Figure 2 is not the same - but readers may think it is :)
We could name it functionally like committed_instants_before and committed_instants_after and the diff is committed_instants_during.

[zhangyue19921010]

Nice catch! Changed!

[yihua]

+1

[prasannarajaperumal]

also since OCC Early detection is pure optimization play - Can we skip on these complex logic and keep the design simple. Worst case there are some false positives that could go through till pre-commit check.

Personally, I would rather have OCC early detection logic very simple and catch 95% of the cases than being very complicated and catch 99% of the cases.

[zhangyue19921010]

Maybe we can set a functional flag default as false.
For common users they don't need to care about this flag.
For users who are more sensitive to this early conflict detect, they can enable this hard check.
Hi @yihua what do you think here :)

[prasannarajaperumal]

Thats sounds okay to me.

[yihua]

Without checking the metadata of completed commits from other concurrent writers, we miss the type of conflicts happening between t3 and t10 in the case below, which I think happens much more than 5% of the cases if there are conflicts, and lead to resource waste. For timeline-server-based markers, since the timeline server is going to do such detection asynchronously, it should not add latency to the hot path of writer data files.

writer 1: t1-----t3
writer 2:     t2---------------<conflict>------------------t10

IMO, given the commit metadata check is sth pre-commit conflict detection also does, we can reuse the same logic and generalize it for the early conflict detection as well. I agree that we can add a feature flag for such logic, but we can keep the default as true for timeline-server-based markers and false for direct markers. If the user really sees performance issues, they can turn it off. What do you guys think?

[zhangyue19921010]

LGTM !

[zhangyue19921010]

Hi @yihua and @pratyakshsharma . Really appreciate for your attention here! Address the comments. PTAL :)

[yihua]

@zhangyue19921010 Thanks for addressing the comments. Why are some code changes involved? Could you properly rebase this PR on the latest master?

[zhangyue19921010]

@zhangyue19921010 Thanks for addressing the comments. Why are some code changes involved? Could you properly rebase this PR on the latest master?

ooooh. Sorry for that. Just rebased the master branch. PTAL

[prasannarajaperumal]

Looks good to land the RFC. Thanks for taking care of my comments.

[zhangyue19921010]

Thanks all for your efforts here!