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.
 */
use crate::Result;
use crate::config::table::TimelineTimezoneValue;
use crate::error::CoreError;
use crate::metadata::HUDI_METADATA_DIR;
use crate::storage::error::StorageError;
use chrono::{DateTime, Local, NaiveDateTime, TimeZone, Timelike, Utc};
use std::cmp::Ordering;
use std::path::PathBuf;
use std::str::FromStr;

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum Action {
    Commit,
    DeltaCommit,
    ReplaceCommit,
    /// A rollback of one earlier instant.
    ///
    /// Deliberately absent from [`DEFAULT_LOADING_ACTIONS`]: a rollback is not a
    /// commit, and adding it there would put it in `completed_commits` and
    /// change every existing read. It is loaded only by a caller that names it,
    /// which today is the metadata table's valid-instant set -- Hudi counts the
    /// commits a rollback rolled back as valid, because their log blocks were
    /// written and then re-applied.
    ///
    /// [`DEFAULT_LOADING_ACTIONS`]: crate::timeline::DEFAULT_LOADING_ACTIONS
    Rollback,
    /// A restore, which is made up of several rollbacks.
    Restore,
}

impl FromStr for Action {
    type Err = CoreError;

    fn from_str(s: &str) -> Result<Self> {
        match s {
            "commit" => Ok(Action::Commit),
            "deltacommit" => Ok(Action::DeltaCommit),
            "replacecommit" => Ok(Action::ReplaceCommit),
            "rollback" => Ok(Action::Rollback),
            "restore" => Ok(Action::Restore),
            _ => Err(CoreError::Timeline(format!("Invalid action: {s}"))),
        }
    }
}

impl AsRef<str> for Action {
    fn as_ref(&self) -> &str {
        match self {
            Action::Commit => "commit",
            Action::DeltaCommit => "deltacommit",
            Action::ReplaceCommit => "replacecommit",
            Action::Rollback => "rollback",
            Action::Restore => "restore",
        }
    }
}

impl Action {
    pub fn is_replacecommit(&self) -> bool {
        self == &Action::ReplaceCommit
    }
}

/// The [State] of an [Instant] represents the status of the action performed on the table.
#[allow(dead_code)]
#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd)]
pub enum State {
    Requested,
    Inflight,
    Completed,
}

impl FromStr for State {
    type Err = CoreError;

    fn from_str(suffix: &str) -> Result<Self> {
        match suffix {
            "requested" => Ok(State::Requested),
            "inflight" => Ok(State::Inflight),
            "" => Ok(State::Completed),
            _ => Err(CoreError::Timeline(format!(
                "Invalid state suffix: {suffix}"
            ))),
        }
    }
}

impl AsRef<str> for State {
    fn as_ref(&self) -> &str {
        match self {
            State::Requested => "requested",
            State::Inflight => "inflight",
            State::Completed => "",
        }
    }
}

/// An [Instant] represents a point in time when an action was performed on the table.
///
/// For table version 8+, completed instants have a different filename format:
/// `{requestedTimestamp}_{completionTimestamp}.{action}` instead of `{timestamp}.{action}`.
/// The `timestamp` field stores the requested timestamp, and `completion_timestamp` stores
/// the completion timestamp for v8+ completed instants.
#[allow(dead_code)]
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Instant {
    /// The timestamp when the action was requested (used for ordering and identification).
    /// TODO rename to requested_timestamp for clarity in v8+?
    pub timestamp: String,
    /// The timestamp when the action completed (only present for v8+ completed instants).
    pub completion_timestamp: Option<String>,
    pub action: Action,
    pub state: State,
    pub epoch_millis: i64,
}

impl PartialOrd for Instant {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for Instant {
    fn cmp(&self, other: &Self) -> Ordering {
        self.epoch_millis
            .cmp(&other.epoch_millis)
            .then_with(|| self.state.cmp(&other.state))
    }
}

impl FromStr for Instant {
    type Err = CoreError;

    /// Parse a timeline file name into an [Instant]. Timezone is assumed to be UTC.
    fn from_str(file_name: &str) -> Result<Self, Self::Err> {
        Self::try_from_file_name_and_timezone(file_name, "UTC")
    }
}

impl Instant {
    pub fn try_from_file_name_and_timezone(file_name: &str, timezone: &str) -> Result<Self> {
        let (timestamp_part, action_suffix) = file_name
            .split_once('.')
            .ok_or_else(|| CoreError::Timeline(format!("Invalid file name: {file_name}")))?;

        let (action, state) = Self::parse_action_and_state(action_suffix)?;

        // Check for v8+ completed instant format: {requestedTimestamp}_{completedTimestamp}.{action}
        // This format is only used for completed instants (state == Completed)
        if let Some((requested_ts, completed_ts)) = timestamp_part.split_once('_') {
            // This is a v8+ completed instant with both requested and completed timestamps
            Self::validate_timestamp(requested_ts)?;
            Self::validate_timestamp(completed_ts)?;
            let dt = Self::parse_datetime(requested_ts, timezone)?;

            if state != State::Completed {
                return Err(CoreError::Timeline(format!(
                    "Underscore timestamp format is only valid for completed instants: {file_name}"
                )));
            }

            Ok(Self {
                timestamp: requested_ts.to_string(),
                completion_timestamp: Some(completed_ts.to_string()),
                state,
                action,
                epoch_millis: dt.timestamp_millis(),
            })
        } else {
            // pre v8 format: {timestamp}.{action}[.{state}]
            Self::validate_timestamp(timestamp_part)?;
            let dt = Self::parse_datetime(timestamp_part, timezone)?;

            Ok(Self {
                timestamp: timestamp_part.to_string(),
                completion_timestamp: None,
                state,
                action,
                epoch_millis: dt.timestamp_millis(),
            })
        }
    }

    /// Instant times Hudi uses to mark a table's beginning, which are valid
    /// instants but not dates.
    ///
    /// `HoodieTimeline` defines `INIT_INSTANT_TS`, `METADATA_BOOTSTRAP_INSTANT_TS`
    /// and `FULL_BOOTSTRAP_INSTANT_TS` as these three strings. Hudi orders
    /// instants by comparing the strings themselves, so it never has to read
    /// them as dates — and its own date parser rejects them (`0000` is not a
    /// year, `00` not a month). `HoodieSqlCommonUtils.validateInstant` therefore
    /// admits them by name before it attempts any parse, which is what makes
    /// `INIT_INSTANT_TS` usable as an incremental read's "from the beginning"
    /// bound.
    ///
    /// This crate compares instants as `DateTime`s rather than as strings, so it
    /// has to map them somewhere; epoch 0/1/2 ms places them before every real
    /// instant and in the same order as the 17-character forms the metadata
    /// table writes for the same purpose.
    const BOOTSTRAP_INSTANT_TIMES: [&'static str; 3] =
        ["00000000000000", "00000000000001", "00000000000002"];

    fn validate_timestamp(timestamp: &str) -> Result<()> {
        if !matches!(timestamp.len(), 14 | 17) {
            return Err(CoreError::Timeline(format!(
                "Timestamp must be in format yyyyMMddHHmmss or yyyyMMddHHmmssSSS, but got: {timestamp}"
            )));
        }
        Ok(())
    }

    /// Parse a timestamp string into a UTC DateTime.
    ///
    /// Supports two formats:
    /// 1. Date format: `yyyyMMddHHmmss` (14 chars) or `yyyyMMddHHmmssSSS` (17 chars)
    /// 2. Epoch milliseconds: 17-digit number representing milliseconds since Unix epoch
    ///    (used by metadata table for timestamps like `00000000000000000`), or one
    ///    of the [`Self::BOOTSTRAP_INSTANT_TIMES`] sentinels.
    ///
    /// The function tries date format first, then falls back to epoch milliseconds
    /// if the date parsing fails (e.g., invalid month/day like `00`).
    pub fn parse_datetime(timestamp: &str, timezone: &str) -> Result<DateTime<Utc>> {
        // First try parsing as yyyyMMddHHmmssSSS date format
        if let Ok(naive_dt) = Self::parse_naive_datetime(timestamp) {
            return Self::convert_to_timezone(naive_dt, timezone);
        }

        // Fallback: treat as epoch milliseconds (zero-padded 17-digit number)
        // This handles metadata table timestamps like 00000000000000000, 00000000000000001, etc.
        //
        // The sentinels are admitted by name rather than by relaxing the length
        // check to 14. Every other 14-character value that fails the date parse
        // is a malformed date, and reading one as epoch millis would silently
        // move it centuries rather than report it.
        if (timestamp.len() == 17 || Self::BOOTSTRAP_INSTANT_TIMES.contains(&timestamp))
            && timestamp.chars().all(|c| c.is_ascii_digit())
        {
            let epoch_ms: i64 = timestamp
                .parse()
                .map_err(|e| CoreError::Timeline(format!("Invalid epoch timestamp: {e}")))?;
            return DateTime::from_timestamp_millis(epoch_ms)
                .ok_or_else(|| CoreError::Timeline("Invalid epoch millis".to_string()));
        }

        Err(CoreError::Timeline(format!(
            "Cannot parse timestamp '{timestamp}': not a valid date format or epoch millis"
        )))
    }

    fn parse_naive_datetime(timestamp: &str) -> Result<NaiveDateTime> {
        // Guard the slice: a timestamp shorter than the date-time prefix would
        // otherwise panic here. Callers reach this with values read from data
        // (a `_hoodie_commit_time` cell), not just with instants parsed from
        // well-formed file names, so a short value has to be an error.
        if timestamp.len() < 14 {
            return Err(CoreError::Timeline(format!(
                "Cannot parse timestamp '{timestamp}': shorter than yyyyMMddHHmmss"
            )));
        }
        let naive_dt = NaiveDateTime::parse_from_str(&timestamp[..14], "%Y%m%d%H%M%S")
            .map_err(|e| CoreError::Timeline(format!("Failed to parse timestamp: {e}")))?;

        if timestamp.len() == 17 {
            let millis: u32 = timestamp[14..]
                .parse()
                .map_err(|e| CoreError::Timeline(format!("Failed to parse milliseconds: {e}")))?;
            naive_dt
                .with_nanosecond(millis * 1_000_000)
                .ok_or_else(|| CoreError::Timeline("Invalid milliseconds".to_string()))
        } else {
            Ok(naive_dt)
        }
    }

    fn convert_to_timezone(naive_dt: NaiveDateTime, timezone: &str) -> Result<DateTime<Utc>> {
        match TimelineTimezoneValue::from_str(timezone)? {
            TimelineTimezoneValue::UTC => Ok(DateTime::from_naive_utc_and_offset(naive_dt, Utc)),
            TimelineTimezoneValue::Local => Ok(Local
                .from_local_datetime(&naive_dt)
                .single()
                .ok_or_else(|| CoreError::Timeline("Invalid local datetime".to_string()))?
                .to_utc()),
        }
    }

    pub fn parse_action_and_state(action_suffix: &str) -> Result<(Action, State)> {
        match action_suffix.split_once('.') {
            Some((action_str, state_str)) => {
                Ok((Action::from_str(action_str)?, State::from_str(state_str)?))
            }
            None => match action_suffix {
                "inflight" => Ok((Action::Commit, State::Inflight)),
                action_str => Ok((Action::from_str(action_str)?, State::Completed)),
            },
        }
    }

    pub fn file_name(&self) -> String {
        match (&self.action, &self.state) {
            (_, State::Completed) => {
                // For v8+ completed instants with completion_timestamp, use the underscore format
                if let Some(completion_ts) = &self.completion_timestamp {
                    format!(
                        "{}_{}.{}",
                        self.timestamp,
                        completion_ts,
                        self.action.as_ref()
                    )
                } else {
                    format!("{}.{}", self.timestamp, self.action.as_ref())
                }
            }
            (Action::Commit, State::Inflight) => {
                format!("{}.{}", self.timestamp, self.state.as_ref())
            }
            _ => format!(
                "{}.{}.{}",
                self.timestamp,
                self.action.as_ref(),
                self.state.as_ref()
            ),
        }
    }

    /// Get the relative path with the default `.hoodie/` base directory.
    /// For v8+ tables with Layout Version 2, use `relative_path_with_base` instead.
    pub fn relative_path(&self) -> Result<String> {
        self.relative_path_with_base(HUDI_METADATA_DIR)
    }

    /// Get the relative path with a specified base directory.
    /// Use `.hoodie/` for pre-v8 tables (Layout Version 1) or `.hoodie/timeline/` for v8+ tables (Layout Version 2).
    pub fn relative_path_with_base(&self, base_dir: &str) -> Result<String> {
        let mut commit_file_path = PathBuf::from(base_dir);
        commit_file_path.push(self.file_name());
        commit_file_path
            .to_str()
            .ok_or(StorageError::InvalidPath(format!(
                "Failed to get file path for {self:?}"
            )))
            .map_err(CoreError::Storage)
            .map(|s| s.to_string())
    }

    pub fn is_replacecommit(&self) -> bool {
        self.action.is_replacecommit()
    }
}

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

    #[test]
    fn test_action_methods() {
        assert_eq!(Action::Commit.as_ref(), "commit");
        assert_eq!(Action::ReplaceCommit.as_ref(), "replacecommit");

        assert!(!Action::Commit.is_replacecommit());
        assert!(Action::ReplaceCommit.is_replacecommit());
    }

    #[test]
    fn test_action_from_str() {
        assert_eq!(Action::from_str("commit").unwrap(), Action::Commit);
        assert_eq!(
            Action::from_str("replacecommit").unwrap(),
            Action::ReplaceCommit
        );
        assert!(Action::from_str("invalid").is_err());
    }

    #[test]
    fn test_state_methods() {
        assert_eq!(State::Requested.as_ref(), "requested");
        assert_eq!(State::Inflight.as_ref(), "inflight");
        assert_eq!(State::Completed.as_ref(), "");
    }

    #[test]
    fn test_state_from_str() {
        assert_eq!(State::from_str("requested").unwrap(), State::Requested);
        assert_eq!(State::from_str("inflight").unwrap(), State::Inflight);
        assert_eq!(State::from_str("").unwrap(), State::Completed);
        assert!(State::from_str("invalid").is_err());
    }

    #[test]
    fn test_state_ordering() {
        assert!(State::Requested < State::Inflight);
        assert!(State::Inflight < State::Completed);
        assert!(State::Requested < State::Completed);

        assert!(State::Completed > State::Inflight);
        assert!(State::Inflight > State::Requested);
        assert!(State::Completed > State::Requested);
    }

    #[test]
    fn test_instant_from_file_name() -> Result<()> {
        // Test completed commit
        let instant = Instant::from_str("20240103153000.commit")?;
        assert_eq!(instant.timestamp, "20240103153000");
        assert_eq!(instant.action, Action::Commit);
        assert_eq!(instant.state, State::Completed);

        // Test inflight replacecommit with milliseconds
        let instant = Instant::from_str("20240103153000123.replacecommit.inflight")?;
        assert_eq!(instant.timestamp, "20240103153000123");
        assert_eq!(instant.action, Action::ReplaceCommit);
        assert_eq!(instant.state, State::Inflight);

        Ok(())
    }

    #[test]
    fn test_invalid_file_names() {
        // Invalid timestamp format
        assert!(Instant::from_str("2024010315.commit").is_err());

        // Invalid action
        assert!(Instant::from_str("20240103153000.invalid").is_err());

        // Invalid state
        assert!(Instant::from_str("20240103153000.commit.invalid").is_err());

        // No dot separator
        assert!(Instant::from_str("20240103153000commit").is_err());
    }

    #[test]
    fn test_file_name_roundtrip() -> Result<()> {
        let test_cases = vec![
            "20240103153000.commit",
            "20240103153000.commit.requested",
            "20240103153000.replacecommit",
            "20240103153000123.inflight",
        ];

        for original_name in test_cases {
            let instant = Instant::from_str(original_name)?;
            assert_eq!(instant.file_name(), original_name);
        }

        Ok(())
    }

    #[test]
    fn test_instant_ordering() -> Result<()> {
        let instant1 = Instant::from_str("20240103153000.commit")?;
        let instant2 = Instant::from_str("20240103153000001.commit")?;
        let instant3 = Instant::from_str("20240103153000999.commit.requested")?;
        let instant4 = Instant::from_str("20240103153000999.inflight")?;
        let instant5 = Instant::from_str("20240103153000999.commit")?;
        let instant6 = Instant::from_str("20240103153001.commit")?;

        assert!(instant1 < instant2);
        assert!(instant2 < instant3);
        assert!(instant3 < instant4);
        assert!(instant4 < instant5);
        assert!(instant5 < instant6);

        Ok(())
    }

    #[test]
    fn test_relative_path() {
        let instant = Instant::from_str("20240103153000.commit").unwrap();
        assert_eq!(
            instant.relative_path().unwrap(),
            ".hoodie/20240103153000.commit"
        );
    }

    #[test]
    #[serial(env_vars)]
    fn test_create_instant_using_local_timezone() {
        // Set a fixed timezone for consistent testing
        let original_tz = std::env::var("TZ").ok();
        unsafe {
            std::env::set_var("TZ", "Etc/GMT+5"); // UTC-5 fixed timezone with no DST
        }

        let file_name = "20240103153000.commit";
        let instant_local = Instant::try_from_file_name_and_timezone(file_name, "local").unwrap();
        let instant_utc = Instant::try_from_file_name_and_timezone(file_name, "utc").unwrap();
        let offset_seconds = (instant_local.epoch_millis - instant_utc.epoch_millis) / 1000;

        // Expected: In UTC+5, the offset should be exactly 5 hours
        assert_eq!(offset_seconds, 5 * 3600);

        // Restore original TZ
        unsafe {
            match original_tz {
                Some(tz) => std::env::set_var("TZ", tz),
                None => std::env::remove_var("TZ"),
            }
        }
    }

    #[test]
    fn test_parse_action_and_state() -> Result<()> {
        // Test action with state suffix
        let (action, state) = Instant::parse_action_and_state("commit.inflight")?;
        assert_eq!(action, Action::Commit);
        assert_eq!(state, State::Inflight);

        let (action, state) = Instant::parse_action_and_state("deltacommit.requested")?;
        assert_eq!(action, Action::DeltaCommit);
        assert_eq!(state, State::Requested);

        // Test standalone "inflight" special case
        let (action, state) = Instant::parse_action_and_state("inflight")?;
        assert_eq!(action, Action::Commit);
        assert_eq!(state, State::Inflight);

        // Test completed state (no suffix)
        let (action, state) = Instant::parse_action_and_state("commit")?;
        assert_eq!(action, Action::Commit);
        assert_eq!(state, State::Completed);

        Ok(())
    }

    #[test]
    fn test_relative_path_with_base() -> Result<()> {
        let instant = Instant::from_str("20240101120000.commit")?;

        let path = instant.relative_path_with_base(".hoodie")?;
        assert_eq!(path, ".hoodie/20240101120000.commit");

        let path = instant.relative_path_with_base(".hoodie/timeline")?;
        assert_eq!(path, ".hoodie/timeline/20240101120000.commit");

        Ok(())
    }

    #[test]
    fn test_is_replacecommit() -> Result<()> {
        let replace_instant = Instant::from_str("20240101120000.replacecommit")?;
        assert!(replace_instant.is_replacecommit());

        let commit_instant = Instant::from_str("20240101120000.commit")?;
        assert!(!commit_instant.is_replacecommit());

        Ok(())
    }

    #[test]
    fn test_v8_instant_with_completion_timestamp() -> Result<()> {
        // v8+ format: {requestedTimestamp}_{completionTimestamp}.{action}
        let instant = Instant::from_str("20240101120000000_20240101120005000.commit")?;
        assert_eq!(instant.timestamp, "20240101120000000");
        assert_eq!(
            instant.completion_timestamp,
            Some("20240101120005000".to_string())
        );
        assert_eq!(instant.action, Action::Commit);
        assert_eq!(instant.state, State::Completed);

        Ok(())
    }

    #[test]
    fn test_v8_instant_underscore_format_requires_completed_state() {
        // Underscore format with non-completed state should fail
        let result = Instant::from_str("20240101120000000_20240101120005000.commit.inflight");
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("only valid for completed")
        );
    }

    #[test]
    fn test_file_name_with_completion_timestamp() -> Result<()> {
        let instant = Instant::from_str("20240101120000000_20240101120005000.commit")?;
        let file_name = instant.file_name();
        assert_eq!(file_name, "20240101120000000_20240101120005000.commit");

        Ok(())
    }

    #[test]
    fn test_parse_datetime_with_milliseconds() -> Result<()> {
        let dt = Instant::parse_datetime("20240315142530500", "UTC")?;
        assert_eq!(dt.timestamp_millis() % 1000, 500);

        Ok(())
    }

    #[test]
    fn test_validate_timestamp_errors() {
        // Too short
        let result = Instant::from_str("2024.commit");
        assert!(result.is_err());

        // Too long (not 14 or 17)
        let result = Instant::from_str("202403151425301.commit");
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_datetime_epoch_millis() -> Result<()> {
        // Epoch 0 (1970-01-01 00:00:00.000 UTC)
        let dt = Instant::parse_datetime("00000000000000000", "UTC")?;
        assert_eq!(dt.timestamp_millis(), 0);

        // Epoch 1ms
        let dt = Instant::parse_datetime("00000000000000001", "UTC")?;
        assert_eq!(dt.timestamp_millis(), 1);

        // Epoch 1000ms (1 second)
        let dt = Instant::parse_datetime("00000000000001000", "UTC")?;
        assert_eq!(dt.timestamp_millis(), 1000);

        // A larger epoch value
        let dt = Instant::parse_datetime("00001734567890123", "UTC")?;
        assert_eq!(dt.timestamp_millis(), 1734567890123);

        Ok(())
    }

    /// Hudi's bootstrapping sentinels are valid instant times, and a caller may
    /// legitimately hand one to an incremental read as "from the beginning".
    ///
    /// They are 14 characters, so the epoch-millis fallback — which only applies
    /// to 17 — did not reach them, and they are not dates, so the date parse
    /// rejected them too. `hoodie.datasource.read.begin.instanttime` set to
    /// `HoodieTimeline.INIT_INSTANT_TS`, which Spark accepts, failed the read.
    #[test]
    fn test_parse_datetime_bootstrap_sentinels() -> Result<()> {
        // The values Hudi defines in HoodieTimeline, with the epoch instant each
        // stands for. They order the same way as their 17-character metadata
        // table counterparts, which is what callers rely on.
        for (sentinel, expected_millis) in [
            ("00000000000000", 0),
            ("00000000000001", 1),
            ("00000000000002", 2),
        ] {
            let dt = Instant::parse_datetime(sentinel, "UTC")?;
            assert_eq!(
                dt.timestamp_millis(),
                expected_millis,
                "sentinel '{sentinel}' must parse to epoch {expected_millis}ms"
            );
            assert!(
                dt < Instant::parse_datetime("20240315142530500", "UTC")?,
                "sentinel '{sentinel}' must order before a real instant"
            );
        }
        Ok(())
    }

    /// Accepting the sentinels must not turn every unparseable 14-character
    /// value into an epoch reading: `20241332000000` is a malformed date (month
    /// 13, day 32), and reading it as epoch millis would silently place it in
    /// the year 2611 instead of reporting it.
    #[test]
    fn test_parse_datetime_malformed_14_char_still_errors() {
        let err = Instant::parse_datetime("20241332000000", "UTC")
            .expect_err("a malformed date must not be reinterpreted as epoch millis");
        assert!(
            format!("{err}").contains("20241332000000"),
            "the error must name the offending value, got: {err}"
        );
    }

    #[test]
    fn test_parse_datetime_epoch_ordering() -> Result<()> {
        // Verify ordering: epoch 0 < epoch 1 < epoch 2 < ... < real timestamp
        let epoch_0 = Instant::parse_datetime("00000000000000000", "UTC")?;
        let epoch_1 = Instant::parse_datetime("00000000000000001", "UTC")?;
        let epoch_2 = Instant::parse_datetime("00000000000000002", "UTC")?;
        let real_ts = Instant::parse_datetime("20240315142530500", "UTC")?;

        assert!(epoch_0 < epoch_1);
        assert!(epoch_1 < epoch_2);
        assert!(epoch_2 < real_ts);

        Ok(())
    }

    #[test]
    fn test_parse_datetime_date_format_still_works() -> Result<()> {
        // Valid date format should still parse correctly (not treated as epoch)
        let dt = Instant::parse_datetime("20240101120000000", "UTC")?;
        // This is Jan 1, 2024, 12:00:00.000 UTC - NOT epoch 20240101120000000
        assert_eq!(
            dt.format("%Y-%m-%d %H:%M:%S").to_string(),
            "2024-01-01 12:00:00"
        );

        Ok(())
    }

    /// Every action round-trips through its on-disk suffix.
    ///
    /// The suffix is the file extension Hudi writes, so a mismatch here means an
    /// instant file is silently discarded by the loader rather than failing --
    /// which is exactly how rollback and restore went unseen before they were
    /// added.
    #[test]
    fn every_action_round_trips_through_its_suffix() {
        for action in [
            Action::Commit,
            Action::DeltaCommit,
            Action::ReplaceCommit,
            Action::Rollback,
            Action::Restore,
        ] {
            let suffix = action.as_ref();
            assert_eq!(
                Action::from_str(suffix).unwrap(),
                action,
                "`{suffix}` must parse back to the action that produced it"
            );
        }
    }

    /// Rollback and restore are deliberately **not** in the default loading set.
    ///
    /// They are not commits. Adding them there would put them in
    /// `completed_commits`, which drives file-slice discovery and the commit
    /// visibility gate -- so every existing read would change. They are loaded
    /// only by a caller that names them.
    #[test]
    fn rollback_and_restore_are_not_loaded_by_default() {
        let defaults = crate::timeline::DEFAULT_LOADING_ACTIONS;
        assert!(
            !defaults.contains(&Action::Rollback) && !defaults.contains(&Action::Restore),
            "adding these to the default set would change every existing read, \
             got: {defaults:?}"
        );
        // The premise: the set is not empty, so the assertion above is not
        // vacuously true.
        assert!(defaults.contains(&Action::Commit));
    }
}