razeri_plugin_contract/

storage_strategy.rs

// Copyright (C) 2026 Gabriel Shea <gshea@ideapivot.com> - All Rights Reserved

//! Agnostic storage strategy types: partition, compression, index.
//!
//! These platform-neutral types are consumed by the `StorageRenderer` trait
//! (added in Phase 4 of the time-partitioning plan) so each dialect can
//! translate them to native SQL. No platform-specific data ever appears here.
//!
//! See `prompt/archive/2026/20260616-incorporate-time-partitioning-plan/03-phase2-storage-types.md`
//! and ADRs `TP-001..TP-012`.

use serde::{Deserialize, Serialize};

/// Agnostic storage strategy for a single materialized table.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct StoragePlan {
    /// Partitioning strategy applied to the table.
    pub partition: PartitionStrategy,
    /// Compression strategy applied to the table's storage pages.
    pub compression: CompressionStrategy,
    /// Index strategy applied to the table (clustered, columnstore, Z-order, etc.).
    pub index: IndexStrategy,
}

/// Partitioning strategy for a table.
///
/// `RangeByTimeDimension` is the only supported partitioning strategy; it
/// requires the entity to expose a single `FieldRole::TimeDimension` field.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "strategy", rename_all = "snake_case")]
pub enum PartitionStrategy {
    /// No partitioning; the table is stored as a single segment.
    None,
    /// Range partition on a single `FieldRole::TimeDimension` field.
    RangeByTimeDimension {
        /// Name of the time-dimension field used as the partition key.
        field: String,
        /// Granularity of each partition range (day, week, month, etc.).
        granularity: PartitionGranularity,
        /// Optional retention policy that removes or archives old partitions.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        retention: Option<RetentionPolicy>,
        /// How many partitions to pre-create ahead of `now()` for the extend
        /// procedure. Defaults: 7 for `Day`, 4 for `Week`, 2 for `Month`,
        /// 1 for `Quarter`/`Year`.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        look_ahead: Option<u32>,
        /// Time-window sizing inputs used to auto-select an appropriate
        /// granularity that fits within `partition_budget` (default 2000)
        /// when the dialect imposes a per-table partition cap (e.g. SQL
        /// Server's 5000-partition limit).
        ///
        /// These fields are advisory: when present, the materializer's
        /// `select_grain` helper uses them to coarsen the granularity from
        /// the requested grain (e.g. `Day` → `Week` → `Month` → `Year`)
        /// until `(history_days + forecast_days) / grain ≤ partition_budget`.
        ///
        /// Authoring sources (in priority order):
        ///   1. Explicit values in the design YAML (highest authority).
        ///   2. AI wizard inferring from `Strand` profiling
        ///      (`min_value`/`max_value`).
        ///   3. Defaults baked into the materializer.
        #[serde(default, skip_serializing_if = "PartitionWindow::is_default")]
        window: PartitionWindow,
    },
}

/// Time-window sizing inputs that drive automatic granularity selection
/// and forecast-aware partition extension.
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
pub struct PartitionWindow {
    /// Expected historical horizon in days. Combined with `forecast_days`
    /// to compute the total time span the partition function must cover.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub history_days: Option<u32>,

    /// Expected forward horizon in days (forecast / projection data).
    /// Causes the `extend_proc` to pre-allocate partitions far enough
    /// ahead to receive future-dated rows.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub forecast_days: Option<u32>,

    /// Maximum number of partitions before the materializer auto-coarsens
    /// the granularity. Defaults to 2000 (well under SQL Server's 5000
    /// hard limit, leaving headroom for years of `extend_proc` runs).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub partition_budget: Option<u32>,
}

impl PartitionWindow {
    /// Returns `true` when no sizing fields are populated (used by serde
    /// `skip_serializing_if` to keep YAML/JSON outputs tidy).
    #[must_use]
    pub fn is_default(&self) -> bool {
        self == &Self::default()
    }
}

/// Granularity of each partition range for time-based partitioning.
///
/// Coarser granularities reduce partition count but increase scan cost per query.
/// Use [`PartitionGranularity::next_coarser`] to step down when the dialect's
/// partition limit would be exceeded.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PartitionGranularity {
    /// One partition per calendar day.
    Day,
    /// One partition per ISO week (Monday–Sunday).
    Week,
    /// One partition per calendar month.
    Month,
    /// One partition per calendar quarter.
    Quarter,
    /// One partition per calendar year.
    Year,
}

impl PartitionGranularity {
    /// Returns the next coarser granularity (`Day` → `Week` → `Month` →
    /// `Quarter` → `Year`), or `None` when already at `Year`.
    ///
    /// Drives the auto-coarsening loop in the materializer's `select_grain`
    /// helper when the requested grain would exceed `partition_budget`.
    #[must_use]
    pub const fn next_coarser(self) -> Option<Self> {
        match self {
            Self::Day => Some(Self::Week),
            Self::Week => Some(Self::Month),
            Self::Month => Some(Self::Quarter),
            Self::Quarter => Some(Self::Year),
            Self::Year => None,
        }
    }

    /// Approximate number of days per granule. Used for partition-count
    /// arithmetic in `select_grain`.
    #[must_use]
    pub const fn approx_days(self) -> u32 {
        match self {
            Self::Day => 1,
            Self::Week => 7,
            Self::Month => 30,
            Self::Quarter => 92,
            Self::Year => 365,
        }
    }
}

/// Policy for removing or archiving old partitions during the retain procedure.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RetentionPolicy {
    /// Number of granules (days/weeks/etc.) to keep before the retain
    /// procedure removes older partitions.
    pub keep: u32,
    /// Whether the retain procedure DROPs (destructive) or SWITCHes/DETACHes
    /// to a sibling archive table.
    #[serde(default)]
    pub action: RetentionAction,
}

/// How old partitions are disposed of when the retain procedure runs.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RetentionAction {
    /// Destructively `DROP` the partition (or detach and delete).
    #[default]
    Drop,
    /// Move the partition to a sibling archive table instead of dropping it.
    Archive,
}

/// Compression strategy for a materialized table.
///
/// Platform renderers translate these variants into the appropriate dialect-native
/// `WITH (DATA_COMPRESSION = ...)`, `USING DELTA`, or `CODEC(...)` syntax.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum CompressionStrategy {
    /// No compression applied.
    None,
    /// SQL Server page-level compression.
    Page,
    /// SQL Server row-level compression.
    Row,
    /// Columnstore compression (columnar layout; SQL Server, Synapse, Azure Fabric).
    Columnstore,
    /// Columnstore archival compression (higher ratio, slower scans).
    ColumnstoreArchive,
    /// LZ4 block compression (Delta Lake, `ClickHouse`).
    Lz4,
    /// Zstd compression (Delta Lake, `ClickHouse`, `DuckDB`).
    Zstd {
        /// Compression level (1 = fastest, higher = smaller; dialect-specific range).
        #[serde(default, skip_serializing_if = "Option::is_none")]
        level: Option<u8>,
    },
    /// Snappy block compression (Parquet, Delta Lake).
    Snappy,
}

/// Index strategy for a materialized table.
///
/// Platform renderers translate these variants into dialect-native DDL
/// (`CREATE CLUSTERED COLUMNSTORE INDEX`, `CLUSTER BY`, `Z-ORDER BY`, etc.).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum IndexStrategy {
    /// No explicit index DDL generated.
    None,
    /// Clustered columnstore index (SQL Server, Synapse, Fabric).
    ClusteredColumnstore,
    /// Non-clustered columnstore index (SQL Server).
    NonClusteredColumnstore,
    /// Delta-style `Z-ORDER` on the listed columns.
    ZOrder {
        /// Columns to include in the Z-order optimization.
        columns: Vec<String>,
    },
    /// `Snowflake` `CLUSTER BY` (or Delta Liquid Clustering, `ClickHouse` `ORDER BY`).
    ClusterBy {
        /// Columns to include in the cluster key.
        columns: Vec<String>,
    },
}

impl StoragePlan {
    /// Returns a plan that produces no SQL on any dialect.
    #[must_use]
    pub fn none() -> Self {
        Self {
            partition: PartitionStrategy::None,
            compression: CompressionStrategy::None,
            index: IndexStrategy::None,
        }
    }

    /// Returns `true` if the plan declares a partitioning strategy other than `None`.
    #[must_use]
    pub fn is_partitioned(&self) -> bool {
        !matches!(self.partition, PartitionStrategy::None)
    }
}

/// Names of the maintenance procedures generated for a partitioned entity.
///
/// Populated by the materializer (Phase 7) and consumed by orchestration so
/// schedulers know what to invoke for sliding-window upkeep.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PartitionMaintenance {
    /// Name of the procedure that pre-creates upcoming partitions.
    pub extend_proc_name: String,
    /// Name of the procedure that drops/archives expired partitions
    /// (only present when a `RetentionPolicy` is configured).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub retain_proc_name: Option<String>,
}

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

    #[test]
    fn storage_plan_none_is_empty() {
        let plan = StoragePlan::none();
        assert!(!plan.is_partitioned());
        assert!(matches!(plan.partition, PartitionStrategy::None));
        assert!(matches!(plan.compression, CompressionStrategy::None));
        assert!(matches!(plan.index, IndexStrategy::None));
    }

    #[test]
    fn range_partition_round_trips_via_yaml() {
        let plan = StoragePlan {
            partition: PartitionStrategy::RangeByTimeDimension {
                field: "sale_date".into(),
                granularity: PartitionGranularity::Day,
                retention: Some(RetentionPolicy { keep: 730, action: RetentionAction::Drop }),
                look_ahead: Some(7),
                window: PartitionWindow::default(),
            },
            compression: CompressionStrategy::Page,
            index: IndexStrategy::ClusteredColumnstore,
        };
        let yaml = serde_yaml::to_string(&plan).unwrap();
        let parsed: StoragePlan = serde_yaml::from_str(&yaml).unwrap();
        assert_eq!(plan, parsed);
        assert!(plan.is_partitioned());
    }

    #[test]
    fn partition_window_default_is_skipped_in_yaml() {
        let plan = StoragePlan {
            partition: PartitionStrategy::RangeByTimeDimension {
                field: "sale_date".into(),
                granularity: PartitionGranularity::Day,
                retention: None,
                look_ahead: None,
                window: PartitionWindow::default(),
            },
            compression: CompressionStrategy::None,
            index: IndexStrategy::None,
        };
        let yaml = serde_yaml::to_string(&plan).unwrap();
        assert!(!yaml.contains("window"), "default window should not serialize: {yaml}");
    }

    #[test]
    fn partition_window_round_trips_via_yaml() {
        let window = PartitionWindow {
            history_days: Some(1825),
            forecast_days: Some(365),
            partition_budget: Some(2000),
        };
        let plan = StoragePlan {
            partition: PartitionStrategy::RangeByTimeDimension {
                field: "sale_date".into(),
                granularity: PartitionGranularity::Day,
                retention: None,
                look_ahead: None,
                window: window.clone(),
            },
            compression: CompressionStrategy::None,
            index: IndexStrategy::None,
        };
        let yaml = serde_yaml::to_string(&plan).unwrap();
        let parsed: StoragePlan = serde_yaml::from_str(&yaml).unwrap();
        assert_eq!(plan, parsed);
        let PartitionStrategy::RangeByTimeDimension { window: w, .. } = parsed.partition else {
            unreachable!()
        };
        assert_eq!(w, window);
    }

    #[test]
    fn next_coarser_walks_to_year_then_stops() {
        assert_eq!(PartitionGranularity::Day.next_coarser(), Some(PartitionGranularity::Week));
        assert_eq!(PartitionGranularity::Week.next_coarser(), Some(PartitionGranularity::Month));
        assert_eq!(PartitionGranularity::Month.next_coarser(), Some(PartitionGranularity::Quarter));
        assert_eq!(PartitionGranularity::Quarter.next_coarser(), Some(PartitionGranularity::Year));
        assert_eq!(PartitionGranularity::Year.next_coarser(), None);
    }

    #[test]
    fn zstd_with_level_serializes() {
        let c = CompressionStrategy::Zstd { level: Some(3) };
        let json = serde_json::to_string(&c).unwrap();
        assert!(json.contains("\"zstd\""));
        assert!(json.contains('3'));
    }

    #[test]
    fn retention_action_defaults_to_drop() {
        let yaml = "keep: 90\n";
        let policy: RetentionPolicy = serde_yaml::from_str(yaml).unwrap();
        assert_eq!(policy.keep, 90);
        assert_eq!(policy.action, RetentionAction::Drop);
    }

    #[test]
    fn z_order_round_trips_via_json() {
        let idx = IndexStrategy::ZOrder { columns: vec!["sale_date".into(), "store_id".into()] };
        let json = serde_json::to_string(&idx).unwrap();
        let parsed: IndexStrategy = serde_json::from_str(&json).unwrap();
        assert_eq!(idx, parsed);
    }

    #[test]
    fn cluster_by_round_trips_via_json() {
        let idx = IndexStrategy::ClusterBy { columns: vec!["created_at".into()] };
        let json = serde_json::to_string(&idx).unwrap();
        let parsed: IndexStrategy = serde_json::from_str(&json).unwrap();
        assert_eq!(idx, parsed);
    }

    #[test]
    fn partition_strategy_none_serializes_with_tag() {
        let p = PartitionStrategy::None;
        let json = serde_json::to_string(&p).unwrap();
        assert!(json.contains("\"none\""));
    }

    #[test]
    fn partition_maintenance_round_trips() {
        let m = PartitionMaintenance {
            extend_proc_name: "core_sales_partition_extend".into(),
            retain_proc_name: Some("core_sales_partition_retain".into()),
        };
        let json = serde_json::to_string(&m).unwrap();
        let parsed: PartitionMaintenance = serde_json::from_str(&json).unwrap();
        assert_eq!(m, parsed);

        let no_retain = PartitionMaintenance {
            extend_proc_name: "core_orders_partition_extend".into(),
            retain_proc_name: None,
        };
        let json = serde_json::to_string(&no_retain).unwrap();
        assert!(!json.contains("retain_proc_name"));
    }
}