Traditional Approach:
  1. User requests "SELECT * FROM orders"
  2. Database returns 1 million rows
  3. Application layer checks permissions row by row
  4. Returns 1000 filtered rows to user

Problems:
  - Database transmits 99.9% useless data
  - Massive application-layer memory usage
  - Row-by-row permission check grows O(n) linearly
  - Pagination, sorting, aggregation all redo in application layer

Query Rewrite Approach:
  1. User requests "SELECT * FROM orders"
  2. Policy engine rewrites request to:
     "SELECT * FROM orders WHERE region IN ('east', 'south')
      AND classification <= 'C2'"
  3. Database returns only 1000 authorized rows
  4. Pagination, sorting, aggregation handled efficiently by database

Advantages:
  - Database processes only authorized data
  - Leverages indexes for filter acceleration
  - Memory usage reduced by 99%+
  - Native pagination/sorting/aggregation support

┌──────────────┐     ┌──────────────────┐     ┌──────────────┐
│  User Query   │────>│  Policy Compiler  │────>│  Rewritten   │
│  Original    │     │                  │     │  Query       │
│  Query       │     │                  │     │              │
└──────────────┘     └────────┬─────────┘     └──────┬───────┘
                              │                       │
                     ┌────────▼─────────┐             │
                     │  Policy Store     │             │
                     │  - RBAC Roles    │             │
                     │  - ABAC Attrs    │             ▼
                     │  - ReBAC Rels    │     ┌──────────────┐
                     └──────────────────┘     │  Database    │
                                              │  (Doris/PG)  │
                                              └──────────────┘

class PolicyCompiler:
    """Policy compiler: transforms permission policies into SQL conditions"""

    def __init__(
        self,
        rbac_engine: RBACEngine,
        abac_engine: ABACEngine,
        rebac_engine: ReBACheckEngine,
        schema_registry: SchemaRegistry,
    ):
        self._rbac = rbac_engine
        self._abac = abac_engine
        self._rebac = rebac_engine
        self._schema = schema_registry

    async def compile(
        self,
        query: ParsedQuery,
        subject: AuthenticatedUser,
        context: RequestContext,
    ) -> RewrittenQuery:
        """
        Compilation pipeline:
        1. Parse original query, extract involved tables and columns
        2. Look up row-level policies for each table
        3. Look up column-level policies for each column
        4. Compile policies into SQL conditions
        5. Inject WHERE clauses and SELECT column filters
        """
        # Step 1: Parse query
        tables = query.extract_tables()
        columns = query.extract_columns()

        # Step 2: Row-level policy compilation
        row_filters = []
        for table in tables:
            policies = await self._get_row_policies(table, subject, context)
            for policy in policies:
                sql_condition = self._compile_policy_to_sql(policy, table)
                row_filters.append(sql_condition)

        # Step 3: Column-level policy compilation
        column_masks = {}
        for table, col in columns:
            mask = await self._get_column_mask(table, col, subject, context)
            if mask:
                column_masks[(table, col)] = mask

        # Step 4: Rewrite query
        rewritten = self._rewrite_query(query, row_filters, column_masks)

        return RewrittenQuery(
            sql=rewritten.sql,
            params=rewritten.params,
            audit_info=AuditInfo(
                original_query=query.sql,
                applied_policies=[p.id for p in row_filters],
                applied_masks=list(column_masks.keys()),
            ),
        )

# Policy definition examples
policies:
  - id: "pol-region-isolation"
    name: "Region Data Isolation"
    description: "Users can only see data from their own region"
    target:
      object_types: ["order", "customer", "inventory"]
    condition:
      operator: "in"
      field: "resource.region"
      value_source: "subject.allowed_regions"
    effect: "FILTER"

  - id: "pol-classification-ceiling"
    name: "Classification Level Ceiling"
    description: "Users can only see data up to their clearance level"
    target:
      object_types: ["*"]
    condition:
      operator: "lte"
      field: "resource.classification_level"
      value_source: "subject.clearance_level"
    effect: "FILTER"

  - id: "pol-owner-full-access"
    name: "Data Owner Full Access"
    description: "Data creators have full access"
    target:
      object_types: ["*"]
    condition:
      operator: "eq"
      field: "resource.owner_id"
      value_source: "subject.user_id"
    effect: "ALLOW_ALL"

  - id: "pol-time-restriction"
    name: "Business Hours Restriction"
    description: "Sensitive data accessible only during business hours"
    target:
      object_types: ["sensitive_report"]
      classification_min: "B1"
    condition:
      operator: "and"
      conditions:
        - operator: "gte"
          field: "environment.hour"
          value: 9
        - operator: "lte"
          field: "environment.hour"
          value: 18
        - operator: "in"
          field: "environment.day_of_week"
          value: [1, 2, 3, 4, 5]
    effect: "FILTER"

class ExpressionCompiler:
    """Compile policy expressions to SQL conditions"""

    def compile_condition(
        self,
        condition: PolicyCondition,
        table_alias: str,
        subject: AuthenticatedUser,
        context: RequestContext,
    ) -> SQLFragment:
        """Recursively compile policy conditions to SQL fragments"""

        match condition.operator:
            case "eq":
                value = self._resolve_value(condition.value_source, subject, context)
                return SQLFragment(
                    sql=f"{table_alias}.{condition.field} = %s",
                    params=[value],
                )

            case "neq":
                value = self._resolve_value(condition.value_source, subject, context)
                return SQLFragment(
                    sql=f"{table_alias}.{condition.field} != %s",
                    params=[value],
                )

            case "in":
                values = self._resolve_value(condition.value_source, subject, context)
                placeholders = ", ".join(["%s"] * len(values))
                return SQLFragment(
                    sql=f"{table_alias}.{condition.field} IN ({placeholders})",
                    params=values,
                )

            case "not_in":
                values = self._resolve_value(condition.value_source, subject, context)
                placeholders = ", ".join(["%s"] * len(values))
                return SQLFragment(
                    sql=f"{table_alias}.{condition.field} NOT IN ({placeholders})",
                    params=values,
                )

            case "lte":
                value = self._resolve_value(condition.value_source, subject, context)
                return SQLFragment(
                    sql=f"{table_alias}.{condition.field} <= %s",
                    params=[value],
                )

            case "gte":
                value = self._resolve_value(condition.value_source, subject, context)
                return SQLFragment(
                    sql=f"{table_alias}.{condition.field} >= %s",
                    params=[value],
                )

            case "between":
                low = self._resolve_value(condition.low_source, subject, context)
                high = self._resolve_value(condition.high_source, subject, context)
                return SQLFragment(
                    sql=f"{table_alias}.{condition.field} BETWEEN %s AND %s",
                    params=[low, high],
                )

            case "like":
                pattern = self._resolve_value(condition.value_source, subject, context)
                return SQLFragment(
                    sql=f"{table_alias}.{condition.field} LIKE %s",
                    params=[pattern],
                )

            case "is_null":
                return SQLFragment(
                    sql=f"{table_alias}.{condition.field} IS NULL",
                    params=[],
                )

            case "is_not_null":
                return SQLFragment(
                    sql=f"{table_alias}.{condition.field} IS NOT NULL",
                    params=[],
                )

            case "and":
                children = [
                    self.compile_condition(c, table_alias, subject, context)
                    for c in condition.conditions
                ]
                sql = " AND ".join(f"({c.sql})" for c in children)
                params = [p for c in children for p in c.params]
                return SQLFragment(sql=f"({sql})", params=params)

            case "or":
                children = [
                    self.compile_condition(c, table_alias, subject, context)
                    for c in condition.conditions
                ]
                sql = " OR ".join(f"({c.sql})" for c in children)
                params = [p for c in children for p in c.params]
                return SQLFragment(sql=f"({sql})", params=params)

            case "not":
                child = self.compile_condition(
                    condition.conditions[0], table_alias, subject, context
                )
                return SQLFragment(sql=f"NOT ({child.sql})", params=child.params)

    def _resolve_value(
        self,
        source: str,
        subject: AuthenticatedUser,
        context: RequestContext,
    ) -> Any:
        """Resolve value source"""
        if source.startswith("subject."):
            attr = source[len("subject."):]
            return getattr(subject, attr)
        elif source.startswith("environment."):
            attr = source[len("environment."):]
            return context.get_env(attr)
        elif source.startswith("constant."):
            return source[len("constant."):]
        else:
            return source  # Literal value

User Query: SELECT * FROM orders WHERE status = 'pending'

After Policy Compilation:
  SELECT * FROM orders
  WHERE status = 'pending'                    -- User's original condition
    AND region IN ('east', 'south')            -- Region isolation policy
    AND classification_level <= 3              -- Classification level policy
    AND (owner_id = 'user123'                  -- Owner policy
         OR department_id IN (10, 20, 30))     -- Department visibility policy

class RowLevelSecurityRewriter:
    """Row-level security rewriter"""

    async def apply_rls(
        self,
        query: ParsedQuery,
        table: TableRef,
        subject: AuthenticatedUser,
        context: RequestContext,
    ) -> list[SQLFragment]:
        """
        Apply row-level security policies.

        Merging logic:
        - Conditions within same policy group: OR (satisfy any)
        - Conditions across different groups: AND (satisfy all)
        - ALLOW_ALL policy can bypass subsequent FILTER policies
        """
        applicable_policies = await self._find_applicable_policies(
            table.object_type, subject, context
        )

        if not applicable_policies:
            # No matching policies -> default deny all rows
            return [SQLFragment(sql="1 = 0", params=[])]

        # Check for ALLOW_ALL policies
        for policy in applicable_policies:
            if policy.effect == "ALLOW_ALL":
                condition = self._compiler.compile_condition(
                    policy.condition, table.alias, subject, context
                )
                return []  # No additional filtering needed

        # Group by policy group
        groups: dict[str, list[SQLFragment]] = defaultdict(list)
        for policy in applicable_policies:
            compiled = self._compiler.compile_condition(
                policy.condition, table.alias, subject, context
            )
            groups[policy.group_id].append(compiled)

        # OR within groups, AND across groups
        group_conditions = []
        for group_id, conditions in groups.items():
            if len(conditions) == 1:
                group_conditions.append(conditions[0])
            else:
                sql = " OR ".join(f"({c.sql})" for c in conditions)
                params = [p for c in conditions for p in c.params]
                group_conditions.append(SQLFragment(sql=f"({sql})", params=params))

        return group_conditions

class ReBACRowFilter:
    """ReBAC relationship-based row-level filtering"""

    async def compile_rebac_filter(
        self,
        table: TableRef,
        subject: SubjectRef,
        permission: str,
    ) -> SQLFragment:
        """
        Convert ReBAC relationships to SQL row filter conditions.

        Strategy: Pre-computation + Materialized Views
        - Periodically expand ReBAC relationships into (user, resource) pairs
        - Store in materialized view for JOIN
        """
        return SQLFragment(
            sql=f"""
                {table.alias}.id IN (
                    SELECT resource_id
                    FROM rebac_materialized_permissions
                    WHERE subject_type = %s
                      AND subject_id = %s
                      AND resource_type = %s
                      AND permission = %s
                )
            """,
            params=[subject.type, subject.id, table.object_type, permission],
        )

    async def compile_rebac_filter_realtime(
        self,
        table: TableRef,
        subject: SubjectRef,
        permission: str,
    ) -> SQLFragment:
        """
        Real-time ReBAC filtering (for frequently changing relationships)
        Pre-query accessible resource IDs via LookupResources
        """
        accessible_ids = []
        async for resource in self._rebac_engine.lookup_resources(
            resource_type=table.object_type,
            permission=permission,
            subject=subject,
        ):
            accessible_ids.append(resource.id)

        if not accessible_ids:
            return SQLFragment(sql="1 = 0", params=[])

        if len(accessible_ids) > 1000:
            temp_table = await self._create_temp_table(accessible_ids)
            return SQLFragment(
                sql=f"{table.alias}.id IN (SELECT id FROM {temp_table})",
                params=[],
            )

        placeholders = ", ".join(["%s"] * len(accessible_ids))
        return SQLFragment(
            sql=f"{table.alias}.id IN ({placeholders})",
            params=accessible_ids,
        )

class ColumnPolicy:
    """Column-level security policy"""

    column: str
    object_type: str
    action: ColumnAction  # MASK / HIDE / TRANSFORM / ALLOW

class ColumnAction(Enum):
    ALLOW = "allow"        # Fully visible
    HIDE = "hide"          # Completely hidden (removed from SELECT)
    MASK = "mask"          # Masked display
    TRANSFORM = "transform"  # Custom transformation

class ColumnLevelSecurityRewriter:
    """Column-level security rewriter"""

    async def rewrite_columns(
        self,
        query: ParsedQuery,
        subject: AuthenticatedUser,
        context: RequestContext,
    ) -> ParsedQuery:
        """Rewrite query's SELECT list"""
        new_columns = []

        for col in query.select_columns:
            policy = await self._get_column_policy(
                col.table, col.name, subject, context
            )

            match policy.action:
                case ColumnAction.ALLOW:
                    new_columns.append(col)

                case ColumnAction.HIDE:
                    continue  # Remove column entirely

                case ColumnAction.MASK:
                    mask_expr = self._get_mask_expression(
                        col, policy.mask_type, policy.mask_config
                    )
                    new_columns.append(
                        ColumnExpr(expression=mask_expr, alias=col.name)
                    )

                case ColumnAction.TRANSFORM:
                    transform_expr = policy.transform_expression
                    new_columns.append(
                        ColumnExpr(expression=transform_expr, alias=col.name)
                    )

        query.select_columns = new_columns
        return query

    def _get_mask_expression(
        self,
        col: ColumnRef,
        mask_type: str,
        config: dict,
    ) -> str:
        """Generate masking SQL expression"""
        full_name = f"{col.table_alias}.{col.name}"

        match mask_type:
            case "partial":
                prefix = config.get("prefix_length", 3)
                suffix = config.get("suffix_length", 4)
                mask_char = config.get("mask_char", "*")
                return (
                    f"CONCAT("
                    f"  LEFT({full_name}, {prefix}), "
                    f"  REPEAT('{mask_char}', GREATEST(LENGTH({full_name}) - {prefix + suffix}, 0)), "
                    f"  RIGHT({full_name}, {suffix})"
                    f")"
                )

            case "hash":
                return f"MD5({full_name})"

            case "null":
                return "NULL"

            case "constant":
                value = config.get("value", "***")
                return f"'{value}'"

            case "range":
                bucket_size = config.get("bucket_size", 10)
                return f"FLOOR({full_name} / {bucket_size}) * {bucket_size}"

            case "date_truncate":
                precision = config.get("precision", "month")
                return f"DATE_TRUNC('{precision}', {full_name})"

class SafeSQLBuilder:
    """Safe SQL builder with enforced parameterization"""

    ALLOWED_COLUMN_PATTERN = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_.]*$')
    ALLOWED_TABLE_PATTERN = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_.]*$')

    def validate_identifier(self, identifier: str, kind: str) -> str:
        """Validate SQL identifier (table name, column name) is safe"""
        pattern = (
            self.ALLOWED_TABLE_PATTERN if kind == "table"
            else self.ALLOWED_COLUMN_PATTERN
        )
        if not pattern.match(identifier):
            raise UnsafeIdentifierError(
                f"Unsafe {kind} identifier: {identifier}"
            )
        return identifier

    def build_where_clause(self, fragments: list[SQLFragment]) -> SQLFragment:
        """Safely merge multiple WHERE conditions"""
        if not fragments:
            return SQLFragment(sql="1 = 1", params=[])

        sql_parts = []
        all_params = []
        for fragment in fragments:
            sql_parts.append(f"({fragment.sql})")
            all_params.extend(fragment.params)

        return SQLFragment(
            sql=" AND ".join(sql_parts),
            params=all_params,
        )

class QueryAuditor:
    """Query auditor: records all rewrite operations"""

    async def audit_rewrite(
        self,
        original_query: str,
        rewritten_query: str,
        subject: AuthenticatedUser,
        applied_policies: list[str],
        context: RequestContext,
    ):
        """Record query rewrite audit log"""
        audit_event = QueryRewriteAuditEvent(
            timestamp=datetime.utcnow(),
            user_id=subject.user_id,
            original_query_hash=hashlib.sha256(original_query.encode()).hexdigest(),
            applied_policies=applied_policies,
            source_ip=context.source_ip,
            query_type=self._classify_query(original_query),
            tables_accessed=self._extract_tables(original_query),
            row_filters_applied=len(applied_policies),
            column_masks_applied=len([p for p in applied_policies if p.startswith("col-")]),
        )

        await self._audit_store.write(audit_event)

class SQLDialect(ABC):
    """SQL dialect abstract base class"""

    @abstractmethod
    def quote_identifier(self, name: str) -> str:
        """Quote identifier"""

    @abstractmethod
    def param_placeholder(self, index: int) -> str:
        """Parameter placeholder"""

    @abstractmethod
    def concat_function(self, *args: str) -> str:
        """String concatenation"""

    @abstractmethod
    def date_trunc(self, precision: str, column: str) -> str:
        """Date truncation"""

    @abstractmethod
    def md5_hash(self, column: str) -> str:
        """MD5 hash"""


class DorisDialect(SQLDialect):
    """Apache Doris SQL dialect"""

    def quote_identifier(self, name: str) -> str:
        return f"`{name}`"

    def param_placeholder(self, index: int) -> str:
        return "%s"

    def concat_function(self, *args: str) -> str:
        return f"CONCAT({', '.join(args)})"

    def date_trunc(self, precision: str, column: str) -> str:
        return f"DATE_TRUNC({column}, '{precision}')"

    def md5_hash(self, column: str) -> str:
        return f"MD5({column})"


class PostgreSQLDialect(SQLDialect):
    """PostgreSQL SQL dialect"""

    def quote_identifier(self, name: str) -> str:
        return f'"{name}"'

    def param_placeholder(self, index: int) -> str:
        return f"${index}"

    def concat_function(self, *args: str) -> str:
        return " || ".join(args)

    def date_trunc(self, precision: str, column: str) -> str:
        return f"DATE_TRUNC('{precision}', {column})"

    def md5_hash(self, column: str) -> str:
        return f"MD5({column}::text)"

class OQLPolicyIntegration:
    """OQL query policy integration"""

    async def rewrite_oql(
        self,
        oql: OQLQuery,
        subject: AuthenticatedUser,
        context: RequestContext,
    ) -> OQLQuery:
        """
        OQL query rewrite flow:
        1. OQL -> AST
        2. Inject policy conditions at AST level
        3. AST -> target SQL (with policy conditions)
        """
        ast = self._parser.parse(oql)

        for object_type_ref in ast.referenced_types():
            policies = await self._policy_store.get_policies(
                object_type=object_type_ref.type_name,
                subject=subject,
            )
            for policy in policies:
                ast.inject_filter(object_type_ref, policy.to_ast_condition())

        return ast.to_oql()

class MaterializedPermissionView:
    """Materialized permission view manager"""

    async def refresh(self, object_type: str):
        """Refresh materialized permission view"""
        await self._db.execute(f"""
            REFRESH MATERIALIZED VIEW CONCURRENTLY
            mv_permissions_{object_type}
        """)

    async def create_materialized_view(self, object_type: str):
        """Create materialized permission view"""
        await self._db.execute(f"""
            CREATE MATERIALIZED VIEW mv_permissions_{object_type} AS
            SELECT
                rt.resource_id,
                rt.subject_type,
                rt.subject_id,
                rt.relation AS permission,
                rt.created_at
            FROM relation_tuples rt
            WHERE rt.resource_type = '{object_type}'
              AND rt.deleted_at IS NULL
            UNION
            -- Expand inherited relationships
            SELECT
                child.resource_id,
                parent_rt.subject_type,
                parent_rt.subject_id,
                parent_rt.relation AS permission,
                parent_rt.created_at
            FROM relation_tuples child
            JOIN relation_tuples parent_rt
              ON child.subject_type = parent_rt.resource_type
             AND child.subject_id = parent_rt.resource_id
            WHERE child.resource_type = '{object_type}'
              AND child.deleted_at IS NULL
              AND parent_rt.deleted_at IS NULL
        """)

        await self._db.execute(f"""
            CREATE INDEX idx_mv_perm_{object_type}_subject
            ON mv_permissions_{object_type} (subject_type, subject_id);

            CREATE INDEX idx_mv_perm_{object_type}_resource
            ON mv_permissions_{object_type} (resource_id);
        """)

class IncrementalPermissionUpdater:
    """Incremental permission updater"""

    async def on_tuple_change(self, change: TupleChange):
        """When relationship tuples change, incrementally update materialized views"""
        affected_types = await self._get_affected_types(change)

        for obj_type in affected_types:
            if change.operation == "TOUCH":
                await self._add_permission_entries(obj_type, change)
            elif change.operation == "DELETE":
                await self._remove_permission_entries(obj_type, change)

        await self._invalidate_query_cache(affected_types)

    async def _get_affected_types(self, change: TupleChange) -> list[str]:
        """Compute affected object types"""
        affected = {change.resource_type}
        child_types = await self._type_system.get_child_types(
            change.resource_type
        )
        affected.update(child_types)
        return list(affected)

class PolicyCompilationCache:
    """Policy compilation result cache"""

    def __init__(self, max_size: int = 5000, ttl: int = 300):
        self._cache = TTLCache(maxsize=max_size, ttl=ttl)

    def cache_key(
        self,
        table: str,
        subject_fingerprint: str,
        context_fingerprint: str,
    ) -> str:
        """Generate cache key"""
        return f"pqr:{table}:{subject_fingerprint}:{context_fingerprint}"

    def get(self, key: str) -> Optional[list[SQLFragment]]:
        return self._cache.get(key)

    def set(self, key: str, fragments: list[SQLFragment]):
        self._cache[key] = fragments

Test Environment: 1M row orders table, 5 policy rules

| Method                     | Latency (p50) | Latency (p99) | Peak Memory |
|---------------------------|--------------|--------------|-------------|
| Application-layer filter   | 2400ms       | 5800ms       | 1.2 GB      |
| Query rewrite (no cache)   | 35ms         | 120ms        | 45 MB       |
| Query rewrite (cached)     | 28ms         | 95ms         | 45 MB       |
| Query rewrite (mat. view)  | 15ms         | 55ms         | 30 MB       |
| Database native RLS        | 12ms         | 48ms         | 28 MB       |

Conclusion: Query rewrite vs application-layer: 98% latency reduction, 96% memory reduction

class QueryPlanAnalyzer:
    """Query plan analyzer: ensure rewritten queries use indexes"""

    async def analyze_rewritten_query(
        self,
        original: str,
        rewritten: str,
        params: list,
    ) -> QueryPlanReport:
        """Analyze execution plan of rewritten query"""
        plan = await self._db.execute(f"EXPLAIN ANALYZE {rewritten}", params)

        report = QueryPlanReport(
            uses_index=self._check_index_usage(plan),
            estimated_rows=self._extract_row_estimate(plan),
            actual_rows=self._extract_actual_rows(plan),
            seq_scan_tables=self._find_seq_scans(plan),
            total_cost=self._extract_total_cost(plan),
        )

        if report.seq_scan_tables:
            logger.warning(
                "Policy rewrite caused sequential scan on tables: %s. "
                "Consider adding indexes for policy filter columns.",
                report.seq_scan_tables,
            )

        return report

class OntologyAwarePolicyResolver:
    """Ontology-aware policy resolver"""

    async def resolve_policies(
        self,
        object_type: str,
        subject: AuthenticatedUser,
    ) -> list[CompiledPolicy]:
        """
        Resolve policies based on Ontology type inheritance.
        Child types automatically inherit parent type policies.
        """
        type_hierarchy = await self._ontology.get_type_hierarchy(object_type)

        all_policies = []
        for type_in_chain in type_hierarchy:
            policies = await self._policy_store.get_policies(type_in_chain)
            all_policies.extend(policies)

        return self._merge_with_override(all_policies)

class RelationPropertyFilter:
    """Relationship property-based filtering"""

    async def compile_relation_filter(
        self,
        source_type: str,
        relation: str,
        target_type: str,
        subject: AuthenticatedUser,
    ) -> SQLFragment:
        """
        When queries involve relationship traversal, apply filtering at the
        relationship level. E.g., when querying a Project's Datasets,
        only return Datasets the user has permission to access.
        """
        return SQLFragment(
            sql=f"""
                EXISTS (
                    SELECT 1 FROM relation_tuples rt
                    WHERE rt.resource_type = %s
                      AND rt.resource_id = {target_type}_table.id
                      AND rt.relation IN ('viewer', 'editor', 'owner')
                      AND rt.subject_type = 'user'
                      AND rt.subject_id = %s
                      AND rt.deleted_at IS NULL
                )
            """,
            params=[target_type, subject.user_id],
        )

class PolicyChangeManager:
    """Policy change manager"""

    async def apply_policy_change(
        self,
        change: PolicyChange,
        approval: ApprovalRecord,
    ):
        """Apply policy change (requires approval)"""
        # 1. Validate approval
        if not approval.is_approved:
            raise PolicyChangeNotApprovedError()

        # 2. Impact analysis
        impact = await self._analyze_impact(change)
        logger.info(
            "Policy change impact: %d object types, ~%d rows affected",
            len(impact.affected_types),
            impact.estimated_affected_rows,
        )

        # 3. Gradual rollout
        if impact.estimated_affected_rows > 100000:
            await self._gradual_rollout(change, impact)
        else:
            await self._immediate_apply(change)

        # 4. Refresh materialized views
        for obj_type in impact.affected_types:
            await self._materialized_view.refresh(obj_type)

        # 5. Clear compilation cache
        self._compilation_cache.clear()

POLICY_REWRITE_METRICS = {
    "pqr_compilation_total": Counter("Total policy compilations"),
    "pqr_compilation_latency": Histogram("Compilation latency", buckets=[1, 5, 10, 50, 100]),
    "pqr_cache_hit_rate": Gauge("Compilation cache hit rate"),
    "pqr_rewrite_total": Counter("Total query rewrites"),
    "pqr_row_filter_count": Histogram("Row filter condition count"),
    "pqr_column_mask_count": Histogram("Column mask count"),
    "pqr_query_slowdown": Histogram("Rewrite-caused query slowdown ratio"),
    "pqr_seq_scan_warnings": Counter("Sequential scan warning count"),
    "pqr_policy_count": Gauge("Total active policies"),
}