linliu-code commented on code in PR #660: URL: https://github.com/apache/hudi-rs/pull/660#discussion_r3786603345
########## 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: Done — swept, and it found more than a lint. Every file-scope `#![allow(dead_code)]` in `reader_v2` is gone (23 of them). Deleted outright: - **The per-record spill codec.** `to_spill_bytes` / `from_spill_bytes`, `encode_`/`decode_ordering_value` and nine `SPILL_*` tags. Its doc claimed *"Used by `SpillableRecordMap` to store an entry in the on-disk (RocksDB) tier"* — it is not. `spillable_map` spills whole Arrow batches through `row_serde::to_binary_row_body` and keeps ordering values in the in-memory `DiskEntry` index. A superseded parallel implementation whose doc asserted the opposite. Eight tests went with it, including three I had added the round before — they were covering dead code. - `ProjectingBatchReader`, constructed nowhere in the lib or under `cfg(test)`. - `InputSplit::{start,length}` (always `0`/`-1`), `DeleteRecord::partition_path` (every construction site passed `String::new()`), `RecordPayload::into_record`, `BufferedRecords::from_engine_record`. On your two specifically: `ProjectingBatchReader` is gone; `HoodieFileGroupReaderBuilder` turned out to be **used by the test harness**, which is why I checked rather than deleted — my first attempt removed it and broke the harness build. What remains is 48 targeted allows, each naming why that item is kept, so the argument about any one of them can be had per item rather than per file. Mostly Java-parity surface reached only by the harness. One thing worth flagging separately: `allow_inflight_instants` is plumbed from the loader into a field nothing reads, so the knob is **inert**. Nothing sets it true, so behaviour is correct today — I annotated it saying so rather than deleting it, since deleting would hide that the setting does nothing. Happy to make the builder refuse `true` instead if you'd prefer it loud. `make check-rust` clean with zero dead-code warnings; 1267 lib tests green. ########## 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: Deleted, agreeing with your read. All three of its tests were subsumed: `merged_reads_match_the_spark_snapshot` covered 6 fixtures where `tests/gold_parity_tests.rs` covers 62 with a live ratchet; `known_gaps_still_fail` was `#[ignore]`d with all ten entries stale; and `fixtures_without_gold_are_present` is the same assertion `EXPECTED_WITHOUT_GOLD` already makes. Your framing was the deciding argument — a second copy that cannot fail is worse than no copy, and its rotting data is exactly what that predicts. ########## 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: Done — zero `ENG-*` references left, plus the phase labels (`A1`, `A2`, `B2`) and three pointers to an internal design doc. 103 references, but only **15 distinct ids**, so each one meant a single thing and could be replaced consistently with the behaviour it stood for: `ENG-42992` → the lazy base-file source, `ENG-42993` → the spillable merge map, `ENG-43009` → the vectorized merge path, `ENG-38318` → event-time ordering, and so on. Where the id was a bare parenthetical the prose already stood on its own, so it just goes. Worth admitting how it went: my first pass carried a whitespace-normalising rule that destroyed indentation across all 15 files — code bodies, Apache license headers, and doc-comment list continuations. `rustfmt` recovered the code, and I restored the rest by diffing each file against the previous commit and taking the original wherever only whitespace differed. It is clean and verified now, but it was self-inflicted and I would not script prose edits that way again. ########## 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: Done — `final_merge` now delegates, matching Java's `finalMerge`. I worked the case analysis first rather than trusting that they looked the same. `final_merge` returned the older record iff `!ctod(newer) && !ctod(older) && same_class && old > new`; `should_keep_newer_record` returns false under exactly that condition — the `new >= old` negation lines up. So: ```rust if should_keep_newer_record(older_record, newer_record) { Ok(newer_record.clone()) } else { Ok(older_record.clone()) } ``` The gold corpus is green behind it: 62 fixtures and 183 option cases against Spark's output, plus the position-merge sweep. Thanks for confirming the CTO-delete and cross-type cases — that made this a rename rather than an investigation. -- 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]
