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 a1e6103e perf(core): build commit sequence numbers without a per-row
allocation (#896)
a1e6103e is described below
commit a1e6103e1817cce146a402b418c031fda2c5e617
Author: Y Ethan Guo <[email protected]>
AuthorDate: Fri Oct 30 08:59:30 2026 -0700
perf(core): build commit sequence numbers without a per-row allocation
(#896)
`_hoodie_commit_seqno` was built with a `format!` per row into a
`Vec<String>` before becoming an Arrow array. A `commit_seqnos` helper now
writes each value straight into a `StringBuilder`, so the column is one buffer
with no per-row allocation, and both the append path and the copy-on-write
restamp of updated rows use it. The values, type and nullability are unchanged.
Locally this cut append CPU time by about 0.25 s for 2.5M rows and 0.5 s
for 10M rows.
---
crates/core/src/write/append.rs | 42 +++++++++++++++++++++++++++++-------
crates/core/src/write/rewrite/cow.rs | 9 ++++----
2 files changed, 39 insertions(+), 12 deletions(-)
diff --git a/crates/core/src/write/append.rs b/crates/core/src/write/append.rs
index 3d869039..c3831d1b 100644
--- a/crates/core/src/write/append.rs
+++ b/crates/core/src/write/append.rs
@@ -1055,9 +1055,23 @@ fn schemas_equal_by_fields(a: &arrow_schema::Schema, b:
&arrow_schema::Schema) -
.all(|(fa, fb)| fa.name() == fb.name() && fa.data_type() ==
fb.data_type())
}
-/// `_hoodie_commit_seqno`, in Java's `HoodieRecord.generateSequenceId` form.
-pub(crate) fn commit_seqno(instant: &str, partition_id: u32, row: usize) ->
String {
- format!("{instant}_{partition_id}_{row}")
+/// `_hoodie_commit_seqno` for `rows` rows numbered from `row_id_offset`, in
Java's
+/// `HoodieRecord.generateSequenceId` form, written straight into the array's
buffer.
+pub(crate) fn commit_seqnos(
+ instant: &str,
+ partition_id: u32,
+ row_id_offset: usize,
+ rows: usize,
+) -> StringArray {
+ use std::fmt::Write;
+ let prefix = format!("{instant}_{partition_id}_");
+ let mut builder = arrow::array::StringBuilder::with_capacity(rows, rows *
(prefix.len() + 10));
+ for row in row_id_offset..row_id_offset + rows {
+ // The builder's `fmt::Write` appends to its value buffer and never
fails.
+ let _ = write!(builder, "{prefix}{row}");
+ builder.append_value("");
+ }
+ builder.finish()
}
pub(crate) fn prepare_batches_for_write(
@@ -1114,11 +1128,7 @@ pub(crate) fn prepare_batches_for_write_with_offset(
fields.extend(batch.schema().fields().iter().cloned());
let mut columns: Vec<ArrayRef> = vec![
Arc::new(StringArray::from(vec![instant; rows])),
- Arc::new(StringArray::from(
- (0..rows)
- .map(|row| commit_seqno(instant, partition_id,
row_id_offset + row))
- .collect::<Vec<_>>(),
- )),
+ Arc::new(commit_seqnos(instant, partition_id, row_id_offset,
rows)),
Arc::new(StringArray::from(key_strs)),
Arc::new(StringArray::from(partition_strs)),
Arc::new(StringArray::from(vec![file_name; rows])),
@@ -1161,6 +1171,22 @@ mod tests {
Ok(())
}
+ #[test]
+ fn test_commit_seqnos_number_rows_from_the_offset() {
+ use arrow::array::Array;
+ let seqnos = commit_seqnos("20260101000000000", 3, 98, 3);
+ let got: Vec<&str> = seqnos.iter().map(Option::unwrap).collect();
+ assert_eq!(
+ got,
+ [
+ "20260101000000000_3_98",
+ "20260101000000000_3_99",
+ "20260101000000000_3_100",
+ ]
+ );
+ assert_eq!(commit_seqnos("20260101000000000", 0, 0, 0).len(), 0);
+ }
+
#[test]
fn test_arrow_schema_to_avro_json_type_matrix() {
let schema = Schema::new(vec![
diff --git a/crates/core/src/write/rewrite/cow.rs
b/crates/core/src/write/rewrite/cow.rs
index c3225d54..0ef063fb 100644
--- a/crates/core/src/write/rewrite/cow.rs
+++ b/crates/core/src/write/rewrite/cow.rs
@@ -443,10 +443,11 @@ pub(super) fn restamp_matched_rows(
return Ok(batch.clone());
};
let new_ct: ArrayRef = std::sync::Arc::new(StringArray::from(vec![instant;
batch.num_rows()]));
- let new_sq: ArrayRef = std::sync::Arc::new(StringArray::from(
- (0..batch.num_rows())
- .map(|row| crate::write::append::commit_seqno(instant, 0, row))
- .collect::<Vec<_>>(),
+ let new_sq: ArrayRef =
std::sync::Arc::new(crate::write::append::commit_seqnos(
+ instant,
+ 0,
+ 0,
+ batch.num_rows(),
));
let mut columns = batch.columns().to_vec();
columns[ct_idx] = zip(mask, &new_ct, &columns[ct_idx])?;
