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 73d214b7 perf(core): read only the requested keys' record index shards
and blocks (#869)
73d214b7 is described below
commit 73d214b76e7dd1b5508a482fd2cd66b624690277
Author: Y Ethan Guo <[email protected]>
AuthorDate: Thu Oct 29 21:47:46 2026 -0700
perf(core): read only the requested keys' record index shards and blocks
(#869)
Every upsert and delete tagged its keys by loading the whole `record_index`
metadata partition and Avro-decoding every entry, so the cost grew with the
table rather than with the batch. The lookup now lists the partition's live
slices once, hashes each key to its shard the way the writer placed it, reads
each touched shard's base through its HFile block index and its log files with
the keys pushed into the HFile block decoder, and decodes only the requested
entries. Shards are read wi [...]
On a 1M-row table with the default ten shards, a one-key lookup went from
about 11 s to about 0.01 s, and a 10,000-key lookup takes about 0.16 s (local
SSD, measured with keys that do not exist so the time is the lookup alone).
---
crates/core/src/file_group/log_file/reader.rs | 16 ++
crates/core/src/file_group/log_file/scanner.rs | 74 ++++-
crates/core/src/index/record.rs | 335 ++++++++++++++++++++---
crates/core/src/write/metadata.rs | 43 ++-
crates/core/tests/table_write_lifecycle_tests.rs | 63 +++++
5 files changed, 478 insertions(+), 53 deletions(-)
diff --git a/crates/core/src/file_group/log_file/reader.rs
b/crates/core/src/file_group/log_file/reader.rs
index 54e04227..0d09ba6c 100644
--- a/crates/core/src/file_group/log_file/reader.rs
+++ b/crates/core/src/file_group/log_file/reader.rs
@@ -21,6 +21,7 @@ use crate::Result;
use crate::config::HudiConfigs;
use crate::config::table::HudiTableConfig;
use crate::error::CoreError;
+use crate::file_group::base_file::reader::KeyPredicate;
use crate::file_group::log_file::content::Decoder;
use crate::file_group::log_file::log_block::{
BlockMetadataKey, BlockMetadataType, BlockType, DeferredContent, LogBlock,
LogBlockContent,
@@ -52,6 +53,9 @@ pub struct LogFileReader {
/// Predicate to push into parquet log blocks. Unset by default, so a
caller
/// that has not decided whether pushing is sound reads every row.
row_filter: Option<RowFilterBuilder>,
+ /// Keys to narrow HFile log blocks to. Unset by default, so every record
of
+ /// every block is returned.
+ key_predicate: Option<KeyPredicate>,
/// Schema Avro blocks are resolved up to. Unset by default, so a caller
that
/// has none reads each block at the schema it was written with.
reader_schema_json: Option<String>,
@@ -85,6 +89,7 @@ impl LogFileReader {
reader,
timezone,
row_filter: None,
+ key_predicate: None,
reader_schema_json: None,
})
}
@@ -108,6 +113,7 @@ impl LogFileReader {
reader,
timezone,
row_filter: None,
+ key_predicate: None,
reader_schema_json: None,
})
}
@@ -191,6 +197,15 @@ impl LogFileReader {
self
}
+ /// Narrow HFile log blocks to `key_predicate`.
+ ///
+ /// Only the HFile data blocks the keys can fall in are parsed, and only
the
+ /// admitted records come back; see [`Decoder::with_key_predicate`].
+ pub(crate) fn with_key_predicate(mut self, key_predicate:
Option<KeyPredicate>) -> Self {
+ self.key_predicate = key_predicate;
+ self
+ }
+
/// Resolve Avro blocks up to this schema. See
/// [`Decoder::with_reader_schema`].
pub fn with_reader_schema(mut self, reader_schema_json: Option<String>) ->
Self {
@@ -579,6 +594,7 @@ impl LogFileReader {
let decoder = Decoder::new(self.hudi_configs.clone())
.with_row_filter(self.row_filter.clone())
+ .with_key_predicate(self.key_predicate.clone())
.with_reader_schema(self.reader_schema_json.clone());
// Decode out of the block's own remaining bytes rather than straight
off
// the reader: the reader is async and the decoders are not. The span
diff --git a/crates/core/src/file_group/log_file/scanner.rs
b/crates/core/src/file_group/log_file/scanner.rs
index cdfb4592..567b6d9d 100644
--- a/crates/core/src/file_group/log_file/scanner.rs
+++ b/crates/core/src/file_group/log_file/scanner.rs
@@ -18,6 +18,7 @@
*/
use crate::Result;
use crate::config::HudiConfigs;
+use crate::file_group::base_file::reader::KeyPredicate;
use crate::file_group::log_file::log_block::{BlockType, LogBlock,
LogBlockContent};
use crate::file_group::log_file::reader::LogFileReader;
use crate::file_group::record_batches::RecordBatches;
@@ -128,6 +129,8 @@ impl CollectedBlocks {
pub struct LogFileScanner {
hudi_configs: Arc<HudiConfigs>,
storage: Arc<Storage>,
+ /// Keys to narrow HFile log blocks to; see
[`LogFileReader::with_key_predicate`].
+ key_predicate: Option<KeyPredicate>,
}
impl LogFileScanner {
@@ -135,9 +138,16 @@ impl LogFileScanner {
Self {
hudi_configs,
storage,
+ key_predicate: None,
}
}
+ /// Narrow HFile log blocks to `key_predicate` on every file this scanner
reads.
+ pub(crate) fn with_key_predicate(mut self, key_predicate:
Option<KeyPredicate>) -> Self {
+ self.key_predicate = key_predicate;
+ self
+ }
+
/// Read all blocks from multiple log files and collect rollback targets.
async fn collect_blocks(
&self,
@@ -149,7 +159,9 @@ impl LogFileScanner {
for path in relative_paths {
let mut reader =
- LogFileReader::new(self.hudi_configs.clone(),
self.storage.clone(), &path).await?;
+ LogFileReader::new(self.hudi_configs.clone(),
self.storage.clone(), &path)
+ .await?
+ .with_key_predicate(self.key_predicate.clone());
let blocks = reader.read_all_blocks(instant_range).await?;
// Collect rollback targets from command blocks
@@ -499,6 +511,66 @@ mod tests {
const SAN_FRANCISCO_UUID: &str = "036ded81-9ed4-479f-bcea-7145dfa0079b";
const SAO_PAULO_UUID: &str = "8aa68f7e-afd6-4c94-b86c-8a886552e08d";
+ /// The key predicate narrows HFile data blocks but must leave delete
+ /// blocks whole: a tombstone for a requested key may sit in a block that
+ /// also names keys nobody asked for, and dropping it would bring the
+ /// deleted key back.
+ #[tokio::test]
+ async fn test_scan_hfile_records_key_predicate_keeps_delete_blocks_whole()
-> crate::Result<()>
+ {
+ use crate::file_group::base_file::reader::KeyPredicate;
+ use crate::file_group::log_file::writer::write_log_block;
+ use crate::file_group::log_file::{BlockMetadataKey, BlockType};
+ use std::collections::{BTreeMap, HashMap};
+
+ let dir = tempfile::tempdir()?;
+ let instant = "20260101000000000";
+ let entries: Vec<(String, Vec<u8>)> = ["a", "b", "c"]
+ .iter()
+ .map(|k| (k.to_string(), k.as_bytes().to_vec()))
+ .collect();
+ let hfile = crate::hfile::HFileWriter::write(&entries, BTreeMap::new())
+ .map_err(|e| crate::error::CoreError::HFile(e.to_string()))?;
+ let header = HashMap::from([(BlockMetadataKey::InstantTime,
instant.to_string())]);
+ let mut bytes = write_log_block(BlockType::HfileData, header, &hfile);
+ bytes.extend(crate::write::build_delete_log_block(
+ instant,
+ &[
+ ("b".to_string(), String::new()),
+ ("d".to_string(), String::new()),
+ ],
+ 0,
+ )?);
+ let log = ".f-0_20260101000000000.log.1_0-0-0";
+ std::fs::write(dir.path().join(log), bytes)?;
+
+ let storage =
Storage::new_with_base_url(parse_uri(dir.path().to_str().unwrap())?)?;
+ let scanner = LogFileScanner::new(Arc::new(HudiConfigs::empty()),
storage)
+
.with_key_predicate(Some(KeyPredicate::Keys(vec!["b".to_string()])));
+ let records = scanner
+ .scan_hfile_records(
+ vec![log.to_string()],
+ &InstantRange::up_to("99991231235959999", "utc"),
+ )
+ .await?;
+ let mut keys: Vec<(String, bool)> = records
+ .iter()
+ .map(|r| (r.key_as_str().unwrap().to_string(),
r.value().is_empty()))
+ .collect();
+ keys.sort();
+ // `b` from the data block, then the tombstones for `b` and `d`; `a`
+ // and `c` were narrowed away.
+ assert_eq!(
+ keys,
+ vec![
+ ("b".to_string(), false),
+ ("b".to_string(), true),
+ ("d".to_string(), true)
+ ]
+ );
+ Ok(())
+ }
+
#[tokio::test]
async fn test_scan_metadata_table_log_files() -> crate::Result<()> {
use crate::metadata::table_record::MetadataRecordType;
diff --git a/crates/core/src/index/record.rs b/crates/core/src/index/record.rs
index fe7875e1..2ebece78 100644
--- a/crates/core/src/index/record.rs
+++ b/crates/core/src/index/record.rs
@@ -22,16 +22,24 @@ use std::collections::HashMap;
use std::sync::Arc;
use crate::Result;
+use crate::config::read::HudiReadConfig;
use crate::error::CoreError;
+use crate::file_group::base_file::reader::{KeyMatcher, KeyPredicate};
use crate::file_group::log_file::scanner::LogFileScanner;
use crate::hfile::{HFileReader, HFileRecord};
use crate::index::{HoodieIndex, HoodieKey, RecordLocation};
use crate::metadata::table::encode::decode_record_index_entry;
use crate::metadata::table::records::MetadataPartitionType;
use crate::storage::Storage;
+use crate::storage::reader::{HFILE_WHOLE_READ_WITH_KEYS_MAX_SIZE,
hfile_whole_read_max_size};
use crate::table::Table;
use crate::timeline::selector::InstantRange;
-use crate::write::metadata::epoch_millis_to_instant;
+use crate::util::concurrency::bounded_in_order;
+use crate::write::metadata::{
+ DEFAULT_RLI_NUM_FILE_GROUPS, RECORD_INDEX_FILE_ID_PREFIX,
epoch_millis_to_instant,
+ map_record_key_to_file_group_index, mdt_partition_latest_slices,
record_index_file_id,
+ shard_count_from_slices,
+};
/// Metadata-table record-level index (global: key = record key).
#[derive(Clone, Debug, Default)]
@@ -53,9 +61,10 @@ impl HoodieIndex for RecordIndex {
}
let storage = table.file_system_view.storage.clone();
- let merged = load_record_index_map(table, storage).await?;
+ let record_keys: Vec<&str> = keys.iter().map(|key|
key.record_key.as_str()).collect();
+ let found = lookup_record_index(table, storage, &record_keys).await?;
for key in keys {
- if let Some(entry) = merged.get(&key.record_key) {
+ if let Some(entry) = found.get(&key.record_key) {
locations.insert(
key.clone(),
Some(RecordLocation {
@@ -76,70 +85,145 @@ struct LoadedLocation {
instant_time_millis: i64,
}
-async fn load_record_index_map(
+/// Locations of `record_keys`, read from only the index shards and HFile
+/// blocks that can hold them.
+///
+/// Keys are hashed to their `record_index` file group the way the writer
+/// placed them, so a lookup opens the touched shards rather than the whole
+/// partition, and inside a shard the HFile block index narrows the read to the
+/// blocks the keys fall in. Values are Avro-decoded only for the requested
keys.
+async fn lookup_record_index(
table: &Table,
storage: Arc<Storage>,
+ record_keys: &[&str],
) -> Result<HashMap<String, LoadedLocation>> {
let partition = MetadataPartitionType::RecordIndex.partition_name();
// Slice-resolved: only each file group's LIVE slice (multi-base aware
- // after a Spark MDT compaction), logs in version order.
- let mut base_paths = Vec::new();
- let mut log_paths = Vec::new();
- for (_, base, logs) in
- crate::write::metadata::mdt_partition_latest_slices(storage.as_ref(),
partition).await?
- {
- base_paths.extend(base);
- log_paths.extend(logs);
- }
-
- let mut merged: HashMap<String, Option<LoadedLocation>> = HashMap::new();
- for base in base_paths {
- let mut reader = HFileReader::open(storage.as_ref(), &base)
- .await
- .map_err(|e| CoreError::HFile(format!("failed to open record_index
base: {e:?}")))?;
- let records = reader
- .collect_records()
- .map_err(|e| CoreError::HFile(format!("failed to read record_index
base: {e:?}")))?;
- apply_hfile_records(&mut merged, &records)?;
+ // after a Spark MDT compaction), logs in version order. One listing serves
+ // both the shard modulus and the routing, so the two cannot disagree.
+ let slices = mdt_partition_latest_slices(storage.as_ref(),
partition).await?;
+ let num_groups = shard_count_from_slices(
+ &slices,
+ RECORD_INDEX_FILE_ID_PREFIX,
+ DEFAULT_RLI_NUM_FILE_GROUPS,
+ )?;
+ let mut keys_by_shard: HashMap<String, Vec<String>> = HashMap::new();
+ for key in record_keys {
+ let shard = map_record_key_to_file_group_index(key, num_groups);
+ keys_by_shard
+ .entry(record_index_file_id(shard))
+ .or_default()
+ .push((*key).to_string());
}
- if !log_paths.is_empty() {
- // Only trust log blocks whose instant completed on the data timeline
- // (Java getValidInstantTimestamps) — orphan MDT commits are skipped.
- let valid = table.valid_metadata_instants().await?;
- let scanner = LogFileScanner::new(table.hudi_configs.clone(), storage);
- let range = InstantRange::exact_match(valid, "UTC");
- let records = scanner.scan_hfile_records(log_paths, &range).await?;
- apply_hfile_records(&mut merged, &records)?;
+ let touched: Vec<(KeyPredicate, Option<String>, Vec<String>)> = slices
+ .into_iter()
+ .filter_map(|(file_id, base, logs)| {
+ keys_by_shard
+ .remove(&file_id)
+ .map(|keys| (KeyPredicate::Keys(keys), base, logs))
+ })
+ .collect();
+ // Shard numbers are contiguous, but a listed id spelled other than
+ // `record-index-NNNN-0` would still leave its keys unread, and an absent
+ // key turns an upsert into a duplicate insert.
+ if let Some(missing) = keys_by_shard.keys().next() {
+ return Err(CoreError::MetadataTable(format!(
+ "record keys route to record index file group {missing}, which is
not in the record_index partition ({num_groups} file groups listed)"
+ )));
}
- Ok(merged
- .into_iter()
- .filter_map(|(key, location)| location.map(|loc| (key, loc)))
- .collect())
+ // Only trust log blocks whose instant completed on the data timeline
+ // (Java getValidInstantTimestamps): orphan MDT commits are skipped.
+ let range = if touched.iter().any(|(_, _, logs)| !logs.is_empty()) {
+ Some(InstantRange::exact_match(
+ table.valid_metadata_instants().await?,
+ "UTC",
+ ))
+ } else {
+ None
+ };
+ // A Spark-compacted shard can be large; read whole only below the
+ // threshold the base file reader uses for a keyed read.
+ let whole_below =
+
hfile_whole_read_max_size(&table.hudi_configs)?.min(HFILE_WHOLE_READ_WITH_KEYS_MAX_SIZE);
+
+ // Shards hold disjoint keys, so each is folded on its own and the maps
+ // are combined afterwards, with the same fan-out ceiling as a table read.
+ let concurrency: usize = table
+ .hudi_configs
+ .get_or_default(HudiReadConfig::FileSliceReadConcurrency)
+ .into();
+ let range = range.as_ref();
+ let per_shard = bounded_in_order(&touched, concurrency, |(predicate, base,
logs)| {
+ let storage = storage.clone();
+ async move {
+ let matcher = predicate.matcher();
+ let mut merged: HashMap<String, LoadedLocation> = HashMap::new();
+ if let Some(base) = base {
+ let reader = HFileReader::open_sized(storage.as_ref(), base,
whole_below, None)
+ .await
+ .map_err(|e| {
+ CoreError::HFile(format!("failed to open record_index
base: {e:?}"))
+ })?;
+ let blocks = reader.blocks_for_predicate(predicate);
+ // A scattered batch can select most blocks of a large shard,
so
+ // they are fetched a window at a time. A resident file is
under
+ // the keyed whole-read bound and needs no windows.
+ let windows = match reader.window_budget() {
+ Some(budget) => HFileReader::plan_windows(&blocks, budget),
+ None => vec![blocks],
+ };
+ for window in windows {
+ let records =
reader.read_records_batched(&window).await.map_err(|e| {
+ CoreError::HFile(format!("failed to read record_index
base: {e:?}"))
+ })?;
+ apply_hfile_records(&mut merged, &records, &matcher)?;
+ }
+ }
+ if let (false, Some(range)) = (logs.is_empty(), range) {
+ let scanner = LogFileScanner::new(table.hudi_configs.clone(),
storage)
+ .with_key_predicate(Some(predicate.clone()));
+ let records = scanner.scan_hfile_records(logs.clone(),
range).await?;
+ apply_hfile_records(&mut merged, &records, &matcher)?;
+ }
+ Ok::<_, CoreError>(merged)
+ }
+ })
+ .await?;
+ Ok(per_shard.into_iter().flatten().collect())
}
+/// Fold `records` into `merged`, later records overriding earlier ones and a
+/// tombstone removing the key.
+///
+/// A record `matcher` does not admit is dropped before its value is decoded: a
+/// selected base block holds every key in its range, and a delete block names
+/// every key its commit removed.
fn apply_hfile_records(
- merged: &mut HashMap<String, Option<LoadedLocation>>,
+ merged: &mut HashMap<String, LoadedLocation>,
records: &[HFileRecord],
+ matcher: &KeyMatcher<'_>,
) -> Result<()> {
for record in records {
let key = record
.key_as_str()
- .ok_or_else(|| CoreError::MetadataTable("record_index key must be
utf8".to_string()))?
- .to_string();
- match decode_record_index_entry(&key, record.value(),
record.avro_schema())? {
+ .ok_or_else(|| CoreError::MetadataTable("record_index key must be
utf8".to_string()))?;
+ if !matcher.admits(key) {
+ continue;
+ }
+ match decode_record_index_entry(key, record.value(),
record.avro_schema())? {
Some(entry) => {
merged.insert(
- key,
- Some(LoadedLocation {
+ key.to_string(),
+ LoadedLocation {
file_id: entry.file_id,
partition_path: entry.partition_path,
instant_time_millis: entry.instant_time_millis,
- }),
+ },
);
}
None => {
- merged.insert(key, None);
+ merged.remove(key);
}
}
}
@@ -185,3 +269,168 @@ pub(crate) async fn record_index_has_file_groups(table:
&Table) -> bool {
.map(|slices| !slices.is_empty())
.unwrap_or(false)
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::hfile::HFileReader;
+ use crate::metadata::table::encode::decode_record_index_entry;
+
+ /// A missing shard also shrinks the modulus, so the keys it misroutes are
+ /// mostly ones that live in intact shards, not ones in the missing shard.
+ #[tokio::test]
+ async fn test_lookup_errors_when_a_shard_is_missing() {
+ use arrow_array::{Int64Array, RecordBatch, StringArray};
+ use arrow_schema::{DataType, Field, Schema};
+
+ let dir = tempfile::tempdir().unwrap();
+ let mut table = Table::create(dir.path().to_str().unwrap())
+ .with_table_name("t")
+ .with_record_key_fields(["id"])
+ .create()
+ .await
+ .unwrap();
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("value", DataType::Int64, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(vec!["d"])),
+ Arc::new(Int64Array::from(vec![1])),
+ ],
+ )
+ .unwrap();
+ table.append([batch]).await.unwrap();
+ // `d` lives in shard 0 of 10, and mod 9 it would route to shard 1,
+ // which is intact.
+ assert_eq!(map_record_key_to_file_group_index("d", 10), 0);
+ assert_eq!(map_record_key_to_file_group_index("d", 9), 1);
+ let storage = table.file_system_view.storage.clone();
+ let found = lookup_record_index(&table, storage,
&["d"]).await.unwrap();
+ assert!(found.contains_key("d"));
+
+ let rli = dir.path().join(".hoodie/metadata/record_index");
+ for entry in std::fs::read_dir(&rli).unwrap() {
+ let path = entry.unwrap().path();
+ if path.to_string_lossy().contains("record-index-0005-") {
+ std::fs::remove_file(path).unwrap();
+ }
+ }
+ let table = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ let storage = table.file_system_view.storage.clone();
+ let err = lookup_record_index(&table, storage, &["d"])
+ .await
+ .err()
+ .unwrap()
+ .to_string();
+ assert!(err.contains("missing shards [5]"), "{err}");
+ }
+
+ #[tokio::test]
+ async fn test_lookup_reads_selected_blocks_of_multi_block_spark_base() {
+ let dir = tempfile::tempdir().unwrap();
+ Table::create(dir.path().to_str().unwrap())
+ .with_table_name("t")
+ .with_record_key_fields(["id"])
+ .create()
+ .await
+ .unwrap();
+ let mdt = dir.path().join(".hoodie/metadata");
+ let rli = mdt.join("record_index");
+ for entry in std::fs::read_dir(&rli).unwrap() {
+ let p = entry.unwrap().path();
+ let name = p.file_name().unwrap().to_string_lossy().to_string();
+ if name.starts_with("record-index-") &&
!name.starts_with("record-index-0000-") {
+ std::fs::remove_file(&p).unwrap();
+ }
+ }
+ let fixture = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join(
+
"../test/data/metadata_multi_block_hfile/record-index-0009-0_10-71-248_20260830075534042.hfile",
+ );
+ let bytes = std::fs::read(&fixture).unwrap();
+ let compaction_instant =
crate::write::append::generate_instant_time().await;
+ std::fs::write(
+ rli.join(format!(
+ "record-index-0000-0_0-0-0_{compaction_instant}.hfile"
+ )),
+ &bytes,
+ )
+ .unwrap();
+ let timeline = mdt.join(".hoodie/timeline");
+ let latest = std::fs::read_dir(&timeline)
+ .unwrap()
+ .map(|e| e.unwrap().path())
+ .filter(|p| p.to_string_lossy().ends_with(".deltacommit"))
+ .max()
+ .unwrap();
+ std::fs::copy(
+ latest,
+
timeline.join(format!("{compaction_instant}_{compaction_instant}.commit")),
+ )
+ .unwrap();
+ let table = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ let storage = table.file_system_view.storage.clone();
+
+ let mut reader = HFileReader::new(bytes.clone()).unwrap();
+ let entries = reader.data_block_entries();
+ let mut firsts = Vec::new();
+ let mut lasts = Vec::new();
+ for e in &entries {
+ let recs = reader.records_in_block(e).unwrap();
+
firsts.push(recs.first().unwrap().key_as_str().unwrap().to_string());
+ lasts.push(recs.last().unwrap().key_as_str().unwrap().to_string());
+ }
+ let all = reader.collect_records().unwrap();
+ let mut expected = HashMap::new();
+ for r in &all {
+ let k = r.key_as_str().unwrap().to_string();
+ let e = decode_record_index_entry(&k, r.value(), r.avro_schema())
+ .unwrap()
+ .unwrap();
+ expected.insert(k, (e.file_id, e.partition_path,
e.instant_time_millis));
+ }
+ // A Spark-compacted shard above the whole-read threshold, so the
+ // base is opened in ranges and only the selected blocks are fetched.
+ assert!(entries.len() > 1);
+ assert!(bytes.len() as u64 >
crate::storage::reader::HFILE_WHOLE_READ_WITH_KEYS_MAX_SIZE);
+ // Block boundary keys, one at a time (the exact-separator path).
+ for k in firsts.iter().chain(lasts.iter()) {
+ let n = lookup_record_index(&table, storage.clone(), &[k.as_str()])
+ .await
+ .unwrap();
+ assert_eq!(
+ n.get(k).map(|l| (
+ l.file_id.clone(),
+ l.partition_path.clone(),
+ l.instant_time_millis
+ )),
+ expected.get(k).cloned(),
+ "boundary {k}"
+ );
+ }
+ // Every key at once.
+ let keys: Vec<&str> = expected.keys().map(String::as_str).collect();
+ let n = lookup_record_index(&table, storage.clone(), &keys)
+ .await
+ .unwrap();
+ assert_eq!(n.len(), expected.len());
+ for (k, v) in &expected {
+ let got = n.get(k).map(|l| {
+ (
+ l.file_id.clone(),
+ l.partition_path.clone(),
+ l.instant_time_millis,
+ )
+ });
+ assert_eq!(got.as_ref(), Some(v), "key {k}");
+ }
+ // Keys just below the first key and between blocks must miss, not
error.
+ let below = format!("{}\u{0}", &firsts[0][..firsts[0].len() - 1]);
+ let n = lookup_record_index(&table, storage.clone(), &["",
below.as_str(), "\u{10FFFF}"])
+ .await
+ .unwrap();
+ assert!(n.is_empty());
+ }
+}
diff --git a/crates/core/src/write/metadata.rs
b/crates/core/src/write/metadata.rs
index 984e789b..00f4fb69 100644
--- a/crates/core/src/write/metadata.rs
+++ b/crates/core/src/write/metadata.rs
@@ -74,8 +74,11 @@ pub struct StatsFileUpdate {
pub ranges: Vec<ColumnRangeStats>,
}
-fn record_index_file_id(shard: usize) -> String {
- format!("record-index-{shard:04}-0")
+/// Prefix of a global record index file group id, `record-index-NNNN-0`.
+pub(crate) const RECORD_INDEX_FILE_ID_PREFIX: &str = "record-index-";
+
+pub(crate) fn record_index_file_id(shard: usize) -> String {
+ format!("{RECORD_INDEX_FILE_ID_PREFIX}{shard:04}-0")
}
/// Java `HoodieTableMetadataUtil.mapRecordKeyToFileGroupIndex` (31-hash,
signed i32).
@@ -468,7 +471,7 @@ async fn detect_rli_num_file_groups(storage: &Storage) ->
Result<usize> {
detect_num_file_groups(
storage,
MetadataPartitionType::RecordIndex.partition_name(),
- "record-index-",
+ RECORD_INDEX_FILE_ID_PREFIX,
DEFAULT_RLI_NUM_FILE_GROUPS,
)
.await
@@ -484,8 +487,24 @@ async fn detect_num_file_groups(
// change the shard modulus (it would diverge from what Java computes),
// and listing errors must propagate rather than silently changing it.
let slices = mdt_partition_latest_slices(storage, partition_name).await?;
+ shard_count_from_slices(&slices, file_id_prefix, default)
+}
+
+/// The shard modulus implied by a partition's committed slices: the number of
+/// distinct shard ids, or `default` when there are none.
+///
+/// Keys are routed by hashing modulo this count, so a gap in the shard ids is
+/// an error rather than a smaller count: with shard 5 of 10 missing, a count
of
+/// 9 would send most keys to intact shards that do not hold them. A missing
+/// highest shard cannot be told from a smaller partition, since nothing on
+/// disk records the count; Java routes by the listed count as well.
+pub(crate) fn shard_count_from_slices(
+ slices: &[(String, Option<String>, Vec<String>)],
+ file_id_prefix: &str,
+ default: usize,
+) -> Result<usize> {
let mut shards = std::collections::BTreeSet::new();
- for (file_id, _, _) in &slices {
+ for (file_id, _, _) in slices {
if let Some(rest) = file_id.strip_prefix(file_id_prefix)
&& let Some(shard_str) = rest.get(..4)
&& let Ok(shard) = shard_str.parse::<usize>()
@@ -493,11 +512,17 @@ async fn detect_num_file_groups(
shards.insert(shard);
}
}
- if shards.is_empty() {
- Ok(default)
- } else {
- Ok(shards.len())
+ let Some(&highest) = shards.last() else {
+ return Ok(default);
+ };
+ if highest + 1 != shards.len() {
+ let missing: Vec<usize> = (0..highest).filter(|s|
!shards.contains(s)).collect();
+ return Err(crate::error::CoreError::MetadataTable(format!(
+ "metadata table file groups {file_id_prefix}NNNN are missing
shards {missing:?} \
+ below the highest listed shard {highest}, so keys cannot be
routed to them"
+ )));
}
+ Ok(shards.len())
}
/// Refuse to write a table whose metadata table has a layout this writer does
@@ -566,7 +591,7 @@ pub(crate) async fn ensure_supported_metadata_layout(table:
&Table) -> Result<()
/// partitioned ones `record-index-<escapedPartition>-NNNN-0`.
fn is_global_record_index_file_id(file_id: &str) -> bool {
file_id
- .strip_prefix("record-index-")
+ .strip_prefix(RECORD_INDEX_FILE_ID_PREFIX)
.and_then(|rest| rest.split_once('-'))
.is_some_and(|(index, suffix)| {
[index, suffix]
diff --git a/crates/core/tests/table_write_lifecycle_tests.rs
b/crates/core/tests/table_write_lifecycle_tests.rs
index 31e747ba..7bd36d31 100644
--- a/crates/core/tests/table_write_lifecycle_tests.rs
+++ b/crates/core/tests/table_write_lifecycle_tests.rs
@@ -1036,3 +1036,66 @@ fn walkdir_count_parquet(root: &std::path::Path) ->
usize {
}
n
}
+
+/// Keys spread over every record index shard, so a lookup has to route each
+/// key to its shard and honour later log blocks and delete blocks over
+/// earlier ones.
+#[tokio::test]
+async fn test_record_index_lookup_across_shards_blocks_and_tombstones() {
+ let dir = tempdir().unwrap();
+ let mut table = Table::create(dir.path().to_str().unwrap())
+ .with_table_name("trips")
+ .with_record_key_fields(["id"])
+ .create()
+ .await
+ .unwrap();
+ let total = 20_000usize;
+ let key = |i: usize| format!("{:08x}-{i}", i.wrapping_mul(2_654_435_761));
+ let keys: Vec<String> = (0..total).map(key).collect();
+ table
+ .append([batch(
+ keys.iter().map(|k| (k.as_str(), 1)).collect::<Vec<_>>(),
+ )])
+ .await
+ .unwrap();
+
+ // Every 7th key is updated, every 11th deleted; a key that is both is
+ // deleted last, so the index must answer "absent" for it afterwards.
+ let updated: Vec<&str> =
keys.iter().step_by(7).map(String::as_str).collect();
+ let result = table
+ .upsert([batch(updated.iter().map(|k| (*k, 2)).collect())])
+ .await
+ .unwrap();
+ assert_eq!((result.num_updates, result.num_inserts), (updated.len(), 0));
+
+ let deleted: Vec<&str> =
keys.iter().step_by(11).map(String::as_str).collect();
+ let result = table
+ .delete_keys(deleted.iter().map(|k| HoodieKey {
+ record_key: (*k).to_string(),
+ partition_path: String::new(),
+ }))
+ .await
+ .unwrap();
+ assert_eq!(result.num_deletes, deleted.len());
+
+ // Re-upserting a mix of live, deleted and unknown keys exercises the
+ // lookup against both the log blocks and the delete blocks.
+ let probe: Vec<(&str, i64)> = vec![
+ (updated[1], 3), // updated, still live
+ (deleted[1], 3), // deleted: must come back as an insert
+ (keys[77].as_str(), 3), // updated, then deleted: also an insert
+ (keys[2].as_str(), 3),
+ ("never-written", 3),
+ ];
+ let result = table.upsert([batch(probe)]).await.unwrap();
+ assert_eq!((result.num_updates, result.num_inserts), (2, 3));
+
+ let rows = rows_by_id(&table.read(&ReadOptions::new()).await.unwrap());
+ assert_eq!(rows.len(), total - deleted.len() + 3);
+ let value = |k: &str| rows.iter().find(|(id, _)| id == k).map(|(_, v)| *v);
+ assert_eq!(value(updated[1]), Some(3));
+ assert_eq!(value(deleted[1]), Some(3));
+ assert_eq!(value(deleted[2]), None);
+ assert_eq!(value(updated[2]), Some(2));
+ assert_eq!(value("never-written"), Some(3));
+}
