linliu-code commented on code in PR #660:
URL: https://github.com/apache/hudi-rs/pull/660#discussion_r3786087285
##########
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:
Confirmed and fixed — thank you, this was a real bug and the reproduction
was exact. I reproduced it before changing anything:
```
20250713010500000 -> 20250713010499999 parse: ERR Timeline("Invalid epoch
millis")
20250713010501000 -> 20250713010500999 parse: OK 2025-07-13 01:05:00.999
UTC
```
The `ss` field becomes 99. It sorts correctly, so every range comparison
looked right, and it is not a time, so the selector rejects it.
**I tried your suggested fix first — carrying the exclusivity as a flag on
the selector — and it broke `hudi_table_get_file_slices_incremental` (2 slices
instead of 3).** `StartTimestamp` has four consumers building ranges, not one:
`create_instant_range_for_log_file_scan`, `resolver::resolve_instant_range`,
the row-mask path, and the metadata-table reader. Making only the log-scan
range inclusive dropped a slice from the file-slice selection path, which never
goes through it. So the flag approach is right in principle but needs all four
moved together, and I did not want to land a partial version of it inside this
PR.
What I did instead removes the invalid value rather than the encoding: the
decrement now steps in the **time** domain, so the result is both a real
instant and lexicographically just below the input.
```rust
const FORMATS: [&str; 2] = ["%Y%m%d%H%M%S%3f", "%Y%m%d%H%M%S"];
if let Ok(dt) = Instant::parse_datetime(instant_time, "UTC") {
for format in FORMATS {
if dt.format(format).to_string() == instant_time
&& let Some(stepped) =
dt.checked_sub_signed(TimeDelta::milliseconds(1))
{ ... }
}
}
```
The round-trip is what distinguishes a date-formatted instant from an
epoch-millis one (a metadata table's `00000000000000000`) — both parse, only
the former renders back to itself — so the sentinels keep the integer decrement.
Two tests, both mutation-checked against the old integer decrement:
- `test_instant_time_minus_one_stays_a_valid_instant` — every borrow
boundary: whole second, whole minute, whole hour, midnight (the date steps
back), second-precision instants, and the epoch-millis sentinels.
- `test_resolve_incremental_window_start_bound_is_a_real_instant` — the
bound sorts below the earliest admitted instant **and** parses. No fixture has
a commit on a whole second, which is why this went unseen, so it asserts the
property rather than reproducing the one input that broke.
I'd still like the flag version eventually — it removes the dual contract
instead of satisfying it — but as its own change, with all four consumers moved
at once.
##########
.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:
Fixed — you and I found this within about an hour of each other; it was
pushed just before your comment landed.
Confirmed the same way you did: `gh api
repos/Swatinem/rust-cache/commits/6323deb1c3ea99e6f4dae6b7d3ac1b73ac21ff77`
returns `422 No commit found`. It shares the `6323deb1` prefix with the real
v2.9.2 SHA and carries the same `# v2.9.2` comment, so it reads as correct at a
glance — and the job died in "Set up job" after two seconds, before checkout,
so nothing of ours ever ran.
Now pinned to `6323deb102c322ba6fcbdcafc7e3dddab59af2b6`, matching the other
two jobs. I also checked every pinned action SHA across all workflow files
resolves against the API — `astral-sh/setup-uv` and
`docker/setup-buildx-action` both do. The `rust-tests-no-spill-backend` leg
passes on the current head.
##########
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:
You're right, and it was my paragraph. Fixed both halves.
The doc now carves the case out explicitly rather than claiming something
the code below contradicts:
> Every reason here **but one** means version 1 serves the read instead… The
exception is a `CUSTOM` record merge mode with a merge to perform, which
errors. That one *does* turn a working read into a failing one: version 2 is
the default, so a merge-on-read table declaring `CUSTOM` that reads today stops
reading. It is deliberate — falling back would merge with version 1's own
derivation, which drops deletes, and wrong rows are worse than a refusal — but
it is a behaviour change, and the error names the way back.
And the error now names the escape hatch:
> A table with a CUSTOM record merge mode needs its own merger, which no
reader here implements. Set `hoodie.read.file.group.reader.version=1` to read
it with the reader that served it before, which merges without that merger.
Worth recording alongside this: the only merge-on-read tables in the fixture
corpus with `record.merge.mode=CUSTOM` are the metadata tables, which route to
version 1 before the capability check runs. So this refusal is covered by unit
tests only — no gold fixture exercises it.
##########
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:
Confirmed, and I've made a start rather than a sweep.
You're right that the framing is stale: 25 files still say "nothing consumes
it yet", written when these were stacked PRs, and this is the change that wires
the reader in. I removed the file-scope `allow(dead_code)` from
`input_split.rs` and `iterator_mode.rs` (their module docs needed unsplitting
anyway — see the two nits on those files), and it immediately surfaced exactly
what you describe: `InputSplit::start` and `InputSplit::length`, always `0` and
`-1`, never read. Both removed.
I have **not** done the other 13 allows. Clearing them properly means
deciding per item whether something is scaffolding the port still needs or an
accident, and I would rather that be a deliberate pass — with your read on
`HoodieFileGroupReaderBuilder` and `ProjectingBatchReader` in particular, since
"never constructed" could mean either. Happy to do it in this PR if you'd
prefer it cleared before merge; my instinct is a follow-up so the diff stays
reviewable, but it's your call and I don't feel strongly.
##########
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:
Confirmed — running it with `--ignored` fails exactly as you say:
```
these now read correctly and should move to GOLD_FIXTURES:
["table_column_projection", "table_all_data_types", "table_null_containers",
"table_corrupt_tail_block", "table_delete_ord_int", "table_delete_ord_long",
"table_delete_ord_double", "table_delete_ord_decimal",
"table_delete_ord_string", "table_delete_ord_timestamp"]
```
All ten, including the four `table_delete_ord_*` you named, and the "two
causes remain" doc above is stale with them.
On whether it earns its keep: I think you're right that it doesn't.
`tests/gold_parity_tests.rs` covers the same fixtures against the same Spark
snapshots with a live ratchet — a new disagreement fails the build and a fixed
one has to be removed — whereas this harness is `#[ignore]`d, so its data can
rot without anything noticing, which is precisely what happened. A second copy
that cannot fail is worse than no copy.
I've left it alone for now rather than deleting it unilaterally, since
removing a test harness on someone else's read is not my call to make silently.
Say the word and I'll fold what it uniquely covers into `gold_parity_tests.rs`
and delete it.
##########
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:
Confirmed — `--ignored` gives 3 passed, 0 failed, and the orphan-exclusion
probe prints "OK (orphan excluded)" for all six combinations. The reason on the
`#[ignore]` no longer reproduces, and the module doc above says that is exactly
when it should be updated.
I don't yet know which of your two possibilities it is — fixed somewhere in
this change, or needing a different fixture to show up. The honest answer is
that I would have to bisect the stack to tell, and I did not want to guess in
the ignore reason. Left as-is pending that; if you'd rather it not merge in
this state I'll dig into which commit changed the behaviour and either drop the
test or re-point it at a fixture that still shows the divergence.
##########
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:
Your reading is right, and it makes the feature's stated rationale not
currently true for the two consumers it names. `spill-rocksdb` is default-on,
`crates/hudi` takes `hudi-core` with default features, and both the Python
wheel and the cxx bridge go through the umbrella crate — so both still build
rocksdb, libclang and all.
The intent was the second of your two options — external consumers opt out —
but the doc reads as though the wheel and the bridge already benefit, which
they do not. Two ways to make it honest: forward the feature from `crates/hudi`
so those two can actually opt out, or correct the rationale to say only
external consumers benefit today.
I'd rather not pick between those inside this PR — forwarding the feature
changes the umbrella crate's public feature surface, which deserves its own
diff and its own note. Which would you prefer?
##########
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:
Confirmed and agreed: 105 `ENG-XXXXX` references across 16 files, plus the
phase labels. They are internal tracker ids and mean nothing to a reader on the
ASF side — the project's own guidance says not to reference internal plans.
I have not done this yet. It is mechanical in shape but not by `sed`: each
one is standing in for a behaviour, and replacing it well means reading the
surrounding code and writing down what the id was pointing at — otherwise the
comment loses information rather than gaining it. Doing that badly across 105
sites would be worse than leaving them.
So: happy to do it, and I think it should be done before this merges into an
ASF repo, but as a focused pass. Would you rather it land in this PR or
immediately after?
##########
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:
No good reason — you're right that it is one rule in two places, and thank
you for working through the CTO-delete and cross-type cases to confirm they
agree today.
I have not changed it yet, only because `final_merge` is on the merge hot
path and I want to confirm the delegation is behaviour-identical under the gold
sweep rather than by reading, given the two spellings agree by inspection but
the whole point of the finding is that they could stop agreeing. That check is
cheap and I'll do it — I just did not want to slip a hot-path change in
unverified alongside everything else in this round.
If you'd rather see it in this PR, say so and I'll make the change and run
the corpus against it.
--
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]
