fraiseql_core/compiler/fact_table/

detector.rs

use super::{
    CalendarBucket, CalendarDimension, CalendarGranularity, DatabaseIntrospector, DatabaseType,
    DimensionColumn, DimensionPath, FactTableMetadata, FilterColumn, MeasureColumn, SqlType,
};
use crate::error::{FraiseQLError, Result};

/// Detects and introspects fact tables
pub struct FactTableDetector;

impl FactTableDetector {
    /// Detect if a table name follows the fact table pattern.
    ///
    /// Fact tables must follow the naming convention: `tf_<table_name>`
    /// where the table name contains only lowercase letters, numbers, and underscores.
    ///
    /// # Arguments
    ///
    /// * `table_name` - Table name to check
    ///
    /// # Returns
    ///
    /// `true` if the table name starts with `tf_` and follows naming conventions,
    /// `false` otherwise.
    ///
    /// # Notes
    ///
    /// - The check is strict: `tf_` is required as a prefix
    /// - Table names like `TF_sales` (uppercase prefix) are NOT recognized as fact tables
    /// - Empty strings and tables named just `tf_` without additional suffix are not valid
    ///
    /// # Examples
    ///
    /// ```
    /// use fraiseql_core::compiler::fact_table::FactTableDetector;
    ///
    /// assert!(FactTableDetector::is_fact_table("tf_sales"));
    /// assert!(FactTableDetector::is_fact_table("tf_events"));
    /// assert!(FactTableDetector::is_fact_table("tf_page_views_daily"));
    /// assert!(!FactTableDetector::is_fact_table("ta_sales_by_day"));
    /// assert!(!FactTableDetector::is_fact_table("v_user"));
    /// assert!(!FactTableDetector::is_fact_table("TF_sales")); // uppercase prefix not recognized
    /// assert!(!FactTableDetector::is_fact_table("tf_")); // incomplete name
    /// ```
    #[must_use]
    pub fn is_fact_table(table_name: &str) -> bool {
        // Must start with "tf_" and have at least one more character
        table_name.len() > 3 && table_name.starts_with("tf_")
    }

    /// Introspect a fact table from the database
    ///
    /// Queries the database schema to extract:
    /// - Measures (numeric columns)
    /// - Dimensions (JSONB/JSON columns)
    /// - Denormalized filters (indexed columns)
    ///
    /// # Arguments
    ///
    /// * `introspector` - Database introspection implementation
    /// * `table_name` - Fact table name (must start with "tf_")
    ///
    /// # Errors
    ///
    /// Returns error if:
    /// - Table is not a fact table (doesn't start with "tf_")
    /// - Database query fails
    /// - Table structure is invalid
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use fraiseql_core::compiler::fact_table::{FactTableDetector, DatabaseIntrospector};
    ///
    /// # async fn example(db: impl DatabaseIntrospector) -> Result<(), Box<dyn std::error::Error>> {
    /// let metadata = FactTableDetector::introspect(&db, "tf_sales").await?;
    /// println!("Found {} measures", metadata.measures.len());
    /// # Ok(())
    /// # }
    /// ```
    pub async fn introspect(
        introspector: &impl DatabaseIntrospector,
        table_name: &str,
    ) -> Result<FactTableMetadata> {
        // Validate table name follows fact table pattern
        if !Self::is_fact_table(table_name) {
            return Err(FraiseQLError::Validation {
                message: format!(
                    "Table '{}' is not a fact table (must start with 'tf_')",
                    table_name
                ),
                path:    None,
            });
        }

        // Query column information
        let columns = introspector.get_columns(table_name).await?;
        if columns.is_empty() {
            return Err(FraiseQLError::Validation {
                message: format!("Table '{}' not found or has no columns", table_name),
                path:    None,
            });
        }

        // Query indexed columns
        let indexed_columns = introspector.get_indexed_columns(table_name).await?;
        let indexed_set: std::collections::HashSet<String> = indexed_columns.into_iter().collect();

        // Parse SQL types based on database
        let db_type = introspector.database_type();

        let mut measures = Vec::new();
        let mut dimension_candidates: Vec<String> = Vec::new();
        let mut filters = Vec::new();

        for (name, data_type, is_nullable) in &columns {
            let sql_type = Self::parse_sql_type(data_type, db_type);

            match sql_type {
                SqlType::Jsonb | SqlType::Json => {
                    // A dimension-column CANDIDATE. The choice is made by role
                    // after the scan (#825): the old code assigned here
                    // unconditionally, so with several JSONB columns the last
                    // one in ordinal order silently won — guaranteed wrong for
                    // the documented calendar pattern, whose `*_info` columns
                    // come after the real dimensions column.
                    dimension_candidates.push(name.clone());
                },
                SqlType::Int | SqlType::BigInt | SqlType::Decimal | SqlType::Float => {
                    // Skip common non-measure columns
                    if name != "id" && !name.ends_with("_id") {
                        measures.push(MeasureColumn {
                            name:     name.clone(),
                            sql_type: sql_type.clone(),
                            nullable: *is_nullable,
                        });
                    }

                    // A numeric `*_id` column is a denormalized filter whether
                    // or not it is indexed — dropping the unindexed ones made
                    // them vanish from the metadata entirely (#825 secondary).
                    if name.ends_with("_id") {
                        filters.push(FilterColumn {
                            name:      name.clone(),
                            sql_type:  sql_type.clone(),
                            indexed:   indexed_set.contains(name.as_str()),
                            hierarchy: None,
                        });
                    }
                },
                _ => {
                    // Other types might be denormalized filters
                    if name != "id"
                        && name != "created_at"
                        && name != "updated_at"
                        && name != "occurred_at"
                    {
                        filters.push(FilterColumn {
                            name: name.clone(),
                            sql_type,
                            indexed: indexed_set.contains(name.as_str()),
                            hierarchy: None,
                        });
                    } else if (name == "occurred_at" || name == "created_at")
                        && indexed_set.contains(name.as_str())
                    {
                        // Timestamp columns are important filters if indexed
                        filters.push(FilterColumn {
                            name: name.clone(),
                            sql_type,
                            indexed: true,
                            hierarchy: None,
                        });
                    }
                },
            }
        }

        // Detect calendar dimensions
        let calendar_dimensions = Self::detect_calendar_dimensions(&columns, &indexed_set)?;

        // Choose the dimensions column by role, then sample only the chosen one.
        let dimension_column = match Self::choose_dimension_column(&dimension_candidates)? {
            None => None,
            Some(name) => {
                let paths = if let Ok(Some(sample)) =
                    introspector.get_sample_jsonb(table_name, &name).await
                {
                    Self::extract_dimension_paths(&sample, &name, db_type)
                } else {
                    Vec::new()
                };
                Some(DimensionColumn { name, paths })
            },
        };

        let metadata = FactTableMetadata {
            table_name: table_name.to_string(),
            type_name: None,
            measures,
            dimensions: dimension_column.unwrap_or(DimensionColumn {
                name:  "dimensions".to_string(),
                paths: Vec::new(),
            }),
            denormalized_filters: filters,
            calendar_dimensions,
            partial_period: None,
            native_measures: std::collections::HashMap::new(),
            native_dimension_mapping: std::collections::HashMap::new(),
        };

        Self::validate(&metadata)?;
        Ok(metadata)
    }

    /// Parse SQL type string to `SqlType` enum
    fn parse_sql_type(type_name: &str, db_type: DatabaseType) -> SqlType {
        match db_type {
            DatabaseType::PostgreSQL => SqlType::from_str_postgres(type_name),
        }
    }

    /// Choose the dimensions column from the JSONB/JSON candidates **by role,
    /// never by position** (#825).
    ///
    /// `*_info` columns belong to the calendar-dimension pattern
    /// ([`Self::detect_calendar_dimensions`]) and are excluded. Of what
    /// remains: a single candidate wins; with several, a conventional name
    /// (`data`, then `dimensions`) wins; several unconventional candidates are
    /// a reported ambiguity rather than a silent order-dependent pick.
    ///
    /// # Errors
    ///
    /// Returns `FraiseQLError::Validation` naming the candidates when the
    /// choice is ambiguous.
    fn choose_dimension_column(candidates: &[String]) -> Result<Option<String>> {
        let non_calendar: Vec<&String> =
            candidates.iter().filter(|n| !n.ends_with("_info")).collect();
        match non_calendar.as_slice() {
            [] => Ok(None),
            [only] => Ok(Some((*only).clone())),
            several => {
                for conventional in ["data", "dimensions"] {
                    if let Some(name) = several.iter().find(|n| n.as_str() == conventional) {
                        return Ok(Some((*name).clone()));
                    }
                }
                let names = several.iter().map(|n| n.as_str()).collect::<Vec<_>>().join(", ");
                Err(FraiseQLError::Validation {
                    message: format!(
                        "ambiguous dimension columns: several JSONB columns qualify ({names}) \
                         and none is named 'data' or 'dimensions' — rename the dimensions \
                         column to a conventional name"
                    ),
                    path:    None,
                })
            },
        }
    }

    /// Validate fact table structure
    ///
    /// # Errors
    ///
    /// Returns error if:
    /// - No measures found
    /// - No dimension column found
    /// - Measures are not numeric types
    pub fn validate(metadata: &FactTableMetadata) -> Result<()> {
        // Must have at least one measure
        if metadata.measures.is_empty() {
            return Err(FraiseQLError::Validation {
                message: format!(
                    "Fact table '{}' must have at least one measure column",
                    metadata.table_name
                ),
                path:    None,
            });
        }

        // Validate all measures are numeric
        for measure in &metadata.measures {
            if !Self::is_numeric_type(&measure.sql_type) {
                return Err(FraiseQLError::Validation {
                    message: format!(
                        "Measure column '{}' must be numeric type, found {:?}",
                        measure.name, measure.sql_type
                    ),
                    path:    None,
                });
            }
        }

        // Must have dimension column
        if metadata.dimensions.name.is_empty() {
            return Err(FraiseQLError::Validation {
                message: format!(
                    "Fact table '{}' must have a dimension column (JSONB)",
                    metadata.table_name
                ),
                path:    None,
            });
        }

        Ok(())
    }

    /// Check if SQL type is numeric (suitable for aggregation)
    pub(super) const fn is_numeric_type(sql_type: &SqlType) -> bool {
        matches!(sql_type, SqlType::Int | SqlType::BigInt | SqlType::Decimal | SqlType::Float)
    }

    /// Extract dimension paths from a sample JSON value
    ///
    /// Walks through the JSON structure and extracts top-level keys as dimension paths.
    /// Nested objects are represented with dot notation (e.g., "customer.region").
    ///
    /// # Arguments
    ///
    /// * `sample` - Sample JSON value from the dimension column
    /// * `column_name` - Name of the JSONB column (e.g., "dimensions")
    /// * `db_type` - Database type for generating correct JSON path syntax
    ///
    /// # Returns
    ///
    /// Vec of `DimensionPath` extracted from the sample data
    #[must_use]
    pub fn extract_dimension_paths(
        sample: &serde_json::Value,
        column_name: &str,
        db_type: DatabaseType,
    ) -> Vec<DimensionPath> {
        let mut paths = Vec::new();
        Self::extract_paths_recursive(sample, column_name, "", &mut paths, db_type, 0);
        paths
    }

    /// Recursively extract paths from JSON structure
    fn extract_paths_recursive(
        value: &serde_json::Value,
        column_name: &str,
        prefix: &str,
        paths: &mut Vec<DimensionPath>,
        db_type: DatabaseType,
        depth: usize,
    ) {
        // Limit depth to avoid infinite recursion on deeply nested structures
        if depth > 3 {
            return;
        }

        if let Some(obj) = value.as_object() {
            for (key, val) in obj {
                let full_path = if prefix.is_empty() {
                    key.clone()
                } else {
                    format!("{}.{}", prefix, key)
                };

                // Determine data type from the value
                let data_type = Self::infer_json_type(val);

                // Generate database-specific JSON path syntax
                let json_path = Self::generate_json_path(column_name, &full_path, db_type);

                paths.push(DimensionPath {
                    name: full_path.replace('.', "_"), /* Convert dots to underscores for field
                                                        * names */
                    json_path,
                    data_type,
                });

                // Recurse into nested objects
                if val.is_object() {
                    Self::extract_paths_recursive(
                        val,
                        column_name,
                        &full_path,
                        paths,
                        db_type,
                        depth + 1,
                    );
                }
            }
        }
    }

    /// Infer JSON data type from a value
    pub(super) fn infer_json_type(value: &serde_json::Value) -> String {
        match value {
            serde_json::Value::Bool(_) => "boolean".to_string(),
            serde_json::Value::Number(n) => {
                if n.is_i64() || n.is_u64() {
                    "integer".to_string()
                } else {
                    "float".to_string()
                }
            },
            serde_json::Value::Array(_) => "array".to_string(),
            serde_json::Value::Object(_) => "object".to_string(),
            // Null and String both infer as "string" type
            serde_json::Value::Null | serde_json::Value::String(_) => "string".to_string(),
        }
    }

    /// Generate database-specific JSON path syntax
    pub(super) fn generate_json_path(
        column_name: &str,
        path: &str,
        db_type: DatabaseType,
    ) -> String {
        let parts: Vec<&str> = path.split('.').collect();

        match db_type {
            DatabaseType::PostgreSQL => {
                // PostgreSQL: column->>'key' for top-level, column->'nested'->>'key' for nested
                if parts.is_empty() {
                    // Safety: handle empty path by returning raw column name
                    column_name.to_string()
                } else if parts.len() == 1 {
                    format!("{}->>'{}'", column_name, parts[0])
                } else {
                    // Safe: parts.len() >= 2 is guaranteed here
                    if let Some(last) = parts.last() {
                        let rest = &parts[..parts.len() - 1];
                        let nav = rest.iter().fold(String::new(), |mut acc, p| {
                            use std::fmt::Write;
                            let _ = write!(acc, "->'{}'", p);
                            acc
                        });
                        format!("{}{}->>'{}'", column_name, nav, last)
                    } else {
                        // This branch is unreachable due to length check, but safe fallback
                        column_name.to_string()
                    }
                }
            },
        }
    }

    /// Detect calendar dimension columns (`date_info`, `week_info`, etc.)
    ///
    /// Looks for `*_info` JSONB/JSON columns following the calendar dimension pattern.
    /// Returns calendar dimension metadata if calendar columns are found.
    ///
    /// # Arguments
    ///
    /// * `columns` - List of (name, `data_type`, nullable) tuples
    /// * `_indexed_set` - Set of indexed columns (unused, for future optimization detection)
    ///
    /// # Returns
    ///
    /// Vec of calendar dimensions (empty if none found)
    pub(super) fn detect_calendar_dimensions(
        columns: &[(String, String, bool)],
        _indexed_set: &std::collections::HashSet<String>,
    ) -> Result<Vec<CalendarDimension>> {
        // Look for *_info columns with JSONB/JSON type
        let calendar_columns: Vec<String> = columns
            .iter()
            .filter(|(name, data_type, _)| {
                name.ends_with("_info")
                    && (data_type.to_lowercase().contains("json")
                        || data_type.to_lowercase().contains("jsonb"))
            })
            .map(|(name, _, _)| name.clone())
            .collect();

        if calendar_columns.is_empty() {
            return Ok(Vec::new());
        }

        // Build granularities based on calendar dimension pattern
        let mut granularities = Vec::new();
        for col_name in calendar_columns {
            let buckets = Self::infer_calendar_buckets(&col_name);
            if !buckets.is_empty() {
                granularities.push(CalendarGranularity {
                    column_name: col_name,
                    buckets,
                });
            }
        }

        if granularities.is_empty() {
            return Ok(Vec::new());
        }

        // Assume single source column "occurred_at"
        // (could be enhanced to detect from schema later)
        Ok(vec![CalendarDimension {
            source_column: "occurred_at".to_string(),
            granularities,
        }])
    }

    /// Map calendar column names to available buckets (standard pattern)
    ///
    /// # Arguments
    ///
    /// * `column_name` - Name of the calendar column (e.g., "`date_info`", "`month_info`")
    ///
    /// # Returns
    ///
    /// Vec of calendar buckets available in this column
    pub(super) fn infer_calendar_buckets(column_name: &str) -> Vec<CalendarBucket> {
        use crate::compiler::aggregate_types::TemporalBucket;

        match column_name {
            "date_info" => vec![
                CalendarBucket {
                    json_key:    "date".to_string(),
                    bucket_type: TemporalBucket::Day,
                    data_type:   "date".to_string(),
                },
                CalendarBucket {
                    json_key:    "week".to_string(),
                    bucket_type: TemporalBucket::Week,
                    data_type:   "integer".to_string(),
                },
                CalendarBucket {
                    json_key:    "month".to_string(),
                    bucket_type: TemporalBucket::Month,
                    data_type:   "integer".to_string(),
                },
                CalendarBucket {
                    json_key:    "quarter".to_string(),
                    bucket_type: TemporalBucket::Quarter,
                    data_type:   "integer".to_string(),
                },
                CalendarBucket {
                    json_key:    "year".to_string(),
                    bucket_type: TemporalBucket::Year,
                    data_type:   "integer".to_string(),
                },
            ],
            "week_info" => vec![
                CalendarBucket {
                    json_key:    "week".to_string(),
                    bucket_type: TemporalBucket::Week,
                    data_type:   "integer".to_string(),
                },
                CalendarBucket {
                    json_key:    "month".to_string(),
                    bucket_type: TemporalBucket::Month,
                    data_type:   "integer".to_string(),
                },
                CalendarBucket {
                    json_key:    "quarter".to_string(),
                    bucket_type: TemporalBucket::Quarter,
                    data_type:   "integer".to_string(),
                },
                CalendarBucket {
                    json_key:    "year".to_string(),
                    bucket_type: TemporalBucket::Year,
                    data_type:   "integer".to_string(),
                },
            ],
            "month_info" => vec![
                CalendarBucket {
                    json_key:    "month".to_string(),
                    bucket_type: TemporalBucket::Month,
                    data_type:   "integer".to_string(),
                },
                CalendarBucket {
                    json_key:    "quarter".to_string(),
                    bucket_type: TemporalBucket::Quarter,
                    data_type:   "integer".to_string(),
                },
                CalendarBucket {
                    json_key:    "year".to_string(),
                    bucket_type: TemporalBucket::Year,
                    data_type:   "integer".to_string(),
                },
            ],
            "quarter_info" => vec![
                CalendarBucket {
                    json_key:    "quarter".to_string(),
                    bucket_type: TemporalBucket::Quarter,
                    data_type:   "integer".to_string(),
                },
                CalendarBucket {
                    json_key:    "year".to_string(),
                    bucket_type: TemporalBucket::Year,
                    data_type:   "integer".to_string(),
                },
            ],
            "semester_info" => vec![
                CalendarBucket {
                    json_key:    "semester".to_string(),
                    bucket_type: TemporalBucket::Quarter, // Map to Quarter for now
                    data_type:   "integer".to_string(),
                },
                CalendarBucket {
                    json_key:    "year".to_string(),
                    bucket_type: TemporalBucket::Year,
                    data_type:   "integer".to_string(),
                },
            ],
            "year_info" => vec![CalendarBucket {
                json_key:    "year".to_string(),
                bucket_type: TemporalBucket::Year,
                data_type:   "integer".to_string(),
            }],
            _ => Vec::new(),
        }
    }

    /// Create metadata from column definitions (for testing)
    ///
    /// # Errors
    ///
    /// Returns `FraiseQLError::Validation` if no time dimension column is found.
    pub fn from_columns(
        table_name: String,
        columns: Vec<(&str, SqlType, bool)>,
    ) -> Result<FactTableMetadata> {
        let mut measures = Vec::new();
        let mut dimension_candidates: Vec<String> = Vec::new();
        let mut filters = Vec::new();

        for (name, sql_type, nullable) in columns {
            match sql_type {
                SqlType::Jsonb | SqlType::Json => {
                    // Dimension-column candidate; chosen by role below (#825).
                    dimension_candidates.push(name.to_string());
                },
                SqlType::Int | SqlType::BigInt | SqlType::Decimal | SqlType::Float => {
                    // Skip id column
                    if name != "id" && !name.ends_with("_id") {
                        // This is a measure
                        measures.push(MeasureColumn {
                            name: name.to_string(),
                            sql_type,
                            nullable,
                        });
                    } else if name != "id" {
                        // This is a filter (_id columns)
                        filters.push(FilterColumn {
                            name: name.to_string(),
                            sql_type,
                            indexed: false,
                            hierarchy: None,
                        });
                    }
                },
                _ => {
                    // This might be a filter column (if not id/created_at/updated_at)
                    if name != "id" && name != "created_at" && name != "updated_at" {
                        filters.push(FilterColumn {
                            name: name.to_string(),
                            sql_type,
                            indexed: false, // Would need to query indexes to determine,
                            hierarchy: None,
                        });
                    }
                },
            }
        }

        let dimensions = Self::choose_dimension_column(&dimension_candidates)?.map_or_else(
            || DimensionColumn {
                name:  "dimensions".to_string(),
                paths: Vec::new(),
            },
            |name| DimensionColumn {
                name,
                paths: Vec::new(),
            },
        );

        let metadata = FactTableMetadata {
            table_name,
            type_name: None,
            measures,
            dimensions,
            denormalized_filters: filters,
            calendar_dimensions: Vec::new(), // No calendar detection in test helper
            partial_period: None,
            native_measures: std::collections::HashMap::new(),
            native_dimension_mapping: std::collections::HashMap::new(),
        };

        Self::validate(&metadata)?;
        Ok(metadata)
    }
}