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.
 */
//! This module is responsible for Hudi table APIs.
//!
//! It provides a quick entry point for reading Hudi table metadata and data,
//! facilitating adaptation and compatibility across various engines.
//!
//! **Example**
//! 1. create hudi table
//! ```rust
//! use url::Url;
//! use hudi_core::table::Table;
//!
//! pub async fn test() {
//!     let base_uri = Url::from_file_path("/tmp/hudi_data").unwrap();
//!     let hudi_table = Table::new(base_uri.path()).await.unwrap();
//! }
//! ```
//! 2. get hudi table schema(arrow_schema::Schema)
//! ```rust
//! use url::Url;
//! use hudi_core::table::Table;
//!
//! pub async fn test() {
//!     use arrow_schema::Schema;
//!     let base_uri = Url::from_file_path("/tmp/hudi_data").unwrap();
//!     let hudi_table = Table::new(base_uri.path()).await.unwrap();
//!     let schema = hudi_table.get_schema().await.unwrap();
//! }
//! ```
//! 3. read hudi table
//! ```rust
//! use url::Url;
//! use hudi_core::table::{ReadOptions, Table};
//!
//! pub async fn test() {
//!     let base_uri = Url::from_file_path("/tmp/hudi_data").unwrap();
//!     let hudi_table = Table::new(base_uri.path()).await.unwrap();
//!     let record_read = hudi_table.read(&ReadOptions::new()).await.unwrap();
//! }
//! ```
//! 4. get file slice
//!    Users can obtain metadata to customize reading methods, read in batches, perform parallel reads, and more.
//! ```rust
//! use url::Url;
//! use hudi_core::table::{ReadOptions, Table};
//! use hudi_core::storage::util::parse_uri;
//! use hudi_core::storage::util::join_url_segments;
//!
//! pub async fn test() {
//!     let base_uri = Url::from_file_path("/tmp/hudi_data").unwrap();
//!     let hudi_table = Table::new(base_uri.path()).await.unwrap();
//!     let flat_slices = hudi_table
//!             .get_file_slices(&ReadOptions::new())
//!             .await.unwrap();
//!     let file_slices = hudi_core::util::collection::split_into_chunks(flat_slices, 2);
//!     // define every parquet task reader how many slice
//!     let mut parquet_file_groups: Vec<Vec<String>> = Vec::new();
//!         for file_slice_vec in file_slices {
//!             let file_group_vec = file_slice_vec
//!                 .iter()
//!                 .filter_map(|f| {
//!                     // None when the slice's records live entirely in log
//!                     // files, so there is no base file to read.
//!                     let relative_path = f.base_file_relative_path().unwrap()?;
//!                     let url = join_url_segments(&base_uri, &[relative_path.as_str()]).unwrap();
//!                     Some(url.path().to_string())
//!                 })
//!                 .collect();
//!             parquet_file_groups.push(file_group_vec)
//!         }
//! }
//! ```

pub mod builder;
pub mod file_pruner;
pub(crate) mod fs_view;
mod listing;
pub mod partition;
mod validation;

pub use crate::config::read_options::{QueryType, ReadOptions};

use crate::Result;
use crate::config::HudiConfigs;
use crate::config::internal::HudiInternalConfig;
use crate::config::read::HudiReadConfig;
use crate::config::table::HudiTableConfig::PartitionFields;
use crate::config::table::{BaseFileFormatValue, HudiTableConfig, TableTypeValue};
use crate::error::CoreError;
use crate::expr::filter::{Filter, validate_fields_against_schemas};
use crate::file_group::file_slice::FileSlice;
use crate::file_group::reader::FileGroupReader;
use crate::file_group::reader_v2::reader_context::CompletionGateInputs;
use crate::keygen::is_timestamp_based_keygen;
use crate::metadata::METADATA_TABLE_PARTITION_FIELD;
use crate::metadata::commit::HoodieCommitMetadata;
use crate::metadata::meta_field::MetaField;
use crate::schema::resolver::{
    resolve_avro_schema, resolve_avro_schema_with_meta_fields, resolve_data_schema, resolve_schema,
};
use crate::statistics::estimator::FileStatsEstimator;
use crate::table::builder::TableBuilder;
use crate::table::file_pruner::FilePruner;
use crate::table::fs_view::FileSystemView;
use crate::table::partition::{PartitionPruner, project_partition_schema};
use crate::timeline::util::format_timestamp;
use crate::timeline::{EARLIEST_START_TIMESTAMP, Timeline};
use arrow::record_batch::RecordBatch;
use arrow_schema::{Field, Schema};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use tokio::sync::OnceCell;
use url::Url;

/// The main struct that provides table APIs for interacting with a Hudi table.
#[derive(Debug)]
pub struct Table {
    pub hudi_configs: Arc<HudiConfigs>,
    pub storage_options: Arc<HashMap<String, String>>,
    pub timeline: Timeline,
    pub file_system_view: FileSystemView,
    /// Cached metadata table instance, lazily initialized on first use.
    /// Only populated when metadata table is enabled (v8+ with files partition).
    /// Shared across clones via `Arc` so all scan() calls reuse the same instance.
    cached_metadata_table: Arc<OnceCell<Table>>,
    /// Cached file stats estimator. Materialized on first successful call to
    /// [`Table::get_or_init_estimator`]. Failed or inapplicable attempts do not
    /// populate the cache, allowing later calls with newer timestamps to retry.
    cached_estimator: Arc<OnceCell<FileStatsEstimator>>,
}

impl Clone for Table {
    fn clone(&self) -> Self {
        Self {
            hudi_configs: self.hudi_configs.clone(),
            storage_options: self.storage_options.clone(),
            timeline: self.timeline.clone(),
            file_system_view: self.file_system_view.clone(),
            cached_metadata_table: self.cached_metadata_table.clone(),
            cached_estimator: self.cached_estimator.clone(),
        }
    }
}

/// One tick below `instant_time`, as an instant string of the same shape.
///
/// Turns an exclusive start bound into one that admits `instant_time` itself.
///
/// Stepped in the *time* domain, not as an integer. An instant time is a
/// `yyyyMMddHHmmss[SSS]` string, so integer arithmetic borrows across its field
/// boundaries: `20250713010500000 - 1` is `20250713010499999`, whose `ss` field
/// is 99. That value sorts correctly, so every range comparison looked right,
/// and it is not a time, so `Instant::parse_datetime` rejected it — an
/// incremental read whose earliest admitted commit landed on a whole second
/// died with "Invalid epoch millis". The bound is read back by several
/// consumers and at least one of them parses it, so it has to be both.
///
/// A value that is not a date — the epoch-millis timestamps a metadata table
/// uses, the bootstrap sentinels — has no fields to borrow across and keeps the
/// integer decrement. Zero and unparseable values are returned unchanged, which
/// keeps the bound no narrower than it was.
fn instant_time_minus_one(instant_time: &str) -> String {
    // Round-tripping is what tells a date-formatted instant from an epoch-millis
    // one: both parse, only the former renders back to itself.
    const FORMATS: [&str; 2] = ["%Y%m%d%H%M%S%3f", "%Y%m%d%H%M%S"];
    if let Ok(dt) = crate::timeline::instant::Instant::parse_datetime(instant_time, "UTC") {
        for format in FORMATS {
            if dt.format(format).to_string() == instant_time
                && let Some(stepped) = dt.checked_sub_signed(chrono::TimeDelta::milliseconds(1))
            {
                let rendered = stepped.format(format).to_string();
                // A step that changes the string's width would change how it
                // sorts; leave those to the integer path.
                if rendered.len() == instant_time.len() {
                    return rendered;
                }
            }
        }
    }
    match instant_time.parse::<u64>() {
        Ok(0) | Err(_) => instant_time.to_string(),
        Ok(n) => format!("{:0width$}", n - 1, width = instant_time.len()),
    }
}

impl Table {
    /// Get or initialize the cached metadata table instance.
    ///
    /// Returns `Ok(&Table)` if metadata table is successfully created or was already cached.
    /// The instance is created once and reused across all subsequent calls.
    pub(crate) async fn get_or_init_metadata_table(&self) -> Result<&Table> {
        self.cached_metadata_table
            .get_or_try_init(|| async {
                log::debug!("Initializing cached metadata table instance");
                self.new_metadata_table().await
            })
            .await
    }

    /// Get or initialize the cached `FileStatsEstimator` for this **data table**.
    ///
    /// This is the single point where eligibility for footer-based stats
    /// estimation is validated. Returns `None` when:
    /// - The base file format is not Parquet (e.g., HFile metadata tables).
    /// - No sample base-file path can be derived from commit metadata at or
    ///   before `sample_at_timestamp`.
    /// - The parquet footer read fails for the sample path.
    ///
    /// The cache is only materialized on successful initialization. This avoids
    /// pinning a "no sample yet" state from an early timestamp.
    ///
    /// Not intended for use on metadata tables. MDTs use HFile, which already
    /// short-circuits via the format check below; this is documented for the
    /// reader, not enforced via a separate flag.
    pub(crate) async fn get_or_init_estimator(
        &self,
        sample_at_timestamp: &str,
    ) -> Option<&FileStatsEstimator> {
        if let Some(estimator) = self.cached_estimator.get() {
            return Some(estimator);
        }

        let configured_base_file_format =
            self.file_system_view.configured_base_file_format().ok()?;
        if configured_base_file_format
            .as_ref()
            .is_some_and(|format| !matches!(format, BaseFileFormatValue::Parquet))
        {
            return None;
        }

        let path = self
            .sample_base_file_path_at_or_before(sample_at_timestamp)
            .await?;
        if !BaseFileFormatValue::Parquet.matches_extension(&path) {
            return None;
        }

        self.cached_estimator
            .get_or_try_init(|| async {
                FileStatsEstimator::from_parquet_footer(&self.file_system_view.storage, &path).await
            })
            .await
            .map(Some)
            .unwrap_or_else(|e| {
                log::warn!(
                    "Failed to initialize file stats estimator from sample base file '{path}' \
                     (as-of timestamp: '{sample_at_timestamp}'): {e}"
                );
                None
            })
    }

    /// Sample one base file path active at or before the given timestamp.
    ///
    /// Walks completed commits ≤ `timestamp` in reverse (latest-first) and
    /// returns the first recorded base-file path. This handles MOR tables
    /// whose latest commit may be a delta commit with only log files —
    /// in that case we fall back to earlier commits that wrote base files.
    ///
    /// Format-agnostic by design: callers that require a specific format
    /// must validate before invoking. Returns `None` if no commit in range
    /// has a recorded base-file path.
    pub(crate) async fn sample_base_file_path_at_or_before(
        &self,
        timestamp: &str,
    ) -> Option<String> {
        let commits = self
            .timeline
            .get_completed_instants_at_or_before(timestamp)
            .ok()?;
        for commit in commits.iter().rev() {
            let Ok(metadata) = self.timeline.get_instant_metadata(commit).await else {
                continue;
            };
            let Ok(parsed) = HoodieCommitMetadata::from_json_map(&metadata) else {
                continue;
            };
            // Pick a stable sample path so estimator-derived stats are reproducible.
            // `partition_to_write_stats` is a HashMap, so raw iteration order is
            // non-deterministic across process runs.
            if let Some(path) = parsed.iter_base_file_paths().min() {
                return Some(path);
            }
        }
        None
    }

    /// Create hudi table by base_uri
    pub async fn new(base_uri: &str) -> Result<Self> {
        TableBuilder::from_base_uri(base_uri).build().await
    }

    /// Create hudi table with options
    pub async fn new_with_options<I, K, V>(base_uri: &str, options: I) -> Result<Self>
    where
        I: IntoIterator<Item = (K, V)>,
        K: AsRef<str>,
        V: Into<String>,
    {
        TableBuilder::from_base_uri(base_uri)
            .with_options(options)
            .build()
            .await
    }

    pub fn hudi_options(&self) -> HashMap<String, String> {
        self.hudi_configs.as_options()
    }

    pub fn storage_options(&self) -> HashMap<String, String> {
        self.storage_options.as_ref().clone()
    }

    #[cfg(feature = "datafusion")]
    pub fn register_storage(
        &self,
        runtime_env: Arc<datafusion::execution::runtime_env::RuntimeEnv>,
    ) {
        self.timeline
            .storage
            .register_object_store(runtime_env.clone());
        self.file_system_view
            .storage
            .register_object_store(runtime_env.clone());
    }

    pub fn base_url(&self) -> Url {
        let err_msg = format!("{:?} is missing or invalid.", HudiTableConfig::BasePath);
        self.hudi_configs
            .get(HudiTableConfig::BasePath)
            .expect(&err_msg)
            .to_url()
            .expect(&err_msg)
    }

    pub fn table_name(&self) -> String {
        let err_msg = format!("{:?} is missing or invalid.", HudiTableConfig::TableName);
        self.hudi_configs
            .get(HudiTableConfig::TableName)
            .expect(&err_msg)
            .into()
    }

    pub fn table_type(&self) -> String {
        let err_msg = format!("{:?} is missing or invalid.", HudiTableConfig::TableType);
        self.hudi_configs
            .get(HudiTableConfig::TableType)
            .expect(&err_msg)
            .into()
    }

    pub fn is_mor(&self) -> bool {
        self.table_type() == TableTypeValue::MergeOnRead.as_ref()
    }

    fn is_base_file_only(&self, options: &ReadOptions) -> crate::Result<bool> {
        Ok(!self.is_mor() || options.is_read_optimized()?)
    }

    pub fn timezone(&self) -> String {
        self.hudi_configs
            .get_or_default(HudiTableConfig::TimelineTimezone)
            .into()
    }

    /// Get the latest Avro schema string of the table, without Hudi meta fields (`_hoodie_*`).
    ///
    /// The implementation looks for the schema in the following order:
    /// 1. Timeline commit metadata.
    /// 2. `hoodie.properties` file's [HudiTableConfig::CreateSchema].
    pub async fn get_schema_in_avro_str(&self) -> Result<String> {
        self.get_schema_in_avro_str_inner(false).await
    }

    /// Get the latest Avro schema string of the table, with Hudi meta fields (`_hoodie_*`)
    /// prepended.
    pub async fn get_schema_in_avro_str_with_meta_fields(&self) -> Result<String> {
        self.get_schema_in_avro_str_inner(true).await
    }

    async fn get_schema_in_avro_str_inner(&self, includes_meta_fields: bool) -> Result<String> {
        if includes_meta_fields {
            resolve_avro_schema_with_meta_fields(self).await
        } else {
            resolve_avro_schema(self).await
        }
    }

    /// Get the latest [arrow_schema::Schema] of the table, without Hudi meta fields
    /// (`_hoodie_*`).
    ///
    /// The implementation looks for the schema in the following order:
    /// 1. Timeline commit metadata.
    /// 2. Base file schema.
    /// 3. `hoodie.properties` file's [HudiTableConfig::CreateSchema].
    pub async fn get_schema(&self) -> Result<Schema> {
        self.get_schema_inner(false).await
    }

    /// Get the latest [arrow_schema::Schema] of the table, with Hudi meta fields (`_hoodie_*`)
    /// prepended.
    pub async fn get_schema_with_meta_fields(&self) -> Result<Schema> {
        self.get_schema_inner(true).await
    }

    /// The schema a file slice's records actually carry.
    ///
    /// Meta fields are only in the data when the table populates them; handing
    /// a reader a schema that names columns the files do not have fails the
    /// evolution step rather than helping it.
    pub(crate) async fn data_schema_for_read(&self) -> Result<Schema> {
        let populates_meta_fields: bool = self
            .hudi_configs
            .get_or_default(HudiTableConfig::PopulatesMetaFields)
            .into();
        self.get_schema_inner(populates_meta_fields).await
    }

    async fn get_schema_inner(&self, includes_meta_fields: bool) -> Result<Schema> {
        if includes_meta_fields {
            resolve_schema(self).await
        } else {
            resolve_data_schema(self).await
        }
    }

    /// Get the latest partition [arrow_schema::Schema] of the table.
    ///
    /// For metadata tables, returns a schema with a single `partition` field
    /// typed as [arrow_schema::DataType::Utf8], since metadata tables use a single partition
    /// column to identify partitions like "files", "column_stats", etc.
    ///
    /// For regular tables, returns the partition fields with their actual data types
    /// derived from the table schema.
    pub async fn get_partition_schema(&self) -> Result<Schema> {
        if self.is_metadata_table() {
            return Ok(Schema::new(vec![Field::new(
                METADATA_TABLE_PARTITION_FIELD,
                arrow_schema::DataType::Utf8,
                false,
            )]));
        }

        // Timestamp-based keygen: the source field is transformed into partition path
        // strings, so use a single _hoodie_partition_path field.
        if is_timestamp_based_keygen(&self.hudi_configs)? {
            return Ok(Schema::new(vec![Field::new(
                MetaField::PartitionPath.as_ref(),
                arrow_schema::DataType::Utf8,
                false,
            )]));
        }

        let partition_field_names: Vec<String> =
            self.hudi_configs.get_or_default(PartitionFields).into();

        let schema = self.get_schema().await?;
        project_partition_schema(&schema, &partition_field_names)
    }

    /// Get the [Timeline] of the table.
    pub fn get_timeline(&self) -> &Timeline {
        &self.timeline
    }

    /// Get the [FileSlice]s the read targets, dispatching on `options.query_type`.
    ///
    /// - [`QueryType::Snapshot`]: returns slices visible at `options.as_of_timestamp`,
    ///   defaulting to the latest commit. `options.filters` drive both partition
    ///   pruning and file-level stats pruning (when min/max stats are available).
    /// - [`QueryType::Incremental`]: returns slices changed in
    ///   (`options.start_timestamp`, `options.end_timestamp`], defaulting to earliest
    ///   and latest respectively. `options.filters` drive partition pruning only;
    ///   data-column filters do not prune files at planning time.
    ///
    /// Returns an empty vector when the table has no commits.
    ///
    /// To bucket the result for parallel reads, use
    /// [`crate::util::collection::split_into_chunks`] or your engine's preferred
    /// partitioning policy.
    pub async fn get_file_slices(&self, options: &ReadOptions) -> Result<Vec<FileSlice>> {
        let prepared = self.prepare_reader_options(options)?;
        let base_file_only = self.is_base_file_only(&prepared)?;
        match prepared.query_type()? {
            QueryType::Snapshot => {
                let Some(timestamp) = prepared.end_timestamp() else {
                    return Ok(Vec::new());
                };
                self.get_file_slices_inner(timestamp, &prepared.filters, base_file_only)
                    .await
            }
            QueryType::Incremental => {
                let (Some(start), Some(end)) =
                    (prepared.start_timestamp(), prepared.end_timestamp())
                else {
                    return Ok(Vec::new());
                };
                self.get_file_slices_between_inner(start, end, &prepared.filters, base_file_only)
                    .await
            }
        }
    }

    async fn get_file_slices_inner(
        &self,
        timestamp: &str,
        filters: &[Filter],
        base_file_only: bool,
    ) -> Result<Vec<FileSlice>> {
        let timeline_view = self.timeline.create_view_as_of(timestamp).await?;

        let partition_schema = self.get_partition_schema().await?;
        // Validate against the meta-inclusive schema so filters on Hudi meta fields
        // (e.g. `_hoodie_record_key`) are accepted — those columns are present in
        // returned batches and the row-level mask applies them. The pruners are
        // tolerant of meta-field filters: PartitionPruner ignores non-partition
        // columns, and FilePruner skips columns without stats.
        let table_schema = self.get_schema_with_meta_fields().await?;
        validate_fields_against_schemas(filters, [&table_schema, &partition_schema])?;

        let partition_pruner =
            PartitionPruner::new(filters, &partition_schema, self.hudi_configs.as_ref())?;

        // File-level stats pruning using base file Parquet footers is only safe
        // when log files cannot introduce records that contradict the base file's
        // min/max stats — i.e., COW tables or MOR read-optimized mode.
        let file_pruner = if base_file_only {
            FilePruner::new(filters, &table_schema, &partition_schema)?
        } else {
            FilePruner::empty()
        };

        // Use cached metadata table instance if enabled
        let metadata_table = if self.is_metadata_table_enabled() {
            match self.get_or_init_metadata_table().await {
                Ok(mdt) => Some(mdt),
                Err(e) => {
                    log::warn!(
                        "Failed to create metadata table, falling back to storage listing: {e}"
                    );
                    None
                }
            }
        } else {
            None
        };

        // Estimator-backed metadata enrichment is used by both MDT-backed loading
        // and fallback storage listing.
        let estimator = self.get_or_init_estimator(timestamp).await;

        // Built here rather than inside the view: it needs the data table's
        // timeline as well as the metadata table's, and only this scope holds
        // both.
        let valid_instants = match metadata_table {
            Some(mdt) => Some(self.valid_instant_timestamps(mdt).await?),
            None => None,
        };
        let metadata_listing = match (metadata_table, valid_instants.as_ref()) {
            (Some(table), Some(valid_instants)) => Some(crate::table::fs_view::MetadataListing {
                table,
                valid_instants,
            }),
            _ => None,
        };

        let mut file_slices = self
            .file_system_view
            .get_file_slices(
                &partition_pruner,
                &file_pruner,
                &table_schema,
                &timeline_view,
                metadata_listing,
                estimator,
            )
            .await?;

        if base_file_only {
            for fs in &mut file_slices {
                fs.log_files.clear();
            }
        }
        Ok(file_slices)
    }

    async fn get_file_slices_between_inner(
        &self,
        start_timestamp: &str,
        end_timestamp: &str,
        filters: &[Filter],
        base_file_only: bool,
    ) -> Result<Vec<FileSlice>> {
        // Seed the cached estimator from a sample base file at or before
        // end_timestamp so the file group builder can populate FileMetadata
        // (size, byte_size, num_records) on each base file.
        let estimator = self.get_or_init_estimator(end_timestamp).await;

        let file_groups = self
            .timeline
            .get_file_groups_between(Some(start_timestamp), Some(end_timestamp), estimator)
            .await?;

        // The commits in range say *which* file groups changed; they are not a
        // reliable source for the slices themselves. A delta commit that only
        // appends names no base file, and a commit that writes one does not know
        // about log files appended after it — so a slice assembled from range
        // metadata alone is missing either its base file or its newest logs.
        //
        // Read the slice that is live at the end of the range instead, exactly
        // as a snapshot read at `end_timestamp` would. The commit-time mask then
        // narrows its rows back to the window.
        let touched: HashSet<(&str, &str)> = file_groups
            .iter()
            .map(|file_group| {
                (
                    file_group.partition_path.as_str(),
                    file_group.file_id.as_str(),
                )
            })
            .collect();

        let mut file_slices: Vec<FileSlice> = self
            .get_file_slices_inner(end_timestamp, filters, base_file_only)
            .await?
            .into_iter()
            .filter(|slice| touched.contains(&(slice.partition_path.as_str(), slice.file_id())))
            .collect();

        if base_file_only {
            for fs in &mut file_slices {
                fs.log_files.clear();
            }
        }
        Ok(file_slices)
    }

    /// Create a [FileGroupReader] using the [Table]'s Hudi configs.
    ///
    /// `read_options.hudi_options` override table-level Hudi configs
    /// (last-writer-wins). `extra_storage_overrides` override table-level
    /// storage options (cloud credentials, endpoints, etc).
    ///
    /// When `read_options` is `Some`, timestamps are resolved into the
    /// `StartTimestamp` / `EndTimestamp` that [`FileGroupReader`] needs for
    /// log-scan bounds and commit-time filtering — the same normalization
    /// that [`Table::read`] performs internally.
    ///
    /// The returned reader also carries the table's current data schema, which is
    /// why this is async: resolving it reads the timeline. Without it a reader
    /// falls back to whatever schema the base file happens to carry, and a base
    /// file written before a column was widened or added would force newer
    /// records back into that older shape — wrong values, with no error. Only a
    /// caller with no timeline at all (the cxx bridge, which is handed paths) is
    /// left with that fallback.
    pub async fn create_file_group_reader_with_options<S, K, V>(
        &self,
        read_options: Option<&ReadOptions>,
        extra_storage_overrides: S,
    ) -> Result<FileGroupReader>
    where
        S: IntoIterator<Item = (K, V)>,
        K: AsRef<str>,
        V: Into<String>,
    {
        let hudi_opts: HashMap<String, String> = match read_options {
            Some(opts) => self.prepare_reader_options(opts)?.hudi_options,
            None => HashMap::new(),
        };
        let reader = self.build_file_group_reader(hudi_opts, extra_storage_overrides)?;
        self.finish_reader(reader).await
    }

    /// Carry this table's per-read state onto a freshly built reader.
    ///
    /// Every read path needs the same two steps, and a path performing only some
    /// of them fails silently rather than loudly: without the data schema an
    /// evolved table reads against the base file's, and without the gate inputs
    /// the log scan admits uncommitted blocks. Both return plausible rows. So the
    /// steps live here and each path calls this rather than repeating them.
    async fn finish_reader(&self, mut reader: FileGroupReader) -> Result<FileGroupReader> {
        reader.set_data_schema(std::sync::Arc::new(self.data_schema_for_read().await?));
        // This table holds the timeline, so it can tell a committed instant from
        // one still inflight — the log-block scan cannot work that out from the
        // slice alone.
        if let Some(inputs) = self.completion_gate_inputs()? {
            reader.set_completion_gate_inputs(inputs);
        }
        Ok(reader)
    }

    /// Inputs for the log-block scan's completed/inflight gate, or `None` when
    /// this table does not need one.
    ///
    /// Mirrors Java `BaseHoodieLogRecordReader`, which runs the check only below
    /// table version 8 (`tableVersion.lesserThan(HoodieTableVersion.EIGHT)`).
    /// From version 8 the timeline records completion times, so a log file whose
    /// delta commit never completed is already dropped when the file slice is
    /// built, and asking again per block would be redundant. Below version 8
    /// there are no completion times to build a slice from, which leaves the
    /// block scan as the only place the question can be asked.
    ///
    /// So from version 8 the exclusion rests entirely on the log file's *name*
    /// carrying the delta commit that wrote it (`file_group::builder`). A table
    /// upgraded from version 6 would seem to break that, since a version-6
    /// writer names log files on the base instant — but the writer's upgrade
    /// closes it from the other side, two ways: it rolls back failed writes and
    /// compacts every log file into a new base file before bumping the version,
    /// and where the pending commit has completed commits after it, rollback is
    /// refused and the upgrade aborts rather than proceeding. Either way no
    /// version-6-named log file reaches a version-8 slice. Verified by running
    /// both paths on Spark 3.5.3 with Hudi 1.2.0-SNAPSHOT.
    fn completion_gate_inputs(&self) -> Result<Option<CompletionGateInputs>> {
        let table_version: isize = self
            .hudi_configs
            .try_get(HudiTableConfig::TableVersion)?
            .map(|v| v.into())
            .unwrap_or(6);
        Ok((table_version < 8).then(|| self.timeline.completion_gate_inputs()))
    }

    /// Build a reader for one of this table's own read paths, carrying the
    /// table's current data schema.
    ///
    /// The three read paths all need the same two steps, and a path that
    /// performed only the first would read an evolved table with a stale schema —
    /// so they share this rather than repeating it.
    async fn reader_for_read_path(&self, prepared: &ReadOptions) -> Result<FileGroupReader> {
        let reader = self.build_file_group_reader(
            prepared.hudi_options.clone(),
            std::iter::empty::<(&str, &str)>(),
        )?;
        self.finish_reader(reader).await
    }

    /// Convert caller-facing [`ReadOptions`] into the form that
    /// [`FileGroupReader`] expects: `AsOfTimestamp` resolved to
    /// `EndTimestamp` for snapshots; `StartTimestamp` / `EndTimestamp`
    /// defaults filled for incremental queries.
    fn prepare_reader_options(&self, options: &ReadOptions) -> Result<ReadOptions> {
        let options = options.with_sanitized_timestamps();
        match options.query_type()? {
            QueryType::Snapshot => {
                if let Some(ts) = self.resolve_snapshot_timestamp(&options)? {
                    Ok(options.clone().with_end_timestamp(&ts))
                } else {
                    Ok(options)
                }
            }
            QueryType::Incremental => {
                let Some((start, end)) = self.resolve_incremental_range(&options)? else {
                    return Ok(options);
                };
                self.resolve_incremental_window(options, &start, &end)
            }
        }
    }

    /// Turn an incremental window into the form the readers below need.
    ///
    /// Resolving this here rather than inside one read path is what keeps the
    /// direct read, `create_file_group_reader_with_options`, and the DataFusion
    /// scan on the same semantics — a caller assembling slices by hand used to get
    /// a different answer from `Table::read` for the same window.
    ///
    /// Two things come out of it:
    ///
    /// 1. The instant times the window admits, for the row-level mask. The window
    ///    may bound *completion* times while `_hoodie_commit_time` holds
    ///    *requested* times, so rows are matched by membership rather than by
    ///    comparing one against the other. See
    ///    [`HudiInternalConfig::IncrementalInstantTimes`].
    /// 2. Start/end bounds re-expressed over those commits' *requested* times.
    ///    Everything below the mask — which base file to open, which log blocks to
    ///    admit — is a requested-time decision, because that is what a file name
    ///    and a block header carry. Left as completion-time bounds they discard the
    ///    very files the window admits. So the range becomes pruning only and the
    ///    mask stays the exact filter, which is how Hudi 1.x divides the same work.
    fn resolve_incremental_window(
        &self,
        options: ReadOptions,
        start: &str,
        end: &str,
    ) -> Result<ReadOptions> {
        let admitted = self
            .timeline
            .get_completed_commits_in_range(Some(start), Some(end))?;

        let mut prepared = options
            .clone()
            .with_start_timestamp(start)
            .with_end_timestamp(end);
        prepared.hudi_options.insert(
            HudiInternalConfig::IncrementalInstantTimes
                .as_ref()
                .to_string(),
            admitted
                .iter()
                .map(|instant| instant.timestamp.as_str())
                .collect::<Vec<_>>()
                .join(","),
        );
        if let (Some(first), Some(last)) = (admitted.first(), admitted.last()) {
            // The start bound is exclusive, so step below the earliest admitted
            // instant to keep it in range. `instant_time_minus_one` steps in the
            // time domain: the bound is read back by several consumers, and at
            // least one of them parses it as a date.
            prepared.hudi_options.insert(
                HudiReadConfig::StartTimestamp.as_ref().to_string(),
                instant_time_minus_one(&first.timestamp),
            );
            prepared.hudi_options.insert(
                HudiReadConfig::EndTimestamp.as_ref().to_string(),
                last.timestamp.clone(),
            );
        }
        Ok(prepared)
    }

    /// Build a [`FileGroupReader`] from already-resolved hudi options.
    fn build_file_group_reader<S, K, V>(
        &self,
        hudi_opts: HashMap<String, String>,
        extra_storage_overrides: S,
    ) -> Result<FileGroupReader>
    where
        S: IntoIterator<Item = (K, V)>,
        K: AsRef<str>,
        V: Into<String>,
    {
        let mut storage_opts: HashMap<String, String> =
            HashMap::with_capacity(self.storage_options.len());
        for (k, v) in self.storage_options.iter() {
            storage_opts.insert(k.clone(), v.clone());
        }
        for (k, v) in extra_storage_overrides {
            storage_opts.insert(k.as_ref().to_string(), v.into());
        }
        FileGroupReader::new_with_overrides(self.hudi_configs.clone(), hudi_opts, storage_opts)
    }

    /// The scan's memory budget, when the table set one.
    ///
    /// Absent means `hoodie.read.file.slice.read.concurrency` stands on its own,
    /// which is the behaviour every existing table keeps.
    fn scan_max_memory_size(&self) -> Option<u64> {
        self.hudi_configs
            .try_get(HudiReadConfig::ScanMaxMemorySize)
            .ok()
            .flatten()
            .map(|v| -> usize { v.into() })
            .map(|v| v as u64)
    }

    /// How many file slices to read at once.
    ///
    /// `try_join_all` over every slice was unbounded, so a table with a thousand
    /// file groups issued a thousand concurrent reads and held a thousand merged
    /// batches at once — and under file group reader version 2 each of those also
    /// carries its own merge map and, on spill, its own RocksDB instance.
    ///
    /// The same derivation the DataFusion plan uses, so the two paths cannot
    /// bound a scan differently. This is the **only** way to obtain a
    /// concurrency for a read: the raw `hoodie.read.file.slice.read.concurrency`
    /// lookup lives inside it rather than in a method of its own, so a read
    /// cannot reach the ceiling while bypassing the budget. A test can miss that
    /// bypass — one did — but a compile error cannot. One `Table` read is one partition's worth of
    /// work, so the budget is not divided further here — dividing twice would
    /// bound this path below what the caller asked for.
    ///
    /// Separated from the read so it can be asserted directly: it is a single
    /// call whose absence compiles perfectly, and the arithmetic behind it is
    /// tested elsewhere in isolation, which says nothing about whether anything
    /// calls it.
    fn bounded_read_concurrency(&self, file_slices: &[FileSlice]) -> usize {
        self.bounded_read_concurrency_from_log_bytes(
            file_slices.iter().map(FileSlice::log_size_bytes),
        )
    }

    /// [`Self::bounded_read_concurrency`] over a borrowed selection.
    ///
    /// The metadata read routes keys to a subset of a partition's slices, so it
    /// holds `&FileSlice` rather than owning them. The budget must see that
    /// subset, not the whole partition: routing a single key to one shard should
    /// admit on that shard's cost.
    pub(crate) fn bounded_read_concurrency_for(&self, file_slices: &[&FileSlice]) -> usize {
        self.bounded_read_concurrency_from_log_bytes(file_slices.iter().map(|s| s.log_size_bytes()))
    }

    /// The one place the ceiling and the budget meet.
    ///
    /// Both public forms funnel through this, over log sizes alone: the
    /// admission decision needs nothing else from a slice, so neither form has
    /// to own or copy one.
    fn bounded_read_concurrency_from_log_bytes(
        &self,
        slice_log_bytes: impl Iterator<Item = Option<u64>>,
    ) -> usize {
        let slice_log_bytes: Vec<Option<u64>> = slice_log_bytes.collect();
        let ceiling: usize = self
            .hudi_configs
            .get_or_default(HudiReadConfig::FileSliceReadConcurrency)
            .into();
        crate::file_group::admission::slices_in_flight(
            self.scan_max_memory_size(),
            1,
            &slice_log_bytes,
            ceiling.max(1),
        )
    }

    /// Read `file_slices` with at most [`Self::bounded_read_concurrency`] in
    /// flight, or fewer when a scan memory budget admits fewer, in slice order.
    ///
    /// Order is preserved (`buffered`, not `buffer_unordered`) because a caller
    /// that concatenates these batches should not see its row order shift with
    /// scheduling. Any single failure aborts the whole read, as `try_join_all`
    /// did — a partially-read table is not a useful answer.
    async fn read_file_slices_bounded(
        &self,
        fg_reader: &FileGroupReader,
        file_slices: &[FileSlice],
        fg_options: &ReadOptions,
    ) -> Result<Vec<RecordBatch>> {
        crate::util::concurrency::bounded_in_order(
            file_slices,
            self.bounded_read_concurrency(file_slices),
            |file_slice| fg_reader.read_file_slice(file_slice, fg_options),
        )
        .await
    }

    /// Warn when an incremental window reaches below the active timeline.
    ///
    /// Archived instants are not read (see
    /// [`TimelineLoader::load_archived_instants`](crate::timeline::loader::TimelineLoader::load_archived_instants)),
    /// so a window whose start predates the archival boundary can only ever
    /// report the commits still in the active timeline. Java's
    /// `IncrementalQueryAnalyzer` falls back to the archived timeline for exactly
    /// this case. Until that is implemented, say so — a range query silently
    /// returning fewer commits than the range contains is the failure mode a
    /// downstream consumer cannot detect for itself.
    fn warn_if_window_predates_active_timeline(&self, start: &str) {
        if let Some(message) = self.window_predates_active_timeline(start) {
            log::warn!("{message}");
        }
    }

    /// The warning [`Self::warn_if_window_predates_active_timeline`] emits, or
    /// `None` when the window is wholly inside the active timeline.
    ///
    /// Split out from the logging so the condition can be asserted: a warning
    /// that stops firing is indistinguishable from one that never did, and this
    /// is the only thing telling a caller their range came back short.
    fn window_predates_active_timeline(&self, start: &str) -> Option<String> {
        let boundary = self.timeline.earliest_active_instant.as_deref()?;
        // Instant times are compared as strings, the way Hudi compares them.
        (start < boundary).then(|| {
            format!(
                "incremental read starts at '{start}', before the active timeline begins at \
                 '{boundary}'; commits archived below that point are not read, so this window \
                 reports fewer changes than it covers"
            )
        })
    }

    /// Read records, dispatching on `options.query_type`.
    ///
    /// - [`QueryType::Snapshot`] reads at `options.as_of_timestamp` or the latest commit.
    /// - [`QueryType::Incremental`] reads the change range
    ///   (`options.start_timestamp`, `options.end_timestamp`].
    ///
    /// `options.filters` drive partition pruning, file-level stats pruning (snapshot
    /// only), and a row-level mask on every returned batch — see [`ReadOptions::filters`]
    /// for the full breakdown. `options.hudi_options` override table-level Hudi configs
    /// for this single read.
    pub async fn read(&self, options: &ReadOptions) -> Result<Vec<RecordBatch>> {
        let prepared = self.prepare_reader_options(options)?;
        match prepared.query_type()? {
            QueryType::Snapshot => self.read_snapshot_inner(&prepared).await,
            QueryType::Incremental => self.read_incremental_inner(&prepared).await,
        }
    }

    async fn read_snapshot_inner(&self, prepared: &ReadOptions) -> Result<Vec<RecordBatch>> {
        let Some(timestamp) = prepared.end_timestamp() else {
            return Ok(Vec::new());
        };
        let base_file_only = self.is_base_file_only(prepared)?;
        let file_slices = self
            .get_file_slices_inner(timestamp, &prepared.filters, base_file_only)
            .await?;
        // The table's current schema, not the base file's: a base file written
        // before a column was widened or added would otherwise force the newer
        // records back into its own narrower shape.
        let fg_reader = self.reader_for_read_path(prepared).await?;
        let fg_options = self.options_for_file_group(prepared);
        self.read_file_slices_bounded(&fg_reader, &file_slices, &fg_options)
            .await
    }

    async fn read_incremental_inner(&self, prepared: &ReadOptions) -> Result<Vec<RecordBatch>> {
        let (Some(start), Some(end)) = (prepared.start_timestamp(), prepared.end_timestamp())
        else {
            return Ok(Vec::new());
        };
        self.warn_if_window_predates_active_timeline(start);
        let base_file_only = self.is_base_file_only(prepared)?;
        let file_slices = self
            .get_file_slices_between_inner(start, end, &prepared.filters, base_file_only)
            .await?;

        // The table's current schema, not the base file's: a base file written
        // before a column was widened or added would otherwise force the newer
        // records back into its own narrower shape.
        let fg_reader = self.reader_for_read_path(prepared).await?;
        let fg_options = self.options_for_file_group(prepared);

        self.read_file_slices_bounded(&fg_reader, &file_slices, &fg_options)
            .await
    }

    /// Build the [`ReadOptions`] passed to `FileGroupReader` for a per-slice read,
    /// stripping filters that target a partition column dropped from data files.
    ///
    /// `FileGroupReader` validates filter columns strictly against the read batch
    /// schema; when `hoodie.datasource.write.drop.partition.columns` is enabled,
    /// partition columns aren't in parquet, so a partition filter would surface as
    /// an error there. The partition pruner has already used those filters at
    /// table level, so dropping them here is safe and avoids the false-positive.
    fn options_for_file_group(&self, options: &ReadOptions) -> ReadOptions {
        let drops: bool = self
            .hudi_configs
            .get_or_default(HudiTableConfig::DropsPartitionFields)
            .into();
        if !drops || options.filters.is_empty() {
            return options.clone();
        }
        let partition_columns: Vec<String> = self
            .hudi_configs
            .get_or_default(HudiTableConfig::PartitionFields)
            .into();
        let mut applicable = options.clone();
        applicable.filters = options
            .filters
            .iter()
            .filter(|filter| !partition_columns.iter().any(|p| p == &filter.field))
            .cloned()
            .collect();
        applicable
    }

    /// Resolve the snapshot timestamp from `options`: explicit `as_of_timestamp` if set,
    /// otherwise the table's latest commit. Returns `None` only when the table has no
    /// commits and no explicit timestamp was given.
    fn resolve_snapshot_timestamp(&self, options: &ReadOptions) -> Result<Option<String>> {
        if let Some(ts) = options.as_of_timestamp() {
            return Ok(Some(format_timestamp(ts, &self.timezone())?));
        }
        Ok(self
            .timeline
            .get_latest_commit_timestamp_as_option()
            .map(|s| s.to_string()))
    }

    /// Resolve the incremental change range `(start, end]` from `options`. `start`
    /// defaults to [`EARLIEST_START_TIMESTAMP`]; `end` defaults to the latest commit.
    ///
    /// Returns `Ok(None)` only when no `end_timestamp` is provided AND the table has
    /// no commits. Invalid timestamp strings propagate as `Err`.
    fn resolve_incremental_range(&self, options: &ReadOptions) -> Result<Option<(String, String)>> {
        let timezone = self.timezone();
        // The default end is the latest COMPLETION time, because that is what an
        // incremental window bounds. Defaulting to the latest requested time
        // instead excluded the newest commit from "everything up to now": it
        // completed strictly after the instant it was requested at.
        let Some(end) = options
            .end_timestamp()
            .or_else(|| self.timeline.get_latest_completion_timestamp_as_option())
        else {
            return Ok(None);
        };
        let end = format_timestamp(end, &timezone)?;
        let start = options
            .start_timestamp()
            .unwrap_or(EARLIEST_START_TIMESTAMP);
        let start = format_timestamp(start, &timezone)?;
        Ok(Some((start, end)))
    }

    // =========================================================================
    // Streaming Read APIs
    // =========================================================================

    /// Streaming read; dispatches on `options.query_type`.
    ///
    /// Snapshot streams batches as they are read from each file slice. Incremental
    /// streaming is not yet supported and returns an `Unsupported` error.
    ///
    /// For MOR file slices with log files, file group reader version 2 streams the
    /// merge; version 1 collects it and yields the slice as a single batch.
    ///
    /// # Example
    /// ```ignore
    /// use futures::StreamExt;
    /// use hudi::table::ReadOptions;
    ///
    /// let options = ReadOptions::new()
    ///     .with_filters([("city", "=", "san_francisco")])?
    ///     .with_batch_size(4096)?;
    /// let mut stream = table.read_stream(&options).await?;
    /// while let Some(result) = stream.next().await {
    ///     println!("Read {} rows", result?.num_rows());
    /// }
    /// ```
    pub async fn read_stream(
        &self,
        options: &ReadOptions,
    ) -> Result<futures::stream::BoxStream<'static, Result<RecordBatch>>> {
        let prepared = self.prepare_reader_options(options)?;
        match prepared.query_type()? {
            QueryType::Snapshot => self.read_snapshot_stream_inner(&prepared).await,
            QueryType::Incremental => Err(CoreError::Unsupported(
                "Streaming for incremental queries is not yet supported".to_string(),
            )),
        }
    }

    async fn read_snapshot_stream_inner(
        &self,
        prepared: &ReadOptions,
    ) -> Result<futures::stream::BoxStream<'static, Result<RecordBatch>>> {
        use futures::stream::{self, StreamExt};

        let Some(timestamp) = prepared.end_timestamp() else {
            return Ok(Box::pin(stream::empty()));
        };

        let base_file_only = self.is_base_file_only(prepared)?;
        let file_slices = self
            .get_file_slices_inner(timestamp, &prepared.filters, base_file_only)
            .await?;

        if file_slices.is_empty() {
            return Ok(Box::pin(stream::empty()));
        }

        // The table's current schema, not the base file's: a base file written
        // before a column was widened or added would otherwise force the newer
        // records back into its own narrower shape.
        let fg_reader = self.reader_for_read_path(prepared).await?;

        // Extract per-batch options. Keep `filters` so they apply at row-level too —
        // the upstream pruning already used them at file/partition level; applying at
        // row-level closes the gap for non-partition column filters. Strip filters
        // on dropped partition columns so they don't trigger FGR validation errors.
        let fg_options_template = self.options_for_file_group(prepared);
        let projection = fg_options_template.projection.clone();
        let row_filters = fg_options_template.filters.clone();
        // Carry batch_size in hudi_options if set; everything else (timestamps,
        // query_type) is irrelevant to the per-slice FG-reader read.
        let mut per_slice_hudi_options: HashMap<String, String> = HashMap::new();
        if let Some(bs) = fg_options_template.batch_size()? {
            per_slice_hudi_options.insert(
                HudiReadConfig::StreamBatchSize.as_ref().to_string(),
                bs.to_string(),
            );
        }

        let streams_iter = file_slices.into_iter().map(move |file_slice| {
            let fg_reader = fg_reader.clone();
            let projection = projection.clone();
            let row_filters = row_filters.clone();
            let options = ReadOptions {
                filters: row_filters,
                projection,
                hudi_options: per_slice_hudi_options.clone(),
            };
            async move {
                // Every await in a small slice's open can resolve without the
                // task ever suspending — a cached base file's reads complete on
                // the blocking pool before their first poll — so opening slice
                // after slice could run without the scheduler once seeing
                // another task. One yield per slice keeps the stream cooperative
                // on a single-worker runtime whatever the I/O timing.
                tokio::task::yield_now().await;
                fg_reader
                    .read_file_slice_stream(&file_slice, &options)
                    .await
            }
        });

        // Chain all file slice streams together, propagating errors to the caller.
        // Sequential, and deliberately so: `.then` awaits each slice's future
        // before starting the next, so exactly one slice is in flight however many
        // the table has.
        //
        // `hoodie.read.file.slice.read.concurrency` therefore does **not** apply
        // here, unlike [`Self::read`], which fans out through
        // `read_file_slices_bounded`. That is not an oversight: a streaming read
        // exists so the whole result is never resident, and fanning out N ways
        // would hold N slices' batches in flight at once — spending the memory the
        // caller chose this API to avoid. A caller who wants the slices read
        // concurrently wants `read`.
        //
        // The consequence worth stating: peak memory here does not grow with slice
        // count, so a scan memory budget has nothing to bound on this path.
        let combined_stream = stream::iter(streams_iter)
            .then(|fut| fut)
            .flat_map(|result| match result {
                Ok(file_stream) => file_stream.left_stream(),
                Err(e) => stream::once(async move { Err(e) }).right_stream(),
            });

        Ok(Box::pin(combined_stream))
    }

    /// Compute estimated table-level statistics for scan planning.
    ///
    /// Returns `(estimated_num_rows, estimated_total_byte_size)` derived from
    /// the metadata table for snapshot queries. Returns `None` if the metadata
    /// table is not enabled, statistics cannot be computed, or the query type
    /// is incremental (commit metadata does not reliably carry base file sizes
    /// for all commit types).
    pub async fn compute_table_stats(&self, options: Option<&ReadOptions>) -> Option<(u64, u64)> {
        if let Some(opts) = options {
            match opts.query_type() {
                Ok(QueryType::Incremental) => return None,
                Ok(QueryType::Snapshot) => {}
                Err(_) => return None,
            }
        }

        if !self.is_metadata_table_enabled() {
            return None;
        }

        let partition_schema = self.get_partition_schema().await.ok()?;
        let hudi_configs = self.hudi_configs.as_ref();
        let partition_pruner = PartitionPruner::new(&[], &partition_schema, hudi_configs).ok()?;
        let mdt = self.get_or_init_metadata_table().await.ok()?;
        let valid = self.valid_instant_timestamps(mdt).await.ok()?;
        let records = mdt
            .fetch_files_partition_records(&partition_pruner, &valid)
            .await
            .ok()?;

        let configured_base_file_format =
            self.file_system_view.configured_base_file_format().ok()?;
        if configured_base_file_format
            .as_ref()
            .is_some_and(|format| !matches!(format, BaseFileFormatValue::Parquet))
        {
            return None;
        }
        let total_on_disk_size = records
            .values()
            .filter(|record| !record.is_all_partitions())
            .flat_map(|record| record.active_files_with_sizes())
            .filter(|(name, _)| BaseFileFormatValue::Parquet.matches_extension(name))
            .map(|(_, size)| size)
            .sum::<u64>();

        if total_on_disk_size == 0 {
            return None;
        }

        let latest_ts = self.timeline.get_latest_commit_timestamp().ok()?;
        let estimator = self.get_or_init_estimator(&latest_ts).await?;
        let (estimated_total_byte_size, estimated_total_rows) =
            estimator.estimate(total_on_disk_size);
        Some((
            estimated_total_rows.max(0) as u64,
            estimated_total_byte_size.max(0) as u64,
        ))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The only thing telling a caller their incremental range came back short.
    ///
    /// Archived commits are not enumerated unless
    /// `hoodie.internal.timeline.archived.enabled` is set, so a window reaching
    /// below the active timeline reports fewer changes than it covers. That is
    /// silent except for this warning — and a warning nothing asserts is one
    /// that can stop firing without anyone noticing.
    ///
    /// Both directions are asserted: a window inside the active timeline must
    /// stay quiet, or the warning would be noise a caller learns to ignore.
    #[tokio::test]
    async fn test_window_predating_the_active_timeline_is_reported() -> Result<()> {
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await?;

        let boundary = hudi_table
            .timeline
            .earliest_active_instant
            .clone()
            .expect("fixture has an active timeline");

        // A start below the boundary: the window covers commits that are no
        // longer enumerable.
        let message = hudi_table
            .window_predates_active_timeline("00000000000000")
            .expect("a start below the active timeline must be reported");
        assert!(
            message.contains("00000000000000") && message.contains(&boundary),
            "the warning must name both the requested start and the boundary, got: {message}"
        );

        // The boundary itself, and anything after it, is wholly inside the
        // active timeline.
        assert!(
            hudi_table
                .window_predates_active_timeline(&boundary)
                .is_none(),
            "a window starting exactly at the boundary covers no archived commit"
        );
        // Hudi compares instant times lexicographically, so appending any
        // character yields a string that sorts strictly after the boundary —
        // i.e. a window that opens inside the active timeline.
        let after = format!("{boundary}9");
        assert!(
            hudi_table.window_predates_active_timeline(&after).is_none(),
            "a window inside the active timeline must not warn"
        );
        Ok(())
    }
    use crate::config::HUDI_CONF_DIR;
    use crate::config::internal::HudiInternalConfig;
    use crate::config::table::BaseFileFormatValue;
    use crate::config::table::HudiTableConfig::{
        BaseFileFormat, Checksum, DatabaseName, DropsPartitionFields, IsHiveStylePartitioning,
        IsPartitionPathUrlencoded, KeyGeneratorClass, OrderingFields, PartitionFields,
        PopulatesMetaFields, RecordKeyFields, TableName, TableType, TableVersion,
        TimelineLayoutVersion, TimelineTimezone,
    };
    use crate::config::util::empty_options;
    use crate::error::CoreError;
    use crate::metadata::meta_field::MetaField;
    use crate::storage::Storage;
    use crate::storage::util::join_url_segments;
    use crate::timeline::EARLIEST_START_TIMESTAMP;
    use hudi_test::{SampleTable, assert_arrow_field_names_eq, assert_avro_field_names_eq};
    use serial_test::serial;
    use std::collections::HashSet;
    use std::fs::canonicalize;
    use std::path::PathBuf;
    use std::{env, panic};

    /// Test helper that loads resolved `HudiConfigs` from a test data directory
    /// without constructing a full `Table`. Useful for testing config parsing
    /// with intentionally invalid values that would prevent table construction.
    async fn get_test_configs(table_dir_name: &str) -> Arc<HudiConfigs> {
        let base_url = Url::from_file_path(
            canonicalize(PathBuf::from("tests").join("data").join(table_dir_name)).unwrap(),
        )
        .unwrap();
        let mut resolver = crate::table::builder::OptionResolver::new_with_options(
            base_url.as_str(),
            [("hoodie.internal.skip.config.validation", "true")],
        );
        resolver.resolve_options().await.unwrap();
        Arc::new(HudiConfigs::new(resolver.hudi_options.iter()))
    }

    /// Test helper to create a new `Table` instance without validating the configuration.
    ///
    /// # Arguments
    ///
    /// * `table_dir_name` - Name of the table root directory; all under `crates/core/tests/data/`.
    async fn get_test_table_without_validation(table_dir_name: &str) -> Table {
        let base_url = Url::from_file_path(
            canonicalize(PathBuf::from("tests").join("data").join(table_dir_name)).unwrap(),
        )
        .unwrap();
        Table::new_with_options(
            base_url.as_str(),
            [("hoodie.internal.skip.config.validation", "true")],
        )
        .await
        .unwrap()
    }

    /// Test helper to get relative file paths from the table with filters.
    async fn get_file_paths_with_filters(
        table: &Table,
        filters: &[(&str, &str, &str)],
    ) -> Result<Vec<String>> {
        let mut file_paths = Vec::new();
        let base_url = table.base_url();
        let options = ReadOptions::new().with_filters(filters.iter().copied())?;
        for f in table.get_file_slices(&options).await? {
            let Some(relative_path) = f.base_file_relative_path()? else {
                continue;
            };
            let file_url = join_url_segments(&base_url, &[relative_path.as_str()])?;
            file_paths.push(file_url.to_string());
        }
        Ok(file_paths)
    }

    /// A scan memory budget lowers the core fan-out's concurrency; without one
    /// the configured ceiling stands.
    ///
    /// Pins the wiring rather than the arithmetic — `slices_in_flight` has its
    /// own tests, and passing those tells you nothing about whether this path
    /// calls it. The equivalent wiring on the DataFusion side was claimed,
    /// compiled, and shipped absent; this is the same class of gap on the core
    /// path.
    #[tokio::test]
    async fn a_scan_memory_budget_lowers_the_core_fan_out() {
        use crate::file_group::base_file::BaseFile;
        use crate::file_group::file_slice::FileSlice;
        use crate::storage::file_metadata::FileMetadata;
        use std::str::FromStr;

        // Log files with recorded sizes: an unmeasured slice is unestimatable and
        // floors at 1, which would make the bounded arm pass for the wrong reason.
        let slices: Vec<FileSlice> = (0..4)
            .map(|i| {
                let name = format!("a{i}0000-0000-0000-0000-00000000000{i}-0_0-1-0_001.parquet");
                let mut fs = FileSlice::new(BaseFile::from_str(&name).unwrap(), String::new());
                let log = format!(".a{i}0000-0000-0000-0000-00000000000{i}-0_002.log.1_0-1-0");
                let mut lf = crate::file_group::log_file::LogFile::from_str(&log).unwrap();
                lf.file_metadata = Some(FileMetadata::new(&log, 8 * 1024 * 1024));
                fs.log_files.insert(lf);
                fs
            })
            .collect();

        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let unbounded = Table::new(base_url.path()).await.unwrap();
        assert_eq!(
            unbounded.bounded_read_concurrency(&slices),
            4,
            "with no budget the configured ceiling stands"
        );

        // Each slice estimates at 33 MiB + 4 x 8 MiB = 65 MiB, so 128 MiB admits one.
        let bounded = Table::new_with_options(
            base_url.path(),
            [("hoodie.read.scan.max.memory.size", "134217728")],
        )
        .await
        .unwrap();
        assert_eq!(
            bounded.bounded_read_concurrency(&slices),
            1,
            "a 128 MiB budget must admit one 65 MiB slice, not the ceiling of 4"
        );
    }

    /// `read_stream` returns exactly what `read` returns, and the slice
    /// concurrency ceiling changes neither.
    ///
    /// The invariant this pins is the documented contract at
    /// `read_snapshot_stream_inner`: the streaming path reads slices sequentially,
    /// so `hoodie.read.file.slice.read.concurrency` does not apply to it. Setting
    /// the ceiling to 1 and to 8 must produce the same rows as the eager read —
    /// if a future change made the stream fan out, the row totals would still
    /// match, but the config would start mattering, and this test is where that
    /// divergence gets noticed.
    ///
    /// It does not prove sequentiality. Nothing observable from outside
    /// distinguishes one slice in flight from four; the contract is asserted by
    /// the doc comment and by `.then`, and this test pins the consequence a caller
    /// can actually check.
    #[tokio::test]
    async fn read_stream_matches_read_and_ignores_the_slice_ceiling() -> Result<()> {
        use futures::StreamExt;

        // A partitioned fixture, because the point is several file slices: on a
        // single-slice table this test cannot tell "read every slice" from "read
        // the first", and a `.take(1)` mutation passes it.
        let base_url = SampleTable::V6ComplexkeygenHivestyle.url_to_cow();
        let eager = Table::new(base_url.path())
            .await?
            .read(&ReadOptions::new())
            .await?;
        let expected: usize = eager.iter().map(|b| b.num_rows()).sum();
        assert!(
            expected > 0,
            "the fixture must return rows, or nothing is pinned"
        );

        for ceiling in ["1", "8"] {
            let table = Table::new_with_options(
                base_url.path(),
                [(HudiReadConfig::FileSliceReadConcurrency.as_ref(), ceiling)],
            )
            .await?;
            let mut stream = table.read_stream(&ReadOptions::new()).await?;
            let mut rows = 0usize;
            while let Some(batch) = stream.next().await {
                rows += batch?.num_rows();
            }
            assert_eq!(
                rows, expected,
                "read_stream at concurrency={ceiling} must return the eager row count"
            );
        }
        Ok(())
    }

    #[tokio::test]
    async fn test_hudi_table_get_hudi_options() {
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let hudi_options = hudi_table.hudi_options();
        for (k, v) in hudi_options.iter() {
            assert!(k.starts_with("hoodie."));
            assert!(!v.is_empty());
        }
    }

    #[tokio::test]
    async fn test_hudi_table_get_storage_options() {
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();

        let cloud_prefixes: HashSet<_> = Storage::CLOUD_STORAGE_PREFIXES
            .iter()
            .map(|prefix| prefix.to_lowercase())
            .collect();

        for (key, value) in hudi_table.storage_options.iter() {
            let key_lower = key.to_lowercase();
            assert!(
                cloud_prefixes
                    .iter()
                    .any(|prefix| key_lower.starts_with(prefix)),
                "Storage option key '{key}' should start with a cloud storage prefix"
            );
            assert!(
                !value.is_empty(),
                "Storage option value for key '{key}' should not be empty"
            );
        }
    }

    #[tokio::test]
    async fn test_hudi_table_storage_options_accessor_and_is_mor() {
        let cow_base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let cow_table = Table::new(cow_base_url.path()).await.unwrap();
        assert_eq!(
            cow_table.storage_options(),
            cow_table.storage_options.as_ref().clone()
        );
        assert!(!cow_table.is_mor());

        let mor_base_url =
            SampleTable::V6SimplekeygenNonhivestyleOverwritetable.url_to_mor_parquet();
        let mor_table = Table::new(mor_base_url.path()).await.unwrap();
        assert!(mor_table.is_mor());
    }

    #[cfg(feature = "datafusion")]
    #[tokio::test]
    async fn test_hudi_table_register_storage() {
        use datafusion::execution::runtime_env::RuntimeEnv;
        use std::sync::Arc;

        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let runtime_env = Arc::new(RuntimeEnv::default());
        hudi_table.register_storage(runtime_env);
    }

    #[tokio::test]
    #[serial(env_vars)]
    async fn hudi_table_get_schema_from_empty_table_without_create_schema() {
        let table = get_test_table_without_validation("table_props_no_create_schema").await;

        let schema = table.get_schema().await;
        assert!(schema.is_err());
        assert!(matches!(schema.unwrap_err(), CoreError::SchemaNotFound(_)));

        let schema = table.get_schema_in_avro_str().await;
        assert!(schema.is_err());
        assert!(matches!(schema.unwrap_err(), CoreError::SchemaNotFound(_)));
    }

    #[tokio::test]
    async fn hudi_table_get_schema_from_empty_table_resolves_to_table_create_schema() {
        for base_url in SampleTable::V6Empty.urls() {
            let hudi_table = Table::new(base_url.path()).await.unwrap();

            // Validate the Arrow schema without meta fields
            let schema = hudi_table.get_schema().await;
            assert!(schema.is_ok());
            let schema = schema.unwrap();
            assert_arrow_field_names_eq!(schema, ["id", "name", "isActive"]);

            // Validate the Arrow schema with meta fields
            let schema = hudi_table.get_schema_with_meta_fields().await;
            assert!(schema.is_ok());
            let schema = schema.unwrap();
            assert_arrow_field_names_eq!(
                schema,
                [MetaField::field_names(), vec!["id", "name", "isActive"]].concat()
            );

            // Validate the Avro schema without meta fields
            let avro_schema = hudi_table.get_schema_in_avro_str().await;
            assert!(avro_schema.is_ok());
            let avro_schema = avro_schema.unwrap();
            assert_avro_field_names_eq!(&avro_schema, ["id", "name", "isActive"]);

            // Validate the Avro schema with meta fields
            let avro_schema = hudi_table.get_schema_in_avro_str_with_meta_fields().await;
            assert!(avro_schema.is_ok());
            let avro_schema = avro_schema.unwrap();
            assert_avro_field_names_eq!(
                &avro_schema,
                [
                    MetaField::field_names().as_slice(),
                    &["id", "name", "isActive"]
                ]
                .concat()
            );
        }
    }

    #[tokio::test]
    async fn hudi_table_get_schema() {
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let original_field_names = [
            "id",
            "name",
            "isActive",
            "byteField",
            "shortField",
            "intField",
            "longField",
            "floatField",
            "doubleField",
            "decimalField",
            "dateField",
            "timestampField",
            "binaryField",
            "arrayField",
            "mapField",
            "structField",
        ];

        // Check Arrow schema without meta fields
        let arrow_schema = hudi_table.get_schema().await;
        assert!(arrow_schema.is_ok());
        let arrow_schema = arrow_schema.unwrap();
        assert_arrow_field_names_eq!(arrow_schema, original_field_names);

        // Check Arrow schema with meta fields
        let arrow_schema = hudi_table.get_schema_with_meta_fields().await;
        assert!(arrow_schema.is_ok());
        let arrow_schema = arrow_schema.unwrap();
        assert_arrow_field_names_eq!(
            arrow_schema,
            [MetaField::field_names(), original_field_names.to_vec()].concat()
        );

        // Check Avro schema without meta fields
        let avro_schema = hudi_table.get_schema_in_avro_str().await;
        assert!(avro_schema.is_ok());
        let avro_schema = avro_schema.unwrap();
        assert_avro_field_names_eq!(&avro_schema, original_field_names);
    }

    #[tokio::test]
    async fn hudi_table_get_partition_schema() {
        let base_url = SampleTable::V6TimebasedkeygenNonhivestyle.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let schema = hudi_table.get_partition_schema().await;
        assert!(schema.is_ok());
        let schema = schema.unwrap();
        assert_arrow_field_names_eq!(schema, [MetaField::PartitionPath.as_ref()]);
    }

    #[tokio::test]
    async fn hudi_table_get_partition_schema_uses_config_order_not_table_schema_order() {
        // V6ComplexkeygenHivestyle declares PARTITIONED BY (byteField, shortField).
        // The returned partition schema must follow the partition.fields config order,
        // which also matches the on-disk partition path order: byteField=.../shortField=...
        let base_url = SampleTable::V6ComplexkeygenHivestyle.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let schema = hudi_table.get_partition_schema().await.unwrap();
        assert_arrow_field_names_eq!(schema, ["byteField", "shortField"]);
    }

    #[tokio::test]
    #[serial(env_vars)]
    async fn validate_invalid_table_props() {
        let configs = get_test_configs("table_props_invalid").await;
        assert!(
            configs.validate(BaseFileFormat).is_err(),
            "required config is missing"
        );
        assert!(configs.validate(Checksum).is_err());
        assert!(
            configs.validate(DatabaseName).is_ok(),
            "non-required config is missing"
        );
        assert!(configs.validate(DropsPartitionFields).is_err());
        assert!(configs.validate(IsHiveStylePartitioning).is_err());
        assert!(configs.validate(IsPartitionPathUrlencoded).is_err());
        assert!(
            configs.validate(KeyGeneratorClass).is_ok(),
            "non-required config is missing"
        );
        assert!(
            configs.validate(PartitionFields).is_ok(),
            "non-required config is missing"
        );
        assert!(
            configs.validate(OrderingFields).is_ok(),
            "non-required config is missing"
        );
        assert!(
            configs.validate(PopulatesMetaFields).is_ok(),
            "non-required config is missing"
        );
        assert!(
            configs.validate(RecordKeyFields).is_ok(),
            "non-required config is missing"
        );
        assert!(
            configs.validate(TableName).is_err(),
            "required config is missing"
        );
        assert!(
            configs.validate(TableType).is_ok(),
            "Valid table type value"
        );
        assert!(configs.validate(TableVersion).is_err());
        assert!(configs.validate(TimelineLayoutVersion).is_err());
        assert!(
            configs.validate(TimelineTimezone).is_ok(),
            "non-required config is missing"
        );
    }

    #[tokio::test]
    #[serial(env_vars)]
    async fn get_invalid_table_props() {
        let configs = get_test_configs("table_props_invalid").await;
        assert!(configs.get(BaseFileFormat).is_err());
        assert!(configs.get(Checksum).is_err());
        assert!(configs.get(DatabaseName).is_err());
        assert!(configs.get(DropsPartitionFields).is_err());
        assert!(configs.get(IsHiveStylePartitioning).is_err());
        assert!(configs.get(IsPartitionPathUrlencoded).is_err());
        assert!(configs.get(KeyGeneratorClass).is_err());
        assert!(configs.get(PartitionFields).is_err());
        assert!(configs.get(OrderingFields).is_err());
        assert!(configs.get(PopulatesMetaFields).is_err());
        assert!(configs.get(RecordKeyFields).is_err());
        assert!(configs.get(TableName).is_err());
        assert!(configs.get(TableType).is_ok(), "Valid table type value");
        assert!(configs.get(TableVersion).is_err());
        assert!(configs.get(TimelineLayoutVersion).is_err());
        assert!(configs.get(TimelineTimezone).is_err());
    }

    #[tokio::test]
    #[serial(env_vars)]
    async fn get_default_for_invalid_table_props() {
        let configs = get_test_configs("table_props_invalid").await;
        let actual: String = configs.get_or_default(BaseFileFormat).into();
        assert_eq!(actual, "parquet");
        assert!(panic::catch_unwind(|| configs.get_or_default(Checksum)).is_err());
        let actual: String = configs.get_or_default(DatabaseName).into();
        assert_eq!(actual, "default");
        let actual: bool = configs.get_or_default(DropsPartitionFields).into();
        assert!(!actual);
        let actual: bool = configs.get_or_default(IsHiveStylePartitioning).into();
        assert!(!actual);
        let actual: bool = configs.get_or_default(IsPartitionPathUrlencoded).into();
        assert!(!actual);
        assert!(panic::catch_unwind(|| configs.get_or_default(KeyGeneratorClass)).is_err());
        let actual: Vec<String> = configs.get_or_default(PartitionFields).into();
        assert!(actual.is_empty());
        assert!(panic::catch_unwind(|| configs.get_or_default(OrderingFields)).is_err());
        let actual: bool = configs.get_or_default(PopulatesMetaFields).into();
        assert!(actual);
        assert!(panic::catch_unwind(|| configs.get_or_default(RecordKeyFields)).is_err());
        assert!(panic::catch_unwind(|| configs.get_or_default(TableName)).is_err());
        let actual: String = configs.get_or_default(TableType).into();
        assert_eq!(actual, "COPY_ON_WRITE");
        assert!(panic::catch_unwind(|| configs.get_or_default(TableVersion)).is_err());
        assert!(panic::catch_unwind(|| configs.get_or_default(TimelineLayoutVersion)).is_err());
        let actual: String = configs.get_or_default(TimelineTimezone).into();
        assert_eq!(actual, "utc");
    }

    #[tokio::test]
    #[serial(env_vars)]
    async fn get_valid_table_props() {
        let table = get_test_table_without_validation("table_props_valid").await;
        let configs = table.hudi_configs;
        let actual: String = configs.get(BaseFileFormat).unwrap().into();
        assert_eq!(actual, "parquet");
        let actual: isize = configs.get(Checksum).unwrap().into();
        assert_eq!(actual, 3761586722);
        let actual: String = configs.get(DatabaseName).unwrap().into();
        assert_eq!(actual, "db");
        let actual: bool = configs.get(DropsPartitionFields).unwrap().into();
        assert!(!actual);
        let actual: bool = configs.get(IsHiveStylePartitioning).unwrap().into();
        assert!(!actual);
        let actual: bool = configs.get(IsPartitionPathUrlencoded).unwrap().into();
        assert!(!actual);
        let actual: String = configs.get(KeyGeneratorClass).unwrap().into();
        assert_eq!(actual, "org.apache.hudi.keygen.SimpleKeyGenerator");
        let actual: Vec<String> = configs.get(PartitionFields).unwrap().into();
        assert_eq!(actual, vec!["city"]);
        let actual: Vec<String> = configs.get(OrderingFields).unwrap().into();
        assert_eq!(actual, vec!["ts"]);
        let actual: bool = configs.get(PopulatesMetaFields).unwrap().into();
        assert!(actual);
        let actual: Vec<String> = configs.get(RecordKeyFields).unwrap().into();
        assert_eq!(actual, vec!["uuid"]);
        let actual: String = configs.get(TableName).unwrap().into();
        assert_eq!(actual, "trips");
        let actual: String = configs.get(TableType).unwrap().into();
        assert_eq!(actual, "COPY_ON_WRITE");
        let actual: isize = configs.get(TableVersion).unwrap().into();
        assert_eq!(actual, 6);
        let actual: isize = configs.get(TimelineLayoutVersion).unwrap().into();
        assert_eq!(actual, 1);
        let actual: String = configs.get(TimelineTimezone).unwrap().into();
        assert_eq!(actual, "local");
    }

    #[tokio::test]
    #[serial(env_vars)]
    async fn get_global_table_props() {
        // Without the environment variable HUDI_CONF_DIR
        let table = get_test_table_without_validation("table_props_partial").await;
        let configs = table.hudi_configs;
        assert!(configs.get(DatabaseName).is_err());
        assert!(configs.get(TableType).is_err());
        let actual: String = configs.get(TableName).unwrap().into();
        assert_eq!(actual, "trips");

        // Environment variable HUDI_CONF_DIR points to nothing
        let base_path = env::current_dir().unwrap();
        let hudi_conf_dir = base_path.join("random/wrong/dir");
        unsafe {
            env::set_var(HUDI_CONF_DIR, hudi_conf_dir.as_os_str());
        }
        let table = get_test_table_without_validation("table_props_partial").await;
        let configs = table.hudi_configs;
        assert!(configs.get(DatabaseName).is_err());
        assert!(configs.get(TableType).is_err());
        let actual: String = configs.get(TableName).unwrap().into();
        assert_eq!(actual, "trips");

        // With global config
        let base_path = env::current_dir().unwrap();
        let hudi_conf_dir = base_path.join("tests/data/hudi_conf_dir");
        unsafe {
            env::set_var(HUDI_CONF_DIR, hudi_conf_dir.as_os_str());
        }
        let table = get_test_table_without_validation("table_props_partial").await;
        let configs = table.hudi_configs;
        let actual: String = configs.get(DatabaseName).unwrap().into();
        assert_eq!(actual, "tmpdb");
        let actual: String = configs.get(TableType).unwrap().into();
        assert_eq!(actual, "MERGE_ON_READ");
        let actual: String = configs.get(TableName).unwrap().into();
        assert_eq!(actual, "trips");
        unsafe {
            env::remove_var(HUDI_CONF_DIR);
        }
    }

    #[tokio::test]
    async fn hudi_table_read_file_slice() {
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let batches = hudi_table
            .create_file_group_reader_with_options(None, empty_options())
            .await
            .unwrap()
            .read_file_slice_from_paths(
                "a079bdb3-731c-4894-b855-abfcd6921007-0_0-203-274_20240418173551906.parquet",
                Vec::<&str>::new(),
                &ReadOptions::new(),
            )
            .await
            .unwrap();
        assert_eq!(batches.num_rows(), 4);
        assert_eq!(batches.num_columns(), 21);
    }

    /// Regression test: the reader handed out by the public constructor must read an
    /// evolved table with the TABLE's schema, not the base file's.
    ///
    /// This constructor is what DataFusion and the Python bindings use. It never
    /// set the data schema, so a v2 read fell back to the base file's — and a
    /// base file written before `num` was promoted from int to long forced the
    /// log's 5000000000 back into i32, returning 705032704 with no error. The
    /// exact value is asserted because that is the whole failure: not a crash,
    /// a wrong number.
    #[tokio::test]
    async fn test_public_file_group_reader_reads_with_the_table_schema() {
        use crate::config::read::HudiReadConfig;
        use hudi_test::QuickstartTripsTable;

        let table_path = QuickstartTripsTable::MorEvoPromotion.path_to_mor_avro();
        let table = Table::new(&table_path).await.unwrap();
        let options = ReadOptions::new()
            .with_hudi_option(HudiReadConfig::FileGroupReaderVersion.as_ref(), "2");

        let reader = table
            .create_file_group_reader_with_options(Some(&options), empty_options())
            .await
            .unwrap();
        let slices = table.get_file_slices(&options).await.unwrap();
        assert_eq!(slices.len(), 1, "fixture is a single non-partitioned slice");

        let batch = reader.read_file_slice(&slices[0], &options).await.unwrap();

        assert_eq!(
            batch.schema().field_with_name("num").unwrap().data_type(),
            &arrow_schema::DataType::Int64,
            "the promoted column must come back at the table's width"
        );
        let num = batch
            .column_by_name("num")
            .unwrap()
            .as_any()
            .downcast_ref::<arrow_array::Int64Array>()
            .expect("num is i64");
        let mut values: Vec<i64> = num.values().to_vec();
        values.sort_unstable();
        assert_eq!(
            values,
            vec![3, 4, 11, 5_000_000_000],
            "5000000000 must survive; reading it as i32 yielded 705032704"
        );
        assert_eq!(
            batch.schema().field_with_name("fnum").unwrap().data_type(),
            &arrow_schema::DataType::Float64,
            "the promoted float must not be narrowed back to f32 either"
        );
    }

    #[tokio::test]
    async fn hudi_table_read_snapshot_and_as_of() {
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let latest_timestamp = hudi_table.timeline.get_latest_commit_timestamp().unwrap();

        let snapshot_batches = hudi_table.read(&ReadOptions::new()).await.unwrap();
        assert!(!snapshot_batches.is_empty());
        let snapshot_rows = snapshot_batches
            .iter()
            .map(|batch| batch.num_rows())
            .sum::<usize>();
        assert!(snapshot_rows > 0);

        let as_of_batches = hudi_table
            .read(&ReadOptions::new().with_as_of_timestamp(&latest_timestamp))
            .await
            .unwrap();
        assert!(!as_of_batches.is_empty());
    }

    #[tokio::test]
    async fn empty_hudi_table_read_apis_return_empty() {
        use futures::StreamExt;

        let base_url = SampleTable::V6Empty.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();

        let snapshot_batches = hudi_table.read(&ReadOptions::new()).await.unwrap();
        assert!(snapshot_batches.is_empty());

        let incremental_batches = hudi_table
            .read(
                &ReadOptions::new()
                    .with_query_type(QueryType::Incremental)
                    .with_start_timestamp(EARLIEST_START_TIMESTAMP),
            )
            .await
            .unwrap();
        assert!(incremental_batches.is_empty());

        let mut snapshot_stream = hudi_table.read_stream(&ReadOptions::new()).await.unwrap();
        assert!(snapshot_stream.next().await.is_none());
    }

    #[tokio::test]
    async fn hudi_table_read_snapshot_stream_returns_batches_with_options() -> Result<()> {
        use futures::TryStreamExt;

        let base_url = SampleTable::V6SimplekeygenNonhivestyle.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let options = ReadOptions::new()
            .with_filters([("byteField", ">=", "10")])?
            .with_projection(["id"])
            .with_batch_size(2)?;

        let stream = hudi_table.read_stream(&options).await.unwrap();
        let batches = stream.try_collect::<Vec<_>>().await.unwrap();
        assert!(!batches.is_empty());
        assert!(batches[0].column_by_name("id").is_some());
        Ok(())
    }

    /// The streaming read carries the completion gate too.
    ///
    /// It reaches the gate through its own pass-through, separate from the eager read's, and
    /// it is the path DataFusion and the Python binding use. The gold sweep only exercises
    /// the eager one, so without this a regression that disarmed the gate for every streaming
    /// read would leave the whole suite green — which is exactly what a mutation of the
    /// streaming pass-through did before this test existed.
    ///
    /// The fixture's orphaned delta commit sets `rider = 'ORPHANED-B'` at `ts = 300`, above
    /// every other row's ordering value, so an admitted orphan wins the row outright rather
    /// than losing the merge for an unrelated reason.
    #[tokio::test]
    async fn hudi_table_read_stream_excludes_an_uncommitted_instants_blocks() -> Result<()> {
        use arrow_array::Array;
        use futures::TryStreamExt;
        use hudi_test::QuickstartTripsTable;

        let base_url = QuickstartTripsTable::MorUncommittedLogV6.url_to_mor_avro();
        let hudi_table = Table::new(base_url.path()).await.unwrap();

        let stream = hudi_table.read_stream(&ReadOptions::new()).await.unwrap();
        let batches = stream.try_collect::<Vec<_>>().await.unwrap();

        let riders: Vec<String> = batches
            .iter()
            .filter_map(|b| b.column_by_name("rider").cloned())
            .flat_map(|c| {
                let arr = c
                    .as_any()
                    .downcast_ref::<arrow_array::StringArray>()
                    .expect("rider is a string column")
                    .clone();
                (0..arr.len()).map(move |i| arr.value(i).to_string())
            })
            .collect();

        assert!(
            !riders.is_empty(),
            "the streaming read returned no rows, so it cannot show the gate ran"
        );
        assert!(
            !riders.iter().any(|r| r == "ORPHANED-B"),
            "the streaming read merged a block from an instant that never completed: {riders:?}"
        );
        assert!(
            riders.iter().any(|r| r == "rider-B"),
            "the base row the orphan would have overwritten must survive: {riders:?}"
        );
        Ok(())
    }

    #[tokio::test]
    async fn hudi_table_read_snapshot_stream_returns_empty_when_no_file_slices_match_filters()
    -> Result<()> {
        use futures::StreamExt;

        let base_url = SampleTable::V6SimplekeygenNonhivestyle.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let options = ReadOptions::new().with_filters([("byteField", "=", "999")])?;

        let mut stream = hudi_table.read_stream(&options).await.unwrap();
        assert!(stream.next().await.is_none());
        Ok(())
    }

    #[tokio::test]
    async fn hudi_table_read_with_incremental_query_type() {
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let batches = hudi_table
            .read(
                &ReadOptions::new()
                    .with_query_type(QueryType::Incremental)
                    .with_start_timestamp(EARLIEST_START_TIMESTAMP),
            )
            .await
            .unwrap();
        assert!(!batches.is_empty());
    }

    #[tokio::test]
    async fn hudi_table_read_dispatches_on_query_type() {
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();

        // Default: Snapshot.
        let snapshot = hudi_table.read(&ReadOptions::new()).await.unwrap();
        assert!(!snapshot.is_empty());

        // Explicit Incremental returns the same shape as the dedicated shortcut.
        let incremental = hudi_table
            .read(
                &ReadOptions::new()
                    .with_query_type(QueryType::Incremental)
                    .with_start_timestamp(EARLIEST_START_TIMESTAMP),
            )
            .await
            .unwrap();
        assert!(!incremental.is_empty());

        // Shortcuts override query_type set on the input options.
        let forced_snapshot = hudi_table
            .read(&ReadOptions::new().with_query_type(QueryType::Incremental))
            .await
            .unwrap();
        assert!(!forced_snapshot.is_empty());
    }

    #[tokio::test]
    async fn hudi_table_read_stream_errors_on_incremental() {
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let result = hudi_table
            .read_stream(&ReadOptions::new().with_query_type(QueryType::Incremental))
            .await;
        match result {
            Ok(_) => panic!("incremental streaming must error"),
            Err(e) => {
                assert!(matches!(e, CoreError::Unsupported(_)));
                assert!(e.to_string().contains("not yet supported"));
            }
        }
    }

    #[tokio::test]
    async fn hudi_table_get_file_slices_dispatches_on_query_type() {
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();

        let snapshot_slices = hudi_table
            .get_file_slices(&ReadOptions::new())
            .await
            .unwrap();
        let incremental_slices = hudi_table
            .get_file_slices(
                &ReadOptions::new()
                    .with_query_type(QueryType::Incremental)
                    .with_start_timestamp(EARLIEST_START_TIMESTAMP),
            )
            .await
            .unwrap();
        assert!(!snapshot_slices.is_empty());
        assert!(!incremental_slices.is_empty());
    }

    #[tokio::test]
    async fn read_with_invalid_as_of_timestamp_errors() {
        // `as_of_timestamp` is parsed via `format_timestamp` before any IO. A
        // malformed value must surface as `TimestampParsingError` rather than
        // silently being treated as "latest" or sliced into a no-op result.
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let options = ReadOptions::new().with_as_of_timestamp("not-a-timestamp");

        let snapshot_err = hudi_table.read(&options).await.unwrap_err();
        assert!(
            matches!(snapshot_err, CoreError::TimestampParsingError(_)),
            "expected TimestampParsingError, got: {snapshot_err}"
        );

        let stream_result = hudi_table.read_stream(&options).await;
        match stream_result {
            Ok(_) => panic!("read_stream must propagate the parse error synchronously"),
            Err(e) => assert!(
                matches!(e, CoreError::TimestampParsingError(_)),
                "expected TimestampParsingError on stream path, got: {e}"
            ),
        }
    }

    #[tokio::test]
    async fn hudi_table_read_options_hudi_options_plumbed_to_reader() {
        // Per-read hudi_options should override table-level configs without
        // mutating the table. Here we set the per-read StartTimestamp
        // via hudi_options on a snapshot read; the read should still succeed.
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let options =
            ReadOptions::new().with_hudi_option(HudiReadConfig::StartTimestamp.as_ref(), "0");
        let batches = hudi_table.read(&options).await.unwrap();
        assert!(!batches.is_empty());
    }

    #[tokio::test]
    async fn hudi_table_get_file_slices_as_of_replacecommit() {
        // Insert-overwrite-table replacecommit: as_of before vs at the replacecommit
        // returns different slice sets. Splitter behavior is covered by
        // util::collection::split_into_chunks tests.
        let base_url = SampleTable::V6SimplekeygenNonhivestyleOverwritetable.url_to_mor_parquet();
        let hudi_table = Table::new(base_url.path()).await.unwrap();

        // before replacecommit
        let second_latest_timestamp = "20250121000656060";
        let file_slices = hudi_table
            .get_file_slices(&ReadOptions::new().with_as_of_timestamp(second_latest_timestamp))
            .await
            .unwrap();
        assert_eq!(file_slices.len(), 3);
        let p10: Vec<_> = file_slices
            .iter()
            .filter(|f| f.partition_path == "10")
            .collect();
        assert_eq!(p10.len(), 1, "Partition 10 should have 1 file slice");
        let file_slice = p10[0];
        assert_eq!(
            file_slice.base_file.as_ref().unwrap().file_name(),
            "92e64357-e4d1-4639-a9d3-c3535829d0aa-0_1-53-79_20250121000647668.parquet"
        );
        assert_eq!(file_slice.log_files.len(), 1);
        assert_eq!(
            file_slice.log_files.iter().next().unwrap().file_name(),
            ".92e64357-e4d1-4639-a9d3-c3535829d0aa-0_20250121000647668.log.1_0-73-101"
        );

        // as of replacecommit
        let latest_timestamp = "20250121000702475";
        let file_slices = hudi_table
            .get_file_slices(&ReadOptions::new().with_as_of_timestamp(latest_timestamp))
            .await
            .unwrap();
        assert_eq!(file_slices.len(), 1);
    }

    #[tokio::test]
    async fn hudi_table_get_file_slices_as_of_timestamps() {
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();

        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let file_slices = hudi_table
            .get_file_slices(&ReadOptions::new())
            .await
            .unwrap();
        assert_eq!(
            file_slices
                .iter()
                .map(|f| f.base_file_relative_path().unwrap().unwrap())
                .collect::<Vec<_>>(),
            vec!["a079bdb3-731c-4894-b855-abfcd6921007-0_0-203-274_20240418173551906.parquet",]
        );

        // as of the latest timestamp
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let file_slices = hudi_table
            .get_file_slices(&ReadOptions::new().with_as_of_timestamp("20240418173551906"))
            .await
            .unwrap();
        assert_eq!(
            file_slices
                .iter()
                .map(|f| f.base_file_relative_path().unwrap().unwrap())
                .collect::<Vec<_>>(),
            vec!["a079bdb3-731c-4894-b855-abfcd6921007-0_0-203-274_20240418173551906.parquet",]
        );

        // as of just smaller than the latest timestamp
        let hudi_table = Table::new_with_options(base_url.path(), empty_options())
            .await
            .unwrap();
        let file_slices = hudi_table
            .get_file_slices(&ReadOptions::new().with_as_of_timestamp("20240418173551905"))
            .await
            .unwrap();
        assert_eq!(
            file_slices
                .iter()
                .map(|f| f.base_file_relative_path().unwrap().unwrap())
                .collect::<Vec<_>>(),
            vec!["a079bdb3-731c-4894-b855-abfcd6921007-0_0-182-253_20240418173550988.parquet",]
        );

        // as of non-exist old timestamp
        let hudi_table = Table::new_with_options(base_url.path(), empty_options())
            .await
            .unwrap();
        let file_slices = hudi_table
            .get_file_slices(&ReadOptions::new().with_as_of_timestamp("19700101000000"))
            .await
            .unwrap();
        assert_eq!(
            file_slices
                .iter()
                .map(|f| f.base_file_relative_path().unwrap().unwrap())
                .collect::<Vec<_>>(),
            Vec::<String>::new()
        );
    }

    #[tokio::test]
    async fn empty_hudi_table_get_file_slices_incremental() {
        let base_url = SampleTable::V6Empty.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let file_slices = hudi_table
            .get_file_slices(
                &ReadOptions::new()
                    .with_query_type(QueryType::Incremental)
                    .with_start_timestamp(EARLIEST_START_TIMESTAMP),
            )
            .await
            .unwrap();
        assert!(file_slices.is_empty())
    }

    #[tokio::test]
    async fn hudi_table_get_file_slices_incremental() {
        let base_url = SampleTable::V6SimplekeygenNonhivestyleOverwritetable.url_to_mor_parquet();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        let mut file_slices = hudi_table
            .get_file_slices(
                &ReadOptions::new()
                    .with_query_type(QueryType::Incremental)
                    .with_end_timestamp("20250121000656060"),
            )
            .await
            .unwrap();
        assert_eq!(file_slices.len(), 3);

        file_slices.sort_unstable_by_key(|f| f.partition_path.clone());

        let file_slice_0 = &file_slices[0];
        assert_eq!(file_slice_0.partition_path, "10");
        assert_eq!(
            file_slice_0.file_id(),
            "92e64357-e4d1-4639-a9d3-c3535829d0aa-0"
        );
        assert_eq!(file_slice_0.log_files.len(), 1);

        let file_slice_1 = &file_slices[1];
        assert_eq!(file_slice_1.partition_path, "20");
        assert_eq!(
            file_slice_1.file_id(),
            "d49ae379-4f20-4549-8e23-a5f9604412c0-0"
        );
        assert!(file_slice_1.log_files.is_empty());

        let file_slice_2 = &file_slices[2];
        assert_eq!(file_slice_2.partition_path, "30");
        assert_eq!(
            file_slice_2.file_id(),
            "de3550df-e12c-4591-9335-92ff992258a2-0"
        );
        assert!(file_slice_2.log_files.is_empty());

        // FileMetadata is populated for incremental queries (issue #401).
        // size comes from HoodieWriteStat.fileSizeInBytes; byte_size and num_records
        // are estimated from the cached FileStatsEstimator (seeded from a sample
        // base file in commit metadata at or before end_timestamp).
        let m0 = file_slice_0
            .base_file
            .as_ref()
            .unwrap()
            .file_metadata
            .as_ref()
            .unwrap();
        assert_eq!(m0.size, 440878);
        assert_eq!(m0.byte_size, 326703);
        assert_eq!(m0.num_records, 458);

        let m1 = file_slice_1
            .base_file
            .as_ref()
            .unwrap()
            .file_metadata
            .as_ref()
            .unwrap();
        assert_eq!(m1.size, 440616);
        assert_eq!(m1.byte_size, 326509);
        assert_eq!(m1.num_records, 458);

        let m2 = file_slice_2
            .base_file
            .as_ref()
            .unwrap()
            .file_metadata
            .as_ref()
            .unwrap();
        assert_eq!(m2.size, 440638);
        assert_eq!(m2.byte_size, 326525);
        assert_eq!(m2.num_records, 458);
    }

    #[tokio::test]
    async fn hudi_table_get_file_paths_for_simple_keygen_non_hive_style() {
        let base_url = SampleTable::V6SimplekeygenNonhivestyle.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        assert_eq!(hudi_table.timeline.completed_commits.len(), 2);

        let partition_filters = &[];
        let actual = get_file_paths_with_filters(&hudi_table, partition_filters)
            .await
            .unwrap()
            .into_iter()
            .collect::<HashSet<_>>();
        let expected = [
            "10/97de74b1-2a8e-4bb7-874c-0a74e1f42a77-0_0-119-166_20240418172804498.parquet",
            "20/76e0556b-390d-4249-b7ad-9059e2bc2cbd-0_0-98-141_20240418172802262.parquet",
            "30/6db57019-98ee-480e-8eb1-fb3de48e1c24-0_1-119-167_20240418172804498.parquet",
        ]
        .map(|f| join_url_segments(&base_url, &[f]).unwrap().to_string())
        .into_iter()
        .collect::<HashSet<_>>();
        assert_eq!(actual, expected);

        let filters = [("byteField", ">=", "10"), ("byteField", "<", "30")];
        let actual = get_file_paths_with_filters(&hudi_table, &filters)
            .await
            .unwrap()
            .into_iter()
            .collect::<HashSet<_>>();
        let expected = [
            "10/97de74b1-2a8e-4bb7-874c-0a74e1f42a77-0_0-119-166_20240418172804498.parquet",
            "20/76e0556b-390d-4249-b7ad-9059e2bc2cbd-0_0-98-141_20240418172802262.parquet",
        ]
        .map(|f| join_url_segments(&base_url, &[f]).unwrap().to_string())
        .into_iter()
        .collect::<HashSet<_>>();
        assert_eq!(actual, expected);

        let actual = get_file_paths_with_filters(&hudi_table, &[("byteField", ">", "30")])
            .await
            .unwrap()
            .into_iter()
            .collect::<HashSet<_>>();
        let expected = HashSet::new();
        assert_eq!(actual, expected);
    }

    #[tokio::test]
    async fn hudi_table_get_file_paths_for_complex_keygen_hive_style() {
        let base_url = SampleTable::V6ComplexkeygenHivestyle.url_to_cow();
        let hudi_table = Table::new(base_url.path()).await.unwrap();
        assert_eq!(hudi_table.timeline.completed_commits.len(), 2);

        let partition_filters = &[];
        let actual = get_file_paths_with_filters(&hudi_table, partition_filters)
            .await
            .unwrap()
            .into_iter()
            .collect::<HashSet<_>>();
        let expected= [
            "byteField=10/shortField=300/a22e8257-e249-45e9-ba46-115bc85adcba-0_0-161-223_20240418173235694.parquet",
            "byteField=20/shortField=100/bb7c3a45-387f-490d-aab2-981c3f1a8ada-0_0-140-198_20240418173213674.parquet",
            "byteField=30/shortField=100/4668e35e-bff8-4be9-9ff2-e7fb17ecb1a7-0_1-161-224_20240418173235694.parquet",
        ]
            .map(|f| { join_url_segments(&base_url, &[f]).unwrap().to_string() })
            .into_iter()
            .collect::<HashSet<_>>();
        assert_eq!(actual, expected);

        let filters = [
            ("byteField", ">=", "10"),
            ("byteField", "<", "20"),
            ("shortField", "!=", "100"),
        ];
        let actual = get_file_paths_with_filters(&hudi_table, &filters)
            .await
            .unwrap()
            .into_iter()
            .collect::<HashSet<_>>();
        let expected = [
            "byteField=10/shortField=300/a22e8257-e249-45e9-ba46-115bc85adcba-0_0-161-223_20240418173235694.parquet",
        ]
            .map(|f| { join_url_segments(&base_url, &[f]).unwrap().to_string() })
            .into_iter()
            .collect::<HashSet<_>>();
        assert_eq!(actual, expected);

        let filters = [("byteField", ">=", "20"), ("shortField", "=", "300")];
        let actual = get_file_paths_with_filters(&hudi_table, &filters)
            .await
            .unwrap()
            .into_iter()
            .collect::<HashSet<_>>();
        let expected = HashSet::new();
        assert_eq!(actual, expected);
    }

    #[tokio::test]
    async fn test_get_or_init_estimator_returns_none_for_non_parquet_format() {
        let base_url = SampleTable::V9TxnsSimpleMeta.url_to_cow();
        let table = Table::new_with_options(
            base_url.path(),
            [
                (BaseFileFormat.as_ref(), BaseFileFormatValue::HFile.as_ref()),
                (HudiInternalConfig::SkipConfigValidation.as_ref(), "true"),
            ],
        )
        .await
        .unwrap();
        let latest_ts = table.timeline.get_latest_commit_timestamp().unwrap();
        assert!(table.get_or_init_estimator(&latest_ts).await.is_none());
    }

    #[tokio::test]
    async fn test_get_or_init_estimator_retries_after_early_timestamp_without_sample() {
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let table = Table::new(base_url.path()).await.unwrap();

        // No completed commits at or before this timestamp, so no sample file can be found.
        let early_ts = "19700101000000";
        assert!(table.get_or_init_estimator(early_ts).await.is_none());

        // A later request should still be able to initialize and cache the estimator,
        // and the subsequent early-timestamp call must hit the same cached instance.
        let latest_ts = table.timeline.get_latest_commit_timestamp().unwrap();
        let initialized = table.get_or_init_estimator(&latest_ts).await.unwrap();
        let cached = table.get_or_init_estimator(early_ts).await.unwrap();
        assert!(std::ptr::eq(initialized, cached));
    }

    /// Regression test: the slice fan-out honours
    /// `hoodie.read.file.slice.read.concurrency`.
    ///
    /// Both read paths used a bare `try_join_all` over every file slice, so the
    /// knob that already bounds the DataFusion scan was ignored here — the direct
    /// and Python paths issued one concurrent read per slice with no ceiling. The
    /// bogus-value half is what keeps this honest: it fails if the config stops
    /// being consulted, even when the fixture is too small to show a difference
    /// in row counts.
    #[tokio::test]
    async fn test_read_honours_the_file_slice_concurrency_bound() {
        use crate::config::read::HudiReadConfig;

        let base_url = SampleTable::V6ComplexkeygenHivestyle.url_to_cow();

        let unbounded = Table::new(base_url.path()).await.unwrap();
        let expected = unbounded
            .read(&ReadOptions::new())
            .await
            .unwrap()
            .iter()
            .map(|b| b.num_rows())
            .sum::<usize>();
        assert!(expected > 0, "fixture must return rows");

        // A ceiling of one serialises the reads and must not change the answer.
        let serial = Table::new_with_options(
            base_url.path(),
            [(HudiReadConfig::FileSliceReadConcurrency.as_ref(), "1")],
        )
        .await
        .unwrap();
        assert_eq!(serial.bounded_read_concurrency(&[]), 1);
        let serial_rows = serial
            .read(&ReadOptions::new())
            .await
            .unwrap()
            .iter()
            .map(|b| b.num_rows())
            .sum::<usize>();
        assert_eq!(
            serial_rows, expected,
            "bounding concurrency must not change what is read"
        );

        // A ceiling of zero would stall the fan-out forever. It never reaches it:
        // the value is rejected when the read resolves its options.
        let zero = Table::new_with_options(
            base_url.path(),
            [(HudiReadConfig::FileSliceReadConcurrency.as_ref(), "0")],
        )
        .await
        .unwrap();
        let err = zero
            .read(&ReadOptions::new())
            .await
            .expect_err("a concurrency ceiling of zero must be rejected, not used");
        assert!(
            err.to_string()
                .contains(HudiReadConfig::FileSliceReadConcurrency.as_ref()),
            "the error must name the offending key, got: {err}"
        );
        // And the ceiling the fan-out would have used is never zero regardless.
        assert!(zero.bounded_read_concurrency(&[]) >= 1);
    }

    /// Regression test: the file group reader version set at TABLE level must
    /// actually select the reader, and a typo in it must fail the read.
    ///
    /// `Table::build` used to drop every `hoodie.read.*` key, so both halves of
    /// this were broken in the same way and were indistinguishable: a table-level
    /// version 2 read with version 1, and a table-level `not-a-version` also read
    /// with version 1 instead of reporting the bad value. The bogus half is what
    /// makes this test non-vacuous — it fails if the key stops reaching the
    /// reader, even if the two versions happen to agree on the fixture.
    #[tokio::test]
    async fn test_table_level_reader_version_reaches_the_reader() {
        use crate::config::read::HudiReadConfig;
        use hudi_test::QuickstartTripsTable;
        let table_path = QuickstartTripsTable::MorEvoPromotion.path_to_mor_avro();

        // A promoted column reads as i64 only when version 2 actually ran;
        // version 1 cannot widen it (see the gold-parity known list).
        let table = Table::new_with_options(
            &table_path,
            [(HudiReadConfig::FileGroupReaderVersion.as_ref(), "2")],
        )
        .await
        .unwrap();
        let batches = table.read(&ReadOptions::new()).await.unwrap();
        assert_eq!(
            batches[0]
                .schema()
                .field_with_name("num")
                .unwrap()
                .data_type(),
            &arrow_schema::DataType::Int64,
            "a table-level reader version of 2 must select version 2"
        );

        // And a value that is not a version must be reported, not ignored.
        let table = Table::new_with_options(
            &table_path,
            [(
                HudiReadConfig::FileGroupReaderVersion.as_ref(),
                "not-a-version",
            )],
        )
        .await
        .unwrap();
        let err = table
            .read(&ReadOptions::new())
            .await
            .expect_err("a bogus table-level reader version must fail the read");
        assert!(
            err.to_string().contains("not-a-version"),
            "the error must name the offending value, got: {err}"
        );
    }

    /// Regression test: the start bound handed to the log scan is a real instant, not
    /// a synthesized "one tick below" string.
    ///
    /// It used to be the earliest admitted instant decremented as an integer, to
    /// bring it back inside an exclusive bound. But an instant time is a
    /// `yyyyMMddHHmmssSSS` string and the decrement borrows across its fields:
    /// `20250713010500000` becomes `20250713010499999`, whose `ss` is 99. That
    /// sorts correctly, so the log scan looked right, and fails
    /// `Instant::parse_datetime`, so any read whose earliest admitted commit
    /// landed on a whole second died with "Invalid epoch millis".
    ///
    /// No fixture has a commit on a whole second, which is why this went unseen.
    /// So the assertion is the property that makes the parse safe for every
    /// fixture rather than a reproduction of the one input that broke: the bound
    /// is an instant the timeline actually contains, so it parses by
    /// construction.
    /// Regression test: a table-built reader carries the timeline's
    /// committed/inflight sets, so the log-block scan can gate on them.
    ///
    /// The gate exists to skip blocks from an instant that never completed — the
    /// straddling case where a writer is still inflight when a later one
    /// commits, so its blocks sort below the latest instant and pass every other
    /// gate. It was inert for every read through this crate because nothing
    /// populated its inputs.
    #[tokio::test]
    async fn test_table_reader_carries_the_completion_gate_inputs() {
        use hudi_test::SampleTable;

        let base_url = SampleTable::V6Nonpartitioned.url_to_mor_parquet();
        let table = Table::new(base_url.path()).await.unwrap();

        let inputs = table.timeline.completion_gate_inputs();
        assert!(
            !inputs.completed_instants.is_empty(),
            "the fixture's completed commits must reach the gate"
        );
        assert!(
            inputs.archived_boundary.is_some(),
            "the archival boundary is the gate's second half — an archived \
             instant is committed by definition"
        );

        // And the reader the read paths use actually receives them.
        let reader = table
            .create_file_group_reader_with_options(None, empty_options())
            .await
            .unwrap();
        assert!(
            reader.has_completion_gate_inputs(),
            "a reader built from a table must be able to gate the log scan"
        );
    }

    /// From table version 8 the timeline records completion times, so a log file
    /// whose delta commit never completed is already dropped when the file slice
    /// is built. Java stops applying the per-block gate there
    /// (`BaseHoodieLogRecordReader`, `tableVersion.lesserThan(EIGHT)`), and so
    /// does this. Paired with the version-6 test above, the two pin the
    /// condition rather than only the armed case.
    #[tokio::test]
    async fn test_completion_gate_is_not_armed_from_table_version_eight() {
        use hudi_test::QuickstartTripsTable;

        let table_path = QuickstartTripsTable::V9MorNonpart3Commits.path_to_mor_avro();
        let table = Table::new(&table_path).await.unwrap();
        assert!(
            table.completion_gate_inputs().unwrap().is_none(),
            "a version 9 table must not arm the per-block gate"
        );

        let reader = table
            .create_file_group_reader_with_options(None, empty_options())
            .await
            .unwrap();
        assert!(
            !reader.has_completion_gate_inputs(),
            "the slice already excludes an uncommitted log file on this layout"
        );
    }

    #[tokio::test]
    async fn test_resolve_incremental_window_start_bound_is_a_real_instant() {
        use crate::config::internal::HudiInternalConfig;
        use crate::config::read::HudiReadConfig;
        use hudi_test::QuickstartTripsTable;

        let table_path = QuickstartTripsTable::V9MorCompactedIncremental.path_to_mor_avro();
        let table = Table::new(&table_path).await.unwrap();

        // A window spanning the whole fixture, so every one of its delta commits
        // is admitted and the earliest of them becomes the start bound.
        let prepared = table
            .prepare_reader_options(
                &ReadOptions::new()
                    .with_query_type(QueryType::Incremental)
                    .with_start_timestamp("20260807223520000")
                    .with_end_timestamp("20260807223532000"),
            )
            .unwrap();

        let start = prepared
            .hudi_options
            .get(HudiReadConfig::StartTimestamp.as_ref())
            .expect("an incremental read pins a start bound");

        // It sits strictly below the earliest admitted instant, so the
        // exclusive bound still admits it...
        let admitted: Vec<&str> = prepared
            .hudi_options
            .get(HudiInternalConfig::IncrementalInstantTimes.as_ref())
            .expect("the admitted instants travel with the read")
            .split(',')
            .collect();
        assert!(
            start.as_str() < admitted[0],
            "the start bound must sort below the earliest admitted instant, \
             got {start} vs {}",
            admitted[0]
        );

        // ...and it is a real time, which the integer decrement was not.
        crate::timeline::instant::Instant::parse_datetime(start, "UTC")
            .expect("the start bound must parse as an instant");
    }

    /// The decrement steps in the time domain, so it never produces an
    /// out-of-range field. Each input below is a boundary the integer decrement
    /// borrowed across.
    #[test]
    fn test_instant_time_minus_one_stays_a_valid_instant() {
        for (input, want) in [
            // whole second — the case that broke: `ss` became 99
            ("20250713010500000", "20250713010459999"),
            // whole minute, and whole hour: two and three fields borrow
            ("20250713010000000", "20250713005959999"),
            ("20250713000000000", "20250712235959999"),
            // midnight: the date itself steps back
            ("20250101000000000", "20241231235959999"),
            // no borrow at all — unchanged behaviour
            ("20250713010501000", "20250713010500999"),
            // second-precision instants borrow the same way
            ("20250713010500", "20250713010459"),
        ] {
            let got = instant_time_minus_one(input);
            assert_eq!(got, want, "decrementing {input}");
            assert!(got.as_str() < input, "{got} must sort below {input}");
            crate::timeline::instant::Instant::parse_datetime(&got, "UTC")
                .unwrap_or_else(|e| panic!("{got} must parse as an instant: {e}"));
        }

        // A metadata table's epoch-millis timestamps have no date fields, so
        // they keep the integer decrement; zero stays put rather than wrapping.
        assert_eq!(
            instant_time_minus_one("00000000000000001"),
            "00000000000000000"
        );
        assert_eq!(
            instant_time_minus_one("00000000000000000"),
            "00000000000000000"
        );
    }

    /// Regression test: `read` and `read_stream` must agree on the schema of an
    /// evolved table.
    ///
    /// `read_stream` resolved the table's schema and then handed it to a base-file
    /// path that ignored it, so read-optimized streaming returned the base file's
    /// narrow types — i32/f32 where the eager read gave i64/f64. A caller that
    /// declared the table's schema up front (the DataFusion scan does) then got
    /// batches that did not match its own plan.
    #[tokio::test]
    async fn test_stream_and_eager_agree_on_an_evolved_schema() {
        use crate::config::read::HudiReadConfig;
        use futures::StreamExt;
        use hudi_test::QuickstartTripsTable;

        let table_path = QuickstartTripsTable::MorEvoPromotion.path_to_mor_avro();
        let table = Table::new(&table_path).await.unwrap();
        let table_schema = table.get_schema().await.unwrap();

        // Read-optimized is the case that diverged: it returns before the
        // reader version is consulted, so it took the base-file-only streaming
        // path.
        for read_optimized in [true, false] {
            let mut options = ReadOptions::new()
                .with_hudi_option(HudiReadConfig::FileGroupReaderVersion.as_ref(), "2");
            if read_optimized {
                options =
                    options.with_hudi_option(HudiReadConfig::UseReadOptimizedMode.as_ref(), "true");
            }

            let eager = table.read(&options).await.unwrap();
            let mut stream = table.read_stream(&options).await.unwrap();
            let mut streamed = Vec::new();
            while let Some(batch) = stream.next().await {
                streamed.push(batch.unwrap());
            }

            for column in ["num", "fnum"] {
                let want = table_schema.field_with_name(column).unwrap().data_type();
                assert_eq!(
                    eager[0]
                        .schema()
                        .field_with_name(column)
                        .unwrap()
                        .data_type(),
                    want,
                    "eager read of '{column}' (read_optimized={read_optimized})"
                );
                assert_eq!(
                    streamed[0]
                        .schema()
                        .field_with_name(column)
                        .unwrap()
                        .data_type(),
                    want,
                    "streamed read of '{column}' (read_optimized={read_optimized}) must match \
                     the eager read and the table's declared schema"
                );
            }

            let eager_rows: usize = eager.iter().map(|b| b.num_rows()).sum();
            let streamed_rows: usize = streamed.iter().map(|b| b.num_rows()).sum();
            assert_eq!(
                eager_rows, streamed_rows,
                "row counts must agree (read_optimized={read_optimized})"
            );
        }
    }

    /// A column a later writer added must appear in a streamed read too, and a
    /// projection naming it must not fail just because the base file predates it.
    #[tokio::test]
    async fn test_stream_null_fills_a_column_added_after_the_base_file() {
        use crate::config::read::HudiReadConfig;
        use futures::StreamExt;
        use hudi_test::QuickstartTripsTable;

        let table_path = QuickstartTripsTable::MorEvoAddCol.path_to_mor_avro();
        let table = Table::new(&table_path).await.unwrap();
        let options = ReadOptions::new()
            .with_hudi_option(HudiReadConfig::FileGroupReaderVersion.as_ref(), "2")
            .with_hudi_option(HudiReadConfig::UseReadOptimizedMode.as_ref(), "true")
            .with_projection(["key", "extra"]);

        let mut stream = table.read_stream(&options).await.unwrap();
        let mut streamed = Vec::new();
        while let Some(batch) = stream.next().await {
            streamed.push(batch.unwrap());
        }

        let schema = streamed[0].schema();
        let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(
            names,
            vec!["key", "extra"],
            "the projection must be honored even though the base file has no 'extra'"
        );
        // The base file predates the column, so every value is null.
        let extra = streamed[0].column_by_name("extra").unwrap();
        assert_eq!(
            extra.null_count(),
            extra.len(),
            "a column added after this base file must read as null, not error"
        );
    }

    /// Version 2 streams a merged slice instead of materializing it, and must
    /// return exactly what the eager read returns.
    ///
    /// The streaming path used to collect-and-merge for any slice with log files
    /// and wrap the result in a one-item stream — `batch_size` was ignored and
    /// output memory was the whole merged slice, even though version 2 already
    /// had a bounded row-group-at-a-time form with no caller. Asserting more than
    /// one batch is what keeps this honest: with the old fallback it was always
    /// exactly one.
    /// A real merge-on-read read completes on a single-worker runtime without
    /// starving a task sharing it, and returns the same rows as the eager read.
    ///
    /// `current_thread` is the point: one worker, so a read that monopolised it
    /// outright - or that reached the deleted blocking bridge from inside the
    /// runtime - would deadlock or starve the ticker completely, and this test
    /// would hang or fail. It is a liveness check over the whole production path:
    /// table open, log scan and gating, log-content fetch, base file, merge.
    ///
    /// **What it does NOT prove, established by mutation rather than assumed:**
    /// it cannot detect blocking. Making `StorageReader::fill_window` sleep
    /// 200 ms on the worker thread - twice, inside the log read - leaves this
    /// test passing, because the assertions only require the ticker to advance
    /// somewhere in each window and a real read has many other await points that
    /// mask a blocked one. A pass here is not evidence that nothing blocks.
    ///
    /// What does pin non-blocking, and where:
    ///   - the merge loop: `merge_stream_lets_other_tasks_run_between_chunks`,
    ///     which owns every await in its base source and so cannot be masked;
    ///   - the log read: no `block_on` remains in this crate outside test
    ///     helpers - a structural property, asserted by no test.
    #[tokio::test]
    async fn test_a_merge_on_read_read_runs_on_a_single_worker_runtime() {
        use crate::config::read::HudiReadConfig;
        use futures::StreamExt;
        use hudi_test::QuickstartTripsTable;
        use std::sync::Arc;
        use std::sync::atomic::{AtomicUsize, Ordering};

        let ticks = Arc::new(AtomicUsize::new(0));
        let counter = ticks.clone();
        let ticker = tokio::spawn(async move {
            loop {
                counter.fetch_add(1, Ordering::SeqCst);
                tokio::task::yield_now().await;
            }
        });

        let table_path = QuickstartTripsTable::V8Trips8I3U1D.path_to_mor_avro();
        let table = Table::new(&table_path).await.unwrap();
        let options = ReadOptions::new()
            .with_hudi_option(HudiReadConfig::FileGroupReaderVersion.as_ref(), "2");

        let mut stream = table.read_stream(&options).await.unwrap();
        let before_first = ticks.load(Ordering::SeqCst);

        let first = stream
            .next()
            .await
            .expect("the fixture yields at least one chunk")
            .unwrap();
        let after_first = ticks.load(Ordering::SeqCst);
        assert!(
            after_first > before_first,
            "the ticker never ran while the log scan and base open happened \
             ({before_first} -> {after_first}); that phase starved the only worker"
        );

        let mut rows = first.num_rows();
        let mut chunks = 1usize;
        while let Some(batch) = stream.next().await {
            rows += batch.unwrap().num_rows();
            chunks += 1;
        }
        let after_all = ticks.load(Ordering::SeqCst);
        assert!(
            after_all > after_first,
            "the ticker did not advance across the remaining {} chunk(s) \
             ({after_first} -> {after_all}); the merge held the only worker thread",
            chunks - 1
        );

        // The read still has to be a read.
        assert!(rows > 0, "the fixture must return rows");
        let eager: usize = table
            .read(&options)
            .await
            .unwrap()
            .iter()
            .map(|b| b.num_rows())
            .sum();
        assert_eq!(rows, eager, "the streamed read must return the same rows");

        ticker.abort();
    }

    #[tokio::test]
    async fn test_stream_merges_a_slice_incrementally_and_matches_the_eager_read() {
        use crate::config::read::HudiReadConfig;
        use futures::StreamExt;
        use hudi_test::QuickstartTripsTable;

        let table_path = QuickstartTripsTable::V8Trips8I3U1D.path_to_mor_avro();
        let table = Table::new(&table_path).await.unwrap();
        let options = ReadOptions::new()
            .with_hudi_option(HudiReadConfig::FileGroupReaderVersion.as_ref(), "2")
            .with_hudi_option(HudiReadConfig::StreamBatchSize.as_ref(), "2");

        let eager = table.read(&options).await.unwrap();
        let eager_rows: usize = eager.iter().map(|b| b.num_rows()).sum();
        assert!(eager_rows > 2, "fixture must exceed one batch to be a test");

        let mut stream = table.read_stream(&options).await.unwrap();
        let mut streamed = Vec::new();
        while let Some(batch) = stream.next().await {
            streamed.push(batch.unwrap());
        }
        let streamed_rows: usize = streamed.iter().map(|b| b.num_rows()).sum();

        assert_eq!(
            streamed_rows, eager_rows,
            "the streamed merge must produce the same rows as the eager one"
        );
        assert!(
            streamed.len() > 1,
            "a batch size of 2 over {eager_rows} rows must yield more than one batch; got {} \
             (one batch means the merge was materialized whole)",
            streamed.len()
        );
        assert_eq!(
            streamed[0].schema(),
            eager[0].schema(),
            "both paths must return the same schema"
        );

        // Same rows in the same ORDER, not merely the same multiset. Both entry
        // points merge the base a row group at a time, so the sequence is a
        // shared property and sorting here would stop guarding it.
        //
        // Honest limit: this fixture cannot yet detect a violation. The order
        // only diverges when one base batch holds both surviving and replaced
        // rows, and every MOR fixture in this repo updates every key, so
        // collapsing the base on one path is invisible here — verified by
        // mutation. `batch_boundaries_change_the_merged_row_order` in
        // `buffer::key_based` is what discriminates; this pins the two entry
        // points together and will discriminate given a fixture with surviving
        // base rows across more than one batch.
        let key_of = |batches: &[RecordBatch]| {
            let mut keys: Vec<String> = Vec::new();
            for batch in batches {
                let column = batch.column_by_name("uuid").unwrap();
                let strings = column
                    .as_any()
                    .downcast_ref::<arrow_array::StringArray>()
                    .unwrap();
                keys.extend((0..batch.num_rows()).map(|i| strings.value(i).to_string()));
            }
            keys
        };
        assert_eq!(
            key_of(&streamed),
            key_of(&eager),
            "the two entry points must return the same rows in the same order"
        );
    }

    /// A read config that selects WHICH read to perform is dropped at table
    /// level, so it cannot silently redirect every later read.
    #[tokio::test]
    async fn test_table_level_query_shape_config_is_not_baked_in() {
        use crate::config::read::HudiReadConfig;
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let table = Table::new_with_options(
            base_url.path(),
            [(HudiReadConfig::AsOfTimestamp.as_ref(), "20240101000000000")],
        )
        .await
        .unwrap();
        assert!(
            !table.hudi_configs.contains(HudiReadConfig::AsOfTimestamp),
            "a table-level as-of timestamp would pin every read to that instant"
        );
        // The read still resolves against the latest commit, not the dropped one.
        let batches = table.read(&ReadOptions::new()).await.unwrap();
        assert!(batches.iter().map(|b| b.num_rows()).sum::<usize>() > 0);
    }

    #[tokio::test]
    async fn test_compute_table_stats_with_mdt() {
        use hudi_test::QuickstartTripsTable;
        let table_path = QuickstartTripsTable::V8Trips8I3U1D.path_to_mor_avro();
        let table = Table::new(&table_path).await.unwrap();
        assert!(table.is_metadata_table_enabled());

        let stats = table.compute_table_stats(None).await;
        assert!(
            stats.is_some(),
            "Stats should be Some for MDT-enabled table"
        );
        let (rows, bytes) = stats.unwrap();
        assert!(rows > 0, "Should have estimated rows > 0, got {rows}");
        assert!(bytes > 0, "Should have estimated bytes > 0, got {bytes}");
    }

    #[tokio::test]
    async fn test_compute_table_stats_with_sample_mdt_table() {
        let base_url = SampleTable::V9TxnsSimpleMeta.url_to_cow();
        let table = Table::new(base_url.path()).await.unwrap();
        assert!(table.is_metadata_table_enabled());

        let stats = table.compute_table_stats(None).await;
        assert!(stats.is_some(), "Stats should be Some for sample MDT table");
        let (rows, bytes) = stats.unwrap();
        assert!(rows > 0);
        assert!(bytes > 0);
    }

    #[tokio::test]
    async fn test_compute_table_stats_returns_none_when_base_file_extension_does_not_match() {
        let base_url = SampleTable::V9TxnsSimpleMeta.url_to_cow();
        let table = Table::new_with_options(
            base_url.path(),
            [
                (BaseFileFormat.as_ref(), BaseFileFormatValue::HFile.as_ref()),
                (HudiInternalConfig::SkipConfigValidation.as_ref(), "true"),
            ],
        )
        .await
        .unwrap();
        assert!(table.is_metadata_table_enabled());
        assert!(table.compute_table_stats(None).await.is_none());
    }

    #[tokio::test]
    async fn test_compute_table_stats_without_mdt() {
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let table = Table::new(base_url.path()).await.unwrap();
        assert!(!table.is_metadata_table_enabled());

        let stats = table.compute_table_stats(None).await;
        assert!(stats.is_none(), "Stats should be None for non-MDT table");
    }

    #[tokio::test]
    async fn test_compute_table_stats_returns_none_for_incremental() {
        let base_url = SampleTable::V6SimplekeygenNonhivestyleOverwritetable.url_to_mor_parquet();
        let table = Table::new(base_url.path()).await.unwrap();

        let options = ReadOptions::new()
            .with_query_type(QueryType::Incremental)
            .with_end_timestamp("20250121000656060");
        assert!(
            table.compute_table_stats(Some(&options)).await.is_none(),
            "Incremental stats should be None"
        );
    }

    #[tokio::test]
    async fn test_clone_table_with_mdt() {
        let base_url = SampleTable::V9TxnsNonpartMeta.url_to_mor_avro();
        let table = Table::new(base_url.path()).await.unwrap();
        assert!(table.is_metadata_table_enabled());

        let cloned = table.clone();
        assert_eq!(cloned.table_name(), table.table_name());
        assert_eq!(cloned.table_type(), table.table_type());

        // Clone shares the cached metadata table via Arc<OnceCell>
        let file_slices = cloned.get_file_slices(&ReadOptions::new()).await.unwrap();
        assert!(!file_slices.is_empty());

        // compute_table_stats works on cloned table
        let stats = cloned.compute_table_stats(None).await;
        assert!(stats.is_some());
        let (rows, bytes) = stats.unwrap();
        assert!(rows > 0);
        assert!(bytes > 0);
    }

    #[tokio::test]
    async fn test_get_file_slices_falls_back_to_storage_when_metadata_table_init_fails() {
        let base_url = SampleTable::V9TxnsSimpleNometa.url_to_cow();
        let table = Table::new_with_options(base_url.path(), [("hoodie.metadata.enable", "true")])
            .await
            .unwrap();
        assert!(table.is_metadata_table_enabled());

        let file_slices = table.get_file_slices(&ReadOptions::new()).await.unwrap();
        assert!(!file_slices.is_empty());
    }

    #[tokio::test]
    async fn test_get_file_slices_with_mdt() {
        let base_url = SampleTable::V9TxnsSimpleMeta.url_to_cow();
        let table = Table::new(base_url.path()).await.unwrap();
        assert!(table.is_metadata_table_enabled());

        // This exercises the MDT code path in get_file_slices_inner:
        // metadata table init, fetch_files_partition_records, and
        // fs_view's load_file_groups with estimator
        let file_slices = table.get_file_slices(&ReadOptions::new()).await.unwrap();
        assert!(!file_slices.is_empty());

        // Verify file metadata is populated from MDT with estimated stats
        for fsl in &file_slices {
            let metadata = fsl
                .base_file
                .as_ref()
                .unwrap()
                .file_metadata
                .as_ref()
                .unwrap();
            assert!(metadata.size > 0);
        }
    }

    #[tokio::test]
    async fn test_get_file_slices_with_mdt_quickstart_table() {
        use hudi_test::QuickstartTripsTable;

        let table_path = QuickstartTripsTable::V8Trips8I3U1D.path_to_mor_avro();
        let table = Table::new(&table_path).await.unwrap();
        assert!(table.is_metadata_table_enabled());

        let metadata_table = table.get_or_init_metadata_table().await.unwrap();
        let partition_schema = table.get_partition_schema().await.unwrap();
        let partition_pruner =
            PartitionPruner::new(&[], &partition_schema, table.hudi_configs.as_ref()).unwrap();
        let valid = table
            .valid_instant_timestamps(metadata_table)
            .await
            .unwrap();
        let records = metadata_table
            .fetch_files_partition_records(&partition_pruner, &valid)
            .await
            .unwrap();
        assert!(!records.is_empty());

        let file_slices = table.get_file_slices(&ReadOptions::new()).await.unwrap();
        assert!(!file_slices.is_empty());
    }
}