You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "yihua (via GitHub)" <gi...@apache.org> on 2026/10/29 07:53:40 UTC
[PR] perf(core): read only the requested keys' record index shards and blocks [hudi-rs]
yihua opened a new pull request, #869:
URL: https://github.com/apache/hudi-rs/pull/869
## Description
closes #865
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 with the table's file slice read concurrency, and a key routed to a shard the listing does not have is an error instead of a silent miss that would turn an upsert into a duplicate insert.
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).
## How are the changes test-covered
- [ ] N/A
- [x] Automated tests (unit and/or integration tests)
- [ ] Manual tests
- [ ] Details are described below
New tests cover a lookup against a Spark-written nine-block record index HFile installed as a committed compaction base (every block's first and last key, all keys at once, and misses between blocks), the key predicate narrowing HFile log blocks while leaving delete blocks whole, the missing-shard error, and a 20,000-key lifecycle across all shards with updates, deletes, and re-inserts of deleted keys. The existing write suites pass unchanged.
https://claude.ai/code/session_01PpvixgKyY2VqVo8UAdtUeF
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(core): read only the requested keys' record index shards and blocks [hudi-rs]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua merged PR #869:
URL: https://github.com/apache/hudi-rs/pull/869
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(core): read only the requested keys' record index shards and blocks [hudi-rs]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #869:
URL: https://github.com/apache/hudi-rs/pull/869#issuecomment-6492791344
## [Codecov](https://app.codecov.io/gh/apache/hudi-rs/pull/869?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `97.70115%` with `2 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 90.23%. Comparing base ([`4dd6542`](https://app.codecov.io/gh/apache/hudi-rs/commit/4dd654223284f6529b3e0a2fb3523564d0d87af7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`bd5a791`](https://app.codecov.io/gh/apache/hudi-rs/commit/bd5a791c494e910307989e5a7c49fc8ae89a5022?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi-rs/pull/869?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [crates/core/src/index/record.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/869?src=pr&el=tree&filepath=crates%2Fcore%2Fsrc%2Findex%2Frecord.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2NvcmUvc3JjL2luZGV4L3JlY29yZC5ycw==) | 96.96% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/869?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #869 +/- ##
==========================================
+ Coverage 90.21% 90.23% +0.02%
==========================================
Files 147 147
Lines 20411 20466 +55
==========================================
+ Hits 18413 18467 +54
- Misses 1998 1999 +1
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/hudi-rs/pull/869?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(core): read only the requested keys' record index shards and blocks [hudi-rs]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #869:
URL: https://github.com/apache/hudi-rs/pull/869#discussion_r4420103469
##########
crates/core/src/index/record.rs:
##########
@@ -76,40 +85,112 @@ 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);
+ // 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);
Review Comment:
Suggestion (non-blocking): consider documenting or strengthening the missing-shard guard's limits. Correction to my original review: the P1 classification was too strong. The example below requires an already incomplete/corrupt index; I have not established a supported table lifecycle that produces it. Initialization creates all shards, and Java also derives its routing modulus from the file-slice count.
The narrower observation still holds: with 10 original shards, existing key `"d"` (hash 100) belongs to shard 0. Removing every file of shard 5 makes the inferred count 9, so lookup routes `"d"` to existing shard 1 and the guard does not fire. The old full scan could still find this key in surviving shard 0. I confirmed the routing with the extracted Rust functions, not an end-to-end write test.
If detecting missing-shard corruption is an intended guarantee of this PR, consider validating gaps in the shard layout before routing and add a test using an existing key in an intact shard. Otherwise, clarify that this guard only detects a requested shard ID absent from the listing; it does not validate overall index completeness. I would not block the PR on this without evidence that the incomplete state can arise during supported operation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(core): read only the requested keys' record index shards and blocks [hudi-rs]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on PR #869:
URL: https://github.com/apache/hudi-rs/pull/869#issuecomment-6515295155
Non-blocking follow-ups are tracked in #917.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(core): read only the requested keys' record index shards and blocks [hudi-rs]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #869:
URL: https://github.com/apache/hudi-rs/pull/869#discussion_r4420343045
##########
crates/core/src/index/record.rs:
##########
@@ -76,40 +85,112 @@ 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);
+ // 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);
Review Comment:
Fixed in 3ef07976: `shard_count_from_slices` now rejects non-contiguous shard ids, so the lookup and the writer both refuse before routing, and the test now looks up your `d` (shard 0 of 10) after dropping shard 5. Nothing on disk records the count, so a missing highest shard stays undetectable, which matches Java routing by the listed count.
##########
crates/core/src/index/record.rs:
##########
@@ -76,40 +85,112 @@ 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);
+ // 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());
+ }
+ 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();
+ // A key that hashed to a shard the listing does not have would read as
+ // absent and turn 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)"
+ )));
}
+ // 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 read on its own and the results
+ // are merged 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 mut found: Vec<HFileRecord> = Vec::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);
+ let matcher = predicate.matcher();
+ // A selected block holds every key in its range; keep only the
+ // requested ones so each shard holds no more than it returns.
+ found.extend(
+ reader
+ .read_records_batched(&blocks)
Review Comment:
Done in 3ef07976: selected blocks are now fetched in `plan_windows` windows under the reader's budget and folded per window.
##########
crates/core/src/index/record.rs:
##########
@@ -76,40 +85,112 @@ 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);
+ // 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());
+ }
+ 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();
+ // A key that hashed to a shard the listing does not have would read as
+ // absent and turn 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)"
+ )));
}
+ // 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 read on its own and the results
+ // are merged 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 mut found: Vec<HFileRecord> = Vec::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);
+ let matcher = predicate.matcher();
+ // A selected block holds every key in its range; keep only the
+ // requested ones so each shard holds no more than it returns.
+ found.extend(
+ reader
+ .read_records_batched(&blocks)
+ .await
+ .map_err(|e| {
+ CoreError::HFile(format!("failed to read record_index base: {e:?}"))
+ })?
+ .into_iter()
+ .filter(|r| r.key_as_str().is_some_and(|k| matcher.admits(k))),
+ );
+ }
+ if let (false, Some(range)) = (logs.is_empty(), range) {
+ let scanner = LogFileScanner::new(table.hudi_configs.clone(), storage)
+ .with_key_predicate(Some(predicate.clone()));
+ found.extend(scanner.scan_hfile_records(logs.clone(), range).await?);
+ }
+ Ok::<_, CoreError>(found)
Review Comment:
Done in 3ef07976: each shard task folds into its own location map, and the maps are combined at the end.
##########
crates/core/src/index/record.rs:
##########
@@ -139,7 +228,7 @@ fn apply_hfile_records(
);
}
None => {
- merged.insert(key, None);
+ merged.insert(key.to_string(), None);
Review Comment:
Done in 3ef07976: tombstones now `remove` from a `HashMap<String, LoadedLocation>`, and the `Option` and final `filter_map` are gone.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(core): read only the requested keys' record index shards and blocks [hudi-rs]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #869:
URL: https://github.com/apache/hudi-rs/pull/869#discussion_r4420103469
##########
crates/core/src/index/record.rs:
##########
@@ -76,40 +85,112 @@ 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);
+ // 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);
Review Comment:
[P1] Validate shard completeness before using the surviving count as the hash modulus. With 10 shards, existing key `"d"` (hash 100) is stored in shard 0. If shard 5 disappears, `shard_count_from_slices` returns 9 and this lookup routes `"d"` to shard 1. Shard 1 exists, so the missing-shard guard below passes and the key silently appears absent, potentially turning an upsert into a duplicate insert or a delete into a no-op. I reproduced the routing with these extracted Rust functions. The new missing-shard test only chooses a key that hashes directly to the missing shard under modulus 9. Please also test an existing key in an intact shard, and reject incomplete layouts before routing (using an authoritative expected count where available).
##########
crates/core/src/index/record.rs:
##########
@@ -76,40 +85,112 @@ 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);
+ // 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());
+ }
+ 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();
+ // A key that hashed to a shard the listing does not have would read as
+ // absent and turn 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)"
+ )));
}
+ // 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 read on its own and the results
+ // are merged 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 mut found: Vec<HFileRecord> = Vec::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);
+ let matcher = predicate.matcher();
+ // A selected block holds every key in its range; keep only the
+ // requested ones so each shard holds no more than it returns.
+ found.extend(
+ reader
+ .read_records_batched(&blocks)
Review Comment:
[P2] Bound the selected-block read with windows. `read_records_batched(&blocks)` fetches/decompresses every selected block and materializes all their records before the filter below runs. A scattered batch can select most blocks of a large compacted shard, and concurrent shard reads multiply that peak memory. `whole_below` only controls opening the file; it does not bound this allocation. Could this use `reader.window_budget()` and `HFileReader::plan_windows`, filtering each window before fetching the next, as the existing HFile base reader does?
##########
crates/core/src/index/record.rs:
##########
@@ -76,40 +85,112 @@ 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);
+ // 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());
+ }
+ 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();
+ // A key that hashed to a shard the listing does not have would read as
+ // absent and turn 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)"
+ )));
}
+ // 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 read on its own and the results
+ // are merged 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 mut found: Vec<HFileRecord> = Vec::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);
+ let matcher = predicate.matcher();
+ // A selected block holds every key in its range; keep only the
+ // requested ones so each shard holds no more than it returns.
+ found.extend(
+ reader
+ .read_records_batched(&blocks)
+ .await
+ .map_err(|e| {
+ CoreError::HFile(format!("failed to read record_index base: {e:?}"))
+ })?
+ .into_iter()
+ .filter(|r| r.key_as_str().is_some_and(|k| matcher.admits(k))),
+ );
+ }
+ if let (false, Some(range)) = (logs.is_empty(), range) {
+ let scanner = LogFileScanner::new(table.hudi_configs.clone(), storage)
+ .with_key_predicate(Some(predicate.clone()));
+ found.extend(scanner.scan_hfile_records(logs.clone(), range).await?);
+ }
+ Ok::<_, CoreError>(found)
Review Comment:
Suggestion (non-blocking): fold the base/log records into a location map inside each shard task and return that map. `bounded_in_order` currently retains all completed shards' raw record vectors until every shard finishes, including obsolete versions and unrelated delete tombstones. Since shards have disjoint keys, their final maps can simply be combined afterward. This reduces retained data and makes the per-shard merge explicit.
##########
crates/core/src/index/record.rs:
##########
@@ -139,7 +228,7 @@ fn apply_hfile_records(
);
}
None => {
- merged.insert(key, None);
+ merged.insert(key.to_string(), None);
Review Comment:
Suggestion (non-blocking): this tombstone branch can use `merged.remove(key)` with a `HashMap<String, LoadedLocation>`. Records are applied in order, so removal preserves delete semantics and a later put still reinserts the key. That also eliminates the `Option` map value and the final `filter_map`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org