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 c9366eb2 perf(core): merge only matched keys in event-time
copy-on-write upserts (#882)
c9366eb2 is described below
commit c9366eb288ed8904852803fb8ce43ac145cdd025
Author: Y Ethan Guo <[email protected]>
AuthorDate: Thu Oct 29 22:37:07 2026 -0700
perf(core): merge only matched keys in event-time copy-on-write upserts
(#882)
With an ordering field, a copy-on-write upsert passed every stored row of a
touched file group through the record merger, although only stored rows
whose
key is incoming can change. The merger now sees those rows plus the incoming
ones; every other stored row stays in place. A file group with repeated
stored keys still goes through the full merge. Rows are written in the order
the commit-time path already uses.
Closes #881
---
crates/core/src/write/rewrite/cow.rs | 8 +-
crates/core/src/write/rewrite/rows.rs | 130 +++++++++++++++++++++++++++++++--
crates/core/tests/table_write_tests.rs | 111 ++++++++++++++++++++++++++++
3 files changed, 242 insertions(+), 7 deletions(-)
diff --git a/crates/core/src/write/rewrite/cow.rs
b/crates/core/src/write/rewrite/cow.rs
index a0b45e46..93ff5fa7 100644
--- a/crates/core/src/write/rewrite/cow.rs
+++ b/crates/core/src/write/rewrite/cow.rs
@@ -569,7 +569,13 @@ pub(super) fn merge_file_group_task(
concat_batches(&old.schema(), [&old, &incoming])?
};
let selected = if ctx.event_time && old.num_rows() > 0 {
- merge_with_event_time(ctx.hudi_configs.clone(), &old,
&incoming, &combined)?
+ merge_with_event_time(
+ ctx.hudi_configs.clone(),
+ &combined,
+ old.num_rows(),
+ &old_by_key,
+ &incoming_keys,
+ )?
} else {
selected
};
diff --git a/crates/core/src/write/rewrite/rows.rs
b/crates/core/src/write/rewrite/rows.rs
index bbce1e07..e6d628be 100644
--- a/crates/core/src/write/rewrite/rows.rs
+++ b/crates/core/src/write/rewrite/rows.rs
@@ -21,18 +21,40 @@
use super::*;
+/// Indices into `combined` (the group's `num_old` stored rows, then the
+/// incoming rows) that survive an event-time merge, in `combined` order.
+///
+/// Only the stored rows whose key is incoming can change, so the merger sees
+/// just those and the incoming rows. A group with a repeated stored key sends
+/// every stored row through it, since the merger also collapses repeats.
pub(super) fn merge_with_event_time(
hudi_configs: std::sync::Arc<crate::config::HudiConfigs>,
- old: &RecordBatch,
- incoming: &RecordBatch,
combined: &RecordBatch,
+ num_old: usize,
+ old_by_key: &HashMap<String, u32>,
+ incoming_keys: &[String],
) -> Result<Vec<u32>> {
- let merger = RecordMerger::new(old.schema(), hudi_configs);
+ let mut touched = vec![old_by_key.len() < num_old; num_old];
+ for key in incoming_keys {
+ if let Some(&index) = old_by_key.get(key) {
+ touched[index as usize] = true;
+ }
+ }
+ let (mut candidates, mut selected): (Vec<u32>, Vec<u32>) =
+ (0..num_old as u32).partition(|&index| touched[index as usize]);
+ candidates.extend(num_old as u32..combined.num_rows() as u32);
+ let candidate_rows = take_batch(combined, &candidates)?;
+ let merger = RecordMerger::new(combined.schema(), hudi_configs);
let merged =
merger.merge_record_batches(RecordBatches::new_with_data_batches([
- old.clone(),
- incoming.clone(),
+ candidate_rows.clone(),
]))?;
- selected_indices_for_merged(combined, &merged)
+ selected.extend(
+ selected_indices_for_merged(&candidate_rows, &merged)?
+ .into_iter()
+ .map(|index| candidates[index as usize]),
+ );
+ selected.sort_unstable();
+ Ok(selected)
}
pub(super) fn selected_indices_for_merged(
@@ -480,3 +502,99 @@ pub(super) fn partial_merge(
}
RecordBatch::try_new(combined.schema(), columns).map_err(Into::into)
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::config::HudiConfigs;
+ use arrow::array::Int64Array;
+ use arrow_schema::{DataType, Field, Schema};
+ use std::sync::Arc;
+
+ /// Rows are `(commit time, key, ts)`; the seqno is unique per row.
+ fn combined(rows: &[(&str, &str, i64)]) -> RecordBatch {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new(MetaField::CommitTime.as_ref(), DataType::Utf8, false),
+ Field::new(MetaField::CommitSeqno.as_ref(), DataType::Utf8, false),
+ Field::new(MetaField::RecordKey.as_ref(), DataType::Utf8, false),
+ Field::new("ts", DataType::Int64, false),
+ ]));
+ let seqnos: Vec<String> = rows
+ .iter()
+ .enumerate()
+ .map(|(index, (commit, _, _))| format!("{commit}_0-{index}-0"))
+ .collect();
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from_iter_values(rows.iter().map(|r|
r.0))),
+ Arc::new(StringArray::from(seqnos)),
+ Arc::new(StringArray::from_iter_values(rows.iter().map(|r|
r.1))),
+ Arc::new(Int64Array::from_iter_values(rows.iter().map(|r|
r.2))),
+ ],
+ )
+ .unwrap()
+ }
+
+ /// The selection over every row of the group, sorted into `combined`
order.
+ fn full_merge(configs: Arc<HudiConfigs>, combined: &RecordBatch) ->
Vec<u32> {
+ let merged = RecordMerger::new(combined.schema(), configs)
+
.merge_record_batches(RecordBatches::new_with_data_batches([combined.clone()]))
+ .unwrap();
+ let mut selected = selected_indices_for_merged(combined,
&merged).unwrap();
+ selected.sort_unstable();
+ selected
+ }
+
+ #[test]
+ fn test_merge_with_event_time_matches_full_merge() {
+ let configs = Arc::new(HudiConfigs::new([
+ (RecordMergeStrategy, "OVERWRITE_WITH_LATEST".to_string()),
+ (OrderingFields, "ts".to_string()),
+ ]));
+ // Incoming: lower (a), equal (b) and higher (c) ordering than stored,
+ // two versions of d, and a new key f twice; e is untouched.
+ let incoming = [
+ ("002", "a", 50),
+ ("002", "b", 100),
+ ("002", "c", 150),
+ ("002", "d", 120),
+ ("002", "d", 130),
+ ("002", "f", 10),
+ ("002", "f", 20),
+ ];
+ let unique = vec![
+ ("001", "e", 100),
+ ("001", "a", 100),
+ ("001", "b", 100),
+ ("001", "c", 100),
+ ("001", "d", 100),
+ ];
+ let mut repeated = unique.clone();
+ repeated.extend([("001", "e", 90), ("001", "a", 110)]);
+ for old in [unique, repeated] {
+ let rows: Vec<_> =
old.iter().chain(incoming.iter()).copied().collect();
+ let combined = combined(&rows);
+ let keys = keys(&combined, MetaField::RecordKey.as_ref()).unwrap();
+ let num_old = old.len();
+ let old_by_key: HashMap<String, u32> = keys[..num_old]
+ .iter()
+ .enumerate()
+ .map(|(index, key)| (key.clone(), index as u32))
+ .collect();
+ let selected = merge_with_event_time(
+ configs.clone(),
+ &combined,
+ num_old,
+ &old_by_key,
+ &keys[num_old..],
+ )
+ .unwrap();
+ assert_eq!(selected, full_merge(configs.clone(), &combined),
"{old:?}");
+ if num_old == 5 {
+ // e and a stored; b, c, the later d and the later f incoming.
+ assert_eq!(selected, vec![0, 1, 6, 7, 9, 11]);
+ }
+ }
+ }
+}
diff --git a/crates/core/tests/table_write_tests.rs
b/crates/core/tests/table_write_tests.rs
index 533dbd82..fa3e9c0c 100644
--- a/crates/core/tests/table_write_tests.rs
+++ b/crates/core/tests/table_write_tests.rs
@@ -885,6 +885,117 @@ async fn test_upsert_uses_event_time_ordering() {
assert_eq!(values.value(0), 4);
}
+/// Rows are `(id, value, event_time, _hoodie_is_deleted)`.
+fn ordered_flagged_batch(rows: Vec<(&str, i64, i64, bool)>) -> RecordBatch {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("value", DataType::Int64, false),
+ Field::new("event_time", DataType::Int64, false),
+ Field::new("_hoodie_is_deleted", DataType::Boolean, true),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.0))),
+ Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.1))),
+ Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.2))),
+ Arc::new(BooleanArray::from(
+ rows.iter().map(|r| r.3).collect::<Vec<_>>(),
+ )),
+ ],
+ )
+ .unwrap()
+}
+
+fn ordered_flagged_rows(batches: &[RecordBatch]) -> Vec<(String, i64, i64,
bool)> {
+ let mut rows = Vec::new();
+ for batch in batches {
+ let column = |name: &str| batch.column_by_name(name).unwrap().as_any();
+ let ids = column("id").downcast_ref::<StringArray>().unwrap();
+ let values = column("value").downcast_ref::<Int64Array>().unwrap();
+ let times = column("event_time").downcast_ref::<Int64Array>().unwrap();
+ let deleted = column("_hoodie_is_deleted")
+ .downcast_ref::<BooleanArray>()
+ .unwrap();
+ for row in 0..batch.num_rows() {
+ rows.push((
+ ids.value(row).to_string(),
+ values.value(row),
+ times.value(row),
+ deleted.value(row),
+ ));
+ }
+ }
+ rows.sort();
+ rows
+}
+
+/// Copy-on-write event-time upserts into a group where most stored rows are
+/// untouched: each touched key resolves by ordering value (ties go to the
+/// incoming row), a winning delete marker removes the row, in-batch duplicates
+/// keep the highest ordering value, and untouched rows pass through.
+#[tokio::test]
+async fn test_cow_event_time_upsert_merges_only_touched_keys() {
+ let dir = tempdir().unwrap();
+ let mut table = Table::create(dir.path().to_str().unwrap())
+ .with_table_name("t")
+ .with_record_key_fields(["id"])
+ .with_ordering_fields(["event_time"])
+ .with_populates_meta_fields(true)
+ .create()
+ .await
+ .unwrap();
+ let stored: Vec<_> = ["a", "b", "c", "d", "e", "f", "g", "h"]
+ .into_iter()
+ .zip(1..)
+ .map(|(id, value)| (id, value, 100, false))
+ .collect();
+ table.upsert([ordered_flagged_batch(stored)]).await.unwrap();
+ table
+ .upsert([ordered_flagged_batch(vec![
+ ("a", 10, 50, false),
+ ("b", 20, 100, false),
+ ("c", 30, 150, false),
+ ("c", 31, 120, false),
+ ("d", 40, 150, true),
+ ("e", 50, 50, true),
+ ("i", 90, 10, false),
+ ("j", 100, 10, false),
+ ("j", 101, 20, false),
+ ])])
+ .await
+ .unwrap();
+ table
+ .upsert_with(
+ [ordered_flagged_batch(vec![
+ ("g", 70, 50, false),
+ ("h", 80, 200, false),
+ ])],
+ UpsertOptions {
+ update_columns: Some(vec!["value".to_string()]),
+ },
+ )
+ .await
+ .unwrap();
+
+ let batches = table.read(&ReadOptions::new()).await.unwrap();
+ let expected = [
+ ("a", 1, 100),
+ ("b", 20, 100),
+ ("c", 30, 150),
+ ("e", 5, 100),
+ ("f", 6, 100),
+ ("g", 7, 100),
+ ("h", 80, 100),
+ ("i", 90, 10),
+ ("j", 101, 20),
+ ]
+ .into_iter()
+ .map(|(id, value, event_time)| (id.to_string(), value, event_time, false))
+ .collect::<Vec<_>>();
+ assert_eq!(ordered_flagged_rows(&batches), expected);
+}
+
#[tokio::test]
async fn test_merge_on_read_writes_parquet_logs_and_merges_snapshot() {
let dir = tempdir().unwrap();
