Query Layer and Optimization.

# query_optimization/partitioning.py

"""
Partitioning strategies for OLAP tables.

Partitioning divides data into manageable chunks that can be
queried independently. Good partitioning = orders of magnitude
faster queries.
"""

from dataclasses import dataclass
from typing import List, Optional
from datetime import date, timedelta
import zlib


@dataclass
class PartitionConfig:
    """Configuration for table partitioning."""
    column: str
    type: str  # "range", "hash", "list"
    granularity: Optional[str] = None  # For time: "day", "month", "year"


class PartitionPruner:
    """
    Determines which partitions to read for a query.
    """
    
    def __init__(self, partition_config: PartitionConfig):
        self.config = partition_config
    
    def get_partitions_for_query(
        self,
        filters: dict,
        all_partitions: List[str]
    ) -> List[str]:
        """
        Return only partitions needed for query filters.
        """
        
        if self.config.column not in filters:
            # No filter on partition column - read all
            return all_partitions
        
        filter_value = filters[self.config.column]
        
        if self.config.type == "range":
            return self._prune_range(filter_value, all_partitions)
        elif self.config.type == "list":
            return self._prune_list(filter_value, all_partitions)
        elif self.config.type == "hash":
            return self._prune_hash(filter_value, all_partitions)
        
        return all_partitions
    
    def _prune_range(self, filter_value, partitions: List[str]) -> List[str]:
        """Prune partitions for range filter."""
        
        # filter_value is tuple (start, end)
        start, end = filter_value
        
        matching = []
        for p in partitions:
            partition_value = self._extract_partition_value(p)
            if start <= partition_value <= end:
                matching.append(p)
        
        return matching
    
    def _prune_list(self, filter_value, partitions: List[str]) -> List[str]:
        """Prune partitions for IN list filter."""
        
        # filter_value is list of values
        matching = []
        for p in partitions:
            partition_value = self._extract_partition_value(p)
            if partition_value in filter_value:
                matching.append(p)
        
        return matching
    
    def _prune_hash(self, filter_value, partitions: List[str]) -> List[str]:
        """Prune partitions for hash-partitioned table."""
        
        # Calculate which partition the value hashes to.
        # Use a stable hash: Python's built-in hash() of a str is randomised
        # per process, so it would not match the buckets written at load time.
        hash_bucket = zlib.crc32(str(filter_value).encode()) % len(partitions)
        return [partitions[hash_bucket]]
    
    def _extract_partition_value(self, partition_path: str):
        """Extract partition value from path."""
        # e.g., "date=2024-01-15" -> "2024-01-15"
        return partition_path.split("=")[1]


def estimate_partition_savings(
    total_partitions: int,
    partitions_scanned: int,
    avg_partition_size_gb: float
) -> dict:
    """Estimate savings from partition pruning."""
    
    full_scan_gb = total_partitions * avg_partition_size_gb
    actual_scan_gb = partitions_scanned * avg_partition_size_gb
    
    return {
        "full_scan_gb": full_scan_gb,
        "actual_scan_gb": actual_scan_gb,
        "savings_gb": full_scan_gb - actual_scan_gb,
        "savings_pct": (1 - partitions_scanned / total_partitions) * 100,
        "partitions_skipped": total_partitions - partitions_scanned
    }

# query_optimization/clustering.py

"""
Clustering (sorting) within partitions for better query performance.

Partitioning decides WHICH files to read.
Clustering decides HOW data is organized WITHIN files.
"""

from typing import List


class ClusteringConfig:
    """Configure clustering for a table."""
    
    def __init__(
        self,
        columns: List[str],
        method: str = "sort",  # "sort", "zorder", "hilbert"
        auto_cluster: bool = True
    ):
        self.columns = columns
        self.method = method
        self.auto_cluster = auto_cluster


# BigQuery clustering example
BIGQUERY_CLUSTERING = """
-- Create clustered table in BigQuery
CREATE TABLE project.dataset.fact_sales
PARTITION BY DATE(order_timestamp)
CLUSTER BY customer_region, product_category  -- filter column first
AS SELECT * FROM source_table;

-- Query benefits from both partition AND cluster pruning
SELECT product_category, SUM(revenue)
FROM project.dataset.fact_sales
WHERE DATE(order_timestamp) = '2024-01-15'  -- Partition pruning
  AND customer_region = 'US-WEST'            -- Cluster pruning
GROUP BY product_category;
"""


# Delta Lake Z-ordering example
DELTA_ZORDER = """
-- Optimize Delta table with Z-ordering
OPTIMIZE fact_sales
ZORDER BY (product_id, customer_id);

-- This helps queries that filter on either or both columns
SELECT * FROM fact_sales
WHERE product_id = 'P123'
  AND customer_id = 'C456';
"""


# ClickHouse ordering
CLICKHOUSE_ORDER = """
-- ClickHouse uses ORDER BY for clustering
CREATE TABLE fact_sales (
    date Date,
    category String,
    region String,
    revenue Float64
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (category, region, date);  -- Data sorted by these columns

-- Queries on category, region are very fast
SELECT category, SUM(revenue)
FROM fact_sales
WHERE category = 'Electronics'
GROUP BY category;
"""

# query_optimization/rollups.py

"""
Rollup table implementation for pre-aggregation.

Rollups trade storage for query speed.
"""

from dataclasses import dataclass
from typing import List, Dict, Any, Optional
from datetime import date


@dataclass
class RollupDefinition:
    """Defines a rollup table."""
    name: str
    source_table: str
    dimensions: List[str]  # GROUP BY columns
    measures: Dict[str, str]  # column -> aggregation
    refresh_schedule: str
    retention_days: int = 365


class RollupManager:
    """
    Manages rollup table creation and refresh.
    """
    
    def __init__(self, db_connection):
        self.db = db_connection
        self.rollups: Dict[str, RollupDefinition] = {}
    
    def create_rollup(self, rollup: RollupDefinition):
        """Create a new rollup table."""
        
        # Build CREATE TABLE statement
        dimension_cols = ", ".join(rollup.dimensions)
        measure_cols = ", ".join([
            f"{agg}({col}) as {col}_{agg.lower()}"
            for col, agg in rollup.measures.items()
        ])
        
        sql = f"""
        CREATE TABLE {rollup.name} AS
        SELECT
            {dimension_cols},
            {measure_cols},
            CURRENT_TIMESTAMP as refreshed_at
        FROM {rollup.source_table}
        GROUP BY {dimension_cols}
        """
        
        self.db.execute(sql)
        self.rollups[rollup.name] = rollup
    
    def refresh_rollup(
        self,
        rollup_name: str,
        incremental: bool = True,
        target_date: date = None
    ):
        """
        Refresh a rollup table.
        
        Incremental: Only refresh specific date partition
        Full: Rebuild entire rollup
        """
        
        rollup = self.rollups[rollup_name]
        
        if incremental and target_date:
            # Delete and re-insert for specific date
            self.db.execute(f"""
                DELETE FROM {rollup.name}
                WHERE date = '{target_date}'
            """)
            
            dimension_cols = ", ".join(rollup.dimensions)
            measure_cols = ", ".join([
                f"{agg}({col}) as {col}_{agg.lower()}"
                for col, agg in rollup.measures.items()
            ])
            
            self.db.execute(f"""
                INSERT INTO {rollup.name}
                SELECT
                    {dimension_cols},
                    {measure_cols},
                    CURRENT_TIMESTAMP as refreshed_at
                FROM {rollup.source_table}
                WHERE date = '{target_date}'
                GROUP BY {dimension_cols}
            """)
        else:
            # Full refresh - recreate
            self.db.execute(f"DROP TABLE IF EXISTS {rollup.name}")
            self.create_rollup(rollup)
    
    def get_best_rollup_for_query(
        self,
        dimensions_needed: List[str],
        time_granularity: str
    ) -> Optional[str]:
        """
        Find the most efficient rollup for a query.
        
        Returns rollup name or None if should use base table.
        """
        
        candidates = []
        
        for name, rollup in self.rollups.items():
            # Check if rollup has all needed dimensions
            if all(d in rollup.dimensions for d in dimensions_needed):
                # Check if granularity matches
                if self._matches_granularity(rollup, time_granularity):
                    candidates.append((name, len(rollup.dimensions)))
        
        if not candidates:
            return None
        
        # Return rollup with fewest dimensions (most aggregated)
        candidates.sort(key=lambda x: x[1])
        return candidates[0][0]
    
    def _matches_granularity(self, rollup: RollupDefinition, granularity: str) -> bool:
        """Check if rollup supports the requested time granularity."""
        
        granularity_hierarchy = ["hour", "day", "week", "month", "quarter", "year"]
        
        rollup_granularity = "day"  # Default
        for dim in rollup.dimensions:
            if dim in granularity_hierarchy:
                rollup_granularity = dim
                break
        
        # Rollup must be at same or finer granularity
        return granularity_hierarchy.index(rollup_granularity) <= \
               granularity_hierarchy.index(granularity)

CREATE MATERIALIZED VIEW mv_daily_revenue AS
SELECT
    DATE_TRUNC('day', order_timestamp) as date,
    product_category,
    SUM(revenue) as total_revenue,
    COUNT(*) as order_count
FROM fact_sales
GROUP BY 1, 2;

-- Snowflake automatically:
-- 1. Keeps view in sync with base table
-- 2. Uses view when query matches pattern
-- 3. Charges for storage and maintenance

CREATE MATERIALIZED VIEW project.dataset.mv_daily_revenue
OPTIONS (
    enable_refresh = true,
    refresh_interval_minutes = 60
)
AS
SELECT
    DATE(order_timestamp) as date,
    product_category,
    SUM(revenue) as total_revenue,
    COUNT(*) as order_count
FROM project.dataset.fact_sales
GROUP BY 1, 2;

-- BigQuery automatically:
-- 1. Incrementally refreshes
-- 2. Rewrites queries to use MV
-- 3. Combines MV + base table for fresh data

-- ClickHouse MVs are triggers that populate a target table
CREATE MATERIALIZED VIEW mv_daily_revenue
ENGINE = SummingMergeTree()
ORDER BY (date, category)
AS
SELECT
    toDate(order_timestamp) as date,
    category,
    sum(revenue) as total_revenue,
    count() as order_count
FROM fact_sales
GROUP BY date, category;

-- ClickHouse:
-- 1. Populates MV on INSERT to source (each insert block aggregated separately)
-- 2. SummingMergeTree sums rows with the same key during background merges
--    (eventually), so still query with sum(...) GROUP BY date, category
-- 3. No automatic query rewrite - query MV directly

# query_optimization/materialized_views.py

"""
Materialized view strategies for different databases.
"""

from typing import Dict


class MaterializedViewManager:
    """
    Manages materialized views with freshness tracking.
    """
    
    def __init__(self, db_connection):
        self.db = db_connection
        self.views: Dict[str, Dict] = {}
    
    def create_view(
        self,
        name: str,
        query: str,
        refresh_interval_minutes: int = 60
    ):
        """Create a materialized view."""
        
        # Create the view
        self.db.execute(f"""
            CREATE MATERIALIZED VIEW {name} AS
            {query}
        """)
        
        self.views[name] = {
            "query": query,
            "refresh_interval": refresh_interval_minutes,
            "last_refresh": None,
            "row_count": 0
        }
    
    def get_freshness(self, name: str) -> Dict:
        """Get view freshness information."""
        
        if name not in self.views:
            return {"error": "View not found"}
        
        view = self.views[name]
        
        # Query for actual freshness
        result = self.db.execute(f"""
            SELECT 
                COUNT(*) as row_count,
                MAX(refreshed_at) as last_refresh
            FROM {name}
        """)
        
        return {
            "name": name,
            "last_refresh": result["last_refresh"],
            "row_count": result["row_count"],
            "refresh_interval_minutes": view["refresh_interval"],
            "is_stale": self._is_stale(result["last_refresh"], view["refresh_interval"])
        }
    
    def _is_stale(self, last_refresh, interval_minutes: int) -> bool:
        """Check if view is stale."""
        if not last_refresh:
            return True
        
        from datetime import datetime, timedelta, timezone
        threshold = datetime.now(timezone.utc) - timedelta(minutes=interval_minutes)
        return last_refresh < threshold

# query_optimization/caching.py

"""
Query result caching implementation.
"""

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Dict, Any, Optional
import hashlib
import json
import redis


@dataclass
class CacheConfig:
    """Cache configuration."""
    default_ttl_seconds: int = 300  # 5 minutes
    max_result_size_mb: int = 10
    redis_url: str = "redis://localhost:6379"


class QueryCache:
    """
    Caches query results for fast repeated access.
    """
    
    def __init__(self, config: CacheConfig):
        self.config = config
        self.redis = redis.from_url(config.redis_url)
        
        # Metrics
        self.hits = 0
        self.misses = 0
    
    def get(self, query: str, params: Dict = None) -> Optional[Dict]:
        """
        Get cached result for query.
        
        Returns None if not cached or expired.
        """
        
        cache_key = self._make_key(query, params)
        
        cached = self.redis.get(cache_key)
        
        if cached:
            self.hits += 1
            result = json.loads(cached)
            result["_cache"] = {
                "hit": True,
                "cached_at": result.get("_cached_at"),
                "key": cache_key[:20] + "..."
            }
            return result
        
        self.misses += 1
        return None
    
    def set(
        self,
        query: str,
        params: Dict,
        result: Dict,
        ttl_seconds: int = None
    ):
        """
        Cache a query result.
        """
        
        cache_key = self._make_key(query, params)
        ttl = ttl_seconds or self.config.default_ttl_seconds
        
        # Add metadata
        result["_cached_at"] = datetime.now(timezone.utc).isoformat()
        
        # Check size
        result_json = json.dumps(result)
        size_mb = len(result_json) / (1024 * 1024)
        
        if size_mb > self.config.max_result_size_mb:
            # Too large to cache
            return False
        
        self.redis.setex(cache_key, ttl, result_json)
        return True
    
    def invalidate(self, pattern: str = None, query: str = None):
        """
        Invalidate cached results.
        
        pattern: Redis key pattern (e.g., "query:sales:*")
        query: Specific query to invalidate
        """
        
        if query:
            cache_key = self._make_key(query, {})
            # Drops the entry cached with no params only: the key is a hash
            # that includes params, so other variants have different keys.
            # (SCAN, not KEYS: KEYS is O(N) and blocks Redis while it runs.)
            keys = list(self.redis.scan_iter(match=f"{cache_key}*"))
            if keys:
                self.redis.unlink(*keys)
        
        elif pattern:
            keys = list(self.redis.scan_iter(match=pattern))
            if keys:
                self.redis.unlink(*keys)
    
    def _make_key(self, query: str, params: Dict = None) -> str:
        """Generate cache key from query and params."""
        
        # Normalize query
        normalized = " ".join(query.lower().split())
        
        # Include params in key
        key_data = {
            "query": normalized,
            "params": params or {}
        }
        
        key_hash = hashlib.sha256(
            json.dumps(key_data, sort_keys=True).encode()
        ).hexdigest()[:16]
        
        return f"query:{key_hash}"
    
    def get_stats(self) -> Dict:
        """Get cache statistics."""
        
        total = self.hits + self.misses
        
        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": self.hits / total if total > 0 else 0,
            "total_requests": total
        }


class SmartQueryCache(QueryCache):
    """
    Smart caching with query-specific TTLs and warming.
    """
    
    def __init__(self, config: CacheConfig):
        super().__init__(config)
        
        # Query-specific TTL rules
        self.ttl_rules = {
            "realtime": 60,       # 1 minute for real-time queries
            "hourly": 300,        # 5 minutes for hourly data
            "daily": 3600,        # 1 hour for daily data
            "historical": 86400,  # 24 hours for historical
        }
    
    def get_ttl_for_query(self, query: str, params: Dict) -> int:
        """Determine appropriate TTL based on query characteristics."""
        
        query_lower = query.lower()
        
        # Real-time queries (today's data)
        if "today" in query_lower or "current" in query_lower:
            return self.ttl_rules["realtime"]
        
        # Check date range in params
        if params:
            if "date" in params:
                date_param = params["date"]
                # If querying old data, cache longer
                if isinstance(date_param, str) and date_param < "2024-01-01":
                    return self.ttl_rules["historical"]
        
        # Default to daily TTL
        return self.ttl_rules["daily"]
    
    async def warm_cache(self, queries: list):
        """
        Pre-warm cache with common queries.
        
        Run during off-peak hours.
        """
        
        for query_def in queries:
            query = query_def["query"]
            params = query_def.get("params", {})
            
            # Execute query and cache result
            result = await self._execute_query(query, params)
            
            ttl = self.get_ttl_for_query(query, params)
            self.set(query, params, result, ttl)
    
    async def _execute_query(self, query: str, params: Dict) -> Dict:
        """Execute query against database."""
        # Implementation depends on database
        pass

# query_optimization/multi_tenant.py

"""
Multi-tenant query management with fair resource allocation.
"""

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Dict, Optional
import asyncio


@dataclass
class TenantQuota:
    """Resource quota for a tenant."""
    max_concurrent_queries: int = 10
    max_scanned_bytes_per_day: int = 100 * 1024**3  # 100GB
    max_query_timeout_seconds: int = 300
    priority: int = 1  # Higher = more priority


class TenantResourceManager:
    """
    Manages query resources across tenants fairly.
    """
    
    def __init__(self):
        self.quotas: Dict[str, TenantQuota] = {}
        self.usage: Dict[str, Dict] = {}
        self.active_queries: Dict[str, int] = {}
    
    def register_tenant(self, tenant_id: str, quota: TenantQuota):
        """Register a tenant with their quota."""
        self.quotas[tenant_id] = quota
        self.usage[tenant_id] = {
            "bytes_scanned_today": 0,
            "queries_today": 0,
            "last_reset": datetime.now(timezone.utc).date()
        }
        self.active_queries[tenant_id] = 0
    
    async def acquire_query_slot(self, tenant_id: str) -> bool:
        """
        Try to acquire a query slot for tenant.
        
        Returns False if quota exceeded.
        """
        
        self._maybe_reset_daily_usage(tenant_id)
        
        quota = self.quotas.get(tenant_id)
        if not quota:
            return False
        
        # Check concurrent query limit
        if self.active_queries[tenant_id] >= quota.max_concurrent_queries:
            return False
        
        # Check daily byte limit
        if self.usage[tenant_id]["bytes_scanned_today"] >= quota.max_scanned_bytes_per_day:
            return False
        
        self.active_queries[tenant_id] += 1
        return True
    
    def release_query_slot(self, tenant_id: str, bytes_scanned: int):
        """Release query slot and record usage."""
        
        self.active_queries[tenant_id] -= 1
        self.usage[tenant_id]["bytes_scanned_today"] += bytes_scanned
        self.usage[tenant_id]["queries_today"] += 1
    
    def get_query_timeout(self, tenant_id: str) -> int:
        """Get query timeout for tenant."""
        quota = self.quotas.get(tenant_id)
        return quota.max_query_timeout_seconds if quota else 60
    
    def _maybe_reset_daily_usage(self, tenant_id: str):
        """Reset usage counters if new day."""
        today = datetime.now(timezone.utc).date()
        if self.usage[tenant_id]["last_reset"] < today:
            self.usage[tenant_id] = {
                "bytes_scanned_today": 0,
                "queries_today": 0,
                "last_reset": today
            }
    
    def get_usage_report(self, tenant_id: str) -> Dict:
        """Get usage report for tenant."""
        
        self._maybe_reset_daily_usage(tenant_id)
        
        quota = self.quotas.get(tenant_id)
        usage = self.usage.get(tenant_id, {})
        
        return {
            "tenant_id": tenant_id,
            "today": {
                "bytes_scanned": usage.get("bytes_scanned_today", 0),
                "bytes_limit": quota.max_scanned_bytes_per_day if quota else 0,
                "bytes_remaining": max(0, 
                    (quota.max_scanned_bytes_per_day if quota else 0) - 
                    usage.get("bytes_scanned_today", 0)
                ),
                "queries_executed": usage.get("queries_today", 0)
            },
            "active_queries": self.active_queries.get(tenant_id, 0),
            "max_concurrent": quota.max_concurrent_queries if quota else 0
        }


class QueryRouter:
    """
    Routes queries to appropriate resources based on tenant and query type.
    """
    
    def __init__(
        self,
        resource_manager: TenantResourceManager,
        query_cache: QueryCache
    ):
        self.resources = resource_manager
        self.cache = query_cache
        
        # Query executors by type
        self.executors = {}
    
    async def execute_query(
        self,
        tenant_id: str,
        query: str,
        params: Dict = None
    ) -> Dict:
        """
        Execute query with tenant isolation and resource management.
        """
        
        # Check cache first
        cached = self.cache.get(f"{tenant_id}:{query}", params)
        if cached:
            return cached
        
        # Acquire resources
        if not await self.resources.acquire_query_slot(tenant_id):
            return {
                "error": "Resource quota exceeded",
                "retry_after_seconds": 60
            }
        
        try:
            # Get timeout for tenant
            timeout = self.resources.get_query_timeout(tenant_id)
            
            # Execute with timeout
            result = await asyncio.wait_for(
                self._execute(tenant_id, query, params),
                timeout=timeout
            )
            
            # Cache result
            self.cache.set(f"{tenant_id}:{query}", params, result)
            
            # Release resources
            bytes_scanned = result.get("_metadata", {}).get("bytes_scanned", 0)
            self.resources.release_query_slot(tenant_id, bytes_scanned)
            
            return result
            
        except asyncio.TimeoutError:
            self.resources.release_query_slot(tenant_id, 0)
            return {"error": f"Query timeout after {timeout}s"}
        
        except Exception as e:
            self.resources.release_query_slot(tenant_id, 0)
            raise
    
    async def _execute(self, tenant_id: str, query: str, params: Dict) -> Dict:
        """Execute query with tenant context."""
        
        # Add tenant filter to query
        safe_query = self._add_tenant_filter(query, tenant_id)
        
        # Route to appropriate executor
        executor = self._get_executor(query)
        
        return await executor.execute(safe_query, params)
    
    def _add_tenant_filter(self, query: str, tenant_id: str) -> str:
        """
        Add tenant filter to query for row-level security.
        
        In production, use database-level RLS instead.
        """
        
        # Simple approach - inject WHERE clause
        # WARNING: This is simplified - use parameterized queries!
        if "WHERE" in query.upper():
            return query.replace("WHERE", f"WHERE tenant_id = '{tenant_id}' AND")
        else:
            # Find insertion point
            return query + f" WHERE tenant_id = '{tenant_id}'"
    
    def _get_executor(self, query: str):
        """Get appropriate executor based on query type."""
        
        query_lower = query.lower()
        
        if "sum(" in query_lower or "count(" in query_lower:
            return self.executors.get("olap", self.executors["default"])
        
        return self.executors["default"]

# query_optimization/query_service.py

"""
Production query service with all optimizations.
"""

from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Dict, Any, List, Optional
import logging

logger = logging.getLogger(__name__)


@dataclass
class QueryServiceConfig:
    """Query service configuration."""
    cache_enabled: bool = True
    cache_ttl_seconds: int = 300
    max_result_rows: int = 10000
    default_timeout_seconds: int = 30
    enable_query_rewriting: bool = True


class QueryService:
    """
    Production query service with caching, optimization, and monitoring.
    """
    
    def __init__(
        self,
        config: QueryServiceConfig,
        cache: QueryCache,
        rollup_manager: RollupManager,
        resource_manager: TenantResourceManager,
        db_connection
    ):
        self.config = config
        self.cache = cache
        self.rollups = rollup_manager
        self.resources = resource_manager
        self.db = db_connection
        
        # Metrics
        self.query_count = 0
        self.cache_hits = 0
        self.rollup_hits = 0
        self.avg_latency_ms = 0
    
    async def execute(
        self,
        tenant_id: str,
        query: str,
        params: Dict = None,
        options: Dict = None
    ) -> Dict:
        """
        Execute query with full optimization pipeline.
        
        Pipeline:
        1. Check cache
        2. Acquire resources
        3. Rewrite query (use rollups)
        4. Execute
        5. Cache result
        6. Return with metadata
        """
        
        start_time = datetime.now(timezone.utc)
        self.query_count += 1
        
        options = options or {}
        result_metadata = {
            "query_id": f"q_{self.query_count}_{start_time.timestamp()}",
            "tenant_id": tenant_id,
            "cache_hit": False,
            "rollup_used": None,
            "bytes_scanned": 0
        }
        
        try:
            # Step 1: Check cache
            if self.config.cache_enabled and not options.get("skip_cache"):
                cache_key = f"{tenant_id}:{query}"
                cached = self.cache.get(cache_key, params)
                
                if cached:
                    self.cache_hits += 1
                    result_metadata["cache_hit"] = True
                    result_metadata["latency_ms"] = self._elapsed_ms(start_time)
                    return self._format_response(cached, result_metadata)
            
            # Step 2: Acquire resources
            if not await self.resources.acquire_query_slot(tenant_id):
                return self._quota_exceeded_response(tenant_id)
            
            try:
                # Step 3: Query rewriting
                optimized_query = query
                if self.config.enable_query_rewriting:
                    optimized_query, rollup_used = self._try_rewrite_to_rollup(
                        query, params
                    )
                    if rollup_used:
                        self.rollup_hits += 1
                        result_metadata["rollup_used"] = rollup_used
                
                # Step 4: Add tenant filter
                safe_query = self._add_tenant_filter(optimized_query, tenant_id)
                
                # Step 5: Execute with timeout
                timeout = options.get("timeout", self.config.default_timeout_seconds)
                result = await self._execute_query(safe_query, params, timeout)
                
                result_metadata["bytes_scanned"] = result.get("_bytes_scanned", 0)
                result_metadata["rows_returned"] = len(result.get("data", []))
                
                # Step 6: Cache result
                if self.config.cache_enabled:
                    ttl = self._get_cache_ttl(query, params)
                    self.cache.set(f"{tenant_id}:{query}", params, result, ttl)
                
                # Release resources
                self.resources.release_query_slot(
                    tenant_id,
                    result_metadata["bytes_scanned"]
                )
                
                result_metadata["latency_ms"] = self._elapsed_ms(start_time)
                self._update_avg_latency(result_metadata["latency_ms"])
                
                return self._format_response(result, result_metadata)
                
            except Exception as e:
                self.resources.release_query_slot(tenant_id, 0)
                raise
        
        except Exception as e:
            logger.error(f"Query execution failed: {e}", exc_info=True)
            result_metadata["latency_ms"] = self._elapsed_ms(start_time)
            result_metadata["error"] = str(e)
            return self._error_response(e, result_metadata)
    
    def _try_rewrite_to_rollup(
        self,
        query: str,
        params: Dict
    ) -> tuple:
        """
        Try to rewrite query to use a rollup table.
        
        Returns (rewritten_query, rollup_name) or (original_query, None)
        """
        
        # Parse query to understand dimensions and granularity
        dimensions = self._extract_dimensions(query)
        granularity = self._extract_time_granularity(query, params)
        
        # Find best rollup
        rollup_name = self.rollups.get_best_rollup_for_query(
            dimensions, granularity
        )
        
        if rollup_name:
            # Rewrite query to use rollup
            rewritten = self._rewrite_query_for_rollup(query, rollup_name)
            return rewritten, rollup_name
        
        return query, None
    
    def _extract_dimensions(self, query: str) -> List[str]:
        """Extract GROUP BY dimensions from query."""
        
        # Simplified parsing - use proper SQL parser in production
        query_upper = query.upper()
        
        if "GROUP BY" not in query_upper:
            return []
        
        group_by_idx = query_upper.index("GROUP BY")
        group_by_clause = query[group_by_idx + 8:].split("ORDER BY")[0].split("LIMIT")[0]
        
        dimensions = [d.strip() for d in group_by_clause.split(",")]
        return dimensions
    
    def _extract_time_granularity(self, query: str, params: Dict) -> str:
        """Determine time granularity of query."""
        
        query_lower = query.lower()
        
        if "hour" in query_lower or "hourly" in query_lower:
            return "hour"
        elif "day" in query_lower or "daily" in query_lower:
            return "day"
        elif "week" in query_lower or "weekly" in query_lower:
            return "week"
        elif "month" in query_lower or "monthly" in query_lower:
            return "month"
        
        return "day"  # Default
    
    def _rewrite_query_for_rollup(self, query: str, rollup_name: str) -> str:
        """Rewrite query to use rollup table."""
        
        # Replace fact table with rollup
        # Simplified - use proper SQL rewriting in production
        rewritten = query.replace("fact_sales", rollup_name)
        rewritten = rewritten.replace("fact_orders", rollup_name)
        
        return rewritten
    
    def _add_tenant_filter(self, query: str, tenant_id: str) -> str:
        """Add tenant isolation filter."""
        
        # In production, use database-level RLS
        # This is a simplified approach
        
        if "WHERE" in query.upper():
            # Insert after WHERE
            where_idx = query.upper().index("WHERE")
            return (
                query[:where_idx + 5] +
                f" tenant_id = '{tenant_id}' AND" +
                query[where_idx + 5:]
            )
        else:
            # Add WHERE clause
            # Find safe insertion point
            for keyword in ["GROUP BY", "ORDER BY", "LIMIT", ";"]:
                if keyword in query.upper():
                    idx = query.upper().index(keyword)
                    return (
                        query[:idx] +
                        f" WHERE tenant_id = '{tenant_id}' " +
                        query[idx:]
                    )
            
            return query + f" WHERE tenant_id = '{tenant_id}'"
    
    async def _execute_query(
        self,
        query: str,
        params: Dict,
        timeout: int
    ) -> Dict:
        """Execute query against database."""
        
        # Add LIMIT if not present
        if "LIMIT" not in query.upper():
            query = query + f" LIMIT {self.config.max_result_rows}"
        
        # Execute with parameterized query
        result = await self.db.execute(query, params, timeout=timeout)
        
        return result
    
    def _get_cache_ttl(self, query: str, params: Dict) -> int:
        """Determine cache TTL based on query characteristics."""
        
        query_lower = query.lower()
        
        # Real-time queries: short TTL
        if "today" in query_lower or "last_hour" in query_lower:
            return 60
        
        # Historical queries: long TTL
        if params and params.get("end_date", "").startswith("2023"):
            return 3600
        
        return self.config.cache_ttl_seconds
    
    def _elapsed_ms(self, start: datetime) -> int:
        """Calculate elapsed time in milliseconds."""
        return int((datetime.now(timezone.utc) - start).total_seconds() * 1000)
    
    def _update_avg_latency(self, latency_ms: int):
        """Update rolling average latency."""
        # Simple exponential moving average
        alpha = 0.1
        self.avg_latency_ms = alpha * latency_ms + (1 - alpha) * self.avg_latency_ms
    
    def _format_response(self, result: Dict, metadata: Dict) -> Dict:
        """Format successful response."""
        return {
            "data": result.get("data", []),
            "metadata": metadata,
            "status": "success"
        }
    
    def _error_response(self, error: Exception, metadata: Dict) -> Dict:
        """Format error response."""
        return {
            "data": [],
            "metadata": metadata,
            "status": "error",
            "error": {
                "type": type(error).__name__,
                "message": str(error)
            }
        }
    
    def _quota_exceeded_response(self, tenant_id: str) -> Dict:
        """Format quota exceeded response."""
        usage = self.resources.get_usage_report(tenant_id)
        
        return {
            "data": [],
            "metadata": {"tenant_id": tenant_id},
            "status": "error",
            "error": {
                "type": "QuotaExceeded",
                "message": "Query quota exceeded",
                "usage": usage,
                "retry_after_seconds": 60
            }
        }
    
    def get_stats(self) -> Dict:
        """Get service statistics."""
        return {
            "total_queries": self.query_count,
            "cache_hit_rate": self.cache_hits / self.query_count if self.query_count > 0 else 0,
            "rollup_hit_rate": self.rollup_hits / self.query_count if self.query_count > 0 else 0,
            "avg_latency_ms": self.avg_latency_ms,
            "cache_stats": self.cache.get_stats()
        }