hudi-agent commented on code in PR #660: URL: https://github.com/apache/hudi-rs/pull/660#discussion_r3780692640
########## crates/core/src/file_group/reader_v2/input_split.rs: ########## @@ -0,0 +1,259 @@ +/* + * 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.InputSplit`. Review Comment: 🤖 nit: the `#![allow(dead_code)]` attribute is sandwiched between two `//!` module-doc blocks, splitting what reads as a single description across an attribute. Could you move `#![allow(dead_code)]` before the first `//!` line (or after the last), so the module doc is contiguous? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## crates/core/src/file_group/reader_v2/iterator_mode.rs: ########## @@ -0,0 +1,57 @@ +/* + * 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.IteratorMode`. Review Comment: 🤖 nit: same split `//!` block issue as `input_split.rs` - the `#![allow(dead_code)]` attribute breaks the module doc in two. Could you move it so the description is contiguous? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## crates/core/src/file_group/reader_v2/iterator_mode.rs: ########## @@ -0,0 +1,57 @@ +/* + * 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.IteratorMode`. +//! +//! Controls what form the output records take when iterating. + +/// The mode in which the file group reader yields records. +/// +/// In Java Hudi, this controls whether the iterator produces engine-native +/// records, HoodieRecord wrappers, or just record keys. +/// +/// In hudi-rs, we always work with Arrow RecordBatch, so `EngineRecord` +/// is the primary mode. The other modes are kept for API symmetry. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum IteratorMode { + /// Yield engine-native records (Arrow RecordBatch in Rust). + #[default] + EngineRecord, + + /// Yield HoodieRecord wrappers (not yet implemented in Rust). + HoodieRecord, + + /// Yield only record keys (not yet implemented in Rust). + RecordKey, +} + +impl IteratorMode { Review Comment: 🤖 nit: `from_str_opt` is a non-idiomatic name - the `_opt` suffix for returning `Option` is not a Rust convention. Have you considered implementing `std::str::FromStr` (enabling `.parse::<IteratorMode>()`) or naming it `try_from_str` to align with how the ecosystem spells fallible conversions? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## crates/core/src/file_group/reader_v2/buffered_record.rs: ########## @@ -0,0 +1,1094 @@ +/* + * 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.BufferedRecord` and +//! `org.apache.hudi.common.model.DeleteRecord`. +//! +//! In Java Hudi, `BufferedRecord<T>` wraps a single engine-native record +//! with its key, ordering value, and operation type. In hudi-rs, the +//! "engine record" is an Arrow `RecordBatch`. +//! +//! During log scanning, records are accumulated into the record buffer. +//! At read time, base file records are merged with log records to produce +//! the final output. +//! +//! ## In-memory representation (A2) +//! +//! Per the A2 alignment work (05-a2a1-design.md), the in-memory payload is a +//! zero-copy [`RecordPayload::BatchRef`] into the shared source batch that the +//! row was decoded from — NOT a per-row Arrow IPC blob. IPC serialization is a +//! spill-only concern (A1) and lives in +//! [`row_serde`](super::row_serde) (`to_binary_row` / +//! `from_binary`); it is no longer paid on the in-memory merge hot path. The +//! data-driven motivation: the per-row IPC blob cost ~27x the build memory and +//! ~12x the build CPU of the batch-ref representation (spike Part 1). + +use std::sync::Arc; + +use arrow_array::RecordBatch; +use arrow_buffer::i256; + +use crate::Result; +use crate::error::CoreError; +use crate::file_group::reader_v2::row_serde; + +// ── Spill value encoding (A1) ────────────────────────────────────────────── +// On spill, a `BufferedRecord` is serialized to a self-describing byte blob +// stored as the RocksDB value. The record key is the RocksDB key, so it is NOT +// repeated in the value. The layout is: +// +// [payload_tag: u8] +// 0x00 = Delete (no further payload bytes) +// 0x01 = Data (followed by an Arrow IPC single-row stream) +// [ordering_tag: u8] +// 0x00 = None +// 0x01 = Long followed by 8 bytes (i64 little-endian) +// 0x02 = String followed by [len: u32 LE][utf8 bytes] +// [if payload_tag == Data] Arrow IPC stream bytes (rest of the blob) +// +// All multi-byte integers are little-endian. The format is internal to the +// reader's own spill round-trip (never persisted across runs), so no on-disk +// version negotiation is needed — the spill dir is created and destroyed within +// a single read (see `SpillableRecordMap`). + +/// Spill payload tag: the record is a delete tombstone (no data follows). +const SPILL_PAYLOAD_TAG_DELETE: u8 = 0x00; +/// Spill payload tag: the record carries data (an Arrow IPC stream follows the +/// ordering-value header). +const SPILL_PAYLOAD_TAG_DATA: u8 = 0x01; +/// Spill ordering-value tag: no ordering value. +const SPILL_ORDERING_TAG_NONE: u8 = 0x00; +/// Spill ordering-value tag: `OrderingValue::Long` (8-byte i64 follows). +const SPILL_ORDERING_TAG_LONG: u8 = 0x01; +/// Spill ordering-value tag: `OrderingValue::String` (u32 length + utf8 follow). +const SPILL_ORDERING_TAG_STRING: u8 = 0x02; +/// Spill ordering-value tag: `OrderingValue::Composite` (u32 element count, then +/// each element encoded with this same tag scheme). +const SPILL_ORDERING_TAG_COMPOSITE: u8 = 0x03; +/// Spill ordering-value tag: `OrderingValue::Double` (8-byte IEEE-754 f64 follows). +const SPILL_ORDERING_TAG_DOUBLE: u8 = 0x04; +/// Spill ordering-value tag: `OrderingValue::Decimal` (16-byte little-endian i128 +/// unscaled value, then a 1-byte i8 scale follow). +const SPILL_ORDERING_TAG_DECIMAL: u8 = 0x05; +/// Spill ordering-value tag: `OrderingValue::Default` (no payload). The null-coerced +/// default ordering value (Java `HoodieRecord.DEFAULT_ORDERING_VALUE` == `Integer(0)`). +const SPILL_ORDERING_TAG_DEFAULT: u8 = 0x06; + +/// In-memory payload of a buffered record (A2). +/// +/// The default representation is a zero-copy reference into a shared source +/// batch ([`BatchRef`](RecordPayload::BatchRef)). Serialization (Arrow IPC) +/// happens only on spill (A1), never on the in-memory hot path — so this enum +/// deliberately carries no binary variant. +/// +/// ## Pinning caveat +/// +/// A [`BatchRef`](RecordPayload::BatchRef) keeps its WHOLE source +/// `Arc<RecordBatch>` alive (a 1-row [`RecordBatch::slice`] still references the +/// parent's full buffers). When survivors are sparse across many source batches, +/// the merge map can pin far more Arrow memory than the live rows occupy. The +/// `compact_pinned_batches` safety valve (see `key_based.rs`) re-batches the +/// survivors of sparsely-populated source batches and repoints their entries to +/// release the dead-row memory. +#[derive(Debug, Clone)] +pub enum RecordPayload { + /// Zero-copy reference into a shared source batch (the A2 default). `row_idx` + /// addresses a single row within `batch`. Reading the payload slices that row + /// out via [`RecordBatch::slice`] (cheap, shares buffers). + BatchRef { + /// The shared source batch this row was decoded from. Interned once per + /// decoded block batch (one `Arc` per batch) so that `Arc::as_ptr` + /// identity is a valid grouping key for compaction. + batch: Arc<RecordBatch>, + /// Row index of this record within `batch`. + row_idx: usize, + }, + /// Owned single-row batch. Produced by `compact_pinned_batches` (a freshly + /// re-batched survivor set) or for records that have no shared source batch. + Owned(RecordBatch), + /// Tombstone — the record represents a deletion and carries no payload. + Delete, +} + +impl RecordPayload { + /// Returns the payload as a single-row `RecordBatch`, or `None` for a delete. + /// + /// `BatchRef` slices the addressed row out of its source batch (zero-copy + /// buffer slice); `Owned` clones the held batch. Mirrors the read side of + /// Java's `BufferedRecord.getRecord()`. + pub fn get_record(&self) -> Option<RecordBatch> { + match self { + RecordPayload::BatchRef { batch, row_idx } => Some(batch.slice(*row_idx, 1)), + RecordPayload::Owned(batch) => Some(batch.clone()), + RecordPayload::Delete => None, + } + } + + /// Consume the payload and return its single-row `RecordBatch`, or `None` + /// for a delete. The move-not-clone counterpart to [`get_record`](Self::get_record). + pub fn into_record(self) -> Option<RecordBatch> { + match self { + RecordPayload::BatchRef { batch, row_idx } => Some(batch.slice(row_idx, 1)), + RecordPayload::Owned(batch) => Some(batch), + RecordPayload::Delete => None, + } + } + + /// Returns true if this payload is a delete tombstone. + pub fn is_delete(&self) -> bool { + matches!(self, RecordPayload::Delete) + } +} + +/// The universal record envelope flowing through the merge pipeline. +/// +/// Mirrors Java's `BufferedRecord<T>`. In Rust/Arrow, a single "record" is a +/// single-row view ([`RecordPayload::BatchRef`]) into a shared decoded batch; +/// the merge map stores one of these per live key. +/// +/// During the log scanning phase, `BufferedRecord`s are stored in the record +/// buffer's map keyed by record key. +#[derive(Debug, Clone)] +pub struct BufferedRecord { + /// The record key extracted from the record. + pub record_key: String, + + /// The in-memory record payload (zero-copy batch ref, owned batch, or delete). + pub payload: RecordPayload, + + /// The ordering value used for merge conflict resolution. + /// Higher ordering value wins during delta merge. + pub ordering_value: Option<OrderingValue>, +} + +impl BufferedRecord { + /// Create a new data record from a zero-copy reference into a shared source + /// batch (the A2 hot-path constructor). + /// + /// `batch` must be the interned `Arc` for the decoded block batch (one `Arc` + /// per batch) so that `Arc::as_ptr` grouping during compaction is valid. + pub fn new_batch_ref( + record_key: String, + batch: Arc<RecordBatch>, + row_idx: usize, + ordering_value: Option<OrderingValue>, + ) -> Self { + Self { + record_key, + payload: RecordPayload::BatchRef { batch, row_idx }, + ordering_value, + } + } + + /// Create a new data record from an owned single-row batch. + /// + /// Used by tests and by call sites that hold a single-row batch with no + /// shared source (e.g. base-record construction in tests). The hot path + /// uses [`new_batch_ref`](Self::new_batch_ref) instead. + pub fn new_data( + record_key: String, + data: RecordBatch, + ordering_value: Option<OrderingValue>, + ) -> Self { + Self { + record_key, + payload: RecordPayload::Owned(data), + ordering_value, + } + } + + /// Create a new delete record (tombstone). + pub fn new_delete(record_key: String, ordering_value: Option<OrderingValue>) -> Self { + Self { + record_key, + payload: RecordPayload::Delete, + ordering_value, + } + } + + /// Returns true if this record represents a deletion. + pub fn is_delete(&self) -> bool { + self.payload.is_delete() + } + + /// Returns true if this record has no data payload (a delete tombstone). + pub fn is_empty(&self) -> bool { + self.payload.is_delete() + } + + /// Return the record data as a single-row `RecordBatch`. + /// + /// Mirrors Java's `BufferedRecord.getRecord()`. Delegates to + /// [`RecordPayload::get_record`]: `BatchRef` slices the row out (zero-copy), + /// `Owned` clones, `Delete` returns `None`. + pub fn get_record(&self) -> Option<RecordBatch> { + self.payload.get_record() + } + + /// Consume the record and return its data batch. The move-not-clone + /// counterpart to [`get_record`](Self::get_record), used on the merge-output + /// hot path where the record is no longer needed afterward. + pub fn into_record(self) -> Option<RecordBatch> { + self.payload.into_record() + } + + /// Serialize this record to a self-describing spill blob (**spill-only, A1**). + /// + /// Used by [`SpillableRecordMap`](super::buffer::spillable_map::SpillableRecordMap) + /// to store an entry in the on-disk (RocksDB) tier when the in-memory budget + /// is exhausted. The record key is NOT included (it is the RocksDB key); the + /// blob carries the payload (delete tombstone or a single-row Arrow IPC + /// stream) and the ordering value. See the module-level spill-encoding notes. + /// + /// Start with single-row IPC for correctness; a batched-spill optimization + /// (group survivors by source batch, spill multi-row blobs) is tracked as B5 + /// in the backlog (05-a2a1-design.md) and deliberately NOT built here. + pub fn to_spill_bytes(&self) -> Vec<u8> { + let mut buf = Vec::new(); + match self.get_record() { + None => buf.push(SPILL_PAYLOAD_TAG_DELETE), + Some(batch) => { + buf.push(SPILL_PAYLOAD_TAG_DATA); + encode_ordering_value(&mut buf, self.ordering_value.as_ref()); + // Arrow IPC single-row stream (the spill serialization primitive, + // row_serde::to_binary_row). For deletes we skip this entirely. + let ipc = row_serde::to_binary_row(&batch.schema(), &batch); + buf.extend_from_slice(&ipc); + return buf; + } + } + // Delete path: still record the ordering value (it can matter when a + // spilled delete tombstone is later compared during the base merge). + encode_ordering_value(&mut buf, self.ordering_value.as_ref()); + buf + } + + /// Reconstruct a record from a spill blob produced by + /// [`to_spill_bytes`](Self::to_spill_bytes) (**spill-only, A1**). + /// + /// `record_key` is supplied by the caller (it is the RocksDB key, not stored + /// in the blob). Returns a typed [`CoreError`] on a malformed blob rather + /// than panicking — a corrupt spill entry is a recoverable read-path error, + /// not an internal invariant violation. + pub fn from_spill_bytes(record_key: String, bytes: &[u8]) -> Result<Self> { + let mut cursor = 0usize; + let payload_tag = read_u8(bytes, &mut cursor)?; + match payload_tag { + SPILL_PAYLOAD_TAG_DELETE => { + let ordering_value = decode_ordering_value(bytes, &mut cursor)?; + Ok(BufferedRecord::new_delete(record_key, ordering_value)) + } + SPILL_PAYLOAD_TAG_DATA => { + let ordering_value = decode_ordering_value(bytes, &mut cursor)?; + let batch = row_serde::from_binary(&bytes[cursor..])?; + // Reloaded from disk: there is no shared source batch to reference, + // so the payload is necessarily `Owned`. + Ok(BufferedRecord::new_data(record_key, batch, ordering_value)) + } + other => Err(CoreError::ReadFileSliceError(format!( + "spill decode: unknown payload tag {other:#04x}" + ))), + } + } +} + +/// Append an ordering value to a spill blob using the tag scheme documented at +/// the module level. +fn encode_ordering_value(buf: &mut Vec<u8>, ordering_value: Option<&OrderingValue>) { + match ordering_value { + None => buf.push(SPILL_ORDERING_TAG_NONE), + Some(OrderingValue::Long(v)) => { + buf.push(SPILL_ORDERING_TAG_LONG); + buf.extend_from_slice(&v.to_le_bytes()); + } + Some(OrderingValue::Double(v)) => { + buf.push(SPILL_ORDERING_TAG_DOUBLE); + buf.extend_from_slice(&v.to_le_bytes()); + } + Some(OrderingValue::Decimal { unscaled, scale }) => { + buf.push(SPILL_ORDERING_TAG_DECIMAL); + buf.extend_from_slice(&unscaled.to_le_bytes()); + buf.push(*scale as u8); + } + Some(OrderingValue::String(s)) => { + buf.push(SPILL_ORDERING_TAG_STRING); + let sb = s.as_bytes(); + buf.extend_from_slice(&(sb.len() as u32).to_le_bytes()); + buf.extend_from_slice(sb); + } + Some(OrderingValue::Default) => buf.push(SPILL_ORDERING_TAG_DEFAULT), + Some(OrderingValue::Composite(elems)) => { + buf.push(SPILL_ORDERING_TAG_COMPOSITE); + buf.extend_from_slice(&(elems.len() as u32).to_le_bytes()); + for e in elems { + encode_ordering_value(buf, Some(e)); + } + } + } +} + +/// Read a single byte at `*cursor`, advancing it. Errors if out of bounds. +fn read_u8(bytes: &[u8], cursor: &mut usize) -> Result<u8> { + let b = *bytes.get(*cursor).ok_or_else(|| { + CoreError::ReadFileSliceError("spill decode: unexpected end of blob".to_string()) + })?; + *cursor += 1; + Ok(b) +} + +/// Decode an ordering value written by [`encode_ordering_value`], advancing +/// `*cursor` past it. +fn decode_ordering_value(bytes: &[u8], cursor: &mut usize) -> Result<Option<OrderingValue>> { + let tag = read_u8(bytes, cursor)?; + match tag { + SPILL_ORDERING_TAG_NONE => Ok(None), + SPILL_ORDERING_TAG_DEFAULT => Ok(Some(OrderingValue::Default)), + SPILL_ORDERING_TAG_LONG => { + let end = *cursor + 8; + let slice = bytes.get(*cursor..end).ok_or_else(|| { + CoreError::ReadFileSliceError( + "spill decode: truncated Long ordering value".to_string(), + ) + })?; + let v = i64::from_le_bytes(slice.try_into().expect("8-byte slice")); + *cursor = end; + Ok(Some(OrderingValue::Long(v))) + } + SPILL_ORDERING_TAG_DOUBLE => { + let end = *cursor + 8; + let slice = bytes.get(*cursor..end).ok_or_else(|| { + CoreError::ReadFileSliceError( + "spill decode: truncated Double ordering value".to_string(), + ) + })?; + let v = f64::from_le_bytes(slice.try_into().expect("8-byte slice")); + *cursor = end; + Ok(Some(OrderingValue::Double(v))) + } + SPILL_ORDERING_TAG_DECIMAL => { + let end = *cursor + 16; + let slice = bytes.get(*cursor..end).ok_or_else(|| { + CoreError::ReadFileSliceError( + "spill decode: truncated Decimal ordering value".to_string(), + ) + })?; + let unscaled = i128::from_le_bytes(slice.try_into().expect("16-byte slice")); + *cursor = end; + let scale = read_u8(bytes, cursor)? as i8; + Ok(Some(OrderingValue::Decimal { unscaled, scale })) + } + SPILL_ORDERING_TAG_STRING => { + let len_end = *cursor + 4; + let len_slice = bytes.get(*cursor..len_end).ok_or_else(|| { + CoreError::ReadFileSliceError( + "spill decode: truncated String ordering length".to_string(), + ) + })?; + let len = u32::from_le_bytes(len_slice.try_into().expect("4-byte slice")) as usize; + *cursor = len_end; + let str_end = *cursor + len; + let str_slice = bytes.get(*cursor..str_end).ok_or_else(|| { + CoreError::ReadFileSliceError( + "spill decode: truncated String ordering value".to_string(), + ) + })?; + let s = std::str::from_utf8(str_slice) + .map_err(|e| { + CoreError::ReadFileSliceError(format!( + "spill decode: invalid utf8 in String ordering value: {e}" + )) + })? + .to_string(); + *cursor = str_end; + Ok(Some(OrderingValue::String(s))) + } + SPILL_ORDERING_TAG_COMPOSITE => { + let len_end = *cursor + 4; + let len_slice = bytes.get(*cursor..len_end).ok_or_else(|| { + CoreError::ReadFileSliceError( + "spill decode: truncated composite ordering length".to_string(), + ) + })?; + let n = u32::from_le_bytes(len_slice.try_into().expect("4-byte slice")) as usize; + *cursor = len_end; + let mut elems = Vec::with_capacity(n); + for _ in 0..n { + match decode_ordering_value(bytes, cursor)? { + Some(v) => elems.push(v), + None => { + return Err(CoreError::ReadFileSliceError( + "spill decode: null element in composite ordering value".to_string(), + )); + } + } + } + Ok(Some(OrderingValue::Composite(elems))) + } + other => Err(CoreError::ReadFileSliceError(format!( Review Comment: 🤖 nit: `is_empty` and `is_delete` are identical — could you remove `is_empty` and let callers use `is_delete` directly? The name `is_empty` implies something subtly different (a record with no meaningful content might not be a delete in a richer model), and having two names for the same predicate creates a maintenance surface where they could drift apart. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- 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]
