mito2/manifest/

action.rs

// Copyright 2023 Greptime Team
//
// Licensed 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.

//! Defines [RegionMetaAction] related structs and [RegionCheckpoint].

use std::collections::{BTreeMap, HashMap, HashSet};
use std::time::Duration;

use chrono::Utc;
use common_telemetry::warn;
use serde::{Deserialize, Serialize};
use snafu::{OptionExt, ResultExt};
use store_api::ManifestVersion;
use store_api::metadata::RegionMetadataRef;
use store_api::storage::{FileId, IndexVersion, RegionId, SequenceNumber};
use strum::Display;

use crate::error::{RegionMetadataNotFoundSnafu, Result, SerdeJsonSnafu, Utf8Snafu};
use crate::manifest::manager::RemoveFileOptions;
use crate::region::ManifestStats;
use crate::sst::FormatType;
use crate::sst::file::{self, FileMeta, FileMetaRef};
use crate::sst::file_metadata::FileMetadataInternerRef;
use crate::wal::EntryId;

/// Actions that can be applied to region manifest.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Display)]
pub enum RegionMetaAction {
    /// Change region's metadata for request like ALTER
    Change(RegionChange),
    /// Change only region partition expression metadata.
    PartitionExprChange(RegionPartitionExprChange),
    /// Edit region's state for changing options or file list.
    Edit(RegionEdit),
    /// Remove the region.
    Remove(RegionRemove),
    /// Truncate the region.
    Truncate(RegionTruncate),
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct RegionPartitionExprChange {
    /// Partition expression serialized as JSON.
    pub partition_expr: Option<String>,
}

impl RegionMetaAction {
    /// Returns true if the action is a change action.
    pub fn is_change(&self) -> bool {
        matches!(self, RegionMetaAction::Change(_))
    }

    /// Returns true if the action is an edit action.
    pub fn is_edit(&self) -> bool {
        matches!(self, RegionMetaAction::Edit(_))
    }

    /// Returns true if the action is a partition expr change action.
    pub fn is_partition_expr_change(&self) -> bool {
        matches!(self, RegionMetaAction::PartitionExprChange(_))
    }
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct RegionChange {
    /// The metadata after changed.
    pub metadata: RegionMetadataRef,
    /// Format of the SST.
    #[serde(default)]
    pub sst_format: FormatType,
    /// Whether the region is in append mode.
    #[serde(default)]
    pub append_mode: Option<bool>,
}

#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)]
pub struct RegionEdit {
    pub files_to_add: Vec<FileMeta>,
    pub files_to_remove: Vec<FileMeta>,
    /// event unix timestamp in milliseconds, help to determine file deletion time.
    #[serde(default)]
    pub timestamp_ms: Option<i64>,
    #[serde(with = "humantime_serde")]
    pub compaction_time_window: Option<Duration>,
    pub flushed_entry_id: Option<EntryId>,
    pub flushed_sequence: Option<SequenceNumber>,
    pub committed_sequence: Option<SequenceNumber>,
    #[serde(flatten)]
    pub wal_recovery: WalRecoveryEdit,
}

/// Changes an edit makes to the WAL ranges the region still has to recover.
#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)]
pub struct WalRecoveryEdit {
    /// The new [RegionManifest::pending_tails], whose segments of each node keep the
    /// order and bounds described there. `None` keeps them unchanged.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pending_tails: Option<Vec<PendingTail>>,
    /// Replaces the handoff floor of one node.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub handoff_floor: Option<HandoffFloor>,
}

/// WAL entries of one node in `(after, upto]` that the region still has to recover.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct PendingTail {
    /// Id of the datanode whose WAL holds the entries.
    pub node_id: u64,
    pub after: EntryId,
    pub upto: EntryId,
    pub upto_kind: UptoKind,
}

/// Where the upper bound of a [PendingTail] comes from.
#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum UptoKind {
    /// The last entry id the node reported, which names an entry that can be replayed.
    Reported,
    /// An upper bound found by listing the WAL objects of the node, which may name no entry.
    Listed,
}

/// Entries of a node at or below `entry_id` are never recovered from its WAL, except
/// those in a pending tail of the node.
#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq)]
pub struct HandoffFloor {
    pub node_id: u64,
    pub entry_id: EntryId,
}

/// Caps a release of the WAL of node `node_id` at `entry_id`, so that it keeps every
/// pending tail of the node.
pub(crate) fn cap_wal_release(
    pending_tails: &[PendingTail],
    node_id: Option<u64>,
    entry_id: EntryId,
) -> EntryId {
    pending_tails
        .iter()
        .filter(|tail| Some(tail.node_id) == node_id)
        .map(|tail| tail.after)
        .fold(entry_id, EntryId::min)
}

impl RegionEdit {
    /// Shared reclassification must be add-only for a file ID: the manifest removes
    /// by ID, while the in-memory Version removes from the specified level.
    pub(crate) fn validate_shared_reclassification(&self, region_id: RegionId) -> Result<()> {
        let shared_ids: HashSet<_> = self
            .files_to_add
            .iter()
            .filter(|file| file.level == file::LEVEL_SHARED)
            .map(|file| file.file_id)
            .collect();
        snafu::ensure!(
            shared_ids.is_empty()
                || self
                    .files_to_remove
                    .iter()
                    .all(|file| !shared_ids.contains(&file.file_id)),
            crate::error::InvalidRequestSnafu {
                region_id,
                reason: "Cannot add and remove the same shared SST in one edit",
            }
        );
        Ok(())
    }
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct RegionRemove {
    pub region_id: RegionId,
}

/// Last data truncated in the region.
///
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct RegionTruncate {
    pub region_id: RegionId,
    #[serde(flatten)]
    pub kind: TruncateKind,
    /// event unix timestamp in milliseconds, help to determine file deletion time.
    #[serde(default)]
    pub timestamp_ms: Option<i64>,
}

/// The kind of truncate operation.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
#[serde(untagged)]
pub enum TruncateKind {
    /// Truncate all data in the region, marked by all data before the given entry id&sequence.
    All {
        /// Last WAL entry id of truncated data.
        truncated_entry_id: EntryId,
        // Last sequence number of truncated data.
        truncated_sequence: SequenceNumber,
    },
    /// Only remove certain files in the region.
    Partial { files_to_remove: Vec<FileMeta> },
}

/// The region manifest data.
#[derive(Serialize, Deserialize, Clone, Debug)]
#[cfg_attr(test, derive(Eq))]
pub struct RegionManifest {
    /// Metadata of the region.
    pub metadata: RegionMetadataRef,
    /// SST files.
    pub files: HashMap<FileId, FileMetaRef>,
    /// Removed files, which are not in the current manifest but may still be kept for a while.
    /// This is a list of (set of files, timestamp) pairs, where the timestamp is the time when
    /// the files are removed from manifest. The timestamp is in milliseconds since unix epoch.
    ///
    /// Using same checkpoint files and action files, the recovered manifest may differ in this
    /// `removed_files` field, because the checkpointer may evict some removed files using
    /// current machine time. This is acceptable because the removed files are not used in normal
    /// read/write path.
    ///
    #[serde(default)]
    pub removed_files: RemovedFilesRecord,
    /// Last WAL entry id of flushed data.
    pub flushed_entry_id: EntryId,
    /// Last sequence of flushed data.
    pub flushed_sequence: SequenceNumber,
    pub committed_sequence: Option<SequenceNumber>,
    /// Current manifest version.
    pub manifest_version: ManifestVersion,
    /// Last WAL entry id of truncated data.
    pub truncated_entry_id: Option<EntryId>,
    /// Inferred compaction time window.
    #[serde(with = "humantime_serde")]
    pub compaction_time_window: Option<Duration>,
    /// Format of the SST file.
    #[serde(default)]
    pub sst_format: FormatType,
    /// Whether the region is in append mode.
    #[serde(default)]
    pub append_mode: Option<bool>,
    /// WAL ranges the region still has to recover, in the order they were added.
    ///
    /// The segments of one node are increasing and do not overlap, and each has
    /// `after < upto`: a new segment of a node takes as `after` the largest of the
    /// flushed entry id, the `upto`s of the node's existing segments and its handoff
    /// floor, and is added only if that is below its `upto`. So the entries of a node
    /// between two of its segments are flushed or at or below a handoff floor of the
    /// node, and need no replay.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub pending_tails: Vec<PendingTail>,
    /// The [HandoffFloor] of each node, removed once the flushed entry id reaches it.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub handoff_floors: BTreeMap<u64, EntryId>,
}

impl RegionManifest {
    /// Returns the entry id after which node `node_id` replays its WAL when it opens
    /// the region. Entries up to it are flushed, left to the pending tails, or below
    /// the handoff floor of the node.
    ///
    /// Entries of the node in `(after, upto]` of one of its tails belong to that tail.
    /// Whoever adds a tail ensures that every entry id the node allocates afterwards is
    /// above the tail's `upto`, so no later write of the node falls in a skipped range.
    pub(crate) fn wal_replay_start(&self, node_id: Option<u64>) -> EntryId {
        let floor = node_id.and_then(|node_id| self.handoff_floors.get(&node_id).copied());
        self.pending_tails
            .iter()
            .filter(|tail| Some(tail.node_id) == node_id)
            .map(|tail| tail.upto)
            .chain(floor)
            .fold(self.flushed_entry_id, EntryId::max)
    }
}

#[cfg(test)]
impl PartialEq for RegionManifest {
    fn eq(&self, other: &Self) -> bool {
        self.metadata == other.metadata
            && self.files == other.files
            && self.flushed_entry_id == other.flushed_entry_id
            && self.flushed_sequence == other.flushed_sequence
            && self.manifest_version == other.manifest_version
            && self.truncated_entry_id == other.truncated_entry_id
            && self.compaction_time_window == other.compaction_time_window
            && self.committed_sequence == other.committed_sequence
            && self.pending_tails == other.pending_tails
            && self.handoff_floors == other.handoff_floors
    }
}

impl RegionManifest {
    /// Rejects prototype task state before deserializing, so callers cannot
    /// silently drop recovery evidence that this build cannot interpret.
    pub(crate) fn decode(bytes: &[u8]) -> Result<Self> {
        crate::manifest::compat::validate_manifest(bytes)?;
        serde_json::from_slice(bytes).context(SerdeJsonSnafu)
    }
}

#[derive(Debug, Default)]
pub struct RegionManifestBuilder {
    file_metadata_interner: FileMetadataInternerRef,
    metadata: Option<RegionMetadataRef>,
    files: HashMap<FileId, FileMetaRef>,
    pub removed_files: RemovedFilesRecord,
    flushed_entry_id: EntryId,
    flushed_sequence: SequenceNumber,
    manifest_version: ManifestVersion,
    truncated_entry_id: Option<EntryId>,
    compaction_time_window: Option<Duration>,
    committed_sequence: Option<SequenceNumber>,
    sst_format: FormatType,
    append_mode: Option<bool>,
    pending_tails: Vec<PendingTail>,
    handoff_floors: BTreeMap<u64, EntryId>,
}

impl RegionManifestBuilder {
    /// Creates an empty builder with the owner's metadata interner.
    pub(crate) fn new(file_metadata_interner: FileMetadataInternerRef) -> Self {
        Self {
            file_metadata_interner,
            ..Default::default()
        }
    }

    /// Canonicalizes independently decoded checkpoint entries before publication.
    pub(crate) fn with_checkpoint(
        checkpoint: Option<RegionManifest>,
        file_metadata_interner: FileMetadataInternerRef,
    ) -> Self {
        let mut builder = Self::from_shared_manifest(checkpoint, file_metadata_interner);
        for meta in builder.files.values_mut() {
            builder.file_metadata_interner.intern_in_place(meta);
        }
        builder
    }

    fn removed_file(&self, file: &FileMeta) -> RemovedFile {
        let index_version = self
            .files
            .get(&file.file_id)
            .and_then(|meta| meta.index_version())
            .or_else(|| file.index_version());
        RemovedFile::File(file.file_id, index_version)
    }

    /// Reuses entries from a snapshot already canonicalized by the same owner.
    pub(crate) fn from_shared_manifest(
        checkpoint: Option<RegionManifest>,
        file_metadata_interner: FileMetadataInternerRef,
    ) -> Self {
        if let Some(s) = checkpoint {
            Self {
                file_metadata_interner,
                metadata: Some(s.metadata),
                files: s.files,
                removed_files: s.removed_files,
                flushed_entry_id: s.flushed_entry_id,
                manifest_version: s.manifest_version,
                flushed_sequence: s.flushed_sequence,
                truncated_entry_id: s.truncated_entry_id,
                compaction_time_window: s.compaction_time_window,
                committed_sequence: s.committed_sequence,
                sst_format: s.sst_format,
                append_mode: s.append_mode,
                pending_tails: s.pending_tails,
                handoff_floors: s.handoff_floors,
            }
        } else {
            Self::new(file_metadata_interner)
        }
    }

    pub fn apply_change(&mut self, manifest_version: ManifestVersion, change: RegionChange) {
        self.metadata = Some(change.metadata);
        self.manifest_version = manifest_version;
        self.sst_format = change.sst_format;
        // Only update append_mode if the change specifies a value.
        self.append_mode = change.append_mode.or(self.append_mode);
    }

    /// Applies a partition-expression-only metadata change.
    ///
    /// This path updates only `partition_expr` (and its derived
    /// `partition_rule_version`) on current metadata and does not touch
    /// `sst_format`.
    pub fn apply_partition_expr_change(
        &mut self,
        manifest_version: ManifestVersion,
        change: RegionPartitionExprChange,
    ) {
        if let Some(metadata) = &self.metadata {
            let mut metadata = metadata.as_ref().clone();
            metadata.set_partition_expr(change.partition_expr);
            self.metadata = Some(metadata.into());
            self.manifest_version = manifest_version;
        } else {
            warn!(
                "metadata is not set in region manifest builder, ignore partition expr change: {:?}",
                change
            );
        }
    }

    pub fn apply_edit(&mut self, manifest_version: ManifestVersion, edit: RegionEdit) {
        self.manifest_version = manifest_version;

        let mut removed_files = vec![];
        for file in edit.files_to_add {
            let file = FileMetaRef::from(file);
            let file = self.file_metadata_interner.intern(file);
            if let Some(old_file) = self.files.insert(file.file_id, file.clone())
                && let Some(old_index) = old_file.index_version()
                && !old_file.is_index_up_to_date(&file)
            {
                // The old file has an index that is now outdated.
                removed_files.push(RemovedFile::Index(old_file.file_id, old_index));
            }
        }
        removed_files.extend(
            edit.files_to_remove
                .iter()
                .map(|file| self.removed_file(file)),
        );
        let at = edit
            .timestamp_ms
            .unwrap_or_else(|| Utc::now().timestamp_millis());
        self.removed_files.add_removed_files(removed_files, at);

        for file in edit.files_to_remove {
            self.files.remove(&file.file_id);
        }
        if let Some(flushed_entry_id) = edit.flushed_entry_id {
            self.flushed_entry_id = self.flushed_entry_id.max(flushed_entry_id);
        }
        if let Some(flushed_sequence) = edit.flushed_sequence {
            self.flushed_sequence = self.flushed_sequence.max(flushed_sequence);
        }

        if let Some(committed_sequence) = edit.committed_sequence {
            self.committed_sequence = Some(
                self.committed_sequence
                    .map_or(committed_sequence, |exist| exist.max(committed_sequence)),
            );
        }
        if let Some(window) = edit.compaction_time_window {
            self.compaction_time_window = Some(window);
        }
        if let Some(pending_tails) = edit.wal_recovery.pending_tails {
            self.pending_tails = pending_tails;
        }
        if let Some(floor) = edit.wal_recovery.handoff_floor {
            self.handoff_floors.insert(floor.node_id, floor.entry_id);
        }
    }

    pub fn apply_truncate(&mut self, manifest_version: ManifestVersion, truncate: RegionTruncate) {
        self.manifest_version = manifest_version;
        match truncate.kind {
            TruncateKind::All {
                truncated_entry_id,
                truncated_sequence,
            } => {
                self.flushed_entry_id = truncated_entry_id;
                self.flushed_sequence = truncated_sequence;
                self.truncated_entry_id = Some(truncated_entry_id);
                self.removed_files.add_removed_files(
                    self.files
                        .values()
                        .map(|f| RemovedFile::File(f.file_id, f.index_version()))
                        .collect(),
                    truncate
                        .timestamp_ms
                        .unwrap_or_else(|| Utc::now().timestamp_millis()),
                );
                self.files.clear();
            }
            TruncateKind::Partial { files_to_remove } => {
                // With GC disabled, VersionControl may still hold an older
                // FileMeta and LocalFilePurger can miss a just-committed newer
                // index generation. We accept this narrow local-mode orphan
                // window; object-store deployments enable GC and collect the
                // generation recorded here.
                self.removed_files.add_removed_files(
                    files_to_remove
                        .iter()
                        .map(|file| self.removed_file(file))
                        .collect(),
                    truncate
                        .timestamp_ms
                        .unwrap_or_else(|| Utc::now().timestamp_millis()),
                );
                for file in files_to_remove {
                    self.files.remove(&file.file_id);
                }
            }
        }
    }

    pub fn files(&self) -> &HashMap<FileId, FileMetaRef> {
        &self.files
    }

    /// Check if the builder keeps a [RegionMetadata](store_api::metadata::RegionMetadata).
    pub fn contains_metadata(&self) -> bool {
        self.metadata.is_some()
    }

    pub fn try_build(mut self) -> Result<RegionManifest> {
        let metadata = self.metadata.context(RegionMetadataNotFoundSnafu)?;
        // A floor at or below the flushed entry id no longer raises the replay start.
        self.handoff_floors
            .retain(|_, floor| *floor > self.flushed_entry_id);
        Ok(RegionManifest {
            metadata,
            files: self.files,
            removed_files: self.removed_files,
            flushed_entry_id: self.flushed_entry_id,
            flushed_sequence: self.flushed_sequence,
            committed_sequence: self.committed_sequence,
            manifest_version: self.manifest_version,
            truncated_entry_id: self.truncated_entry_id,
            compaction_time_window: self.compaction_time_window,
            sst_format: self.sst_format,
            append_mode: self.append_mode,
            pending_tails: self.pending_tails,
            handoff_floors: self.handoff_floors,
        })
    }
}

/// A record of removed files in the region manifest.
/// This is used to keep track of files that have been removed from the manifest but may still
/// be kept for a while
#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)]
pub struct RemovedFilesRecord {
    /// a list of `(FileIds, timestamp)` pairs, where the timestamp is the time when
    /// the files are removed from manifest. The timestamp is in milliseconds since unix epoch.
    pub removed_files: Vec<RemovedFiles>,
}

impl RemovedFilesRecord {
    /// Clear the actually deleted files from the list of removed files
    pub fn clear_deleted_files(&mut self, deleted_files: Vec<RemovedFile>) {
        let deleted_file_set: HashSet<_> = HashSet::from_iter(deleted_files);
        for files in self.removed_files.iter_mut() {
            files
                .files
                .retain(|removed| !deleted_file_set.contains(removed));
        }

        self.removed_files.retain(|fs| !fs.files.is_empty());
    }

    pub fn update_file_removed_cnt_to_stats(&self, stats: &ManifestStats) {
        let cnt = self
            .removed_files
            .iter()
            .map(|r| r.files.len() as u64)
            .sum();
        stats
            .file_removed_cnt
            .store(cnt, std::sync::atomic::Ordering::Relaxed);
    }
}

#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)]
pub struct RemovedFiles {
    /// The timestamp is the time when
    /// the files are removed from manifest. The timestamp is in milliseconds since unix epoch.
    pub removed_at: i64,
    /// The set of file ids that are removed.
    #[serde(default)]
    pub files: HashSet<RemovedFile>,
}

/// A removed file, which can be a data file(optional paired with a index file) or an outdated index file.
#[derive(Serialize, Hash, Clone, Debug, PartialEq, Eq)]
pub enum RemovedFile {
    File(FileId, Option<IndexVersion>),
    Index(FileId, IndexVersion),
}

/// Support deserialize from old format(just FileId as string) for backward compatibility
/// into current format(RemovedFile enum).
/// This is needed just in case there are old manifests with removed files recorded.
impl<'de> Deserialize<'de> for RemovedFile {
    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        #[derive(Deserialize)]
        #[serde(untagged)]
        enum CompatRemovedFile {
            Enum(RemovedFileEnum),
            FileId(FileId),
        }

        #[derive(Deserialize)]
        enum RemovedFileEnum {
            File(FileId, Option<IndexVersion>),
            Index(FileId, IndexVersion),
        }

        let compat = CompatRemovedFile::deserialize(deserializer)?;
        match compat {
            CompatRemovedFile::FileId(file_id) => Ok(RemovedFile::File(file_id, None)),
            CompatRemovedFile::Enum(e) => match e {
                RemovedFileEnum::File(file_id, version) => Ok(RemovedFile::File(file_id, version)),
                RemovedFileEnum::Index(file_id, version) => {
                    Ok(RemovedFile::Index(file_id, version))
                }
            },
        }
    }
}

impl RemovedFile {
    pub fn file_id(&self) -> FileId {
        match self {
            RemovedFile::File(file_id, _) => *file_id,
            RemovedFile::Index(file_id, _) => *file_id,
        }
    }

    pub fn index_version(&self) -> Option<IndexVersion> {
        match self {
            RemovedFile::File(_, index_version) => *index_version,
            RemovedFile::Index(_, index_version) => Some(*index_version),
        }
    }
}

impl RemovedFilesRecord {
    /// Add a record of removed files with the current timestamp.
    pub fn add_removed_files(&mut self, removed: Vec<RemovedFile>, at: i64) {
        if removed.is_empty() {
            return;
        }
        let files = removed.into_iter().collect();
        self.removed_files.push(RemovedFiles {
            removed_at: at,
            files,
        });
    }

    pub fn evict_old_removed_files(&mut self, opt: &RemoveFileOptions) -> Result<()> {
        if !opt.enable_gc {
            // If GC is not enabled, always keep removed files empty.
            self.removed_files.clear();
            return Ok(());
        }

        // if GC is enabled, rely on gc worker to delete files, and evict removed files based on options.

        Ok(())
    }
}

// The checkpoint of region manifest, generated by checkpointer.
#[derive(Serialize, Deserialize, Debug, Clone)]
#[cfg_attr(test, derive(PartialEq, Eq))]
pub struct RegionCheckpoint {
    /// The last manifest version that this checkpoint compacts(inclusive).
    pub last_version: ManifestVersion,
    // The number of manifest actions that this checkpoint compacts.
    pub compacted_actions: usize,
    // The checkpoint data
    pub checkpoint: Option<RegionManifest>,
}

impl RegionCheckpoint {
    pub fn last_version(&self) -> ManifestVersion {
        self.last_version
    }

    pub fn encode(&self) -> Result<Vec<u8>> {
        let json = serde_json::to_string(&self).context(SerdeJsonSnafu)?;

        Ok(json.into_bytes())
    }

    pub fn decode(bytes: &[u8]) -> Result<Self> {
        let data = std::str::from_utf8(bytes).context(Utf8Snafu)?;
        crate::manifest::compat::validate_checkpoint(data)?;
        serde_json::from_str(data).context(SerdeJsonSnafu)
    }
}

#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq)]
pub struct RegionMetaActionList {
    pub actions: Vec<RegionMetaAction>,
}

impl RegionMetaActionList {
    pub fn with_action(action: RegionMetaAction) -> Self {
        Self {
            actions: vec![action],
        }
    }

    pub fn new(actions: Vec<RegionMetaAction>) -> Self {
        Self { actions }
    }

    /// Split the actions into a partition expr change, a region change and an edit.
    pub fn split_region_change_and_edit(
        self,
    ) -> (
        Option<RegionPartitionExprChange>,
        Option<RegionChange>,
        RegionEdit,
    ) {
        let mut edit = RegionEdit {
            files_to_add: Vec::new(),
            files_to_remove: Vec::new(),
            timestamp_ms: None,
            compaction_time_window: None,
            flushed_entry_id: None,
            flushed_sequence: None,
            committed_sequence: None,
            wal_recovery: Default::default(),
        };
        let mut partition_expr_change = None;
        let mut region_change = None;
        for action in self.actions {
            match action {
                RegionMetaAction::PartitionExprChange(change) => {
                    partition_expr_change = Some(change);
                }
                RegionMetaAction::Change(change) => {
                    region_change = Some(change);
                }
                RegionMetaAction::Edit(region_edit) => {
                    // Merge file adds/removes
                    edit.files_to_add.extend(region_edit.files_to_add);
                    edit.files_to_remove.extend(region_edit.files_to_remove);
                    // Max of flushed entry id / sequence
                    if let Some(eid) = region_edit.flushed_entry_id {
                        edit.flushed_entry_id =
                            Some(edit.flushed_entry_id.map_or(eid, |v| v.max(eid)));
                    }
                    if let Some(seq) = region_edit.flushed_sequence {
                        edit.flushed_sequence =
                            Some(edit.flushed_sequence.map_or(seq, |v| v.max(seq)));
                    }
                    if let Some(seq) = region_edit.committed_sequence {
                        edit.committed_sequence =
                            Some(edit.committed_sequence.map_or(seq, |v| v.max(seq)));
                    }
                    // Prefer the latest non-none time window
                    if region_edit.compaction_time_window.is_some() {
                        edit.compaction_time_window = region_edit.compaction_time_window;
                    }
                }
                _ => {}
            }
        }

        (partition_expr_change, region_change, edit)
    }
}

impl RegionMetaActionList {
    /// Encode self into json in the form of string lines.
    pub fn encode(&self) -> Result<Vec<u8>> {
        let json = serde_json::to_string(&self).context(SerdeJsonSnafu)?;

        Ok(json.into_bytes())
    }

    pub fn decode(bytes: &[u8]) -> Result<Self> {
        let data = std::str::from_utf8(bytes).context(Utf8Snafu)?;
        crate::manifest::compat::validate_actions(data)?;
        serde_json::from_str(data).context(SerdeJsonSnafu)
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use common_time::Timestamp;

    use crate::sst::file_metadata::MetadataRef;

    use super::*;

    // These tests are used to ensure backward compatibility of manifest files.
    // DO NOT modify the serialized string when they fail, check if your
    // modification to manifest-related structs is compatible with older manifests.
    #[test]
    fn test_sketch_index_manifest_fixture() {
        use crate::sst::file::IndexType;

        // Frozen persisted names. Keep this fixture unchanged when extending the format.
        let fixture = r#"{"actions":[{"Edit":{"files_to_add":[{
                "region_id":0,"file_id":"bc5896ec-e4d8-4017-a80d-f2de73188d55",
                "time_range":[{"value":0,"unit":"Millisecond"},{"value":1,"unit":"Millisecond"}],
                "level":0,"index_version":7,"index_file_size":128,
                "available_indexes":["InvertedIndex","HllIndex","KllIndex","MgIndex"],
                "indexes":[{"column_id":3,"created_indexes":["InvertedIndex","HllIndex","KllIndex","MgIndex"]}]
            }],"files_to_remove":[],"compaction_time_window":null}}]}"#;
        let actions = RegionMetaActionList::decode(fixture.as_bytes()).unwrap();
        let RegionMetaAction::Edit(edit) = &actions.actions[0] else {
            panic!("expected a file edit");
        };
        assert_eq!(actions.actions.len(), 1);
        assert_eq!(edit.files_to_add.len(), 1);
        let file = &edit.files_to_add[0];
        let expected = [
            IndexType::InvertedIndex,
            IndexType::HllIndex,
            IndexType::KllIndex,
            IndexType::MgIndex,
        ];
        assert_eq!(file.available_indexes.as_slice(), expected);
        assert_eq!(file.indexes.len(), 1);
        assert_eq!(file.indexes[0].column_id, 3);
        assert_eq!(file.indexes[0].created_indexes.as_slice(), expected);
        assert_eq!(file.index_version(), Some(7));
        assert_eq!(file.index_file_size, 128);
        let encoded = actions.encode().unwrap();
        assert_eq!(RegionMetaActionList::decode(&encoded).unwrap(), actions);
        let encoded: serde_json::Value = serde_json::from_slice(&encoded).unwrap();
        let names = serde_json::json!(["InvertedIndex", "HllIndex", "KllIndex", "MgIndex"]);
        assert_eq!(
            encoded["actions"][0]["Edit"]["files_to_add"][0]["available_indexes"],
            names
        );
        assert_eq!(
            encoded["actions"][0]["Edit"]["files_to_add"][0]["indexes"][0]["created_indexes"],
            names
        );
    }

    #[test]
    fn test_region_action_compatibility() {
        let region_edit = r#"{
            "flushed_entry_id":null,
            "compaction_time_window":null,
            "files_to_add":[
            {"region_id":4402341478400,"file_id":"4b220a70-2b03-4641-9687-b65d94641208","time_range":[{"value":1451609210000,"unit":"Millisecond"},{"value":1451609520000,"unit":"Millisecond"}],"level":1,"file_size":100}
            ],
            "files_to_remove":[
            {"region_id":4402341478400,"file_id":"34b6ebb9-b8a5-4a4b-b744-56f67defad02","time_range":[{"value":1451609210000,"unit":"Millisecond"},{"value":1451609520000,"unit":"Millisecond"}],"level":0,"file_size":100}
            ]
        }"#;
        let _ = serde_json::from_str::<RegionEdit>(region_edit).unwrap();

        let region_edit = r#"{
            "flushed_entry_id":10,
            "flushed_sequence":10,
            "compaction_time_window":null,
            "files_to_add":[
            {"region_id":4402341478400,"file_id":"4b220a70-2b03-4641-9687-b65d94641208","time_range":[{"value":1451609210000,"unit":"Millisecond"},{"value":1451609520000,"unit":"Millisecond"}],"level":1,"file_size":100}
            ],
            "files_to_remove":[
            {"region_id":4402341478400,"file_id":"34b6ebb9-b8a5-4a4b-b744-56f67defad02","time_range":[{"value":1451609210000,"unit":"Millisecond"},{"value":1451609520000,"unit":"Millisecond"}],"level":0,"file_size":100}
            ]
        }"#;
        let _ = serde_json::from_str::<RegionEdit>(region_edit).unwrap();

        // Note: For backward compatibility, the test accepts a RegionChange without sst_format
        let region_change = r#" {
            "metadata":{
                "column_metadatas":[
                {"column_schema":{"name":"a","data_type":{"Int64":{}},"is_nullable":false,"is_time_index":false,"default_constraint":null,"metadata":{}},"semantic_type":"Tag","column_id":1},{"column_schema":{"name":"b","data_type":{"Float64":{}},"is_nullable":false,"is_time_index":false,"default_constraint":null,"metadata":{}},"semantic_type":"Field","column_id":2},{"column_schema":{"name":"c","data_type":{"Timestamp":{"Millisecond":null}},"is_nullable":false,"is_time_index":false,"default_constraint":null,"metadata":{}},"semantic_type":"Timestamp","column_id":3}
                ],
                "primary_key":[1],
                "region_id":5299989648942,
                "schema_version":0
            }
            }"#;
        let _ = serde_json::from_str::<RegionChange>(region_change).unwrap();

        let region_remove = r#"{"region_id":42}"#;
        let _ = serde_json::from_str::<RegionRemove>(region_remove).unwrap();

        let region_partition_expr_change = r#"{
            "partition_expr": "{\"expr\":\"x < 100\"}"
        }"#;
        let _ = serde_json::from_str::<RegionPartitionExprChange>(region_partition_expr_change)
            .unwrap();
    }

    #[test]
    fn test_region_manifest_compatibility() {
        // Test deserializing RegionManifest from old schema where FileId is a UUID string
        let region_manifest_json = r#"{
            "metadata": {
                "column_metadatas": [
                    {
                        "column_schema": {
                            "name": "a",
                            "data_type": {"Int64": {}},
                            "is_nullable": false,
                            "is_time_index": false,
                            "default_constraint": null,
                            "metadata": {}
                        },
                        "semantic_type": "Tag",
                        "column_id": 1
                    },
                    {
                        "column_schema": {
                            "name": "b",
                            "data_type": {"Float64": {}},
                            "is_nullable": false,
                            "is_time_index": false,
                            "default_constraint": null,
                            "metadata": {}
                        },
                        "semantic_type": "Field",
                        "column_id": 2
                    },
                    {
                        "column_schema": {
                            "name": "c",
                            "data_type": {"Timestamp": {"Millisecond": null}},
                            "is_nullable": false,
                            "is_time_index": false,
                            "default_constraint": null,
                            "metadata": {}
                        },
                        "semantic_type": "Timestamp",
                        "column_id": 3
                    }
                ],
                "primary_key": [1],
                "region_id": 4402341478400,
                "schema_version": 0
            },
            "files": {
                "4b220a70-2b03-4641-9687-b65d94641208": {
                    "region_id": 4402341478400,
                    "file_id": "4b220a70-2b03-4641-9687-b65d94641208",
                    "time_range": [
                        {"value": 1451609210000, "unit": "Millisecond"},
                        {"value": 1451609520000, "unit": "Millisecond"}
                    ],
                    "level": 1,
                    "file_size": 100
                },
                "34b6ebb9-b8a5-4a4b-b744-56f67defad02": {
                    "region_id": 4402341478400,
                    "file_id": "34b6ebb9-b8a5-4a4b-b744-56f67defad02",
                    "time_range": [
                        {"value": 1451609210000, "unit": "Millisecond"},
                        {"value": 1451609520000, "unit": "Millisecond"}
                    ],
                    "level": 0,
                    "file_size": 100
                }
            },
            "flushed_entry_id": 10,
            "flushed_sequence": 20,
            "manifest_version": 1,
            "truncated_entry_id": null,
            "compaction_time_window": null
        }"#;

        let manifest = serde_json::from_str::<RegionManifest>(region_manifest_json).unwrap();

        // Verify that the files were correctly deserialized
        assert_eq!(manifest.files.len(), 2);
        assert_eq!(manifest.flushed_entry_id, 10);
        assert_eq!(manifest.flushed_sequence, 20);
        assert_eq!(manifest.manifest_version, 1);

        // Verify that FileIds were correctly parsed from UUID strings
        let mut file_ids: Vec<String> = manifest.files.keys().map(|id| id.to_string()).collect();
        file_ids.sort_unstable();
        assert_eq!(
            file_ids,
            vec![
                "34b6ebb9-b8a5-4a4b-b744-56f67defad02",
                "4b220a70-2b03-4641-9687-b65d94641208",
            ]
        );

        // Roundtrip test with current FileId format
        let serialized_manifest = serde_json::to_string(&manifest).unwrap();
        let deserialized_manifest: RegionManifest =
            serde_json::from_str(&serialized_manifest).unwrap();
        assert_eq!(manifest, deserialized_manifest);
        assert_ne!(serialized_manifest, region_manifest_json);
    }

    #[test]
    fn test_region_truncate_compat() {
        // Test deserializing RegionTruncate from old schema
        let region_truncate_json = r#"{
            "region_id": 4402341478400,
            "truncated_entry_id": 10,
            "truncated_sequence": 20
        }"#;

        let truncate_v1: RegionTruncate = serde_json::from_str(region_truncate_json).unwrap();
        assert_eq!(truncate_v1.region_id, 4402341478400);
        assert_eq!(
            truncate_v1.kind,
            TruncateKind::All {
                truncated_entry_id: 10,
                truncated_sequence: 20,
            }
        );

        // Test deserializing RegionTruncate from new schema
        let region_truncate_v2_json = r#"{
    "region_id": 4402341478400,
    "files_to_remove": [
        {
            "region_id": 4402341478400,
            "file_id": "4b220a70-2b03-4641-9687-b65d94641208",
            "time_range": [
                {
                    "value": 1451609210000,
                    "unit": "Millisecond"
                },
                {
                    "value": 1451609520000,
                    "unit": "Millisecond"
                }
            ],
            "level": 1,
            "file_size": 100
        }
    ]
}"#;

        let truncate_v2: RegionTruncate = serde_json::from_str(region_truncate_v2_json).unwrap();
        assert_eq!(truncate_v2.region_id, 4402341478400);
        assert_eq!(
            truncate_v2.kind,
            TruncateKind::Partial {
                files_to_remove: vec![FileMeta {
                    region_id: RegionId::from_u64(4402341478400),
                    file_id: FileId::parse_str("4b220a70-2b03-4641-9687-b65d94641208").unwrap(),
                    time_range: (
                        Timestamp::new_millisecond(1451609210000),
                        Timestamp::new_millisecond(1451609520000)
                    ),
                    level: 1,
                    file_size: 100,
                    ..Default::default()
                }]
            }
        );
    }

    #[test]
    fn test_region_manifest_removed_files() {
        let region_metadata = r#"{
                "column_metadatas": [
                    {
                        "column_schema": {
                            "name": "a",
                            "data_type": {"Int64": {}},
                            "is_nullable": false,
                            "is_time_index": false,
                            "default_constraint": null,
                            "metadata": {}
                        },
                        "semantic_type": "Tag",
                        "column_id": 1
                    },
                    {
                        "column_schema": {
                            "name": "b",
                            "data_type": {"Float64": {}},
                            "is_nullable": false,
                            "is_time_index": false,
                            "default_constraint": null,
                            "metadata": {}
                        },
                        "semantic_type": "Field",
                        "column_id": 2
                    },
                    {
                        "column_schema": {
                            "name": "c",
                            "data_type": {"Timestamp": {"Millisecond": null}},
                            "is_nullable": false,
                            "is_time_index": false,
                            "default_constraint": null,
                            "metadata": {}
                        },
                        "semantic_type": "Timestamp",
                        "column_id": 3
                    }
                ],
                "primary_key": [1],
                "region_id": 4402341478400,
                "schema_version": 0
            }"#;

        let metadata: RegionMetadataRef =
            serde_json::from_str(region_metadata).expect("Failed to parse region metadata");
        let manifest = RegionManifest {
            metadata: metadata.clone(),
            files: HashMap::new(),
            flushed_entry_id: 0,
            flushed_sequence: 0,
            committed_sequence: None,
            manifest_version: 0,
            truncated_entry_id: None,
            compaction_time_window: None,
            removed_files: RemovedFilesRecord {
                removed_files: vec![RemovedFiles {
                    removed_at: 0,
                    files: HashSet::from([RemovedFile::File(
                        FileId::parse_str("4b220a70-2b03-4641-9687-b65d94641208").unwrap(),
                        None,
                    )]),
                }],
            },
            sst_format: FormatType::PrimaryKey,
            append_mode: None,
            pending_tails: Default::default(),
            handoff_floors: Default::default(),
        };

        let json = serde_json::to_string(&manifest).unwrap();
        let new: RegionManifest = serde_json::from_str(&json).unwrap();

        assert_eq!(manifest, new);
    }

    #[test]
    fn test_remove_tracks_current_manifest_index_version() {
        let file_id = FileId::random();
        let current = FileMeta {
            region_id: RegionId::new(1, 1),
            file_id,
            available_indexes: smallvec::smallvec![crate::sst::file::IndexType::InvertedIndex],
            index_file_size: 1024,
            index_version: 2,
            ..Default::default()
        };
        let mut stale = current.clone();
        stale.available_indexes.clear();
        stale.index_file_size = 0;
        stale.index_version = 0;

        let mut builder = RegionManifestBuilder::default();
        builder.apply_edit(
            1,
            RegionEdit {
                files_to_add: vec![current.clone()],
                files_to_remove: Vec::new(),
                timestamp_ms: None,
                compaction_time_window: None,
                flushed_entry_id: None,
                flushed_sequence: None,
                committed_sequence: None,
                wal_recovery: Default::default(),
            },
        );
        builder.apply_edit(
            2,
            RegionEdit {
                files_to_add: Vec::new(),
                files_to_remove: vec![stale.clone()],
                timestamp_ms: Some(42),
                compaction_time_window: None,
                flushed_entry_id: None,
                flushed_sequence: None,
                committed_sequence: None,
                wal_recovery: Default::default(),
            },
        );

        assert_eq!(
            builder.removed_files.removed_files[0].files,
            HashSet::from([RemovedFile::File(file_id, Some(2))])
        );

        let mut builder = RegionManifestBuilder::default();
        builder.apply_edit(
            1,
            RegionEdit {
                files_to_add: vec![current],
                files_to_remove: Vec::new(),
                timestamp_ms: None,
                compaction_time_window: None,
                flushed_entry_id: None,
                flushed_sequence: None,
                committed_sequence: None,
                wal_recovery: Default::default(),
            },
        );
        builder.apply_truncate(
            2,
            RegionTruncate {
                region_id: RegionId::new(1, 1),
                kind: TruncateKind::Partial {
                    files_to_remove: vec![stale],
                },
                timestamp_ms: Some(42),
            },
        );

        assert_eq!(
            builder.removed_files.removed_files[0].files,
            HashSet::from([RemovedFile::File(file_id, Some(2))])
        );
    }

    /// Test if old version can still be deserialized then serialized to the new version.
    #[test]
    fn test_old_region_manifest_compat() {
        #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
        pub struct RegionManifestV1 {
            /// Metadata of the region.
            pub metadata: RegionMetadataRef,
            /// SST files.
            pub files: HashMap<FileId, FileMeta>,
            /// Last WAL entry id of flushed data.
            pub flushed_entry_id: EntryId,
            /// Last sequence of flushed data.
            pub flushed_sequence: SequenceNumber,
            /// Current manifest version.
            pub manifest_version: ManifestVersion,
            /// Last WAL entry id of truncated data.
            pub truncated_entry_id: Option<EntryId>,
            /// Inferred compaction time window.
            #[serde(with = "humantime_serde")]
            pub compaction_time_window: Option<Duration>,
        }

        let region_metadata = r#"{
                "column_metadatas": [
                    {
                        "column_schema": {
                            "name": "a",
                            "data_type": {"Int64": {}},
                            "is_nullable": false,
                            "is_time_index": false,
                            "default_constraint": null,
                            "metadata": {}
                        },
                        "semantic_type": "Tag",
                        "column_id": 1
                    },
                    {
                        "column_schema": {
                            "name": "b",
                            "data_type": {"Float64": {}},
                            "is_nullable": false,
                            "is_time_index": false,
                            "default_constraint": null,
                            "metadata": {}
                        },
                        "semantic_type": "Field",
                        "column_id": 2
                    },
                    {
                        "column_schema": {
                            "name": "c",
                            "data_type": {"Timestamp": {"Millisecond": null}},
                            "is_nullable": false,
                            "is_time_index": false,
                            "default_constraint": null,
                            "metadata": {}
                        },
                        "semantic_type": "Timestamp",
                        "column_id": 3
                    }
                ],
                "primary_key": [1],
                "region_id": 4402341478400,
                "schema_version": 0
            }"#;

        let metadata: RegionMetadataRef =
            serde_json::from_str(region_metadata).expect("Failed to parse region metadata");

        // first test v1 empty to new
        let v1 = RegionManifestV1 {
            metadata: metadata.clone(),
            files: HashMap::new(),
            flushed_entry_id: 0,
            flushed_sequence: 0,
            manifest_version: 0,
            truncated_entry_id: None,
            compaction_time_window: None,
        };
        let json = serde_json::to_string(&v1).unwrap();
        let new_from_old: RegionManifest = serde_json::from_str(&json).unwrap();
        assert_eq!(
            new_from_old,
            RegionManifest {
                metadata: metadata.clone(),
                files: HashMap::new(),
                removed_files: Default::default(),
                flushed_entry_id: 0,
                flushed_sequence: 0,
                committed_sequence: None,
                manifest_version: 0,
                truncated_entry_id: None,
                compaction_time_window: None,
                sst_format: FormatType::PrimaryKey,
                append_mode: None,
                pending_tails: Default::default(),
                handoff_floors: Default::default(),
            }
        );

        let new_manifest = RegionManifest {
            metadata: metadata.clone(),
            files: HashMap::new(),
            removed_files: Default::default(),
            flushed_entry_id: 0,
            flushed_sequence: 0,
            committed_sequence: None,
            manifest_version: 0,
            truncated_entry_id: None,
            compaction_time_window: None,
            sst_format: FormatType::PrimaryKey,
            append_mode: None,
            pending_tails: Default::default(),
            handoff_floors: Default::default(),
        };
        let json = serde_json::to_string(&new_manifest).unwrap();
        let old_from_new: RegionManifestV1 = serde_json::from_str(&json).unwrap();
        assert_eq!(
            old_from_new,
            RegionManifestV1 {
                metadata: metadata.clone(),
                files: HashMap::new(),
                flushed_entry_id: 0,
                flushed_sequence: 0,
                manifest_version: 0,
                truncated_entry_id: None,
                compaction_time_window: None,
            }
        );

        #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
        pub struct RegionEditV1 {
            pub files_to_add: Vec<FileMeta>,
            pub files_to_remove: Vec<FileMeta>,
            #[serde(with = "humantime_serde")]
            pub compaction_time_window: Option<Duration>,
            pub flushed_entry_id: Option<EntryId>,
            pub flushed_sequence: Option<SequenceNumber>,
        }

        let json = serde_json::to_string(&RegionEditV1 {
            files_to_add: vec![],
            files_to_remove: vec![],
            compaction_time_window: None,
            flushed_entry_id: None,
            flushed_sequence: None,
        })
        .unwrap();
        let new_from_old: RegionEdit = serde_json::from_str(&json).unwrap();
        assert_eq!(
            RegionEdit {
                files_to_add: vec![],
                files_to_remove: vec![],
                timestamp_ms: None,
                compaction_time_window: None,
                flushed_entry_id: None,
                flushed_sequence: None,
                committed_sequence: None,
                wal_recovery: Default::default(),
            },
            new_from_old
        );

        // test new version with timestamp_ms set can deserialize to old version
        let new = RegionEdit {
            files_to_add: vec![],
            files_to_remove: vec![],
            timestamp_ms: Some(42),
            compaction_time_window: None,
            flushed_entry_id: None,
            flushed_sequence: None,
            committed_sequence: None,
            wal_recovery: Default::default(),
        };

        let new_json = serde_json::to_string(&new).unwrap();

        let old_from_new: RegionEditV1 = serde_json::from_str(&new_json).unwrap();
        assert_eq!(
            RegionEditV1 {
                files_to_add: vec![],
                files_to_remove: vec![],
                compaction_time_window: None,
                flushed_entry_id: None,
                flushed_sequence: None,
            },
            old_from_new
        );
    }

    #[test]
    fn test_region_change_backward_compatibility() {
        // Test that we can deserialize a RegionChange without sst_format
        let region_change_json = r#"{
            "metadata": {
                "column_metadatas": [
                    {"column_schema":{"name":"a","data_type":{"Int64":{}},"is_nullable":false,"is_time_index":false,"default_constraint":null,"metadata":{}},"semantic_type":"Tag","column_id":1},
                    {"column_schema":{"name":"b","data_type":{"Int64":{}},"is_nullable":false,"is_time_index":false,"default_constraint":null,"metadata":{}},"semantic_type":"Field","column_id":2},
                    {"column_schema":{"name":"c","data_type":{"Timestamp":{"Millisecond":null}},"is_nullable":false,"is_time_index":false,"default_constraint":null,"metadata":{}},"semantic_type":"Timestamp","column_id":3}
                ],
                "primary_key": [
                    1
                ],
                "region_id": 42,
                "schema_version": 0
            }
        }"#;

        let region_change: RegionChange = serde_json::from_str(region_change_json).unwrap();
        assert_eq!(region_change.sst_format, FormatType::PrimaryKey);

        // Test serialization and deserialization with sst_format
        let region_change = RegionChange {
            metadata: region_change.metadata.clone(),
            sst_format: FormatType::Flat,
            append_mode: None,
        };

        let serialized = serde_json::to_string(&region_change).unwrap();
        let deserialized: RegionChange = serde_json::from_str(&serialized).unwrap();
        assert_eq!(deserialized.sst_format, FormatType::Flat);
    }

    #[test]
    fn test_removed_file_compatibility() {
        let file_id = FileId::random();
        // Case 1: Deserialize from FileId string (Legacy format)
        let json_str = format!("\"{}\"", file_id);
        let removed_file: RemovedFile = serde_json::from_str(&json_str).unwrap();
        assert_eq!(removed_file, RemovedFile::File(file_id, None));

        // Case 2: Deserialize from new format (File)
        let removed_file_v2 = RemovedFile::File(file_id, Some(10));
        let json_v2 = serde_json::to_string(&removed_file_v2).unwrap();
        let deserialized_v2: RemovedFile = serde_json::from_str(&json_v2).unwrap();
        assert_eq!(removed_file_v2, deserialized_v2);

        // Case 3: Deserialize from new format (Index)
        let removed_index = RemovedFile::Index(file_id, 20);
        let json_index = serde_json::to_string(&removed_index).unwrap();
        let deserialized_index: RemovedFile = serde_json::from_str(&json_index).unwrap();
        assert_eq!(removed_index, deserialized_index);

        // Case 4: Round-trip serialization/deserialization of new enum format with None as index version
        let removed_file = RemovedFile::File(file_id, None);
        let json = serde_json::to_string(&removed_file).unwrap();
        let deserialized: RemovedFile = serde_json::from_str(&json).unwrap();
        assert_eq!(removed_file, deserialized);

        // Case 5: Deserialize mixed set in RemovedFilesRecord
        // This simulates a Set<RemovedFile> which might contain old strings or new objects if manually constructed or from old versions.
        // Actually, if it was HashSet<FileId>, the JSON is ["id1", "id2"].
        // If it is HashSet<RemovedFile>, the JSON is [{"File":...}, "id2"] if mixed (which shouldn't happen usually but good to test).

        let json_set = format!("[\"{}\"]", file_id);
        let removed_files_set: HashSet<RemovedFile> = serde_json::from_str(&json_set).unwrap();
        assert!(removed_files_set.contains(&RemovedFile::File(file_id, None)));
    }

    /// It is intentionally acceptable to ignore the legacy `file_ids` field when
    /// deserializing [`RemovedFiles`].
    ///
    /// In older manifests, `file_ids` recorded the set of SSTable files that were
    /// candidates for garbage collection at a given `removed_at` timestamp. The
    /// newer format stores this information in the `files` field instead. When we
    /// deserialize an old manifest entry into the new struct, we *drop* the
    /// `file_ids` field instead of trying to recover or merge it.
    ///
    /// Dropping `file_ids` does **not** risk deleting live data: a file is only
    /// physically removed when it is both (a) no longer referenced by any region
    /// metadata and (b) selected by the GC worker as safe to delete. Losing the
    /// historical list of candidate `file_ids` merely means some obsolete files
    /// may stay on disk longer than strictly necessary.
    ///
    /// The GC worker periodically scans storage (e.g. by walking the data
    /// directories and/or consulting the latest manifest) to discover files that
    /// are no longer referenced anywhere. Any files that were only referenced via
    /// the dropped `file_ids` field will be rediscovered during these scans and
    /// eventually deleted. Thus the system converges to a correct, fully-collected
    /// state without relying on `file_ids`, and the only potential impact of
    /// ignoring it is temporary disk space overhead, not data loss.
    #[test]
    fn test_removed_files_backward_compatibility() {
        // Define the old version struct with file_ids field
        #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
        struct OldRemovedFiles {
            pub removed_at: i64,
            pub file_ids: HashSet<FileId>,
        }

        // Create an old version instance
        let mut file_ids = HashSet::new();
        file_ids.insert(FileId::random());
        file_ids.insert(FileId::random());

        let old_removed_files = OldRemovedFiles {
            removed_at: 1234567890,
            file_ids,
        };

        // Serialize the old version
        let old_json = serde_json::to_string(&old_removed_files).unwrap();

        // Try to deserialize into new version - file_ids should be ignored
        let result: Result<RemovedFiles, _> = serde_json::from_str(&old_json);

        // This should succeed and create a default RemovedFiles (empty files set)
        assert!(result.is_ok(), "{:?}", result);
        let removed_files = result.unwrap();
        assert_eq!(removed_files.removed_at, 1234567890);
        assert!(removed_files.files.is_empty());

        // Test that new format still works
        let file_id = FileId::random();
        let new_json = format!(
            r#"{{
            "removed_at": 1234567890,
            "files": ["{}"]
        }}"#,
            file_id
        );

        let result: Result<RemovedFiles, _> = serde_json::from_str(&new_json);
        assert!(result.is_ok());
        let removed_files = result.unwrap();
        assert_eq!(removed_files.removed_at, 1234567890);
        assert_eq!(removed_files.files.len(), 1);
        assert!(
            removed_files
                .files
                .contains(&RemovedFile::File(file_id, None))
        );
    }

    fn tail(node_id: u64, after: EntryId, upto: EntryId) -> PendingTail {
        PendingTail {
            node_id,
            after,
            upto,
            upto_kind: UptoKind::Reported,
        }
    }

    fn wal_recovery_edit(
        flushed_entry_id: Option<EntryId>,
        wal_recovery: WalRecoveryEdit,
    ) -> RegionEdit {
        RegionEdit {
            files_to_add: vec![],
            files_to_remove: vec![],
            timestamp_ms: None,
            compaction_time_window: None,
            flushed_entry_id,
            flushed_sequence: None,
            committed_sequence: None,
            wal_recovery,
        }
    }

    fn manifest_builder() -> RegionManifestBuilder {
        let mut builder = RegionManifestBuilder::default();
        builder.apply_change(
            0,
            RegionChange {
                metadata: Arc::new(crate::manifest::tests::utils::basic_region_metadata()),
                sst_format: FormatType::PrimaryKey,
                append_mode: None,
            },
        );
        builder
    }

    #[test]
    fn test_checkpoint_sharing_preserves_owned_allocations() {
        let mut checkpoint = manifest_builder().try_build().unwrap();
        let expr = partition::expr::col("tag").lt(datatypes::value::Value::UInt32(10));
        let mut allocations = HashMap::new();
        for _ in 0..4 {
            let meta = FileMetaRef::from(FileMeta {
                region_id: RegionId::new(1, 1),
                file_id: FileId::random(),
                indexes: vec![crate::sst::file::ColumnIndexMetadata::default()],
                partition_expr: Some(expr.clone().into()),
                ..Default::default()
            });
            allocations.insert(
                meta.file_id,
                (std::ptr::from_ref(meta.as_ref()), meta.indexes.as_ptr()),
            );
            checkpoint.files.insert(meta.file_id, meta);
        }
        let interner = Arc::new(crate::sst::file_metadata::FileMetadataInterner::default());
        let manifest = RegionManifestBuilder::with_checkpoint(Some(checkpoint), interner.clone())
            .try_build()
            .unwrap();
        let expression = manifest
            .files
            .values()
            .next()
            .unwrap()
            .partition_expr
            .as_ref()
            .unwrap();
        for meta in manifest.files.values() {
            assert_eq!(
                allocations[&meta.file_id],
                (std::ptr::from_ref(meta.as_ref()), meta.indexes.as_ptr())
            );
            assert!(MetadataRef::ptr_eq(
                expression,
                meta.partition_expr.as_ref().unwrap()
            ));
        }
        let decoded = serde_json::from_slice(&serde_json::to_vec(&manifest).unwrap()).unwrap();
        let recovered = RegionManifestBuilder::with_checkpoint(Some(decoded), interner)
            .try_build()
            .unwrap();
        for (id, meta) in &manifest.files {
            assert!(MetadataRef::ptr_eq(meta, &recovered.files[id]));
        }
    }

    #[test]
    fn test_pending_tails_change_only_when_set() {
        let mut builder = manifest_builder();
        builder.apply_edit(
            1,
            wal_recovery_edit(
                None,
                WalRecoveryEdit {
                    pending_tails: Some(vec![tail(1, 10, 20)]),
                    handoff_floor: None,
                },
            ),
        );
        builder.apply_edit(2, wal_recovery_edit(Some(5), WalRecoveryEdit::default()));
        assert_eq!(builder.pending_tails, vec![tail(1, 10, 20)]);

        builder.apply_edit(
            3,
            wal_recovery_edit(
                None,
                WalRecoveryEdit {
                    pending_tails: Some(vec![]),
                    handoff_floor: None,
                },
            ),
        );
        assert!(builder.pending_tails.is_empty());
    }

    #[test]
    fn test_flushed_entry_id_drops_reached_floors_and_keeps_tails() {
        let mut builder = manifest_builder();
        for (node_id, entry_id) in [(1, 20), (2, 30)] {
            builder.apply_edit(
                1,
                wal_recovery_edit(
                    None,
                    WalRecoveryEdit {
                        pending_tails: Some(vec![tail(1, 10, 20)]),
                        handoff_floor: Some(HandoffFloor { node_id, entry_id }),
                    },
                ),
            );
        }
        builder.apply_edit(2, wal_recovery_edit(Some(20), WalRecoveryEdit::default()));

        let manifest = builder.try_build().unwrap();
        assert_eq!(manifest.pending_tails, vec![tail(1, 10, 20)]);
        assert_eq!(manifest.handoff_floors, BTreeMap::from([(2, 30)]));
    }

    #[test]
    fn test_truncate_drops_reached_floors() {
        let mut builder = manifest_builder();
        builder.apply_edit(
            1,
            wal_recovery_edit(
                None,
                WalRecoveryEdit {
                    pending_tails: None,
                    handoff_floor: Some(HandoffFloor {
                        node_id: 1,
                        entry_id: 20,
                    }),
                },
            ),
        );
        builder.apply_truncate(
            2,
            RegionTruncate {
                region_id: RegionId::new(1, 1),
                kind: TruncateKind::All {
                    truncated_entry_id: 20,
                    truncated_sequence: 20,
                },
                timestamp_ms: None,
            },
        );

        assert!(builder.try_build().unwrap().handoff_floors.is_empty());
    }

    #[test]
    fn test_wal_replay_start() {
        let mut builder = manifest_builder();
        // Node 1 was handed off at 40 after its first tail, so its second tail starts
        // there and entries in (20, 40] belong to no tail.
        builder.apply_edit(
            1,
            wal_recovery_edit(
                Some(10),
                WalRecoveryEdit {
                    pending_tails: Some(vec![tail(1, 10, 20), tail(2, 30, 60), tail(1, 40, 50)]),
                    handoff_floor: Some(HandoffFloor {
                        node_id: 1,
                        entry_id: 40,
                    }),
                },
            ),
        );
        builder.apply_edit(
            2,
            wal_recovery_edit(
                None,
                WalRecoveryEdit {
                    pending_tails: None,
                    handoff_floor: Some(HandoffFloor {
                        node_id: 3,
                        entry_id: 40,
                    }),
                },
            ),
        );
        let manifest = builder.try_build().unwrap();

        assert_eq!(manifest.wal_replay_start(Some(1)), 50);
        assert_eq!(manifest.wal_replay_start(Some(2)), 60);
        assert_eq!(manifest.wal_replay_start(Some(3)), 40);
        assert_eq!(manifest.wal_replay_start(Some(4)), 10);
        assert_eq!(manifest.wal_replay_start(None), 10);
    }

    #[test]
    fn test_cap_wal_release() {
        let tails = [tail(1, 20, 25), tail(2, 5, 60), tail(1, 30, 40)];

        assert_eq!(cap_wal_release(&tails, Some(1), 50), 20);
        assert_eq!(cap_wal_release(&tails, Some(1), 10), 10);
        assert_eq!(cap_wal_release(&tails, Some(3), 50), 50);
        assert_eq!(cap_wal_release(&tails, None, 50), 50);
    }

    const PENDING_TAILS_DELTA: &str =
        include_str!("../../../../tests/compatibility/manifest/pending_tails/delta.json");
    const PENDING_TAILS_CHECKPOINT: &str =
        include_str!("../../../../tests/compatibility/manifest/pending_tails/checkpoint.json");

    fn fixture_pending_tails() -> Vec<PendingTail> {
        vec![
            tail(1, 1 << 32, (1 << 32) + 5),
            PendingTail {
                upto_kind: UptoKind::Listed,
                ..tail(2, 2 << 32, 3 << 32)
            },
        ]
    }

    #[test]
    fn test_pending_tails_fixtures() {
        let delta = PENDING_TAILS_DELTA.trim_end();
        let actions = RegionMetaActionList::decode(delta.as_bytes()).unwrap();
        let [RegionMetaAction::Edit(edit)] = actions.actions.as_slice() else {
            panic!("unexpected actions: {actions:?}");
        };
        let expected = WalRecoveryEdit {
            pending_tails: Some(fixture_pending_tails()),
            handoff_floor: Some(HandoffFloor {
                node_id: 3,
                entry_id: 4 << 32,
            }),
        };
        assert_eq!(edit.wal_recovery, expected);
        assert_eq!(actions.encode().unwrap(), delta.as_bytes());

        let checkpoint = PENDING_TAILS_CHECKPOINT.trim_end();
        let decoded = RegionCheckpoint::decode(checkpoint.as_bytes()).unwrap();
        let manifest = decoded.checkpoint.as_ref().unwrap();
        assert_eq!(manifest.pending_tails, fixture_pending_tails());
        assert_eq!(manifest.handoff_floors, BTreeMap::from([(3, 4 << 32)]));
        let encoded: serde_json::Value =
            serde_json::from_slice(&decoded.encode().unwrap()).unwrap();
        let frozen: serde_json::Value = serde_json::from_str(checkpoint).unwrap();
        for key in ["pending_tails", "handoff_floors"] {
            assert_eq!(encoded["checkpoint"][key], frozen["checkpoint"][key]);
        }
    }

    /// Checks the fixtures against the shapes before pending tails were added.
    #[test]
    fn test_pending_tails_old_shape_compat() {
        #[derive(Serialize, Deserialize)]
        struct OldRegionEdit {
            files_to_add: Vec<FileMeta>,
            files_to_remove: Vec<FileMeta>,
            #[serde(default)]
            timestamp_ms: Option<i64>,
            #[serde(with = "humantime_serde")]
            compaction_time_window: Option<Duration>,
            flushed_entry_id: Option<EntryId>,
            flushed_sequence: Option<SequenceNumber>,
            committed_sequence: Option<SequenceNumber>,
        }

        #[derive(Serialize, Deserialize)]
        struct OldRegionManifest {
            metadata: RegionMetadataRef,
            files: HashMap<FileId, FileMeta>,
            #[serde(default)]
            removed_files: RemovedFilesRecord,
            flushed_entry_id: EntryId,
            flushed_sequence: SequenceNumber,
            committed_sequence: Option<SequenceNumber>,
            manifest_version: ManifestVersion,
            truncated_entry_id: Option<EntryId>,
            #[serde(with = "humantime_serde")]
            compaction_time_window: Option<Duration>,
            #[serde(default)]
            sst_format: FormatType,
            #[serde(default)]
            append_mode: Option<bool>,
        }

        let delta: serde_json::Value = serde_json::from_str(PENDING_TAILS_DELTA).unwrap();
        let old_edit: OldRegionEdit =
            serde_json::from_value(delta["actions"][0]["Edit"].clone()).unwrap();
        assert_eq!(old_edit.committed_sequence, Some((1 << 40) + 5));
        let old_json = serde_json::to_string(&old_edit).unwrap();
        let edit: RegionEdit = serde_json::from_str(&old_json).unwrap();
        assert_eq!(edit.wal_recovery, WalRecoveryEdit::default());
        assert_eq!(serde_json::to_string(&edit).unwrap(), old_json);

        let checkpoint: serde_json::Value = serde_json::from_str(PENDING_TAILS_CHECKPOINT).unwrap();
        let old_manifest: OldRegionManifest =
            serde_json::from_value(checkpoint["checkpoint"].clone()).unwrap();
        assert_eq!(old_manifest.flushed_entry_id, 1 << 32);
        let old_json = serde_json::to_string(&old_manifest).unwrap();
        let manifest: RegionManifest = serde_json::from_str(&old_json).unwrap();
        assert!(manifest.pending_tails.is_empty() && manifest.handoff_floors.is_empty());
        assert_eq!(serde_json::to_string(&manifest).unwrap(), old_json);
    }
}