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.
 */
pub mod builder;
pub mod completion_time;
pub mod instant;
pub mod loader;
pub mod lsm_tree;
pub(crate) mod selector;
pub(crate) mod util;
pub mod view;

use crate::Result;
use crate::config::HudiConfigs;
use crate::error::CoreError;
use crate::file_group::FileGroup;
use crate::file_group::builder::replaced_file_groups_from_replace_commit;
use crate::file_group::reader_v2::reader_context::CompletionGateInputs;
use crate::schema::resolver::{
    resolve_avro_schema_from_commit_metadata, resolve_data_schema_from_commit_metadata,
};
use crate::statistics::estimator::FileStatsEstimator;
use crate::storage::Storage;
use crate::timeline::builder::TimelineBuilder;
use crate::timeline::instant::{Action, State};
use crate::timeline::loader::TimelineLoader;
use crate::timeline::selector::TimelineSelector;
use crate::timeline::view::TimelineView;
use arrow_schema::Schema;
use instant::Instant;

use serde_json::{Map, Value};
use std::collections::{HashMap, HashSet};
use std::fmt::Debug;
use std::sync::Arc;

/// A [Timeline] contains transaction logs of all actions performed on the table at different [Instant]s of time.
#[derive(Clone, Debug)]
#[allow(dead_code)]
pub struct Timeline {
    hudi_configs: Arc<HudiConfigs>,
    pub(crate) storage: Arc<Storage>,
    active_loader: TimelineLoader,
    archived_loader: Option<TimelineLoader>,
    pub completed_commits: Vec<Instant>,
    /// Request timestamp of the earliest instant still in the active timeline,
    /// in any state — the archival boundary.
    ///
    /// A data file whose instant sorts *below* this was written by a commit that
    /// has since been archived, and archived instants are committed by
    /// definition: archival only ever moves completed ones, and never moves past
    /// the oldest pending instant. So it is the second half of
    /// [`CompletionTimeView::is_committed`], mirroring Java's
    /// `containsInstant(ts) || isBeforeTimelineStarts(ts)`
    /// (`BaseHoodieTimeline.java:494`). `None` means the active timeline is
    /// empty, so nothing can be treated as archived.
    pub(crate) earliest_active_instant: Option<String>,
    /// Request timestamps of the active instants that have NOT completed —
    /// requested or inflight.
    ///
    /// Read from the same listing that yields [`Self::completed_commits`] and
    /// the archival boundary, so retaining it costs nothing. The log-block scan
    /// needs it: a log file admitted at listing time on its own committed
    /// instant can still carry blocks appended by a later instant that never
    /// completed, and only the instant's state can tell.
    pub(crate) pending_instants: HashSet<String>,
}

pub const EARLIEST_START_TIMESTAMP: &str = "19700101000000000";

/// The actions a read loads from the timeline.
///
/// Narrowing this is no longer only a listing decision: [`Timeline::completion_gate_inputs`]
/// derives the log scan's committed set from these, so an action that completes outside this
/// list and wrote data or delete log blocks would have its rows dropped from a merge-on-read
/// read — silently, since the gate excludes rather than errors. The three here cover every
/// action that writes such blocks (compaction completes as `commit`, clustering as
/// `replacecommit`, log compaction as `deltacommit`), which is also the set Java's gate is
/// built from via `getCommitsTimeline()`.
pub const DEFAULT_LOADING_ACTIONS: &[Action] =
    &[Action::Commit, Action::DeltaCommit, Action::ReplaceCommit];

impl Timeline {
    pub(crate) fn new(
        hudi_configs: Arc<HudiConfigs>,
        storage: Arc<Storage>,
        active_loader: TimelineLoader,
        archived_loader: Option<TimelineLoader>,
    ) -> Self {
        Self {
            hudi_configs,
            storage,
            active_loader,
            archived_loader,
            completed_commits: Vec::new(),
            earliest_active_instant: None,
            pending_instants: HashSet::new(),
        }
    }

    pub(crate) async fn new_from_storage(
        hudi_configs: Arc<HudiConfigs>,
        storage_options: Arc<HashMap<String, String>>,
    ) -> Result<Self> {
        let storage = Storage::new(storage_options.clone(), hudi_configs.clone())?;
        let mut timeline = TimelineBuilder::new(hudi_configs, storage).build().await?;
        // Every state, not just completed: one listing then yields both the
        // completed commits and the archival boundary. See
        // `Timeline::earliest_active_instant` for why the boundary needs the
        // pending ones.
        let selector = TimelineSelector::actions_in_range(
            DEFAULT_LOADING_ACTIONS,
            &[State::Requested, State::Inflight, State::Completed],
            timeline.hudi_configs.clone(),
            None,
            None,
        )?;
        let all_active = timeline.load_instants(&selector, false).await?;
        timeline.earliest_active_instant = all_active
            .iter()
            .map(|instant| instant.timestamp.clone())
            .min();
        let (completed, pending): (Vec<Instant>, Vec<Instant>) = all_active
            .into_iter()
            .partition(|instant| instant.state == State::Completed);
        // An instant is listed once per state file it has, so a completed one
        // also appears as requested and inflight. Pending means it reached NO
        // completed state — subtracting is what makes that true, and without the
        // subtraction every committed instant reads as inflight and the gate
        // rejects the whole timeline.
        let completed_times: HashSet<String> = completed
            .iter()
            .map(|instant| instant.timestamp.clone())
            .collect();
        timeline.pending_instants = pending
            .into_iter()
            .map(|instant| instant.timestamp)
            .filter(|timestamp| !completed_times.contains(timestamp))
            .collect();
        timeline.completed_commits = completed;
        Ok(timeline)
    }

    /// The inputs the log-block scan needs to tell a committed instant from one
    /// that never finished.
    ///
    /// A log file is admitted to a slice on its own instant, but blocks inside
    /// it carry their own — including a writer that was still inflight when a
    /// later writer committed. Without these sets that block merges as if it
    /// were committed, because it sorts below the latest instant and so passes
    /// every other gate.
    pub(crate) fn completion_gate_inputs(&self) -> CompletionGateInputs {
        CompletionGateInputs {
            completed_instants: self
                .completed_commits
                .iter()
                .map(|instant| instant.timestamp.clone())
                .collect(),
            inflight_instants: self.pending_instants.clone(),
            archived_boundary: self.earliest_active_instant.clone(),
        }
    }

    /// Instant times with a requested or inflight file and no completed one,
    /// across every action rather than only the three in
    /// [`DEFAULT_LOADING_ACTIONS`].
    ///
    /// Separate from [`Self::pending_instants`], which this deliberately does
    /// not widen: that field also feeds [`Self::completion_gate_inputs`], where
    /// a wider inflight set would start rejecting log blocks the gate admits
    /// today. The metadata table's valid-instant set is the one caller that
    /// needs Java's whole-timeline view.
    pub(crate) async fn all_pending_instant_times(&self) -> Result<HashSet<String>> {
        self.active_loader.list_pending_instant_times().await
    }

    /// Load instants from the timeline based on the selector criteria.
    ///
    /// # Archived Timeline Loading
    ///
    /// Archived instants are loaded only when BOTH conditions are met:
    /// 1. The selector has a time filter (start or end timestamp)
    /// 2. `TimelineArchivedReadEnabled` config is set to `true`
    ///
    /// This double-gate design ensures:
    /// - Queries without time filters only read active timeline (optimization)
    /// - Historical time-range queries can include archived data when explicitly enabled
    ///
    /// # Arguments
    ///
    /// * `selector` - The criteria for selecting instants (actions, states, time range)
    /// * `desc` - If true, return instants in descending order by timestamp
    pub async fn load_instants(
        &self,
        selector: &TimelineSelector,
        desc: bool,
    ) -> Result<Vec<Instant>> {
        // If a time filter is present and we have an archived loader, include archived as well.
        if selector.has_time_filter() {
            let mut instants = self.active_loader.load_instants(selector, desc).await?;
            if let Some(archived_loader) = &self.archived_loader {
                let mut archived = archived_loader
                    .load_archived_instants(selector, desc)
                    .await?;
                if !archived.is_empty() {
                    // Each side is sorted, but archived instants are OLDER than
                    // active ones, so appending leaves the whole vector unsorted.
                    // `TimelineSelector::select` binary-searches this with
                    // `partition_point`, which silently returns nonsense on
                    // unsorted input — so re-sort rather than merely concatenate.
                    instants.append(&mut archived);
                    instants.sort_unstable();
                    if desc {
                        instants.reverse();
                    }
                }
            }
            Ok(instants)
        } else {
            self.active_loader.load_instants(selector, desc).await
        }
    }

    async fn load_instants_inner(
        &self,
        selector: &TimelineSelector,
        desc: bool,
    ) -> Result<Vec<Instant>> {
        // For now, just load active. Archived support will be added internally later
        // based on selector ranges.
        self.active_loader.load_instants(selector, desc).await
    }

    /// Get the completed commit [Instant]s in the timeline.
    ///
    /// * For Copy-on-write tables, this includes commit instants.
    /// * For Merge-on-read tables, this includes compaction commit instants.
    ///
    /// # Arguments
    ///
    /// * `desc` - If true, the [Instant]s are sorted in descending order.
    pub async fn get_completed_commits(&self, desc: bool) -> Result<Vec<Instant>> {
        let selector =
            TimelineSelector::completed_commits_in_range(self.hudi_configs.clone(), None, None)?;
        self.load_instants_inner(&selector, desc).await
    }

    /// Get the completed deltacommit [Instant]s in the timeline.
    ///
    /// Only applicable for Merge-on-read tables. Empty vector will be returned for Copy-on-write tables.
    ///
    /// # Arguments
    ///
    /// * `desc` - If true, the [Instant]s are sorted in descending order.
    pub async fn get_completed_deltacommits(&self, desc: bool) -> Result<Vec<Instant>> {
        let selector = TimelineSelector::completed_deltacommits_in_range(
            self.hudi_configs.clone(),
            None,
            None,
        )?;
        self.load_instants_inner(&selector, desc).await
    }

    /// Get the completed replacecommit [Instant]s in the timeline.
    ///
    /// # Arguments
    ///
    /// * `desc` - If true, the [Instant]s are sorted in descending order.
    pub async fn get_completed_replacecommits(&self, desc: bool) -> Result<Vec<Instant>> {
        let selector = TimelineSelector::completed_replacecommits_in_range(
            self.hudi_configs.clone(),
            None,
            None,
        )?;
        self.load_instants_inner(&selector, desc).await
    }

    /// Get the completed clustering commit [Instant]s in the timeline.
    ///
    /// # Arguments
    ///
    /// * `desc` - If true, the [Instant]s are sorted in descending order.
    pub async fn get_completed_clustering_commits(&self, desc: bool) -> Result<Vec<Instant>> {
        let selector = TimelineSelector::completed_replacecommits_in_range(
            self.hudi_configs.clone(),
            None,
            None,
        )?;
        let instants = self.load_instants_inner(&selector, desc).await?;
        let mut clustering_instants = Vec::new();
        for instant in instants {
            let metadata = self.get_instant_metadata(&instant).await?;
            let op_type = metadata
                .get("operationType")
                .and_then(|v| v.as_str())
                .ok_or_else(|| {
                    CoreError::CommitMetadata("Failed to get operation type".to_string())
                })?;
            if op_type == "cluster" {
                clustering_instants.push(instant);
            }
        }
        Ok(clustering_instants)
    }

    pub(crate) async fn get_instant_metadata(
        &self,
        instant: &Instant,
    ) -> Result<Map<String, Value>> {
        self.active_loader.load_instant_metadata(instant).await
    }

    /// The raw bytes of one instant file, for readers that decode a record
    /// other than commit metadata — a rollback, a restore, or a rollback plan.
    ///
    pub(crate) async fn load_instant_bytes(&self, instant: &Instant) -> Result<Vec<u8>> {
        self.active_loader.load_instant_bytes(instant).await
    }

    /// Get the instant metadata in JSON format.
    pub async fn get_instant_metadata_in_json(&self, instant: &Instant) -> Result<String> {
        self.active_loader
            .load_instant_metadata_as_json(instant)
            .await
    }

    pub(crate) async fn get_latest_commit_metadata(&self) -> Result<Map<String, Value>> {
        match self.completed_commits.iter().next_back() {
            Some(instant) => self.get_instant_metadata(instant).await,
            None => Err(CoreError::TimelineNoCommit),
        }
    }

    pub(crate) fn get_latest_commit_timestamp_as_option(&self) -> Option<&str> {
        self.completed_commits
            .iter()
            .next_back()
            .map(|instant| instant.timestamp.as_str())
    }

    /// The greatest completion timestamp across completed commits — "everything
    /// committed so far", expressed the way an incremental window is bounded.
    ///
    /// Not the completion time of the latest-*requested* commit: completion order
    /// need not follow requested order, so the maximum has to be taken over the
    /// completion timestamps themselves. Falls back to the latest requested time
    /// on timeline layout v1, which records no completion times.
    pub(crate) fn get_latest_completion_timestamp_as_option(&self) -> Option<&str> {
        self.completed_commits
            .iter()
            .filter_map(|instant| instant.completion_timestamp.as_deref())
            .max()
            .or_else(|| self.get_latest_commit_timestamp_as_option())
    }

    /// Get the latest commit timestamp from the [Timeline].
    ///
    /// Only completed commits are considered.
    pub fn get_latest_commit_timestamp(&self) -> Result<String> {
        self.get_latest_commit_timestamp_as_option()
            .map_or_else(|| Err(CoreError::TimelineNoCommit), |t| Ok(t.to_string()))
    }

    /// Create a [TimelineView] as of the given timestamp.
    pub async fn create_view_as_of(&self, timestamp: &str) -> Result<TimelineView> {
        let excludes = self.get_replaced_file_groups_as_of(timestamp).await?;
        Ok(TimelineView::new_with_archival_boundary(
            timestamp.to_string(),
            None,
            &self.completed_commits,
            excludes,
            &self.hudi_configs,
            self.earliest_active_instant.clone(),
        ))
    }

    /// Get the latest [apache_avro::schema::Schema] as [String] from the [Timeline].
    ///
    /// ### Note
    /// This API behaves differently from [crate::table::Table::get_avro_schema],
    /// which additionally looks for [HudiTableConfig::CreateSchema] in the table config.
    pub async fn get_latest_avro_schema(&self) -> Result<String> {
        let commit_metadata = self.get_latest_commit_metadata().await?;
        resolve_avro_schema_from_commit_metadata(&commit_metadata)
    }

    /// Get the latest data [arrow_schema::Schema] from the [Timeline], without Hudi meta fields.
    ///
    /// ### Note
    /// This API behaves differently from [crate::table::Table::get_schema],
    /// which additionally looks for [HudiTableConfig::CreateSchema] in the table config.
    pub async fn get_latest_schema(&self) -> Result<Schema> {
        let commit_metadata = self.get_latest_commit_metadata().await?;
        resolve_data_schema_from_commit_metadata(&commit_metadata, self.storage.clone()).await
    }

    /// Get all completed instants (commit/deltacommit/replacecommit) whose
    /// request timestamp is ≤ `timestamp`, sorted ascending by timestamp.
    pub(crate) fn get_completed_instants_at_or_before(
        &self,
        timestamp: &str,
    ) -> Result<Vec<Instant>> {
        let selector = TimelineSelector::completed_actions_in_range(
            DEFAULT_LOADING_ACTIONS,
            self.hudi_configs.clone(),
            None,
            Some(timestamp),
        )?;
        selector.select(self)
    }

    pub(crate) async fn get_replaced_file_groups_as_of(
        &self,
        timestamp: &str,
    ) -> Result<HashSet<FileGroup>> {
        let mut file_groups: HashSet<FileGroup> = HashSet::new();
        let selector = TimelineSelector::completed_replacecommits_in_range(
            self.hudi_configs.clone(),
            None,
            Some(timestamp),
        )?;
        for instant in selector.select(self)? {
            let commit_metadata = self.get_instant_metadata(&instant).await?;
            file_groups.extend(replaced_file_groups_from_replace_commit(&commit_metadata)?);
        }

        // TODO: return file group and instants, and handle multi-writer fg id conflicts

        Ok(file_groups)
    }

    /// Get file groups from commit metadata for commits in a time range.
    ///
    /// This is used for incremental queries where we only want file groups
    /// that were modified in the time range (start, end].
    ///
    /// # Arguments
    /// * `start_timestamp` - Start of the time range (exclusive), None means no lower bound
    /// * `end_timestamp` - End of the time range (inclusive), None means no upper bound
    ///
    /// # Returns
    /// File groups that were modified in the time range, excluding replaced file groups.
    /// The completed commits an incremental window `(start, end]` admits, in
    /// requested-time order.
    ///
    /// Which timestamp the window bounds depends on the timeline layout — see
    /// [`TimelineSelector::select`].
    pub(crate) fn get_completed_commits_in_range(
        &self,
        start_timestamp: Option<&str>,
        end_timestamp: Option<&str>,
    ) -> Result<Vec<Instant>> {
        let selector = TimelineSelector::completed_actions_in_completion_time_range(
            DEFAULT_LOADING_ACTIONS,
            self.hudi_configs.clone(),
            start_timestamp,
            end_timestamp,
        )?;
        selector.select(self)
    }

    pub(crate) async fn get_file_groups_between(
        &self,
        start_timestamp: Option<&str>,
        end_timestamp: Option<&str>,
        estimator: Option<&FileStatsEstimator>,
    ) -> Result<HashSet<FileGroup>> {
        use crate::file_group::builder::{
            FileGroupMerger, file_groups_from_commit_metadata_with_estimator,
            replaced_file_groups_from_replace_commit,
        };

        // Requested-time bounds by this point: an incremental window is
        // translated once, in `Table::resolve_incremental_window`, which resolves
        // the user's (possibly completion-time) window into the instant times it
        // admits and re-expresses the bounds over those commits' requested times.
        // Interpreting them as completion times again here would translate twice
        // and select nothing.
        let selector = TimelineSelector::completed_actions_in_range(
            DEFAULT_LOADING_ACTIONS,
            self.hudi_configs.clone(),
            start_timestamp,
            end_timestamp,
        )?;
        let commits = selector.select(self)?;
        if commits.is_empty() {
            return Ok(HashSet::new());
        }

        // Build completion time view from selected commits only.
        let completion_time_view = TimelineView::new(
            commits.last().unwrap().timestamp.clone(),
            Some(commits.first().unwrap().timestamp.clone()),
            &commits,
            HashSet::new(),
            &self.hudi_configs,
        );

        let mut file_groups: HashSet<FileGroup> = HashSet::new();
        let mut replaced_file_groups: HashSet<FileGroup> = HashSet::new();

        for commit in commits {
            let commit_metadata = self.get_instant_metadata(&commit).await?;
            let contribution = file_groups_from_commit_metadata_with_estimator(
                &commit_metadata,
                &completion_time_view,
                estimator,
            )?;
            file_groups.merge(contribution.file_groups)?;

            // A delta commit that only appends names no base file, so it
            // contributes no slice — but the file group it touched still
            // changed in the range, and the caller reads it as of the range's
            // end. Carry the identity alone.
            for unattached in contribution.unattached_log_files {
                let touched = FileGroup::new(unattached.file_id, unattached.partition);
                if !file_groups.contains(&touched) {
                    file_groups.insert(touched);
                }
            }

            if commit.is_replacecommit() {
                replaced_file_groups
                    .extend(replaced_file_groups_from_replace_commit(&commit_metadata)?);
            }
        }

        Ok(file_groups
            .difference(&replaced_file_groups)
            .cloned()
            .collect())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap;
    use std::fs::canonicalize;
    use std::path::Path;
    use std::str::FromStr;
    use std::sync::Arc;

    use url::Url;

    use hudi_test::{SampleTable, assert_arrow_field_names_eq, assert_avro_field_names_eq};

    use crate::config::table::HudiTableConfig;
    use crate::metadata::meta_field::MetaField;
    use crate::timeline::instant::{Action, State};
    #[tokio::test]
    async fn test_timeline_v8_nonpartitioned() {
        let base_url = SampleTable::V8Nonpartitioned.url_to_cow();
        let timeline = create_test_timeline(base_url).await;
        assert_eq!(timeline.completed_commits.len(), 2);
        assert!(timeline.active_loader.is_layout_two_active());
        // Archived loader should be None when TimelineArchivedReadEnabled is false (default)
        assert!(timeline.archived_loader.is_none());
    }

    #[tokio::test]
    async fn test_timeline_v8_with_archived_enabled() {
        use crate::config::internal::HudiInternalConfig::TimelineArchivedReadEnabled;

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

        // Build initial configs with base path and archived read enabled
        let mut options_map = HashMap::new();
        options_map.insert(
            HudiTableConfig::BasePath.as_ref().to_string(),
            base_url.to_string(),
        );
        options_map.insert(
            TimelineArchivedReadEnabled.as_ref().to_string(),
            "true".to_string(),
        );

        let storage = Storage::new(
            Arc::new(HashMap::new()),
            Arc::new(HudiConfigs::new(options_map.clone())),
        )
        .unwrap();

        let table_properties = crate::config::util::parse_data_for_options(
            &storage
                .get_file_data(".hoodie/hoodie.properties")
                .await
                .unwrap(),
            "=",
        )
        .unwrap();
        options_map.extend(table_properties);
        let hudi_configs = Arc::new(HudiConfigs::new(options_map));

        let timeline = TimelineBuilder::new(hudi_configs, storage)
            .build()
            .await
            .unwrap();

        // When TimelineArchivedReadEnabled is true, archived loader should be created
        assert!(timeline.active_loader.is_layout_two_active());
        assert!(
            timeline
                .archived_loader
                .as_ref()
                .map(|l| l.is_layout_two_archived())
                .unwrap_or(false)
        );
    }

    async fn create_test_timeline(base_url: Url) -> Timeline {
        let storage = Storage::new(
            Arc::new(HashMap::new()),
            Arc::new(HudiConfigs::new([(
                HudiTableConfig::BasePath,
                base_url.to_string(),
            )])),
        )
        .unwrap();

        let hudi_configs = HudiConfigs::new([(HudiTableConfig::BasePath, base_url.to_string())]);
        let table_properties = crate::config::util::parse_data_for_options(
            &storage
                .get_file_data(".hoodie/hoodie.properties")
                .await
                .unwrap(),
            "=",
        )
        .unwrap();
        let mut hudi_configs_map = hudi_configs.as_options();
        hudi_configs_map.extend(table_properties);
        let hudi_configs = Arc::new(HudiConfigs::new(hudi_configs_map));

        let mut timeline = TimelineBuilder::new(hudi_configs, storage)
            .build()
            .await
            .unwrap();

        let selector = TimelineSelector::completed_actions_in_range(
            DEFAULT_LOADING_ACTIONS,
            timeline.hudi_configs.clone(),
            None,
            None,
        )
        .unwrap();
        timeline.completed_commits = timeline.load_instants(&selector, false).await.unwrap();
        timeline
    }

    #[tokio::test]
    async fn timeline_read_latest_schema() {
        let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
        let timeline = create_test_timeline(base_url).await;
        let table_schema = timeline.get_latest_schema().await.unwrap();
        // get_latest_schema returns data schema without meta fields
        assert_eq!(table_schema.fields.len(), 16)
    }

    #[tokio::test]
    async fn timeline_read_latest_schema_from_empty_table() {
        let base_url = SampleTable::V6Empty.url_to_cow();
        let timeline = create_test_timeline(base_url).await;
        let table_schema = timeline.get_latest_schema().await;
        assert!(table_schema.is_err());
        assert!(matches!(
            table_schema.unwrap_err(),
            CoreError::TimelineNoCommit
        ))
    }

    #[tokio::test]
    async fn init_commits_timeline() {
        let base_url = Url::from_file_path(
            canonicalize(Path::new("tests/data/timeline/commits_stub")).unwrap(),
        )
        .unwrap();
        let timeline = create_test_timeline(base_url).await;
        assert_eq!(
            timeline.completed_commits,
            vec![
                Instant::from_str("20240402123035233.commit").unwrap(),
                Instant::from_str("20240402144910683.commit").unwrap(),
            ]
        )
    }

    #[tokio::test]
    async fn get_commit_metadata_returns_error() {
        let base_url = Url::from_file_path(
            canonicalize(Path::new(
                "tests/data/timeline/commits_with_invalid_content",
            ))
            .unwrap(),
        )
        .unwrap();
        let timeline = create_test_timeline(base_url).await;
        let instant = Instant::from_str("20240402123035233.commit").unwrap();

        // Test error when reading empty commit metadata file
        let result = timeline.get_instant_metadata(&instant).await;
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(matches!(err, CoreError::Timeline(_)));
        // Error message changed to be more specific about JSON parsing
        assert!(
            err.to_string()
                .contains("Failed to parse JSON commit metadata")
                || err.to_string().contains("EOF while parsing")
        );

        let instant = Instant::from_str("20240402144910683.commit").unwrap();

        // Test error when reading a commit metadata file with invalid JSON
        let result = timeline.get_instant_metadata(&instant).await;
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(matches!(err, CoreError::Timeline(_)));
        // Error message changed to be more specific about JSON parsing
        assert!(
            err.to_string()
                .contains("Failed to parse JSON commit metadata")
                || err.to_string().contains("expected value")
        );
    }

    #[tokio::test]
    async fn timeline_get_schema_returns_error_for_no_schema_and_write_stats() {
        let base_url = Url::from_file_path(
            canonicalize(Path::new(
                "tests/data/timeline/commits_with_no_schema_and_write_stats",
            ))
            .unwrap(),
        )
        .unwrap();
        let timeline = create_test_timeline(base_url).await;

        // Check Arrow schema
        let arrow_schema = timeline.get_latest_schema().await;
        assert!(arrow_schema.is_err());
        assert!(
            matches!(arrow_schema.unwrap_err(), CoreError::CommitMetadata(_)),
            "Getting Arrow schema includes base file lookup, therefore expect CommitMetadata error when write stats are missing"
        );

        // Check Avro schema
        let avro_schema = timeline.get_latest_avro_schema().await;
        assert!(avro_schema.is_err());
        assert!(
            matches!(avro_schema.unwrap_err(), CoreError::SchemaNotFound(_)),
            "Getting Avro schema does not include base file lookup, therefore expect SchemaNotFound error when `extraMetadata.schema` is missing"
        );
    }

    #[tokio::test]
    async fn timeline_get_schema_from_commit_metadata() {
        let base_url = Url::from_file_path(
            canonicalize(Path::new(
                "tests/data/timeline/commits_with_valid_schema_in_commit_metadata",
            ))
            .unwrap(),
        )
        .unwrap();
        let timeline = create_test_timeline(base_url).await;

        // Check Arrow schema — get_latest_schema returns data schema without meta fields
        let arrow_schema = timeline.get_latest_schema().await;
        assert!(arrow_schema.is_ok());
        let arrow_schema = arrow_schema.unwrap();
        assert_arrow_field_names_eq!(
            arrow_schema,
            vec!["ts", "uuid", "rider", "driver", "fare", "city"]
        );

        // Check Avro schema
        let avro_schema = timeline.get_latest_avro_schema().await;
        assert!(avro_schema.is_ok());
        let avro_schema = avro_schema.unwrap();
        assert_avro_field_names_eq!(
            &avro_schema,
            ["ts", "uuid", "rider", "driver", "fare", "city"]
        );
    }

    #[tokio::test]
    async fn timeline_get_schema_from_empty_commit_metadata() {
        let base_url = Url::from_file_path(
            canonicalize(Path::new(
                "tests/data/timeline/commits_with_empty_commit_metadata",
            ))
            .unwrap(),
        )
        .unwrap();
        let timeline = create_test_timeline(base_url).await;

        // Check Arrow schema
        let result = timeline.get_latest_schema().await;
        assert!(result.is_err());
        assert!(matches!(result.unwrap_err(), CoreError::CommitMetadata(_)));

        // Check Avro schema
        let result = timeline.get_latest_avro_schema().await;
        assert!(result.is_err());
        assert!(matches!(result.unwrap_err(), CoreError::CommitMetadata(_)));
    }

    #[tokio::test]
    async fn timeline_get_schema_from_base_file() {
        let timeline_base_urls = [
            "tests/data/timeline/commits_load_schema_from_base_file_cow",
            "tests/data/timeline/commits_load_schema_from_base_file_mor",
        ];
        for base_url in timeline_base_urls {
            let base_url = Url::from_file_path(canonicalize(Path::new(base_url)).unwrap()).unwrap();
            let timeline = create_test_timeline(base_url).await;

            let arrow_schema = timeline.get_latest_schema().await;
            assert!(arrow_schema.is_ok());
            let arrow_schema = arrow_schema.unwrap();
            assert_arrow_field_names_eq!(
                arrow_schema,
                [
                    MetaField::field_names(),
                    vec!["ts", "uuid", "rider", "driver", "fare", "city"]
                ]
                .concat()
            );
        }
    }

    #[tokio::test]
    async fn test_get_completed_commits() {
        let base_url = SampleTable::V8Nonpartitioned.url_to_cow();
        let timeline = create_test_timeline(base_url).await;

        let commits = timeline.get_completed_commits(false).await.unwrap();
        assert!(!commits.is_empty());
        // All should be commits in completed state
        for instant in &commits {
            assert_eq!(instant.action, Action::Commit);
            assert_eq!(instant.state, State::Completed);
        }
    }

    #[tokio::test]
    async fn test_get_completed_deltacommits() {
        let base_url = SampleTable::V8Nonpartitioned.url_to_cow();
        let timeline = create_test_timeline(base_url).await;

        let deltacommits = timeline.get_completed_deltacommits(false).await.unwrap();
        // All should be deltacommits (or empty if none exist)
        for instant in &deltacommits {
            assert_eq!(instant.action, Action::DeltaCommit);
            assert_eq!(instant.state, State::Completed);
        }
    }

    #[tokio::test]
    async fn test_get_completed_replacecommits() {
        let base_url = SampleTable::V8Nonpartitioned.url_to_cow();
        let timeline = create_test_timeline(base_url).await;

        let replacecommits = timeline.get_completed_replacecommits(false).await.unwrap();
        // All should be replacecommits (or empty if none exist)
        for instant in &replacecommits {
            assert!(instant.action.is_replacecommit());
            assert_eq!(instant.state, State::Completed);
        }
    }

    #[tokio::test]
    async fn test_get_completed_replacecommits_v9_overwrite() {
        let base_url = SampleTable::V9TxnsSimpleOverwrite.url_to_cow();
        let timeline = create_test_timeline(base_url).await;

        let commits = timeline.get_completed_commits(false).await.unwrap();
        assert_eq!(commits.len(), 2);
        for instant in &commits {
            assert_eq!(instant.action, Action::Commit);
            assert_eq!(instant.state, State::Completed);
        }

        let replacecommits = timeline.get_completed_replacecommits(false).await.unwrap();
        assert_eq!(replacecommits.len(), 1);
        for instant in &replacecommits {
            assert_eq!(instant.action, Action::ReplaceCommit);
            assert_eq!(instant.state, State::Completed);
        }
    }

    #[tokio::test]
    async fn test_get_completed_deltacommits_v9_nonpartitioned_rollback() {
        let base_url = SampleTable::V9NonpartitionedRollback.url_to_mor_avro();
        let timeline = create_test_timeline(base_url).await;

        let commits = timeline.get_completed_commits(false).await.unwrap();
        assert!(
            commits.is_empty(),
            "Rollback MOR fixture should not contain completed commit instants"
        );

        let deltacommits = timeline.get_completed_deltacommits(false).await.unwrap();
        assert_eq!(deltacommits.len(), 2);
        for instant in &deltacommits {
            assert_eq!(instant.action, Action::DeltaCommit);
            assert_eq!(instant.state, State::Completed);
        }
    }

    #[tokio::test]
    async fn test_get_commits_descending_order() {
        let base_url = SampleTable::V8Nonpartitioned.url_to_cow();
        let timeline = create_test_timeline(base_url).await;

        let commits_asc = timeline.get_completed_commits(false).await.unwrap();
        let commits_desc = timeline.get_completed_commits(true).await.unwrap();

        assert_eq!(commits_asc.len(), commits_desc.len());
        if !commits_asc.is_empty() {
            // Verify descending order is reverse of ascending
            assert_eq!(commits_asc.first(), commits_desc.last());
            assert_eq!(commits_asc.last(), commits_desc.first());
        }
    }

    #[tokio::test]
    async fn test_get_instant_metadata_in_json() {
        let base_url = SampleTable::V8Nonpartitioned.url_to_cow();
        let timeline = create_test_timeline(base_url).await;

        let commits = timeline.get_completed_commits(false).await.unwrap();
        if let Some(instant) = commits.first() {
            let json = timeline
                .get_instant_metadata_in_json(instant)
                .await
                .unwrap();
            // Should be valid JSON
            assert!(serde_json::from_str::<serde_json::Value>(&json).is_ok());
        }
    }

    #[tokio::test]
    async fn test_get_latest_commit_timestamp() {
        let base_url = SampleTable::V8Nonpartitioned.url_to_cow();
        let timeline = create_test_timeline(base_url).await;

        let timestamp = timeline.get_latest_commit_timestamp().unwrap();
        assert!(!timestamp.is_empty());
        // Should be in timeline timestamp format
        assert!(timestamp.len() >= 14);
    }

    #[tokio::test]
    async fn test_get_latest_commit_timestamp_as_option() {
        let base_url = SampleTable::V8Nonpartitioned.url_to_cow();
        let timeline = create_test_timeline(base_url).await;

        let timestamp = timeline.get_latest_commit_timestamp_as_option();
        assert!(timestamp.is_some());
        assert!(!timestamp.unwrap().is_empty());
    }

    /// Regression test: a completed instant is not also reported as pending.
    ///
    /// The active timeline lists an instant once per state file it has, so a
    /// completed one appears as requested and inflight too. Reading pending
    /// straight off the non-completed rows therefore marks every committed
    /// instant inflight, and the log-scan gate — which admits only
    /// `committed && !inflight` — then rejects the whole timeline and returns
    /// base-file-only data. That is silent: no error, just missing log deltas.
    #[tokio::test]
    async fn test_completion_gate_inputs_do_not_report_completed_instants_as_pending() {
        let base_url = SampleTable::V6Nonpartitioned.url_to_mor_parquet();
        let hudi_configs = Arc::new(HudiConfigs::new([(
            HudiTableConfig::BasePath,
            base_url.to_string(),
        )]));
        let timeline = Timeline::new_from_storage(hudi_configs, Arc::new(HashMap::new()))
            .await
            .unwrap();

        let inputs = timeline.completion_gate_inputs();
        assert!(
            !inputs.completed_instants.is_empty(),
            "the fixture has completed commits"
        );
        let both: Vec<&String> = inputs
            .inflight_instants
            .iter()
            .filter(|t| inputs.completed_instants.contains(*t))
            .collect();
        assert!(
            both.is_empty(),
            "an instant cannot be both completed and pending, got {both:?}"
        );
    }
}