hamelin_translation 0.3.10

//! Window frame specification for WINDOW commands
//!
//! This module defines the `WindowFrame` struct that represents a validated window frame
//! specification. It is created during lowering by evaluating the WITHIN expression
//! and is consumed by backend translators (DataFusion, SQL, etc.).
//!
//! # Design
//!
//! The WITHIN clause in Hamelin uses sign conventions:
//! - **Negative values = PRECEDING**: `-1h` means 1 hour preceding
//! - **Positive values = FOLLOWING**: `1h` means 1 hour following
//! - **Zero = CURRENT ROW**: A single value like `-1h` becomes "1h PRECEDING to CURRENT ROW"
//! - **Range syntax**: `-2h..-1h` means "2h PRECEDING to 1h PRECEDING"
//! - **Unbounded**: `..0` means "UNBOUNDED PRECEDING to CURRENT ROW"
//!
//! # Frame Units
//!
//! - **ROWS**: Row-based frame (`-10r..10r`, `0r`, `-5r`)
//! - **RANGE**: Value-based frame using intervals or calendar intervals (`-1h`, `-2h..-1h`)
//!
//! # Conversion from eval'd Value
//!
//! During lowering, the WITHIN expression is evaluated (must not reference columns)
//! and converted to `WindowFrame`. The sign of the value determines Preceding/Following,
//! and zero values map to CurrentRow.

use chrono::Duration;

use hamelin_eval::value::{RangeValue, Value};

/// A time-based offset for TimeRange bounds
#[derive(Debug, Clone, PartialEq)]
pub enum TimeOffset {
    /// Fixed interval (always non-negative duration)
    Interval(Duration),
    /// Calendar interval (always non-negative number of months)
    CalendarInterval(i32),
}

/// A bound for row-based frames
#[derive(Debug, Clone, PartialEq)]
pub enum RowBound {
    /// UNBOUNDED PRECEDING or UNBOUNDED FOLLOWING
    Unbounded,
    /// CURRENT ROW
    CurrentRow,
    /// N PRECEDING (value is always non-negative)
    Preceding(u64),
    /// N FOLLOWING (value is always non-negative)
    Following(u64),
}

/// A bound for value-based range frames (matches sort column type)
#[derive(Debug, Clone, PartialEq)]
pub enum RangeBound {
    /// UNBOUNDED PRECEDING or UNBOUNDED FOLLOWING
    Unbounded,
    /// CURRENT ROW
    CurrentRow,
    /// N PRECEDING (value is always non-negative)
    Preceding(Value),
    /// N FOLLOWING (value is always non-negative)
    Following(Value),
}

/// A bound for time-based range frames
#[derive(Debug, Clone, PartialEq)]
pub enum TimeRangeBound {
    /// UNBOUNDED PRECEDING or UNBOUNDED FOLLOWING
    Unbounded,
    /// CURRENT ROW
    CurrentRow,
    /// N PRECEDING (offset is always non-negative)
    Preceding(TimeOffset),
    /// N FOLLOWING (offset is always non-negative)
    Following(TimeOffset),
}

/// A validated window frame specification
#[derive(Debug, Clone)]
pub enum WindowFrame {
    /// Row-based frame (e.g., `-10r..10r`)
    Rows { start: RowBound, end: RowBound },
    /// Value-based range frame matching the sort column type
    Range { start: RangeBound, end: RangeBound },
    /// Time-based range frame using intervals
    TimeRange {
        start: TimeRangeBound,
        end: TimeRangeBound,
    },
}

impl WindowFrame {
    /// Convert an evaluated Value to a WindowFrame
    ///
    /// # Arguments
    ///
    /// * `value` - The evaluated WITHIN expression value
    ///
    /// # Value Types
    ///
    /// - `Value::Interval(d)` - Single interval: TimeRange with start/end based on sign
    /// - `Value::CalendarInterval(rd)` - Single calendar interval: TimeRange
    /// - `Value::Rows(n)` - Single row count: Rows frame
    /// - `Value::Range(rv)` - Range of bounds: frame type determined by bound types
    pub fn from_value(value: Value) -> Result<Self, String> {
        match value {
            // Single interval value -> TimeRange
            Value::Interval(d) => {
                let zero = Duration::zero();
                let (start, end) = if d < zero {
                    // Negative interval: N PRECEDING to CURRENT ROW
                    let abs_duration = -d;
                    (
                        TimeRangeBound::Preceding(TimeOffset::Interval(abs_duration)),
                        TimeRangeBound::CurrentRow,
                    )
                } else if d > zero {
                    // Positive interval: CURRENT ROW to N FOLLOWING
                    (
                        TimeRangeBound::CurrentRow,
                        TimeRangeBound::Following(TimeOffset::Interval(d)),
                    )
                } else {
                    // Zero interval: CURRENT ROW to CURRENT ROW
                    (TimeRangeBound::CurrentRow, TimeRangeBound::CurrentRow)
                };

                Ok(WindowFrame::TimeRange { start, end })
            }

            // Single calendar interval value -> TimeRange
            Value::CalendarInterval(months) => {
                let (start, end) = if months < 0 {
                    // Negative: N PRECEDING to CURRENT ROW
                    (
                        TimeRangeBound::Preceding(TimeOffset::CalendarInterval(-months)),
                        TimeRangeBound::CurrentRow,
                    )
                } else if months > 0 {
                    // Positive: CURRENT ROW to N FOLLOWING
                    (
                        TimeRangeBound::CurrentRow,
                        TimeRangeBound::Following(TimeOffset::CalendarInterval(months)),
                    )
                } else {
                    // Zero: CURRENT ROW to CURRENT ROW
                    (TimeRangeBound::CurrentRow, TimeRangeBound::CurrentRow)
                };

                Ok(WindowFrame::TimeRange { start, end })
            }

            // Single rows value -> Rows
            Value::Rows(n) => {
                let (start, end) = if n < 0 {
                    // Negative: N PRECEDING to CURRENT ROW
                    let abs_n = n.unsigned_abs();
                    (RowBound::Preceding(abs_n), RowBound::CurrentRow)
                } else if n > 0 {
                    // Positive: CURRENT ROW to N FOLLOWING
                    (RowBound::CurrentRow, RowBound::Following(n as u64))
                } else {
                    // Zero: CURRENT ROW to CURRENT ROW
                    (RowBound::CurrentRow, RowBound::CurrentRow)
                };

                Ok(WindowFrame::Rows { start, end })
            }

            // Range value - determine frame type from bounds
            Value::Range(rv) => {
                let RangeValue { lower, upper } = *rv;
                convert_range_value(lower, upper)
            }

            // Single int value -> Range
            Value::Int(n) => {
                let (start, end) = if n < 0 {
                    // Negative: N PRECEDING to CURRENT ROW
                    (
                        RangeBound::Preceding(Value::Int(-n)),
                        RangeBound::CurrentRow,
                    )
                } else if n > 0 {
                    // Positive: CURRENT ROW to N FOLLOWING
                    (RangeBound::CurrentRow, RangeBound::Following(Value::Int(n)))
                } else {
                    // Zero: CURRENT ROW to CURRENT ROW
                    (RangeBound::CurrentRow, RangeBound::CurrentRow)
                };

                Ok(WindowFrame::Range { start, end })
            }

            // Single double value -> Range
            Value::Double(f) => {
                let (start, end) = if f < 0.0 {
                    // Negative: N PRECEDING to CURRENT ROW
                    (
                        RangeBound::Preceding(Value::Double(-f)),
                        RangeBound::CurrentRow,
                    )
                } else if f > 0.0 {
                    // Positive: CURRENT ROW to N FOLLOWING
                    (
                        RangeBound::CurrentRow,
                        RangeBound::Following(Value::Double(f)),
                    )
                } else {
                    // Zero: CURRENT ROW to CURRENT ROW
                    (RangeBound::CurrentRow, RangeBound::CurrentRow)
                };

                Ok(WindowFrame::Range { start, end })
            }

            _ => Err(format!(
                "Cannot convert {} to window frame",
                value.type_name()
            )),
        }
    }
}

/// The type of frame determined from range bound values
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FrameType {
    Rows,
    TimeRange,
    Range,
}

/// Determine the frame type from a single value
fn value_frame_type(v: &Value) -> Result<FrameType, String> {
    match v {
        Value::Rows(_) => Ok(FrameType::Rows),
        Value::Interval(_) | Value::CalendarInterval(_) => Ok(FrameType::TimeRange),
        // Other types (Int, Float, etc.) -> generic Range
        _ => Ok(FrameType::Range),
    }
}

/// Convert a range value (with lower and upper bounds) to WindowFrame
fn convert_range_value(lower: Option<Value>, upper: Option<Value>) -> Result<WindowFrame, String> {
    // Determine frame type from bounds
    let lower_type = lower.as_ref().map(value_frame_type).transpose()?;
    let upper_type = upper.as_ref().map(value_frame_type).transpose()?;

    let frame_type = match (lower_type, upper_type) {
        (Some(t1), Some(t2)) if t1 == t2 => t1,
        (Some(t1), Some(t2)) => {
            return Err(format!(
                "Mismatched frame bound types: {:?} and {:?}",
                t1, t2
            ));
        }
        (Some(t), None) | (None, Some(t)) => t,
        (None, None) => {
            // Both unbounded - default to TimeRange (most common case)
            FrameType::TimeRange
        }
    };

    match frame_type {
        FrameType::Rows => {
            let start = match lower {
                None => RowBound::Unbounded,
                Some(v) => convert_row_bound(v)?,
            };
            let end = match upper {
                None => RowBound::Unbounded,
                Some(v) => convert_row_bound(v)?,
            };
            Ok(WindowFrame::Rows { start, end })
        }
        FrameType::TimeRange => {
            let start = match lower {
                None => TimeRangeBound::Unbounded,
                Some(v) => convert_time_range_bound(v)?,
            };
            let end = match upper {
                None => TimeRangeBound::Unbounded,
                Some(v) => convert_time_range_bound(v)?,
            };
            Ok(WindowFrame::TimeRange { start, end })
        }
        FrameType::Range => {
            let start = match lower {
                None => RangeBound::Unbounded,
                Some(v) => convert_generic_range_bound(v)?,
            };
            let end = match upper {
                None => RangeBound::Unbounded,
                Some(v) => convert_generic_range_bound(v)?,
            };
            Ok(WindowFrame::Range { start, end })
        }
    }
}

/// Convert a value to a RowBound
fn convert_row_bound(v: Value) -> Result<RowBound, String> {
    match v {
        Value::Rows(n) => {
            if n < 0 {
                Ok(RowBound::Preceding(n.unsigned_abs()))
            } else if n > 0 {
                Ok(RowBound::Following(n as u64))
            } else {
                // Zero: Following(0)
                Ok(RowBound::Following(0))
            }
        }
        _ => Err(format!("Cannot convert {} to row bound", v.type_name())),
    }
}

/// Convert a value to a TimeRangeBound
fn convert_time_range_bound(v: Value) -> Result<TimeRangeBound, String> {
    match v {
        Value::Interval(d) => {
            let zero = Duration::zero();
            if d < zero {
                Ok(TimeRangeBound::Preceding(TimeOffset::Interval(-d)))
            } else if d > zero {
                Ok(TimeRangeBound::Following(TimeOffset::Interval(d)))
            } else {
                // Zero: Following(0) - per Hamelin semantics
                Ok(TimeRangeBound::Following(TimeOffset::Interval(
                    Duration::zero(),
                )))
            }
        }
        Value::CalendarInterval(months) => {
            if months < 0 {
                Ok(TimeRangeBound::Preceding(TimeOffset::CalendarInterval(
                    -months,
                )))
            } else {
                // Zero or positive: Following
                Ok(TimeRangeBound::Following(TimeOffset::CalendarInterval(
                    months,
                )))
            }
        }
        _ => Err(format!(
            "Cannot convert {} to time range bound",
            v.type_name()
        )),
    }
}

/// Convert a value to a generic RangeBound
///
/// For generic range bounds, we store the absolute value and determine
/// preceding/following based on sign.
fn convert_generic_range_bound(v: Value) -> Result<RangeBound, String> {
    // For generic values, we need to check if they're negative
    // This works for numeric types that support comparison
    match &v {
        Value::Int(n) => {
            if *n < 0 {
                Ok(RangeBound::Preceding(Value::Int(-*n)))
            } else if *n > 0 {
                Ok(RangeBound::Following(v))
            } else {
                Ok(RangeBound::Following(Value::Int(0)))
            }
        }
        Value::Double(f) => {
            if *f < 0.0 {
                Ok(RangeBound::Preceding(Value::Double(-*f)))
            } else if *f > 0.0 {
                Ok(RangeBound::Following(v))
            } else {
                Ok(RangeBound::Following(Value::Double(0.0)))
            }
        }
        _ => Err(format!(
            "Cannot convert {} to range bound - unsupported type for generic range",
            v.type_name()
        )),
    }
}

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

    #[test]
    fn test_negative_interval_to_frame() {
        let v = Value::Interval(Duration::hours(-2));
        let frame = WindowFrame::from_value(v).unwrap();

        match frame {
            WindowFrame::TimeRange { start, end } => {
                assert!(matches!(
                    start,
                    TimeRangeBound::Preceding(TimeOffset::Interval(d)) if d == Duration::hours(2)
                ));
                assert_eq!(end, TimeRangeBound::CurrentRow);
            }
            _ => panic!("Expected TimeRange frame"),
        }
    }

    #[test]
    fn test_positive_interval_to_frame() {
        let v = Value::Interval(Duration::hours(1));
        let frame = WindowFrame::from_value(v).unwrap();

        match frame {
            WindowFrame::TimeRange { start, end } => {
                assert_eq!(start, TimeRangeBound::CurrentRow);
                assert!(matches!(
                    end,
                    TimeRangeBound::Following(TimeOffset::Interval(d)) if d == Duration::hours(1)
                ));
            }
            _ => panic!("Expected TimeRange frame"),
        }
    }

    #[test]
    fn test_zero_interval_to_frame() {
        let v = Value::Interval(Duration::zero());
        let frame = WindowFrame::from_value(v).unwrap();

        match frame {
            WindowFrame::TimeRange { start, end } => {
                assert_eq!(start, TimeRangeBound::CurrentRow);
                assert_eq!(end, TimeRangeBound::CurrentRow);
            }
            _ => panic!("Expected TimeRange frame"),
        }
    }

    #[test]
    fn test_negative_rows_to_frame() {
        let v = Value::Rows(-5);
        let frame = WindowFrame::from_value(v).unwrap();

        match frame {
            WindowFrame::Rows { start, end } => {
                assert_eq!(start, RowBound::Preceding(5));
                assert_eq!(end, RowBound::CurrentRow);
            }
            _ => panic!("Expected Rows frame"),
        }
    }

    #[test]
    fn test_range_interval_to_frame() {
        // -2h..-1h
        let lower = Value::Interval(Duration::hours(-2));
        let upper = Value::Interval(Duration::hours(-1));
        let v = Value::Range(Box::new(RangeValue {
            lower: Some(lower),
            upper: Some(upper),
        }));
        let frame = WindowFrame::from_value(v).unwrap();

        match frame {
            WindowFrame::TimeRange { start, end } => {
                assert!(matches!(
                    start,
                    TimeRangeBound::Preceding(TimeOffset::Interval(d)) if d == Duration::hours(2)
                ));
                assert!(matches!(
                    end,
                    TimeRangeBound::Preceding(TimeOffset::Interval(d)) if d == Duration::hours(1)
                ));
            }
            _ => panic!("Expected TimeRange frame"),
        }
    }

    #[test]
    fn test_unbounded_preceding_to_zero() {
        // ..0 -> UNBOUNDED PRECEDING to 0 FOLLOWING
        let v = Value::Range(Box::new(RangeValue {
            lower: None,
            upper: Some(Value::Interval(Duration::zero())),
        }));
        let frame = WindowFrame::from_value(v).unwrap();

        match frame {
            WindowFrame::TimeRange { start, end } => {
                assert_eq!(start, TimeRangeBound::Unbounded);
                // Zero in range context becomes Following(0), not CurrentRow
                assert!(matches!(
                    end,
                    TimeRangeBound::Following(TimeOffset::Interval(d)) if d == Duration::zero()
                ));
            }
            _ => panic!("Expected TimeRange frame"),
        }
    }

    #[test]
    fn test_rows_range() {
        // -10r..10r
        let v = Value::Range(Box::new(RangeValue {
            lower: Some(Value::Rows(-10)),
            upper: Some(Value::Rows(10)),
        }));
        let frame = WindowFrame::from_value(v).unwrap();

        match frame {
            WindowFrame::Rows { start, end } => {
                assert_eq!(start, RowBound::Preceding(10));
                assert_eq!(end, RowBound::Following(10));
            }
            _ => panic!("Expected Rows frame"),
        }
    }
}