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 9f5d1c08 perf(core): match record index keys in place before building
records (#906)
9f5d1c08 is described below
commit 9f5d1c081011962156e33911ffc359acfb69aa41
Author: Y Ethan Guo <[email protected]>
AuthorDate: Fri Oct 30 01:42:27 2026 -0700
perf(core): match record index keys in place before building records (#906)
`DataBlock::entries` walks an HFile data block's entries as borrowed key
and value slices, and fails on an entry that does not fit the block rather than
ending the walk early, since a lookup that stopped quietly would report its
keys as absent. `HFileReader::records_in_block` now takes a predicate and
builds records only for the keys it admits, and the record index log-block scan
passes its key matcher, so a lookup of a few thousand keys among millions of
entries pays key comparisons [...]
On a 10M-row table, a 10,000-key upsert spent about 90% of its time in this
decode before the change.
---
crates/core/src/file_group/log_file/content.rs | 17 ++--
crates/core/src/hfile/block.rs | 114 ++++++++++++++++++++++++-
crates/core/src/hfile/key.rs | 36 ++++----
crates/core/src/hfile/reader.rs | 80 ++++++++++++++---
crates/core/src/index/record.rs | 2 +-
5 files changed, 208 insertions(+), 41 deletions(-)
diff --git a/crates/core/src/file_group/log_file/content.rs
b/crates/core/src/file_group/log_file/content.rs
index 0ed1a49b..64d170ac 100644
--- a/crates/core/src/file_group/log_file/content.rs
+++ b/crates/core/src/file_group/log_file/content.rs
@@ -674,16 +674,13 @@ impl Decoder {
if let Some(predicate) = self.key_predicate.as_ref() {
let matcher = predicate.matcher();
for entry in hfile_reader.blocks_for_predicate(predicate) {
- for record in hfile_reader
- .records_in_block(&entry)
- .map_err(|e| CoreError::HFile(e.to_string()))?
- {
- // A block is the smallest readable unit, so a selected
block
- // holds keys nobody asked for.
- if record.key_as_str().is_some_and(|key|
matcher.admits(key)) {
- records.push(record);
- }
- }
+ // A block is the smallest readable unit, so a selected block
+ // holds keys nobody asked for; only the admitted ones are
built.
+ records.extend(
+ hfile_reader
+ .records_in_block(&entry, |key| matcher.admits(key))
+ .map_err(|e| CoreError::HFile(e.to_string()))?,
+ );
}
return Ok(records);
}
diff --git a/crates/core/src/hfile/block.rs b/crates/core/src/hfile/block.rs
index 67849305..60437569 100644
--- a/crates/core/src/hfile/block.rs
+++ b/crates/core/src/hfile/block.rs
@@ -21,7 +21,7 @@
use crate::hfile::block_type::{HFileBlockType, MAGIC_LENGTH};
use crate::hfile::compression::CompressionCodec;
use crate::hfile::error::{HFileError, Result};
-use crate::hfile::key::{KEY_VALUE_HEADER_SIZE, Key, KeyValue};
+use crate::hfile::key::{KEY_VALUE_HEADER_SIZE, Key, KeyValue, key_content};
/// Size constants for block header
const SIZEOF_INT32: usize = 4;
@@ -211,6 +211,19 @@ impl DataBlock {
offset < self.content_end
}
+ /// The record key content and value of every entry, as slices of the
+ /// block.
+ ///
+ /// [`Self::iter`] copies each entry out of the block; a caller that only
+ /// wants to find a few keys among thousands of entries can look at the
+ /// keys in place and copy the entries it keeps.
+ pub fn entries(&self) -> DataBlockEntries<'_> {
+ DataBlockEntries {
+ block: self,
+ offset: 0,
+ }
+ }
+
/// Iterate over all key-value pairs in the block.
#[allow(dead_code)]
pub fn iter(&self) -> DataBlockIterator<'_> {
@@ -252,6 +265,52 @@ impl<'a> Iterator for DataBlockIterator<'a> {
}
}
+/// Borrowing walk over a data block's entries: see [`DataBlock::entries`].
+pub struct DataBlockEntries<'a> {
+ block: &'a DataBlock,
+ offset: usize,
+}
+
+impl<'a> Iterator for DataBlockEntries<'a> {
+ /// `(key content, value)`, the bytes `KeyValue::key().content()` and
+ /// `KeyValue::value()` return for the entry, or an error for an entry
+ /// that does not fit the block. The error ends the walk: a lookup that
+ /// quietly stopped at a corrupt entry would report its keys as absent.
+ type Item = Result<(&'a [u8], &'a [u8])>;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ let data = &self.block.data[..self.block.content_end];
+ let offset = self.offset;
+ if offset + KEY_VALUE_HEADER_SIZE > data.len() {
+ return None;
+ }
+ let truncated = || {
+ HFileError::InvalidFormat(format!(
+ "HFile data block entry at offset {offset} runs past the block"
+ ))
+ };
+ let key_length = i32::from_be_bytes(data[offset..offset +
4].try_into().ok()?);
+ let value_length = i32::from_be_bytes(data[offset + 4..offset +
8].try_into().ok()?);
+ let (Ok(key_length), Ok(value_length)) =
+ (usize::try_from(key_length), usize::try_from(value_length))
+ else {
+ self.offset = data.len();
+ return Some(Err(truncated()));
+ };
+ let key_start = offset + KEY_VALUE_HEADER_SIZE;
+ let value_start = key_start.saturating_add(key_length);
+ let end = value_start.saturating_add(value_length);
+ // One MVCC timestamp byte follows the value (`KeyValue::record_size`).
+ if end.saturating_add(1) > data.len() {
+ self.offset = data.len();
+ return Some(Err(truncated()));
+ }
+ self.offset = end + 1;
+ let content = key_content(&data[key_start..value_start]);
+ Some(Ok((content, &data[value_start..end])))
+ }
+}
+
/// Read a Hadoop VLong encoded integer from bytes.
/// This is the encoding used in HFile root/leaf index blocks.
/// Returns (value, bytes_consumed).
@@ -301,6 +360,59 @@ pub fn var_long_size_on_disk(bytes: &[u8], offset: usize)
-> usize {
#[cfg(test)]
mod tests {
+ /// One entry as the writer lays it out: lengths, structured key, value,
+ /// MVCC byte.
+ fn entry(content: &[u8], value: &[u8]) -> Vec<u8> {
+ let mut key = (content.len() as u16).to_be_bytes().to_vec();
+ key.extend_from_slice(content);
+ key.push(0);
+ key.extend_from_slice(&i64::MAX.to_be_bytes());
+ key.push(4);
+ let mut out = (key.len() as i32).to_be_bytes().to_vec();
+ out.extend_from_slice(&(value.len() as i32).to_be_bytes());
+ out.extend_from_slice(&key);
+ out.extend_from_slice(value);
+ out.push(0);
+ out
+ }
+
+ fn data_block(bytes: Vec<u8>) -> super::DataBlock {
+ let content_end = bytes.len();
+ super::DataBlock {
+ data: bytes,
+ content_end,
+ }
+ }
+
+ /// The borrowing walk sees what the copying iterator sees, and a
+ /// truncated entry is an error rather than a silent end.
+ #[test]
+ fn entries_match_key_values_and_reject_a_truncated_entry() {
+ let mut bytes = entry(b"k1", b"v1");
+ bytes.extend(entry(b"k2", b"value two"));
+ let block = data_block(bytes.clone());
+ let walked: Vec<(Vec<u8>, Vec<u8>)> = block
+ .entries()
+ .map(|e| e.map(|(k, v)| (k.to_vec(), v.to_vec())).unwrap())
+ .collect();
+ let copied: Vec<(Vec<u8>, Vec<u8>)> = block
+ .iter()
+ .map(|kv| (kv.key().content().to_vec(), kv.value().to_vec()))
+ .collect();
+ assert_eq!(walked, copied);
+ assert_eq!(walked[1], (b"k2".to_vec(), b"value two".to_vec()));
+
+ // Drop the tail of the second entry.
+ let truncated = data_block(bytes[..bytes.len() - 4].to_vec());
+ let results: Vec<_> = truncated.entries().collect();
+ assert_eq!(results.len(), 2);
+ assert!(results[0].is_ok());
+ assert!(matches!(
+ results[1],
+ Err(super::HFileError::InvalidFormat(_))
+ ));
+ }
+
use super::*;
#[test]
diff --git a/crates/core/src/hfile/key.rs b/crates/core/src/hfile/key.rs
index 4733ebca..822cee0f 100644
--- a/crates/core/src/hfile/key.rs
+++ b/crates/core/src/hfile/key.rs
@@ -91,15 +91,6 @@ impl Key {
Some(Self::from_bytes(bytes))
}
- /// Returns the offset to the key content (after the length prefix).
- ///
- /// Private: this is an offset into *this key's* bytes, so it is only
meaningful
- /// with [`Self::bytes`]. It used to be an offset into the enclosing
block, and a
- /// caller pairing it with a block buffer would now index the wrong place.
- fn content_offset(&self) -> usize {
- self.offset + SIZEOF_INT16
- }
-
/// Returns the length of the key content.
pub fn content_length(&self) -> usize {
if self.bytes.len() < self.offset + SIZEOF_INT16 {
@@ -117,12 +108,7 @@ impl Key {
/// follows it. A well-formed HFile cannot reach that: `key_length` is
/// `2 + row + 1 + family + qualifier + 9`, always at least `2 + content`.
pub fn content(&self) -> &[u8] {
- let start = self.content_offset();
- let len = self.content_length();
- if start + len > self.bytes.len() {
- return &[];
- }
- &self.bytes[start..start + len]
+ key_content(&self.bytes[self.offset..])
}
/// Returns the key content as a UTF-8 string.
@@ -141,6 +127,20 @@ impl Key {
}
}
+/// The content of a key: a 2-byte signed length, then that many bytes.
+///
+/// Empty when the length is missing, negative, or longer than the key, rather
+/// than reading on into the value that follows the key in a block.
+pub(crate) fn key_content(key: &[u8]) -> &[u8] {
+ let Some((length, rest)) = key.split_first_chunk::<SIZEOF_INT16>() else {
+ return &[];
+ };
+ usize::try_from(i16::from_be_bytes(*length))
+ .ok()
+ .and_then(|length| rest.get(..length))
+ .unwrap_or(&[])
+}
+
impl PartialEq for Key {
fn eq(&self, other: &Self) -> bool {
self.content() == other.content()
@@ -401,6 +401,12 @@ mod tests {
assert_eq!(key.content_length(), 0);
}
+ #[test]
+ fn test_key_content_rejects_a_negative_length() {
+ let key = Key::from_bytes(vec![0xFF, 0xFF, b'a', b'b']);
+ assert_eq!(key.content(), b"");
+ }
+
#[test]
fn test_key_content_out_of_bounds() {
// Key claims content length of 10 but only has 3 bytes
diff --git a/crates/core/src/hfile/reader.rs b/crates/core/src/hfile/reader.rs
index adbf4a3a..c097cc9d 100644
--- a/crates/core/src/hfile/reader.rs
+++ b/crates/core/src/hfile/reader.rs
@@ -413,14 +413,39 @@ impl HFileReader {
}
}
- /// The records of one data block, from a resident source.
+ /// The records of one data block whose key `admits`, from a resident
source.
///
/// The synchronous counterpart to [`Self::read_records_batched`], which
needs a
/// ranged source because it fetches. An HFile arriving as a log block's
content
/// is already in memory, so a caller that has narrowed the blocks with
/// [`Self::blocks_for_predicate`] wants to parse just those rather than
walk the
- /// file — the saving is parsing and decompression, not I/O.
- pub fn records_in_block(&self, entry: &BlockIndexEntry) ->
Result<Vec<HFileRecord>> {
+ /// file — the saving is parsing and decompression, not I/O. Keys are
compared
+ /// in place and only the admitted entries are copied out, so a lookup of
a few
+ /// keys in a block of thousands costs the key comparisons rather than a
record
+ /// allocation per entry.
+ pub fn records_in_block(
+ &self,
+ entry: &BlockIndexEntry,
+ admits: impl Fn(&str) -> bool,
+ ) -> Result<Vec<HFileRecord>> {
+ let data_block = self.data_block(entry)?;
+ // Records carry the file's own writer schema so a decoder can honor
the
+ // shape this file was written with rather than a vendored one.
+ let schema = self.get_avro_schema_arc()?;
+ let mut records = Vec::new();
+ for entry in data_block.entries() {
+ let (key, value) = entry?;
+ if std::str::from_utf8(key).is_ok_and(&admits) {
+ records.push(
+ HFileRecord::new(key.to_vec(),
value.to_vec()).with_avro_schema(schema.clone()),
+ );
+ }
+ }
+ Ok(records)
+ }
+
+ /// The data block an index entry points at.
+ fn data_block(&self, entry: &BlockIndexEntry) -> Result<DataBlock> {
let block = self.read_block_at(entry.offset as usize, entry.size as
usize)?;
if block.block_type() != HFileBlockType::Data {
return Err(HFileError::UnexpectedBlockType {
@@ -428,17 +453,7 @@ impl HFileReader {
actual: block.block_type().to_string(),
});
}
- let data_block = DataBlock::from_block(block);
- // Records carry the file's own writer schema so a decoder can honor
the
- // shape this file was written with rather than a vendored one.
- let schema = self.get_avro_schema_arc()?;
- // The shared helper, not a second copy of it: two places building an
- // `HFileRecord` from a `KeyValue` is how the predicate path and the
scan path
- // would drift on what a record key is.
- Ok(data_block
- .iter()
- .map(|kv| Self::key_value_to_record(&kv, schema.clone()))
- .collect())
+ Ok(DataBlock::from_block(block))
}
/// The data blocks a key predicate can be satisfied from.
@@ -1993,6 +2008,43 @@ mod tests {
assert_eq!(record.key_as_str(), Some("hudi-key-000000000"));
}
+ /// The in-place key walk sees the same entries as the copying iterator,
+ /// and an admitting predicate keeps exactly the entries it admits.
+ #[test]
+ fn test_records_in_block_matches_the_copying_iterator() {
+ let bytes =
read_test_hfile("hudi_1_0_hbase_2_4_9_16KB_NONE_5000.hfile");
+ let reader = HFileReader::new(bytes).expect("Failed to create reader");
+ let entry = reader.blocks_for_keys(&["hudi-key-000000100"])[0].clone();
+ let pairs = |records: Vec<HFileRecord>| -> Vec<(Vec<u8>, Vec<u8>)> {
+ records
+ .iter()
+ .map(|r| (r.key().to_vec(), r.value().to_vec()))
+ .collect()
+ };
+ let copied: Vec<(Vec<u8>, Vec<u8>)> = reader
+ .data_block(&entry)
+ .unwrap()
+ .iter()
+ .map(|kv| (kv.key().content().to_vec(), kv.value().to_vec()))
+ .collect();
+ assert!(copied.len() > 1);
+ assert_eq!(
+ pairs(reader.records_in_block(&entry, |_| true).unwrap()),
+ copied
+ );
+
+ let wanted = |key: &str| key.ends_with("100") || key.ends_with("101");
+ let expected: Vec<(Vec<u8>, Vec<u8>)> = copied
+ .into_iter()
+ .filter(|(k, _)| wanted(std::str::from_utf8(k).unwrap()))
+ .collect();
+ assert_eq!(expected.len(), 2);
+ assert_eq!(
+ pairs(reader.records_in_block(&entry, wanted).unwrap()),
+ expected
+ );
+ }
+
#[test]
fn test_lookup_records() {
let bytes =
read_test_hfile("hudi_1_0_hbase_2_4_9_16KB_NONE_5000.hfile");
diff --git a/crates/core/src/index/record.rs b/crates/core/src/index/record.rs
index 2ebece78..808141d1 100644
--- a/crates/core/src/index/record.rs
+++ b/crates/core/src/index/record.rs
@@ -378,7 +378,7 @@ mod tests {
let mut firsts = Vec::new();
let mut lasts = Vec::new();
for e in &entries {
- let recs = reader.records_in_block(e).unwrap();
+ let recs = reader.records_in_block(e, |_| true).unwrap();
firsts.push(recs.first().unwrap().key_as_str().unwrap().to_string());
lasts.push(recs.last().unwrap().key_as_str().unwrap().to_string());
}
