hudi-core 0.5.0

/*
 * 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.
 */

//! HFile implementation of [`BaseFileReader`].

use std::sync::Arc;

use arrow::array::{RecordBatch, RecordBatchOptions};
use arrow_schema::{Schema, SchemaRef};
use futures::StreamExt;
use futures::future::BoxFuture;

use super::reader::{BaseFileReadOptions, BaseFileReader, BaseFileStream, KeyPredicate};
use crate::file_group::log_file::avro::{AvroBlockDecoder, RegisteredWriterSchema};
use crate::hfile::HFileReader;
use crate::hfile::record_key::fill_empty_entry_keys;
use crate::statistics::{StatisticsContainer, StatsGranularity};
use crate::storage::Storage;
use crate::storage::error::{Result, StorageError};
use crate::storage::file_metadata::FileMetadata;
use crate::util::arrow::project_batch_by_names;

/// Records per Arrow batch while decoding an HFile's values.
const DECODE_BATCH_SIZE: usize = 1024;

/// Only reached if a reader reports no budget, which a ranged reader always does.
const DEFAULT_WINDOW_BUDGET_FALLBACK: u64 = 16 * 1024 * 1024;

/// Reads HFile base files.
#[derive(Debug)]
pub struct HFileBaseFileReader {
    storage: Arc<Storage>,
}

impl HFileBaseFileReader {
    pub fn new(storage: Arc<Storage>) -> Self {
        Self { storage }
    }

    /// The record schema an HFile was written with, as Avro JSON and as Arrow.
    ///
    /// An HFile stores each value Avro-encoded and carries the schema it used in
    /// its own file info. Decoding against that is what makes a base file and a
    /// log block of the same table yield the same columns, which is the whole
    /// reason they can merge; handing the value on as bytes does not.
    ///
    /// The decoder and the registration come back rather than being dropped, because
    /// building a decoder is the dominant cost of reading a small HFile: `arrow_avro`
    /// re-parses the writer schema's JSON on every construction, which for the
    /// metadata table's eight-kilobyte record schema is more than reading the file.
    /// The decoder has decoded nothing yet, so the first window can decode through
    /// it; the registration is immutable and serves every later window, which then
    /// pays only to build.
    fn decoded_schema(
        reader: &HFileReader,
        relative_path: &str,
    ) -> Result<(SchemaRef, AvroBlockDecoder, RegisteredWriterSchema)> {
        let json = reader
            .avro_schema_json()
            .map_err(|e| {
                StorageError::Creation(format!(
                    "Failed to read the Avro schema of HFile {relative_path}: {e:?}"
                ))
            })?
            .ok_or_else(|| {
                StorageError::Creation(format!(
                    "HFile {relative_path} carries no Avro schema, so its values cannot be decoded"
                ))
            })?
            .to_string();
        // The schema comes from the decoder, not from converting the Avro JSON:
        // `avro_to_arrow` does not handle named-type references, and the metadata
        // table's record schema uses them.
        let registered = RegisteredWriterSchema::new(&json)
            .map_err(|e| StorageError::Creation(format!("{e}")))?;
        let decoder =
            AvroBlockDecoder::try_new_with_registered(&registered, None, DECODE_BATCH_SIZE)
                .map_err(|e| StorageError::Creation(format!("{e}")))?;
        let schema = decoder.schema();
        Ok((schema, decoder, registered))
    }

    /// The projected schema, or an error naming a column the file does not have.
    /// An empty projection is the row-count-only request shape.
    fn project(full: &SchemaRef, projection: Option<&[String]>) -> Result<SchemaRef> {
        let Some(names) = projection else {
            return Ok(full.clone());
        };
        let mut fields = Vec::with_capacity(names.len());
        for name in names {
            let field = full.field_with_name(name).map_err(|_| {
                StorageError::InvalidColumn(format!("{name} is not a column of this HFile"))
            })?;
            fields.push(field.clone());
        }
        Ok(Arc::new(Schema::new(fields)))
    }

    /// Open the file, reading it whole when it is small enough and in ranges when it
    /// is not.
    ///
    /// The bound depends on what the read is going to do, because a scan and a seek
    /// want opposite things: see [`HFILE_WHOLE_READ_WITH_KEYS_MAX_SIZE`] for the
    /// measurement and for why Hudi's single threshold is not followed here.
    /// `known_file_size` lets the caller spare the size lookup when the listing
    /// already told it.
    ///
    /// [`HFILE_WHOLE_READ_WITH_KEYS_MAX_SIZE`]: crate::storage::reader::HFILE_WHOLE_READ_WITH_KEYS_MAX_SIZE
    async fn open(
        &self,
        relative_path: &str,
        options: &BaseFileReadOptions,
    ) -> Result<HFileReader> {
        let mut whole_below =
            crate::storage::reader::hfile_whole_read_max_size(&self.storage.hudi_configs)
                .map_err(|e| StorageError::Creation(format!("{e}")))?;
        if options.key_predicate.is_some() {
            whole_below =
                whole_below.min(crate::storage::reader::HFILE_WHOLE_READ_WITH_KEYS_MAX_SIZE);
        }
        HFileReader::open_sized(
            &self.storage,
            relative_path,
            whole_below,
            options.known_file_size,
        )
        .await
        .map_err(|e| StorageError::Creation(format!("Failed to read HFile {relative_path}: {e:?}")))
    }

    /// The read itself, once the file is open. Split out of [`Self::read_stream`]
    /// so a caller can hold the reader across the read and read its
    /// `FetchCounts` — the only way to observe that a key predicate narrowed
    /// what was fetched, since selection over-includes and `decode_window`
    /// filters afterwards, leaving the rows identical either way.
    fn stream_from(
        &self,
        reader: HFileReader,
        relative_path: &str,
        options: &BaseFileReadOptions,
    ) -> Result<BaseFileStream> {
        // `Some(vec![])` asks for the row count only. The trailer carries it,
        // so no data block is read and no schema is needed.
        //
        // Not when a key predicate is set: the trailer counts the whole file,
        // so answering from it would report every record as though it matched.
        // A caller asking how many of five keys a file holds must get five or
        // fewer, so that combination reads blocks like any other.
        if options.key_predicate.is_none()
            && options.projection.as_ref().is_some_and(|p| p.is_empty())
        {
            let schema: SchemaRef = Arc::new(Schema::empty());
            let row_count = usize::try_from(reader.num_entries()).unwrap_or(usize::MAX);
            let batch = RecordBatch::try_new_with_options(
                schema.clone(),
                vec![],
                &RecordBatchOptions::new().with_row_count(Some(row_count)),
            )
            .map_err(StorageError::ArrowError)?;
            return Ok(BaseFileStream::new(
                schema,
                futures::stream::once(async move { Ok(batch) }).boxed(),
            ));
        }

        let (full_schema, decoder, registered) = Self::decoded_schema(&reader, relative_path)?;
        let schema = Self::project(&full_schema, options.projection.as_deref())?;
        let projection: Option<Vec<String>> =
            options.projection.as_ref().map(|names| names.to_vec());

        let budget = reader
            .window_budget()
            .unwrap_or(DEFAULT_WINDOW_BUDGET_FALLBACK);
        // Seek when the caller named keys, scan when it did not. Selection
        // over-includes, so `decode_window` filters below; this only changes
        // which blocks are fetched.
        let entries = match options.key_predicate.as_ref() {
            Some(predicate) => reader.blocks_for_predicate(predicate),
            None => reader.data_block_entries(),
        };
        // Windows still bound peak memory over whatever was selected.
        let windows = HFileReader::plan_windows(&entries, budget);
        let key_predicate = options.key_predicate.clone();

        let stream = futures::stream::unfold(
            (
                reader,
                windows.into_iter(),
                full_schema,
                projection,
                key_predicate,
                // The decoder the schema was resolved with, for the first window.
                // Later windows build their own: a decoder is flushed at the end of
                // a window and `arrow_avro` does not fully reset a union's state on
                // flush (arrow-rs#10876), so carrying one across a window boundary
                // would decode the next window against stale offsets.
                Some(decoder),
                registered,
                false,
            ),
            |(
                reader,
                mut windows,
                full_schema,
                projection,
                key_predicate,
                decoder,
                registered,
                failed,
            )| async move {
                // Sticky: once a window fails the read is not whole, so no
                // later window is handed out as though it were.
                if failed {
                    return None;
                }
                let window = windows.next()?;
                let item = decode_window(
                    &reader,
                    &window,
                    &full_schema,
                    projection.as_deref(),
                    key_predicate.as_ref(),
                    decoder,
                    &registered,
                )
                .await;
                let failed = item.is_err();
                Some((
                    item,
                    (
                        reader,
                        windows,
                        full_schema,
                        projection,
                        key_predicate,
                        None,
                        registered,
                        failed,
                    ),
                ))
            },
        )
        .boxed();

        Ok(BaseFileStream::new(schema, stream))
    }
}

impl BaseFileReader for HFileBaseFileReader {
    fn read_stream<'a>(
        &'a self,
        relative_path: &'a str,
        options: BaseFileReadOptions,
    ) -> BoxFuture<'a, Result<BaseFileStream>> {
        Box::pin(async move {
            let reader = self.open(relative_path, &options).await?;
            self.stream_from(reader, relative_path, &options)
        })
    }

    /// The record count comes from the trailer, and the ranged open already
    /// learned the file length, so neither costs an extra request.
    fn get_metadata_and_stats<'a>(
        &'a self,
        relative_path: &'a str,
        _table_schema: &'a Schema,
    ) -> BoxFuture<'a, Result<(FileMetadata, StatisticsContainer)>> {
        Box::pin(async move {
            // Ranged unconditionally here, whatever the threshold: this reports the
            // file's length and record count, both of which come from the trailer, so
            // buffering the file would be paid for nothing.
            let reader = HFileReader::open_ranged(&self.storage, relative_path)
                .await
                .map_err(|e| {
                    StorageError::Creation(format!("Failed to read HFile {relative_path}: {e:?}"))
                })?;
            let size = reader.file_len();

            let name = std::path::Path::new(relative_path)
                .file_name()
                .and_then(|n| n.to_str())
                .unwrap_or(relative_path)
                .to_string();

            let mut metadata = FileMetadata::new(name, size);
            metadata.num_records = i64::try_from(reader.num_entries()).unwrap_or(i64::MAX);

            let stats = StatisticsContainer {
                granularity: StatsGranularity::File,
                num_rows: Some(metadata.num_records),
                columns: std::collections::HashMap::new(),
            };
            Ok((metadata, stats))
        })
    }
}

/// Decode one window's records into a single batch, projected if asked.
///
/// Each value is an Avro datum on its own, so it needs no de-framing; the
/// decoder batches them and is flushed once per window.
async fn decode_window(
    reader: &HFileReader,
    window: &[crate::hfile::BlockIndexEntry],
    decoded_schema: &SchemaRef,
    projection: Option<&[String]>,
    key_predicate: Option<&KeyPredicate>,
    decoder: Option<AvroBlockDecoder>,
    registered: &RegisteredWriterSchema,
) -> Result<RecordBatch> {
    let mut records = reader
        .read_records_batched(window)
        .await
        .map_err(|e| StorageError::Creation(format!("Failed to read HFile data blocks: {e:?}")))?;

    // A block is the smallest thing that can be read, so a selected block holds
    // keys nobody asked for. Dropping them here rather than after decoding keeps
    // the entry-key fill below aligned with the rows it fills, since that fill is
    // positional over `records`.
    if let Some(predicate) = key_predicate {
        // Built once per window, not per record: a record-index lookup names
        // thousands of keys and this runs on every record the window holds.
        let matcher = predicate.matcher();
        records.retain(|record| match std::str::from_utf8(&record.key) {
            Ok(key) => matcher.admits(key),
            // A key that is not UTF-8 cannot match a predicate expressed as
            // strings. Keeping it would put a row through that the caller asked
            // not to see.
            Err(_) => false,
        });
    }

    // A later window builds its own decoder, since the previous one was flushed and
    // `arrow_avro` does not fully reset a union's state on flush (arrow-rs#10876).
    // It builds from the registration rather than the JSON, which is the schema-sized
    // half of the cost and is immutable.
    let mut decoder = match decoder {
        Some(decoder) => decoder,
        None => AvroBlockDecoder::try_new_with_registered(registered, None, DECODE_BATCH_SIZE)
            .map_err(|e| StorageError::Creation(format!("{e}")))?,
    };
    let mut batches: Vec<RecordBatch> = Vec::new();
    for record in &records {
        if let Some(batch) = decoder
            .decode(&record.value)
            .map_err(|e| StorageError::Creation(format!("{e}")))?
        {
            batches.push(batch);
        }
    }
    if let Some(batch) = decoder
        .flush()
        .map_err(|e| StorageError::Creation(format!("{e}")))?
        && batch.num_rows() > 0
    {
        batches.push(batch);
    }

    // The stream declares the decoded schema, so an empty window must carry that
    // schema too: a batch whose schema disagrees with the stream's breaks
    // projection and the merge downstream.
    let schema = batches
        .first()
        .map(|b| b.schema())
        .unwrap_or_else(|| decoded_schema.clone());
    let combined =
        arrow::compute::concat_batches(&schema, &batches).map_err(StorageError::ArrowError)?;

    // A writer may leave the record's key field empty because the HFile entry key
    // already carries it. Positional, so the decode order above is load-bearing.
    let entry_keys: Vec<&str> = records
        .iter()
        .map(|r| {
            r.key_as_str().ok_or_else(|| {
                StorageError::Creation("an HFile record key is not valid UTF-8".to_string())
            })
        })
        .collect::<Result<Vec<&str>>>()?;
    let combined = fill_empty_entry_keys(combined, &entry_keys)
        .map_err(|e| StorageError::Creation(format!("{e}")))?;

    project_batch_by_names(combined, projection).map_err(|e| StorageError::Creation(format!("{e}")))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::HudiConfigs;
    use crate::config::read::HudiReadConfig;
    use crate::config::table::HudiTableConfig;
    use crate::file_group::FileGroup;
    use crate::file_group::reader_v2::MAX_INSTANT_TIME;
    use crate::file_group::reader_v2::engine::HoodieFileGroupReader;
    use crate::file_group::reader_v2::input_split::InputSplit;
    use crate::file_group::reader_v2::reader_parameters::ReaderParameters;
    use crate::file_group::reader_v2::resolver::resolve_reader_context;
    use crate::hfile::{BlockIndexEntry, Key};
    use crate::metadata::table::reader::MetadataTableFileGroupReader;
    use arrow_array::cast::AsArray;
    use std::collections::{HashMap, HashSet};
    use std::fs::canonicalize;
    use std::path::PathBuf;
    use url::Url;

    /// The metadata table's `files` partition, base file only. The seven log
    /// files beside it belong to the log-block merge path, not this one.
    const MDT_FILES_BASE_FILE: &str = "files/files-0000-0_0-955-2690_00000000000000000.hfile";
    /// The `files` partition's base file *after* the compaction at
    /// 20251220210130942, which carries the whole listing: `__all_partitions__`
    /// and one record per partition. The pre-compaction base file above holds a
    /// single record, so it cannot show a predicate filtering anything.
    const MDT_FILES_COMPACTED_BASE_FILE: &str =
        "files/files-0000-0_23-1133-3302_20251220210130942.hfile";

    const MDT_FILES_FILE_GROUP: &str = "files-0000-0";
    const MDT_FILES_PARTITION: &str = "files";

    fn metadata_table_uri() -> String {
        use hudi_test::QuickstartTripsTable;
        let table_path = QuickstartTripsTable::V8Trips8I3U1D.path_to_mor_avro();
        let mdt = PathBuf::from(table_path).join(".hoodie").join("metadata");
        Url::from_file_path(canonicalize(&mdt).unwrap())
            .unwrap()
            .as_ref()
            .to_string()
    }

    /// The metadata table's own properties: HFILE base files, and a CUSTOM merge
    /// mode that a slice with no log files never consults.
    fn mdt_configs() -> Arc<HudiConfigs> {
        Arc::new(HudiConfigs::new([
            (HudiTableConfig::BasePath.as_ref(), metadata_table_uri()),
            (
                HudiTableConfig::BaseFileFormat.as_ref(),
                "hfile".to_string(),
            ),
            (
                HudiReadConfig::EndTimestamp.as_ref(),
                MAX_INSTANT_TIME.to_string(),
            ),
            // The metadata table's own record-key config. Without it the key
            // resolves to `_hoodie_record_key`, which every MDT record leaves
            // empty, and a merge collapses all keys into one.
            (HudiTableConfig::RecordKeyFields.as_ref(), "key".to_string()),
            // The metadata table's real merge semantics: CUSTOM, deferred to the
            // metadata payload by the all-zeros strategy id.
            ("hoodie.record.merge.mode", "CUSTOM".to_string()),
            (
                "hoodie.record.merge.strategy.id",
                "00000000-0000-0000-0000-000000000000".to_string(),
            ),
            (
                "hoodie.compaction.payload.class",
                "org.apache.hudi.metadata.HoodieMetadataPayload".to_string(),
            ),
            (
                HudiTableConfig::PopulatesMetaFields.as_ref(),
                "false".to_string(),
            ),
        ]))
    }

    /// What `MetadataTableFileGroupReader` returns for the same base-file-only
    /// slice: the oracle this read is checked against.
    async fn reference_keys() -> crate::Result<HashSet<String>> {
        let configs = mdt_configs();
        let storage = Storage::new(Arc::new(HashMap::new()), configs.clone())?;
        let reader = MetadataTableFileGroupReader::new(configs, storage);

        let mut fg = FileGroup::new(
            MDT_FILES_FILE_GROUP.to_string(),
            MDT_FILES_PARTITION.to_string(),
        );
        fg.add_base_file_from_name(MDT_FILES_BASE_FILE.strip_prefix("files/").unwrap())?;
        let slice = fg
            .get_file_slice_as_of(MAX_INSTANT_TIME)
            .expect("the file group has a slice")
            .clone();

        let records = reader.read_files_partition(&slice, &[]).await?;
        Ok(records.into_keys().collect())
    }

    /// An HFile base file read through the v2 file group reader returns the same
    /// keys the metadata-table reader returns for the same base-file-only slice.
    /// The `files` partition's pre-compaction slice: the initial base HFile plus
    /// every log file written before the compaction at 20251220210130942.
    const MDT_FILES_PRECOMPACT_BASE: &str = "files-0000-0_0-955-2690_00000000000000000.hfile";
    const MDT_FILES_PRECOMPACT_LOGS: &[&str] = &[
        ".files-0000-0_20251220210108078.log.1_10-999-2838",
        ".files-0000-0_20251220210123755.log.1_3-1032-2950",
        ".files-0000-0_20251220210125441.log.1_5-1057-3024",
        ".files-0000-0_20251220210127080.log.1_3-1082-3100",
        ".files-0000-0_20251220210128625.log.1_5-1107-3174",
        ".files-0000-0_20251220210129235.log.1_3-1118-3220",
        ".files-0000-0_20251220210130911.log.1_3-1149-3338",
    ];

    fn precompact_slice() -> crate::Result<crate::file_group::FileSlice> {
        let mut fg = FileGroup::new(
            MDT_FILES_FILE_GROUP.to_string(),
            MDT_FILES_PARTITION.to_string(),
        );
        fg.add_base_file_from_name(MDT_FILES_PRECOMPACT_BASE)?;
        fg.add_log_files_from_names(MDT_FILES_PRECOMPACT_LOGS.iter().copied())?;
        Ok(fg
            .get_file_slice_as_of(MAX_INSTANT_TIME)
            .expect("the file group has a slice")
            .clone())
    }

    /// Every `filesystemMetadata` entry the metadata-table reader's own fold
    /// produces, so the comparison is on file names and sizes rather than counts.
    fn entries_by_key(
        records: &HashMap<String, crate::metadata::table::records::FilesPartitionRecord>,
    ) -> std::collections::BTreeMap<String, Vec<(String, i64, bool)>> {
        records
            .iter()
            .map(|(key, record)| {
                let mut entries: Vec<(String, i64, bool)> = record
                    .files
                    .iter()
                    .map(|(name, info)| (name.clone(), info.size, info.is_deleted))
                    .collect();
                entries.sort();
                (key.clone(), entries)
            })
            .collect()
    }

    /// The same, read off v2's merged Arrow batch.
    fn entries_from_batch(
        batch: &arrow_array::RecordBatch,
    ) -> std::collections::BTreeMap<String, Vec<(String, i64, bool)>> {
        use arrow_array::Array;
        let keys = batch
            .column_by_name("key")
            .expect("the metadata record key column")
            .as_string::<i32>();
        let map = batch
            .column_by_name("filesystemMetadata")
            .expect("the files-partition map column")
            .as_map();
        (0..batch.num_rows())
            .map(|row| {
                let entries = map.value(row);
                let names = entries.column(0).as_string::<i32>();
                let values = entries.column(1).as_struct();
                let sizes = values
                    .column_by_name("size")
                    .expect("size")
                    .as_primitive::<arrow_array::types::Int64Type>();
                let deleted = values
                    .column_by_name("isDeleted")
                    .expect("isDeleted")
                    .as_boolean();
                let mut out: Vec<(String, i64, bool)> = (0..names.len())
                    .map(|i| (names.value(i).to_string(), sizes.value(i), deleted.value(i)))
                    .collect();
                out.sort();
                (keys.value(row).to_string(), out)
            })
            .collect()
    }

    /// A metadata-table `files` slice with log blocks, merged by v2 under the
    /// table's own CUSTOM merge mode, lists exactly what the metadata-table
    /// reader's fold lists.
    ///
    /// Under `COMMIT_TIME_ORDERING` the same read returns all four keys but one
    /// file entry each, because the newest log block's map replaces the base
    /// record's rather than folding into it: four entries where the fold has
    /// fourteen. That is the regression this pins.
    #[tokio::test]
    async fn v2_folds_a_metadata_files_slice_like_the_metadata_table_reader() -> crate::Result<()> {
        let configs = mdt_configs();
        let storage = Storage::new(Arc::new(HashMap::new()), configs.clone())?;

        let expected = entries_by_key(
            &MetadataTableFileGroupReader::new(configs.clone(), storage.clone())
                .read_files_partition(&precompact_slice()?, &[])
                .await?,
        );
        assert!(
            expected.values().map(Vec::len).sum::<usize>() > expected.len(),
            "the oracle must fold several entries per key, or the comparison is vacuous"
        );

        let base = format!("{MDT_FILES_PARTITION}/{MDT_FILES_PRECOMPACT_BASE}");
        let mut context =
            resolve_reader_context(&configs, /* has_log_files */ true, Some(&base))?;
        context.rebuild_record_context(MDT_FILES_PARTITION.to_string());
        assert_eq!(
            context.merge_mode, "CUSTOM",
            "the metadata table's own merge mode must reach the reader"
        );

        let mut reader = HoodieFileGroupReader::new(
            Arc::new(context),
            storage,
            InputSplit::new(
                Some(base.clone()),
                Some("00000000000000000".to_string()),
                MDT_FILES_PRECOMPACT_LOGS
                    .iter()
                    .map(|f| format!("{MDT_FILES_PARTITION}/{f}"))
                    .collect(),
                MDT_FILES_PARTITION.to_string(),
            ),
            ReaderParameters::default(),
            None,
            None,
        )?;
        let batch = reader.read().await?;

        assert_eq!(entries_from_batch(&batch), expected);
        Ok(())
    }

    /// The `partition_stats` file group's log-only slice: eight log files, no base
    /// file, written before the compaction at 20251220210130942.
    const MDT_PARTITION_STATS_LOGS: &[&str] = &[
        ".partition-stats-0000-0_00000000000000003.log.1_0-0-0",
        ".partition-stats-0000-0_20251220210108078.log.1_9-999-2837",
        ".partition-stats-0000-0_20251220210123755.log.1_2-1032-2949",
        ".partition-stats-0000-0_20251220210125441.log.1_4-1057-3023",
        ".partition-stats-0000-0_20251220210127080.log.1_2-1082-3099",
        ".partition-stats-0000-0_20251220210128625.log.1_4-1107-3173",
        ".partition-stats-0000-0_20251220210129235.log.1_2-1118-3219",
        ".partition-stats-0000-0_20251220210130911.log.1_2-1149-3337",
    ];
    const MDT_PARTITION_STATS_PARTITION: &str = "partition_stats";

    /// Read a `partition_stats` slice through v2, keyed by record key.
    async fn read_partition_stats(
        logs: &[String],
    ) -> crate::Result<Option<arrow_array::RecordBatch>> {
        let configs = mdt_configs();
        let storage = Storage::new(Arc::new(HashMap::new()), configs.clone())?;
        let mut context = resolve_reader_context(
            &configs, /* has_log_files */ true, /* base */ None,
        )?;
        context.rebuild_record_context(MDT_PARTITION_STATS_PARTITION.to_string());
        let mut reader = HoodieFileGroupReader::new(
            Arc::new(context),
            storage,
            InputSplit::new(
                None,
                Some("00000000000000003".to_string()),
                logs.to_vec(),
                MDT_PARTITION_STATS_PARTITION.to_string(),
            ),
            ReaderParameters::default(),
            None,
            None,
        )?;
        // The first log file carries no data blocks, so a single-file read of it
        // has no schema to build an output from. That is the input's shape, not a
        // failure of the merge.
        Ok(reader.read().await.ok())
    }

    /// One row of a partition-statistics record, rendered so a comparison reads
    /// as values rather than as Arrow internals.
    fn stats_row(batch: &arrow_array::RecordBatch, row: usize) -> (String, String, String, i64) {
        use arrow_array::Array;
        let stats = batch
            .column_by_name("ColumnStatsMetadata")
            .expect("the column-statistics column")
            .as_struct();
        let render = |name: &str| -> String {
            let union = stats
                .column_by_name(name)
                .unwrap()
                .as_any()
                .downcast_ref::<arrow_array::UnionArray>()
                .expect("a bound is a union of typed wrappers");
            let child = union.child(union.type_id(row));
            let offset = union.value_offset(row);
            match child.as_struct_opt() {
                Some(wrapper) => arrow_cast::display::array_value_to_string(
                    wrapper.column_by_name("value").unwrap(),
                    offset,
                )
                .unwrap(),
                None => "null".to_string(),
            }
        };
        (
            stats
                .column_by_name("columnName")
                .unwrap()
                .as_string::<i32>()
                .value(row)
                .to_string(),
            render("minValue"),
            render("maxValue"),
            stats
                .column_by_name("valueCount")
                .unwrap()
                .as_primitive::<arrow_array::types::Int64Type>()
                .value(row),
        )
    }

    /// A `partition_stats` slice merges through v2's CUSTOM path and lands on the
    /// newest record for each key.
    ///
    /// Every record this table writes is tight-bound, so
    /// `mergeColumnStatsRecords`'s second short-circuit applies to all of them and
    /// the merged result must equal the newest log file's record for that key.
    /// The oracle is therefore derived from the inputs: read each log file alone,
    /// in commit order, and keep the last occurrence.
    ///
    /// This pins the short-circuit and that the fold survives real dense unions.
    /// It does **not** cover the bound widening or the counter sums; no fixture in
    /// this repo writes a non-tight-bound statistics record, so those are covered
    /// by unit tests on `fold_column_stats` instead.
    #[tokio::test]
    async fn v2_merges_a_partition_stats_slice_to_the_newest_record() -> crate::Result<()> {
        let logs: Vec<String> = MDT_PARTITION_STATS_LOGS
            .iter()
            .map(|f| format!("{MDT_PARTITION_STATS_PARTITION}/{f}"))
            .collect();

        // The oracle: last writer per key, across the log files in commit order.
        let mut expected: std::collections::BTreeMap<String, (String, String, String, i64)> =
            std::collections::BTreeMap::new();
        let mut records_seen = 0usize;
        for log in &logs {
            let Some(batch) = read_partition_stats(std::slice::from_ref(log)).await? else {
                continue;
            };
            let keys = batch.column_by_name("key").unwrap().as_string::<i32>();
            for row in 0..batch.num_rows() {
                records_seen += 1;
                expected.insert(keys.value(row).to_string(), stats_row(&batch, row));
            }
        }
        assert!(
            records_seen > expected.len(),
            "the slice must contain repeated keys, or nothing merges and this test \
             proves nothing: {records_seen} records over {} keys",
            expected.len()
        );

        let merged = read_partition_stats(&logs)
            .await?
            .expect("the full slice must read");
        let keys = merged.column_by_name("key").unwrap().as_string::<i32>();
        let actual: std::collections::BTreeMap<String, (String, String, String, i64)> = (0..merged
            .num_rows())
            .map(|row| (keys.value(row).to_string(), stats_row(&merged, row)))
            .collect();

        assert_eq!(actual, expected);
        Ok(())
    }

    /// The `secondary_index` file group's earliest slice: a base HFile written at
    /// instant 4, and one log file written later that deletes its record.
    const MDT_SECONDARY_INDEX_PARTITION: &str = "secondary_index_rider_idx";
    const MDT_SECONDARY_INDEX_BASE: &str =
        "secondary-index-rider-idx-0000-0_0-1008-2875_00000000000000004.hfile";
    const MDT_SECONDARY_INDEX_LOG: &str =
        ".secondary-index-rider-idx-0000-0_20251220210128625.log.1_0-1107-3169";

    /// A delete in a metadata log block cancels the base record, under the
    /// metadata table's own CUSTOM merge mode.
    ///
    /// This is the only end-to-end coverage of the custom merger's delete path:
    /// `delta_merge_delete` returns the tombstone whatever the partition type,
    /// mirroring `preCombine`'s short-circuit on `isDeletedRecord`, and the base
    /// record must not survive it.
    ///
    /// It covers a third partition type reaching the merger (secondary index,
    /// type 7). It does **not** cover that type's data-vs-data rule: no fixture
    /// here holds the same index key as live data in both a base file and a log,
    /// so "the newer record wins" is covered by a unit test and by nothing else.
    #[tokio::test]
    async fn a_metadata_log_delete_cancels_the_base_record() -> crate::Result<()> {
        let base = format!("{MDT_SECONDARY_INDEX_PARTITION}/{MDT_SECONDARY_INDEX_BASE}");
        let log = format!("{MDT_SECONDARY_INDEX_PARTITION}/{MDT_SECONDARY_INDEX_LOG}");

        async fn read(
            base: String,
            logs: Vec<String>,
        ) -> crate::Result<(arrow_array::RecordBatch, u64)> {
            let configs = mdt_configs();
            let storage = Storage::new(Arc::new(HashMap::new()), configs.clone())?;
            let mut context =
                resolve_reader_context(&configs, !logs.is_empty(), Some(base.as_str()))?;
            context.rebuild_record_context(MDT_SECONDARY_INDEX_PARTITION.to_string());
            let mut reader = HoodieFileGroupReader::new(
                Arc::new(context),
                storage,
                InputSplit::new(
                    Some(base),
                    Some("00000000000000004".to_string()),
                    logs,
                    MDT_SECONDARY_INDEX_PARTITION.to_string(),
                ),
                ReaderParameters::default(),
                None,
                None,
            )?;
            let batch = reader.read().await?;
            let deletes = reader.read_stats().num_deletes;
            Ok((batch, deletes))
        }

        let (base_only, _) = read(base.clone(), vec![]).await?;
        assert_eq!(
            base_only.num_rows(),
            1,
            "the base file must hold the record the log then deletes, or this test \
             would pass on an empty base"
        );

        let (merged, deletes) = read(base, vec![log]).await?;
        assert_eq!(deletes, 1, "the log block must contribute one delete");
        assert_eq!(
            merged.num_rows(),
            0,
            "a delete must cancel the base record rather than leaving it readable"
        );
        Ok(())
    }

    /// A key predicate given to the reader narrows what is read and returns only
    /// the keys asked for.
    ///
    /// Driven through `read_data`, the trait method callers use, rather than
    /// through `blocks_for_keys` directly — the selection has its own tests in
    /// `hfile::reader`; this one is about the predicate surviving the trip through
    /// `BaseFileReadOptions` and being both applied and filtered by.
    #[tokio::test]
    async fn a_key_predicate_narrows_the_read_and_filters_the_rows() -> crate::Result<()> {
        let configs = mdt_configs();
        let storage = Storage::new(Arc::new(HashMap::new()), configs)?;
        let reader = HFileBaseFileReader::new(storage);

        let all = reader
            .read_data(
                MDT_FILES_COMPACTED_BASE_FILE,
                BaseFileReadOptions::default(),
            )
            .await?;
        let keys: Vec<String> = all
            .column_by_name("key")
            .expect("the metadata record key column")
            .as_string::<i32>()
            .iter()
            .flatten()
            .map(str::to_string)
            .collect();
        assert!(
            keys.len() > 1,
            "the fixture must hold several keys, or filtering cannot be observed"
        );

        let wanted = keys.last().unwrap().clone();
        let filtered = reader
            .read_data(
                MDT_FILES_COMPACTED_BASE_FILE,
                BaseFileReadOptions {
                    key_predicate: Some(KeyPredicate::Keys(vec![wanted.clone()])),
                    ..Default::default()
                },
            )
            .await?;

        let got: Vec<String> = filtered
            .column_by_name("key")
            .unwrap()
            .as_string::<i32>()
            .iter()
            .flatten()
            .map(str::to_string)
            .collect();
        assert_eq!(
            got,
            vec![wanted],
            "a key predicate must return exactly the keys it named, since a \
             selected block holds others"
        );
        Ok(())
    }

    #[tokio::test]
    async fn v2_reads_an_hfile_base_file_matching_the_metadata_table_reader() -> crate::Result<()> {
        let expected = reference_keys().await?;
        assert!(
            !expected.is_empty(),
            "the oracle must return keys, or the comparison is vacuous"
        );

        let configs = mdt_configs();
        let storage = Storage::new(Arc::new(HashMap::new()), configs.clone())?;
        let mut context = resolve_reader_context(
            &configs,
            /* has_log_files */ false,
            Some(MDT_FILES_BASE_FILE),
        )?;
        // The production path rebuilds this immediately (`adapter.rs`), and without
        // it the merge keys on `_hoodie_record_key` whatever the table says. A test
        // that skips it is not exercising the context the reader is really given.
        context.rebuild_record_context(MDT_FILES_PARTITION.to_string());

        let mut reader = HoodieFileGroupReader::new(
            Arc::new(context),
            storage,
            InputSplit::new(
                Some(MDT_FILES_BASE_FILE.to_string()),
                Some("00000000000000000".to_string()),
                vec![],
                MDT_FILES_PARTITION.to_string(),
            ),
            ReaderParameters::default(),
            None,
            None,
        )?;
        let batch = reader.read().await?;

        let keys: HashSet<String> = batch
            .column_by_name("key")
            .expect("the metadata table's record key column, decoded from the HFile's values")
            .as_string::<i32>()
            .iter()
            .flatten()
            .map(str::to_string)
            .collect();

        assert_eq!(
            keys, expected,
            "v2's key set must equal the metadata-table reader's for the same slice"
        );
        Ok(())
    }

    fn entry(offset: u64, size: u32) -> BlockIndexEntry {
        BlockIndexEntry::new(Key::from_bytes(vec![0]), None, offset, size)
    }

    #[test]
    fn windows_stop_at_the_budget_and_never_split_a_block() {
        let entries = vec![entry(0, 40), entry(40, 40), entry(80, 40)];

        let windows = HFileReader::plan_windows(&entries, 100);
        assert_eq!(
            windows.iter().map(Vec::len).collect::<Vec<_>>(),
            vec![2, 1],
            "40 + 40 fits under 100, the third block starts a new window"
        );

        let windows = HFileReader::plan_windows(&entries, 1_000);
        assert_eq!(
            windows.iter().map(Vec::len).collect::<Vec<_>>(),
            vec![3],
            "a budget above the total is one window"
        );

        // A block bigger than the budget still goes out whole: it cannot decode
        // in pieces.
        let windows = HFileReader::plan_windows(&[entry(0, 500)], 100);
        assert_eq!(windows, vec![vec![entry(0, 500)]]);

        assert!(HFileReader::plan_windows(&[], 100).is_empty());
    }

    /// A read split over several windows must equal a read that used one.
    ///
    /// The first window decodes through the decoder the schema was resolved with,
    /// and later windows build their own, because `arrow_avro` does not fully reset
    /// a union's state on flush (arrow-rs#10876) and a decoder is flushed at the end
    /// of every window. Reusing one across that boundary decodes the next window
    /// against stale offsets, which surfaces as wrong values rather than an error,
    /// so this pins the values and not only the row count.
    ///
    /// The fixture is a fifty-record HFile written with one-kilobyte blocks, because
    /// the tables in this repo hold their `files` partition in a single data block
    /// and a one-block file has only ever one window however small the budget.
    #[tokio::test]
    async fn a_read_split_over_windows_decodes_what_one_window_does() -> crate::Result<()> {
        async fn read(windowed: bool) -> crate::Result<(Vec<(String, i32)>, usize)> {
            let dir =
                std::fs::canonicalize(std::path::Path::new("tests/data/metadata_slices")).unwrap();
            let mut options = HashMap::from([
                (
                    HudiTableConfig::BasePath.as_ref().to_string(),
                    Url::from_directory_path(&dir).unwrap().to_string(),
                ),
                (
                    HudiTableConfig::BaseFileFormat.as_ref().to_string(),
                    "hfile".to_string(),
                ),
            ]);
            if windowed {
                // Ranged, because windows exist only when the file is not read whole,
                // and a one-byte budget plans a window per block without splitting one.
                options.insert(
                    crate::storage::reader::CONFIG_HFILE_WHOLE_READ_MAX_SIZE_MB.to_string(),
                    "0".to_string(),
                );
                options.insert(
                    crate::storage::reader::CONFIG_DFS_BUFFER_MAX_SIZE.to_string(),
                    "1".to_string(),
                );
            }
            let storage = Storage::new(
                Arc::new(HashMap::new()),
                Arc::new(HudiConfigs::new(options)),
            )?;
            let mut stream = HFileBaseFileReader::new(storage)
                .read_stream("files-multiblock.hfile", BaseFileReadOptions::default())
                .await?
                .into_stream();
            let mut rows = Vec::new();
            let mut batches = 0;
            while let Some(batch) = futures::StreamExt::next(&mut stream).await {
                let batch = batch?;
                batches += 1;
                let keys = batch
                    .column_by_name("key")
                    .expect("the metadata record key column")
                    .as_string::<i32>();
                let types = batch
                    .column_by_name("type")
                    .expect("the metadata record type column")
                    .as_primitive::<arrow_array::types::Int32Type>();
                for row in 0..batch.num_rows() {
                    rows.push((keys.value(row).to_string(), types.value(row)));
                }
            }
            rows.sort();
            Ok((rows, batches))
        }

        let (split, split_batches) = read(true).await?;
        let (whole, _) = read(false).await?;
        assert!(
            split_batches > 1,
            "this test is vacuous unless the budget splits the read: got \
             {split_batches} batch(es)"
        );
        assert_eq!(
            split, whole,
            "a windowed read must decode the same keys and types as a single-window read"
        );
        assert_eq!(whole.len(), 50, "the fixture holds fifty records");
        Ok(())
    }

    /// The size threshold picks the read strategy, and the two strategies are told
    /// apart by the requests they issue.
    ///
    /// Rows cannot separate them: whole and ranged return the same records over the
    /// same file, which is the point. So the assertion is on request counts, taken
    /// from a store that wraps the real one and counts what passes through.
    #[tokio::test]
    async fn the_size_threshold_picks_whole_or_ranged() -> crate::Result<()> {
        use crate::storage::counting::CountingObjectStore;

        fn configs_with(threshold_mb: &str) -> Arc<HudiConfigs> {
            let mut options = mdt_configs().as_options();
            options.insert(
                crate::storage::reader::CONFIG_HFILE_WHOLE_READ_MAX_SIZE_MB.to_string(),
                threshold_mb.to_string(),
            );
            Arc::new(HudiConfigs::new(options))
        }

        async fn read(
            threshold_mb: &str,
            known_file_size: Option<u64>,
        ) -> crate::Result<(Vec<String>, usize, usize)> {
            read_with(threshold_mb, known_file_size, None).await
        }

        async fn read_with(
            threshold_mb: &str,
            known_file_size: Option<u64>,
            key_predicate: Option<KeyPredicate>,
        ) -> crate::Result<(Vec<String>, usize, usize)> {
            let (store, counts) =
                CountingObjectStore::new(Arc::new(object_store::local::LocalFileSystem::new()));
            let storage = Storage::new_with_object_store(
                Url::parse(&metadata_table_uri()).unwrap(),
                store,
                configs_with(threshold_mb),
            );
            let reader = HFileBaseFileReader::new(storage);
            let mut options = BaseFileReadOptions::default();
            if let Some(size) = known_file_size {
                options = options.with_known_file_size(size);
            }
            if let Some(predicate) = key_predicate {
                options = options.with_key_predicate(predicate);
            }
            let batch = reader
                .read_data(MDT_FILES_COMPACTED_BASE_FILE, options)
                .await?;
            let keys: Vec<String> = batch
                .column_by_name("key")
                .expect("the metadata record key column")
                .as_string::<i32>()
                .iter()
                .flatten()
                .map(str::to_string)
                .collect();
            Ok((keys, counts.gets(), counts.heads()))
        }

        // The file's own size, so the below-threshold read can be asked for
        // without a size lookup.
        let file_size = std::fs::metadata(
            PathBuf::from(Url::parse(&metadata_table_uri()).unwrap().path())
                .join(MDT_FILES_COMPACTED_BASE_FILE),
        )
        .expect("the fixture base file")
        .len();

        // Below the default threshold, with the size already in hand: one read of
        // the whole file and nothing else, not even a metadata lookup.
        let (whole_keys, whole_gets, whole_heads) = read("50", Some(file_size)).await?;
        assert_eq!(
            (whole_gets, whole_heads),
            (1, 0),
            "a whole read of a known-size file is one request"
        );

        // Same threshold, size not supplied: one lookup to learn the size, then the
        // same single read.
        let (sized_keys, sized_gets, sized_heads) = read("50", None).await?;
        assert_eq!(
            (sized_gets, sized_heads),
            (1, 1),
            "learning the size costs one lookup and no extra read"
        );

        // Zero means never whole. The ranged open reads the trailer and the
        // load-on-open section before any data block, so it cannot come in at one.
        let (ranged_keys, ranged_gets, _) = read("0", Some(file_size)).await?;
        assert!(
            ranged_gets > whole_gets,
            "a ranged read issues more requests than a whole one, got {ranged_gets} \
             against {whole_gets}"
        );

        // A keyed read takes the same whole side here, because the fixture is far
        // below the keyed bound too. What is asserted is that naming keys does not
        // by itself force the ranged side.
        let (keyed_keys, keyed_gets, keyed_heads) = read_with(
            "50",
            Some(file_size),
            Some(KeyPredicate::Keys(vec![
                whole_keys.last().expect("the fixture returns keys").clone(),
            ])),
        )
        .await?;
        assert_eq!(
            (keyed_gets, keyed_heads),
            (1, 0),
            "a keyed read of a file below the keyed bound is still one request"
        );
        assert_eq!(keyed_keys.len(), 1, "the predicate must still filter");

        // Raising the file's apparent size past the keyed bound flips a keyed read
        // to ranged while a scan of the same file stays whole. This is the whole
        // point of the keyed bound, so it is asserted rather than assumed.
        let over_keyed_bound = crate::storage::reader::HFILE_WHOLE_READ_WITH_KEYS_MAX_SIZE + 1;
        let (_, keyed_big_gets, _) = read_with(
            "50",
            Some(over_keyed_bound),
            Some(KeyPredicate::Keys(vec![whole_keys.last().unwrap().clone()])),
        )
        .await?;
        let (_, scan_big_gets, _) = read("50", Some(over_keyed_bound)).await?;
        assert!(
            keyed_big_gets > 1,
            "past the keyed bound a keyed read must go ranged, got {keyed_big_gets} requests"
        );
        assert_eq!(
            scan_big_gets, 1,
            "a scan of the same size must stay whole; only the keyed bound moved"
        );

        assert!(
            !whole_keys.is_empty(),
            "the fixture must return rows, or the counts above prove nothing"
        );
        assert_eq!(whole_keys, sized_keys);
        assert_eq!(
            whole_keys, ranged_keys,
            "both strategies must return the same records; only their request \
             counts differ"
        );
        Ok(())
    }

    /// A prefix predicate returns every key carrying the prefix and nothing else.
    #[tokio::test]
    async fn a_prefix_predicate_returns_the_matching_keys() -> crate::Result<()> {
        let configs = mdt_configs();
        let storage = Storage::new(Arc::new(HashMap::new()), configs)?;
        let reader = HFileBaseFileReader::new(storage);

        let all = reader
            .read_data(
                MDT_FILES_COMPACTED_BASE_FILE,
                BaseFileReadOptions::default(),
            )
            .await?;
        let keys: Vec<String> = all
            .column_by_name("key")
            .unwrap()
            .as_string::<i32>()
            .iter()
            .flatten()
            .map(str::to_string)
            .collect();

        // The metadata table's partition keys share a "city=" head, so that
        // prefix matches a strict subset: the partitions but not
        // `__all_partitions__`.
        let expected: Vec<String> = keys
            .iter()
            .filter(|k| k.starts_with("city="))
            .cloned()
            .collect();
        assert!(
            !expected.is_empty() && expected.len() < keys.len(),
            "the prefix must match a strict subset, got {} of {}",
            expected.len(),
            keys.len()
        );

        let filtered = reader
            .read_data(
                MDT_FILES_COMPACTED_BASE_FILE,
                BaseFileReadOptions {
                    key_predicate: Some(KeyPredicate::Prefixes(vec!["city=".to_string()])),
                    ..Default::default()
                },
            )
            .await?;
        let got: Vec<String> = filtered
            .column_by_name("key")
            .unwrap()
            .as_string::<i32>()
            .iter()
            .flatten()
            .map(str::to_string)
            .collect();
        assert_eq!(got, expected);
        Ok(())
    }

    /// A count-only projection with a key predicate counts the matching records,
    /// not the file's.
    ///
    /// The row count comes from the trailer, which counts everything, so the fast
    /// path had to learn to stand aside. A caller asking how many of one key a
    /// file holds must not be told how many records the file holds.
    #[tokio::test]
    async fn a_count_only_projection_respects_the_predicate() -> crate::Result<()> {
        let configs = mdt_configs();
        let storage = Storage::new(Arc::new(HashMap::new()), configs)?;
        let reader = HFileBaseFileReader::new(storage);

        let count_all = reader
            .read_data(
                MDT_FILES_COMPACTED_BASE_FILE,
                BaseFileReadOptions {
                    projection: Some(vec![]),
                    ..Default::default()
                },
            )
            .await?
            .num_rows();
        assert!(count_all > 1, "the fixture must hold several records");

        let all = reader
            .read_data(
                MDT_FILES_COMPACTED_BASE_FILE,
                BaseFileReadOptions::default(),
            )
            .await?;
        let one = all
            .column_by_name("key")
            .unwrap()
            .as_string::<i32>()
            .value(0)
            .to_string();

        let count_one = reader
            .read_data(
                MDT_FILES_COMPACTED_BASE_FILE,
                BaseFileReadOptions {
                    projection: Some(vec![]),
                    key_predicate: Some(KeyPredicate::Keys(vec![one.clone()])),
                    ..Default::default()
                },
            )
            .await?
            .num_rows();
        assert_eq!(
            count_one, 1,
            "a count with a one-key predicate must be 1, not the file's {count_all}"
        );
        Ok(())
    }

    /// A key predicate set on the reader context reaches the base file reader
    /// through the engine.
    ///
    /// The other predicate tests call `read_data` directly, which is the trait
    /// method but not the path a caller takes. That left the plumbing from
    /// `ReaderContext` to `BaseFileReadOptions` untested, and it was in fact
    /// missing: nothing outside tests could set the predicate at all, so the
    /// criterion "a predicate reaches the reader through the reader context" was
    /// unmet while every predicate test passed. This is the test that fails when
    /// that route is broken.
    #[tokio::test]
    async fn a_key_predicate_on_the_reader_context_reaches_the_reader() -> crate::Result<()> {
        let configs = mdt_configs();
        let storage = Storage::new(Arc::new(HashMap::new()), configs.clone())?;

        async fn read(
            configs: Arc<HudiConfigs>,
            storage: Arc<Storage>,
            predicate: Option<KeyPredicate>,
        ) -> crate::Result<Vec<String>> {
            {
                let mut context = resolve_reader_context(
                    &configs,
                    /* has_log_files */ false,
                    Some(MDT_FILES_COMPACTED_BASE_FILE),
                )?;
                context.rebuild_record_context(MDT_FILES_PARTITION.to_string());
                context.key_predicate = predicate;
                let mut reader = HoodieFileGroupReader::new(
                    Arc::new(context),
                    storage,
                    InputSplit::new(
                        Some(MDT_FILES_COMPACTED_BASE_FILE.to_string()),
                        Some("20251220210130942".to_string()),
                        vec![],
                        MDT_FILES_PARTITION.to_string(),
                    ),
                    ReaderParameters::default(),
                    None,
                    None,
                )?;
                let batch = reader.read().await?;
                Ok(batch
                    .column_by_name("key")
                    .expect("the metadata record key column")
                    .as_string::<i32>()
                    .iter()
                    .flatten()
                    .map(str::to_string)
                    .collect::<Vec<String>>())
            }
        }

        let all = read(configs.clone(), storage.clone(), None).await?;
        assert!(
            all.len() > 1,
            "the fixture must hold several keys, or filtering cannot be observed"
        );

        let wanted = all.last().unwrap().clone();
        let filtered = read(
            configs.clone(),
            storage.clone(),
            Some(KeyPredicate::Keys(vec![wanted.clone()])),
        )
        .await?;
        assert_eq!(
            filtered,
            vec![wanted],
            "a predicate set on the reader context must reach the base file reader; \
             getting every key back means the engine dropped it"
        );
        Ok(())
    }

    /// Lays the HFile fixture down in a temp dir with `extra` appended to its
    /// `hoodie.properties`, and returns the table's URL.
    fn hfile_fixture_with(extra: &str) -> (tempfile::TempDir, String) {
        let src = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("../test/data/hfile_base_file_table");
        let tmp = tempfile::tempdir().unwrap();
        let dst = tmp.path();
        std::fs::create_dir_all(dst.join(".hoodie")).unwrap();
        for entry in std::fs::read_dir(&src).unwrap() {
            let entry = entry.unwrap();
            if entry.file_type().unwrap().is_file() {
                std::fs::copy(entry.path(), dst.join(entry.file_name())).unwrap();
            }
        }
        let props = std::fs::read_to_string(src.join(".hoodie/hoodie.properties")).unwrap();
        std::fs::write(
            dst.join(".hoodie/hoodie.properties"),
            format!("{props}{extra}"),
        )
        .unwrap();
        let url = url::Url::from_file_path(std::fs::canonicalize(dst).unwrap())
            .unwrap()
            .to_string();
        (tmp, url)
    }

    /// A CUSTOM merge mode whose payload class has no merger here, on a table
    /// whose base file is HFile -- the one combination where the CUSTOM refusal's
    /// own advice does not work.
    ///
    /// That refusal tells the reader to set file group reader version 1. For an
    /// HFile table that remedy fails, because version 1 has no HFile base file
    /// reader and its log decoder refuses HFile blocks. The test asserts both
    /// halves: that version 2 names HFile in its refusal, and that the fallback
    /// it used to recommend really does fail. Asserting only the message would
    /// pass even if the advice were sound.
    ///
    /// Reachable only since an HFile base file stopped being refused outside the
    /// metadata table; before that this table could not be constructed.
    #[tokio::test]
    async fn a_custom_merge_mode_without_a_merger_names_hfile_in_its_refusal() -> crate::Result<()>
    {
        use crate::file_group::reader::FileGroupReader;
        use crate::table::ReadOptions;

        const BASE: &str = "f0000000-0000-0000-0000-000000000001-0_0-1-1_20250101000000000.hfile";
        const LOG: &str = ".f0000000-0000-0000-0000-000000000001-0_20250101000000000.log.1_0-2-2";

        // A payload class no merger in this crate implements. Deliberately not
        // HoodieMetadataPayload, which resolves and so would not reach the gate.
        let (_tmp, url) = hfile_fixture_with(
            "\nhoodie.record.merge.mode=CUSTOM\n\
             hoodie.compaction.payload.class=com.example.NoSuchPayload\n",
        );

        let by_v2 = FileGroupReader::new_with_options(&url, crate::config::util::empty_options())
            .await?
            .read_file_slice_from_paths(BASE, vec![LOG], &ReadOptions::new())
            .await
            .expect_err("a CUSTOM mode with no merger must be refused");
        let msg = by_v2.to_string();
        assert!(
            msg.contains("CUSTOM record merge mode"),
            "the CUSTOM gate must be the one refusing, got: {msg}"
        );
        assert!(
            msg.contains("HFile"),
            "the refusal must say the version 1 remedy does not work for HFile, got: {msg}"
        );

        // The other half: the remedy the message qualifies really is a dead end.
        let by_v1 = FileGroupReader::new_with_options(
            &url,
            [("hoodie.read.file.group.reader.version", "1")],
        )
        .await?
        .read_file_slice_from_paths(BASE, vec![LOG], &ReadOptions::new())
        .await
        .expect_err("version 1 cannot read an HFile log block");
        assert!(
            by_v1.to_string().contains("HFile records"),
            "version 1 must fail on the HFile log block, got: {by_v1}"
        );
        Ok(())
    }

    /// Reads the HFile fixture under one merge mode and returns `(key, fare)`
    /// sorted, so the two modes can be compared on the same slice.
    ///
    /// The mode is written into a copy of the table's `hoodie.properties` rather
    /// than passed as a read option, because the merge mode is a property of the
    /// table and a read option does not override it.
    async fn read_hfile_slice(merge_mode: &str) -> crate::Result<Vec<(String, i64)>> {
        use crate::file_group::reader::FileGroupReader;
        use crate::table::ReadOptions;

        let src = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("../test/data/hfile_base_file_table");
        let tmp = tempfile::tempdir().unwrap();
        let dst = tmp.path();
        std::fs::create_dir_all(dst.join(".hoodie")).unwrap();
        for entry in std::fs::read_dir(&src).unwrap() {
            let entry = entry.unwrap();
            if entry.file_type().unwrap().is_file() {
                std::fs::copy(entry.path(), dst.join(entry.file_name())).unwrap();
            }
        }
        let props = std::fs::read_to_string(src.join(".hoodie/hoodie.properties"))
            .unwrap()
            .replace(
                "hoodie.record.merge.mode=COMMIT_TIME_ORDERING",
                &format!("hoodie.record.merge.mode={merge_mode}"),
            );
        assert!(
            props.contains(&format!("hoodie.record.merge.mode={merge_mode}")),
            "the fixture's properties must carry the mode under test, or this is vacuous"
        );
        std::fs::write(dst.join(".hoodie/hoodie.properties"), props).unwrap();

        let base_url = url::Url::from_file_path(std::fs::canonicalize(dst).unwrap())
            .unwrap()
            .to_string();
        let reader =
            FileGroupReader::new_with_options(&base_url, crate::config::util::empty_options())
                .await?;

        let batch = reader
            .read_file_slice_from_paths(
                "f0000000-0000-0000-0000-000000000001-0_0-1-1_20250101000000000.hfile",
                vec![".f0000000-0000-0000-0000-000000000001-0_20250101000000000.log.1_0-2-2"],
                &ReadOptions::new(),
            )
            .await?;

        let keys = batch
            .column_by_name("uuid")
            .expect("uuid column")
            .as_string::<i32>();
        let fares = batch
            .column_by_name("fare")
            .expect("fare column")
            .as_primitive::<arrow_array::types::Float64Type>();
        let mut got: Vec<(String, i64)> = (0..batch.num_rows())
            .map(|i| (keys.value(i).to_string(), fares.value(i) as i64))
            .collect();
        got.sort();
        Ok(got)
    }

    fn expected(uuid0003: i64) -> Vec<(String, i64)> {
        vec![
            ("uuid0001".to_string(), 11),
            ("uuid0002".to_string(), 12),
            ("uuid0003".to_string(), uuid0003),
            ("uuid0004".to_string(), 103),
            ("uuid0005".to_string(), 104),
            ("uuid0006".to_string(), 105),
        ]
    }

    /// An HFile base file and an HFile log block on a table that is **not** the
    /// metadata table, read through the ordinary public path under commit-time
    /// ordering.
    ///
    /// The log is the later commit, so it wins both overlapping keys. Values are
    /// asserted rather than a row count: dropping the log block would return four
    /// rows at fares 11-14, and losing the base would return four rows from
    /// `uuid0003`.
    #[tokio::test]
    async fn an_hfile_slice_merges_by_commit_time_outside_the_metadata_table() -> crate::Result<()>
    {
        assert_eq!(
            read_hfile_slice("COMMIT_TIME_ORDERING").await?,
            expected(102),
            "the log block is the later commit, so it must win both overlapping keys"
        );
        Ok(())
    }

    /// The same slice under event-time ordering, which must reach a *different*
    /// answer.
    ///
    /// `uuid0003` is the discriminator: the log's event time (1000) is older than
    /// the base's (5003), so the base value survives even though the log is the
    /// later commit. `uuid0004` goes the other way (9000 against 5004) so the
    /// test cannot pass by ignoring the log entirely.
    #[tokio::test]
    async fn an_hfile_slice_merges_by_event_time_outside_the_metadata_table() -> crate::Result<()> {
        assert_eq!(
            read_hfile_slice("EVENT_TIME_ORDERING").await?,
            expected(13),
            "uuid0003's log record has the older event time, so the base value must survive"
        );
        Ok(())
    }

    /// One metadata-table file slice: the partition it lives in, its base file if
    /// the slice has one, and the log files written after it.
    struct MdtSlice {
        label: &'static str,
        partition: &'static str,
        base_file: Option<&'static str>,
        log_files: &'static [&'static str],
        /// The base instant the slice is read as of.
        instant: &'static str,
    }

    /// Every metadata partition in the fixture that has a merge-reachable slice.
    ///
    /// `column_stats`, `partition_stats` and `record_index` compact to a base file
    /// written *after* all their log files, so their pre-compaction slice has log
    /// files and no base file at all. That is the sharper case for this change: an
    /// HFile log block that is skipped contributes nothing, so the read has no
    /// records and no schema to build an output from, and fails outright.
    const MDT_SLICES: &[MdtSlice] = &[
        MdtSlice {
            label: "files",
            partition: "files",
            base_file: Some("files-0000-0_0-955-2690_00000000000000000.hfile"),
            log_files: &[
                ".files-0000-0_20251220210108078.log.1_10-999-2838",
                ".files-0000-0_20251220210123755.log.1_3-1032-2950",
                ".files-0000-0_20251220210125441.log.1_5-1057-3024",
                ".files-0000-0_20251220210127080.log.1_3-1082-3100",
                ".files-0000-0_20251220210128625.log.1_5-1107-3174",
                ".files-0000-0_20251220210129235.log.1_3-1118-3220",
                ".files-0000-0_20251220210130911.log.1_3-1149-3338",
            ],
            instant: "00000000000000000",
        },
        MdtSlice {
            label: "column_stats",
            partition: "column_stats",
            base_file: None,
            log_files: &[
                ".col-stats-0000-0_20251220210108078.log.1_7-999-2835",
                ".col-stats-0000-0_20251220210123755.log.1_0-1032-2947",
                ".col-stats-0000-0_20251220210125441.log.1_2-1057-3021",
                ".col-stats-0000-0_20251220210127080.log.1_0-1082-3097",
                ".col-stats-0000-0_20251220210128625.log.1_2-1107-3171",
                ".col-stats-0000-0_20251220210129235.log.1_0-1118-3217",
                ".col-stats-0000-0_20251220210130911.log.1_0-1149-3335",
            ],
            instant: "00000000000000001",
        },
        MdtSlice {
            label: "partition_stats",
            partition: "partition_stats",
            base_file: None,
            log_files: &[
                ".partition-stats-0000-0_20251220210108078.log.1_9-999-2837",
                ".partition-stats-0000-0_20251220210123755.log.1_2-1032-2949",
                ".partition-stats-0000-0_20251220210125441.log.1_4-1057-3023",
                ".partition-stats-0000-0_20251220210127080.log.1_2-1082-3099",
                ".partition-stats-0000-0_20251220210128625.log.1_4-1107-3173",
                ".partition-stats-0000-0_20251220210129235.log.1_2-1118-3219",
                ".partition-stats-0000-0_20251220210130911.log.1_2-1149-3337",
            ],
            instant: "00000000000000003",
        },
        MdtSlice {
            label: "record_index",
            partition: "record_index",
            base_file: None,
            log_files: &[".record-index-0001-0_20251220210108078.log.1_0-999-2828"],
            instant: "00000000000000002",
        },
        MdtSlice {
            label: "secondary_index",
            partition: "secondary_index_rider_idx",
            base_file: Some("secondary-index-rider-idx-0002-0_2-1008-2876_00000000000000004.hfile"),
            log_files: &[".secondary-index-rider-idx-0002-0_20251220210125441.log.1_0-1057-3019"],
            instant: "00000000000000004",
        },
    ];

    /// Read one slice through the v2 reader, optionally without its log files.
    async fn read_mdt_slice(slice: &MdtSlice, with_logs: bool) -> crate::Result<RecordBatch> {
        let configs = mdt_configs();
        let storage = Storage::new(Arc::new(HashMap::new()), configs.clone())?;
        let logs: Vec<String> = if with_logs {
            slice
                .log_files
                .iter()
                .map(|f| format!("{}/{f}", slice.partition))
                .collect()
        } else {
            vec![]
        };
        // The base file's path is threaded in so the format resolves per path rather
        // than from config alone; the metadata table's base files are HFile.
        let base_file_path = slice.base_file.map(|b| format!("{}/{b}", slice.partition));
        let mut context =
            resolve_reader_context(&configs, !logs.is_empty(), base_file_path.as_deref())?;
        context.rebuild_record_context(slice.partition.to_string());
        let mut reader = HoodieFileGroupReader::new(
            Arc::new(context),
            storage,
            InputSplit::new(
                base_file_path.clone(),
                Some(slice.instant.to_string()),
                logs,
                slice.partition.to_string(),
            ),
            ReaderParameters::default(),
            None,
            None,
        )?;
        reader.read().await
    }

    /// The record keys of a metadata read.
    fn record_keys(batch: &RecordBatch) -> HashSet<String> {
        batch
            .column_by_name("key")
            .expect("the metadata record key column, decoded from the HFile's values")
            .as_string::<i32>()
            .iter()
            .flatten()
            .map(str::to_string)
            .collect()
    }

    /// Every metadata partition in the fixture reads through the v2 reader with its
    /// HFile log blocks contributing.
    ///
    /// One case per partition rather than the `files` partition alone, because a
    /// skipped log block is silent: the read succeeds and returns the base file's
    /// rows, so a `files`-only test would pass on four of the five partitions
    /// regardless.
    ///
    /// Two shapes of assertion, chosen by what the slice can prove:
    ///
    /// - **No base file** (`column_stats`, `partition_stats`, `record_index`). Every
    ///   record comes from a log block, so a skipped block leaves the read with no
    ///   records and no schema to build an output from, and it fails. Returning any
    ///   row is the assertion.
    /// - **Base file and log files** (`files`, `secondary_index`). A skipped block
    ///   makes the merged read identical to a base-file-only read, so the two must
    ///   differ.
    ///
    /// What this pins is that the blocks are **decoded and reach the merge**, not
    /// what the merge then does with them. The rule the merge applies is the
    /// metadata payload's own, asserted separately by
    /// `v2_folds_a_metadata_files_slice_like_the_metadata_table_reader`, which
    /// compares the folded values rather than the key set.
    #[tokio::test]
    async fn v2_reads_every_metadata_partition_with_its_log_blocks() -> crate::Result<()> {
        for slice in MDT_SLICES {
            let merged = read_mdt_slice(slice, true).await.map_err(|e| {
                StorageError::Creation(format!(
                    "{}: reading with log files failed: {e:?}",
                    slice.label
                ))
            })?;
            let merged_keys = record_keys(&merged);
            assert!(
                !merged_keys.is_empty(),
                "{}: a read whose log blocks contribute must return records",
                slice.label
            );

            match slice.base_file {
                None => {
                    // Nothing but log blocks, so the rows are proof on their own.
                }
                Some(_) => {
                    let base_only = read_mdt_slice(slice, false).await?;
                    assert_ne!(
                        (record_keys(&base_only), base_only.num_rows()),
                        (merged_keys.clone(), merged.num_rows()),
                        "{}: adding log files changed nothing, so the blocks were skipped",
                        slice.label
                    );
                }
            }
        }
        Ok(())
    }

    /// The `files` partition, checked against an oracle rather than against itself.
    ///
    /// `MetadataTableFileGroupReader` reads the same slice through a separate
    /// implementation that decodes to Rust structs instead of Arrow, so agreeing
    /// with it is evidence about the records rather than about this code path.
    /// It is the only metadata partition that reader serves, which is why the
    /// other four are covered by the weaker assertions above.
    #[tokio::test]
    async fn v2_reads_a_files_slice_matching_the_metadata_table_reader() -> crate::Result<()> {
        let slice = &MDT_SLICES[0];
        assert_eq!(slice.label, "files", "this case reads the files partition");

        let configs = mdt_configs();
        let storage = Storage::new(Arc::new(HashMap::new()), configs.clone())?;
        let mut fg = FileGroup::new(
            MDT_FILES_FILE_GROUP.to_string(),
            MDT_FILES_PARTITION.to_string(),
        );
        fg.add_base_file_from_name(slice.base_file.expect("the files slice has a base file"))?;
        fg.add_log_files_from_names(slice.log_files.iter().copied())?;
        let expected: HashSet<String> = MetadataTableFileGroupReader::new(configs, storage)
            .read_files_partition(
                fg.get_file_slice_as_of(MAX_INSTANT_TIME)
                    .expect("the file group has a slice"),
                &[],
            )
            .await?
            .into_keys()
            .collect();
        assert!(
            !expected.is_empty(),
            "the oracle must return keys, or the comparison is vacuous"
        );

        assert_eq!(
            record_keys(&read_mdt_slice(slice, true).await?),
            expected,
            "the v2 read of a files slice must return the metadata-table reader's keys"
        );
        Ok(())
    }

    /// A key predicate reaching the reader fetches fewer bytes for the same rows.
    ///
    /// The half no committed fixture could show. Every Avro-carrying HFile in the
    /// repo has one data block, and with one block a seek and a scan fetch
    /// identically — so the mutation `Some(_) => reader.data_block_entries()`
    /// passes against all of them. This drives a generated `record_index` HFile
    /// with 9 data blocks.
    ///
    /// Bytes, not rows, and deliberately: selection over-includes and
    /// `decode_window` filters afterwards, so a predicate read returns the same
    /// rows whether it fetched one block or nine. A row assertion here would
    /// reproduce the very hole this test exists to close.
    ///
    /// Both arms run the production path — `open` then `stream_from` — with the
    /// whole-read cache set to 0, documented as "always read in ranges", because
    /// `FetchCounts` exists only on a ranged source.
    ///
    #[tokio::test]
    async fn a_key_predicate_fetches_fewer_bytes_for_the_same_rows() -> crate::Result<()> {
        use crate::file_group::base_file::reader::{BaseFileReadOptions, KeyPredicate};
        use futures::TryStreamExt;
        let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("../test/data/metadata_multi_block_hfile");
        let name = "record-index-0009-0_10-71-248_20260830075534042.hfile".to_string();
        let size = std::fs::metadata(dir.join(&name)).unwrap().len();

        let base_url = url::Url::from_directory_path(std::fs::canonicalize(&dir).unwrap()).unwrap();
        let configs = Arc::new(HudiConfigs::new([
            (
                HudiTableConfig::BasePath.as_ref().to_string(),
                base_url.to_string(),
            ),
            (
                crate::storage::reader::CONFIG_HFILE_WHOLE_READ_MAX_SIZE_MB.to_string(),
                "0".to_string(),
            ),
        ]));
        let storage = Storage::new(Arc::new(HashMap::new()), configs)?;
        let base = HFileBaseFileReader::new(storage);

        // Measure one production read: open, hand the reader to the read, and
        // read the counts back through the fetcher clone that outlives it.
        async fn measure(
            base: &HFileBaseFileReader,
            name: &str,
            options: BaseFileReadOptions,
        ) -> crate::Result<(u64, usize, usize)> {
            let reader = base.open(name, &options).await?;
            let blocks = reader.data_block_entries().len();
            let counts = reader
                .reads_handle()
                .expect("the whole-read cache is 0, so this reader is ranged");
            let stream = base.stream_from(reader, name, &options)?;
            let rows: usize = stream
                .try_fold(0usize, |n, b| async move { Ok(n + b.num_rows()) })
                .await?;
            Ok((counts.reads().bytes(), rows, blocks))
        }

        // A key the file certainly holds, taken from the file rather than
        // guessed: a miss would also fetch few bytes, for the wrong reason.
        let (scan_bytes, scan_rows, blocks) =
            measure(&base, &name, BaseFileReadOptions::default()).await?;
        let sample = base
            .read_data(&name, BaseFileReadOptions::default())
            .await?;
        use arrow_array::Array;
        let keys = sample
            .column_by_name("key")
            .and_then(|c| {
                c.as_any()
                    .downcast_ref::<arrow_array::StringArray>()
                    .cloned()
            })
            .expect("record_index rows carry a key column");
        let wanted = keys.value(keys.len() / 2).to_string();

        let (seek_bytes, seek_rows, _) = measure(
            &base,
            &name,
            BaseFileReadOptions {
                key_predicate: Some(KeyPredicate::Keys(vec![wanted.clone()])),
                ..Default::default()
            },
        )
        .await?;

        assert!(
            blocks > 1,
            "the fixture must be multi-block or this test cannot fail: {blocks} in {size} bytes"
        );
        assert_eq!(seek_rows, 1, "the predicate must select exactly {wanted}");
        assert!(
            scan_rows > seek_rows,
            "the scan must return more than the one matching row: {scan_rows}"
        );
        assert!(
            seek_bytes < scan_bytes,
            "a one-key predicate must fetch fewer bytes than a scan of {blocks} blocks: \
             {seek_bytes} vs {scan_bytes}"
        );
        println!(
            "MEASURED blocks={blocks} bytes={seek_bytes}/{scan_bytes} rows={seek_rows}/{scan_rows}"
        );
        Ok(())
    }

    /// The `bloom_filters` slice, base HFile plus the log written after it.
    const MDT_BLOOM_DIR: &str = "../test/data/metadata_bloom_filters_slice";
    const MDT_BLOOM_BASE: &str = "bloom-filters-0000-0_0-51-131_20260830093346422.hfile";
    const MDT_BLOOM_LOG: &str = ".bloom-filters-0000-0_20260830093352285.log.1_0-109-277";

    /// A `bloom_filters` slice reads through the metadata table's CUSTOM merge
    /// path, base and log both contributing.
    ///
    /// The corpus reaches `files`, `partition_stats` and `secondary_index`; this
    /// is the fourth metadata partition type and nothing covered it. It is the one
    /// whose payload is a raw byte buffer rather than a struct of scalars, so a
    /// decoder that works for the others can still fail here — which is what makes
    /// the type worth a fixture rather than a unit test on constructed records.
    ///
    /// The row count is what gives the log side weight: the base holds 2 records
    /// and the log 2 more, under keys the base does not have, so a read that
    /// dropped the log would return 2 and every other assertion here would still
    /// hold.
    ///
    /// It does **not** pin partition routing. Rebuilding the record context for
    /// `files` instead of `bloom_filters` changes nothing observable here — the
    /// `type` field is decoded from the record, not chosen by the context — so
    /// that is measured, not asserted.
    #[tokio::test]
    async fn v2_reads_a_bloom_filters_slice() -> crate::Result<()> {
        use arrow_array::Array;
        let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join(MDT_BLOOM_DIR);
        let uri = url::Url::from_directory_path(std::fs::canonicalize(&dir).unwrap()).unwrap();
        let configs = Arc::new(HudiConfigs::new([
            (
                HudiTableConfig::BasePath.as_ref().to_string(),
                uri.to_string(),
            ),
            (
                HudiTableConfig::BaseFileFormat.as_ref().to_string(),
                "hfile".to_string(),
            ),
            (
                HudiReadConfig::EndTimestamp.as_ref().to_string(),
                MAX_INSTANT_TIME.to_string(),
            ),
            (
                HudiTableConfig::RecordKeyFields.as_ref().to_string(),
                "key".to_string(),
            ),
            (
                HudiTableConfig::PopulatesMetaFields.as_ref().to_string(),
                "false".to_string(),
            ),
            ("hoodie.record.merge.mode".to_string(), "CUSTOM".to_string()),
            (
                "hoodie.record.merge.strategy.id".to_string(),
                "00000000-0000-0000-0000-000000000000".to_string(),
            ),
            (
                "hoodie.compaction.payload.class".to_string(),
                "org.apache.hudi.metadata.HoodieMetadataPayload".to_string(),
            ),
        ]));
        let storage = Storage::new(Arc::new(HashMap::new()), configs.clone())?;
        let mut context = resolve_reader_context(&configs, true, Some("hfile"))?;
        // The partition drives which metadata payload the merger decodes into.
        context.rebuild_record_context("bloom_filters".to_string());
        let mut reader = HoodieFileGroupReader::new(
            Arc::new(context),
            storage,
            InputSplit::new(
                Some(MDT_BLOOM_BASE.to_string()),
                Some(MAX_INSTANT_TIME.to_string()),
                vec![MDT_BLOOM_LOG.to_string()],
                String::new(),
            ),
            ReaderParameters::default(),
            None,
            None,
        )?;
        let batch = reader.read().await?;

        assert_eq!(
            batch.num_rows(),
            4,
            "2 records from the base and 2 from the log; a dropped log side reads 2"
        );
        let types = batch
            .column_by_name("type")
            .expect("a metadata record carries its partition type")
            .as_primitive::<arrow_array::types::Int32Type>();
        // 4 is BLOOM_FILTERS in Hudi's metadata record schema.
        assert!(
            (0..batch.num_rows()).all(|i| types.value(i) == 4),
            "every record must be a bloom-filter record"
        );

        let bloom = batch
            .column_by_name("BloomFilterMetadata")
            .expect("the bloom-filter column")
            .as_struct();
        let filters = bloom
            .column_by_name("bloomFilter")
            .expect("the filter payload")
            .as_binary::<i32>();
        let non_empty = (0..batch.num_rows())
            .filter(|i| !filters.is_null(*i) && !filters.value(*i).is_empty())
            .count();
        assert_eq!(
            non_empty,
            batch.num_rows(),
            "every bloom-filter record must carry a non-empty filter buffer"
        );
        Ok(())
    }
}