yamlbase 0.7.2 - Docs.rs

use crate::database::Value;
use crate::sql::executor::QueryResult;
use crate::YamlBaseError;
use sqlparser::ast::{BinaryOperator, Expr, OrderByExpr, Select, SelectItem, SetExpr, Statement, TableFactor};
use sqlparser::dialect::PostgreSqlDialect;
use sqlparser::parser::Parser;
use tracing::debug;

use super::postgres_catalog::PostgresCatalog;
use super::postgres_information_schema::PostgresInformationSchema;

/// Represents a schema in the catalog system
#[derive(Debug, Clone, PartialEq)]
pub enum CatalogSchema {
    PgCatalog,
    InformationSchema,
    Public,
    User(String),
}

impl CatalogSchema {
    fn from_name(name: &str) -> Self {
        match name.to_lowercase().as_str() {
            "pg_catalog" => CatalogSchema::PgCatalog,
            "information_schema" => CatalogSchema::InformationSchema,
            "public" => CatalogSchema::Public,
            other => CatalogSchema::User(other.to_string()),
        }
    }
}

/// Routes catalog queries to appropriate handlers
#[derive(Clone)]
pub struct CatalogRouter {
    pg_catalog: PostgresCatalog,
    information_schema: PostgresInformationSchema,
}

impl CatalogRouter {
    pub fn new(pg_catalog: PostgresCatalog, information_schema: PostgresInformationSchema) -> Self {
        Self {
            pg_catalog,
            information_schema,
        }
    }

    /// Check if a query is a catalog query and route it appropriately
    pub fn route_query(&self, query: &str) -> crate::Result<Option<QueryResult>> {
        use tracing::debug;
        
        debug!("CatalogRouter checking query: {}", query);
        
        // Special handling for SQLAlchemy's complex column introspection query
        if self.is_complex_sqlalchemy_column_query(query) {
            debug!("Detected complex SQLAlchemy column query");
            return Ok(Some(self.handle_complex_sqlalchemy_column_query(query)?));
        }
        
        // Special handling for foreign key constraint queries
        if self.is_foreign_key_query(query) {
            debug!("Detected foreign key constraint query");
            return Ok(Some(self.handle_foreign_key_query(query)?));
        }
        
        // Special handling for SQLAlchemy's JOIN query pattern
        if self.is_sqlalchemy_table_query(query) {
            return Ok(Some(self.handle_sqlalchemy_table_query()));
        }
        
        // Special handling for SQLAlchemy's column introspection query
        if self.is_sqlalchemy_column_query(query) {
            return Ok(Some(self.handle_sqlalchemy_column_query(query)?));
        }
        
        // Special handling for simple catalog JOINs
        if self.is_catalog_join_query(query) {
            return self.handle_catalog_join_query(query);
        }
        
        // Parse the SQL to extract schema and table information
        let dialect = PostgreSqlDialect {};
        let statements = Parser::parse_sql(&dialect, query)
            .map_err(YamlBaseError::SqlParse)?;

        if statements.is_empty() {
            return Ok(None);
        }

        // We only handle SELECT statements for catalog queries
        let statement = &statements[0];
        if let Statement::Query(query) = statement {
            if let SetExpr::Select(select) = &*query.body {
                let order_by_exprs = query.order_by.iter()
                    .flat_map(|ob| ob.exprs.iter())
                    .cloned()
                    .collect::<Vec<_>>();
                return self.handle_select(select, query.limit.as_ref(), &order_by_exprs);
            }
        }

        Ok(None)
    }

    fn handle_select(&self, select: &Select, limit: Option<&Expr>, order_by: &[OrderByExpr]) -> crate::Result<Option<QueryResult>> {
        // Extract table information from FROM clause
        for table_with_joins in &select.from {
            if let Some((schema, table)) = Self::extract_schema_table(&table_with_joins.relation) {
                // Check if this is a catalog schema
                match schema {
                    CatalogSchema::PgCatalog => {
                        return self.handle_pg_catalog_query(
                            &table,
                            &select.projection,
                            select.selection.as_ref(),
                            limit,
                            order_by
                        );
                    }
                    CatalogSchema::InformationSchema => {
                        return self.handle_information_schema_query(
                            &table,
                            &select.projection,
                            select.selection.as_ref(),
                            limit,
                            order_by
                        );
                    }
                    _ => {
                        // Not a catalog query
                        return Ok(None);
                    }
                }
            }
        }

        Ok(None)
    }

    fn extract_schema_table(table_factor: &TableFactor) -> Option<(CatalogSchema, String)> {
        match table_factor {
            TableFactor::Table { name, .. } => {
                let parts: Vec<&str> = name.0.iter().map(|ident| ident.value.as_str()).collect();
                
                match parts.len() {
                    1 => {
                        // Just table name - check if it's a known catalog table
                        let table_lower = parts[0].to_lowercase();
                        if table_lower.starts_with("pg_") {
                            Some((CatalogSchema::PgCatalog, parts[0].to_string()))
                        } else {
                            None
                        }
                    }
                    2 => {
                        // schema.table format
                        Some((CatalogSchema::from_name(parts[0]), parts[1].to_string()))
                    }
                    _ => None,
                }
            }
            _ => None,
        }
    }

    fn handle_pg_catalog_query(
        &self,
        table: &str,
        projection: &[SelectItem],
        where_clause: Option<&Expr>,
        limit: Option<&Expr>,
        order_by: &[OrderByExpr],
    ) -> crate::Result<Option<QueryResult>> {
        let table_lower = table.to_lowercase();
        
        // Get the base result from the catalog
        let mut result = match table_lower.as_str() {
            "pg_type" => self.pg_catalog.query_pg_type(None),
            "pg_class" => self.pg_catalog.query_pg_class(None),
            "pg_attribute" => self.pg_catalog.query_pg_attribute(None),
            "pg_namespace" => self.pg_catalog.query_pg_namespace(),
            "pg_database" => self.pg_catalog.query_pg_database(),
            "pg_proc" => self.pg_catalog.query_pg_proc(),
            "pg_index" => self.pg_catalog.query_pg_index(),
            "pg_constraint" => self.pg_catalog.query_pg_constraint(),
            "pg_settings" => self.pg_catalog.query_pg_settings(),
            "pg_description" => self.pg_catalog.query_pg_description(),
            "pg_roles" | "pg_user" => self.pg_catalog.query_pg_roles(),
            "pg_am" => self.pg_catalog.query_pg_am(),
            "pg_operator" => self.pg_catalog.query_pg_operator(),
            "pg_cast" => self.pg_catalog.query_pg_cast(),
            "pg_enum" => self.pg_catalog.query_pg_enum(),
            "pg_range" => self.pg_catalog.query_pg_range(),
            "pg_trigger" => self.pg_catalog.query_pg_trigger(),
            "pg_depend" => self.pg_catalog.query_pg_depend(),
            "pg_aggregate" => self.pg_catalog.query_pg_aggregate(),
            "pg_sequence" => self.pg_catalog.query_pg_sequence(),
            "pg_stat_user_tables" => self.pg_catalog.query_pg_stat_user_tables(),
            "pg_tables" => self.pg_catalog.query_pg_tables(),
            "pg_statio_user_tables" => self.pg_catalog.query_pg_statio_user_tables(),
            _ => return Ok(None),
        };

        // Apply WHERE clause filtering
        if let Some(where_expr) = where_clause {
            result = self.apply_where_clause(result, where_expr)?;
        }

        // Apply projection (column selection)
        result = self.apply_projection(result, projection)?;

        // Apply ORDER BY
        if !order_by.is_empty() {
            result = self.apply_order_by(result, order_by)?;
        }

        // Apply LIMIT
        if let Some(limit_expr) = limit {
            result = self.apply_limit(result, limit_expr)?;
        }

        Ok(Some(result))
    }

    fn handle_information_schema_query(
        &self,
        table: &str,
        projection: &[SelectItem],
        where_clause: Option<&Expr>,
        limit: Option<&Expr>,
        order_by: &[OrderByExpr],
    ) -> crate::Result<Option<QueryResult>> {
        let table_lower = table.to_lowercase();
        
        // Get the base result from the information schema
        let mut result = match table_lower.as_str() {
            "tables" => self.information_schema.query_tables(None),
            "columns" => self.information_schema.query_columns(None),
            "schemata" => self.information_schema.query_schemata(None),
            "table_constraints" => self.information_schema.query_table_constraints(),
            "key_column_usage" => self.information_schema.query_key_column_usage(),
            "referential_constraints" => self.information_schema.query_referential_constraints(),
            "check_constraints" => self.information_schema.query_check_constraints(),
            "routines" => self.information_schema.query_routines(),
            "parameters" => self.information_schema.query_parameters(),
            "views" => self.information_schema.query_views(),
            "sequences" => self.information_schema.query_sequences(),
            _ => return Ok(None),
        };

        // Apply WHERE clause filtering
        if let Some(where_expr) = where_clause {
            result = self.apply_where_clause(result, where_expr)?;
        }

        // Apply projection
        result = self.apply_projection(result, projection)?;

        // Apply ORDER BY
        if !order_by.is_empty() {
            result = self.apply_order_by(result, order_by)?;
        }

        // Apply LIMIT
        if let Some(limit_expr) = limit {
            result = self.apply_limit(result, limit_expr)?;
        }

        Ok(Some(result))
    }

    fn apply_where_clause(&self, mut result: QueryResult, where_expr: &Expr) -> crate::Result<QueryResult> {
        // Filter rows based on WHERE conditions
        debug!("Applying WHERE clause: {:?}", where_expr);
        debug!("Columns: {:?}", result.columns);
        debug!("Initial row count: {}", result.rows.len());
        
        let filtered_rows = result.rows.into_iter()
            .filter(|row| {
                let matches = self.evaluate_where_condition(row, where_expr, &result.columns);
                debug!("Row {:?} matches WHERE: {}", row, matches);
                matches
            })
            .collect::<Vec<_>>();
        
        debug!("Filtered row count: {}", filtered_rows.len());
        result.rows = filtered_rows;
        Ok(result)
    }

    fn evaluate_where_condition(&self, row: &[Value], expr: &Expr, columns: &[String]) -> bool {
        match expr {
            Expr::BinaryOp { left, op, right } => {
                self.evaluate_binary_op(row, left, op, right, columns)
            }
            Expr::InList { expr, list, negated } => {
                self.evaluate_in_list(row, expr, list, *negated, columns)
            }
            Expr::IsNull(expr) => {
                self.evaluate_is_null(row, expr, columns, false)
            }
            Expr::IsNotNull(expr) => {
                self.evaluate_is_null(row, expr, columns, true)
            }
            Expr::Like { expr, pattern, negated, .. } => {
                self.evaluate_like(row, expr, pattern, columns, *negated)
            }
            _ => true, // Unknown expressions pass through
        }
    }

    fn evaluate_binary_op(
        &self,
        row: &[Value],
        left: &Expr,
        op: &sqlparser::ast::BinaryOperator,
        right: &Expr,
        columns: &[String],
    ) -> bool {
        use sqlparser::ast::BinaryOperator;

        // Handle AND/OR operators specially - they work on boolean expressions
        match op {
            BinaryOperator::And => {
                let left_result = self.evaluate_where_condition(row, left, columns);
                let right_result = self.evaluate_where_condition(row, right, columns);
                debug!("AND: left_result={}, right_result={}", left_result, right_result);
                return left_result && right_result;
            }
            BinaryOperator::Or => {
                let left_result = self.evaluate_where_condition(row, left, columns);
                let right_result = self.evaluate_where_condition(row, right, columns);
                debug!("OR: left_result={}, right_result={}", left_result, right_result);
                return left_result || right_result;
            }
            _ => {}
        }

        // For other operators, get the values and compare them
        let left_val = self.get_expr_value(row, left, columns);
        let right_val = self.get_expr_value(row, right, columns);
        
        debug!("Binary op: left_expr={:?}, left_val={:?}, op={:?}, right_expr={:?}, right_val={:?}",
               left, left_val, op, right, right_val);

        match (left_val, right_val) {
            (Some(lv), Some(rv)) => {
                match op {
                    BinaryOperator::Eq => {
                        let result = lv == rv;
                        debug!("Eq comparison: {:?} == {:?} = {}", lv, rv, result);
                        result
                    }
                    BinaryOperator::NotEq => lv != rv,
                    BinaryOperator::Lt => self.compare_values(&lv, &rv) == Some(std::cmp::Ordering::Less),
                    BinaryOperator::LtEq => matches!(self.compare_values(&lv, &rv), Some(std::cmp::Ordering::Less | std::cmp::Ordering::Equal)),
                    BinaryOperator::Gt => self.compare_values(&lv, &rv) == Some(std::cmp::Ordering::Greater),
                    BinaryOperator::GtEq => matches!(self.compare_values(&lv, &rv), Some(std::cmp::Ordering::Greater | std::cmp::Ordering::Equal)),
                    _ => true,
                }
            }
            _ => false,
        }
    }

    fn evaluate_in_list(&self, row: &[Value], expr: &Expr, list: &[Expr], negated: bool, columns: &[String]) -> bool {
        if let Some(val) = self.get_expr_value(row, expr, columns) {
            let in_list = list.iter().any(|item| {
                self.get_expr_value(row, item, columns) == Some(val.clone())
            });
            if negated { !in_list } else { in_list }
        } else {
            false
        }
    }

    fn evaluate_is_null(&self, row: &[Value], expr: &Expr, columns: &[String], is_not: bool) -> bool {
        let val = self.get_expr_value(row, expr, columns);
        let is_null = val.is_none() || matches!(val, Some(Value::Null));
        if is_not { !is_null } else { is_null }
    }

    fn evaluate_like(&self, row: &[Value], expr: &Expr, pattern: &Expr, columns: &[String], negated: bool) -> bool {
        let val = self.get_expr_value(row, expr, columns);
        let pat = self.get_expr_value(row, pattern, columns);

        match (val, pat) {
            (Some(Value::Text(s)), Some(Value::Text(p))) => {
                // Simple LIKE implementation (convert % to .* for regex)
                let regex_pattern = p.replace('%', ".*").replace('_', ".");
                let matches = regex::Regex::new(&format!("^{}$", regex_pattern))
                    .map(|re| re.is_match(&s))
                    .unwrap_or(false);
                
                if negated {
                    !matches
                } else {
                    matches
                }
            }
            _ => false,
        }
    }

    fn get_expr_value(&self, row: &[Value], expr: &Expr, columns: &[String]) -> Option<Value> {
        match expr {
            Expr::Identifier(ident) => {
                let col_name = ident.value.to_lowercase();
                columns.iter().position(|c| c.to_lowercase() == col_name)
                    .and_then(|idx| row.get(idx).cloned())
            }
            Expr::CompoundIdentifier(parts) => {
                // Handle compound identifiers like pg_catalog.pg_namespace.nspname
                // Use the last part as the column name
                let col_name = parts.last().map(|i| i.value.to_lowercase()).unwrap_or_default();
                columns.iter().position(|c| c.to_lowercase() == col_name)
                    .and_then(|idx| row.get(idx).cloned())
            }
            Expr::Value(v) => {
                use sqlparser::ast::Value as SqlValue;
                match v {
                    SqlValue::SingleQuotedString(s) => Some(Value::Text(s.clone())),
                    SqlValue::Number(n, _) => {
                        if let Ok(i) = n.parse::<i32>() {
                            Some(Value::Integer(i as i64))
                        } else if let Ok(f) = n.parse::<f64>() {
                            Some(Value::Float(f as f32))
                        } else {
                            None
                        }
                    }
                    SqlValue::Boolean(b) => Some(Value::Boolean(*b)),
                    SqlValue::Null => Some(Value::Null),
                    _ => None,
                }
            }
            _ => None,
        }
    }

    fn compare_values(&self, left: &Value, right: &Value) -> Option<std::cmp::Ordering> {
        match (left, right) {
            (Value::Integer(l), Value::Integer(r)) => Some(l.cmp(r)),
            (Value::Float(l), Value::Float(r)) => l.partial_cmp(r),
            (Value::Text(l), Value::Text(r)) => Some(l.cmp(r)),
            (Value::Boolean(l), Value::Boolean(r)) => Some(l.cmp(r)),
            _ => None,
        }
    }

    fn apply_projection(&self, result: QueryResult, projection: &[SelectItem]) -> crate::Result<QueryResult> {
        // Handle SELECT * - return all columns
        if projection.len() == 1 {
            if let SelectItem::Wildcard(_) = &projection[0] {
                return Ok(result);
            }
        }

        // Extract requested columns
        let mut selected_columns = Vec::new();
        let mut selected_indices = Vec::new();
        let mut selected_types = Vec::new();

        for item in projection {
            match item {
                SelectItem::UnnamedExpr(Expr::Identifier(ident)) => {
                    let col_name = ident.value.to_lowercase();
                    if let Some(idx) = result.columns.iter().position(|c| c.to_lowercase() == col_name) {
                        selected_columns.push(result.columns[idx].clone());
                        selected_indices.push(idx);
                        selected_types.push(result.column_types[idx].clone());
                    }
                }
                SelectItem::UnnamedExpr(Expr::CompoundIdentifier(parts)) => {
                    // Handle compound identifiers like pg_catalog.pg_namespace.nspname
                    // Use the last part as the column name
                    let col_name = parts.last().map(|i| i.value.to_lowercase()).unwrap_or_default();
                    if let Some(idx) = result.columns.iter().position(|c| c.to_lowercase() == col_name) {
                        selected_columns.push(result.columns[idx].clone());
                        selected_indices.push(idx);
                        selected_types.push(result.column_types[idx].clone());
                    }
                }
                SelectItem::ExprWithAlias { expr: Expr::Identifier(ident), alias } => {
                    let col_name = ident.value.to_lowercase();
                    if let Some(idx) = result.columns.iter().position(|c| c.to_lowercase() == col_name) {
                        selected_columns.push(alias.value.clone());
                        selected_indices.push(idx);
                        selected_types.push(result.column_types[idx].clone());
                    }
                }
                SelectItem::ExprWithAlias { expr: Expr::CompoundIdentifier(parts), alias } => {
                    // Handle compound identifiers with alias
                    let col_name = parts.last().map(|i| i.value.to_lowercase()).unwrap_or_default();
                    if let Some(idx) = result.columns.iter().position(|c| c.to_lowercase() == col_name) {
                        selected_columns.push(alias.value.clone());
                        selected_indices.push(idx);
                        selected_types.push(result.column_types[idx].clone());
                    }
                }
                _ => {} // Ignore complex expressions for now
            }
        }

        // If no columns were selected, return empty result
        if selected_columns.is_empty() && !projection.is_empty() {
            return Ok(QueryResult {
                columns: vec![],
                column_types: vec![],
                rows: vec![],
            });
        }

        // Project the rows
        let projected_rows = result.rows.into_iter()
            .map(|row| {
                selected_indices.iter()
                    .map(|&idx| row.get(idx).cloned().unwrap_or(Value::Null))
                    .collect()
            })
            .collect();

        Ok(QueryResult {
            columns: selected_columns,
            column_types: selected_types,
            rows: projected_rows,
        })
    }

    fn apply_limit(&self, mut result: QueryResult, limit_expr: &Expr) -> crate::Result<QueryResult> {
        if let Expr::Value(sqlparser::ast::Value::Number(n, _)) = limit_expr {
            if let Ok(limit) = n.parse::<usize>() {
                result.rows.truncate(limit);
            }
        }
        Ok(result)
    }

    fn apply_order_by(&self, mut result: QueryResult, order_by: &[OrderByExpr]) -> crate::Result<QueryResult> {
        // Sort rows based on ORDER BY expressions
        for order_expr in order_by.iter().rev() {
            // We process in reverse order to handle multiple columns correctly
            if let Expr::Identifier(ident) = &order_expr.expr {
                let col_name = ident.value.to_lowercase();
                if let Some(col_idx) = result.columns.iter().position(|c| c.to_lowercase() == col_name) {
                    let ascending = order_expr.asc.unwrap_or(true);
                    
                    result.rows.sort_by(|a, b| {
                        let a_val = a.get(col_idx);
                        let b_val = b.get(col_idx);
                        
                        let ordering = match (a_val, b_val) {
                            (Some(av), Some(bv)) => self.compare_values(av, bv).unwrap_or(std::cmp::Ordering::Equal),
                            (Some(_), None) => std::cmp::Ordering::Less,
                            (None, Some(_)) => std::cmp::Ordering::Greater,
                            (None, None) => std::cmp::Ordering::Equal,
                        };
                        
                        if ascending {
                            ordering
                        } else {
                            ordering.reverse()
                        }
                    });
                }
            } else if let Expr::CompoundIdentifier(parts) = &order_expr.expr {
                // Handle compound identifiers like pg_catalog.pg_namespace.nspname
                let col_name = parts.last().map(|i| i.value.to_lowercase()).unwrap_or_default();
                if let Some(col_idx) = result.columns.iter().position(|c| c.to_lowercase() == col_name) {
                    let ascending = order_expr.asc.unwrap_or(true);
                    
                    result.rows.sort_by(|a, b| {
                        let a_val = a.get(col_idx);
                        let b_val = b.get(col_idx);
                        
                        let ordering = match (a_val, b_val) {
                            (Some(av), Some(bv)) => self.compare_values(av, bv).unwrap_or(std::cmp::Ordering::Equal),
                            (Some(_), None) => std::cmp::Ordering::Less,
                            (None, Some(_)) => std::cmp::Ordering::Greater,
                            (None, None) => std::cmp::Ordering::Equal,
                        };
                        
                        if ascending {
                            ordering
                        } else {
                            ordering.reverse()
                        }
                    });
                }
            }
        }
        
        Ok(result)
    }
    
    /// Check if this is SQLAlchemy's specific table introspection query
    fn is_sqlalchemy_table_query(&self, query: &str) -> bool {
        let normalized = query.to_lowercase();
        // Check for the specific pattern SQLAlchemy uses
        normalized.contains("pg_catalog.pg_class") &&
        normalized.contains("pg_catalog.pg_namespace") &&
        normalized.contains("join") &&
        normalized.contains("relkind") &&
        (normalized.contains("pg_table_is_visible") || normalized.contains("relnamespace"))
    }
    
    /// Handle SQLAlchemy's table introspection query
    fn handle_sqlalchemy_table_query(&self) -> QueryResult {
        // Get user tables from pg_class
        let pg_class = self.pg_catalog.query_pg_class(None);
        
        // Filter for user tables (relkind = 'r') in public namespace (oid = 2200)
        let mut result_rows = Vec::new();
        for row in &pg_class.rows {
            // Check relkind (column 15)
            if let Some(Value::Text(relkind)) = row.get(15) {
                if relkind == "r" {
                    // Check relnamespace (column 2)
                    if let Some(Value::Integer(ns_oid)) = row.get(2) {
                        if *ns_oid == 2200 { // public namespace
                            // Get relname (column 1)
                            if let Some(Value::Text(relname)) = row.get(1) {
                                result_rows.push(vec![Value::Text(relname.clone())]);
                            }
                        }
                    }
                }
            }
        }
        
        QueryResult {
            columns: vec!["relname".to_string()],
            column_types: vec![crate::yaml::schema::SqlType::Text],
            rows: result_rows,
        }
    }
    
    /// Check if this is SQLAlchemy's column introspection query
    fn is_sqlalchemy_column_query(&self, query: &str) -> bool {
        let normalized = query.to_lowercase();
        // SQLAlchemy queries pg_attribute with specific aliases and format_type
        normalized.contains("pg_catalog.pg_attribute.attname as name") &&
        normalized.contains("format_type") &&
        normalized.contains("pg_catalog.pg_class") &&
        normalized.contains("join")
    }
    
    /// Handle SQLAlchemy's column introspection query
    fn handle_sqlalchemy_column_query(&self, query: &str) -> crate::Result<QueryResult> {
        use crate::yaml::schema::SqlType;
        
        // Parse the table name from the query
        let query_lower = query.to_lowercase();
        let table_name = if let Some(start) = query_lower.find("relname in (") {
            let after_in = &query[start + 12..];
            if after_in.starts_with("%(filter_names_") {
                // This is SQLAlchemy's parameterized query - return columns for all tables
                // SQLAlchemy will filter them itself
                return self.get_all_table_columns();
            }
            // Look for the quoted table name
            if let Some(quote_start) = after_in.find('\'')
                .or_else(|| after_in.find('\"')) {
                let table_str = &after_in[quote_start + 1..];
                if let Some(quote_end) = table_str.find('\'')
                    .or_else(|| table_str.find('\"')) {
                    table_str[..quote_end].to_string()
                } else {
                    return Ok(QueryResult {
                        columns: vec!["name".to_string(), "format_type".to_string(), 
                                     "default".to_string(), "not_null".to_string(),
                                     "table_name".to_string(), "comment".to_string(),
                                     "generated".to_string(), "identity_options".to_string()],
                        column_types: vec![SqlType::Text; 8],
                        rows: vec![],
                    });
                }
            } else {
                return Ok(QueryResult {
                    columns: vec!["name".to_string(), "format_type".to_string(), 
                                 "default".to_string(), "not_null".to_string(),
                                 "table_name".to_string(), "comment".to_string(),
                                 "generated".to_string(), "identity_options".to_string()],
                    column_types: vec![SqlType::Text; 8],
                    rows: vec![],
                });
            }
        } else {
            // Try to extract from WHERE clause
            return Ok(QueryResult {
                columns: vec!["name".to_string(), "format_type".to_string(), 
                             "default".to_string(), "not_null".to_string(),
                             "table_name".to_string(), "comment".to_string(),
                             "generated".to_string(), "identity_options".to_string()],
                column_types: vec![SqlType::Text; 8],
                rows: vec![],
            });
        };
        
        // Get pg_attribute data for this table
        let pg_attr = self.pg_catalog.query_pg_attribute(Some(query));
        let pg_class = self.pg_catalog.query_pg_class(None);
        
        // Find the table OID
        let mut table_oid = 0i64;
        for row in &pg_class.rows {
            if let (Some(Value::Text(name)), Some(Value::Integer(oid))) = (row.get(1), row.get(0)) {
                if name == &table_name {
                    table_oid = *oid;
                    break;
                }
            }
        }
        
        // Build result rows with SQLAlchemy's expected column names
        let mut result_rows = Vec::new();
        for row in &pg_attr.rows {
            // Only include columns for this table
            if let Some(Value::Integer(attrelid)) = row.get(0) {
                if *attrelid == table_oid {
                    // Extract column info
                    let name = row.get(1).map(|v| v.to_string()).unwrap_or_default();
                    let type_oid = row.get(2).map(|v| v.to_string()).unwrap_or_default();
                    let not_null = row.get(4).map(|v| v.to_string()).unwrap_or_else(|| "f".to_string());
                    let attnum = row.get(5).map(|v| v.to_string()).unwrap_or_default();
                    
                    // Skip system columns (attnum <= 0)
                    if let Ok(num) = attnum.parse::<i32>() {
                        if num <= 0 {
                            continue;
                        }
                    }
                    
                    // Format type based on type OID
                    let format_type = match type_oid.as_str() {
                        "23" => "integer",
                        "20" => "bigint",
                        "25" => "text",
                        "1043" => "character varying",
                        "16" => "boolean",
                        "1082" => "date",
                        "1114" => "timestamp without time zone",
                        "700" => "real",
                        "701" => "double precision",
                        "1700" => "numeric",
                        _ => "text",
                    };
                    
                    result_rows.push(vec![
                        Value::Text(name.clone()),                        // name
                        Value::Text(format_type.to_string()),            // format_type
                        Value::Null,                                      // default
                        Value::Boolean(not_null == "t"),                // not_null
                        Value::Text(table_name.clone()),                 // table_name
                        Value::Null,                                      // comment
                        Value::Text("".to_string()),                     // generated
                        Value::Null,                                      // identity_options
                    ]);
                }
            }
        }
        
        Ok(QueryResult {
            columns: vec!["name".to_string(), "format_type".to_string(), 
                         "default".to_string(), "not_null".to_string(),
                         "table_name".to_string(), "comment".to_string(),
                         "generated".to_string(), "identity_options".to_string()],
            column_types: vec![SqlType::Text, SqlType::Text, SqlType::Text, SqlType::Boolean,
                              SqlType::Text, SqlType::Text, SqlType::Text, SqlType::Text],
            rows: result_rows,
        })
    }
    
    /// Check if this is a JOIN between catalog tables
    fn is_catalog_join_query(&self, query: &str) -> bool {
        let normalized = query.to_lowercase();
        // Check for JOIN between pg_catalog tables
        (normalized.contains("pg_catalog.pg_attribute") || normalized.contains("pg_attribute")) &&
        (normalized.contains("pg_catalog.pg_class") || normalized.contains("pg_class")) &&
        normalized.contains("join")
    }
    
    /// Handle JOIN queries between catalog tables
    fn handle_catalog_join_query(&self, query: &str) -> crate::Result<Option<QueryResult>> {
        use crate::yaml::schema::SqlType;
        
        let query_lower = query.to_lowercase();
        
        // Simple JOIN between pg_attribute and pg_class
        if query_lower.contains("pg_attribute") && query_lower.contains("pg_class") {
            // Get the tables
            let pg_attr = self.pg_catalog.query_pg_attribute(None);
            let pg_class = self.pg_catalog.query_pg_class(None);
            
            // Parse what columns are being selected
            let mut columns = Vec::new();
            let mut column_types = Vec::new();
            
            // Check for specific column patterns in SELECT clause
            if query_lower.contains("attname as name") || query_lower.contains("attname as name") {
                columns.push("name".to_string());
                column_types.push(SqlType::Text);
            } else if query_lower.contains("attname") {
                columns.push("attname".to_string());
                column_types.push(SqlType::Text);
            }
            
            if query_lower.contains("relname as table_name") {
                columns.push("table_name".to_string());
                column_types.push(SqlType::Text);
            } else if query_lower.contains("relname") && !query_lower.contains("where") {
                columns.push("relname".to_string());
                column_types.push(SqlType::Text);
            }
            
            if query_lower.contains("attnotnull as not_null") {
                columns.push("not_null".to_string());
                column_types.push(SqlType::Boolean);
            } else if query_lower.contains("attnotnull") {
                columns.push("attnotnull".to_string());
                column_types.push(SqlType::Boolean);
            }
            
            if query_lower.contains("attnum") && !query_lower.contains("attnum >") && !query_lower.contains("attnum<") {
                columns.push("attnum".to_string());
                column_types.push(SqlType::Integer);
            }
            
            // Build result rows by joining
            let mut result_rows = Vec::new();
            
            // Simple nested loop join
            for attr_row in &pg_attr.rows {
                // pg_attribute columns: attrelid(0), attname(1), atttypid(2), ..., attnotnull(12), ...
                if let (Some(Value::Integer(attrelid)), Some(Value::Integer(attnum))) = 
                    (attr_row.get(0), attr_row.get(5)) {
                    
                    // Skip system columns
                    if *attnum <= 0 {
                        continue;
                    }
                    
                    // Find matching pg_class row
                    for class_row in &pg_class.rows {
                        // pg_class columns: oid(0), relname(1), ...
                        if let (Some(Value::Integer(oid)), Some(Value::Text(relname))) = 
                            (class_row.get(0), class_row.get(1)) {
                            
                            if *oid == *attrelid {
                                // Check WHERE conditions if any
                                let mut include = true;
                                
                                // Check for WHERE relname = 'xxx'
                                if let Some(idx) = query_lower.find("relname = '") {
                                    let after = &query[idx + 11..];
                                    if let Some(end) = after.find("'") {
                                        let target_table = &after[..end];
                                        if relname != target_table {
                                            include = false;
                                        }
                                    }
                                } else if let Some(idx) = query_lower.find("relname='") {
                                    let after = &query[idx + 9..];
                                    if let Some(end) = after.find("'") {
                                        let target_table = &after[..end];
                                        if relname != target_table {
                                            include = false;
                                        }
                                    }
                                }
                                
                                // Check for WHERE attrelid = xxx
                                if query_lower.contains("where") && query_lower.contains("attrelid") {
                                    if let Some(idx) = query_lower.find("attrelid = ") {
                                        let after = &query[idx + 11..];
                                        let num_str: String = after.chars()
                                            .take_while(|c| c.is_ascii_digit())
                                            .collect();
                                        if let Ok(target_oid) = num_str.parse::<i64>() {
                                            if *attrelid != target_oid {
                                                include = false;
                                            }
                                        }
                                    }
                                }
                                
                                if include {
                                    // Build row based on selected columns
                                    let mut row = Vec::new();
                                    
                                    for col in &columns {
                                        match col.as_str() {
                                            "name" | "attname" => {
                                                if let Some(Value::Text(name)) = attr_row.get(1) {
                                                    row.push(Value::Text(name.clone()));
                                                } else {
                                                    row.push(Value::Null);
                                                }
                                            }
                                            "table_name" | "relname" => {
                                                row.push(Value::Text(relname.clone()));
                                            }
                                            "not_null" | "attnotnull" => {
                                                if let Some(Value::Boolean(not_null)) = attr_row.get(12) {
                                                    row.push(Value::Boolean(*not_null));
                                                } else {
                                                    row.push(Value::Boolean(false));
                                                }
                                            }
                                            "attnum" => {
                                                row.push(Value::Integer(*attnum));
                                            }
                                            _ => {}
                                        }
                                    }
                                    
                                    if !row.is_empty() {
                                        result_rows.push(row);
                                    }
                                }
                                break; // Found the matching class row
                            }
                        }
                    }
                }
            }
            
            Ok(Some(QueryResult {
                columns,
                column_types,
                rows: result_rows,
            }))
        } else {
            Ok(None)
        }
    }
    
    /// Get columns for all user tables in the format SQLAlchemy expects
    /// Check if this is a complex SQLAlchemy column introspection query
    fn is_complex_sqlalchemy_column_query(&self, query: &str) -> bool {
        let normalized = query.to_lowercase();
        // SQLAlchemy's complex query pattern includes:
        // - SELECT pg_catalog.pg_attribute.attname AS name
        // - format_type() function
        // - LEFT OUTER JOIN pg_catalog.pg_attribute
        // - LEFT OUTER JOIN pg_catalog.pg_description  
        // - JOIN pg_catalog.pg_namespace
        normalized.contains("pg_catalog.pg_attribute.attname as name") &&
        normalized.contains("format_type") &&
        normalized.contains("left outer join") &&
        normalized.contains("pg_catalog.pg_class")
    }
    
    /// Check if this is a foreign key constraint query
    fn is_foreign_key_query(&self, query: &str) -> bool {
        let normalized = query.to_lowercase();
        // SQLAlchemy queries for foreign keys with:
        // - pg_constraint table
        // - contype = 'f' for foreign keys
        // - LEFT OUTER JOIN with pg_class
        normalized.contains("pg_constraint") &&
        normalized.contains("contype = 'f'") &&
        normalized.contains("left outer join")
    }
    
    /// Handle complex SQLAlchemy column introspection query
    fn handle_complex_sqlalchemy_column_query(&self, query: &str) -> crate::Result<QueryResult> {
        use crate::yaml::schema::SqlType;
        use tracing::debug;
        
        debug!("Handling complex SQLAlchemy column query");
        
        // Extract the table name from the WHERE clause
        let query_lower = query.to_lowercase();
        let mut table_name = None;
        
        // Look for: WHERE ... pg_class.relname IN ('tablename')
        if let Some(in_pos) = query_lower.find("pg_class.relname in") {
            let after_in = &query_lower[in_pos + 20..];
            if let Some(paren_start) = after_in.find('(') {
                let after_paren = &after_in[paren_start + 1..];
                if let Some(paren_end) = after_paren.find(')') {
                    let tables_str = &after_paren[..paren_end];
                    // Extract the table name from quotes
                    if let Some(quote_start) = tables_str.find('\'') {
                        let after_quote = &tables_str[quote_start + 1..];
                        if let Some(quote_end) = after_quote.find('\'') {
                            table_name = Some(after_quote[..quote_end].to_string());
                        }
                    }
                }
            }
        }
        
        debug!("Extracted table name: {:?}", table_name);
        
        // If we found a table name, return its columns
        if let Some(table) = table_name {
            // Get pg_class and pg_attribute data
            let pg_class = self.pg_catalog.query_pg_class(None);
            let pg_attr = self.pg_catalog.query_pg_attribute(None);
            
            let mut result_rows = Vec::new();
            
            // Find the table's OID in pg_class
            let mut table_oid = None;
            for class_row in &pg_class.rows {
                if let (Some(Value::Integer(oid)), Some(Value::Text(relname))) = 
                    (class_row.get(0), class_row.get(1)) {
                    if relname == &table {
                        table_oid = Some(*oid);
                        break;
                    }
                }
            }
            
            debug!("Found table OID: {:?}", table_oid);
            
            if let Some(oid) = table_oid {
                // Get all columns for this table
                for attr_row in &pg_attr.rows {
                    if let (Some(Value::Integer(attrelid)), Some(Value::Text(attname)), 
                            Some(Value::Integer(atttypid)), Some(Value::Boolean(attnotnull)),
                            Some(Value::Integer(attnum))) = 
                        (attr_row.get(0), attr_row.get(1), attr_row.get(2), 
                         attr_row.get(12), attr_row.get(5)) {
                        
                        // Match columns to this table
                        if *attrelid != oid {
                            continue;
                        }
                        
                        // Skip system columns
                        if *attnum <= 0 {
                            continue;
                        }
                        
                        // Format type based on type OID
                        let format_type = match *atttypid {
                            16 => "boolean",
                            20 => "bigint",
                            21 => "smallint",
                            23 => "integer",
                            25 => "text",
                            700 => "real",
                            701 => "double precision",
                            1042 => "character",
                            1043 => "character varying",
                            1082 => "date",
                            1083 => "time without time zone",
                            1114 => "timestamp without time zone",
                            1700 => "numeric",
                            _ => "text",
                        };
                        
                        result_rows.push(vec![
                            Value::Text(attname.clone()),                   // name
                            Value::Text(format_type.to_string()),          // format_type
                            Value::Null,                                    // default
                            Value::Boolean(*attnotnull),                   // not_null
                            Value::Text(table.clone()),                    // table_name
                            Value::Null,                                    // comment
                            Value::Text("".to_string()),                   // generated
                            Value::Null,                                    // identity_options
                        ]);
                    }
                }
            }
            
            debug!("Returning {} columns for table {}", result_rows.len(), table);
            
            Ok(QueryResult {
                columns: vec!["name".to_string(), "format_type".to_string(), 
                             "default".to_string(), "not_null".to_string(),
                             "table_name".to_string(), "comment".to_string(),
                             "generated".to_string(), "identity_options".to_string()],
                column_types: vec![SqlType::Text, SqlType::Text, SqlType::Text, SqlType::Boolean,
                                  SqlType::Text, SqlType::Text, SqlType::Text, SqlType::Text],
                rows: result_rows,
            })
        } else {
            // Fallback to returning all columns for all tables
            self.get_all_table_columns()
        }
    }
    
    /// Handle foreign key constraint query from SQLAlchemy
    fn handle_foreign_key_query(&self, query: &str) -> crate::Result<QueryResult> {
        use crate::yaml::schema::SqlType;
        use tracing::debug;
        
        debug!("Handling foreign key constraint query");
        
        // Extract the table name from the WHERE clause
        let query_lower = query.to_lowercase();
        let mut table_name = None;
        
        // Look for: WHERE ... pg_class.relname IN ('tablename')
        if let Some(in_pos) = query_lower.find("pg_class.relname in") {
            let after_in = &query_lower[in_pos + 20..];
            if let Some(paren_start) = after_in.find('(') {
                let after_paren = &after_in[paren_start + 1..];
                if let Some(paren_end) = after_paren.find(')') {
                    let tables_str = &after_paren[..paren_end];
                    // Extract the table name from quotes
                    if let Some(quote_start) = tables_str.find('\'') {
                        let after_quote = &tables_str[quote_start + 1..];
                        if let Some(quote_end) = after_quote.find('\'') {
                            table_name = Some(after_quote[..quote_end].to_string());
                        }
                    }
                }
            }
        }
        
        debug!("Extracted table name for FK query: {:?}", table_name);
        
        // Get constraint data
        let pg_constraint = self.pg_catalog.query_pg_constraint();
        let pg_class = self.pg_catalog.query_pg_class(None);
        let _pg_namespace = self.pg_catalog.query_pg_namespace();
        
        // Build result based on SQLAlchemy's expected columns:
        // relname, conname, constraintdef, nsp_ref.nspname, description
        let mut result_rows = Vec::new();
        
        // Find the table's OID if we have a specific table
        let table_oid = if let Some(ref table) = table_name {
            let mut oid = None;
            for class_row in &pg_class.rows {
                if let (Some(Value::Integer(row_oid)), Some(Value::Text(relname))) = 
                    (class_row.get(0), class_row.get(1)) {
                    if relname == table {
                        oid = Some(*row_oid);
                        break;
                    }
                }
            }
            oid
        } else {
            None
        };
        
        debug!("Table OID for FK query: {:?}", table_oid);
        
        // Iterate through constraints
        for constraint_row in &pg_constraint.rows {
            if let (Some(Value::Integer(_oid)), Some(Value::Text(conname)), 
                    Some(Value::Text(contype)), Some(Value::Integer(conrelid)),
                    Some(Value::Integer(confrelid))) = 
                (constraint_row.get(0), constraint_row.get(1), 
                 constraint_row.get(3), constraint_row.get(7), constraint_row.get(11)) {
                
                // Only process foreign key constraints
                if contype != "f" {
                    continue;
                }
                
                // If we have a specific table, check if this constraint belongs to it
                if let Some(target_oid) = table_oid {
                    if *conrelid != target_oid {
                        continue;
                    }
                }
                
                // Find the referenced table name
                let mut ref_table_name = String::new();
                let ref_schema = "public".to_string();
                
                for class_row in &pg_class.rows {
                    if let (Some(Value::Integer(class_oid)), Some(Value::Text(relname))) = 
                        (class_row.get(0), class_row.get(1)) {
                        if *class_oid == *confrelid {
                            ref_table_name = relname.clone();
                            break;
                        }
                    }
                }
                
                // Generate constraint definition
                let constraintdef = format!("FOREIGN KEY ({}) REFERENCES {}(id)", 
                    conname.replace("_fkey", "").replace(&format!("{}_", table_name.as_ref().unwrap_or(&String::new())), ""),
                    ref_table_name);
                
                result_rows.push(vec![
                    table_name.as_ref().map_or(Value::Text("".to_string()), |t| Value::Text(t.clone())), // relname
                    Value::Text(conname.clone()),          // conname
                    Value::Text(constraintdef),            // constraint definition
                    Value::Text(ref_schema),               // referenced schema
                    Value::Null,                           // description
                ]);
            }
        }
        
        debug!("Returning {} foreign key constraints", result_rows.len());
        
        Ok(QueryResult {
            columns: vec!["relname".to_string(), "conname".to_string(), 
                         "anon_1".to_string(), "nspname".to_string(), "description".to_string()],
            column_types: vec![SqlType::Text, SqlType::Text, SqlType::Text, SqlType::Text, SqlType::Text],
            rows: result_rows,
        })
    }
    
    fn get_all_table_columns(&self) -> crate::Result<QueryResult> {
        use crate::yaml::schema::SqlType;
        
        // Get all tables from pg_class
        let pg_class = self.pg_catalog.query_pg_class(None);
        let pg_attr = self.pg_catalog.query_pg_attribute(None);
        
        let mut result_rows = Vec::new();
        
        // For each user table
        for class_row in &pg_class.rows {
            if let (Some(Value::Integer(oid)), Some(Value::Text(relname)), Some(Value::Text(relkind))) = 
                (class_row.get(0), class_row.get(1), class_row.get(15)) {
                
                // Only process regular tables
                if relkind != "r" {
                    continue;
                }
                
                // Skip system tables
                if *oid < 16384 {
                    continue;
                }
                
                // Get all columns for this table
                for attr_row in &pg_attr.rows {
                    if let (Some(Value::Integer(attrelid)), Some(Value::Text(attname)), 
                            Some(Value::Integer(atttypid)), Some(Value::Boolean(attnotnull)),
                            Some(Value::Integer(attnum))) = 
                        (attr_row.get(0), attr_row.get(1), attr_row.get(2), 
                         attr_row.get(12), attr_row.get(5)) {
                        
                        // Match columns to this table
                        if *attrelid != *oid {
                            continue;
                        }
                        
                        // Skip system columns
                        if *attnum <= 0 {
                            continue;
                        }
                        
                        // Format type based on type OID
                        let format_type = match *atttypid {
                            16 => "boolean",
                            20 => "bigint",
                            21 => "smallint",
                            23 => "integer",
                            25 => "text",
                            700 => "real",
                            701 => "double precision",
                            1042 => "character",
                            1043 => "character varying",
                            1082 => "date",
                            1083 => "time without time zone",
                            1114 => "timestamp without time zone",
                            1184 => "timestamp with time zone",
                            1700 => "numeric",
                            2950 => "uuid",
                            _ => "text",
                        };
                        
                        result_rows.push(vec![
                            Value::Text(attname.clone()),                  // name
                            Value::Text(format_type.to_string()),          // format_type
                            Value::Null,                                    // default
                            Value::Boolean(*attnotnull),                   // not_null
                            Value::Text(relname.clone()),                  // table_name
                            Value::Null,                                    // comment
                            Value::Text("".to_string()),                   // generated
                            Value::Null,                                    // identity_options
                        ]);
                    }
                }
            }
        }
        
        Ok(QueryResult {
            columns: vec!["name".to_string(), "format_type".to_string(), 
                         "default".to_string(), "not_null".to_string(),
                         "table_name".to_string(), "comment".to_string(),
                         "generated".to_string(), "identity_options".to_string()],
            column_types: vec![SqlType::Text, SqlType::Text, SqlType::Text, SqlType::Boolean,
                              SqlType::Text, SqlType::Text, SqlType::Text, SqlType::Text],
            rows: result_rows,
        })
    }
    
    /// Route a parsed statement (used by extended protocol)
    pub fn route_statement(&self, statement: &Statement) -> crate::Result<Option<QueryResult>> {
        use tracing::debug;
        
        debug!("CatalogRouter routing parsed statement: {:?}", statement);
        
        // Check if this is a catalog query by examining the parsed AST
        if let Statement::Query(query) = statement {
            if let SetExpr::Select(select) = &*query.body {
                // Check if this is a catalog table query
                if let Some(table_ref) = select.from.first() {
                    if let TableFactor::Table { name, .. } = &table_ref.relation {
                        let table_name = name.0.iter()
                            .map(|ident| ident.value.as_str())
                            .collect::<Vec<_>>()
                            .join(".");
                        
                        debug!("Query is for table: {}", table_name);
                        
                        // Check if this is a catalog JOIN query
                        if select.from.len() > 1 || select.from.iter().any(|t| {
                            if let TableFactor::Table { .. } = &t.relation {
                                true
                            } else {
                                false
                            }
                        }) {
                            if table_name.contains("pg_attribute") || table_name.contains("pg_class") {
                                debug!("Detected catalog JOIN in parsed statement");
                                // For now, return all table columns for SQLAlchemy compatibility
                                return Ok(Some(self.get_all_table_columns()?));
                            }
                        }
                        
                        // Route to appropriate catalog handler based on table name
                        match table_name.as_str() {
                            "pg_catalog.pg_attribute" | "pg_attribute" => {
                                return Ok(Some(self.pg_catalog.query_pg_attribute(None)));
                            }
                            "pg_catalog.pg_class" | "pg_class" => {
                                // Check if there's a WHERE clause we need to handle
                                if let Some(where_expr) = &select.selection {
                                    // For now, handle the simple case: WHERE relname = 'X'
                                    // or WHERE relkind = 'r' AND relname = 'X'
                                    return Ok(Some(self.handle_pg_class_with_where(where_expr)?));
                                } else {
                                    return Ok(Some(self.pg_catalog.query_pg_class(None)));
                                }
                            }
                            "pg_catalog.pg_type" | "pg_type" => {
                                return Ok(Some(self.pg_catalog.query_pg_type(None)));
                            }
                            "pg_catalog.pg_namespace" | "pg_namespace" => {
                                return Ok(Some(self.pg_catalog.query_pg_namespace()));
                            }
                            "pg_catalog.pg_database" | "pg_database" => {
                                return Ok(Some(self.pg_catalog.query_pg_database()));
                            }
                            "pg_catalog.pg_tables" | "pg_tables" => {
                                return Ok(Some(self.pg_catalog.query_pg_tables()));
                            }
                            "pg_catalog.pg_statio_user_tables" | "pg_statio_user_tables" => {
                                return Ok(Some(self.pg_catalog.query_pg_statio_user_tables()));
                            }
                            "information_schema.tables" => {
                                return Ok(Some(self.information_schema.query_tables(None)));
                            }
                            "information_schema.columns" => {
                                return Ok(Some(self.information_schema.query_columns(None)));
                            }
                            "information_schema.schemata" => {
                                return Ok(Some(self.information_schema.query_schemata(None)));
                            }
                            _ => {}
                        }
                    }
                }
            }
        }
        
        Ok(None)
    }
    
    fn handle_pg_class_with_where(&self, where_expr: &sqlparser::ast::Expr) -> crate::Result<QueryResult> {
        
        // Get all pg_class rows
        let mut result = self.pg_catalog.query_pg_class(None);
        
        // Apply WHERE filtering
        let mut filtered_rows = Vec::new();
        
        for row in &result.rows {
            if self.evaluate_where_for_pg_class(where_expr, &row) {
                filtered_rows.push(row.clone());
            }
        }
        
        result.rows = filtered_rows;
        Ok(result)
    }
    
    fn evaluate_where_for_pg_class(&self, expr: &sqlparser::ast::Expr, row: &Vec<Value>) -> bool {
        
        match expr {
            Expr::BinaryOp { left, op, right } => {
                match op {
                    BinaryOperator::Eq => {
                        // Check if it's relname = 'something' or relkind = 'something'
                        if let (Expr::Identifier(ident), Expr::Value(sqlparser::ast::Value::SingleQuotedString(value))) = 
                            (left.as_ref(), right.as_ref()) {
                            
                            let col_name = ident.value.to_lowercase();
                            match col_name.as_str() {
                                "relname" => {
                                    // relname is column 1
                                    if let Some(Value::Text(relname)) = row.get(1) {
                                        return relname == value;
                                    }
                                }
                                "relkind" => {
                                    // relkind is column 2  
                                    if let Some(Value::Text(relkind)) = row.get(2) {
                                        return relkind == value;
                                    }
                                }
                                _ => {}
                            }
                        }
                    }
                    BinaryOperator::And => {
                        // Evaluate both sides
                        return self.evaluate_where_for_pg_class(left, row) && 
                               self.evaluate_where_for_pg_class(right, row);
                    }
                    _ => {}
                }
            }
            _ => {}
        }
        
        // Default to true if we can't evaluate
        true
    }
}