This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new b77f5ce5 fix(core): fold partial updates in both ordering directions
(#679)
b77f5ce5 is described below
commit b77f5ce51a07c7faf422f7de81083acd4e857e3e
Author: Lin Liu <[email protected]>
AuthorDate: Tue Sep 1 17:20:55 2026 -0700
fix(core): fold partial updates in both ordering directions (#679)
---
.../src/file_group/reader_v2/buffer/key_based.rs | 193 ++++++++++++++++++---
crates/core/tests/gold_parity_tests.rs | 10 ++
.../mor/avro/table_partial_update_event_time.sql | 74 ++++++++
.../mor/avro/table_partial_update_event_time.zip | Bin 0 -> 16147 bytes
crates/test/src/lib.rs | 21 +++
5 files changed, 272 insertions(+), 26 deletions(-)
diff --git a/crates/core/src/file_group/reader_v2/buffer/key_based.rs
b/crates/core/src/file_group/reader_v2/buffer/key_based.rs
index 676f31f7..92a4aa21 100644
--- a/crates/core/src/file_group/reader_v2/buffer/key_based.rs
+++ b/crates/core/src/file_group/reader_v2/buffer/key_based.rs
@@ -1458,37 +1458,50 @@ impl HoodieFileGroupRecordBuffer for
KeyBasedFileGroupRecordBuffer {
// is the `else`:
//
// Partial-update (IS_PARTIAL / KEEP_VALUES): the incoming record
carries only
- // a subset of columns; overlay its present columns onto the prior
buffered
- // record for this key, producing the UNION of their columns. The
result
- // stays narrower than the reader schema until a base row or later
update
- // supplies the rest, so base-vs-log can still fill the gaps. With no
prior
- // data record it stays narrow and is padded at the base/drain step.
+ // a subset of columns; fold it together with the prior buffered
record for
+ // this key, producing the UNION of their columns. The result stays
narrower
+ // than the reader schema until a base row or later update supplies
the rest,
+ // so base-vs-log can still fill the gaps. With no prior data record
it stays
+ // narrow and is padded at the base/drain step.
let inc_is_partial = match (record.get_record(),
self.base.reader_schema.as_ref()) {
(Some(ib), Some(target)) => schema_is_partial(&ib.schema(),
target),
_ => false,
};
if inc_is_partial {
- let record = match self.base.records.get(key)? {
- Some(prior) => match (record.get_record(), prior.get_record())
{
- (Some(ib), Some(prior_batch)) => {
- let union = union_schema(&ib.schema(),
&prior_batch.schema());
- let merged = overlay_partial_over_prior(&ib,
&prior_batch, &union)?;
- BufferedRecord::new_data(record.record_key, merged,
record.ordering_value)
- }
- // Prior is a delete tombstone (no columns) → keep
incoming as-is.
- _ => record,
- },
- None => record,
- };
- // Single-probe merge (perf): probe `key` once via
`merge_in_place` and
- // overwrite in place, instead of get(probe+clone) → delta_merge →
insert
- // (probe+clone+probe). The merger only reads `existing` by
reference. A2
- // semantics preserved: the merged BatchRef payload is stored
in-memory
- // (IPC serialization deferred to spill only).
+ // Mirrors Java `DefaultSparkRecordMerger.partialMerge`, which
folds in BOTH
+ // directions: the ordering WINNER supplies every column it
carries and the
+ // loser only the columns the winner omits. The union takes the
winner's
+ // ordering value, so a later update is compared against the
winner's
+ // position on the timeline rather than the loser's.
+ //
+ // Single-probe merge (perf): the fold runs inside
`merge_in_place`, which
+ // probes `key` once and overwrites the slot in place rather than
+ // get(probe+clone) → merge → insert(probe+clone+probe). A2
semantics
+ // preserved: the merged payload is stored in-memory (IPC
serialization
+ // deferred to spill only).
let merger = &self.base.buffered_record_merger;
- self.base
- .records
- .merge_in_place(key, |existing| merger.delta_merge(&record,
existing))?;
+ let commit_time_ordering = self.base.record_merge_mode ==
"COMMIT_TIME_ORDERING";
+ self.base.records.merge_in_place(key, |existing| {
+ // Folding needs columns on both sides. With no prior, or a
prior that
+ // is a delete tombstone, the ordering comparison alone
decides.
+ let Some((prior, prior_batch, ib)) = existing
+ .and_then(|prior| Some((prior, prior.get_record()?,
record.get_record()?)))
+ else {
+ return merger.delta_merge(&record, existing);
+ };
+ let union = union_schema(&ib.schema(), &prior_batch.schema());
+ let new_wins = commit_time_ordering ||
should_keep_newer_record(prior, &record);
+ let (winner, loser, ordering_value) = if new_wins {
+ (&ib, &prior_batch, record.ordering_value.clone())
+ } else {
+ (&prior_batch, &ib, prior.ordering_value.clone())
+ };
+ Ok(Some(BufferedRecord::new_data(
+ record.record_key.clone(),
+ overlay_partial_over_prior(winner, loser, &union)?,
+ ordering_value,
+ )))
+ })?;
} else if self.ignore_defaults || self.unavailable_value.is_some() {
// Full-schema partial-update (IGNORE_DEFAULTS or
FILL_UNAVAILABLE),
// log-vs-log blend. Mirrors Java
`EventTimePartialRecordMerger.deltaMerge`,
@@ -6200,7 +6213,10 @@ mod tests {
}
fn build_pu_event_buffer() -> KeyBasedFileGroupRecordBuffer {
- let merge_mode = "EVENT_TIME_ORDERING";
+ build_pu_buffer("EVENT_TIME_ORDERING")
+ }
+
+ fn build_pu_buffer(merge_mode: &str) -> KeyBasedFileGroupRecordBuffer {
let mut ctx = ReaderContext::empty();
ctx.table_config.insert(
HudiTableConfig::OrderingFields.as_ref().to_string(),
@@ -6246,6 +6262,27 @@ mod tests {
BufferedRecord::new_data(key.to_string(), batch,
Some(OrderingValue::Long(ts)))
}
+ /// A partial record carrying `_hoodie_record_key` + `ts` + `note` (omits
`a`),
+ /// with ordering value = `ts`. Disjoint from `pu_event_partial`'s
columns, so a
+ /// fold of the two is observable in both directions.
+ fn pu_event_partial_note(key: &str, ts: i64, note: &str) -> BufferedRecord
{
+ let s = Arc::new(Schema::new(vec![
+ Field::new("_hoodie_record_key", DataType::Utf8, false),
+ Field::new("ts", DataType::Int64, false),
+ Field::new("note", DataType::Utf8, true),
+ ]));
+ let batch = RecordBatch::try_new(
+ s,
+ vec![
+ Arc::new(StringArray::from(vec![key])) as _,
+ Arc::new(Int64Array::from(vec![ts])) as _,
+ Arc::new(StringArray::from(vec![Some(note)])) as _,
+ ],
+ )
+ .unwrap();
+ BufferedRecord::new_data(key.to_string(), batch,
Some(OrderingValue::Long(ts)))
+ }
+
fn pu_event_base() -> RecordBatch {
RecordBatch::try_new(
pu_event_schema(),
@@ -6326,6 +6363,110 @@ mod tests {
);
}
+ /// EVENT_TIME, log-vs-log: the HIGHER-ordering partial arrives FIRST, so
the
+ /// second (stale) partial loses the ordering comparison. Java folds in
both
+ /// directions, so the loser's unique column still lands: `a` from ts=9,
+ /// `note` from ts=2.
+ #[test]
+ fn test_partial_update_event_time_out_of_order_folds_loser_columns() {
+ let mut buffer = build_pu_event_buffer();
+ buffer
+ .process_next_data_record(pu_event_partial("k1", 9, 99), "k1")
+ .unwrap();
+ buffer
+ .process_next_data_record(pu_event_partial_note("k1", 2, "late"),
"k1")
+ .unwrap();
+ assert_eq!(
+ pu_event_drain(buffer),
+ (9, 99, Some("late".to_string())),
+ "winner keeps a=99 at ts=9; the losing partial still contributes
note"
+ );
+ }
+
+ /// The folded union carries the WINNER's ordering value: a later update
at ts=5
+ /// must lose to a union whose winner sat at ts=9. Carrying the loser's
ts=2
+ /// instead would let this update take over.
+ #[test]
+ fn test_partial_update_event_time_fold_keeps_winner_ordering() {
+ let mut buffer = build_pu_event_buffer();
+ buffer
+ .process_next_data_record(pu_event_partial("k1", 9, 99), "k1")
+ .unwrap();
+ buffer
+ .process_next_data_record(pu_event_partial_note("k1", 2, "late"),
"k1")
+ .unwrap();
+ buffer
+ .process_next_data_record(pu_event_partial("k1", 5, 55), "k1")
+ .unwrap();
+ assert_eq!(
+ pu_event_drain(buffer),
+ (9, 99, Some("late".to_string())),
+ "ts=5 loses to the union's winner ordering of 9"
+ );
+ }
+
+ /// EVENT_TIME, incoming wins: on a column BOTH partials carry, the
winner's
+ /// value survives. The mirror of the stale-loses tests above, which use
+ /// disjoint columns and so cannot see the overlay direction at all.
+ #[test]
+ fn test_partial_update_event_time_newer_wins_overlapping_column() {
+ let mut buffer = build_pu_event_buffer();
+ buffer
+ .process_next_data_record(pu_event_partial("k1", 2, 11), "k1")
+ .unwrap();
+ buffer
+ .process_next_data_record(pu_event_partial("k1", 9, 99), "k1")
+ .unwrap();
+ assert_eq!(
+ pu_event_drain(buffer),
+ (9, 99, Some("keep".to_string())),
+ "both partials carry `a`; the ts=9 winner's value must survive,
not the ts=2 loser's"
+ );
+ }
+
+ /// EVENT_TIME, incoming wins: the folded union carries the WINNER's
ordering
+ /// value, so a later write between the two loses. Mirrors
+ /// `test_partial_update_event_time_fold_keeps_winner_ordering`, which pins
+ /// the same property on the other arm.
+ #[test]
+ fn test_partial_update_event_time_newer_wins_fold_keeps_winner_ordering() {
+ let mut buffer = build_pu_event_buffer();
+ buffer
+ .process_next_data_record(pu_event_partial("k1", 2, 11), "k1")
+ .unwrap();
+ buffer
+ .process_next_data_record(pu_event_partial("k1", 9, 99), "k1")
+ .unwrap();
+ buffer
+ .process_next_data_record(pu_event_partial("k1", 5, 55), "k1")
+ .unwrap();
+ assert_eq!(
+ pu_event_drain(buffer),
+ (9, 99, Some("keep".to_string())),
+ "ts=5 loses to a union whose winner sat at ts=9, not to the ts=2
loser it folded"
+ );
+ }
+
+ /// COMMIT_TIME_ORDERING is last-writer-wins, so a later partial update
wins
+ /// even when its ordering value went DOWN. Without the short-circuit the
+ /// ordering comparison would hand it to the earlier record, which is
+ /// event-time behavior on a commit-time table.
+ #[test]
+ fn
test_partial_update_commit_time_later_write_wins_a_lower_ordering_value() {
+ let mut buffer = build_pu_buffer("COMMIT_TIME_ORDERING");
+ buffer
+ .process_next_data_record(pu_event_partial("k1", 9, 99), "k1")
+ .unwrap();
+ buffer
+ .process_next_data_record(pu_event_partial("k1", 2, 11), "k1")
+ .unwrap();
+ assert_eq!(
+ pu_event_drain(buffer),
+ (2, 11, Some("keep".to_string())),
+ "last writer wins on a commit-time table, whichever way the
ordering value moved"
+ );
+ }
+
/// A partial column whose type disagrees with the table schema is a LOUD
error
/// (not a silent null) — a present value must never be dropped.
#[test]
diff --git a/crates/core/tests/gold_parity_tests.rs
b/crates/core/tests/gold_parity_tests.rs
index 56b11d13..3c45923b 100644
--- a/crates/core/tests/gold_parity_tests.rs
+++ b/crates/core/tests/gold_parity_tests.rs
@@ -566,6 +566,16 @@ const KNOWN: &[Known] = &[
reader_version: "1",
reason: "version 1 panics on a partial-update block",
},
+ // The same version 1 limitation as the fixture above, reached by the same
+ // route: the partial block's narrower schema is concatenated onto the base
+ // batch positionally, so the key column meets the ordering column.
Version 2
+ // folds both blocks correctly, which is what this fixture exists to pin.
+ Known {
+ fixture: "table_partial_update_event_time",
+ scope: CaseScope::Any,
+ reader_version: "1",
+ reason: "version 1 concatenates a partial-update block onto a wider
base batch",
+ },
Known {
fixture: "table_all_data_types",
scope: CaseScope::Any,
diff --git
a/crates/test/data/quickstart_trips_table/mor/avro/table_partial_update_event_time.sql
b/crates/test/data/quickstart_trips_table/mor/avro/table_partial_update_event_time.sql
new file mode 100644
index 00000000..460d9e46
--- /dev/null
+++
b/crates/test/data/quickstart_trips_table/mor/avro/table_partial_update_event_time.sql
@@ -0,0 +1,74 @@
+/*
+ * 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.
+ */
+
+-- Two PARTIAL-update log blocks on one key, where the second one loses the
+-- ordering. Generated with Spark 3.5.3 and Hudi 1.2.0-SNAPSHOT; gold_data is
+-- `spark.read.format("hudi").load(<table>)` written beside the table
directory.
+--
+-- MERGE INTO writes a partial block only when the table is MOR, the operation
is
+-- an upsert, the table version is at least 8, and the update touches a STRICT
+-- SUBSET of the columns (MergeIntoHoodieTableCommand). Hence four columns and
+-- two-column updates: touching all of them would silently fall back to a full
+-- update and the fixture would prove nothing.
+--
+-- `hoodie.spark.sql.merge.into.partial.updates` defaults to true and is set
here
+-- so the fixture does not depend on that default holding.
+
+CREATE TABLE table_partial_update_event_time (
+ id STRING,
+ ts BIGINT,
+ a STRING,
+ b STRING
+) USING HUDI
+TBLPROPERTIES (
+ type = 'mor',
+ primaryKey = 'id',
+ preCombineField = 'ts',
+ 'hoodie.write.table.version' = '9',
+ 'hoodie.record.merge.mode' = 'EVENT_TIME_ORDERING',
+ 'hoodie.metadata.enable' = 'false',
+ 'hoodie.parquet.small.file.limit' = '0',
+ 'hoodie.compact.inline' = 'false'
+)
+LOCATION 'FIXTURE_LOCATION';
+
+SET hoodie.merge.small.file.group.candidates.limit=0;
+SET hoodie.spark.sql.merge.into.partial.updates=true;
+
+INSERT INTO table_partial_update_event_time VALUES
+ ('k1', 100, 'a-init', 'b-init'),
+ ('k2', 100, 'a-init', 'b-init');
+
+-- Partial update of (ts, a) only. k1 goes high, k2 goes low.
+MERGE INTO table_partial_update_event_time t
+USING (SELECT 'k1' AS id, 300L AS ts, 'a-high' AS a
+ UNION ALL SELECT 'k2' AS id, 200L AS ts, 'a-low' AS a) s
+ON t.id = s.id
+WHEN MATCHED THEN UPDATE SET t.ts = s.ts, t.a = s.a;
+
+-- Partial update of (ts, b) only. k1 now arrives BELOW its own previous log
+-- record and must lose; k2 arrives above and must win. Each key therefore
folds
+-- in the opposite direction, and in both the column only the LOSING record
+-- carries has to survive: k1 keeps b-low, k2 keeps a-low. A fold that keeps
the
+-- winner whole reads b-init / a-init from the base instead.
+MERGE INTO table_partial_update_event_time t
+USING (SELECT 'k1' AS id, 200L AS ts, 'b-low' AS b
+ UNION ALL SELECT 'k2' AS id, 300L AS ts, 'b-high' AS b) s
+ON t.id = s.id
+WHEN MATCHED THEN UPDATE SET t.ts = s.ts, t.b = s.b;
diff --git
a/crates/test/data/quickstart_trips_table/mor/avro/table_partial_update_event_time.zip
b/crates/test/data/quickstart_trips_table/mor/avro/table_partial_update_event_time.zip
new file mode 100644
index 00000000..53c60a9b
Binary files /dev/null and
b/crates/test/data/quickstart_trips_table/mor/avro/table_partial_update_event_time.zip
differ
diff --git a/crates/test/src/lib.rs b/crates/test/src/lib.rs
index 2eabda68..2749b686 100644
--- a/crates/test/src/lib.rs
+++ b/crates/test/src/lib.rs
@@ -274,6 +274,27 @@ pub enum QuickstartTripsTable {
/// merge-correct truth the applied result must match).
#[strum(serialize = "table_partial_update")]
MorLayoutPartialUpdate,
+ /// v9 MOR non-partitioned, EVENT_TIME_ORDERING, where two PARTIAL-update
log
+ /// blocks touch the same key and the second one LOSES the ordering.
+ ///
+ /// Provenance: `table_partial_update_event_time.sql` beside this zip
+ /// (Spark 3.5.3 / Hudi 1.2.0-SNAPSHOT). MERGE INTO writes a partial block
+ /// when the table is MOR, the operation is an upsert, and the update
touches
+ /// a strict subset of the columns.
+ /// Schema: id INT, ts LONG, a STRING, b STRING (non-partitioned).
+ /// Layout: base .parquet (2 rows at ts 100) + log.1 (partial update of
+ /// `ts`,`a`) + log.1 (partial update of `ts`,`b`).
+ /// Semantics: each key folds in the opposite direction. `id=1` takes its
+ /// second update BELOW the first (200 < 300), so the earlier record wins
and
+ /// still absorbs `b` from the loser; `id=2` takes its second update above
+ /// (300 > 200), so the later record wins and absorbs `a`. Either way the
+ /// column only the LOSING record carries must survive, which is what
+ /// distinguishes a both-direction fold from one that keeps the winner
whole.
+ /// `table_partial_update` cannot show this: it is COMMIT_TIME_ORDERING,
+ /// where the incoming record always wins and the fold is one-directional.
+ /// gold_data = Spark `SELECT *` snapshot, 2 rows.
+ #[strum(serialize = "table_partial_update_event_time")]
+ MorPartialUpdateEventTime,
/// v9 MOR non-partitioned, base + 1 HFILE-format log file
/// (`HFILE_DATA_BLOCK`).
///
