This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 3fd4e88d fix(core): gate the log-block scan on instant state (#678)
3fd4e88d is described below
commit 3fd4e88d9b5413ca65cd1ddf2913e45635faf6a6
Author: Lin Liu <[email protected]>
AuthorDate: Tue Sep 1 17:14:23 2026 -0700
fix(core): gate the log-block scan on instant state (#678)
---
crates/core/src/file_group/reader.rs | 25 +++
crates/core/src/file_group/reader_v2/adapter.rs | 10 ++
.../src/file_group/reader_v2/log_record_reader.rs | 128 +++++++++++++---
.../reader_v2/merged_log_record_reader.rs | 7 +-
.../src/file_group/reader_v2/reader_context.rs | 22 ++-
crates/core/src/table/mod.rs | 170 ++++++++++++++++++++-
crates/core/src/timeline/mod.rs | 94 +++++++++++-
crates/core/tests/gold_parity_tests.rs | 22 +++
.../mor/avro/table_uncommitted_log.sql | 78 ++++++++++
.../mor/avro/table_uncommitted_log_v6.zip | Bin 0 -> 24847 bytes
.../mor/avro/table_uncommitted_log_v9.zip | Bin 0 -> 30414 bytes
crates/test/src/lib.rs | 46 ++++++
12 files changed, 565 insertions(+), 37 deletions(-)
diff --git a/crates/core/src/file_group/reader.rs
b/crates/core/src/file_group/reader.rs
index f3a8e06c..5e49461a 100644
--- a/crates/core/src/file_group/reader.rs
+++ b/crates/core/src/file_group/reader.rs
@@ -31,6 +31,7 @@ use crate::file_group::base_file::reader::{
};
use crate::file_group::file_slice::FileSlice;
use crate::file_group::log_file::scanner::{LogFileScanner, ScanResult};
+use crate::file_group::reader_v2::reader_context::CompletionGateInputs;
use crate::file_group::record_batches::RecordBatches;
use crate::merge::record_merger::RecordMerger;
use crate::metadata::meta_field::MetaField;
@@ -64,6 +65,15 @@ pub struct FileGroupReader {
/// base types. A caller holding the timeline knows better; one reading
from
/// paths alone (the cxx bridge) does not, and falls back to the base file.
data_schema_override: Option<arrow_schema::SchemaRef>,
+ /// The committed/inflight sets the log-block scan gates on.
+ ///
+ /// A log file is admitted to a slice on its own instant, but its blocks
+ /// carry theirs — including a writer still inflight when a later one
+ /// committed. Without these the scan merges such a block, because it sorts
+ /// below the latest instant and passes every other gate. Only a caller
+ /// holding the timeline can supply them; one reading from paths alone (the
+ /// cxx bridge) leaves the gate off, as it always was.
+ completion_gate_inputs: Option<Arc<CompletionGateInputs>>,
}
impl std::fmt::Debug for FileGroupReader {
@@ -105,6 +115,7 @@ impl FileGroupReader {
base_file_format: format,
base_file_reader,
data_schema_override: None,
+ completion_gate_inputs: None,
})
}
@@ -135,6 +146,7 @@ impl FileGroupReader {
base_file_format: format,
base_file_reader,
data_schema_override: None,
+ completion_gate_inputs: None,
})
}
@@ -214,6 +226,17 @@ impl FileGroupReader {
))
}
+ /// Whether this reader can gate the log scan on instant state.
+ #[cfg(test)]
+ pub(crate) fn has_completion_gate_inputs(&self) -> bool {
+ self.completion_gate_inputs.is_some()
+ }
+
+ /// Gate the log-block scan on these committed/inflight sets.
+ pub(crate) fn set_completion_gate_inputs(&mut self, inputs:
CompletionGateInputs) {
+ self.completion_gate_inputs = Some(Arc::new(inputs));
+ }
+
/// Read slices with `schema` rather than whatever the base file carries.
pub(crate) fn set_data_schema(&mut self, schema: arrow_schema::SchemaRef) {
self.data_schema_override = Some(schema);
@@ -341,6 +364,7 @@ impl FileGroupReader {
log_file_paths,
partition_path,
Some(data_schema),
+ self.completion_gate_inputs.clone(),
)
.await?;
@@ -789,6 +813,7 @@ impl FileGroupReader {
log_file_paths,
partition_path,
Some(data_schema),
+ self.completion_gate_inputs.clone(),
)
.await?;
diff --git a/crates/core/src/file_group/reader_v2/adapter.rs
b/crates/core/src/file_group/reader_v2/adapter.rs
index 3d53ba5b..3524ee05 100644
--- a/crates/core/src/file_group/reader_v2/adapter.rs
+++ b/crates/core/src/file_group/reader_v2/adapter.rs
@@ -36,6 +36,7 @@ use crate::error::CoreError;
use crate::file_group::reader_v2::MAX_INSTANT_TIME;
use crate::file_group::reader_v2::engine::HoodieFileGroupReader;
use crate::file_group::reader_v2::input_split::InputSplit;
+use crate::file_group::reader_v2::reader_context::CompletionGateInputs;
use crate::file_group::reader_v2::reader_parameters::ReaderParameters;
use crate::file_group::reader_v2::resolver::resolve_reader_context;
use crate::storage::Storage;
@@ -56,6 +57,7 @@ pub(crate) async fn read_file_slice(
log_file_paths: Vec<String>,
partition_path: String,
data_schema: Option<SchemaRef>,
+ completion_gate_inputs: Option<Arc<CompletionGateInputs>>,
) -> Result<RecordBatch> {
let mut reader = build_reader(
hudi_configs,
@@ -64,6 +66,7 @@ pub(crate) async fn read_file_slice(
log_file_paths,
partition_path,
data_schema,
+ completion_gate_inputs,
)?;
reader.read().await
@@ -81,6 +84,7 @@ pub(crate) async fn read_file_slice_stream(
log_file_paths: Vec<String>,
partition_path: String,
data_schema: Option<SchemaRef>,
+ completion_gate_inputs: Option<Arc<CompletionGateInputs>>,
) -> Result<futures::stream::BoxStream<'static, Result<RecordBatch>>> {
let mut reader = build_reader(
hudi_configs,
@@ -89,6 +93,7 @@ pub(crate) async fn read_file_slice_stream(
log_file_paths,
partition_path,
data_schema,
+ completion_gate_inputs,
)?;
reader.open_blocking_stream().await
@@ -103,10 +108,15 @@ fn build_reader(
log_file_paths: Vec<String>,
partition_path: String,
data_schema: Option<SchemaRef>,
+ completion_gate_inputs: Option<Arc<CompletionGateInputs>>,
) -> Result<HoodieFileGroupReader> {
let has_log_files = !log_file_paths.is_empty();
let hudi_configs = with_unbounded_end_timestamp(hudi_configs);
let mut context = resolve_reader_context(&hudi_configs, has_log_files)?;
+ // The scan's committed/inflight gate. Supplied by a caller holding the
+ // timeline; `None` leaves the gate a no-op, which is what a caller reading
+ // from paths alone (the cxx bridge) can offer.
+ context.completion_gate_inputs = completion_gate_inputs;
context.rebuild_record_context(partition_path.clone());
// A slice with no base file reports an empty path; the engine keys its
diff --git a/crates/core/src/file_group/reader_v2/log_record_reader.rs
b/crates/core/src/file_group/reader_v2/log_record_reader.rs
index 453b4254..63013f29 100644
--- a/crates/core/src/file_group/reader_v2/log_record_reader.rs
+++ b/crates/core/src/file_group/reader_v2/log_record_reader.rs
@@ -97,8 +97,8 @@ impl<'a> CompletionGate<'a> {
/// committedness for any instant, so it admits everything (a no-op) rather
/// than excluding all blocks. Java always builds the gate from a non-empty
/// active timeline (`filterCompletedInstants()`), so an empty set here
means
- /// the inputs were not populated by the caller (e.g. the FFI bridge
forwarded
- /// an empty list). Excluding on that basis would silently drop EVERY log
delta
+ /// the inputs were not populated by the caller — a caller that supplied
the
+ /// struct but left the sets empty. Excluding on that basis would silently
drop EVERY log delta
/// — including committed ones — and return base-file-only data (the
/// C-INFLIGHT silent-wrong: a committed later delta wrongly dropped).
Deferring
/// to the other gates preserves the pre-gate behavior for a mis-wired
gate; a
@@ -118,12 +118,12 @@ impl<'a> CompletionGate<'a> {
/// `completed_instants` set AND no `archived_boundary`. Such a gate
cannot establish
/// committedness for any instant, so [`admits`](Self::admits) degrades to
admit-all.
///
- /// When the gate was supplied deliberately (the caller passes `Some` only
for a table
- /// version < 8 snapshot), being unpopulated is a mis-wire signal, not a
genuine empty
+ /// When the gate was supplied deliberately (the caller passes `Some`
whenever it has a
+ /// timeline to build it from), being unpopulated is a mis-wire signal,
not a genuine empty
/// timeline: Java always builds the gate from a non-empty active timeline
- /// (`filterCompletedInstants()`), and a v1 snapshot always has >= 1
completed instant.
+ /// (`filterCompletedInstants()`), and a readable table always has >= 1
completed instant.
/// `forward_scan_pass1` warns once when this holds so a gate whose inputs
were dropped
- /// across the FFI boundary is diagnosable rather than a silent no-op.
+ /// before reaching the scan is diagnosable rather than a silent no-op.
fn is_unpopulated(&self) -> bool {
self.completed_instants.is_empty() && self.archived_boundary.is_none()
}
@@ -164,17 +164,16 @@ pub fn forward_scan_pass1(
instant_range.is_some(),
);
- // The completion gate was supplied (table version < 8 snapshot) but
carries no positive
- // completion info — its inputs were not populated across the FFI
boundary. Rather than
- // silently degrade to admit-all (Gate 3 becomes a no-op, so a straddling
uncommitted delta
- // could be merged), warn once so the mis-wire is diagnosable. Behavior is
unchanged (still
- // fail-open); this only makes the condition visible.
+ // The completion gate was supplied but carries no positive completion
info — the caller
+ // passed the struct with empty sets. Rather than silently degrade to
admit-all (Gate 3
+ // becomes a no-op, so a straddling uncommitted delta could be merged),
warn so the mis-wire
+ // is diagnosable. Behavior is unchanged (still fail-open); this only
makes it visible.
if completion_gate.is_some_and(CompletionGate::is_unpopulated) {
log::warn!(
"[Pass1] completion gate supplied but unpopulated (empty completed
set, no archived \
- boundary): Gate 3 (C-INFLIGHT) admits all delta blocks. A table
version < 8 snapshot \
- always has >= 1 completed instant, so this indicates the gate
inputs were dropped \
- across the FFI boundary.
latest_instant_time={latest_instant_time}"
+ boundary): Gate 3 admits all delta blocks. A readable table
always has >= 1 \
+ completed instant, so this indicates the gate inputs were dropped
between the \
+ timeline and this scan. latest_instant_time={latest_instant_time}"
);
}
@@ -449,9 +448,9 @@ pub struct BaseHoodieLogRecordReader {
pub allow_inflight_instants: bool,
/// Inputs for the Gate-3 completed/inflight check. `Some` only for
/// table version < 8 (v1 timeline layout); `None` for v8+ and when no
timeline is available,
- /// in which case Gate 3 is a no-op. Populated by the builder / FFI wiring
from the active
- /// timeline (completed + inflight instant sets + the first active
instant).
- pub completion_gate_inputs: Option<CompletionGateInputs>,
+ /// in which case Gate 3 is a no-op. Populated by `Table` from the active
timeline
+ /// (completed + pending instant sets + the first active instant).
+ pub completion_gate_inputs: Option<Arc<CompletionGateInputs>>,
// ── Stats / state (mirrors Java's AtomicLong counters + progress) ──
pub valid_block_instants: Vec<String>,
@@ -536,10 +535,11 @@ impl BaseHoodieLogRecordReader {
});
// Pass 1: Forward scan with 5 gates. The completed/inflight gate
(Gate 3)
- // is applied only when the timeline sets were supplied (table version
< 8).
+ // is applied when the caller supplied the timeline sets — a `Table`
read of a
+ // table below version 8. A v8+ read, or a caller with no timeline,
leaves it a no-op.
let completion_gate = self
.completion_gate_inputs
- .as_ref()
+ .as_deref()
.map(CompletionGate::new);
let mut pass1 = forward_scan_pass1(
all_blocks,
@@ -855,7 +855,7 @@ mod tests {
assert_eq!(result.ordered_instants_list, vec!["20250101000000000"]);
}
- /// Table version < 8: an UNCOMMITTED "straddling" instant -- one whose
+ /// An UNCOMMITTED "straddling" instant -- one whose
/// time is BELOW the latest completed instant (the high-watermark) and
which has NO in-log
/// rollback command block (a pending/failed write, or a concurrent
writer's inflight commit) --
/// must be EXCLUDED (not merged) once the completed/inflight sets are
supplied.
@@ -901,10 +901,10 @@ mod tests {
);
}
- /// Without the completion gate (v8+ tables, or no timeline available)
Gate 3 is a no-op, so the
- /// straddling instant is still admitted -- proving the fix is scoped to
table version < 8 and is
- /// purely additive (prior callers pass `None` and see unchanged
behavior). v8+ safety is
- /// enforced elsewhere: per-delta-commit log files are excluded at the
completion-time file-slice
+ /// Without the completion gate -- a v8+ read, or a caller with no
timeline (the cxx bridge)
+ /// -- Gate 3 is a no-op, so the straddling instant is still admitted,
proving the gate is
+ /// purely additive and that such callers see unchanged behavior. On v8+
that is safe for a
+ /// separate reason: per-delta-commit log files are dropped at the
completion-time file-slice
/// level, so the same block never reaches Pass 1.
#[test]
fn test_pass1_gate3_absent_gate_admits_straddling_instant() {
@@ -923,6 +923,44 @@ mod tests {
);
}
+ /// Where the gate is supplied it applies to incremental reads too, not
just snapshots, so it
+ /// has to COMPOSE with Gate 4's window rather than widen it. Two claims
are pinned here: a
+ /// pending instant inside the incremental window is still excluded (Gate
3 applies even though
+ /// a range is set), and a committed instant outside the window stays
excluded (Gate 3 admitting
+ /// it cannot override Gate 4). Together they show the gate only ever
subtracts, so arming it
+ /// on a read never costs rows that read should have seen.
+ #[test]
+ fn test_pass1_gate3_and_the_incremental_window_only_ever_subtract() {
+ let blocks = vec![
+ make_data_block("20250101000000000"), // committed, inside the
window
+ make_data_block("20250102000000000"), // PENDING, inside the window
+ make_data_block("20250104000000000"), // committed, outside the
window
+ ];
+ let range = Some(InstantRange::up_to("20250103000000000", "utc"));
+
+ let gate_inputs = CompletionGateInputs {
+ completed_instants: [
+ "20250101000000000".to_string(),
+ "20250104000000000".to_string(),
+ ]
+ .into_iter()
+ .collect(),
+ inflight_instants:
["20250102000000000".to_string()].into_iter().collect(),
+ archived_boundary: None,
+ };
+ let gate = CompletionGate::new(&gate_inputs);
+
+ let result =
+ forward_scan_pass1(blocks, MAX_INSTANT_TIME, &range, "utc",
Some(&gate)).unwrap();
+
+ assert_eq!(
+ result.ordered_instants_list,
+ vec!["20250101000000000"],
+ "pending 20250102 excluded by Gate 3 despite being in range; \
+ committed 20250104 excluded by Gate 4 despite Gate 3 admitting it"
+ );
+ }
+
/// Gate 3 treats an instant older than the archived boundary as committed
(archival only removes
/// completed instants), so an old archived data block is admitted while a
genuinely inflight one
/// is still excluded.
@@ -1789,4 +1827,46 @@ mod tests {
"inflight straddling delta t1 (id=2) must be excluded"
);
}
+
+ /// REGRESSION: a block from an instant that never completed is admitted
when
+ /// no completion gate is supplied, and skipped when one is.
+ ///
+ /// This is the "straddling" case: writer A opens an instant at T1 and is
+ /// still inflight when writer B commits T2 > T1. `latest_instant_time` is
+ /// T2, so A's blocks pass the future gate and the instant-range gate, and
+ /// nothing else looks at whether A ever committed. The log file itself was
+ /// admitted at listing time on its own (committed) instant, so the
+ /// file-level check does not see these blocks either.
+ ///
+ /// The pair is the point: same blocks, same bounds, gate absent vs
present.
+ #[test]
+ fn test_pass1_admits_an_uncommitted_instant_without_the_completion_gate() {
+ let blocks = vec![make_data_block("T1_inflight"),
make_data_block("T2_done")];
+
+ // No gate — what every production read did before the gate was wired.
+ let ungated = forward_scan_pass1(blocks.clone(), "T2_done", &None,
"UTC", None).unwrap();
+ assert!(
+ ungated.instant_to_blocks_map.contains_key("T1_inflight"),
+ "without a gate an inflight instant's blocks are merged: {:?}",
+ ungated.ordered_instants_list
+ );
+
+ // With the timeline's own sets, the inflight instant is excluded.
+ let inputs = CompletionGateInputs {
+ completed_instants: HashSet::from(["T2_done".to_string()]),
+ inflight_instants: HashSet::from(["T1_inflight".to_string()]),
+ archived_boundary: Some("T0".to_string()),
+ };
+ let gate = CompletionGate::new(&inputs);
+ let gated = forward_scan_pass1(blocks, "T2_done", &None, "UTC",
Some(&gate)).unwrap();
+
+ assert!(
+ !gated.instant_to_blocks_map.contains_key("T1_inflight"),
+ "the gate must skip an instant that never completed"
+ );
+ assert!(
+ gated.instant_to_blocks_map.contains_key("T2_done"),
+ "the committed instant must still be merged"
+ );
+ }
}
diff --git a/crates/core/src/file_group/reader_v2/merged_log_record_reader.rs
b/crates/core/src/file_group/reader_v2/merged_log_record_reader.rs
index a5c7aef3..667bf1f4 100644
--- a/crates/core/src/file_group/reader_v2/merged_log_record_reader.rs
+++ b/crates/core/src/file_group/reader_v2/merged_log_record_reader.rs
@@ -274,7 +274,7 @@ pub struct Builder {
allow_inflight_instants: bool,
/// Inputs for the Gate-3 completed/inflight check; `Some` only for
/// table version < 8. `None` (default) leaves Gate 3 a no-op.
- completion_gate_inputs: Option<CompletionGateInputs>,
+ completion_gate_inputs: Option<Arc<CompletionGateInputs>>,
}
impl Default for Builder {
@@ -347,7 +347,10 @@ impl Builder {
/// Supply the Gate-3 completed/inflight sets. Callers pass `Some` only for
/// table version < 8; `None` (the default) leaves Gate 3 disabled.
- pub fn with_completion_gate_inputs(mut self, inputs:
Option<CompletionGateInputs>) -> Self {
+ pub fn with_completion_gate_inputs(
+ mut self,
+ inputs: Option<Arc<CompletionGateInputs>>,
+ ) -> Self {
self.completion_gate_inputs = inputs;
self
}
diff --git a/crates/core/src/file_group/reader_v2/reader_context.rs
b/crates/core/src/file_group/reader_v2/reader_context.rs
index fec6b770..9acf81bd 100644
--- a/crates/core/src/file_group/reader_v2/reader_context.rs
+++ b/crates/core/src/file_group/reader_v2/reader_context.rs
@@ -31,6 +31,7 @@ use crate::config::table::HudiTableConfig;
use crate::storage::RowFilterBuilder;
use crate::timeline::selector::InstantRange;
use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
/// Owned inputs for the Gate-3 completed/inflight check in the log scan.
///
@@ -51,6 +52,12 @@ pub struct CompletionGateInputs {
/// Active completed-commit instant times (Java
`filterCompletedInstants()`).
pub completed_instants: HashSet<String>,
/// Active inflight/requested instant times (Java `filterInflights()`).
+ ///
+ /// Carried for parity with Java's two-set check; it cannot change an
outcome for the
+ /// producer in this crate. `Table` subtracts the completed set when
building this one, and
+ /// the archived boundary is the minimum over *all* active instants
including these — so an
+ /// instant listed here fails the committed test already, and the
`!inflight` term never
+ /// decides anything. A different producer, supplying sets that overlap,
would need it.
pub inflight_instants: HashSet<String>,
/// First active-timeline instant. An instant strictly before it is
archived; archival only
/// removes completed instants, so an archived instant is committed.
`None` disables the
@@ -149,12 +156,17 @@ pub struct ReaderContext {
/// the FG reader).
pub mor_pk_safe: bool,
/// Gate-3 completed/inflight inputs (completed/inflight/archived sets).
- /// Carried here mirroring [`Self::instant_range`]; `Some` only for table
- /// version < 8 snapshot reads (set by the FFI bridge when the planner
- /// enabled the gate), `None` otherwise (v8+/incremental/no-timeline) which
- /// leaves Gate 3 a no-op, preserving prior behavior. Consumed by the log
+ /// Carried here mirroring [`Self::instant_range`]. `Some` only when the
caller holds a
+ /// timeline *and* the table version is below 8 — see
+ /// [`crate::table::Table::completion_gate_inputs`]. `None` otherwise: for
v8+, where the
+ /// whole log file is dropped at slice-build time, and for a caller handed
bare paths with
+ /// no timeline (the cxx bridge). Either way Gate 3 becomes a no-op.
Consumed by the log
/// scan builder in `buffer::loader::scan_log_files`.
- pub completion_gate_inputs: Option<CompletionGateInputs>,
+ ///
+ /// Not scoped to snapshot reads, though: the gate only ever EXCLUDES
blocks whose instant
+ /// is pending, so it holds for incremental reads too and cannot admit a
block another gate
+ /// rejected.
+ pub completion_gate_inputs: Option<Arc<CompletionGateInputs>>,
}
// Manual Debug — RowFilterBuilder is a closure (Arc<dyn Fn ...>), not Debug.
diff --git a/crates/core/src/table/mod.rs b/crates/core/src/table/mod.rs
index e07c9180..e4cc9b04 100644
--- a/crates/core/src/table/mod.rs
+++ b/crates/core/src/table/mod.rs
@@ -107,6 +107,7 @@ use crate::error::CoreError;
use crate::expr::filter::{Filter, validate_fields_against_schemas};
use crate::file_group::file_slice::FileSlice;
use crate::file_group::reader::FileGroupReader;
+use crate::file_group::reader_v2::reader_context::CompletionGateInputs;
use crate::keygen::is_timestamp_based_keygen;
use crate::metadata::METADATA_TABLE_PARTITION_FIELD;
use crate::metadata::commit::HoodieCommitMetadata;
@@ -669,11 +670,58 @@ impl Table {
Some(opts) => self.prepare_reader_options(opts)?.hudi_options,
None => HashMap::new(),
};
- let mut reader = self.build_file_group_reader(hudi_opts,
extra_storage_overrides)?;
+ let reader = self.build_file_group_reader(hudi_opts,
extra_storage_overrides)?;
+ self.finish_reader(reader).await
+ }
+
+ /// Carry this table's per-read state onto a freshly built reader.
+ ///
+ /// Every read path needs the same two steps, and a path performing only
some
+ /// of them fails silently rather than loudly: without the data schema an
+ /// evolved table reads against the base file's, and without the gate
inputs
+ /// the log scan admits uncommitted blocks. Both return plausible rows. So
the
+ /// steps live here and each path calls this rather than repeating them.
+ async fn finish_reader(&self, mut reader: FileGroupReader) ->
Result<FileGroupReader> {
reader.set_data_schema(std::sync::Arc::new(self.data_schema_for_read().await?));
+ // This table holds the timeline, so it can tell a committed instant
from
+ // one still inflight — the log-block scan cannot work that out from
the
+ // slice alone.
+ if let Some(inputs) = self.completion_gate_inputs()? {
+ reader.set_completion_gate_inputs(inputs);
+ }
Ok(reader)
}
+ /// Inputs for the log-block scan's completed/inflight gate, or `None` when
+ /// this table does not need one.
+ ///
+ /// Mirrors Java `BaseHoodieLogRecordReader`, which runs the check only
below
+ /// table version 8 (`tableVersion.lesserThan(HoodieTableVersion.EIGHT)`).
+ /// From version 8 the timeline records completion times, so a log file
whose
+ /// delta commit never completed is already dropped when the file slice is
+ /// built, and asking again per block would be redundant. Below version 8
+ /// there are no completion times to build a slice from, which leaves the
+ /// block scan as the only place the question can be asked.
+ ///
+ /// So from version 8 the exclusion rests entirely on the log file's *name*
+ /// carrying the delta commit that wrote it (`file_group::builder`). A
table
+ /// upgraded from version 6 would seem to break that, since a version-6
+ /// writer names log files on the base instant — but the writer's upgrade
+ /// closes it from the other side, two ways: it rolls back failed writes
and
+ /// compacts every log file into a new base file before bumping the
version,
+ /// and where the pending commit has completed commits after it, rollback
is
+ /// refused and the upgrade aborts rather than proceeding. Either way no
+ /// version-6-named log file reaches a version-8 slice. Verified by running
+ /// both paths on Spark 3.5.3 with Hudi 1.2.0-SNAPSHOT.
+ fn completion_gate_inputs(&self) -> Result<Option<CompletionGateInputs>> {
+ let table_version: isize = self
+ .hudi_configs
+ .try_get(HudiTableConfig::TableVersion)?
+ .map(|v| v.into())
+ .unwrap_or(6);
+ Ok((table_version < 8).then(|| self.timeline.completion_gate_inputs()))
+ }
+
/// Build a reader for one of this table's own read paths, carrying the
/// table's current data schema.
///
@@ -681,12 +729,11 @@ impl Table {
/// performed only the first would read an evolved table with a stale
schema —
/// so they share this rather than repeating it.
async fn reader_for_read_path(&self, prepared: &ReadOptions) ->
Result<FileGroupReader> {
- let mut reader = self.build_file_group_reader(
+ let reader = self.build_file_group_reader(
prepared.hudi_options.clone(),
std::iter::empty::<(&str, &str)>(),
)?;
-
reader.set_data_schema(std::sync::Arc::new(self.data_schema_for_read().await?));
- Ok(reader)
+ self.finish_reader(reader).await
}
/// Convert caller-facing [`ReadOptions`] into the form that
@@ -1801,6 +1848,57 @@ mod tests {
Ok(())
}
+ /// The streaming read carries the completion gate too.
+ ///
+ /// It reaches the gate through its own pass-through, separate from the
eager read's, and
+ /// it is the path DataFusion and the Python binding use. The gold sweep
only exercises
+ /// the eager one, so without this a regression that disarmed the gate for
every streaming
+ /// read would leave the whole suite green — which is exactly what a
mutation of the
+ /// streaming pass-through did before this test existed.
+ ///
+ /// The fixture's orphaned delta commit sets `rider = 'ORPHANED-B'` at `ts
= 300`, above
+ /// every other row's ordering value, so an admitted orphan wins the row
outright rather
+ /// than losing the merge for an unrelated reason.
+ #[tokio::test]
+ async fn hudi_table_read_stream_excludes_an_uncommitted_instants_blocks()
-> Result<()> {
+ use arrow_array::Array;
+ use futures::TryStreamExt;
+ use hudi_test::QuickstartTripsTable;
+
+ let base_url =
QuickstartTripsTable::MorUncommittedLogV6.url_to_mor_avro();
+ let hudi_table = Table::new(base_url.path()).await.unwrap();
+
+ let stream =
hudi_table.read_stream(&ReadOptions::new()).await.unwrap();
+ let batches = stream.try_collect::<Vec<_>>().await.unwrap();
+
+ let riders: Vec<String> = batches
+ .iter()
+ .filter_map(|b| b.column_by_name("rider").cloned())
+ .flat_map(|c| {
+ let arr = c
+ .as_any()
+ .downcast_ref::<arrow_array::StringArray>()
+ .expect("rider is a string column")
+ .clone();
+ (0..arr.len()).map(move |i| arr.value(i).to_string())
+ })
+ .collect();
+
+ assert!(
+ !riders.is_empty(),
+ "the streaming read returned no rows, so it cannot show the gate
ran"
+ );
+ assert!(
+ !riders.iter().any(|r| r == "ORPHANED-B"),
+ "the streaming read merged a block from an instant that never
completed: {riders:?}"
+ );
+ assert!(
+ riders.iter().any(|r| r == "rider-B"),
+ "the base row the orphan would have overwritten must survive:
{riders:?}"
+ );
+ Ok(())
+ }
+
#[tokio::test]
async fn
hudi_table_read_snapshot_stream_returns_empty_when_no_file_slices_match_filters()
-> Result<()> {
@@ -2395,6 +2493,70 @@ mod tests {
/// fixture rather than a reproduction of the one input that broke: the
bound
/// is an instant the timeline actually contains, so it parses by
/// construction.
+ /// Regression test: a table-built reader carries the timeline's
+ /// committed/inflight sets, so the log-block scan can gate on them.
+ ///
+ /// The gate exists to skip blocks from an instant that never completed —
the
+ /// straddling case where a writer is still inflight when a later one
+ /// commits, so its blocks sort below the latest instant and pass every
other
+ /// gate. It was inert for every read through this crate because nothing
+ /// populated its inputs.
+ #[tokio::test]
+ async fn test_table_reader_carries_the_completion_gate_inputs() {
+ use hudi_test::SampleTable;
+
+ let base_url = SampleTable::V6Nonpartitioned.url_to_mor_parquet();
+ let table = Table::new(base_url.path()).await.unwrap();
+
+ let inputs = table.timeline.completion_gate_inputs();
+ assert!(
+ !inputs.completed_instants.is_empty(),
+ "the fixture's completed commits must reach the gate"
+ );
+ assert!(
+ inputs.archived_boundary.is_some(),
+ "the archival boundary is the gate's second half — an archived \
+ instant is committed by definition"
+ );
+
+ // And the reader the read paths use actually receives them.
+ let reader = table
+ .create_file_group_reader_with_options(None, empty_options())
+ .await
+ .unwrap();
+ assert!(
+ reader.has_completion_gate_inputs(),
+ "a reader built from a table must be able to gate the log scan"
+ );
+ }
+
+ /// From table version 8 the timeline records completion times, so a log
file
+ /// whose delta commit never completed is already dropped when the file
slice
+ /// is built. Java stops applying the per-block gate there
+ /// (`BaseHoodieLogRecordReader`, `tableVersion.lesserThan(EIGHT)`), and so
+ /// does this. Paired with the version-6 test above, the two pin the
+ /// condition rather than only the armed case.
+ #[tokio::test]
+ async fn test_completion_gate_is_not_armed_from_table_version_eight() {
+ use hudi_test::QuickstartTripsTable;
+
+ let table_path =
QuickstartTripsTable::V9MorNonpart3Commits.path_to_mor_avro();
+ let table = Table::new(&table_path).await.unwrap();
+ assert!(
+ table.completion_gate_inputs().unwrap().is_none(),
+ "a version 9 table must not arm the per-block gate"
+ );
+
+ let reader = table
+ .create_file_group_reader_with_options(None, empty_options())
+ .await
+ .unwrap();
+ assert!(
+ !reader.has_completion_gate_inputs(),
+ "the slice already excludes an uncommitted log file on this layout"
+ );
+ }
+
#[tokio::test]
async fn test_resolve_incremental_window_start_bound_is_a_real_instant() {
use crate::config::internal::HudiInternalConfig;
diff --git a/crates/core/src/timeline/mod.rs b/crates/core/src/timeline/mod.rs
index 5e08df1f..42c4ba0c 100644
--- a/crates/core/src/timeline/mod.rs
+++ b/crates/core/src/timeline/mod.rs
@@ -30,6 +30,7 @@ use crate::config::HudiConfigs;
use crate::error::CoreError;
use crate::file_group::FileGroup;
use crate::file_group::builder::replaced_file_groups_from_replace_commit;
+use crate::file_group::reader_v2::reader_context::CompletionGateInputs;
use crate::schema::resolver::{
resolve_avro_schema_from_commit_metadata,
resolve_data_schema_from_commit_metadata,
};
@@ -69,9 +70,28 @@ pub struct Timeline {
/// (`BaseHoodieTimeline.java:494`). `None` means the active timeline is
/// empty, so nothing can be treated as archived.
pub(crate) earliest_active_instant: Option<String>,
+ /// Request timestamps of the active instants that have NOT completed —
+ /// requested or inflight.
+ ///
+ /// Read from the same listing that yields [`Self::completed_commits`] and
+ /// the archival boundary, so retaining it costs nothing. The log-block
scan
+ /// needs it: a log file admitted at listing time on its own committed
+ /// instant can still carry blocks appended by a later instant that never
+ /// completed, and only the instant's state can tell.
+ pub(crate) pending_instants: HashSet<String>,
}
pub const EARLIEST_START_TIMESTAMP: &str = "19700101000000000";
+
+/// The actions a read loads from the timeline.
+///
+/// Narrowing this is no longer only a listing decision:
[`Timeline::completion_gate_inputs`]
+/// derives the log scan's committed set from these, so an action that
completes outside this
+/// list and wrote data or delete log blocks would have its rows dropped from
a merge-on-read
+/// read — silently, since the gate excludes rather than errors. The three
here cover every
+/// action that writes such blocks (compaction completes as `commit`,
clustering as
+/// `replacecommit`, log compaction as `deltacommit`), which is also the set
Java's gate is
+/// built from via `getCommitsTimeline()`.
pub const DEFAULT_LOADING_ACTIONS: &[Action] =
&[Action::Commit, Action::DeltaCommit, Action::ReplaceCommit];
@@ -89,6 +109,7 @@ impl Timeline {
archived_loader,
completed_commits: Vec::new(),
earliest_active_instant: None,
+ pending_instants: HashSet::new(),
}
}
@@ -114,13 +135,47 @@ impl Timeline {
.iter()
.map(|instant| instant.timestamp.clone())
.min();
- timeline.completed_commits = all_active
+ let (completed, pending): (Vec<Instant>, Vec<Instant>) = all_active
+ .into_iter()
+ .partition(|instant| instant.state == State::Completed);
+ // An instant is listed once per state file it has, so a completed one
+ // also appears as requested and inflight. Pending means it reached NO
+ // completed state — subtracting is what makes that true, and without
the
+ // subtraction every committed instant reads as inflight and the gate
+ // rejects the whole timeline.
+ let completed_times: HashSet<String> = completed
+ .iter()
+ .map(|instant| instant.timestamp.clone())
+ .collect();
+ timeline.pending_instants = pending
.into_iter()
- .filter(|instant| instant.state == State::Completed)
+ .map(|instant| instant.timestamp)
+ .filter(|timestamp| !completed_times.contains(timestamp))
.collect();
+ timeline.completed_commits = completed;
Ok(timeline)
}
+ /// The inputs the log-block scan needs to tell a committed instant from
one
+ /// that never finished.
+ ///
+ /// A log file is admitted to a slice on its own instant, but blocks inside
+ /// it carry their own — including a writer that was still inflight when a
+ /// later writer committed. Without these sets that block merges as if it
+ /// were committed, because it sorts below the latest instant and so passes
+ /// every other gate.
+ pub(crate) fn completion_gate_inputs(&self) -> CompletionGateInputs {
+ CompletionGateInputs {
+ completed_instants: self
+ .completed_commits
+ .iter()
+ .map(|instant| instant.timestamp.clone())
+ .collect(),
+ inflight_instants: self.pending_instants.clone(),
+ archived_boundary: self.earliest_active_instant.clone(),
+ }
+ }
+
/// Load instants from the timeline based on the selector criteria.
///
/// # Archived Timeline Loading
@@ -898,4 +953,39 @@ mod tests {
assert!(timestamp.is_some());
assert!(!timestamp.unwrap().is_empty());
}
+
+ /// Regression test: a completed instant is not also reported as pending.
+ ///
+ /// The active timeline lists an instant once per state file it has, so a
+ /// completed one appears as requested and inflight too. Reading pending
+ /// straight off the non-completed rows therefore marks every committed
+ /// instant inflight, and the log-scan gate — which admits only
+ /// `committed && !inflight` — then rejects the whole timeline and returns
+ /// base-file-only data. That is silent: no error, just missing log deltas.
+ #[tokio::test]
+ async fn
test_completion_gate_inputs_do_not_report_completed_instants_as_pending() {
+ let base_url = SampleTable::V6Nonpartitioned.url_to_mor_parquet();
+ let hudi_configs = Arc::new(HudiConfigs::new([(
+ HudiTableConfig::BasePath,
+ base_url.to_string(),
+ )]));
+ let timeline = Timeline::new_from_storage(hudi_configs,
Arc::new(HashMap::new()))
+ .await
+ .unwrap();
+
+ let inputs = timeline.completion_gate_inputs();
+ assert!(
+ !inputs.completed_instants.is_empty(),
+ "the fixture has completed commits"
+ );
+ let both: Vec<&String> = inputs
+ .inflight_instants
+ .iter()
+ .filter(|t| inputs.completed_instants.contains(*t))
+ .collect();
+ assert!(
+ both.is_empty(),
+ "an instant cannot be both completed and pending, got {both:?}"
+ );
+ }
}
diff --git a/crates/core/tests/gold_parity_tests.rs
b/crates/core/tests/gold_parity_tests.rs
index a1f4dfd7..56b11d13 100644
--- a/crates/core/tests/gold_parity_tests.rs
+++ b/crates/core/tests/gold_parity_tests.rs
@@ -115,6 +115,8 @@ const EXPECTED_OPTION_FIXTURES: &[&str] = &[
"table_null_containers [MorAvro]",
"table_parquet_log_block [MorAvro]",
"table_partial_update [MorAvro]",
+ "table_uncommitted_log_v6 [MorAvro]",
+ "table_uncommitted_log_v9 [MorAvro]",
"v6_trips_8i1u [MorAvro]",
"v6_trips_8i3d [MorAvro]",
"v8_mor_boundary_windows [MorAvro]",
@@ -503,6 +505,26 @@ impl Known {
/// disagreement fails the build, and one that starts passing has to be removed
/// from here, so this list cannot quietly go stale.
const KNOWN: &[Known] = &[
+ // Version 1 only: it applies no completed/inflight check to log blocks, so
+ // the blocks of a delta commit that never completed still reach the merge.
+ // The fixture exists to pin that version 2 does check; this records that
+ // version 1 does not, which is the divergence itself rather than a fixture
+ // flaw. Only the version 6 fixture separates the two readers: on version 9
+ // the log file is dropped when the slice is built, which both of them
share.
+ //
+ // Unlike the other version 1 entries here, which record a narrower or
+ // differently-shaped answer, this one records a wrong one: rows that were
+ // never committed. It is accepted rather than fixed because version 2 is
the
+ // default and version 1 is on its way out. Two things would reopen that: a
+ // version 1 read reachable without asking for it, or version 1 outliving
the
+ // migration. Fixing it means giving `LogFileScanner` the instant state its
+ // `scan` has no notion of, which reaches the metadata table reader too.
+ Known {
+ fixture: "table_uncommitted_log_v6",
+ scope: CaseScope::Any,
+ reader_version: "1",
+ reason: "version 1 admits blocks from an instant that never completed",
+ },
// Version 1 only: it now reads a log-only slice but returns it without the
// base columns. Nothing it is asked to project changes that, so the
// fixture is blanketed.
diff --git
a/crates/test/data/quickstart_trips_table/mor/avro/table_uncommitted_log.sql
b/crates/test/data/quickstart_trips_table/mor/avro/table_uncommitted_log.sql
new file mode 100644
index 00000000..f89dedb0
--- /dev/null
+++ b/crates/test/data/quickstart_trips_table/mor/avro/table_uncommitted_log.sql
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+-- Source for BOTH `table_uncommitted_log_v6.zip` and
`table_uncommitted_log_v9.zip`:
+-- a delta commit that wrote log blocks and never completed.
+--
+-- Run once per table version, substituting 6 and 9 for WRITE_TABLE_VERSION and
+-- naming the table table_uncommitted_log_v6 / table_uncommitted_log_v9.
+-- Generated with Spark 3.5.3 and Hudi 1.2.0-SNAPSHOT.
+--
+-- Two steps follow the SQL and cannot be expressed in it:
+--
+-- 1. Delete the SECOND delta commit's completed timeline file, leaving its
+-- `.inflight` and `.requested` and its log file in place. That is what a
+-- writer killed mid-commit leaves behind. Layout v1 (version 6) keeps the
+-- completed file at `.hoodie/<instant>.deltacommit`; layout v2 (version
9)
+-- at `.hoodie/timeline/<requested>_<completion>.deltacommit`.
+--
+-- 2. Read the table with Spark afterwards and write that to `gold_data`, so
+-- the gold is Hudi's own answer to the doctored layout.
+--
+-- Why the orphan is the middle commit and not the last: an orphan at the end
+-- sorts above the latest committed instant, so the future-block gate discards
it
+-- and the completed/inflight gate is never reached.
+--
+-- Why both table versions: on version 6 every log file carries the base
instant
+-- in its name, so the file cannot be attributed to a delta commit and only the
+-- per-block check can exclude the orphaned blocks. From version 8 each log
file
+-- carries its own delta commit, so the whole file is dropped when the file
slice
+-- is built and the per-block check is redundant, which is the condition Java
+-- applies in `BaseHoodieLogRecordReader`.
+
+CREATE TABLE table_uncommitted_log_vN (
+ ts BIGINT,
+ uuid STRING,
+ rider STRING,
+ fare DOUBLE
+) USING HUDI
+TBLPROPERTIES (
+ type = 'mor',
+ primaryKey = 'uuid',
+ preCombineField = 'ts',
+ 'hoodie.write.table.version' = 'WRITE_TABLE_VERSION',
+ 'hoodie.metadata.enable' = 'false',
+ 'hoodie.parquet.small.file.limit' = '0',
+ 'hoodie.compact.inline' = 'false'
+)
+LOCATION 'FIXTURE_LOCATION';
+
+INSERT INTO table_uncommitted_log_vN VALUES
+ (100, 'a', 'rider-A', 10.0),
+ (100, 'b', 'rider-B', 20.0),
+ (100, 'c', 'rider-C', 30.0),
+ (100, 'd', 'rider-D', 40.0);
+
+-- This commit is the one whose completed timeline file is deleted afterwards.
+-- Its blocks must not reach the merge, so `b` keeps the base row.
+UPDATE table_uncommitted_log_vN SET fare = 99.0, rider = 'ORPHANED-B', ts =
300 WHERE uuid = 'b';
+
+-- Committed, and at a later instant, so the orphan above sits below the latest
+-- committed instant rather than above it.
+UPDATE table_uncommitted_log_vN SET fare = 22.0, rider = 'committed-A', ts =
200 WHERE uuid = 'a';
diff --git
a/crates/test/data/quickstart_trips_table/mor/avro/table_uncommitted_log_v6.zip
b/crates/test/data/quickstart_trips_table/mor/avro/table_uncommitted_log_v6.zip
new file mode 100644
index 00000000..be987586
Binary files /dev/null and
b/crates/test/data/quickstart_trips_table/mor/avro/table_uncommitted_log_v6.zip
differ
diff --git
a/crates/test/data/quickstart_trips_table/mor/avro/table_uncommitted_log_v9.zip
b/crates/test/data/quickstart_trips_table/mor/avro/table_uncommitted_log_v9.zip
new file mode 100644
index 00000000..9a09aa0c
Binary files /dev/null and
b/crates/test/data/quickstart_trips_table/mor/avro/table_uncommitted_log_v9.zip
differ
diff --git a/crates/test/src/lib.rs b/crates/test/src/lib.rs
index 21d4116c..2eabda68 100644
--- a/crates/test/src/lib.rs
+++ b/crates/test/src/lib.rs
@@ -329,6 +329,52 @@ pub enum QuickstartTripsTable {
#[strum(serialize = "table_event_time_stale")]
MorEventTimeStale,
+ //
-------------------------------------------------------------------------
+ // A delta commit that wrote log blocks and never completed: what a writer
+ // killed mid-commit leaves behind. Its blocks must not reach the merge.
+ //
+ // The two versions take different routes to that answer, which is why both
+ // are here. Below table version 8 the timeline records no completion
times,
+ // so the log-block scan itself has to check the instant's state. From
+ // version 8 the completion times exist and the log file is dropped when
the
+ // file slice is built, so the per-block check is redundant and Java skips
it
+ // (`BaseHoodieLogRecordReader`, `tableVersion.lesserThan(EIGHT)`). A
fixture
+ // for only one version would leave the other route unexercised.
+ //
+ // Provenance: `table_uncommitted_log.sql` beside these zips (Spark 3.5.3 /
+ // Hudi 1.2.0-SNAPSHOT). Generated normally — a base write then two
updates —
+ // after which the *middle* delta commit's completed timeline file was
+ // deleted, leaving its `.inflight` and `.requested` and its log file
behind.
+ // gold_data is Spark reading the table in that state, so it is Hudi's own
+ // answer to the doctored layout.
+ //
+ // The middle one, not the last: an orphan at the end sorts above the
latest
+ // committed instant, so the future-block gate discards it and the
+ // completed/inflight gate is never reached. The first attempt at this
+ // fixture orphaned the last commit and passed with the gate disarmed —
+ // regenerate it that way and the sweep goes green proving nothing.
+ //
+ // Schema: ts LONG, uuid STRING, rider STRING, fare DOUBLE
(non-partitioned).
+ // Layout: base .parquet, 4 rows at ts 100, then two log files — the
orphaned
+ // update of `b` at ts 300, then the committed update of `a` at ts 200. The
+ // versions name those files differently, which is the whole point: on v6
+ // both are `.log.1`/`.log.2` carrying the *base* instant, so no file-level
+ // check can attribute either to a delta commit and only the per-block gate
+ // can exclude the orphan; on v9 each is a `.log.1` carrying its own delta
+ // commit, so the orphan's file is dropped when the slice is built.
+ // Semantics: `a` takes its update, `b` keeps the base row. gold_data = 4
rows.
+ // The orphan's ts 300 outranks every other row, so an admitted orphan
would
+ // win under `preCombineField = 'ts'` — the fixture fails loudly, not
silently.
+ //
-------------------------------------------------------------------------
+ /// Table version 6: the orphaned blocks are excluded by the log-block
scan's
+ /// completed/inflight gate.
+ #[strum(serialize = "table_uncommitted_log_v6")]
+ MorUncommittedLogV6,
+ /// Table version 9: the orphaned blocks are excluded when the file slice
is
+ /// built, with no per-block gate involved.
+ #[strum(serialize = "table_uncommitted_log_v9")]
+ MorUncommittedLogV9,
+
//
-------------------------------------------------------------------------
// Delete-block orderingVal wrapper-type fixtures (Task 7).
// Each table: v9 MOR, COMMIT_TIME_ORDERING, NON_PARTITIONED, 4 rows
inserted
