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 efaebc36 perf(core): compute the record index instant once per commit
(#892)
efaebc36 is described below
commit efaebc36b48fd45239f9afe7a8c42ee3acdfc35c
Author: Y Ethan Guo <[email protected]>
AuthorDate: Thu Oct 29 23:49:23 2026 -0700
perf(core): compute the record index instant once per commit (#892)
Each record-level index entry computed `instant_time_millis` from the
commit's instant string and the table's timeline time zone, once per record.
The six writers that build these entries (append, the copy-on-write and
merge-on-read rewrites, merge-on-read deletes) now compute it once per call and
reuse it. The value is identical for every record of a commit, so the written
entries do not change. The append path also no longer builds the entries when
the record index is off, since the [...]
Appending 2.5M rows with the record index off went from about 2.2 s to
about 1.1 s locally from computing the instant once; skipping the entries
removes the rest of that per-row work.
---
crates/core/src/write/append.rs | 27 ++++++++++++++++-----------
crates/core/src/write/rewrite/cow.rs | 9 +++++----
crates/core/src/write/rewrite/mod.rs | 9 +++++----
crates/core/src/write/rewrite/mor.rs | 23 +++++++++++------------
4 files changed, 37 insertions(+), 31 deletions(-)
diff --git a/crates/core/src/write/append.rs b/crates/core/src/write/append.rs
index c54a2e04..7c4100b0 100644
--- a/crates/core/src/write/append.rs
+++ b/crates/core/src/write/append.rs
@@ -124,6 +124,12 @@ pub async fn append_batches(table: &mut Table, batches:
&[RecordBatch]) -> Resul
let mut write_stats: HashMap<String, Vec<HoodieWriteStat>> =
HashMap::new();
let mut files_mdt = Vec::new();
let mut rli_entries = Vec::new();
+ let rli_instant_millis =
crate::index::is_record_index_enabled(table).then(|| {
+ crate::write::metadata::instant_to_epoch_millis_in(
+ &request_instant,
+ &crate::write::metadata::timeline_timezone(&table.hudi_configs),
+ )
+ });
let mut stats_updates = Vec::new();
let mut written_paths: Vec<String> = Vec::new();
let mut primary_base_path = String::new();
@@ -301,17 +307,16 @@ pub async fn append_batches(table: &mut Table, batches:
&[RecordBatch]) -> Resul
plan.file_name.clone(),
output.size,
));
- for key in &plan.chunk_keys {
- rli_entries.push(RecordIndexEntry {
- record_key: key.record_key.clone(),
- partition_path: plan.partition_path.clone(),
- file_id: plan.file_id.clone(),
- instant_time_millis:
crate::write::metadata::instant_to_epoch_millis_in(
- &request_instant,
-
&crate::write::metadata::timeline_timezone(&table.hudi_configs),
- ),
- is_deleted: false,
- });
+ if let Some(instant_time_millis) = rli_instant_millis {
+ for key in &plan.chunk_keys {
+ rli_entries.push(RecordIndexEntry {
+ record_key: key.record_key.clone(),
+ partition_path: plan.partition_path.clone(),
+ file_id: plan.file_id.clone(),
+ instant_time_millis,
+ is_deleted: false,
+ });
+ }
}
}
Err(error) => {
diff --git a/crates/core/src/write/rewrite/cow.rs
b/crates/core/src/write/rewrite/cow.rs
index 93ff5fa7..83538acf 100644
--- a/crates/core/src/write/rewrite/cow.rs
+++ b/crates/core/src/write/rewrite/cow.rs
@@ -291,6 +291,10 @@ pub(super) async fn rewrite(
return Err(error);
}
+ let instant_time_millis =
crate::write::metadata::instant_to_epoch_millis_in(
+ instant,
+ &crate::write::metadata::timeline_timezone(&table.hudi_configs),
+ );
let rli_entries = {
let mut entries = if batch.num_rows() > 0 &&
is_record_index_enabled(table) {
hoodie_keys_for_batch(&table.hudi_configs, batch, Some(instant))?
@@ -302,10 +306,7 @@ pub(super) async fn rewrite(
record_key: key.record_key,
partition_path: key.partition_path,
file_id,
- instant_time_millis:
crate::write::metadata::instant_to_epoch_millis_in(
- instant,
-
&crate::write::metadata::timeline_timezone(&table.hudi_configs),
- ),
+ instant_time_millis,
is_deleted: false,
})
})
diff --git a/crates/core/src/write/rewrite/mod.rs
b/crates/core/src/write/rewrite/mod.rs
index ea57cd25..28f9ac48 100644
--- a/crates/core/src/write/rewrite/mod.rs
+++ b/crates/core/src/write/rewrite/mod.rs
@@ -426,6 +426,10 @@ pub async fn upsert_batches(
}
}
+ let instant_time_millis =
crate::write::metadata::instant_to_epoch_millis_in(
+ &instant,
+ &crate::write::metadata::timeline_timezone(&table.hudi_configs),
+ );
// RLI adds: every input row's final location is known from the plan. A
// REJECTED cross-partition move keeps the old row, so its index entry
// must carry the old partition, not the losing incoming value.
@@ -449,10 +453,7 @@ pub async fn upsert_batches(
record_key: key.record_key.clone(),
partition_path,
file_id,
- instant_time_millis:
crate::write::metadata::instant_to_epoch_millis_in(
- &instant,
-
&crate::write::metadata::timeline_timezone(&table.hudi_configs),
- ),
+ instant_time_millis,
is_deleted: false,
})
})
diff --git a/crates/core/src/write/rewrite/mor.rs
b/crates/core/src/write/rewrite/mor.rs
index 30303736..1febfca2 100644
--- a/crates/core/src/write/rewrite/mor.rs
+++ b/crates/core/src/write/rewrite/mor.rs
@@ -123,6 +123,10 @@ pub(super) async fn mor_upsert_batches(
std::slice::from_ref(&incoming),
)?;
let instant = request_deltacommit(table, "UPSERT").await?;
+ let instant_time_millis =
crate::write::metadata::instant_to_epoch_millis_in(
+ &instant,
+ &crate::write::metadata::timeline_timezone(&table.hudi_configs),
+ );
let tagged_keys = hoodie_keys_for_batch(&table.hudi_configs, &incoming,
Some(&instant))?;
let incoming_keys: Vec<String> = tagged_keys.iter().map(|k|
k.record_key.clone()).collect();
let locations = mor_file_locations(table, &tagged_keys).await?;
@@ -450,10 +454,7 @@ pub(super) async fn mor_upsert_batches(
record_key: key.record_key.clone(),
partition_path: partition_path.clone(),
file_id: file_id.clone(),
- instant_time_millis:
crate::write::metadata::instant_to_epoch_millis_in(
- &instant,
-
&crate::write::metadata::timeline_timezone(&table.hudi_configs),
- ),
+ instant_time_millis,
is_deleted: false,
}
}));
@@ -524,10 +525,7 @@ pub(super) async fn mor_upsert_batches(
record_key: key.record_key.clone(),
partition_path: partition_path.clone(),
file_id: file_id.clone(),
- instant_time_millis:
crate::write::metadata::instant_to_epoch_millis_in(
- &instant,
-
&crate::write::metadata::timeline_timezone(&table.hudi_configs),
- ),
+ instant_time_millis,
is_deleted: false,
}
}));
@@ -784,6 +782,10 @@ pub(super) async fn mor_delete_keys(
..Default::default()
});
}
+ let instant_time_millis =
crate::write::metadata::instant_to_epoch_millis_in(
+ &instant,
+ &crate::write::metadata::timeline_timezone(&table.hudi_configs),
+ );
let record_index = delete_keys
.iter()
.filter(|key| locations.contains_key(&key.record_key))
@@ -791,10 +793,7 @@ pub(super) async fn mor_delete_keys(
record_key: key.record_key.clone(),
partition_path: key.partition_path.clone(),
file_id: String::new(),
- instant_time_millis:
crate::write::metadata::instant_to_epoch_millis_in(
- &instant,
-
&crate::write::metadata::timeline_timezone(&table.hudi_configs),
- ),
+ instant_time_millis,
is_deleted: true,
})
.collect::<Vec<_>>();
