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 3c8ad5e9 feat(core): read any metadata table partition as Arrow (#819)
3c8ad5e9 is described below
commit 3c8ad5e9cd9b6d5194414bc4fa10eb183b52d748
Author: Steven Lee <[email protected]>
AuthorDate: Wed Oct 28 07:29:47 2026 +0800
feat(core): read any metadata table partition as Arrow (#819)
* feat(core): read any metadata table partition as Arrow
Generalises the metadata-table Arrow read so a caller can read one
partition — record_index, column_stats, bloom_filters, partition_stats,
or a secondary_index_* — not just files.
- MetadataTableV2Reader::read_files_partition_batch becomes
read_partition_batch (partition name parameterised); the read path is
identical, only the partition the reader is pointed at changes.
- MetadataTable::read_files_partition_batch becomes a thin wrapper over a
new read_metadata_table_partition_batch.
- New public Table::read_metadata_table_partition_arrow(partition, keys)
for DATA tables, mirroring read_metadata_table_files_partition_arrow.
Motivated by the DuckDB Hudi extension (alitrack/duckdb-hudi) exposing
Hudi index data under DuckDB SQL: the files index is already exposed
(hudi_mdt) and column_stats is derived from parquet footers (hudi_stats);
record index / bloom / secondary indexes need this read.
* test(core): cover reading the record index as Arrow and reject undeclared
partitions
Read `record_index` through `read_metadata_table_partition_arrow` on the v9
txns fixture: a keyed read opens one shard and returns the key as stored, an
unknown key is an empty batch, and a full scan holds one entry per live
record. A partition the table does not declare in
`hoodie.table.metadata.partitions` is now an error rather than an empty
batch, since the two were indistinguishable.
---------
Co-authored-by: lihongyan <[email protected]>
---
crates/core/src/metadata/table/mod.rs | 135 +++++++++++++++++++++++++++-
crates/core/src/metadata/table/v2_reader.rs | 21 +++--
2 files changed, 148 insertions(+), 8 deletions(-)
diff --git a/crates/core/src/metadata/table/mod.rs
b/crates/core/src/metadata/table/mod.rs
index f3b499ab..bc039467 100644
--- a/crates/core/src/metadata/table/mod.rs
+++ b/crates/core/src/metadata/table/mod.rs
@@ -210,6 +210,53 @@ impl Table {
.await
}
+ /// Read one metadata-table partition as an Arrow batch, undecoded.
+ ///
+ /// The generalisation of
[`Self::read_metadata_table_files_partition_arrow`]:
+ /// `partition` is the metadata table partition to read — one of `files`,
+ /// `column_stats`, `partition_stats`, `bloom_filters`, `record_index`, or
a
+ /// `secondary_index_<name>` value (see
+ /// [`Self::get_metadata_table_partitions`]). `files` is the only partition
+ /// with a decoded record type; the others come back as raw Arrow in their
+ /// stored schema, so callers that want them should consume the batch
+ /// directly.
+ ///
+ /// `keys` empty reads every record; otherwise only those keys, matched
+ /// **as stored**.
+ ///
+ /// An empty batch when the partition has no commits yet.
+ ///
+ /// # Errors
+ /// [`CoreError::MetadataTable`] when `partition` is not one the table
+ /// declares in `hoodie.table.metadata.partitions`. An empty batch there
+ /// would be indistinguishable from a declared partition that holds no
+ /// records, and an undeclared name is most likely a typo.
+ ///
+ /// A file group that is log-only and holds nothing but tombstones fails in
+ /// the merge with "No schema available for merge output", since no record
+ /// survives to lend the output its schema. Not yet handled.
+ ///
+ /// # Note
+ /// Must be called on a DATA table, not a METADATA table.
+ pub async fn read_metadata_table_partition_arrow(
+ &self,
+ partition: &str,
+ keys: &[&str],
+ ) -> Result<arrow_array::RecordBatch> {
+ let declared = self.get_metadata_table_partitions();
+ if !declared.iter().any(|p| p == partition) {
+ return Err(CoreError::MetadataTable(format!(
+ "Metadata table partition '{partition}' is not declared on
this table; \
+ declared partitions: {declared:?}"
+ )));
+ }
+ let metadata_table = self.get_or_init_metadata_table().await?;
+ let valid = self.valid_instant_timestamps(metadata_table).await?;
+ metadata_table
+ .read_metadata_table_partition_batch(partition, keys, &valid)
+ .await
+ }
+
/// Fetch records from the `files` partition with optional partition
pruning.
///
/// For non-partitioned tables, directly fetches the "." record.
@@ -306,7 +353,28 @@ impl Table {
keys: &[&str],
valid_instants: &HashSet<String>,
) -> Result<arrow_array::RecordBatch> {
- let Some((reader, file_slices)) = self.files_partition_reader().await?
else {
+ self.read_metadata_table_partition_batch(
+ FilesPartitionRecord::PARTITION_NAME,
+ keys,
+ valid_instants,
+ )
+ .await
+ }
+
+ /// The merged batch of one metadata partition, for a caller that wants
Arrow.
+ ///
+ /// An empty batch when the partition is not enabled or has no commits,
+ /// matching the empty map the decoded form returns in that case.
+ ///
+ /// # Note
+ /// Must be called on a METADATA table instance.
+ pub(crate) async fn read_metadata_table_partition_batch(
+ &self,
+ partition: &str,
+ keys: &[&str],
+ valid_instants: &HashSet<String>,
+ ) -> Result<arrow_array::RecordBatch> {
+ let Some((reader, file_slices)) =
self.partition_reader(partition).await? else {
return Ok(arrow_array::RecordBatch::new_empty(std::sync::Arc::new(
Schema::empty(),
)));
@@ -316,7 +384,7 @@ impl Table {
let batches = crate::util::concurrency::bounded_in_order(
&targets,
self.bounded_read_concurrency_for(&targets),
- |file_slice| reader.read_files_partition_batch(file_slice, keys),
+ |file_slice| reader.read_partition_batch(file_slice, keys,
partition),
)
.await?;
concat_metadata_batches(batches)
@@ -1549,4 +1617,67 @@ mod tests {
);
Ok(())
}
+
+ /// The record index through the public Arrow API, on a table whose index
is
+ /// sharded across file groups. A keyed read opens one shard and comes back
+ /// keyed as stored; an unknown key is empty rather than an error; a full
+ /// scan holds one entry per live record.
+ #[tokio::test]
+ async fn read_metadata_table_partition_arrow_reads_the_record_index() ->
Result<()> {
+ use arrow_array::{Array, StringArray};
+
+ let table =
Table::new(&SampleTable::V9TxnsSimpleMeta.path_to_cow()).await?;
+
+ let batch = table
+ .read_metadata_table_partition_arrow("record_index", &["TXN-011"])
+ .await?;
+ assert_eq!(batch.num_rows(), 1, "one live key is one index entry");
+ let keys = batch
+ .column_by_name("key")
+ .expect("the record index keys on a `key` column")
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .expect("`key` is a string column");
+ assert_eq!(keys.value(0), "TXN-011");
+
+ let batch = table
+ .read_metadata_table_partition_arrow("record_index",
&["no-such-key"])
+ .await?;
+ assert_eq!(
+ batch.num_rows(),
+ 0,
+ "a missing key is an empty batch, not an error"
+ );
+
+ let live_rows: usize = table
+ .read(&crate::table::ReadOptions::new())
+ .await?
+ .iter()
+ .map(|batch| batch.num_rows())
+ .sum();
+ let batch = table
+ .read_metadata_table_partition_arrow("record_index", &[])
+ .await?;
+ assert_eq!(
+ batch.num_rows(),
+ live_rows,
+ "a full scan holds one entry per live record"
+ );
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn
read_metadata_table_partition_arrow_rejects_an_undeclared_partition() ->
Result<()> {
+ let table =
Table::new(&SampleTable::V9TxnsSimpleMeta.path_to_cow()).await?;
+ let err = table
+ .read_metadata_table_partition_arrow("secondary_idx_merchant", &[])
+ .await
+ .unwrap_err();
+ assert!(matches!(err, CoreError::MetadataTable(_)), "got {err}");
+ assert!(
+ err.to_string().contains("secondary_idx_merchant"),
+ "the error must name the partition, got {err}"
+ );
+ Ok(())
+ }
}
diff --git a/crates/core/src/metadata/table/v2_reader.rs
b/crates/core/src/metadata/table/v2_reader.rs
index 69f09125..a0176ceb 100644
--- a/crates/core/src/metadata/table/v2_reader.rs
+++ b/crates/core/src/metadata/table/v2_reader.rs
@@ -103,20 +103,29 @@ impl MetadataTableV2Reader {
file_slice: &FileSlice,
keys: &[&str],
) -> Result<HashMap<String, FilesPartitionRecord>> {
- let batch = self.read_files_partition_batch(file_slice, keys).await?;
+ let batch = self
+ .read_partition_batch(file_slice, keys,
FilesPartitionRecord::PARTITION_NAME)
+ .await?;
Self::records_from_batch(&batch, keys)
}
- /// The merged `files` partition batch, before it is decoded into records.
+ /// The merged batch of one metadata partition, before it is decoded into
+ /// records.
///
/// Separate because the batch is what an Arrow consumer wants and
decoding it
/// only to re-encode would be work in both directions. Nothing about the
read
/// differs between the two; this is the read, and the other is this plus a
/// decode.
- pub(crate) async fn read_files_partition_batch(
+ ///
+ /// `partition` is the metadata table partition to read (its Hudi partition
+ /// value: `files`, `record_index`, `column_stats`, `bloom_filters`,
+ /// `partition_stats`, `secondary_index_*`). The decoded-record layer only
+ /// exists for `files`; the others come back as raw Arrow.
+ pub(crate) async fn read_partition_batch(
&self,
file_slice: &FileSlice,
keys: &[&str],
+ partition: &str,
) -> Result<RecordBatch> {
let base_file_path = file_slice.base_file_relative_path()?;
let log_file_paths = if file_slice.has_log_file() {
@@ -137,7 +146,7 @@ impl MetadataTableV2Reader {
// The metadata table sets `hoodie.populate.meta.fields=false`, so the
record
// key is its own `key` column rather than `_hoodie_record_key`.
Without this
// the merge keys every record on an empty string and collapses them
into one.
-
reader_context.rebuild_record_context(FilesPartitionRecord::PARTITION_NAME.to_string());
+ reader_context.rebuild_record_context(partition.to_string());
// A named-key read is a point lookup, which is the normal shape for
the
// metadata table; an empty key list is the full scan the `files`
partition
@@ -165,7 +174,7 @@ impl MetadataTableV2Reader {
base_file_path,
base_file_commit_time,
log_file_paths,
- FilesPartitionRecord::PARTITION_NAME.to_string(),
+ partition.to_string(),
),
ReaderParameters::default(),
None,
@@ -175,7 +184,7 @@ impl MetadataTableV2Reader {
let batch = reader.read().await?;
log::debug!(
"metadata read of '{}' with {} named key(s): merge map peaked at
{} entries",
- FilesPartitionRecord::PARTITION_NAME,
+ partition,
keys.len(),
reader.read_stats().merge_map_peak_entries
);
