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::BaseFileFormatValue;
use crate::error::CoreError;
use crate::file_group::FileGroup;
use crate::file_group::base_file::BaseFile;
use crate::file_group::log_file::LogFile;
use crate::metadata::commit::{HoodieCommitMetadata, HoodieWriteStat};
use crate::metadata::replace_commit::HoodieReplaceCommitMetadata;
use crate::metadata::table_record::FilesPartitionRecord;
use crate::statistics::estimator::FileStatsEstimator;
use crate::storage::file_metadata::FileMetadata;
use crate::timeline::completion_time::CompletionTimeView;
use dashmap::DashMap;
use serde_json::{Map, Value};
use std::collections::{HashMap, HashSet};
use std::path::Path;
use std::str::FromStr;

pub trait FileGroupMerger {
    fn merge<I>(&mut self, file_groups: I) -> Result<()>
    where
        I: IntoIterator<Item = FileGroup>;
}

impl FileGroupMerger for HashSet<FileGroup> {
    fn merge<I>(&mut self, file_groups: I) -> Result<()>
    where
        I: IntoIterator<Item = FileGroup>,
    {
        for file_group in file_groups {
            if let Some(mut existing) = self.take(&file_group) {
                existing.merge(&file_group)?;
                self.insert(existing);
            } else {
                self.insert(file_group);
            }
        }
        Ok(())
    }
}

/// # Arguments
///
/// * `commit_metadata` - The commit metadata JSON map
/// * `completion_time_view` - View to look up completion timestamps.
/// * `estimator` - Optional [FileStatsEstimator] used to populate `byte_size`
///   and `num_records` on each base file. When `None`, those fields stay at 0.
///   `size` is populated from `HoodieWriteStat::file_size_in_bytes` only when
///   the write stat path identifies the base file.
pub(crate) fn file_groups_from_commit_metadata_with_estimator<V: CompletionTimeView>(
    commit_metadata: &Map<String, Value>,
    completion_time_view: &V,
    estimator: Option<&FileStatsEstimator>,
) -> Result<CommitFileGroups> {
    let metadata = HoodieCommitMetadata::from_json_map(commit_metadata)?;

    let mut file_groups = HashSet::new();
    let mut unattached_log_files = Vec::new();

    for (partition, write_stat) in metadata.iter_write_stats() {
        let file_id = write_stat
            .file_id
            .as_ref()
            .ok_or_else(|| CoreError::CommitMetadata("Missing fileId in write stats".into()))?;

        let mut file_group = FileGroup::new(file_id.clone(), partition.clone());

        // Resolve the base file name: MOR write stats record `baseFile`; COW write
        // stats record `path` (`<partition>/<file>`). Either way, derive the file
        // name string used to construct the BaseFile.
        let base_file_name: Option<String> = match write_stat.base_file.as_deref() {
            // A delta commit that only appends to a file group leaves `baseFile`
            // empty and points `path` at the log file it wrote. That is not a
            // name to parse — the slice it belongs to was written by an earlier
            // commit, so the log file waits until every base file is known.
            Some("") | None => {
                let path = write_stat.path.as_deref().ok_or_else(|| {
                    CoreError::CommitMetadata("Missing path in write stats".into())
                })?;
                let name = Path::new(path)
                    .file_name()
                    .and_then(|name| name.to_str())
                    .ok_or_else(|| CoreError::CommitMetadata("Invalid file name in path".into()))?;
                if LogFile::is_log_file_name(name) {
                    None
                } else {
                    Some(name.to_string())
                }
            }
            Some(name) => Some(name.to_string()),
        };

        let Some(base_file_name) = base_file_name else {
            for log_file_name in log_file_names_in(write_stat) {
                // Parsed to reject a malformed name here rather than later; the
                // file group's identity is all the incremental path needs, since
                // it reads the slice live at the range's end.
                LogFile::from_str(&log_file_name)?;
                unattached_log_files.push(UnattachedLogFile {
                    partition: partition.clone(),
                    file_id: file_id.clone(),
                });
            }
            continue;
        };

        let mut base_file = BaseFile::from_str(&base_file_name)?;
        base_file.set_completion_time(completion_time_view);

        let path_file_name = write_stat
            .path
            .as_deref()
            .and_then(|path| Path::new(path).file_name())
            .and_then(|name| name.to_str());
        let write_stat_size_is_for_base_file = path_file_name == Some(base_file_name.as_str());

        // Populate FileMetadata only when the write-stat size belongs to the base
        // file. MOR delta commits can have `baseFile` set while `path` and
        // `fileSizeInBytes` describe a newly written log file.
        if let Some(file_size) = write_stat
            .file_size_in_bytes
            .filter(|s| write_stat_size_is_for_base_file && *s > 0)
        {
            let size = file_size as u64;
            let (byte_size, num_records) = estimator.map(|e| e.estimate(size)).unwrap_or((0, 0));
            base_file.file_metadata = Some(FileMetadata {
                name: base_file_name,
                size,
                byte_size,
                num_records,
            });
        }

        file_group.add_base_file(base_file)?;

        // Log files are only present in MOR write stats (the `baseFile` branch).
        if let Some(log_file_names) = &write_stat.log_files {
            for log_file_name in log_file_names {
                let mut log_file = LogFile::from_str(log_file_name)?;
                log_file.set_completion_time(completion_time_view);
                file_group.add_log_file(log_file)?;
            }
        }

        file_groups.insert(file_group);
    }

    Ok(CommitFileGroups {
        file_groups,
        unattached_log_files,
    })
}

/// A log file whose file slice was not written by the same commit.
///
/// Its slice is keyed by a base instant an earlier commit produced, so it can
/// only be placed once every commit in the range has contributed its base files.
pub(crate) struct UnattachedLogFile {
    pub partition: String,
    pub file_id: String,
}

/// What one commit's metadata contributes to an incremental read.
pub(crate) struct CommitFileGroups {
    pub file_groups: HashSet<FileGroup>,
    pub unattached_log_files: Vec<UnattachedLogFile>,
}

/// The log files a write stat says it wrote.
///
/// `log_files` carries them when the stat also names a base file; a log-only
/// stat names its single log file in `path`.
fn log_file_names_in(write_stat: &HoodieWriteStat) -> Vec<String> {
    if let Some(names) = &write_stat.log_files
        && !names.is_empty()
    {
        return names.clone();
    }
    write_stat
        .path
        .as_deref()
        .and_then(|path| Path::new(path).file_name())
        .and_then(|name| name.to_str())
        .filter(|name| LogFile::is_log_file_name(name))
        .map(|name| vec![name.to_string()])
        .unwrap_or_default()
}

pub fn replaced_file_groups_from_replace_commit(
    commit_metadata: &Map<String, Value>,
) -> Result<HashSet<FileGroup>> {
    // Replace commits follow HoodieReplaceCommitMetadata schema; parse with the dedicated type
    let metadata = HoodieReplaceCommitMetadata::from_json_map(commit_metadata)?;

    let mut file_groups = HashSet::new();

    for (partition, file_id) in metadata.iter_replace_file_ids() {
        let file_group = FileGroup::new(file_id.clone(), partition.clone());
        file_groups.insert(file_group);
    }

    Ok(file_groups)
}

/// Build FileGroups from metadata-table FilesPartitionRecords.
///
/// Used for **snapshot queries** when the metadata table is enabled. Parses file
/// names from the `.hoodie/metadata` table's files partition records and
/// constructs FileGroup objects. Only includes active (non-deleted) files.
///
/// If `configured_base_file_format` is `Some`, only that suffix is treated as a
/// base file. If it is `None`, any known base-file suffix is accepted so tables
/// without `hoodie.table.base.file.format` can be discovered by extension.
///
/// # Arguments
/// * `records` - Metadata table files partition records (partition_path -> FilesPartitionRecord)
/// * `configured_base_file_format` - Explicit base-file format from table config, or
///   `None` to fall back to extension-based discovery.
/// * `completion_time_view` - View to look up completion timestamps.
/// * `estimator` - Optional [FileStatsEstimator] used to populate `byte_size`
///   and `num_records` on each base file.
///
/// # Returns
/// A map of partition paths to their FileGroups.
pub(crate) fn file_groups_from_files_partition_records<V: CompletionTimeView>(
    records: &HashMap<String, FilesPartitionRecord>,
    configured_base_file_format: Option<&BaseFileFormatValue>,
    completion_time_view: &V,
    estimator: Option<&FileStatsEstimator>,
) -> Result<DashMap<String, Vec<FileGroup>>> {
    let file_groups_map = DashMap::new();

    for (partition_path, record) in records {
        // Skip __all_partitions__ record - it lists partition names, not files
        if record.is_all_partitions() {
            continue;
        }

        let mut file_id_to_base_files: HashMap<String, Vec<BaseFile>> = HashMap::new();
        let mut file_id_to_log_files: HashMap<String, Vec<LogFile>> = HashMap::new();

        for (file_name, file_size) in record.active_files_with_sizes() {
            if file_name.starts_with('.') {
                // Log file: starts with '.'
                let mut log_file = LogFile::from_str(file_name).map_err(|e| {
                    CoreError::FileGroup(format!(
                        "Metadata table contains invalid/unsupported log file name '{file_name}' in partition '{partition_path}': {e}. \
                         This may indicate data corruption."
                    ))
                })?;

                log_file.set_completion_time(completion_time_view);
                // Skip the file if the commit that wrote it never completed.
                if !completion_time_view.is_committed(&log_file.timestamp) {
                    continue;
                }
                if file_size > 0 {
                    log_file.file_metadata =
                        Some(FileMetadata::new(file_name.to_string(), file_size));
                }
                file_id_to_log_files
                    .entry(log_file.file_id.clone())
                    .or_default()
                    .push(log_file);
            } else if configured_base_file_format.as_ref().map_or_else(
                || BaseFileFormatValue::from_extension(file_name).is_some(),
                |format| format.matches_extension(file_name),
            ) {
                // Base file: ends with base file extension
                let mut base_file = BaseFile::from_str(file_name).map_err(|e| {
                    CoreError::FileGroup(format!(
                        "Metadata table contains invalid/unsupported base file name '{file_name}' in partition '{partition_path}': {e}. \
                         This may indicate data corruption."
                    ))
                })?;

                base_file.set_completion_time(completion_time_view);
                // Skip the file if the commit that wrote it never completed.
                if !completion_time_view.is_committed(&base_file.commit_timestamp) {
                    continue;
                }
                // Populate file metadata with on-disk size from MDT,
                // and estimated byte_size/num_records if estimator is available.
                if file_size > 0 {
                    let (byte_size, num_records) =
                        estimator.map(|e| e.estimate(file_size)).unwrap_or((0, 0));
                    base_file.file_metadata = Some(FileMetadata {
                        name: file_name.to_string(),
                        size: file_size,
                        byte_size,
                        num_records,
                    });
                }
                file_id_to_base_files
                    .entry(base_file.file_id.clone())
                    .or_default()
                    .push(base_file);
            }
            // Skip files with unrecognized extensions
        }

        // Build FileGroups from parsed files
        // Note: Currently only supports file groups with base files.
        // TODO: Support file groups with only log files (P1 task)
        let mut file_groups = Vec::new();
        for (file_id, base_files) in file_id_to_base_files {
            let mut fg = FileGroup::new(file_id.clone(), partition_path.clone());
            fg.add_base_files(base_files)?;

            // Add corresponding log files if any
            if let Some(log_files) = file_id_to_log_files.remove(&file_id) {
                fg.add_log_files(log_files)?;
            }

            file_groups.push(fg);
        }

        if !file_groups.is_empty() {
            file_groups_map.insert(partition_path.clone(), file_groups);
        }
    }

    Ok(file_groups_map)
}

#[cfg(test)]
mod tests {
    mod test_file_group_merger {
        use super::super::*;
        use crate::file_group::FileGroup;

        #[test]
        fn test_merge_file_groups() {
            let mut existing = HashSet::new();
            let fg1 = FileGroup::new("file1".to_string(), "p1".to_string());
            existing.insert(fg1);

            let new_groups = vec![
                FileGroup::new("file2".to_string(), "p1".to_string()),
                FileGroup::new("file3".to_string(), "p2".to_string()),
            ];

            existing.merge(new_groups).unwrap();
            assert_eq!(existing.len(), 3);
        }

        #[test]
        fn test_merge_empty() {
            let mut existing = HashSet::new();
            let fg1 = FileGroup::new("file1".to_string(), "p1".to_string());
            existing.insert(fg1);

            let new_groups: Vec<FileGroup> = vec![];
            existing.merge(new_groups).unwrap();
            assert_eq!(existing.len(), 1);
        }
    }

    mod test_file_groups_from_commit_metadata {
        use super::super::*;
        use crate::config::HudiConfigs;
        use crate::timeline::instant::{Action, Instant, State};
        use crate::timeline::view::TimelineView;
        use serde_json::{Map, Value, json};
        use std::collections::HashSet;
        use std::sync::Arc;

        /// The file groups one commit contributes, without the log files whose
        /// slice another commit owns. These tests predate log-only slices and
        /// assert on the groups alone.
        fn file_groups_from_commit_metadata<V: CompletionTimeView>(
            commit_metadata: &Map<String, Value>,
            completion_time_view: &V,
        ) -> Result<HashSet<FileGroup>> {
            file_groups_from_commit_metadata_with_estimator(
                commit_metadata,
                completion_time_view,
                None,
            )
            .map(|contribution| contribution.file_groups)
        }

        /// A view that admits every file, so these tests exercise file-name
        /// parsing and grouping rather than commit visibility. The archival
        /// boundary is the far-future sentinel: every instant sorts below it, so
        /// every instant reads as archived and therefore as committed. Commit
        /// visibility itself is covered in `timeline::view`.
        fn create_layout_v1_view() -> TimelineView {
            let configs = Arc::new(HudiConfigs::new([("hoodie.timeline.layout.version", "1")]));
            TimelineView::new_with_archival_boundary(
                "99999999999999999".to_string(),
                None,
                &[] as &[Instant],
                HashSet::new(),
                &configs,
                Some("99999999999999999".to_string()),
            )
        }

        /// As [`create_layout_v1_view`]: admits every file so the assertions stay
        /// about parsing and grouping.
        fn create_layout_v2_view(instants: &[Instant]) -> TimelineView {
            let configs = Arc::new(HudiConfigs::new([("hoodie.timeline.layout.version", "2")]));
            TimelineView::new_with_archival_boundary(
                "99999999999999999".to_string(),
                None,
                instants,
                HashSet::new(),
                &configs,
                Some("99999999999999999".to_string()),
            )
        }

        #[test]
        fn test_missing_partition_to_write_stats() {
            let metadata: Map<String, Value> = json!({
                "compacted": false,
                "operationType": "UPSERT"
            })
            .as_object()
            .unwrap()
            .clone();

            // Use layout v1 view (no completion time tracking)
            let result = file_groups_from_commit_metadata(&metadata, &create_layout_v1_view());
            // With the new implementation, this returns Ok with an empty HashSet
            // because iter_write_stats() returns an empty iterator when partition_to_write_stats is None
            assert!(result.is_ok());
            assert_eq!(result.unwrap().len(), 0);
        }

        #[test]
        fn test_invalid_write_stats_array() {
            let metadata: Map<String, Value> = json!({
                "partitionToWriteStats": {
                    "byteField=20/shortField=100": "not_an_array"
                }
            })
            .as_object()
            .unwrap()
            .clone();

            let result = file_groups_from_commit_metadata(&metadata, &create_layout_v1_view());
            assert!(matches!(
                result,
                Err(CoreError::CommitMetadata(msg)) if msg.contains("Failed to parse commit metadata")
            ));
        }

        #[test]
        fn test_missing_file_id() {
            let metadata: Map<String, Value> = json!({
                "partitionToWriteStats": {
                    "byteField=20/shortField=100": [{
                        "path": "byteField=20/shortField=100/some-file.parquet"
                    }]
                }
            })
            .as_object()
            .unwrap()
            .clone();

            let result = file_groups_from_commit_metadata(&metadata, &create_layout_v1_view());
            assert!(matches!(
                result,
                Err(CoreError::CommitMetadata(msg)) if msg == "Missing fileId in write stats"
            ));
        }

        #[test]
        fn test_missing_path() {
            let metadata: Map<String, Value> = json!({
                "partitionToWriteStats": {
                    "byteField=20/shortField=100": [{
                        "fileId": "bb7c3a45-387f-490d-aab2-981c3f1a8ada-0"
                    }]
                }
            })
            .as_object()
            .unwrap()
            .clone();

            let result = file_groups_from_commit_metadata(&metadata, &create_layout_v1_view());
            assert!(matches!(
                result,
                Err(CoreError::CommitMetadata(msg)) if msg == "Missing path in write stats"
            ));
        }

        #[test]
        fn test_invalid_path_format() {
            let metadata: Map<String, Value> = json!({
                "partitionToWriteStats": {
                    "byteField=20/shortField=100": [{
                        "fileId": "bb7c3a45-387f-490d-aab2-981c3f1a8ada-0",
                        "path": "" // empty path
                    }]
                }
            })
            .as_object()
            .unwrap()
            .clone();

            let result = file_groups_from_commit_metadata(&metadata, &create_layout_v1_view());
            assert!(matches!(
                result,
                Err(CoreError::CommitMetadata(msg)) if msg == "Invalid file name in path"
            ));
        }

        #[test]
        fn test_non_string_field_types_fail_to_parse() {
            // Non-string `fileId` or `path` should both surface as a serde
            // deserialization error from HoodieCommitMetadata::from_json_map.
            let cases = [
                json!({
                    "partitionToWriteStats": {
                        "byteField=20/shortField=100": [{
                            "fileId": 123,
                            "path": "byteField=20/shortField=100/some-file.parquet"
                        }]
                    }
                }),
                json!({
                    "partitionToWriteStats": {
                        "byteField=20/shortField=100": [{
                            "fileId": "bb7c3a45-387f-490d-aab2-981c3f1a8ada-0",
                            "path": 123
                        }]
                    }
                }),
            ];

            for value in cases {
                let metadata: Map<String, Value> = value.as_object().unwrap().clone();
                let result = file_groups_from_commit_metadata(&metadata, &create_layout_v1_view());
                assert!(matches!(
                    result,
                    Err(CoreError::CommitMetadata(msg)) if msg.contains("Failed to parse commit metadata")
                ));
            }
        }

        #[test]
        fn test_valid_sample_data() {
            let sample_json = r#"{
            "partitionToWriteStats": {
                "byteField=20/shortField=100": [{
                    "fileId": "bb7c3a45-387f-490d-aab2-981c3f1a8ada-0",
                    "path": "byteField=20/shortField=100/bb7c3a45-387f-490d-aab2-981c3f1a8ada-0_0-140-198_20240418173213674.parquet"
                }],
                "byteField=10/shortField=300": [{
                    "fileId": "a22e8257-e249-45e9-ba46-115bc85adcba-0",
                    "path": "byteField=10/shortField=300/a22e8257-e249-45e9-ba46-115bc85adcba-0_1-140-199_20240418173213674.parquet"
                }]
            }
        }"#;

            let metadata: Map<String, Value> = serde_json::from_str(sample_json).unwrap();
            let result = file_groups_from_commit_metadata(&metadata, &create_layout_v1_view());
            assert!(result.is_ok());
            let file_groups = result.unwrap();
            assert_eq!(file_groups.len(), 2);

            let expected_partitions = HashSet::from_iter(vec![
                "byteField=20/shortField=100",
                "byteField=10/shortField=300",
            ]);
            let actual_partitions =
                HashSet::<&str>::from_iter(file_groups.iter().map(|fg| fg.partition_path.as_str()));
            assert_eq!(actual_partitions, expected_partitions);
        }

        #[test]
        fn test_mor_table_with_base_file_and_log_files() {
            let sample_json = r#"{
            "partitionToWriteStats": {
                "partition1": [{
                    "fileId": "file-id-0",
                    "baseFile": "file-id-0_0-7-24_20240418173200000.parquet",
                    "logFiles": [
                        ".file-id-0_20240418173200000.log.1_0-8-25",
                        ".file-id-0_20240418173200000.log.2_0-9-26"
                    ]
                }]
            }
        }"#;

            let metadata: Map<String, Value> = serde_json::from_str(sample_json).unwrap();
            let result = file_groups_from_commit_metadata(&metadata, &create_layout_v1_view());
            assert!(result.is_ok());
            let file_groups = result.unwrap();
            assert_eq!(file_groups.len(), 1);

            let file_group = file_groups.iter().next().unwrap();
            assert_eq!(file_group.file_id, "file-id-0");
            assert_eq!(file_group.partition_path, "partition1");

            // Check that we have a file slice with base file and log files
            assert_eq!(file_group.file_slices.len(), 1);
            let (_, file_slice) = file_group.file_slices.iter().next().unwrap();
            assert_eq!(
                file_slice.base_file.as_ref().unwrap().file_name(),
                "file-id-0_0-7-24_20240418173200000.parquet"
            );
            assert_eq!(file_slice.log_files.len(), 2);
        }

        #[test]
        fn test_mor_table_base_file_without_log_files() {
            // Both `logFiles` absent (initial insert) and `logFiles: []` (empty array)
            // should yield a file slice with no log files.
            let cases = [
                r#"{
                    "partitionToWriteStats": {
                        "partition1": [{
                            "fileId": "file-id-0",
                            "baseFile": "file-id-0_0-7-24_20240418173200000.parquet"
                        }]
                    }
                }"#,
                r#"{
                    "partitionToWriteStats": {
                        "partition1": [{
                            "fileId": "file-id-0",
                            "baseFile": "file-id-0_0-7-24_20240418173200000.parquet",
                            "logFiles": []
                        }]
                    }
                }"#,
            ];

            for sample_json in cases {
                let metadata: Map<String, Value> = serde_json::from_str(sample_json).unwrap();
                let file_groups =
                    file_groups_from_commit_metadata(&metadata, &create_layout_v1_view()).unwrap();
                assert_eq!(file_groups.len(), 1);
                let (_, file_slice) = file_groups
                    .iter()
                    .next()
                    .unwrap()
                    .file_slices
                    .iter()
                    .next()
                    .unwrap();
                assert!(file_slice.log_files.is_empty());
            }
        }

        #[test]
        fn test_file_groups_from_commit_metadata_with_completion_time_view() {
            let sample_json = r#"{
            "partitionToWriteStats": {
                "partition1": [{
                    "fileId": "file-id-0",
                    "path": "partition1/file-id-0_0-7-24_20240418173200000.parquet"
                }]
            }
        }"#;

            let metadata: Map<String, Value> = serde_json::from_str(sample_json).unwrap();

            // Create a completion time view with the mapping for this file's request timestamp
            let instants = vec![Instant {
                timestamp: "20240418173200000".to_string(),
                completion_timestamp: Some("20240418173210000".to_string()),
                action: Action::Commit,
                state: State::Completed,
                epoch_millis: 0,
            }];
            let view = create_layout_v2_view(&instants);

            let result = file_groups_from_commit_metadata(&metadata, &view);
            assert!(result.is_ok());
            let file_groups = result.unwrap();
            assert_eq!(file_groups.len(), 1);

            let file_group = file_groups.iter().next().unwrap();
            let file_slice = file_group.file_slices.values().next().unwrap();
            assert_eq!(
                file_slice.base_file.as_ref().unwrap().completion_timestamp,
                Some("20240418173210000".to_string())
            );
        }

        #[test]
        fn test_public_api_uses_no_estimator_for_cow_path_metadata() {
            let json = r#"{
                "partitionToWriteStats": {
                    "p1": [{
                        "fileId": "fid-0",
                        "path": "p1/fid-0_0-7-24_20240418173200000.parquet",
                        "fileSizeInBytes": 4096
                    }]
                }
            }"#;
            let metadata: Map<String, Value> = serde_json::from_str(json).unwrap();

            let groups =
                file_groups_from_commit_metadata(&metadata, &create_layout_v1_view()).unwrap();
            let file_slice = groups
                .iter()
                .next()
                .unwrap()
                .file_slices
                .values()
                .next()
                .unwrap();
            assert!(file_slice.log_files.is_empty());
            let m = file_slice
                .base_file
                .as_ref()
                .unwrap()
                .file_metadata
                .as_ref()
                .unwrap();
            assert_eq!(m.name, "fid-0_0-7-24_20240418173200000.parquet");
            assert_eq!(m.size, 4096);
            assert_eq!(m.byte_size, 0);
            assert_eq!(m.num_records, 0);
        }

        #[test]
        fn test_metadata_populated_from_write_stat_size_with_estimator() {
            // No-estimator coverage lives in test_public_api_uses_no_estimator_for_cow_path_metadata.
            // This test focuses on estimator-derived byte_size / num_records for the MOR
            // baseFile path.
            let json = r#"{
                "partitionToWriteStats": {
                    "p1": [{
                        "fileId": "fid-0",
                        "baseFile": "fid-0_0-7-24_20240418173200000.parquet",
                        "path": "p1/fid-0_0-7-24_20240418173200000.parquet",
                        "fileSizeInBytes": 4096
                    }]
                }
            }"#;
            let metadata: Map<String, Value> = serde_json::from_str(json).unwrap();

            let estimator = FileStatsEstimator::new(100.0, 2.5);
            let groups = file_groups_from_commit_metadata_with_estimator(
                &metadata,
                &create_layout_v1_view(),
                Some(&estimator),
            )
            .unwrap()
            .file_groups;
            let m = groups
                .iter()
                .next()
                .unwrap()
                .file_slices
                .values()
                .next()
                .unwrap()
                .base_file
                .as_ref()
                .unwrap()
                .file_metadata
                .as_ref()
                .unwrap();
            assert_eq!(m.size, 4096);
            assert_eq!(m.byte_size, 10240); // 4096 * 2.5
            assert_eq!(m.num_records, 40); // 4096 / 100
        }

        #[test]
        fn test_mor_log_write_stat_does_not_assign_log_size_to_base_file() {
            let json = r#"{
                "partitionToWriteStats": {
                    "p1": [{
                        "fileId": "fid-0",
                        "baseFile": "fid-0_0-7-24_20240418173200000.parquet",
                        "path": "p1/.fid-0_20240418173200000.log.1_0-8-25",
                        "fileSizeInBytes": 1148,
                        "logFiles": [
                            ".fid-0_20240418173200000.log.1_0-8-25"
                        ]
                    }]
                }
            }"#;
            let metadata: Map<String, Value> = serde_json::from_str(json).unwrap();
            let estimator = FileStatsEstimator::new(100.0, 2.5);

            let groups = file_groups_from_commit_metadata_with_estimator(
                &metadata,
                &create_layout_v1_view(),
                Some(&estimator),
            )
            .unwrap()
            .file_groups;
            let file_slice = groups
                .iter()
                .next()
                .unwrap()
                .file_slices
                .values()
                .next()
                .unwrap();

            assert_eq!(file_slice.log_files.len(), 1);
            assert!(
                file_slice
                    .base_file
                    .as_ref()
                    .unwrap()
                    .file_metadata
                    .is_none()
            );
        }

        #[test]
        fn test_metadata_absent_when_no_file_size() {
            let json = r#"{
                "partitionToWriteStats": {
                    "p1": [{
                        "fileId": "fid-0",
                        "baseFile": "fid-0_0-7-24_20240418173200000.parquet"
                    }]
                }
            }"#;
            let metadata: Map<String, Value> = serde_json::from_str(json).unwrap();
            let groups =
                file_groups_from_commit_metadata(&metadata, &create_layout_v1_view()).unwrap();
            let fs = groups
                .iter()
                .next()
                .unwrap()
                .file_slices
                .values()
                .next()
                .unwrap();
            assert!(fs.base_file.as_ref().unwrap().file_metadata.is_none());
        }

        /// A log-only delta commit names its log file in `path` with no base
        /// file; it contributes an unattached log file, not a file group.
        #[test]
        fn test_log_only_delta_commit_contributes_unattached_log_files() {
            let json = r#"{
                "partitionToWriteStats": {
                    "p1": [{
                        "fileId": "fid-0",
                        "path": "p1/.fid-0_20240418173200000.log.1_0-8-25"
                    }]
                }
            }"#;
            let metadata: Map<String, Value> = serde_json::from_str(json).unwrap();

            let contribution = file_groups_from_commit_metadata_with_estimator(
                &metadata,
                &create_layout_v1_view(),
                None,
            )
            .unwrap();

            assert!(contribution.file_groups.is_empty());
            assert_eq!(contribution.unattached_log_files.len(), 1);
            assert_eq!(contribution.unattached_log_files[0].partition, "p1");
            assert_eq!(contribution.unattached_log_files[0].file_id, "fid-0");
        }

        /// A malformed log-file name in a log-only write stat fails loudly
        /// instead of contributing a slice that can never be located.
        #[test]
        fn test_log_only_delta_commit_with_invalid_log_name_errors() {
            let json = r#"{
                "partitionToWriteStats": {
                    "p1": [{
                        "fileId": "fid-0",
                        "path": "p1/.not-a-log-name.log",
                        "logFiles": [".not-a-log-name.log"]
                    }]
                }
            }"#;
            let metadata: Map<String, Value> = serde_json::from_str(json).unwrap();

            let result = file_groups_from_commit_metadata_with_estimator(
                &metadata,
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_err());
        }
    }

    mod test_file_groups_from_files_partition_records {
        use super::super::*;
        use crate::config::HudiConfigs;
        use crate::config::table::BaseFileFormatValue;
        use crate::metadata::table_record::{
            FilesPartitionRecord, HoodieMetadataFileInfo, MetadataRecordType,
        };
        use crate::timeline::instant::{Action, Instant, State};
        use crate::timeline::view::TimelineView;
        use std::collections::{HashMap, HashSet};
        use std::sync::Arc;

        /// A view that admits every file, so these tests exercise file-name
        /// parsing and grouping rather than commit visibility. The archival
        /// boundary is the far-future sentinel: every instant sorts below it, so
        /// every instant reads as archived and therefore as committed. Commit
        /// visibility itself is covered in `timeline::view`.
        fn create_layout_v1_view() -> TimelineView {
            let configs = Arc::new(HudiConfigs::new([("hoodie.timeline.layout.version", "1")]));
            TimelineView::new_with_archival_boundary(
                "99999999999999999".to_string(),
                None,
                &[] as &[Instant],
                HashSet::new(),
                &configs,
                Some("99999999999999999".to_string()),
            )
        }

        /// As [`create_layout_v1_view`]: admits every file so the assertions stay
        /// about parsing and grouping.
        fn create_layout_v2_view(instants: &[Instant]) -> TimelineView {
            let configs = Arc::new(HudiConfigs::new([("hoodie.timeline.layout.version", "2")]));
            TimelineView::new_with_archival_boundary(
                "99999999999999999".to_string(),
                None,
                instants,
                HashSet::new(),
                &configs,
                Some("99999999999999999".to_string()),
            )
        }

        /// A view with NO archival boundary, so nothing outside the completed set
        /// is readable — the strict counterpart used by the tests whose subject
        /// IS commit visibility.
        fn create_strict_view(instants: &[Instant]) -> TimelineView {
            let configs = Arc::new(HudiConfigs::new([("hoodie.timeline.layout.version", "2")]));
            TimelineView::new(
                "99999999999999999".to_string(),
                None,
                instants,
                HashSet::new(),
                &configs,
            )
        }

        fn create_file_info(name: &str, size: i64, is_deleted: bool) -> HoodieMetadataFileInfo {
            HoodieMetadataFileInfo::new(name.to_string(), size, is_deleted)
        }

        fn create_files_record(
            key: &str,
            files: Vec<(&str, i64, bool)>,
        ) -> (String, FilesPartitionRecord) {
            let mut files_map = HashMap::new();
            for (name, size, is_deleted) in files {
                files_map.insert(name.to_string(), create_file_info(name, size, is_deleted));
            }
            (
                key.to_string(),
                FilesPartitionRecord {
                    key: key.to_string(),
                    record_type: MetadataRecordType::Files,
                    files: files_map,
                },
            )
        }

        fn create_all_partitions_record(partitions: Vec<&str>) -> (String, FilesPartitionRecord) {
            let mut files_map = HashMap::new();
            for partition in partitions {
                files_map.insert(partition.to_string(), create_file_info(partition, 0, false));
            }
            (
                FilesPartitionRecord::ALL_PARTITIONS_KEY.to_string(),
                FilesPartitionRecord {
                    key: FilesPartitionRecord::ALL_PARTITIONS_KEY.to_string(),
                    record_type: MetadataRecordType::AllPartitions,
                    files: files_map,
                },
            )
        }

        #[test]
        fn test_empty_records() {
            let records: HashMap<String, FilesPartitionRecord> = HashMap::new();
            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();
            assert!(file_groups_map.is_empty());
        }

        /// Files whose commit never completed are skipped, base and log alike:
        /// a crashed or in-flight writer's files must not read as committed.
        #[test]
        fn test_uncommitted_base_and_log_files_are_skipped() {
            let committed = Instant {
                timestamp: "20240418173200000".to_string(),
                completion_timestamp: Some("20240418173210000".to_string()),
                action: Action::Commit,
                state: State::Completed,
                epoch_millis: 0,
            };
            let view = create_strict_view(&[committed]);

            let records: HashMap<String, FilesPartitionRecord> = [create_files_record(
                "p1",
                vec![
                    ("fid-0_0-7-24_20240418173200000.parquet", 1024, false),
                    (".fid-0_20240418173200000.log.1_0-8-25", 512, false),
                    // Same file group, written by an instant the timeline never
                    // completed: both must be dropped from the view.
                    ("fid-1_0-7-24_20240418999999999.parquet", 1024, false),
                    (".fid-0_20240418999999999.log.2_0-9-26", 512, false),
                ],
            )]
            .into();

            let file_groups_map = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &view,
                None,
            )
            .unwrap();

            let file_groups = file_groups_map.get("p1").unwrap();
            assert_eq!(file_groups.len(), 1, "the uncommitted base file is out");
            let fg = &file_groups[0];
            assert_eq!(fg.file_id, "fid-0");
            let slice = fg.file_slices.values().next().unwrap();
            assert_eq!(slice.log_files.len(), 1, "the uncommitted log file is out");
        }

        #[test]
        fn test_all_partitions_record_skipped() {
            let mut records = HashMap::new();
            let (key, record) = create_all_partitions_record(vec!["partition1", "partition2"]);
            records.insert(key, record);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();
            // __all_partitions__ record should be skipped
            assert!(file_groups_map.is_empty());
        }

        #[test]
        fn test_base_files_only() {
            let mut records = HashMap::new();
            let (key, record) = create_files_record(
                "partition1",
                vec![
                    ("file-id-0_0-7-24_20240418173200000.parquet", 1000, false),
                    ("file-id-1_0-8-25_20240418173210000.parquet", 2000, false),
                ],
            );
            records.insert(key, record);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();

            assert_eq!(file_groups_map.len(), 1);
            let file_groups = file_groups_map.get("partition1").unwrap();
            assert_eq!(file_groups.len(), 2);

            // Check file IDs are correctly parsed
            let file_ids: HashSet<_> = file_groups.iter().map(|fg| fg.file_id.as_str()).collect();
            assert!(file_ids.contains("file-id-0"));
            assert!(file_ids.contains("file-id-1"));
        }

        #[test]
        fn test_base_files_with_log_files() {
            let mut records = HashMap::new();
            let (key, record) = create_files_record(
                "partition1",
                vec![
                    // Base file for file-id-0
                    ("file-id-0_0-7-24_20240418173200000.parquet", 1000, false),
                    // Log files for file-id-0
                    (".file-id-0_20240418173200000.log.1_0-8-25", 100, false),
                    (".file-id-0_20240418173200000.log.2_0-9-26", 150, false),
                ],
            );
            records.insert(key, record);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();

            assert_eq!(file_groups_map.len(), 1);
            let file_groups = file_groups_map.get("partition1").unwrap();
            assert_eq!(file_groups.len(), 1);

            let fg = &file_groups[0];
            assert_eq!(fg.file_id, "file-id-0");

            // Check that log files were added
            let file_slice = fg.file_slices.values().next().unwrap();
            assert_eq!(file_slice.log_files.len(), 2);
        }

        #[test]
        fn test_multiple_partitions() {
            let mut records = HashMap::new();

            let (key1, record1) = create_files_record(
                "city=chennai",
                vec![("file-id-0_0-7-24_20240418173200000.parquet", 1000, false)],
            );
            records.insert(key1, record1);

            let (key2, record2) = create_files_record(
                "city=sao_paulo",
                vec![("file-id-1_0-8-25_20240418173210000.parquet", 2000, false)],
            );
            records.insert(key2, record2);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();

            assert_eq!(file_groups_map.len(), 2);
            assert!(file_groups_map.contains_key("city=chennai"));
            assert!(file_groups_map.contains_key("city=sao_paulo"));
        }

        #[test]
        fn test_deleted_files_excluded() {
            let mut records = HashMap::new();
            let (key, record) = create_files_record(
                "partition1",
                vec![
                    ("file-id-0_0-7-24_20240418173200000.parquet", 1000, false),
                    // This file is marked as deleted
                    ("file-id-1_0-8-25_20240418173210000.parquet", 2000, true),
                ],
            );
            records.insert(key, record);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();

            assert_eq!(file_groups_map.len(), 1);
            let file_groups = file_groups_map.get("partition1").unwrap();
            // Only 1 file group because deleted file is excluded
            assert_eq!(file_groups.len(), 1);
            assert_eq!(file_groups[0].file_id, "file-id-0");
        }

        #[test]
        fn test_unrecognized_file_extension_skipped() {
            let mut records = HashMap::new();
            let (key, record) = create_files_record(
                "partition1",
                vec![
                    ("file-id-0_0-7-24_20240418173200000.parquet", 1000, false),
                    // Unrecognized extension
                    ("somefile.txt", 500, false),
                ],
            );
            records.insert(key, record);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();

            assert_eq!(file_groups_map.len(), 1);
            let file_groups = file_groups_map.get("partition1").unwrap();
            // Only 1 file group, .txt file skipped
            assert_eq!(file_groups.len(), 1);
        }

        #[test]
        fn test_invalid_base_file_name_returns_error() {
            let mut records = HashMap::new();
            let (key, record) = create_files_record(
                "partition1",
                vec![
                    // Invalid base file name - missing required parts
                    ("invalid_file.parquet", 1000, false),
                ],
            );
            records.insert(key, record);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_err());
            let err = result.unwrap_err();
            assert!(
                err.to_string()
                    .contains("invalid/unsupported base file name")
            );
            assert!(err.to_string().contains("partition1"));
        }

        #[test]
        fn test_invalid_log_file_name_returns_error() {
            let mut records = HashMap::new();
            let (key, record) = create_files_record(
                "partition1",
                vec![
                    ("file-id-0_0-7-24_20240418173200000.parquet", 1000, false),
                    // Invalid log file name
                    (".invalid_log_file", 100, false),
                ],
            );
            records.insert(key, record);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_err());
            let err = result.unwrap_err();
            assert!(
                err.to_string()
                    .contains("invalid/unsupported log file name")
            );
        }

        #[test]
        fn test_layout_v2_uncommitted_base_file_filtered() {
            let mut records = HashMap::new();
            // A base file with timestamp that is NOT in the completion view (uncommitted)
            let (key, record) = create_files_record(
                "partition1",
                vec![("file-id-0_0-7-24_20240418173200000.parquet", 1000, false)],
            );
            records.insert(key, record);

            // Create a non-empty completion view that does NOT include the file's timestamp
            // This simulates an uncommitted file in timeline layout v2
            // The view has OTHER commits, so should_filter_uncommitted() returns true
            let instants = vec![Instant {
                timestamp: "20240418173199999".to_string(), // Different timestamp
                completion_timestamp: Some("20240418173209999".to_string()),
                action: Action::Commit,
                state: State::Completed,
                epoch_millis: 0,
            }];
            let view = create_strict_view(&instants);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &view,
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();

            // File should be filtered because it has no completion timestamp
            // (its request timestamp is not in the completion view)
            assert!(file_groups_map.is_empty());
        }

        #[test]
        fn test_empty_completion_view_filters_uncommitted() {
            // For timeline layout v2 with no completion time data (empty instants),
            // all files are filtered because their request timestamps don't have
            // corresponding completion timestamps in the view.
            let mut records = HashMap::new();
            let (key, record) = create_files_record(
                "partition1",
                vec![("file-id-0_0-7-24_20240418173200000.parquet", 1000, false)],
            );
            records.insert(key, record);

            let view = create_strict_view(&[]);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &view,
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();

            // Layout v2 with empty view filters all files (no completion timestamps available)
            assert!(file_groups_map.is_empty());
        }

        #[test]
        fn test_layout_v2_committed_base_file_included() {
            let mut records = HashMap::new();
            let (key, record) = create_files_record(
                "partition1",
                vec![("file-id-0_0-7-24_20240418173200000.parquet", 1000, false)],
            );
            records.insert(key, record);

            // Create a completion view that includes the file's timestamp
            let instants = vec![Instant {
                timestamp: "20240418173200000".to_string(),
                completion_timestamp: Some("20240418173210000".to_string()),
                action: Action::Commit,
                state: State::Completed,
                epoch_millis: 0,
            }];
            let view = create_layout_v2_view(&instants);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &view,
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();

            assert_eq!(file_groups_map.len(), 1);
            let file_groups = file_groups_map.get("partition1").unwrap();
            assert_eq!(file_groups.len(), 1);

            // Verify completion timestamp was set
            let file_slice = file_groups[0].file_slices.values().next().unwrap();
            assert_eq!(
                file_slice.base_file.as_ref().unwrap().completion_timestamp,
                Some("20240418173210000".to_string())
            );
        }

        #[test]
        fn test_layout_v2_uncommitted_log_file_filtered() {
            let mut records = HashMap::new();
            // Log file has a DIFFERENT base commit timestamp (20240418173205000)
            // that is NOT in the completion view
            let (key, record) = create_files_record(
                "partition1",
                vec![
                    // Committed base file with timestamp in the completion view
                    ("file-id-0_0-7-24_20240418173200000.parquet", 1000, false),
                    // Log file with a different base_commit_time that's NOT in the view
                    // The log file's timestamp (20240418173205000) won't have a completion time
                    (".file-id-0_20240418173205000.log.1_0-8-25", 100, false),
                ],
            );
            records.insert(key, record);

            // Completion view only includes the base file's timestamp, not the log's
            let instants = vec![Instant {
                timestamp: "20240418173200000".to_string(),
                completion_timestamp: Some("20240418173210000".to_string()),
                action: Action::Commit,
                state: State::Completed,
                epoch_millis: 0,
            }];
            let view = create_strict_view(&instants);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &view,
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();

            assert_eq!(file_groups_map.len(), 1);
            let file_groups = file_groups_map.get("partition1").unwrap();
            assert_eq!(file_groups.len(), 1);

            // Log file should be filtered - only base file remains
            // Because the log's base_commit_time is not in the completion view
            let file_slice = file_groups[0].file_slices.values().next().unwrap();
            assert!(file_slice.log_files.is_empty());
        }

        #[test]
        fn test_layout_v2_committed_log_file_included() {
            let mut records = HashMap::new();
            let (key, record) = create_files_record(
                "partition1",
                vec![
                    // Committed base file
                    ("file-id-0_0-7-24_20240418173200000.parquet", 1000, false),
                    // Log file with same base_commit_time that IS in the completion view
                    // Log file format: .{file_id}_{base_commit_time}.log.{version}_{write_token}
                    (".file-id-0_20240418173200000.log.1_0-8-25", 100, false),
                ],
            );
            records.insert(key, record);

            // Completion view includes the timestamp used by both base file and log file
            let instants = vec![Instant {
                timestamp: "20240418173200000".to_string(),
                completion_timestamp: Some("20240418173210000".to_string()),
                action: Action::Commit,
                state: State::Completed,
                epoch_millis: 0,
            }];
            let view = create_layout_v2_view(&instants);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &view,
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();

            assert_eq!(file_groups_map.len(), 1);
            let file_groups = file_groups_map.get("partition1").unwrap();
            assert_eq!(file_groups.len(), 1);

            // Both base file and log file should be included since they share the same timestamp
            let file_slice = file_groups[0].file_slices.values().next().unwrap();
            assert_eq!(file_slice.log_files.len(), 1);
        }

        #[test]
        fn test_empty_partition_not_added_to_map() {
            let mut records = HashMap::new();
            // Partition with only deleted files
            let (key, record) = create_files_record(
                "partition1",
                vec![(
                    "file-id-0_0-7-24_20240418173200000.parquet",
                    1000,
                    true, // deleted
                )],
            );
            records.insert(key, record);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();

            // Empty partition should not be in the map
            assert!(file_groups_map.is_empty());
        }

        #[test]
        fn test_multiple_base_files_same_file_id() {
            let mut records = HashMap::new();
            // Multiple base files with same file_id (different commit times)
            let (key, record) = create_files_record(
                "partition1",
                vec![
                    ("file-id-0_0-7-24_20240418173200000.parquet", 1000, false),
                    ("file-id-0_0-8-25_20240418173210000.parquet", 1500, false),
                ],
            );
            records.insert(key, record);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();

            assert_eq!(file_groups_map.len(), 1);
            let file_groups = file_groups_map.get("partition1").unwrap();
            // Should have 1 file group with 2 file slices
            assert_eq!(file_groups.len(), 1);
            assert_eq!(file_groups[0].file_slices.len(), 2);
        }

        #[test]
        fn test_log_files_without_base_file_not_included() {
            let mut records = HashMap::new();
            // Only log files, no base file
            let (key, record) = create_files_record(
                "partition1",
                vec![
                    (".file-id-0_20240418173200000.log.1_0-8-25", 100, false),
                    (".file-id-0_20240418173200000.log.2_0-9-26", 150, false),
                ],
            );
            records.insert(key, record);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();

            // Log-only file groups are not yet supported (see P1 task)
            assert!(file_groups_map.is_empty());
        }

        #[test]
        fn test_hfile_base_file_extension() {
            let mut records = HashMap::new();
            let (key, record) = create_files_record(
                "partition1",
                vec![("file-id-0_0-7-24_20240418173200000.hfile", 1000, false)],
            );
            records.insert(key, record);

            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::HFile),
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();

            assert_eq!(file_groups_map.len(), 1);
        }

        #[test]
        fn test_extension_fallback_accepts_lance_and_parquet_metadata_records() {
            let mut records = HashMap::new();
            let (key, record) = create_files_record(
                "partition1",
                vec![
                    ("file-id-0_0-7-24_20240418173200000.lance", 1000, false),
                    ("file-id-1_0-8-25_20240418173210000.parquet", 2000, false),
                    ("ignored.txt", 1, false),
                ],
            );
            records.insert(key, record);

            let result = file_groups_from_files_partition_records(
                &records,
                None,
                &create_layout_v1_view(),
                None,
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();
            let file_groups = file_groups_map.get("partition1").unwrap();
            let extensions: HashSet<_> = file_groups
                .iter()
                .flat_map(|fg| fg.file_slices.values())
                .map(|slice| slice.base_file.as_ref().unwrap().extension.as_str())
                .collect();

            assert_eq!(extensions, HashSet::from(["lance", "parquet"]));
        }

        #[test]
        fn test_with_estimator_populates_file_metadata() {
            use crate::statistics::estimator::FileStatsEstimator;

            let mut records = HashMap::new();
            let (key, record) = create_files_record(
                "partition1",
                vec![("file-id-0_0-7-24_20240418173200000.parquet", 5000, false)],
            );
            records.insert(key, record);

            // 250 bytes/row on disk, 2.0x compression ratio
            let estimator = FileStatsEstimator::new(250.0, 2.0);
            let result = file_groups_from_files_partition_records(
                &records,
                Some(&BaseFileFormatValue::Parquet),
                &create_layout_v1_view(),
                Some(&estimator),
            );
            assert!(result.is_ok());
            let file_groups_map = result.unwrap();
            assert_eq!(file_groups_map.len(), 1);

            let file_groups = file_groups_map.get("partition1").unwrap();
            let fg = &file_groups[0];
            let (_, file_slice) = fg.file_slices.iter().next().unwrap();
            let metadata = file_slice
                .base_file
                .as_ref()
                .unwrap()
                .file_metadata
                .as_ref()
                .unwrap();
            assert_eq!(metadata.size, 5000); // on-disk size
            assert_eq!(metadata.byte_size, 10000); // 5000 * 2.0
            assert_eq!(metadata.num_records, 20); // 5000 / 250
        }
    }

    mod test_replaced_file_groups_from_replace_commit {
        use super::super::*;
        use crate::table::partition::EMPTY_PARTITION_PATH;
        use serde_json::{Map, Value, json};

        #[test]
        fn test_missing_partition_to_replace() {
            let metadata: Map<String, Value> = json!({
                "compacted": false,
                "operationType": "UPSERT"
            })
            .as_object()
            .unwrap()
            .clone();

            let result = replaced_file_groups_from_replace_commit(&metadata);
            // With the new implementation, this returns Ok with an empty HashSet
            // because iter_replace_file_ids() returns an empty iterator when partition_to_replace_file_ids is None
            assert!(result.is_ok());
            assert_eq!(result.unwrap().len(), 0);
        }

        #[test]
        fn test_invalid_file_ids_array() {
            let metadata: Map<String, Value> = json!({
                "partitionToReplaceFileIds": {
                    "20": "not_an_array"
                }
            })
            .as_object()
            .unwrap()
            .clone();

            let result = replaced_file_groups_from_replace_commit(&metadata);
            assert!(matches!(
                result,
                Err(CoreError::CommitMetadata(msg)) if msg.contains("Failed to parse commit metadata")
            ));
        }

        #[test]
        fn test_invalid_file_id_type() {
            let metadata: Map<String, Value> = json!({
                "partitionToReplaceFileIds": {
                    "20": [123] // number instead of string
                }
            })
            .as_object()
            .unwrap()
            .clone();

            let result = replaced_file_groups_from_replace_commit(&metadata);
            // Serde will fail to parse this
            assert!(matches!(
                result,
                Err(CoreError::CommitMetadata(msg)) if msg.contains("Failed to parse commit metadata")
            ));
        }

        #[test]
        fn test_null_value_in_array() {
            let metadata: Map<String, Value> = json!({
                "partitionToReplaceFileIds": {
                    "20": [null]
                }
            })
            .as_object()
            .unwrap()
            .clone();

            let result = replaced_file_groups_from_replace_commit(&metadata);
            // Serde will fail to parse this
            assert!(matches!(
                result,
                Err(CoreError::CommitMetadata(msg)) if msg.contains("Failed to parse commit metadata")
            ));
        }

        #[test]
        fn test_empty_partition() {
            let metadata: Map<String, Value> = json!({
                "partitionToReplaceFileIds": {
                    "": ["d398fae1-c0e6-4098-8124-f55f7098bdba-0"]
                }
            })
            .as_object()
            .unwrap()
            .clone();

            let result = replaced_file_groups_from_replace_commit(&metadata);
            assert!(result.is_ok());
            let file_groups = result.unwrap();
            assert_eq!(file_groups.len(), 1);
            let file_group = file_groups.iter().next().unwrap();
            assert_eq!(file_group.partition_path, EMPTY_PARTITION_PATH);
        }

        #[test]
        fn test_multiple_file_groups_same_partition() {
            let metadata: Map<String, Value> = json!({
                "partitionToReplaceFileIds": {
                    "20": [
                        "88163884-fef0-4aab-865d-c72327a8a1d5-0",
                        "88163884-fef0-4aab-865d-c72327a8a1d5-1"
                    ]
                }
            })
            .as_object()
            .unwrap()
            .clone();

            let result = replaced_file_groups_from_replace_commit(&metadata);
            assert!(result.is_ok());
            let file_groups = result.unwrap();
            let actual_partition_paths = file_groups
                .iter()
                .map(|fg| fg.partition_path.as_str())
                .collect::<Vec<_>>();
            assert_eq!(actual_partition_paths, &["20", "20"]);
        }

        #[test]
        fn test_empty_array() {
            let metadata: Map<String, Value> = json!({
                "partitionToReplaceFileIds": {
                    "20": []
                }
            })
            .as_object()
            .unwrap()
            .clone();

            let result = replaced_file_groups_from_replace_commit(&metadata);
            assert!(result.is_ok());
            let file_groups = result.unwrap();
            assert!(file_groups.is_empty());
        }

        #[test]
        fn test_valid_sample_data() {
            let metadata: Map<String, Value> = json!({
                "partitionToReplaceFileIds": {
                    "30": ["d398fae1-c0e6-4098-8124-f55f7098bdba-0"],
                    "20": ["88163884-fef0-4aab-865d-c72327a8a1d5-0"],
                    "10": ["4f2685a3-614f-49ca-9b2b-e1cb9fb61f27-0"]
                }
            })
            .as_object()
            .unwrap()
            .clone();

            let result = replaced_file_groups_from_replace_commit(&metadata);
            assert!(result.is_ok());
            let file_groups = result.unwrap();
            assert_eq!(file_groups.len(), 3);

            let expected_partitions = HashSet::from_iter(vec!["10", "20", "30"]);
            let actual_partitions =
                HashSet::<&str>::from_iter(file_groups.iter().map(|fg| fg.partition_path.as_str()));
            assert_eq!(actual_partitions, expected_partitions);
        }
    }
}