You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "ad1happy2go (via GitHub)" <gi...@apache.org> on 2026/10/09 18:59:47 UTC
[PR] fix(common): list all partitions in fallback incremental reads when a stale non-write instant precedes the first commit [hudi]
ad1happy2go opened a new pull request, #20273:
URL: https://github.com/apache/hudi/pull/20273
### Describe the issue this Pull Request addresses
An incremental query with `hoodie.datasource.read.incr.fallback.fulltablescan.enable=true` whose start instant is archived silently returns **no rows**, or only some partitions' rows, when a completed non-write instant (rollback or clean) older than every active commit is still on the active timeline.
Two components decide whether the start is archived, and they look at different timelines:
- **The relation** (`MergeOnReadIncrementalRelationV1.startInstantArchived`, and `V2` via `queryContext.getArchivedInstants`) checks the **commit timeline**. It sees the start as archived and falls back to a full table scan.
- **The file index** (`BaseHoodieTableFileIndex.isBeforeTimelineStarts`) checks the **whole active timeline**. The older rollback makes it treat the start as active, so it takes the incremental branch of `listPartitionPaths`. That branch lists only the partitions written by `findInstantsInRange()` on the write timeline.
The log shows `listing full table (0 partitions)` and the scan returns nothing, or only the partitions touched by active commits in the range.
This state is ordinary: the table version 6 archiver only archives rollback instants once there are more than `hoodie.keep.max.commits` of them, so a single rollback from a failed write outlives the commits around it. It also stays on the active timeline when the table is upgraded to version 9, so the V2 (completion-time) path is affected as well.
Reproduced with a table written by 0.14.1 and read by 1.1.1 and 1.2.1, with a full table scan as fallback:
| Window | Expected | Returned |
|---|---|---|
| Archived → archived | 500 | **0** |
| Archived → latest | 1500 | **600** (2 of 5 partitions) |
| Active → latest | 100 | 100 |
The same table without the rollback returns the expected rows.
### Summary and Changelog
The file index now checks whether the start is archived against the **completed write timeline**. That is the timeline its incremental partition listing reads, so a stale rollback or clean no longer makes an archived start look active.
- `BaseHoodieTableFileIndex.isBeforeTimelineStarts()` uses `getActiveTimeline().getWriteTimeline().filterCompletedInstants()`. This applies to both the requested-time path and the completion-time path.
- Unit test `BaseHoodieTableFileIndexTest#testIncrementalStartBeforeFirstWriteIsArchivedDespiteOlderNonWriteInstant` covers both query modes. It fails without the fix.
- Functional test `TestIncrementalReadWithFullTableScan#testFallbackFullTableScanWithStaleRollbackOnActiveTimeline` writes a failed write, its rollback and archived commits rotating over 5 partitions. It then compares fallback incremental reads over archived, mixed and active windows with the expected commits and partitions. It covers COW v6, COW v6→v9 upgraded, and MOR v6. Without the fix all three fail; post-fix CI results are still pending, see the checklist.
### Impact
Fallback incremental reads (V1 and V2, COW and MOR) return correct results on tables that carry such instants. Without the fallback, the relation already filters the listed partitions down to the ones changed in range, so results don't change. The file index may list more partitions in that case, from the metadata table.
### Risk Level
low. This is a one-method change to the guard in front of the incremental partition listing. It can only make the file index take the existing "list all partitions" path more often, which is a superset of the incremental listing.
### Documentation Update
none
### Contributor's checklist
- [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable
- [ ] Post-fix run of the functional test and the existing incremental/CDC suites (in progress)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(common): correct incremental fallback reads when a stale non-write instant precedes the first active commit [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua merged PR #20273:
URL: https://github.com/apache/hudi/pull/20273
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(common): correct incremental fallback reads when a stale non-write instant precedes the first active commit [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #20273:
URL: https://github.com/apache/hudi/pull/20273#discussion_r4266829192
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java:
##########
@@ -188,10 +188,12 @@ public QueryContext analyze() {
}
}
if (!archivedInstantTimes.isEmpty()) {
- archivedReadTimeline = getArchivedReadTimeline(metaClient, archivedInstantTimes.get(0));
+ archivedReadTimeline = getArchivedReadTimeline(metaClient, archivedInstantTimes.stream().min(String::compareTo).get());
archivedInstants = archivedReadTimeline.getInstantsAsStream().filter(instant -> instantTimeSet.contains(instant.requestedTime())).collect(Collectors.toList());
}
- List<String> instants = Stream.concat(archivedInstants.stream(), activeInstants.stream()).map(HoodieInstant::requestedTime).collect(Collectors.toList());
+ // archived and active instants can interleave (see #splitInstantByActiveness), so order them by instant time.
+ List<String> instants = Stream.concat(archivedInstants.stream(), activeInstants.stream()).map(HoodieInstant::requestedTime)
+ .sorted().collect(Collectors.toList());
Review Comment:
Note that this issue is relevant only when the archival beyond savepoint is enabled and there are "holes" in the active timeline, i.e., some commits having instant times between the savepoint and later commits are archived. So this does not affect out-of-the-box use cases where the archival beyond savepoint is disabled.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(common): correct incremental fallback reads when a stale non-write instant precedes the first active commit [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #20273:
URL: https://github.com/apache/hudi/pull/20273#discussion_r4264765992
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java:
##########
@@ -188,10 +188,12 @@ public QueryContext analyze() {
}
}
if (!archivedInstantTimes.isEmpty()) {
- archivedReadTimeline = getArchivedReadTimeline(metaClient, archivedInstantTimes.get(0));
+ archivedReadTimeline = getArchivedReadTimeline(metaClient, archivedInstantTimes.stream().min(String::compareTo).get());
archivedInstants = archivedReadTimeline.getInstantsAsStream().filter(instant -> instantTimeSet.contains(instant.requestedTime())).collect(Collectors.toList());
}
- List<String> instants = Stream.concat(archivedInstants.stream(), activeInstants.stream()).map(HoodieInstant::requestedTime).collect(Collectors.toList());
+ // archived and active instants can interleave (see #splitInstantByActiveness), so order them by instant time.
+ List<String> instants = Stream.concat(archivedInstants.stream(), activeInstants.stream()).map(HoodieInstant::requestedTime)
+ .sorted().collect(Collectors.toList());
Review Comment:
Confirmed this one. With C1 savepointed and C2–C6 archived past it, `getMaxCompletionTime()` returns C1's completion time, so a bounded Flink read caps `getLatestMergedFileSlicesBeforeOrOn` at C1 and drops the C2–C6 file slices; the streaming read picks them up on the next poll since the issued offset is C1's completion time. `getFilesFromMetadata` also walks the merged metadata list in reverse assuming ascending order, so an appended C1 can shadow a newer version of a file group it shares with a hole commit. `HoodieIncrSource` reads all six commits but checkpoints at C1's completion time, so it reprocesses C2–C6 in the next batch. The Spark relations use neither, so they are fine.
All of this needs the savepoint hole, which master already drops entirely, so I'm fine landing this PR as is and taking it as a follow-up: `getMaxCompletionTime()` can fall back to the active timeline's latest completion time whenever an archived instant completed after the newest active one (same reasoning as the all-archived branch), and `getInstants()` / the Flink merge can sort by instant time. @ad1happy2go could you open a tracking issue for that?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(common): correct incremental fallback reads when a stale non-write instant precedes the first active commit [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #20273:
URL: https://github.com/apache/hudi/pull/20273#discussion_r4272609569
##########
hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/CompletionTimeQueryViewV2.java:
##########
@@ -91,9 +102,19 @@ public class CompletionTimeQueryViewV2 implements CompletionTimeQueryView, Seria
public CompletionTimeQueryViewV2(HoodieTableMetaClient metaClient) {
this.metaClient = metaClient;
this.instantTimeToCompletionTimeMap = new ConcurrentHashMap<>();
- this.cursorInstant = metaClient.getActiveTimeline().firstInstant().map(HoodieInstant::requestedTime).orElse("");
+ HoodieTimeline activeTimeline = metaClient.getActiveTimeline();
+ // Instants at or after the cursor are assumed to be on the active timeline, so it must start at the first
+ // write instant that is not savepointed: an older non-write instant (e.g. a rollback the archiver has not reached)
+ // or a savepointed commit left behind by archival would otherwise hide the archived write instants after it.
+ // Instants before the cursor that are still active are loaded eagerly by #load. Without such a write instant,
+ // nothing after the last active instant can be archived.
+ this.firstActiveWriteInstant = activeTimeline.getTimelineOfActions(HoodieTimeline.WRITE_AND_SAVEPOINT_ACTIONS).getFirstNonSavepointCommit()
Review Comment:
[P2] Preserve the no-archive-I/O path for ranges covered by a contiguous active timeline
Moving the cursor from the first active instant to the first non-savepointed write can introduce archive I/O even with `hoodie.archive.beyond.savepoint=false` and no archival holes. For example, let all archived instants precede Oct 1, with an active completed clean on Oct 1 and the first active write on Oct 3. For a query starting Oct 3 at noon, `getInstantTimes()` computes its one-day lookback as Oct 2 at noon. Previously that was after the cursor, so no archive load occurred; now it is before the cursor and calls `loadCompletionTimeIncrementally()` for an interval containing no archived instants. A leading sequence of retained savepointed writes can produce the same effect.
`ArchivedTimelineLoaderV2.getFilteredFiles()` reads the latest archive version/manifest before discovering that no files overlap. This adds storage I/O and manifest processing, even though no archive Parquet file needs to be read. The cursor only amortizes this within one view; `IncrementalQueryAnalyzer.analyze()` creates a new view each time.
Could we retain a safe fast path for contiguous V2 timelines while preserving the legacy-upgrade/hole fix, and add a loader-call-count regression test for this fully active range?
##########
hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/CompletionTimeQueryViewV2.java:
##########
@@ -102,14 +123,18 @@ public CompletionTimeQueryViewV2(HoodieTableMetaClient metaClient) {
*/
public boolean isCompleted(String instantTime) {
// archival does not proceed beyond the first savepoint, so any instant before that is completed.
- return this.instantTimeToCompletionTimeMap.containsKey(instantTime) || isArchived(instantTime);
+ return this.instantTimeToCompletionTimeMap.containsKey(instantTime)
+ || InstantComparison.compareTimestamps(instantTime, LESSER_THAN, this.firstNonSavepointCommit);
}
/**
* Returns whether the instant is archived.
*/
public boolean isArchived(String instantTime) {
- return InstantComparison.compareTimestamps(instantTime, LESSER_THAN, this.firstNonSavepointCommit);
+ // with archival beyond savepoint, the write instants between a savepointed commit and the first active write
+ // instant are archived as well, while the savepointed commit itself stays active.
+ return InstantComparison.compareTimestamps(instantTime, LESSER_THAN, this.firstActiveWriteInstant)
Review Comment:
[P2] Avoid archive lookup caused by an active clean/rollback before this boundary
With archival beyond savepoint disabled, consider an entirely active sequence `C1(savepointed), clean2, C3`. This new boundary is C3, so `isArchived(clean2)` returns true even though clean2 is active. Previously the write boundary was C1 and it returned false. The same applies to a rollback between those writes.
This reaches the query path because `load()` stores completion times for all completed actions, and the bounded `getInstantTimes()` path returns matching map entries without restricting them to writes. If clean2 is in the completion-time window, `splitInstantByActiveness()` now puts it in the archived bucket and `analyze()` calls `getArchivedReadTimeline()`, even when every selected write is active. Subsequent commit filtering removes the non-write action, so the concern here is unnecessary archive access rather than incorrect write rows. This can also happen independently of the cursor/lookback issue when a delayed clean/rollback completes in a recent query window.
Could the analyzer exclude non-write entries from its archival decision (or otherwise recognize these active actions), with a test asserting that an active savepoint/clean/write sequence does not trigger an archived-timeline lookup?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(common): list all partitions in fallback incremental reads when a stale non-write instant precedes the first commit [hudi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20273:
URL: https://github.com/apache/hudi/pull/20273#issuecomment-6087831858
## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/20273?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `0%` with `3 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 80.62%. Comparing base ([`3aa7fe2`](https://app.codecov.io/gh/apache/hudi/commit/3aa7fe2a9f70ef52064c977d28d28b4f4c95d00b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`dddec27`](https://app.codecov.io/gh/apache/hudi/commit/dddec27542cd16a2b1f59a3574b35d849fc11494?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi/pull/20273?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...pache/hudi/core/read/BaseHoodieTableFileIndex.java](https://app.codecov.io/gh/apache/hudi/pull/20273?src=pr&el=tree&filepath=hudi-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhudi%2Fcore%2Fread%2FBaseHoodieTableFileIndex.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29yZS9yZWFkL0Jhc2VIb29kaWVUYWJsZUZpbGVJbmRleC5qYXZh) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/hudi/pull/20273?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #20273 +/- ##
============================================
- Coverage 80.63% 80.62% -0.01%
Complexity 35060 35060
============================================
Files 2552 2552
Lines 143399 143400 +1
Branches 17454 17454
============================================
- Hits 115623 115621 -2
- Misses 19860 19863 +3
Partials 7916 7916
```
| [Components](https://app.codecov.io/gh/apache/hudi/pull/20273/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/20273/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.16% <0.00%> (-0.01%)` | :arrow_down: |
| [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/20273/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.80% <ø> (ø)` | |
| [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/20273/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.97% <ø> (ø)` | |
| [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/20273/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `74.05% <ø> (ø)` | |
| [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/20273/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.19% <ø> (ø)` | |
| [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/20273/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.80% <ø> (ø)` | |
| [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/20273/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `71.11% <ø> (ø)` | |
| [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/20273/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.23% <ø> (ø)` | |
| [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/20273/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.43% <ø> (ø)` | |
| [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/20273/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.41% <ø> (ø)` | |
| [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/20273/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.00% <ø> (ø)` | |
| [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/20273/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `53.96% <ø> (ø)` | |
| [Flag](https://app.codecov.io/gh/apache/hudi/pull/20273/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hadoop-mr-java-client](https://app.codecov.io/gh/apache/hudi/pull/20273/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.95% <0.00%> (-0.02%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi/pull/20273?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...pache/hudi/core/read/BaseHoodieTableFileIndex.java](https://app.codecov.io/gh/apache/hudi/pull/20273?src=pr&el=tree&filepath=hudi-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhudi%2Fcore%2Fread%2FBaseHoodieTableFileIndex.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29yZS9yZWFkL0Jhc2VIb29kaWVUYWJsZUZpbGVJbmRleC5qYXZh) | `88.84% <0.00%> (-1.16%)` | :arrow_down: |
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(common): correct incremental fallback reads when a stale non-write instant precedes the first active commit [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #20273:
URL: https://github.com/apache/hudi/pull/20273#discussion_r4264038457
##########
hudi-common/src/test/java/org/apache/hudi/core/read/BaseHoodieTableFileIndexTest.java:
##########
@@ -32,7 +37,9 @@
import java.util.List;
import java.util.Map;
+import static org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
Review Comment:
nit: this class should be named `TestBaseHoodieTableFileIndex`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(common): correct incremental fallback reads when a stale non-write instant precedes the first active commit [hudi]
Posted by "ad1happy2go (via GitHub)" <gi...@apache.org>.
ad1happy2go commented on code in PR #20273:
URL: https://github.com/apache/hudi/pull/20273#discussion_r4259650503
##########
hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/CompletionTimeQueryViewV2.java:
##########
@@ -102,14 +128,18 @@ public CompletionTimeQueryViewV2(HoodieTableMetaClient metaClient) {
*/
public boolean isCompleted(String instantTime) {
// archival does not proceed beyond the first savepoint, so any instant before that is completed.
- return this.instantTimeToCompletionTimeMap.containsKey(instantTime) || isArchived(instantTime);
+ return this.instantTimeToCompletionTimeMap.containsKey(instantTime)
+ || InstantComparison.compareTimestamps(instantTime, LESSER_THAN, this.firstNonSavepointCommit);
}
/**
* Returns whether the instant is archived.
*/
public boolean isArchived(String instantTime) {
- return InstantComparison.compareTimestamps(instantTime, LESSER_THAN, this.firstNonSavepointCommit);
+ // with archival beyond savepoint, the write instants between a savepointed commit and the first active write
+ // instant are archived as well, while the savepointed commit itself stays active.
+ return InstantComparison.compareTimestamps(instantTime, LESSER_THAN, this.firstActiveWriteInstant)
+ && !this.activeSavepointedInstants.contains(instantTime);
Review Comment:
Good catch, fixed in 4d9be634a. `splitInstantByActiveness` now checks each instant with `isArchived` instead of cutting at the first active one, and the merged instant list is ordered by instant time. `testArchivedInstantsAfterSavepointedCommit` now also covers a window starting at `1000`: 2–6 come back as archived and 1, 7, 8 as active.
##########
hudi-common/src/main/java/org/apache/hudi/core/read/BaseHoodieTableFileIndex.java:
##########
@@ -95,6 +96,10 @@
@Slf4j
public abstract class BaseHoodieTableFileIndex implements AutoCloseable {
+ private static final Set<String> WRITE_AND_SAVEPOINT_ACTIONS = CollectionUtils.createSet(
Review Comment:
Done in 4d9be634a: `WRITE_AND_SAVEPOINT_ACTIONS` is now defined once in `HoodieTimeline` and used by both classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(common): correct incremental fallback reads when a stale non-write instant precedes the first active commit [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
nsivabalan commented on code in PR #20273:
URL: https://github.com/apache/hudi/pull/20273#discussion_r4263875623
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java:
##########
@@ -216,14 +218,11 @@ public QueryContext analyze() {
* Splits the given instant time list into a pair of archived instant list and active instant list.
*/
private static Pair<List<String>, List<String>> splitInstantByActiveness(List<String> instantTimeList, CompletionTimeQueryView completionTimeQueryView) {
- int firstActiveIdx = IntStream.range(0, instantTimeList.size()).filter(i -> !completionTimeQueryView.isArchived(instantTimeList.get(i))).findFirst().orElse(-1);
- if (firstActiveIdx == -1) {
- return Pair.of(instantTimeList, Collections.emptyList());
- } else if (firstActiveIdx == 0) {
- return Pair.of(Collections.emptyList(), instantTimeList);
- } else {
- return Pair.of(instantTimeList.subList(0, firstActiveIdx), instantTimeList.subList(firstActiveIdx, instantTimeList.size()));
- }
+ // with archival beyond savepoint, a savepointed commit stays active while the commits after it are archived,
+ // so the archived instants are not always a prefix of the list: check each instant on its own.
+ Map<Boolean, List<String>> instantTimesByArchived = instantTimeList.stream()
Review Comment:
can we take a much simpler route for oob users (archival beyond saveppoint disabled).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(common): correct incremental fallback reads when a stale non-write instant precedes the first active commit [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #20273:
URL: https://github.com/apache/hudi/pull/20273#discussion_r4245772280
##########
hudi-common/src/main/java/org/apache/hudi/core/read/BaseHoodieTableFileIndex.java:
##########
@@ -95,6 +96,10 @@
@Slf4j
public abstract class BaseHoodieTableFileIndex implements AutoCloseable {
+ private static final Set<String> WRITE_AND_SAVEPOINT_ACTIONS = CollectionUtils.createSet(
Review Comment:
🤖 nit: `WRITE_AND_SAVEPOINT_ACTIONS` is defined identically here and in `CompletionTimeQueryViewV2`; would it make sense to define it once (e.g. in `HoodieTimeline`) and reuse it?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/CompletionTimeQueryViewV2.java:
##########
@@ -102,14 +128,18 @@ public CompletionTimeQueryViewV2(HoodieTableMetaClient metaClient) {
*/
public boolean isCompleted(String instantTime) {
// archival does not proceed beyond the first savepoint, so any instant before that is completed.
- return this.instantTimeToCompletionTimeMap.containsKey(instantTime) || isArchived(instantTime);
+ return this.instantTimeToCompletionTimeMap.containsKey(instantTime)
+ || InstantComparison.compareTimestamps(instantTime, LESSER_THAN, this.firstNonSavepointCommit);
}
/**
* Returns whether the instant is archived.
*/
public boolean isArchived(String instantTime) {
- return InstantComparison.compareTimestamps(instantTime, LESSER_THAN, this.firstNonSavepointCommit);
+ // with archival beyond savepoint, the write instants between a savepointed commit and the first active write
+ // instant are archived as well, while the savepointed commit itself stays active.
+ return InstantComparison.compareTimestamps(instantTime, LESSER_THAN, this.firstActiveWriteInstant)
+ && !this.activeSavepointedInstants.contains(instantTime);
Review Comment:
🤖 non-blocking: `isArchived` can now go false then true along the timeline (savepointed C1 is not archived, but C2–C6 after it are), while `IncrementalQueryAnalyzer.splitInstantByActiveness` still treats everything from the first non-archived instant onward as active.
So with a range that starts before C1's completion, or a hole that has a second savepointed commit inside it, C2–C6 land in the active list. They are not on the active timeline, so they are silently dropped. The full-scan fallback doesn't catch this, because it filters on `includedCommits`.
Could the split go instant by instant instead, e.g. `filter(isArchived)` / `filter(!isArchived)`, with the merged instants sorted by requested time? A window starting at `1000` in `testArchivedInstantsAfterSavepointedCommit` would cover this.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(common): correct incremental fallback reads when a stale non-write instant precedes the first active commit [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #20273:
URL: https://github.com/apache/hudi/pull/20273#discussion_r4259780861
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java:
##########
@@ -188,10 +188,12 @@ public QueryContext analyze() {
}
}
if (!archivedInstantTimes.isEmpty()) {
- archivedReadTimeline = getArchivedReadTimeline(metaClient, archivedInstantTimes.get(0));
+ archivedReadTimeline = getArchivedReadTimeline(metaClient, archivedInstantTimes.stream().min(String::compareTo).get());
archivedInstants = archivedReadTimeline.getInstantsAsStream().filter(instant -> instantTimeSet.contains(instant.requestedTime())).collect(Collectors.toList());
}
- List<String> instants = Stream.concat(archivedInstants.stream(), activeInstants.stream()).map(HoodieInstant::requestedTime).collect(Collectors.toList());
+ // archived and active instants can interleave (see #splitInstantByActiveness), so order them by instant time.
+ List<String> instants = Stream.concat(archivedInstants.stream(), activeInstants.stream()).map(HoodieInstant::requestedTime)
+ .sorted().collect(Collectors.toList());
Review Comment:
🤖 non-blocking: `instants` is sorted now, but `QueryContext.getMaxCompletionTime()`, `getInstants()` and Flink's `mergeList(archived, active)` still assume archived instants come first. For example, a Flink incremental read with start before C1 and end before C7 gets active = [C1], so `getMaxCompletionTime()` returns C1's completion time, and the full-scan branch's `getLatestMergedFileSlicesBeforeOrOn` then drops the C2–C6 file slices. Could `getMaxCompletionTime()` fall back to the active timeline's latest completion time when archived instants completed after the active ones, and could the archived/active lists also be ordered by instant time for downstream readers?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(common): correct incremental fallback reads when a stale non-write instant precedes the first active commit [hudi]
Posted by "nsivabalan (via GitHub)" <gi...@apache.org>.
nsivabalan commented on PR #20273:
URL: https://github.com/apache/hudi/pull/20273#issuecomment-6157370012
Thanks @ad1happy2go, nice debugging on this one. The core fix looks right to me. Treating "first instant on the active timeline" as the archival boundary breaks as soon as a rollback or clean outlives the commits around it, and the archiver archives those on a separate schedule. Moving `BaseHoodieTableFileIndex#isBeforeTimelineStarts` and the `CompletionTimeQueryViewV2` cursor to the write timeline fixes the reported bug.
Most of the remaining diff is about archival beyond savepoint: `activeSavepointedInstants`, the new `isArchived` boundary, the per-instant `splitInstantByActiveness`, and the savepoint tests. I dug into that feature, and I don't think we should keep extending read paths for it.
The hole it leaves after a savepointed commit breaks much more than incremental reads. `getFirstNonSavepointCommit()` only skips a savepointed commit when the timeline it's given contains the SAVEPOINT instant. The file system view (`AbstractTableFileSystemView#refreshTimeline` → `getWriteTimeline()`), the cleaner and the index all pass timelines without it. So a commit in the hole counts as uncommitted.
I reproduced this on master with a Spark functional test on COW. All four combinations behave the same: table version 6 and 9, each with the BLOOM and SIMPLE index.
- C1 inserts into partition 1 and is savepointed.
- C2 inserts 10 records into partition 2.
- Four more commits into partition 3 archive C2, while C1 stays active.
| Step | Expected | Actual |
|---|---|---|
| Snapshot read | 60 rows | **50**: C2's rows are missing |
| Upsert of C2's keys | update C2's file group | **new file group** in partition 2 |
| Delete savepoint C1, archive past it | 100 rows / 100 keys | **110 rows / 100 keys**: 10 duplicates |
That's the same root cause as #11405, which has been open since 2024. With the LSM timeline we're moving away from a separate archived timeline anyway, so I've opened #20310 to remove support for `hoodie.archive.beyond.savepoint`. It deprecates the config, fails fast when it's set to true, and replaces the config-based restore guard with a check on the table itself.
Given that, could we narrow this PR to the stale rollback/clean fix?
- Keep the write-timeline boundary in `isBeforeTimelineStarts` and the `CompletionTimeQueryViewV2` cursor, plus the stale-rollback tests (`testIncrementalStartBeforeFirstWriteIsArchivedDespiteOlderNonWriteInstant`, `testIncrementalStartIsArchivedWithoutActiveWriteInstant`, `testArchivedInstantsAfterStaleNonWriteInstant`, and the Spark functional test).
- Drop the savepoint-hole handling: `activeSavepointedInstants`, the `isArchived` / `isCompleted` split, and the `splitInstantByActiveness` rewrite with its sorting.
That also settles the open Flink comment about `QueryContext#getMaxCompletionTime`. That issue only exists because archived and active instants can now interleave, which can't happen once the feature is gone.
<details><summary>Repro test (TestArchiveBeyondSavepointReads, hudi-spark)</summary>
```java
package org.apache.hudi.functional;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.config.HoodieArchivalConfig;
import org.apache.hudi.config.HoodieCleanConfig;
import org.apache.hudi.config.HoodieIndexConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.testutils.HoodieClientTestBase;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Stream;
import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH;
import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_THIRD_PARTITION_PATH;
import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
import static org.junit.jupiter.api.Assertions.assertAll;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Repro: with hoodie.archive.beyond.savepoint=true, commits archived after a savepointed commit leave a hole in the
* active timeline. File slices written only by commits in the hole are treated as uncommitted by the file system view.
*/
@Tag("functional")
public class TestArchiveBeyondSavepointReads extends HoodieClientTestBase {
private static final int NUM_RECORDS = 10;
private static final String REPORT_FILE = "target/archive-beyond-savepoint-repro.txt";
@BeforeEach
public void setUp() throws Exception {
initPath();
initSparkContexts();
initTestDataGenerator();
initHoodieStorage();
initTimelineService();
}
private static Stream<Arguments> tableVersionAndIndex() {
return Stream.of(
Arguments.of(HoodieTableVersion.SIX, HoodieIndex.IndexType.BLOOM),
Arguments.of(HoodieTableVersion.SIX, HoodieIndex.IndexType.SIMPLE),
Arguments.of(HoodieTableVersion.NINE, HoodieIndex.IndexType.BLOOM),
Arguments.of(HoodieTableVersion.NINE, HoodieIndex.IndexType.SIMPLE));
}
/**
* C1 inserts into partition 1 and is savepointed. C2 inserts into partition 2, a file group no later commit touches.
* C3..C6 insert into partition 3, which archives C2 while C1 stays active. Then:
* 1. a snapshot read should still return C2's rows;
* 2. an upsert of C2's keys should update them in place, not insert them again;
* 3. once the savepoint is deleted and C1 is archived too, the table should not contain duplicate keys.
*/
@ParameterizedTest
@MethodSource("tableVersionAndIndex")
void testHoleCommitsAfterSavepoint(HoodieTableVersion tableVersion, HoodieIndex.IndexType indexType) throws Exception {
HoodieWriteConfig writeConfig = getWriteConfigAndInitTable(tableVersion, indexType);
List<String> observations = new ArrayList<>();
try (SparkRDDWriteClient client = getHoodieWriteClient(writeConfig)) {
// given: savepointed C1, hole commit C2, and enough later commits to archive C2
String c1 = insert(client, DEFAULT_FIRST_PARTITION_PATH);
client.savepoint(c1, "user", "savepoint C1");
List<HoodieRecord> holeRecords = new ArrayList<>();
String c2 = insert(client, DEFAULT_SECOND_PARTITION_PATH, holeRecords);
for (int i = 0; i < 4; i++) {
insert(client, DEFAULT_THIRD_PARTITION_PATH);
}
HoodieTimeline activeTimeline = metaClient.reloadActiveTimeline();
assertTrue(activeTimeline.containsInstant(c1), "precondition: savepointed C1 stays active");
assertFalse(activeTimeline.containsInstant(c2), "precondition: C2 is archived, leaving a hole after C1");
// when/then 1: snapshot read
long snapshotCount = snapshotRead().count();
observations.add("snapshot rows after archival (expected 60): " + snapshotCount);
// when/then 2: upsert C2's keys
String newCommitTime = client.startCommit();
List<HoodieRecord> updates = dataGen.generateUpdates(newCommitTime, holeRecords);
client.commit(newCommitTime, client.upsert(jsc.parallelize(updates, 1), newCommitTime));
Dataset<Row> afterUpsert = snapshotRead();
long upsertRows = afterUpsert.count();
long upsertKeys = afterUpsert.select("_hoodie_record_key").distinct().count();
long holeFileGroupsInStorage = countBaseFileGroups(DEFAULT_SECOND_PARTITION_PATH);
observations.add("after upsert of C2 keys: rows=" + upsertRows + " distinctKeys=" + upsertKeys
+ " fileGroupsInPartition2=" + holeFileGroupsInStorage + " (expected 60/60/1)");
// when/then 3: delete the savepoint and archive C1, so the hole closes
client.deleteSavepoint(c1);
for (int i = 0; i < 4; i++) {
insert(client, DEFAULT_THIRD_PARTITION_PATH);
}
assertFalse(metaClient.reloadActiveTimeline().containsInstant(c1), "precondition: C1 archived after savepoint delete");
Dataset<Row> afterHoleCloses = snapshotRead();
long finalRows = afterHoleCloses.count();
long finalKeys = afterHoleCloses.select("_hoodie_record_key").distinct().count();
observations.add("after savepoint delete and archival: rows=" + finalRows + " distinctKeys=" + finalKeys
+ " (expected " + (60 + 4 * NUM_RECORDS) + "/" + (60 + 4 * NUM_RECORDS) + ")");
report(tableVersion, indexType, observations);
long expectedFinal = 60 + 4 * NUM_RECORDS;
assertAll(
() -> assertEquals(60, snapshotCount, "snapshot read drops rows of the archived hole commit"),
() -> assertEquals(60, upsertRows, "rows after upserting the hole commit's keys"),
() -> assertEquals(60, upsertKeys, "distinct keys after upserting the hole commit's keys"),
() -> assertEquals(1, holeFileGroupsInStorage, "upsert should update the existing file group, not create a new one"),
() -> assertEquals(expectedFinal, finalRows, "rows once the hole closes"),
() -> assertEquals(finalRows, finalKeys, "duplicate record keys once the hole closes"));
}
}
private HoodieWriteConfig getWriteConfigAndInitTable(HoodieTableVersion tableVersion, HoodieIndex.IndexType indexType) throws IOException {
HoodieWriteConfig writeConfig = getConfigBuilder(HoodieFailedWritesCleaningPolicy.EAGER)
.withAutoUpgradeVersion(false)
.withWriteTableVersion(tableVersion.versionCode())
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(indexType).build())
.withCleanConfig(HoodieCleanConfig.newBuilder().retainCommits(1).build())
.withArchivalConfig(HoodieArchivalConfig.newBuilder()
.archiveCommitsWith(2, 3)
.withArchiveBeyondSavepoint(true)
.build())
.withMetadataConfig(HoodieMetadataConfig.newBuilder()
.enable(true)
.withMaxNumDeltaCommitsBeforeCompaction(1)
.build())
.build();
initMetaClient(HoodieTableType.COPY_ON_WRITE, writeConfig.getProps());
return writeConfig;
}
private String insert(SparkRDDWriteClient client, String partition) {
return insert(client, partition, new ArrayList<>());
}
private String insert(SparkRDDWriteClient client, String partition, List<HoodieRecord> insertedRecords) {
String instantTime = client.startCommit();
List<HoodieRecord> records = dataGen.generateInsertsForPartition(instantTime, NUM_RECORDS, partition);
insertedRecords.addAll(records);
List<org.apache.hudi.client.WriteStatus> statuses = client.insert(jsc.parallelize(records, 1), instantTime).collect();
assertNoWriteErrors(statuses);
client.commit(instantTime, jsc.parallelize(statuses, 1));
return instantTime;
}
private Dataset<Row> snapshotRead() {
return sparkSession.read().format("hudi").load(basePath);
}
private long countBaseFileGroups(String partition) throws IOException {
try (Stream<java.nio.file.Path> files = Files.list(Paths.get(basePath, partition))) {
return files.map(path -> path.getFileName().toString())
.filter(name -> name.endsWith(".parquet"))
.map(name -> name.substring(0, name.indexOf('_')))
.distinct()
.count();
}
}
private static void report(HoodieTableVersion tableVersion, HoodieIndex.IndexType indexType, List<String> observations) throws IOException {
StringBuilder sb = new StringBuilder("== table version " + tableVersion.versionCode() + ", index " + indexType + "\n");
observations.forEach(line -> sb.append(" ").append(line).append('\n'));
Files.write(Paths.get(REPORT_FILE), sb.toString().getBytes(StandardCharsets.UTF_8),
StandardOpenOption.CREATE, StandardOpenOption.APPEND);
}
}
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org