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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
