hamelin_translation 0.9.0

//! 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 integers, floats, intervals, or calendar intervals
//!
//! # 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. Range bounds are stored as `IRExpression` so
//! backends can translate them through their normal expression path.

use std::sync::Arc;

use anyhow::bail;
use chrono::Duration;

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

use crate::ir::IRExpression;

/// Convert a Value to an IRExpression (absolute value, always non-negative).
fn value_to_ir_expression(value: Value) -> IRExpression {
    IRExpression::new(Arc::new(value.into()))
}

/// 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 (integers, floats, intervals, calendar intervals)
#[derive(Debug, Clone)]
pub enum RangeBound {
    /// UNBOUNDED PRECEDING or UNBOUNDED FOLLOWING
    Unbounded,
    /// CURRENT ROW
    CurrentRow,
    /// N PRECEDING (expression evaluates to a non-negative value)
    Preceding(IRExpression),
    /// N FOLLOWING (expression evaluates to a non-negative value)
    Following(IRExpression),
}

/// 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 (integers, floats, intervals, calendar intervals)
    Range { start: RangeBound, end: RangeBound },
}

impl WindowFrame {
    /// Convert an evaluated Value to a WindowFrame
    ///
    /// # Arguments
    ///
    /// * `value` - The evaluated WITHIN expression value
    ///
    /// # Value Types
    ///
    /// - `Value::Interval(d)` - Single interval: Range with start/end based on sign
    /// - `Value::CalendarInterval(rd)` - Single calendar interval: Range
    /// - `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) -> anyhow::Result<Self> {
        match value {
            // Single interval value -> Range
            Value::Interval(d) => {
                let zero = Duration::zero();
                let (start, end) = if d < zero {
                    let abs_duration = -d;
                    (
                        RangeBound::Preceding(value_to_ir_expression(Value::Interval(
                            abs_duration,
                        ))),
                        RangeBound::CurrentRow,
                    )
                } else if d > zero {
                    (
                        RangeBound::CurrentRow,
                        RangeBound::Following(value_to_ir_expression(Value::Interval(d))),
                    )
                } else {
                    (RangeBound::CurrentRow, RangeBound::CurrentRow)
                };

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

            // Single calendar interval value -> Range
            Value::CalendarInterval(months) => {
                let (start, end) = if months < 0 {
                    (
                        RangeBound::Preceding(value_to_ir_expression(Value::CalendarInterval(
                            -months,
                        ))),
                        RangeBound::CurrentRow,
                    )
                } else if months > 0 {
                    (
                        RangeBound::CurrentRow,
                        RangeBound::Following(value_to_ir_expression(Value::CalendarInterval(
                            months,
                        ))),
                    )
                } else {
                    (RangeBound::CurrentRow, RangeBound::CurrentRow)
                };

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

            // Single rows value -> Rows
            Value::Rows(n) => {
                let (start, end) = if n < 0 {
                    let abs_n = n.unsigned_abs();
                    (RowBound::Preceding(abs_n), RowBound::CurrentRow)
                } else if n > 0 {
                    (RowBound::CurrentRow, RowBound::Following(n as u64))
                } else {
                    (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 {
                    (
                        RangeBound::Preceding(value_to_ir_expression(Value::Int(-n))),
                        RangeBound::CurrentRow,
                    )
                } else if n > 0 {
                    (
                        RangeBound::CurrentRow,
                        RangeBound::Following(value_to_ir_expression(Value::Int(n))),
                    )
                } else {
                    (RangeBound::CurrentRow, RangeBound::CurrentRow)
                };

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

            // Single double value -> Range
            Value::Double(f) => {
                let (start, end) = if f < 0.0 {
                    (
                        RangeBound::Preceding(value_to_ir_expression(Value::Double(-f))),
                        RangeBound::CurrentRow,
                    )
                } else if f > 0.0 {
                    (
                        RangeBound::CurrentRow,
                        RangeBound::Following(value_to_ir_expression(Value::Double(f))),
                    )
                } else {
                    (RangeBound::CurrentRow, RangeBound::CurrentRow)
                };

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

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

/// Whether the frame uses ROWS or RANGE semantics
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FrameKind {
    Rows,
    Range,
}

/// Determine the frame kind from a single value
fn value_frame_kind(v: &Value) -> FrameKind {
    match v {
        Value::Rows(_) => FrameKind::Rows,
        _ => FrameKind::Range,
    }
}

/// Convert a range value (with lower and upper bounds) to WindowFrame
fn convert_range_value(lower: Option<Value>, upper: Option<Value>) -> anyhow::Result<WindowFrame> {
    let lower_kind = lower.as_ref().map(value_frame_kind);
    let upper_kind = upper.as_ref().map(value_frame_kind);

    let frame_kind = match (lower_kind, upper_kind) {
        (Some(t1), Some(t2)) if t1 == t2 => t1,
        (Some(t1), Some(t2)) => {
            bail!("Mismatched frame bound types: {:?} and {:?}", t1, t2);
        }
        (Some(t), None) | (None, Some(t)) => t,
        (None, None) => FrameKind::Range,
    };

    match frame_kind {
        FrameKind::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 })
        }
        FrameKind::Range => {
            let start = match lower {
                None => RangeBound::Unbounded,
                Some(v) => convert_range_bound(v)?,
            };
            let end = match upper {
                None => RangeBound::Unbounded,
                Some(v) => convert_range_bound(v)?,
            };
            Ok(WindowFrame::Range { start, end })
        }
    }
}

/// Convert a value to a RowBound
fn convert_row_bound(v: Value) -> anyhow::Result<RowBound> {
    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 {
                Ok(RowBound::CurrentRow)
            }
        }
        _ => bail!("Cannot convert {} to row bound", v.type_name()),
    }
}

/// Convert a value to a RangeBound
fn convert_range_bound(v: Value) -> anyhow::Result<RangeBound> {
    match &v {
        Value::Interval(d) => {
            let zero = Duration::zero();
            if *d < zero {
                Ok(RangeBound::Preceding(value_to_ir_expression(
                    Value::Interval(-*d),
                )))
            } else if *d > zero {
                Ok(RangeBound::Following(value_to_ir_expression(v)))
            } else {
                Ok(RangeBound::CurrentRow)
            }
        }
        Value::CalendarInterval(months) => {
            if *months < 0 {
                Ok(RangeBound::Preceding(value_to_ir_expression(
                    Value::CalendarInterval(-*months),
                )))
            } else if *months > 0 {
                Ok(RangeBound::Following(value_to_ir_expression(v)))
            } else {
                Ok(RangeBound::CurrentRow)
            }
        }
        Value::Int(n) => {
            if *n < 0 {
                Ok(RangeBound::Preceding(value_to_ir_expression(Value::Int(
                    -*n,
                ))))
            } else if *n > 0 {
                Ok(RangeBound::Following(value_to_ir_expression(v)))
            } else {
                Ok(RangeBound::CurrentRow)
            }
        }
        Value::Double(f) => {
            if *f < 0.0 {
                Ok(RangeBound::Preceding(value_to_ir_expression(
                    Value::Double(-*f),
                )))
            } else if *f > 0.0 {
                Ok(RangeBound::Following(value_to_ir_expression(v)))
            } else {
                Ok(RangeBound::CurrentRow)
            }
        }
        _ => bail!("Cannot convert {} to range bound", v.type_name()),
    }
}

#[cfg(test)]
mod tests {
    use hamelin_lib::tree::ast::expression::IntervalUnit;

    use super::*;

    /// Helper to extract the inner typed expression kind from an IRExpression
    fn assert_interval_expr(expr: &IRExpression, expected_millis: i64) {
        match &expr.inner().ast.kind {
            hamelin_lib::tree::ast::expression::ExpressionKind::IntervalLiteral(lit) => {
                assert_eq!(lit.unit, IntervalUnit::Millisecond);
                assert_eq!(lit.value, expected_millis);
            }
            other => panic!("Expected IntervalLiteral, got {:?}", other),
        }
    }

    fn assert_calendar_interval_expr(expr: &IRExpression, expected_months: i64) {
        match &expr.inner().ast.kind {
            hamelin_lib::tree::ast::expression::ExpressionKind::IntervalLiteral(lit) => {
                assert_eq!(lit.unit, IntervalUnit::Month);
                assert_eq!(lit.value, expected_months);
            }
            other => panic!("Expected IntervalLiteral (month), got {:?}", other),
        }
    }

    fn assert_int_expr(expr: &IRExpression, expected: i64) {
        match &expr.inner().ast.kind {
            hamelin_lib::tree::ast::expression::ExpressionKind::IntLiteral(lit) => {
                assert_eq!(lit.int, expected);
            }
            other => panic!("Expected IntLiteral, got {:?}", other),
        }
    }

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

        match frame {
            WindowFrame::Range { start, end } => {
                match &start {
                    RangeBound::Preceding(expr) => {
                        assert_interval_expr(expr, 2 * 3600 * 1_000);
                    }
                    other => panic!("Expected Preceding, got {:?}", other),
                }
                assert!(matches!(end, RangeBound::CurrentRow));
            }
            _ => panic!("Expected Range 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::Range { start, end } => {
                assert!(matches!(start, RangeBound::CurrentRow));
                match &end {
                    RangeBound::Following(expr) => {
                        assert_interval_expr(expr, 3600 * 1_000);
                    }
                    other => panic!("Expected Following, got {:?}", other),
                }
            }
            _ => panic!("Expected Range frame"),
        }
    }

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

        match frame {
            WindowFrame::Range { start, end } => {
                assert!(matches!(start, RangeBound::CurrentRow));
                assert!(matches!(end, RangeBound::CurrentRow));
            }
            _ => panic!("Expected Range 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::Range { start, end } => {
                match &start {
                    RangeBound::Preceding(expr) => {
                        assert_interval_expr(expr, 2 * 3600 * 1_000);
                    }
                    other => panic!("Expected Preceding, got {:?}", other),
                }
                match &end {
                    RangeBound::Preceding(expr) => {
                        assert_interval_expr(expr, 3600 * 1_000);
                    }
                    other => panic!("Expected Preceding, got {:?}", other),
                }
            }
            _ => panic!("Expected Range frame"),
        }
    }

    #[test]
    fn test_unbounded_preceding_to_zero() {
        // ..0 -> UNBOUNDED PRECEDING to CURRENT ROW
        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::Range { start, end } => {
                assert!(matches!(start, RangeBound::Unbounded));
                assert!(matches!(end, RangeBound::CurrentRow));
            }
            _ => panic!("Expected Range 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"),
        }
    }

    #[test]
    fn test_negative_int_to_frame() {
        let v = Value::Int(-3);
        let frame = WindowFrame::from_value(v).unwrap();

        match frame {
            WindowFrame::Range { start, end } => {
                match &start {
                    RangeBound::Preceding(expr) => assert_int_expr(expr, 3),
                    other => panic!("Expected Preceding, got {:?}", other),
                }
                assert!(matches!(end, RangeBound::CurrentRow));
            }
            _ => panic!("Expected Range frame"),
        }
    }

    #[test]
    fn test_calendar_interval_to_frame() {
        let v = Value::CalendarInterval(-2);
        let frame = WindowFrame::from_value(v).unwrap();

        match frame {
            WindowFrame::Range { start, end } => {
                match &start {
                    RangeBound::Preceding(expr) => {
                        assert_calendar_interval_expr(expr, 2);
                    }
                    other => panic!("Expected Preceding, got {:?}", other),
                }
                assert!(matches!(end, RangeBound::CurrentRow));
            }
            _ => panic!("Expected Range frame"),
        }
    }
}