[HUDI-5176] Fix incremental source to consider inflight commits before completed commits #7160
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
nsivabalan
reviewed
Nov 8, 2022
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/IncrSourceHelper.java
Show resolved
Hide resolved
…e completed commits
…nd pending table services
yihua
force-pushed
the
HUDI-5176-fix-incr-source-inflights
branch
from
e533073 to
2afa21a
Compare
nsivabalan
approved these changes
Nov 10, 2022
satishkotha
pushed a commit
to satishkotha/incubator-hudi
that referenced
this pull request
Dec 12, 2022
…e completed commits (apache#7160) When multiple writers are writing to the same Hudi table under optimistic concurrency control, one writer may start later and finish the write transaction earlier than the other concurrent writer. In this case, the Hudi timeline contains an inflight commit before a completed commit. As a concrete example, writer 1 starts a commit at t1 and later writer 2 starts another commit at t2 (t2 > t1). Commit t2 finishes earlier than t1. t3 t4 ---------------------------------------------------------> t instant t1 |------------------------------| (writer 1) instant t2 |--------------| (writer 2) When Hudi incremental source is used on such a Hudi table, the logic to determine the next fetch (IncrSourceHelper.calculateBeginAndEndInstants()) is susceptible to data loss, missing the inflight instants. The method filters for the completed instants only and uses that to determine the instants to fetch, thus the checkpoint. For the same example, when incrementally fetching the data at t3, the checkpoint advances to t2 and ignores t1 indefinitely. To fix the problem, this PR adds the logic to only look at contiguous completed commits / deltacommits / non-clustering replacecommits for incremental pulls, e.g., completed instants before t1 in the example, so that the checkpoint never goes over the inflight commits which can cause data loss. To implement this, we figure out the earliest incomplete commit, deltacommit, or non-clustering replacecommit to bound the incremental pulls. This PR fixes the HoodieIncrSource, S3EventsHoodieIncrSource, and GcsEventsHoodieIncrSource for the same issue.
alexeykudinkin
pushed a commit
to onehouseinc/hudi
that referenced
this pull request
Dec 14, 2022
…e completed commits (apache#7160) When multiple writers are writing to the same Hudi table under optimistic concurrency control, one writer may start later and finish the write transaction earlier than the other concurrent writer. In this case, the Hudi timeline contains an inflight commit before a completed commit. As a concrete example, writer 1 starts a commit at t1 and later writer 2 starts another commit at t2 (t2 > t1). Commit t2 finishes earlier than t1. t3 t4 ---------------------------------------------------------> t instant t1 |------------------------------| (writer 1) instant t2 |--------------| (writer 2) When Hudi incremental source is used on such a Hudi table, the logic to determine the next fetch (IncrSourceHelper.calculateBeginAndEndInstants()) is susceptible to data loss, missing the inflight instants. The method filters for the completed instants only and uses that to determine the instants to fetch, thus the checkpoint. For the same example, when incrementally fetching the data at t3, the checkpoint advances to t2 and ignores t1 indefinitely. To fix the problem, this PR adds the logic to only look at contiguous completed commits / deltacommits / non-clustering replacecommits for incremental pulls, e.g., completed instants before t1 in the example, so that the checkpoint never goes over the inflight commits which can cause data loss. To implement this, we figure out the earliest incomplete commit, deltacommit, or non-clustering replacecommit to bound the incremental pulls. This PR fixes the HoodieIncrSource, S3EventsHoodieIncrSource, and GcsEventsHoodieIncrSource for the same issue.
alexeykudinkin
pushed a commit
to onehouseinc/hudi
that referenced
this pull request
Dec 14, 2022
…e completed commits (apache#7160) When multiple writers are writing to the same Hudi table under optimistic concurrency control, one writer may start later and finish the write transaction earlier than the other concurrent writer. In this case, the Hudi timeline contains an inflight commit before a completed commit. As a concrete example, writer 1 starts a commit at t1 and later writer 2 starts another commit at t2 (t2 > t1). Commit t2 finishes earlier than t1. t3 t4 ---------------------------------------------------------> t instant t1 |------------------------------| (writer 1) instant t2 |--------------| (writer 2) When Hudi incremental source is used on such a Hudi table, the logic to determine the next fetch (IncrSourceHelper.calculateBeginAndEndInstants()) is susceptible to data loss, missing the inflight instants. The method filters for the completed instants only and uses that to determine the instants to fetch, thus the checkpoint. For the same example, when incrementally fetching the data at t3, the checkpoint advances to t2 and ignores t1 indefinitely. To fix the problem, this PR adds the logic to only look at contiguous completed commits / deltacommits / non-clustering replacecommits for incremental pulls, e.g., completed instants before t1 in the example, so that the checkpoint never goes over the inflight commits which can cause data loss. To implement this, we figure out the earliest incomplete commit, deltacommit, or non-clustering replacecommit to bound the incremental pulls. This PR fixes the HoodieIncrSource, S3EventsHoodieIncrSource, and GcsEventsHoodieIncrSource for the same issue.
alexeykudinkin
pushed a commit
to onehouseinc/hudi
that referenced
this pull request
Dec 14, 2022
…e completed commits (apache#7160) When multiple writers are writing to the same Hudi table under optimistic concurrency control, one writer may start later and finish the write transaction earlier than the other concurrent writer. In this case, the Hudi timeline contains an inflight commit before a completed commit. As a concrete example, writer 1 starts a commit at t1 and later writer 2 starts another commit at t2 (t2 > t1). Commit t2 finishes earlier than t1. t3 t4 ---------------------------------------------------------> t instant t1 |------------------------------| (writer 1) instant t2 |--------------| (writer 2) When Hudi incremental source is used on such a Hudi table, the logic to determine the next fetch (IncrSourceHelper.calculateBeginAndEndInstants()) is susceptible to data loss, missing the inflight instants. The method filters for the completed instants only and uses that to determine the instants to fetch, thus the checkpoint. For the same example, when incrementally fetching the data at t3, the checkpoint advances to t2 and ignores t1 indefinitely. To fix the problem, this PR adds the logic to only look at contiguous completed commits / deltacommits / non-clustering replacecommits for incremental pulls, e.g., completed instants before t1 in the example, so that the checkpoint never goes over the inflight commits which can cause data loss. To implement this, we figure out the earliest incomplete commit, deltacommit, or non-clustering replacecommit to bound the incremental pulls. This PR fixes the HoodieIncrSource, S3EventsHoodieIncrSource, and GcsEventsHoodieIncrSource for the same issue.
alexeykudinkin
pushed a commit
to onehouseinc/hudi
that referenced
this pull request
Dec 14, 2022
…e completed commits (apache#7160) When multiple writers are writing to the same Hudi table under optimistic concurrency control, one writer may start later and finish the write transaction earlier than the other concurrent writer. In this case, the Hudi timeline contains an inflight commit before a completed commit. As a concrete example, writer 1 starts a commit at t1 and later writer 2 starts another commit at t2 (t2 > t1). Commit t2 finishes earlier than t1. t3 t4 ---------------------------------------------------------> t instant t1 |------------------------------| (writer 1) instant t2 |--------------| (writer 2) When Hudi incremental source is used on such a Hudi table, the logic to determine the next fetch (IncrSourceHelper.calculateBeginAndEndInstants()) is susceptible to data loss, missing the inflight instants. The method filters for the completed instants only and uses that to determine the instants to fetch, thus the checkpoint. For the same example, when incrementally fetching the data at t3, the checkpoint advances to t2 and ignores t1 indefinitely. To fix the problem, this PR adds the logic to only look at contiguous completed commits / deltacommits / non-clustering replacecommits for incremental pulls, e.g., completed instants before t1 in the example, so that the checkpoint never goes over the inflight commits which can cause data loss. To implement this, we figure out the earliest incomplete commit, deltacommit, or non-clustering replacecommit to bound the incremental pulls. This PR fixes the HoodieIncrSource, S3EventsHoodieIncrSource, and GcsEventsHoodieIncrSource for the same issue.
fengjian428
pushed a commit
to fengjian428/hudi
that referenced
this pull request
Apr 5, 2023
…e completed commits (apache#7160) When multiple writers are writing to the same Hudi table under optimistic concurrency control, one writer may start later and finish the write transaction earlier than the other concurrent writer. In this case, the Hudi timeline contains an inflight commit before a completed commit. As a concrete example, writer 1 starts a commit at t1 and later writer 2 starts another commit at t2 (t2 > t1). Commit t2 finishes earlier than t1. t3 t4 ---------------------------------------------------------> t instant t1 |------------------------------| (writer 1) instant t2 |--------------| (writer 2) When Hudi incremental source is used on such a Hudi table, the logic to determine the next fetch (IncrSourceHelper.calculateBeginAndEndInstants()) is susceptible to data loss, missing the inflight instants. The method filters for the completed instants only and uses that to determine the instants to fetch, thus the checkpoint. For the same example, when incrementally fetching the data at t3, the checkpoint advances to t2 and ignores t1 indefinitely. To fix the problem, this PR adds the logic to only look at contiguous completed commits / deltacommits / non-clustering replacecommits for incremental pulls, e.g., completed instants before t1 in the example, so that the checkpoint never goes over the inflight commits which can cause data loss. To implement this, we figure out the earliest incomplete commit, deltacommit, or non-clustering replacecommit to bound the incremental pulls. This PR fixes the HoodieIncrSource, S3EventsHoodieIncrSource, and GcsEventsHoodieIncrSource for the same issue.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Change Logs
When multiple writers are writing to the same Hudi table under optimistic concurrency control, one writer may start later and finish the write transaction earlier than the other concurrent writer. In this case, the Hudi timeline contains an inflight commit before a completed commit. As a concrete example, writer 1 starts a commit at t1 and later writer 2 starts another commit at t2 (t2 > t1). Commit t2 finishes earlier than t1.
When Hudi incremental source is used on such a Hudi table, the logic to determine the next fetch (
IncrSourceHelper.calculateBeginAndEndInstants()) is susceptible to data loss, missing the inflight instants. The method filters for the completed instants only and uses that to determine the instants to fetch, thus the checkpoint. For the same example, when incrementally fetching the data at t3, the checkpoint advances to t2 and ignores t1 indefinitely.To fix the problem, this PR adds the logic to only look at contiguous completed commits / deltacommits / non-clustering replacecommits for incremental pulls, e.g., completed instants before t1 in the example, so that the checkpoint never goes over the inflight commits which can cause data loss. To implement this, we figure out the earliest incomplete commit, deltacommit, or non-clustering replacecommit to bound the incremental pulls.
This PR fixes the
HoodieIncrSource,S3EventsHoodieIncrSource, andGcsEventsHoodieIncrSourcefor the same issue.Impact
This PR avoids missing instants for incremental pulls for the scenarios mentioned above for multi-writer.
Risk level
medium
This PR adds two new tests,
TestHoodieIncrSource#testHoodieIncrSourceInflightCommitBeforeCompletedCommitandTestHoodieIncrSource#testHoodieIncrSourceWithPendingTableServices, to verify the expected behavior. Before this PR, the first test failed.Documentation Update
N/A
Contributor's checklist