yihua commented on code in PR #660:
URL: https://github.com/apache/hudi-rs/pull/660#discussion_r3781318351
##########
crates/core/src/table/mod.rs:
##########
@@ -155,6 +158,19 @@ impl Clone for Table {
}
}
+/// One tick below `instant_time`, as a same-width instant string.
+///
+/// Used to turn an exclusive start bound into an inclusive one. Instant times
are
+/// fixed-width numeric strings, so this is a plain decrement; a value that
does
+/// not parse (the bootstrap sentinels, for instance) is returned unchanged,
which
+/// keeps the bound no narrower than it was.
+fn instant_time_minus_one(instant_time: &str) -> String {
Review Comment:
This decrements the instant as an integer, so it can land on a string that
sorts correctly but isn't a valid time — `20250713010500000 - 1` is
`20250713010499999`, i.e. `SS=99`. That value becomes `StartTimestamp`, and
`actions_in_range` parses it, so an incremental read whose earliest admitted
commit has `ss=00` and `SSS=000` fails with `Timeline("Invalid epoch millis")`
(I reproduced this by renaming a fixture's first commit). Could the exclusive
bound be kept as a flag on the selector instead of being encoded as a "one tick
below" string?
_⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._
##########
.github/workflows/ci.yml:
##########
@@ -81,6 +83,29 @@ jobs:
path: ./cov-reports
if-no-files-found: 'error'
+ # hudi-core without `spill-rocksdb` must keep building and passing. That
build
+ # is the reason the feature exists — it needs no libclang and no C++
toolchain,
+ # so a consumer that cannot take a native build (or does not want one in a
+ # Python wheel) has a configuration to use. Nothing else in CI compiles it,
so
+ # without this job it would break silently.
+ rust-tests-no-spill-backend:
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@v6
+ - name: Cache dependencies
+ uses: swatinem/rust-cache@6323deb1c3ea99e6f4dae6b7d3ac1b73ac21ff77 #
v2.9.2
Review Comment:
This SHA doesn't resolve — `6323deb1c3ea99e6f4dae6b7d3ac1b73ac21ff77` isn't
a commit in swatinem/rust-cache. v2.9.2 is
`6323deb102c322ba6fcbdcafc7e3dddab59af2b6`, which is what the other two jobs in
this file pin. The job currently fails at "Set up job", so the no-spill leg
never runs. (The build itself is fine — I ran both commands locally.)
_⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._
##########
crates/core/src/file_group/reader_v2/gold_tests.rs:
##########
@@ -0,0 +1,335 @@
+/*
+ * 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.
+ */
+
+//! Reads each merge-on-read layout fixture and checks the result against the
+//! Spark snapshot shipped inside it.
+//!
+//! The fixtures carry their own `gold_data/` — a `SELECT *` of the table taken
+//! from Spark — so what is asserted here is agreement with Hudi's reference
+//! reader, not with this reader's own prior output.
+//!
+//! The file slice is discovered from the extracted fixture rather than written
+//! down per case: each of these tables is a single file group, so the base
file
+//! and log files are whatever is on disk. A fixture that gains a file
therefore
+//! cannot silently stop being covered.
+
+#![cfg(test)]
+
+use crate::config::HudiConfigs;
+use crate::config::read::HudiReadConfig;
+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_parameters::ReaderParameters;
+use crate::file_group::reader_v2::resolver::resolve_reader_context;
+use crate::storage::Storage;
+use arrow_array::RecordBatch;
+use std::path::Path;
+use std::sync::Arc;
+
+/// The base file and log files of a fixture's single file group, relative to
+/// the table root, plus the partition they live in.
+struct Slice {
+ base: Option<String>,
+ logs: Vec<String>,
+ partition: String,
+}
+
+/// Walk an extracted fixture and pick out its one file group.
+///
+/// Skips `.hoodie` (table metadata) and `gold_data` (the Spark snapshot this
+/// compares against, which is not table data).
+fn discover_slice(table_root: &Path) -> Slice {
+ fn walk(dir: &Path, root: &Path, out: &mut Vec<(String, String)>) {
+ let Ok(entries) = std::fs::read_dir(dir) else {
+ return;
+ };
+ for entry in entries.flatten() {
+ let path = entry.path();
+ let name = entry.file_name().to_string_lossy().to_string();
+ if path.is_dir() {
+ if name == ".hoodie" || name == "gold_data" {
+ continue;
+ }
+ walk(&path, root, out);
+ } else {
+ let rel = path
+ .strip_prefix(root)
+ .expect("walked path is under the table root")
+ .to_string_lossy()
+ .to_string();
+ let partition = Path::new(&rel)
+ .parent()
+ .map(|p| p.to_string_lossy().to_string())
+ .unwrap_or_default();
+ out.push((rel, partition));
+ }
+ }
+ }
+
+ let mut files = Vec::new();
+ walk(table_root, table_root, &mut files);
+
+ let mut base = None;
+ let mut logs = Vec::new();
+ let mut partition = String::new();
+ for (rel, part) in files {
+ let name = Path::new(&rel)
+ .file_name()
+ .map(|n| n.to_string_lossy().to_string())
+ .unwrap_or_default();
+ // Hadoop writes a `..<file>.crc` checksum sidecar next to each data
+ // file. It is not table data and must not reach the readers.
+ if name.ends_with(".crc") {
+ continue;
+ }
+ if name.ends_with(".parquet") && !name.starts_with('.') {
+ base = Some(rel);
+ partition = part;
+ } else if name.contains(".log.") {
+ logs.push(rel);
+ partition = part;
+ }
+ }
+ // Log files are appended in order; the scan relies on that ordering.
+ logs.sort();
+
+ Slice {
+ base,
+ logs,
+ partition,
+ }
+}
+
+/// Read a fixture's file group and return the merged rows.
+async fn read_fixture(table_path: &str) -> crate::Result<RecordBatch> {
+ let slice = discover_slice(Path::new(table_path));
+
+ // Load the table's own properties rather than inventing them: the merge
+ // semantics live in `hoodie.properties`, and inventing configs would test
+ // the reader against a table that does not exist.
+ let mut resolver = crate::table::builder::OptionResolver::new_with_options(
+ table_path,
+ [(
+ HudiReadConfig::EndTimestamp.as_ref(),
+ MAX_INSTANT_TIME.to_string(),
+ )],
+ );
+ resolver.resolve_options().await?;
+ let configs = HudiConfigs::new(resolver.hudi_options.clone());
+ let storage = Storage::new(
+ Arc::new(resolver.storage_options),
+ Arc::new(configs.clone()),
+ )?;
+
+ let has_logs = !slice.logs.is_empty();
+ let mut context = resolve_reader_context(&configs, has_logs)?;
+ // The fixtures are read whole, so nothing bounds the log scan.
+ context.instant_range = None;
+ context.rebuild_record_context(slice.partition.clone());
+
+ let input_split = InputSplit::new(slice.base, None, slice.logs,
slice.partition);
+
+ // The table's own schema, the way a real read resolves it. Passing `None`
+ // here — as this harness used to — makes the reader fall back to the base
+ // file's schema, which is stale on any table whose columns were widened
+ // after that file was written. The fixtures are then read into the older
+ // shape and the comparison below, which only counts rows, cannot see it.
+ let table = crate::table::Table::new(table_path).await?;
+ let data_schema = Arc::new(table.data_schema_for_read().await?);
+
+ let mut reader = HoodieFileGroupReader::new(
+ Arc::new(context),
+ storage,
+ input_split,
+ ReaderParameters::default(),
+ Some(data_schema),
+ None,
+ )?;
+ reader.read().await
+}
+
+/// Fixtures the reader reproduces today.
+const GOLD_FIXTURES: &[&str] = &[
+ "table_log_only",
+ "table_log_compaction",
+ "table_parquet_log_block",
+ "table_partial_update",
+ "table_evo_add_col",
+ "table_evo_promotion",
+];
+
+/// Fixtures the reader does not read yet, with what stops each one.
+///
+/// Two causes remain, neither in the merge:
+///
+/// **Avro maps are modelled as a malformed Arrow type.** `avro_to_arrow` turns
+/// an Avro map into `Dictionary(Utf8, V)`, but an Arrow dictionary key must be
+/// an integer, so the array builder rejects it. Avro maps belong in Arrow's
+/// `Map` type. Changing that alters the schema of every table with a map
+/// column, including on the existing read path.
+///
+/// **The delete record's ordering value has two shapes in the wild.** Hudi
+/// moved it into per-type wrapper records — `IntWrapper`, `DecimalWrapper` and
+/// so on — because Avro forbids a union holding two branches of the same
+/// underlying type. This crate carries only the older primitive union, so a
log
+/// file written the newer way resolves each branch index to the wrong type.
+///
+/// The two shapes cannot be told apart by decoding: a wrapper is a record
whose
+/// single field is the primitive, so `IntWrapper` and a bare `long` encode to
+/// the same bytes. Only the branch index differs. Reading both therefore needs
+/// to key off the writer version rather than the payload, which is a design
+/// question rather than a decoding one — and swapping to the newer schema
+/// outright breaks the fixtures written the older way.
+///
+/// **A corrupt tail block is not recognised.** `LogFileReader` has no
+/// corrupt-block detection — `create_corrupted_block_if_needed` returns `None`
+/// — so it parses the trailing garbage as a block.
+const KNOWN_GAPS: &[(&str, &str)] = &[
Review Comment:
Running this with `--ignored` fails its own second assertion: all ten
`KNOWN_GAPS` entries now read correctly, including the four
`table_delete_ord_*` ones. The doc above ("two causes remain — avro maps as
Dictionary, wrapped delete ordering values") is stale in the same way. Since
`tests/gold_parity_tests.rs` covers the same fixtures with a live ratchet, is
this harness still earning its keep, or could it be folded in?
_⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._
##########
crates/core/tests/reader_concurrency_tests.rs:
##########
@@ -0,0 +1,341 @@
+/*
+ * 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.
+ */
+//! Reader-vs-writer concurrency repros.
+//!
+//! Most of these are `#[ignore]`d, for one of two reasons: they document a
+//! divergence from Hudi's Java reader that is not fixed yet, or they are
probes
+//! that print what the reader does without asserting anything. Neither belongs
+//! in the suite — the first would fail it, and the second would pass whatever
+//! happened. Run them with
+//! `cargo test -p hudi-core --test reader_concurrency_tests -- --ignored
--nocapture`.
+//!
+//! A divergence that gets fixed should leave this file, or become an ordinary
+//! test that pins the fixed behavior. An `#[ignore]` reason describing
something
+//! that no longer happens is worse than no note at all.
+
+use hudi_core::table::{ReadOptions, Table};
+use std::path::{Path, PathBuf};
+
+fn rows(batches: &[arrow_array::RecordBatch]) -> usize {
+ batches.iter().map(|b| b.num_rows()).sum()
+}
+
+/// Copy an existing base file under a NEW file id at `instant`, leaving no
+/// commit for that instant — a writer that wrote its data file and then died.
+fn plant_orphan_base_file(table_dir: &Path, instant: &str) -> PathBuf {
+ let existing = std::fs::read_dir(table_dir)
+ .unwrap()
+ .filter_map(|e| e.ok())
+ .map(|e| e.path())
+ .find(|p| p.extension().is_some_and(|x| x == "parquet"))
+ .expect("fixture has a base file to copy");
+ let orphan = table_dir.join(format!(
+ "b179bdb3-731c-4894-b855-abfcd6921008-0_0-1-1_{instant}.parquet"
+ ));
+ std::fs::copy(&existing, &orphan).unwrap();
+ orphan
+}
+
+fn fixture_zip(relative: &str) -> PathBuf {
+ PathBuf::from(env!("CARGO_MANIFEST_DIR").replace("/core",
"/test")).join(relative)
+}
+
+async fn probe(zip: &str, label: &str) {
+ probe_with_reader_version(zip, label, "1").await;
+ probe_with_reader_version(zip, label, "2").await;
+}
+
+async fn probe_with_reader_version(zip: &str, label: &str, reader_version:
&str) {
+ use hudi_core::config::read::HudiReadConfig;
+ let dir = hudi_test::extract_test_table_fresh(&fixture_zip(zip));
+ let table_dir = std::fs::read_dir(&dir)
+ .unwrap()
+ .filter_map(|e| e.ok())
+ .map(|e| e.path())
+ .find(|p| p.is_dir() && p.join(".hoodie").exists())
+ .expect("extracted fixture holds one table dir");
+
+ let opts = ReadOptions::new().with_hudi_option(
+ HudiReadConfig::FileGroupReaderVersion.as_ref(),
+ reader_version,
+ );
+ let table = Table::new(table_dir.to_str().unwrap()).await.unwrap();
+ let before = rows(&table.read(&opts).await.unwrap());
+
+ // One tick below the latest commit: above the archival boundary (so it
reads
+ // as pending, not as archived) and at or below the snapshot bound (so the
+ // as-of filter does not exclude it either). That is an in-flight commit.
+ let latest_ts = table
+ .get_timeline()
+ .get_latest_commit_timestamp()
+ .unwrap()
+ .to_string();
+ let orphan_instant = format!(
+ "{:0width$}",
+ latest_ts.parse::<u64>().unwrap() - 1,
+ width = latest_ts.len()
+ );
+ let orphan_instant = orphan_instant.as_str();
+ plant_orphan_base_file(&table_dir, orphan_instant);
+
+ let table = Table::new(table_dir.to_str().unwrap()).await.unwrap();
+ let after = rows(&table.read(&opts).await.unwrap());
+ println!(
+ "{label:34} reader version={reader_version} rows before={before}
after={after} {}",
+ if after == before {
+ "OK (orphan excluded)"
+ } else {
+ "*** UNCOMMITTED ROWS VISIBLE ***"
+ }
+ );
+}
+
+#[ignore = "documents a divergence from Java: layout-v1 tables do not filter
uncommitted files"]
Review Comment:
Running this with `--ignored` prints "OK (orphan excluded)" for all six
combinations, so the reason on the `#[ignore]` no longer reproduces — which the
module doc above says should be updated when it happens. Did the divergence get
fixed somewhere in this change, or does it need a different fixture to show up?
_⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._
##########
crates/core/src/file_group/reader.rs:
##########
@@ -264,9 +430,13 @@ impl FileGroupReader {
/// selecting a version cannot turn a working read into a failing one. Each
/// reason is logged, because a fallback nobody can observe is
/// indistinguishable from a reader that is never used.
+ ///
+ /// Refusals are the exception, not the fallthrough: version 2 is the
reader
Review Comment:
The paragraph just above ("Every reason here means version 1 serves the read
instead, so selecting a version cannot turn a working read into a failing one")
is contradicted by the CUSTOM branch below, which returns `Unsupported`. Since
version 2 is now the default, a MOR table with
`hoodie.record.merge.mode=CUSTOM` that reads today will start failing. Could
the doc carve that case out, and the error message name
`hoodie.read.file.group.reader.version=1` as the way back?
_⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._
##########
crates/core/src/file_group/reader_v2/mod.rs:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.
+ */
+
+//! Context types for the merge-on-read file group reader.
+//!
+//! This module holds the inputs the MOR reader needs and the resolver that
+//! derives them from a table's configs. Nothing consumes it yet — the reader
Review Comment:
This doc and the file-scope `allow(dead_code)`s were written for the stacked
PRs, but this is the change that wires the reader in, so they're now hiding
real dead code in a live module. Dropping the 15 allows surfaces 38 warnings —
`HoodieFileGroupReaderBuilder` and `ProjectingBatchReader` are never
constructed, `RecordContext::partition_path` is threaded through three layers
and never read, and the spill codec's `to_spill_bytes`/`from_spill_bytes` are
unused. Worth clearing before merge while it's still cheap to tell intentional
from accidental?
_⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._
##########
crates/core/Cargo.toml:
##########
@@ -101,12 +107,32 @@ percent-encoding = { workspace = true }
[dev-dependencies]
hudi-test = { path = "../test" }
serial_test = { workspace = true }
+# Also a regular dependency, but only an optional one behind `spill-rocksdb`.
+# Repeated here because tests outside the spill tier use it too, and must keep
+# building in the `--no-default-features` configuration.
tempfile = { workspace = true }
[lints.clippy]
result_large_err = "allow"
[features]
+default = ["spill-rocksdb"]
Review Comment:
The feature's rationale is that the Python wheel and the cxx bridge
shouldn't have to build rocksdb, but it's default-on and `crates/hudi` takes
`hudi-core` with default features, so both still do. Should the umbrella crate
forward the feature so those two can opt out, or is the intent that they keep
it and only external consumers opt out?
_⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._
##########
crates/core/src/file_group/reader_v2/record_merger.rs:
##########
@@ -0,0 +1,799 @@
+/*
+ * 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.
+ */
+
+//! Ported from the merge-on-read reader. Nothing consumes it yet, so its
+//! items are unreachable from the crate's call graph until the reader wires
in.
+#![allow(dead_code)]
+
+//! Mirrors `org.apache.hudi.common.table.read.BufferedRecordMerger` and
+//! `org.apache.hudi.common.table.read.BufferedRecordMergerFactory`.
+//!
+//! The `BufferedRecordMerger` trait defines how records are merged during
+//! log scanning (delta merge: log-vs-log) and at read time (final merge:
+//! base-vs-log).
+
+use super::buffered_record::{BufferedRecord, DeleteRecord, OrderingValue};
+use crate::Result;
+
+/// Mirrors Java `OrderingValues.isCommitTimeOrderingValue` (`orderingValue ==
null
+/// || OrderingValues.isDefault(orderingValue)`, where `isDefault` ==
+/// `Integer(0).equals(orderingValue)`). A DELETE carrying the default ordering
+/// value is a "commit-time ordering delete" and wins unconditionally,
regardless
+/// of the existing record's ordering value (mirrors
+/// `BufferedRecord.isCommitTimeOrderingDelete`).
+///
+/// The default is ONLY the null-coerced [`OrderingValue::Default`] sentinel
(Java
+/// `Integer(0)`) or an absent (`None`) ordering. A GENUINE `Long(0)` field
value
+/// is NOT default (GAP-2): `Integer(0).equals(Long(0))` is `false` in Java,
so a
+/// delete carrying a real ordering value of `0` is ordering-compared, not
treated
+/// as natural-order.
+fn is_default_ordering(ordering_value: &Option<OrderingValue>) -> bool {
+ matches!(ordering_value, None | Some(OrderingValue::Default))
+}
+
+/// Mirrors Java `BufferedRecord.isCommitTimeOrderingDelete`: a DELETE carrying
+/// the default ordering value. Such deletes win unconditionally (natural
order).
+fn is_commit_time_ordering_delete(record: &BufferedRecord) -> bool {
+ record.is_delete() && is_default_ordering(&record.ordering_value)
+}
+
+/// Mirrors Java `OrderingValues.isSameClass`: ordering values are only
compared
+/// when they are the same concrete type. Java throws on a cross-type
+/// `compareTo`; we instead skip the comparison (caller lets the newer/delete
+/// record win), avoiding the arbitrary cross-variant `Ord` on
[`OrderingValue`].
+///
+/// Thin free-function wrapper over [`OrderingValue::is_same_class`] so the
four
+/// buffer-side call sites below read as `is_same_class(a, b)`; the single
source
+/// of truth is the method.
+fn is_same_class(a: &OrderingValue, b: &OrderingValue) -> bool {
+ a.is_same_class(b)
+}
+
+/// Trait for merging buffered records during the file group read pipeline.
+///
+/// Mirrors Java's `BufferedRecordMerger<T>` interface.
+///
+/// Consumed by `KeyBasedFileGroupRecordBuffer`:
+/// - `delta_merge` in `process_next_data_record` (log-vs-log within buffer)
+/// - `delta_merge_delete` in `process_next_deleted_record`
(delete-vs-existing)
+/// - `final_merge` in `has_next_base_record` (base-vs-log at read time)
+pub trait BufferedRecordMerger: Send + Sync + std::fmt::Debug {
+ /// Merge a new log record with an existing buffered record.
+ ///
+ /// Returns `Some(merged)` if the record should be kept, `None` if dropped.
+ ///
+ /// Called during log scanning when a new record for the same key arrives.
+ fn delta_merge(
+ &self,
+ new_record: &BufferedRecord,
+ existing_record: Option<&BufferedRecord>,
+ ) -> Result<Option<BufferedRecord>>;
+
+ /// Merge a delete record with an existing buffered record.
+ ///
+ /// Returns `Some(delete)` if the delete wins, `None` if the existing
record survives.
+ fn delta_merge_delete(
+ &self,
+ delete_record: &DeleteRecord,
+ existing_record: Option<&BufferedRecord>,
+ ) -> Result<Option<DeleteRecord>>;
+
+ /// Merge a base file record with a log record (final merge at read time).
+ ///
+ /// The `older_record` is from the base file, `newer_record` is from the
log buffer.
+ fn final_merge(
+ &self,
+ older_record: &BufferedRecord,
+ newer_record: &BufferedRecord,
+ ) -> Result<BufferedRecord>;
+}
+
+/// Event-time based record merger.
+///
+/// Resolves conflicts by comparing ordering values: the record with the
+/// higher ordering value wins. This is the merge strategy for the
+/// `EVENT_TIME_ORDERING` mode only; `COMMIT_TIME_ORDERING` uses
+/// [`CommitTimeRecordMerger`] (last writer wins, no ordering comparison).
+///
+/// Created by `BufferedRecordMergerFactory` based on the merge mode. The read
+/// path now accepts `EVENT_TIME_ORDERING`
(`buffer/loader.rs::get_record_buffer`),
+/// so this merger is live in production — it drives the scalar base-vs-log
+/// `final_merge`, and the vectorized `pick_winner` kernel mirrors its
semantics.
+#[derive(Debug)]
+pub struct EventTimeRecordMerger;
+
+impl BufferedRecordMerger for EventTimeRecordMerger {
+ /// Mirrors Java `EventTimeRecordMerger.deltaMerge` via
`shouldKeepNewerRecord`.
+ fn delta_merge(
+ &self,
+ new_record: &BufferedRecord,
+ existing_record: Option<&BufferedRecord>,
+ ) -> Result<Option<BufferedRecord>> {
+ match existing_record {
+ None => Ok(Some(new_record.clone())),
+ Some(existing) if should_keep_newer_record(existing, new_record)
=> {
+ Ok(Some(new_record.clone()))
+ }
+ Some(existing) => Ok(Some(existing.clone())),
+ }
+ }
+
+ /// Mirrors Java `BufferedRecordMergerFactory.deltaMergeDeleteRecord`.
+ ///
+ /// The DELETE is obsolete (existing survives) ONLY when the existing
record
+ /// has a strictly greater ordering value of the SAME type AND the delete
+ /// carries a non-default ordering value (`0` == natural order = always
+ /// delete). A default-ordering delete, or one against an existing record
+ /// that is itself a commit-time-ordering delete, follows Java's branches.
+ fn delta_merge_delete(
+ &self,
+ delete_record: &DeleteRecord,
+ existing_record: Option<&BufferedRecord>,
+ ) -> Result<Option<DeleteRecord>> {
+ match existing_record {
+ None => Ok(Some(delete_record.clone())),
+ Some(existing) if is_commit_time_ordering_delete(existing) =>
Ok(None),
+ Some(existing) => {
+ let choose_existing =
!is_default_ordering(&delete_record.ordering_value)
+ && match (&delete_record.ordering_value,
&existing.ordering_value) {
+ (Some(del_val), Some(existing_val)) => {
+ is_same_class(del_val, existing_val) &&
existing_val > del_val
+ }
+ _ => false,
+ };
+ if choose_existing {
+ Ok(None) // existing record survives; DELETE is obsolete
+ } else {
+ Ok(Some(delete_record.clone()))
+ }
+ }
+ }
+ }
+
+ /// Mirrors Java `EventTimeRecordMerger.finalMerge`.
+ ///
+ /// The newer (log) record wins unless it is NOT a commit-time-ordering
delete
+ /// and the older record (also not such a delete) has a strictly greater,
+ /// same-typed ordering value.
+ fn final_merge(
Review Comment:
Java's `finalMerge` delegates straight to `shouldKeepNewerRecord`; this
re-derives the same rule inline. They agree today (I worked through the
CTO-delete and cross-type cases), but it's one rule in two places. Any reason
not to call `should_keep_newer_record(older_record, newer_record)` here?
_⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._
##########
crates/core/src/file_group/reader_v2/buffer/spillable_map.rs:
##########
@@ -0,0 +1,2627 @@
+/*
+ * 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.
+ */
+
+//! Ported from the merge-on-read reader. Nothing consumes it yet, so its
+//! items are unreachable from the crate's call graph until the reader wires
in.
+#![allow(dead_code)]
+
+//! Size-tracked, RocksDB-spillable merge map (A1, ENG-42993).
Review Comment:
There are a few dozen `ENG-XXXXX` ids and phase labels (A1, A2, A3c, A6e,
B2, I-3, I-33, G-16) left in the source — e.g. this line,
`record_context.rs:1790`, `read_stats.rs:72`. They don't resolve for anyone
reading this on the ASF side. Could they be replaced with the behaviour they're
standing in for, or a hudi-rs issue number?
_⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality._
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
