feat(common): query LAYOUT_VERSION_1 timelines in CompletionTimeQueryView by using instant time as completion time - #19338
Conversation
…View by using instant time as completion time
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
test. One minor readability nit on parameter mutation in the new method body.
| rangeStart = Option.empty(); | ||
| } | ||
|
|
||
| if (rangeStart.isEmpty() && rangeEnd.isEmpty()) { |
There was a problem hiding this comment.
🤖 nit: could you avoid mutating the rangeStart parameter here? When a reader reaches the second if (rangeStart.isEmpty() && rangeEnd.isEmpty()) check they have to mentally track whether the earlier reassignment fired. Something like Option<String> effectiveStart = startFromEarliest ? Option.empty() : rangeStart; introduced upfront, then used in the downstream checks and the builder, would make the control flow self-evident.
There was a problem hiding this comment.
Good suggestion. I replaced the rangeStart reassignment with an effectiveStart variable to make the range handling clearer.
| Function<String, String> earliestInstantTimeFunc) { | ||
| throw new RuntimeException("Incremental query view for timeline version 1 not yet implemented"); | ||
| final boolean startFromEarliest = START_COMMIT_EARLIEST.equalsIgnoreCase(rangeStart.orElse(null)); | ||
| HoodieTimeline completedTimeline = timeline.filterCompletedInstants(); |
There was a problem hiding this comment.
This turns the archived-offset case from a hard failure into a silent skip. Everything below only sees the active timeline, so if a streaming job was down long enough for archival to pass its resume offset, the commits between rangeStart and the first active instant just vanish from the read. The V2 view covers that window by lazily loading the archive; the old RuntimeException here at least made the gap loud. (Same story for the (_, end] branch: V2 falls back to the archived map, this returns empty.)
The view already has isArchived(), so a cheap guard keeps the fail-stop semantics:
if (rangeStart.isPresent() && !startFromEarliest && isArchived(rangeStart.get())) {
throw new HoodieException("Start instant " + rangeStart.get()
+ " is already archived; incremental read on a LAYOUT_VERSION_1 timeline does not consult the archived timeline");
}Reading the archived timeline would be even better (analyze() already handles archived instants via getArchivedReadTimeline), but at minimum a loud failure beats silent data loss on resume.
There was a problem hiding this comment.
This turns the archived-offset case from a hard failure into a silent skip. Everything below only sees the active timeline, so if a streaming job was down long enough for archival to pass its resume offset, the commits between
rangeStartand the first active instant just vanish from the read. The V2 view covers that window by lazily loading the archive; the oldRuntimeExceptionhere at least made the gap loud. (Same story for the(_, end]branch: V2 falls back to the archived map, this returns empty.)The view already has
isArchived(), so a cheap guard keeps the fail-stop semantics:if (rangeStart.isPresent() && !startFromEarliest && isArchived(rangeStart.get())) { throw new HoodieException("Start instant " + rangeStart.get() + " is already archived; incremental read on a LAYOUT_VERSION_1 timeline does not consult the archived timeline"); }Reading the archived timeline would be even better (
analyze()already handles archived instants viagetArchivedReadTimeline), but at minimum a loud failure beats silent data loss on resume.
Thanks for catching this. I updated the V1 query view to load archived instants on demand, so archived offsets are no longer silently skipped. I also added tests covering archived/active ranges and the end-only fallback.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR implements CompletionTimeQueryViewV1.getInstantTimes so incremental/streaming reads work against LAYOUT_VERSION_1 timelines (table versions 5–7) by treating the instant time as the completion time, mirroring the V2 branch structure. I traced the range logic against the V2 view and the IncrementalQueryAnalyzer caller (including the splitInstantByActiveness ordering assumption, the re-filtering of unfiltered archived candidates, and the getArchivedTimeline inclusive-start semantics) and the core behavior looks consistent and correct. One non-blocking efficiency note is in the inline comment about redundantly loading the archived timeline on the catch-up path. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here.
| .collect(Collectors.toList()); | ||
| } | ||
|
|
||
| private Stream<String> getArchivedInstantTimes(Option<String> startInstant) { |
There was a problem hiding this comment.
🤖 When the query range includes archived instants, the archived timeline ends up loaded twice per analyze() — once here with useCache=false, and again in IncrementalQueryAnalyzer.getArchivedReadTimeline (also useCache=false). The getArchivedTimeline javadoc flags this as costly, and the (_, end] fallback loads from "" (the whole archived timeline). Since V2 amortizes this via its in-memory map, is there a way to avoid the second full load on the catch-up path (e.g. useCache=true, or letting the view/analyzer share one load)?
|
A potential improvement for v1 archived timeline: The latest PR already performs a start-instant-based archive lookup: metaClient.getArchivedTimeline(startInstant, false)For V1, the missing improvement is to make that the actual incremental-query path, not just the candidate-discovery path. Currently:
A cleaner V1 design would have That would preserve V2-compatible range semantics while eliminating the redundant archive scan. Simply adding another start-time lookup inside |
5 participants
Describe the issue this Pull Request addresses
The 1.x incremental read stack (
IncrementalQueryAnalyzer→CompletionTimeQueryView.getInstantTimes) resolves the query range by completion time. For aLAYOUT_VERSION_1timeline (table versions 5–7, releases 0.12–0.16),CompletionTimeQueryViewV1.getInstantTimescurrently throws:As a result, any incremental (and streaming) read against these older tables via the 1.x reader fails outright.
The key observation — as raised in review — is that this is not a "missing completion time" problem: for a
LAYOUT_VERSION_1timeline, an instant's completion time is its instant (request) time, and the semantics are backward compatible. This is already how the V1 view behaves elsewhere (load()mapsrequestedTime → completionTime, andgetCompletionTimereturns the begin time for archived instants). This PR applies that same equivalence to the range query, so V1 falls out as the "completion time = instant time" special case of the existing V2 range logic rather than a separate implementation.Summary and Changelog
Implements
CompletionTimeQueryViewV1.getInstantTimes(...)by mirroring the branch structure already used inCompletionTimeQueryViewV2, driven by requested (instant) time instead of a persisted completion time:(_, end]— returns the single latest instant whose requested time is<= end.['earliest', _)— clears the start bound so it degenerates to consuming from the earliest instant.(_, _)— returns the latest snapshot instant.InstantRange(honoring theOPEN_CLOSED/CLOSED_CLOSEDbounds and nullable boundaries), sorted ascending.Notes:
HoodieTimelinepassed in by the caller (already filtered per user configs such asskipCompaction/skipClustering), consistent with the V2 code path.earliestInstantTimeFuncparameter is unused for V1 (it only serves V2's archive lazy-load boundary).Tests: adds
TestCompletionTimeQueryViewV1covering closed/closed first read, open/closed streaming resume (excluding the last issued offset),earlieststart, end-only, and no-bounds latest-snapshot cases.No code was copied verbatim; the implementation follows the shape of
CompletionTimeQueryViewV2#getInstantTimes.Impact
LAYOUT_VERSION_1) through the 1.x reader; these previously threw.getInstantTimesbody is implemented.Risk Level
Medium. This changes read semantics for older tables from "throws" to "returns instants". Mitigated by:
OPEN_CLOSEDmust exclude the issued offset,CLOSED_CLOSEDfirst read includes the start commit,earliest/ end-only / no-bounds).hudi-commonand runningTestCompletionTimeQueryViewV1locally — 5/5 tests pass.Documentation Update
None. No new config, and no user-facing option changes; this makes an existing documented feature (incremental/streaming query) work on
LAYOUT_VERSION_1tables.Contributor's checklist