fraiseql_core/runtime/

window_parser.rs

//! Window Query Parser
//!
//! Parses GraphQL window queries into `WindowRequest` for execution.
//!
//! # GraphQL Query Format
//!
//! ```graphql
//! query {
//!   sales_window(
//!     where: { customer_id_eq: "uuid-123" }
//!     orderBy: { occurred_at: ASC }
//!     limit: 100
//!   ) {
//!     revenue
//!     category
//!     rank: row_number(partitionBy: ["category"], orderBy: { revenue: DESC })
//!     running_total: sum(field: "revenue", orderBy: { occurred_at: ASC })
//!     prev_revenue: lag(field: "revenue", offset: 1, default: 0)
//!   }
//! }
//! ```
//!
//! # JSON Query Format
//!
//! ```json
//! {
//!   "table": "tf_sales",
//!   "select": [
//!     {"type": "measure", "name": "revenue", "alias": "revenue"},
//!     {"type": "dimension", "path": "category", "alias": "category"}
//!   ],
//!   "windows": [
//!     {
//!       "function": {"type": "row_number"},
//!       "alias": "rank",
//!       "partitionBy": [{"type": "dimension", "path": "category"}],
//!       "orderBy": [{"field": "revenue", "direction": "DESC"}]
//!     },
//!     {
//!       "function": {"type": "running_sum", "measure": "revenue"},
//!       "alias": "running_total",
//!       "orderBy": [{"field": "occurred_at", "direction": "ASC"}],
//!       "frame": {"frame_type": "ROWS", "start": {"type": "unbounded_preceding"}, "end": {"type": "current_row"}}
//!     }
//!   ],
//!   "orderBy": [{"field": "occurred_at", "direction": "ASC"}],
//!   "limit": 100
//! }
//! ```

use serde_json::Value;

use super::aggregate_parser::split_where_key;
use crate::{
    backend::where_clause::WhereClause,
    compiler::{
        aggregation::OrderDirection,
        fact_table::FactTableMetadata,
        window_functions::{
            FrameBoundary, FrameExclusion, FrameType, PartitionByColumn, WindowFrame,
            WindowFunctionRequest, WindowFunctionSpec, WindowOrderBy, WindowRequest,
            WindowSelectColumn,
        },
    },
    error::{FraiseQLError, Result},
};

/// Window query parser
pub struct WindowQueryParser;

impl WindowQueryParser {
    /// Parse a window query JSON into `WindowRequest`.
    ///
    /// # Arguments
    ///
    /// * `query_json` - JSON representation of the window query
    /// * `metadata` - Fact table metadata: its declared fields settle an ambiguous `where` key
    ///
    /// # Errors
    ///
    /// Returns error if the query structure is invalid.
    pub fn parse(query_json: &Value, metadata: &FactTableMetadata) -> Result<WindowRequest> {
        // Extract table name
        let table_name = query_json
            .get("table")
            .and_then(|v| v.as_str())
            .ok_or_else(|| FraiseQLError::Validation {
                message: "Missing 'table' field in window query".to_string(),
                path:    None,
            })?
            .to_string();

        // Parse SELECT columns
        let select = if let Some(select_array) = query_json.get("select") {
            Self::parse_select_columns(select_array)?
        } else {
            vec![]
        };

        // Parse window functions
        let windows = if let Some(windows_array) = query_json.get("windows") {
            Self::parse_window_functions(windows_array)?
        } else {
            vec![]
        };

        // Parse WHERE clause
        let where_clause = if let Some(where_obj) = query_json.get("where") {
            Some(Self::parse_where_clause(where_obj, metadata)?)
        } else {
            None
        };

        // Parse final ORDER BY
        let order_by = if let Some(order_array) = query_json.get("orderBy") {
            Self::parse_order_by(order_array)?
        } else {
            vec![]
        };

        // Parse LIMIT/OFFSET
        let limit = query_json
            .get("limit")
            .and_then(|v| v.as_u64())
            .map(|n| u32::try_from(n).unwrap_or(u32::MAX));

        let offset = query_json
            .get("offset")
            .and_then(|v| v.as_u64())
            .map(|n| u32::try_from(n).unwrap_or(u32::MAX));

        Ok(WindowRequest {
            table_name,
            select,
            windows,
            where_clause,
            order_by,
            limit,
            offset,
        })
    }

    /// Parse SELECT columns from JSON array.
    fn parse_select_columns(select_array: &Value) -> Result<Vec<WindowSelectColumn>> {
        let Some(arr) = select_array.as_array() else {
            return Ok(vec![]);
        };

        arr.iter().map(Self::parse_single_select_column).collect()
    }

    fn parse_single_select_column(col: &Value) -> Result<WindowSelectColumn> {
        let col_type =
            col.get("type")
                .and_then(|v| v.as_str())
                .ok_or_else(|| FraiseQLError::Validation {
                    message: "Missing 'type' in select column".to_string(),
                    path:    None,
                })?;

        let alias = col
            .get("alias")
            .and_then(|v| v.as_str())
            .ok_or_else(|| FraiseQLError::Validation {
                message: "Missing 'alias' in select column".to_string(),
                path:    None,
            })?
            .to_string();

        match col_type {
            "measure" => {
                let name = col
                    .get("name")
                    .and_then(|v| v.as_str())
                    .ok_or_else(|| FraiseQLError::Validation {
                        message: "Missing 'name' in measure select column".to_string(),
                        path:    None,
                    })?
                    .to_string();
                Ok(WindowSelectColumn::Measure { name, alias })
            },
            "dimension" => {
                let path = col
                    .get("path")
                    .and_then(|v| v.as_str())
                    .ok_or_else(|| FraiseQLError::Validation {
                        message: "Missing 'path' in dimension select column".to_string(),
                        path:    None,
                    })?
                    .to_string();
                Ok(WindowSelectColumn::Dimension { path, alias })
            },
            "filter" => {
                let name = col
                    .get("name")
                    .and_then(|v| v.as_str())
                    .ok_or_else(|| FraiseQLError::Validation {
                        message: "Missing 'name' in filter select column".to_string(),
                        path:    None,
                    })?
                    .to_string();
                Ok(WindowSelectColumn::Filter { name, alias })
            },
            _ => Err(FraiseQLError::Validation {
                message: format!("Unknown select column type: {col_type}"),
                path:    None,
            }),
        }
    }

    /// Parse window functions from JSON array.
    fn parse_window_functions(windows_array: &Value) -> Result<Vec<WindowFunctionRequest>> {
        let Some(arr) = windows_array.as_array() else {
            return Ok(vec![]);
        };

        arr.iter().map(Self::parse_single_window_function).collect()
    }

    fn parse_single_window_function(window: &Value) -> Result<WindowFunctionRequest> {
        // Parse function spec
        let function = window
            .get("function")
            .ok_or_else(|| FraiseQLError::Validation {
                message: "Missing 'function' in window definition".to_string(),
                path:    None,
            })
            .and_then(Self::parse_function_spec)?;

        // Parse alias
        let alias = window
            .get("alias")
            .and_then(|v| v.as_str())
            .ok_or_else(|| FraiseQLError::Validation {
                message: "Missing 'alias' in window definition".to_string(),
                path:    None,
            })?
            .to_string();

        // Parse PARTITION BY
        let partition_by = if let Some(partition_array) = window.get("partitionBy") {
            Self::parse_partition_by(partition_array)?
        } else {
            vec![]
        };

        // Parse ORDER BY within window
        let order_by = if let Some(order_array) = window.get("orderBy") {
            Self::parse_order_by(order_array)?
        } else {
            vec![]
        };

        // Parse frame
        let frame = window.get("frame").map(Self::parse_frame).transpose()?;

        Ok(WindowFunctionRequest {
            function,
            alias,
            partition_by,
            order_by,
            frame,
        })
    }

    /// Parse window function specification.
    fn parse_function_spec(func: &Value) -> Result<WindowFunctionSpec> {
        let func_type =
            func.get("type")
                .and_then(|v| v.as_str())
                .ok_or_else(|| FraiseQLError::Validation {
                    message: "Missing 'type' in function spec".to_string(),
                    path:    None,
                })?;

        match func_type {
            // Ranking functions
            "row_number" => Ok(WindowFunctionSpec::RowNumber),
            "rank" => Ok(WindowFunctionSpec::Rank),
            "dense_rank" => Ok(WindowFunctionSpec::DenseRank),
            "ntile" => {
                let n = u32::try_from(func.get("n").and_then(|v| v.as_u64()).ok_or_else(|| {
                    FraiseQLError::Validation {
                        message: "Missing 'n' in NTILE function".to_string(),
                        path:    None,
                    }
                })?)
                .unwrap_or(u32::MAX);
                Ok(WindowFunctionSpec::Ntile { n })
            },
            "percent_rank" => Ok(WindowFunctionSpec::PercentRank),
            "cume_dist" => Ok(WindowFunctionSpec::CumeDist),

            // Value functions
            "lag" => {
                let field = Self::extract_string_field(func, "field")?;
                let offset =
                    i32::try_from(func.get("offset").and_then(|v| v.as_i64()).unwrap_or(1))
                        .unwrap_or(1);
                let default = func.get("default").cloned();
                Ok(WindowFunctionSpec::Lag {
                    field,
                    offset,
                    default,
                })
            },
            "lead" => {
                let field = Self::extract_string_field(func, "field")?;
                let offset =
                    i32::try_from(func.get("offset").and_then(|v| v.as_i64()).unwrap_or(1))
                        .unwrap_or(1);
                let default = func.get("default").cloned();
                Ok(WindowFunctionSpec::Lead {
                    field,
                    offset,
                    default,
                })
            },
            "first_value" => {
                let field = Self::extract_string_field(func, "field")?;
                Ok(WindowFunctionSpec::FirstValue { field })
            },
            "last_value" => {
                let field = Self::extract_string_field(func, "field")?;
                Ok(WindowFunctionSpec::LastValue { field })
            },
            "nth_value" => {
                let field = Self::extract_string_field(func, "field")?;
                let n = u32::try_from(func.get("n").and_then(|v| v.as_u64()).ok_or_else(|| {
                    FraiseQLError::Validation {
                        message: "Missing 'n' in NTH_VALUE function".to_string(),
                        path:    None,
                    }
                })?)
                .unwrap_or(u32::MAX);
                Ok(WindowFunctionSpec::NthValue { field, n })
            },

            // Aggregate as window functions
            "running_sum" => {
                let measure = Self::extract_string_field(func, "measure")?;
                Ok(WindowFunctionSpec::RunningSum { measure })
            },
            "running_avg" => {
                let measure = Self::extract_string_field(func, "measure")?;
                Ok(WindowFunctionSpec::RunningAvg { measure })
            },
            "running_count" => {
                if let Some(field) = func.get("field").and_then(|v| v.as_str()) {
                    Ok(WindowFunctionSpec::RunningCountField {
                        field: field.to_string(),
                    })
                } else {
                    Ok(WindowFunctionSpec::RunningCount)
                }
            },
            "running_min" => {
                let measure = Self::extract_string_field(func, "measure")?;
                Ok(WindowFunctionSpec::RunningMin { measure })
            },
            "running_max" => {
                let measure = Self::extract_string_field(func, "measure")?;
                Ok(WindowFunctionSpec::RunningMax { measure })
            },
            "running_stddev" => {
                let measure = Self::extract_string_field(func, "measure")?;
                Ok(WindowFunctionSpec::RunningStddev { measure })
            },
            "running_variance" => {
                let measure = Self::extract_string_field(func, "measure")?;
                Ok(WindowFunctionSpec::RunningVariance { measure })
            },

            _ => Err(FraiseQLError::Validation {
                message: format!("Unknown window function type: {func_type}"),
                path:    None,
            }),
        }
    }

    /// Extract a required string field from JSON object.
    fn extract_string_field(obj: &Value, field_name: &str) -> Result<String> {
        obj.get(field_name).and_then(|v| v.as_str()).map(String::from).ok_or_else(|| {
            FraiseQLError::Validation {
                message: format!("Missing '{field_name}' in function spec"),
                path:    None,
            }
        })
    }

    /// Parse PARTITION BY from JSON array.
    fn parse_partition_by(partition_array: &Value) -> Result<Vec<PartitionByColumn>> {
        let Some(arr) = partition_array.as_array() else {
            return Ok(vec![]);
        };

        arr.iter()
            .map(|item| {
                let col_type = item.get("type").and_then(|v| v.as_str()).ok_or_else(|| {
                    FraiseQLError::Validation {
                        message: "Missing 'type' in partitionBy column".to_string(),
                        path:    None,
                    }
                })?;

                match col_type {
                    "dimension" => {
                        let path = item
                            .get("path")
                            .and_then(|v| v.as_str())
                            .ok_or_else(|| FraiseQLError::Validation {
                                message: "Missing 'path' in dimension partition column".to_string(),
                                path:    None,
                            })?
                            .to_string();
                        Ok(PartitionByColumn::Dimension { path })
                    },
                    "filter" => {
                        let name = item
                            .get("name")
                            .and_then(|v| v.as_str())
                            .ok_or_else(|| FraiseQLError::Validation {
                                message: "Missing 'name' in filter partition column".to_string(),
                                path:    None,
                            })?
                            .to_string();
                        Ok(PartitionByColumn::Filter { name })
                    },
                    "measure" => {
                        let name = item
                            .get("name")
                            .and_then(|v| v.as_str())
                            .ok_or_else(|| FraiseQLError::Validation {
                                message: "Missing 'name' in measure partition column".to_string(),
                                path:    None,
                            })?
                            .to_string();
                        Ok(PartitionByColumn::Measure { name })
                    },
                    _ => Err(FraiseQLError::Validation {
                        message: format!("Unknown partition column type: {col_type}"),
                        path:    None,
                    }),
                }
            })
            .collect()
    }

    /// Parse ORDER BY from JSON array.
    fn parse_order_by(order_array: &Value) -> Result<Vec<WindowOrderBy>> {
        let Some(arr) = order_array.as_array() else {
            return Ok(vec![]);
        };

        arr.iter()
            .map(|item| {
                let field = item
                    .get("field")
                    .and_then(|v| v.as_str())
                    .ok_or_else(|| FraiseQLError::Validation {
                        message: "Missing 'field' in orderBy".to_string(),
                        path:    None,
                    })?
                    .to_string();

                let direction = match item.get("direction").and_then(|v| v.as_str()) {
                    Some("DESC" | "desc") => OrderDirection::Desc,
                    _ => OrderDirection::Asc,
                };

                Ok(WindowOrderBy { field, direction })
            })
            .collect()
    }

    /// Parse WHERE clause from JSON: `{ "<field>_<operator>": value, … }`.
    fn parse_where_clause(where_obj: &Value, metadata: &FactTableMetadata) -> Result<WhereClause> {
        let Some(obj) = where_obj.as_object() else {
            return Ok(WhereClause::And(vec![]));
        };

        let no_native_columns = std::collections::HashMap::new();
        let mut conditions = Vec::new();
        for (key, value) in obj {
            let (field, operator) = split_where_key(key, metadata, &no_native_columns)?;
            conditions.push(WhereClause::Field {
                // Recase the JSONB key so a camelCase window filter
                // (`organizationId_eq`) builds `data->>'organization_id'`
                // rather than a never-matching `organizationId` key (#486).
                path: vec![crate::utils::to_snake_case(field)],
                operator,
                value: value.clone(),
            });
        }

        Ok(WhereClause::And(conditions))
    }

    /// Parse window frame from JSON.
    fn parse_frame(frame: &Value) -> Result<WindowFrame> {
        let frame_type = match frame.get("frame_type").and_then(|v| v.as_str()) {
            Some("ROWS") => FrameType::Rows,
            Some("RANGE") => FrameType::Range,
            Some("GROUPS") => FrameType::Groups,
            _ => {
                return Err(FraiseQLError::Validation {
                    message: "Invalid or missing 'frame_type' in frame".to_string(),
                    path:    None,
                });
            },
        };

        let start = frame
            .get("start")
            .ok_or_else(|| FraiseQLError::Validation {
                message: "Missing 'start' in frame".to_string(),
                path:    None,
            })
            .and_then(Self::parse_frame_boundary)?;

        let end = frame
            .get("end")
            .ok_or_else(|| FraiseQLError::Validation {
                message: "Missing 'end' in frame".to_string(),
                path:    None,
            })
            .and_then(Self::parse_frame_boundary)?;

        let exclusion = frame.get("exclusion").and_then(|v| v.as_str()).map(|s| match s {
            "current_row" => FrameExclusion::CurrentRow,
            "group" => FrameExclusion::Group,
            "ties" => FrameExclusion::Ties,
            _ => FrameExclusion::NoOthers,
        });

        Ok(WindowFrame {
            frame_type,
            start,
            end,
            exclusion,
        })
    }

    /// Parse frame boundary from JSON.
    fn parse_frame_boundary(boundary: &Value) -> Result<FrameBoundary> {
        let boundary_type = boundary.get("type").and_then(|v| v.as_str()).ok_or_else(|| {
            FraiseQLError::Validation {
                message: "Missing 'type' in frame boundary".to_string(),
                path:    None,
            }
        })?;

        match boundary_type {
            "unbounded_preceding" => Ok(FrameBoundary::UnboundedPreceding),
            "n_preceding" => {
                let n =
                    u32::try_from(boundary.get("n").and_then(|v| v.as_u64()).ok_or_else(|| {
                        FraiseQLError::Validation {
                            message: "Missing 'n' in N PRECEDING boundary".to_string(),
                            path:    None,
                        }
                    })?)
                    .unwrap_or(u32::MAX);
                Ok(FrameBoundary::NPreceding { n })
            },
            "current_row" => Ok(FrameBoundary::CurrentRow),
            "n_following" => {
                let n =
                    u32::try_from(boundary.get("n").and_then(|v| v.as_u64()).ok_or_else(|| {
                        FraiseQLError::Validation {
                            message: "Missing 'n' in N FOLLOWING boundary".to_string(),
                            path:    None,
                        }
                    })?)
                    .unwrap_or(u32::MAX);
                Ok(FrameBoundary::NFollowing { n })
            },
            "unbounded_following" => Ok(FrameBoundary::UnboundedFollowing),
            _ => Err(FraiseQLError::Validation {
                message: format!("Unknown frame boundary type: {boundary_type}"),
                path:    None,
            }),
        }
    }
}