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 4c259bdc feat(core): read only the file groups the record index names
for key filters (#907)
4c259bdc is described below
commit 4c259bdc551ce92629e0952b775730d3eda2a09c
Author: Y Ethan Guo <[email protected]>
AuthorDate: Fri Oct 30 01:42:55 2026 -0700
feat(core): read only the file groups the record index names for key
filters (#907)
`Table::get_file_slices_inner` now asks the record-level index which file
groups hold the keys an `=` or `IN` filter on the record key field (or
`_hoodie_record_key`) names, and plans only those file slices; the row-level
mask still applies the filter to what is read. The lookup runs for a snapshot
read at the latest commit, on a table with a single string record key field and
a populated index; a time-travel read, a read-optimized read of a merge-on-read
table, a composite key or an [...]
The index is read as of the reader's snapshot rather than its latest state:
log blocks count only for instants at or before the snapshot's latest commit
that had completed when the table was opened, and a record index base file
compacted after the snapshot falls back to the full scan. One listing of the
record index partition serves both that check and the lookup. Metadata-table
instants newer than the snapshot that do not touch the index (a clean, an
indexing action, a metadata-table [...]
Reading one key out of a 10M-row table went from a scan of every file slice
to one file slice.
---
crates/core/src/index/mod.rs | 1 +
crates/core/src/index/record.rs | 73 +++-
crates/core/src/table/mod.rs | 130 +++++-
crates/core/src/timeline/mod.rs | 26 +-
crates/core/tests/table_read_record_index_tests.rs | 478 +++++++++++++++++++++
5 files changed, 685 insertions(+), 23 deletions(-)
diff --git a/crates/core/src/index/mod.rs b/crates/core/src/index/mod.rs
index e6cd0bee..806e8321 100644
--- a/crates/core/src/index/mod.rs
+++ b/crates/core/src/index/mod.rs
@@ -21,6 +21,7 @@
mod record;
mod simple;
+pub(crate) use record::locate_file_groups;
pub use record::{RecordIndex, is_record_index_enabled};
pub use simple::SimpleIndex;
diff --git a/crates/core/src/index/record.rs b/crates/core/src/index/record.rs
index 808141d1..c11a561f 100644
--- a/crates/core/src/index/record.rs
+++ b/crates/core/src/index/record.rs
@@ -18,12 +18,13 @@
*/
//! Record-level index backed by the metadata table `record_index` partition.
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use crate::Result;
use crate::config::read::HudiReadConfig;
use crate::error::CoreError;
+use crate::file_group::base_file::BaseFile;
use crate::file_group::base_file::reader::{KeyMatcher, KeyPredicate};
use crate::file_group::log_file::scanner::LogFileScanner;
use crate::hfile::{HFileReader, HFileRecord};
@@ -85,23 +86,72 @@ struct LoadedLocation {
instant_time_millis: i64,
}
+/// Locations of `record_keys` in the index's latest state.
+async fn lookup_record_index(
+ table: &Table,
+ storage: Arc<Storage>,
+ record_keys: &[&str],
+) -> Result<HashMap<String, LoadedLocation>> {
+ Ok(lookup_as_of(table, storage, record_keys, None)
+ .await?
+ .unwrap_or_default())
+}
+
+/// The file groups holding `record_keys` in the snapshot whose latest commit
+/// is `as_of`, or `None` when the index cannot answer for that snapshot.
+pub(crate) async fn locate_file_groups(
+ table: &Table,
+ record_keys: &[&str],
+ as_of: &str,
+) -> Result<Option<HashSet<(String, String)>>> {
+ let storage = table.file_system_view.storage.clone();
+ Ok(lookup_as_of(table, storage, record_keys, Some(as_of))
+ .await?
+ .map(|found| {
+ found
+ .into_values()
+ .map(|location| (location.partition_path, location.file_id))
+ .collect()
+ }))
+}
+
/// Locations of `record_keys`, read from only the index shards and HFile
-/// blocks that can hold them.
+/// blocks that can hold them; with `as_of`, as of the snapshot whose latest
+/// commit that is, and `None` when the index cannot describe it.
///
/// 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(
+async fn lookup_as_of(
table: &Table,
storage: Arc<Storage>,
record_keys: &[&str],
-) -> Result<HashMap<String, LoadedLocation>> {
+ as_of: Option<&str>,
+) -> Result<Option<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. 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?;
+ if let Some(as_of) = as_of {
+ // An empty partition answers every key with "absent", and a base file
+ // compacted after the snapshot may hold later commits' changes, which
+ // no instant filter can take back out.
+ let compacted_after = |base: &str| {
+ base.rsplit('/')
+ .next()
+ .and_then(|name| name.parse::<BaseFile>().ok())
+ .is_none_or(|base| base.commit_timestamp.as_str() > as_of)
+ };
+ if slices.is_empty()
+ || slices
+ .iter()
+ .any(|(_, base, _)|
base.as_deref().is_some_and(compacted_after))
+ {
+ return Ok(None);
+ }
+ }
let num_groups = shard_count_from_slices(
&slices,
RECORD_INDEX_FILE_ID_PREFIX,
@@ -135,10 +185,15 @@ async fn lookup_record_index(
// 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",
- ))
+ let mut valid = table.valid_metadata_instants().await?;
+ if let Some(as_of) = as_of {
+ // The snapshot holds the commits that had completed by `as_of`
when
+ // the table was opened; one pending then may have completed since.
+ valid.retain(|instant| {
+ instant.as_str() <= as_of &&
!table.timeline.pending_instants.contains(instant)
+ });
+ }
+ Some(InstantRange::exact_match(valid, "UTC"))
} else {
None
};
@@ -190,7 +245,7 @@ async fn lookup_record_index(
}
})
.await?;
- Ok(per_shard.into_iter().flatten().collect())
+ Ok(Some(per_shard.into_iter().flatten().collect()))
}
/// Fold `records` into `merged`, later records overriding earlier ones and a
diff --git a/crates/core/src/table/mod.rs b/crates/core/src/table/mod.rs
index 57d20cae..dd9f19a9 100644
--- a/crates/core/src/table/mod.rs
+++ b/crates/core/src/table/mod.rs
@@ -129,7 +129,7 @@ use crate::table::partition::{PartitionPruner,
project_partition_schema};
use crate::timeline::util::format_timestamp;
use crate::timeline::{EARLIEST_START_TIMESTAMP, Timeline};
use arrow::record_batch::RecordBatch;
-use arrow_schema::{Field, Schema};
+use arrow_schema::{DataType, Field, Schema};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use tokio::sync::OnceCell;
@@ -721,6 +721,30 @@ impl Table {
)
.await?;
+ // The index describes merged rows, which a read-optimized read of a
+ // merge-on-read table does not see.
+ let read_optimized = base_file_only && self.is_mor();
+ if !read_optimized
+ && let Some(located) = self
+ .file_groups_for_record_keys(filters, timestamp, &table_schema)
+ .await
+ {
+ let planned: HashSet<(&str, &str)> = file_slices
+ .iter()
+ .map(|fs| (fs.partition_path.as_str(), fs.file_id()))
+ .collect();
+ // A file group the index names that this plan lacks (another
+ // filter pruned it, or the index disagrees with the snapshot)
+ // leaves the plan as it is.
+ if located.iter().all(|(partition, file_id)| {
+ planned.contains(&(partition.as_str(), file_id.as_str()))
+ }) {
+ file_slices.retain(|fs| {
+ located.contains(&(fs.partition_path.clone(),
fs.file_id().to_string()))
+ });
+ }
+ }
+
if base_file_only {
for fs in &mut file_slices {
fs.log_files.clear();
@@ -729,6 +753,110 @@ impl Table {
Ok(file_slices)
}
+ /// The file groups the record-level index names for the record keys
+ /// `filters` ask for, or `None` when the index cannot answer.
+ ///
+ /// Only an equality or `IN` filter on the table's single string record key
+ /// field (or on `_hoodie_record_key`) is a key lookup, and only for a read
+ /// of the latest commit on table version 8 or later, whose metadata table
+ /// this reader can list. Anything the lookup cannot handle, including an
+ /// error, leaves the full scan in place; the row-level mask still applies
+ /// the filter to whatever is read.
+ async fn file_groups_for_record_keys(
+ &self,
+ filters: &[Filter],
+ timestamp: &str,
+ table_schema: &Schema,
+ ) -> Option<HashSet<(String, String)>> {
+ use crate::expr::ExprOperator;
+ use crate::index::is_record_index_enabled;
+ use crate::keygen::{KeyGeneratorKind, key_generator_kind};
+
+ let metadata_off = self
+ .hudi_configs
+ .get_raw(HudiTableConfig::MetadataTableEnabled.as_ref())
+ .is_some_and(|v| v.trim().eq_ignore_ascii_case("false"));
+ if filters.is_empty()
+ || metadata_off
+ || !self.is_metadata_table_enabled()
+ || !is_record_index_enabled(self)
+ || self.timeline.get_latest_commit_timestamp_as_option() !=
Some(timestamp)
+ {
+ return None;
+ }
+ let version: isize = self
+ .hudi_configs
+ .try_get(HudiTableConfig::TableVersion)
+ .ok()
+ .flatten()
+ .map(Into::into)?;
+ if version < 8 {
+ return None;
+ }
+ let key_fields: Vec<String> = self
+ .hudi_configs
+ .try_get(HudiTableConfig::RecordKeyFields)
+ .ok()
+ .flatten()
+ .map(Into::into)?;
+ let [key_field] = key_fields.as_slice() else {
+ return None;
+ };
+ // The index stores the key as a string; only a string column's filter
+ // value is that same text.
+ let key_is_string = matches!(
+ table_schema.field_with_name(key_field).ok()?.data_type(),
+ DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View
+ );
+ // A complex key generator stores a single-field key as `field:value`
+ // from table version 9; below it the stored form depends on the writer
+ // release, so the index cannot be asked.
+ let prefixed = match key_generator_kind(&self.hudi_configs).ok()? {
+ KeyGeneratorKind::Simple | KeyGeneratorKind::NonPartitioned =>
false,
+ KeyGeneratorKind::Complex | KeyGeneratorKind::SparkSql if version
>= 9 => true,
+ _ => return None,
+ };
+ let mut keys: Option<HashSet<String>> = None;
+ for filter in filters {
+ if !matches!(filter.operator, ExprOperator::Eq | ExprOperator::In)
{
+ continue;
+ }
+ let values: HashSet<String> = if filter.field == *key_field {
+ if !key_is_string {
+ return None;
+ }
+ filter
+ .values
+ .iter()
+ .map(|v| {
+ if prefixed {
+ format!("{key_field}:{v}")
+ } else {
+ v.clone()
+ }
+ })
+ .collect()
+ } else if filter.field == MetaField::RecordKey.as_ref() {
+ filter.values.iter().cloned().collect()
+ } else {
+ continue;
+ };
+ keys = Some(match keys {
+ Some(prior) => prior.intersection(&values).cloned().collect(),
+ None => values,
+ });
+ }
+ let keys = keys?;
+ let keys: Vec<&str> = keys.iter().map(String::as_str).collect();
+ match crate::index::locate_file_groups(self, &keys, timestamp).await {
+ Ok(located) => located,
+ Err(e) => {
+ log::warn!("record index lookup failed, reading every file
slice: {e}");
+ None
+ }
+ }
+ }
+
async fn get_file_slices_between_inner(
&self,
start_timestamp: &str,
diff --git a/crates/core/src/timeline/mod.rs b/crates/core/src/timeline/mod.rs
index 5c709230..70d8613b 100644
--- a/crates/core/src/timeline/mod.rs
+++ b/crates/core/src/timeline/mod.rs
@@ -135,7 +135,14 @@ impl Timeline {
None,
)?;
let all_active = timeline.load_instants(&selector, false).await?;
- timeline.earliest_active_instant = all_active
+ timeline.set_active_instants(all_active);
+ Ok(timeline)
+ }
+
+ /// Set the completed commits, the pending instants and the archival
+ /// boundary from one listing of the active timeline in every state.
+ fn set_active_instants(&mut self, all_active: Vec<Instant>) {
+ self.earliest_active_instant = all_active
.iter()
.map(|instant| instant.timestamp.clone())
.min();
@@ -151,13 +158,12 @@ impl Timeline {
.iter()
.map(|instant| instant.timestamp.clone())
.collect();
- timeline.pending_instants = pending
+ self.pending_instants = pending
.into_iter()
.map(|instant| instant.timestamp)
.filter(|timestamp| !completed_times.contains(timestamp))
.collect();
- timeline.completed_commits = completed;
- Ok(timeline)
+ self.completed_commits = completed;
}
/// The inputs the log-block scan needs to tell a committed instant from
one
@@ -196,7 +202,8 @@ impl Timeline {
/// Reload completed commits from storage after a write.
pub(crate) async fn reload_completed_commits(&mut self) -> Result<()> {
// Same listing as `new_from_storage`: every state, so the archival
- // boundary can be refreshed together with the completed commits.
+ // boundary and the pending instants are refreshed together with the
+ // completed commits.
// Archival (running inside writes) raises the boundary; a reload that
// kept the old one would leave newly-archived instants neither in the
// completed set nor below the boundary — read as uncommitted, silently
@@ -209,14 +216,7 @@ impl Timeline {
None,
)?;
let all_active = self.load_instants(&selector, false).await?;
- self.earliest_active_instant = all_active
- .iter()
- .map(|instant| instant.timestamp.clone())
- .min();
- self.completed_commits = all_active
- .into_iter()
- .filter(|instant| instant.state == State::Completed)
- .collect();
+ self.set_active_instants(all_active);
Ok(())
}
diff --git a/crates/core/tests/table_read_record_index_tests.rs
b/crates/core/tests/table_read_record_index_tests.rs
new file mode 100644
index 00000000..8863814f
--- /dev/null
+++ b/crates/core/tests/table_read_record_index_tests.rs
@@ -0,0 +1,478 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+//! Record-level index lookups on the read path: a key filter reads only the
+//! file groups the index names.
+
+use std::sync::Arc;
+
+use arrow_array::{Int64Array, RecordBatch, StringArray};
+use arrow_schema::{DataType, Field, Schema};
+use hudi_core::config::table::TableTypeValue;
+use hudi_core::index::HoodieKey;
+use hudi_core::table::{ReadOptions, Table};
+use tempfile::tempdir;
+
+/// Ten keys `k{group + 3i}` with values `base + i`: the three groups' key
+/// ranges overlap, so footer min/max statistics cannot tell them apart and
+/// only the index can.
+fn batch(group: usize, base: i64) -> RecordBatch {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Utf8, false),
+ Field::new("value", DataType::Int64, false),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(
+ (0..10)
+ .map(|i| format!("k{:02}", group + 3 * i))
+ .collect::<Vec<_>>(),
+ )),
+ Arc::new(Int64Array::from_iter_values((0..10).map(|i| base + i))),
+ ],
+ )
+ .unwrap()
+}
+
+/// Three file groups: group 0 holds `k00, k03, ...`, group 1 `k01, k04, ...`,
+/// group 2 `k02, k05, ...`.
+async fn three_groups(
+ dir: &std::path::Path,
+ table_type: TableTypeValue,
+ record_index: bool,
+) -> Table {
+ let mut table = Table::create(dir.to_str().unwrap())
+ .with_table_name("t")
+ .with_record_key_fields(["id"])
+ .with_table_type(table_type)
+ .with_record_index(record_index)
+ // One file group per append.
+ .with_option("hoodie.parquet.small.file.limit", "0")
+ .create()
+ .await
+ .unwrap();
+ for (group, base) in [(0, 0), (1, 100), (2, 200)] {
+ table.append([batch(group, base)]).await.unwrap();
+ }
+ Table::new(dir.to_str().unwrap()).await.unwrap()
+}
+
+fn filters(filters: &[(&str, &str, &str)]) -> ReadOptions {
+ ReadOptions::new()
+ .with_filters(filters.iter().copied())
+ .unwrap()
+}
+
+/// How many file slices the read plans, and the ids it returns.
+async fn slices_and_ids(table: &Table, options: &ReadOptions) -> (usize,
Vec<String>) {
+ let slices = table.get_file_slices(options).await.unwrap().len();
+ let mut ids: Vec<String> = table
+ .read(options)
+ .await
+ .unwrap()
+ .iter()
+ .flat_map(|b| {
+ b.column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .iter()
+ .map(|v| v.unwrap().to_string())
+ .collect::<Vec<_>>()
+ })
+ .collect();
+ ids.sort();
+ (slices, ids)
+}
+
+#[tokio::test]
+async fn test_key_filter_reads_only_the_indexed_file_group() {
+ let dir = tempdir().unwrap();
+ let table = three_groups(dir.path(), TableTypeValue::CopyOnWrite,
true).await;
+
+ assert_eq!(
+ table
+ .get_file_slices(&ReadOptions::new())
+ .await
+ .unwrap()
+ .len(),
+ 3
+ );
+ assert_eq!(
+ slices_and_ids(&table, &filters(&[("id", "=", "k04")])).await,
+ (1, vec!["k04".to_string()])
+ );
+ assert_eq!(
+ slices_and_ids(&table, &filters(&[("_hoodie_record_key", "IN",
"k03,k05")])).await,
+ (2, vec!["k03".to_string(), "k05".to_string()])
+ );
+ // Two key filters intersect; a filter on another column changes nothing.
+ assert_eq!(
+ slices_and_ids(
+ &table,
+ &filters(&[
+ ("id", "IN", "k03,k04"),
+ ("id", "=", "k04"),
+ ("value", ">", "0")
+ ])
+ )
+ .await,
+ (1, vec!["k04".to_string()])
+ );
+ // A key the index does not know reads nothing.
+ assert_eq!(
+ slices_and_ids(&table, &filters(&[("id", "=", "zz")])).await,
+ (0, vec![])
+ );
+}
+
+#[tokio::test]
+async fn test_key_filter_does_not_use_the_index_for_an_earlier_snapshot() {
+ let dir = tempdir().unwrap();
+ let table = three_groups(dir.path(), TableTypeValue::CopyOnWrite,
true).await;
+ let mut instants: Vec<String> = table
+ .get_timeline()
+ .completed_commits
+ .iter()
+ .map(|i| i.timestamp.clone())
+ .collect();
+ instants.sort();
+ let options = filters(&[("id", "=",
"k04")]).with_as_of_timestamp(&instants[1]);
+ // Both file groups of that snapshot are read; the row mask still applies.
+ assert_eq!(
+ slices_and_ids(&table, &options).await,
+ (2, vec!["k04".to_string()])
+ );
+}
+
+#[tokio::test]
+async fn test_key_filter_without_the_index_reads_every_file_group() {
+ let dir = tempdir().unwrap();
+ let table = three_groups(dir.path(), TableTypeValue::CopyOnWrite,
false).await;
+ assert_eq!(
+ slices_and_ids(&table, &filters(&[("id", "=", "k04")])).await,
+ (3, vec!["k04".to_string()])
+ );
+}
+
+#[tokio::test]
+async fn test_key_filter_on_a_merge_on_read_table_reads_the_updated_row() {
+ let dir = tempdir().unwrap();
+ three_groups(dir.path(), TableTypeValue::MergeOnRead, true).await;
+ let mut writer = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ writer.upsert([batch(1, 1000)]).await.unwrap();
+
+ let reader = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ let options = filters(&[("id", "=", "k04")]);
+ let slices = reader.get_file_slices(&options).await.unwrap();
+ assert_eq!(slices.len(), 1);
+ assert!(slices[0].has_log_file(), "the upsert went to a log file");
+ let values: Vec<i64> = reader
+ .read(&options)
+ .await
+ .unwrap()
+ .iter()
+ .flat_map(|b| {
+ b.column_by_name("value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect();
+ assert_eq!(values, vec![1001]);
+}
+
+/// A reader opened before a delete keeps its snapshot: the index is read as
+/// of it, so the key still reads back from its own file group.
+#[tokio::test]
+async fn test_a_stale_reader_reads_the_index_as_of_its_snapshot() {
+ let dir = tempdir().unwrap();
+ let reader = three_groups(dir.path(), TableTypeValue::CopyOnWrite,
true).await;
+ let mut writer = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ writer
+ .delete_keys([HoodieKey {
+ record_key: "k04".to_string(),
+ partition_path: String::new(),
+ }])
+ .await
+ .unwrap();
+ assert_eq!(
+ slices_and_ids(&reader, &filters(&[("id", "=", "k04")])).await,
+ (1, vec!["k04".to_string()])
+ );
+ let fresh = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ assert_eq!(
+ slices_and_ids(&fresh, &filters(&[("id", "=", "k04")])).await,
+ (0, vec![])
+ );
+}
+
+/// `hoodie.metadata.enable=false` on the reader turns the lookup off.
+#[tokio::test]
+async fn test_metadata_disabled_on_the_reader_reads_every_file_group() {
+ let dir = tempdir().unwrap();
+ three_groups(dir.path(), TableTypeValue::CopyOnWrite, true).await;
+ let table = Table::new_with_options(
+ dir.path().to_str().unwrap(),
+ [("hoodie.metadata.enable", "false")],
+ )
+ .await
+ .unwrap();
+ assert_eq!(
+ slices_and_ids(&table, &filters(&[("id", "=", "k04")])).await,
+ (3, vec!["k04".to_string()])
+ );
+}
+
+/// Sorted values of the string column `column` across `batches`.
+fn strings(batches: &[RecordBatch], column: &str) -> Vec<String> {
+ let mut values: Vec<String> = batches
+ .iter()
+ .flat_map(|b| {
+ b.column_by_name(column)
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .iter()
+ .map(|v| v.unwrap().to_string())
+ .collect::<Vec<_>>()
+ })
+ .collect();
+ values.sort();
+ values
+}
+
+/// A commit still pending when the reader opened is not part of its
+/// snapshot, so its index entries are not either, even when it completes
+/// before the read.
+#[tokio::test]
+async fn
test_a_commit_pending_when_the_reader_opened_is_not_read_from_the_index() {
+ let dir = tempdir().unwrap();
+ three_groups(dir.path(), TableTypeValue::CopyOnWrite, true).await;
+ let mut writer = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ writer
+ .delete_keys([HoodieKey {
+ record_key: "k04".to_string(),
+ partition_path: String::new(),
+ }])
+ .await
+ .unwrap();
+ let delete_instant = writer
+ .get_timeline()
+ .completed_commits
+ .iter()
+ .map(|i| i.timestamp.clone())
+ .max()
+ .unwrap();
+ writer.upsert([batch(0, 500)]).await.unwrap();
+
+ // Open the reader while the delete looks in flight, then let it complete.
+ let timeline = dir.path().join(".hoodie/timeline");
+ let completed = std::fs::read_dir(&timeline)
+ .unwrap()
+ .map(|e| e.unwrap().path())
+ .find(|p| {
+ let name = p.file_name().unwrap().to_string_lossy().to_string();
+ name.starts_with(&format!("{delete_instant}_")) &&
name.ends_with(".commit")
+ })
+ .unwrap();
+ let hidden = dir.path().join("hidden.commit");
+ std::fs::rename(&completed, &hidden).unwrap();
+ let reader = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ std::fs::rename(&hidden, &completed).unwrap();
+
+ let (_, ids) = slices_and_ids(&reader, &filters(&[("id", "=",
"k04")])).await;
+ assert_eq!(ids, vec!["k04".to_string()]);
+}
+
+/// Spark-written partitioned tables whose metadata table holds an instant
+/// newer than the latest data commit (an indexing action). Footer statistics
+/// already narrow the copy-on-write read; the merge-on-read one, whose slices
+/// carry logs, narrows only through the index.
+#[tokio::test]
+async fn test_key_filter_on_spark_written_tables() {
+ use hudi_test::{SampleTable, TableFormat};
+ for format in [TableFormat::Cow, TableFormat::MorAvro] {
+ let path = SampleTable::V9TxnsSimpleMeta.path(format);
+ let table = Table::new(&path).await.unwrap();
+ let all = table
+ .get_file_slices(&ReadOptions::new())
+ .await
+ .unwrap()
+ .len();
+ let keys = strings(&table.read(&ReadOptions::new()).await.unwrap(),
"txn_id");
+ let probe = format!("{},{}", keys[0], keys[keys.len() - 1]);
+ let options = filters(&[("txn_id", "IN", probe.as_str())]);
+ let slices = table.get_file_slices(&options).await.unwrap().len();
+ let got = strings(&table.read(&options).await.unwrap(), "txn_id");
+
+ let scan = Table::new_with_options(&path, [("hoodie.metadata.enable",
"false")])
+ .await
+ .unwrap();
+ let want = strings(&scan.read(&options).await.unwrap(), "txn_id");
+ assert_eq!(got, want, "{format:?}");
+ assert!(slices < all, "{format:?}: {slices} of {all} slices planned");
+ }
+}
+
+/// Create the three-group table with a declared complex key generator at
+/// `version`, so a single-field key is stored as `id:<value>`.
+async fn complex_three_groups(dir: &std::path::Path, version: isize) {
+ Table::create(dir.to_str().unwrap())
+ .with_table_name("t")
+ .with_table_version(version)
+ .with_record_key_fields(["id"])
+ .with_record_index(true)
+ .with_option("hoodie.parquet.small.file.limit", "0")
+ .create()
+ .await
+ .unwrap();
+ let props = dir.join(".hoodie/hoodie.properties");
+ let rewritten: String = std::fs::read_to_string(&props)
+ .unwrap()
+ .lines()
+ .map(|line| {
+ if line.starts_with("hoodie.table.keygenerator.type=") {
+ "hoodie.table.keygenerator.type=COMPLEX\n".to_string()
+ } else {
+ format!("{line}\n")
+ }
+ })
+ .collect();
+ std::fs::write(&props, rewritten).unwrap();
+ let mut writer = Table::new_with_options(
+ dir.to_str().unwrap(),
+ [("hoodie.write.complex.keygen.validation.enable", "false")],
+ )
+ .await
+ .unwrap();
+ for (group, base) in [(0, 0), (1, 100), (2, 200)] {
+ writer.append([batch(group, base)]).await.unwrap();
+ }
+}
+
+#[tokio::test]
+async fn test_key_filter_with_a_complex_key_generator() {
+ let dir = tempdir().unwrap();
+ complex_three_groups(dir.path(), 9).await;
+ let table = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ assert_eq!(
+ strings(
+ &table.read(&ReadOptions::new()).await.unwrap(),
+ "_hoodie_record_key"
+ )[0],
+ "id:k00"
+ );
+ assert_eq!(
+ slices_and_ids(&table, &filters(&[("id", "=", "k04")])).await,
+ (1, vec!["k04".to_string()])
+ );
+}
+
+/// Below table version 9 the stored form is not knowable from the table, so
+/// the lookup stays off whatever the reader claims.
+#[tokio::test]
+async fn test_key_filter_with_a_complex_key_generator_below_version_9() {
+ let dir = tempdir().unwrap();
+ complex_three_groups(dir.path(), 8).await;
+ let table = Table::new_with_options(
+ dir.path().to_str().unwrap(),
+ [("hoodie.write.complex.keygen.new.encoding", "true")],
+ )
+ .await
+ .unwrap();
+ assert_eq!(
+ slices_and_ids(&table, &filters(&[("id", "=", "k04")])).await,
+ (3, vec!["k04".to_string()])
+ );
+}
+
+/// The index describes merged rows, so a read-optimized read, which reads
+/// base files only, does not narrow by it: a key deleted in a log file is
+/// still in its base file.
+#[tokio::test]
+async fn test_a_read_optimized_read_does_not_use_the_index() {
+ let dir = tempdir().unwrap();
+ three_groups(dir.path(), TableTypeValue::MergeOnRead, true).await;
+ let mut writer = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ writer
+ .delete_keys([HoodieKey {
+ record_key: "k04".to_string(),
+ partition_path: String::new(),
+ }])
+ .await
+ .unwrap();
+
+ let reader = Table::new(dir.path().to_str().unwrap()).await.unwrap();
+ let options = filters(&[("id", "=", "k04")])
+ .with_hudi_option("hoodie.read.use.read_optimized.mode", "true");
+ let (_, ids) = slices_and_ids(&reader, &options).await;
+ assert_eq!(ids, vec!["k04".to_string()]);
+}
+
+/// A commit pending when the table was opened and completed before this
+/// handle writes is part of the snapshot the handle reads after its write.
+#[tokio::test]
+async fn test_a_read_after_a_write_sees_a_commit_that_was_pending_at_open() {
+ let dir = tempdir().unwrap();
+ let path = dir.path().to_str().unwrap();
+ let mut table = Table::create(path)
+ .with_table_name("t")
+ .with_record_key_fields(["id"])
+ .with_record_index(true)
+ .with_option("hoodie.parquet.small.file.limit", "0")
+ .create()
+ .await
+ .unwrap();
+ table.append([batch(0, 0)]).await.unwrap();
+ table.append([batch(1, 100)]).await.unwrap();
+ let pending_instant = Table::new(path)
+ .await
+ .unwrap()
+ .get_timeline()
+ .completed_commits
+ .iter()
+ .map(|i| i.timestamp.clone())
+ .max()
+ .unwrap();
+
+ // Open the handle while the second append looks in flight.
+ let timeline = dir.path().join(".hoodie/timeline");
+ let completed = std::fs::read_dir(&timeline)
+ .unwrap()
+ .map(|e| e.unwrap().path())
+ .find(|p| {
+ let name = p.file_name().unwrap().to_string_lossy().to_string();
+ name.starts_with(&format!("{pending_instant}_")) &&
name.ends_with(".commit")
+ })
+ .unwrap();
+ let hidden = dir.path().join("hidden.commit");
+ std::fs::rename(&completed, &hidden).unwrap();
+ let mut handle = Table::new(path).await.unwrap();
+ std::fs::rename(&hidden, &completed).unwrap();
+
+ handle.append([batch(2, 200)]).await.unwrap();
+ let (_, ids) = slices_and_ids(&handle, &filters(&[("id", "=",
"k04")])).await;
+ assert_eq!(ids, vec!["k04".to_string()]);
+}
