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