传统模式:
  1. 用户请求 "SELECT * FROM orders"
  2. 数据库返回 100 万行
  3. 应用层逐行检查权限
  4. 过滤后返回 1000 行给用户

问题:
  - 数据库传输了 99.9% 的无用数据
  - 应用层内存占用巨大
  - 逐行权限检查 O(n) 线性增长
  - 分页、排序、聚合全部在应用层重做

查询重写模式:
  1. 用户请求 "SELECT * FROM orders"
  2. 策略引擎将请求重写为:
     "SELECT * FROM orders WHERE region IN ('east', 'south')
      AND classification <= 'C2'"
  3. 数据库只返回 1000 行授权数据
  4. 分页、排序、聚合由数据库高效执行

优势:
  - 数据库只处理授权数据
  - 利用索引加速过滤
  - 内存占用降低 99%+
  - 分页/排序/聚合原生支持

┌──────────────┐     ┌──────────────────┐     ┌──────────────┐
│  用户查询     │────>│  策略编译器       │────>│  重写后查询   │
│  Original    │     │  Policy Compiler  │     │  Rewritten   │
│  Query       │     │                  │     │  Query       │
└──────────────┘     └────────┬─────────┘     └──────┬───────┘
                              │                       │
                     ┌────────▼─────────┐             │
                     │  策略存储         │             │
                     │  - RBAC 角色     │             │
                     │  - ABAC 属性     │             ▼
                     │  - ReBAC 关系    │     ┌──────────────┐
                     └──────────────────┘     │  数据库引擎   │
                                              │  (Doris/PG)  │
                                              └──────────────┘

class PolicyCompiler:
    """策略编译器：将权限策略转换为 SQL 条件"""

    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:
        """
        编译管道:
        1. 解析原始查询，提取涉及的表和列
        2. 查找每个表的行级策略
        3. 查找每个列的列级策略
        4. 编译策略为 SQL 条件
        5. 注入 WHERE 子句和 SELECT 列过滤
        """
        # Step 1: 解析查询
        tables = query.extract_tables()
        columns = query.extract_columns()

        # Step 2: 行级策略编译
        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_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: 重写查询
        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()),
            ),
        )

# 策略定义示例
policies:
  - id: "pol-region-isolation"
    name: "区域数据隔离"
    description: "用户只能看到自己区域的数据"
    target:
      object_types: ["order", "customer", "inventory"]
    condition:
      operator: "in"
      field: "resource.region"
      value_source: "subject.allowed_regions"
    effect: "FILTER"

  - id: "pol-classification-ceiling"
    name: "分类等级天花板"
    description: "用户只能看到不超过自己等级的数据"
    target:
      object_types: ["*"]
    condition:
      operator: "lte"
      field: "resource.classification_level"
      value_source: "subject.clearance_level"
    effect: "FILTER"

  - id: "pol-owner-full-access"
    name: "数据所有者完全访问"
    description: "数据的创建者有完全访问权"
    target:
      object_types: ["*"]
    condition:
      operator: "eq"
      field: "resource.owner_id"
      value_source: "subject.user_id"
    effect: "ALLOW_ALL"

  - id: "pol-time-restriction"
    name: "工作时间限制"
    description: "敏感数据仅在工作时间内可访问"
    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:
    """将策略表达式编译为 SQL 条件"""

    def compile_condition(
        self,
        condition: PolicyCondition,
        table_alias: str,
        subject: AuthenticatedUser,
        context: RequestContext,
    ) -> SQLFragment:
        """递归编译策略条件为 SQL 片段"""

        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:
        """解析值来源"""
        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  # 字面值

用户查询: SELECT * FROM orders WHERE status = 'pending'

策略编译后:
  SELECT * FROM orders
  WHERE status = 'pending'                    -- 用户原始条件
    AND region IN ('east', 'south')            -- 区域隔离策略
    AND classification_level <= 3              -- 分类等级策略
    AND (owner_id = 'user123'                  -- 所有者策略
         OR department_id IN (10, 20, 30))     -- 部门可见性策略

class RowLevelSecurityRewriter:
    """行级安全重写器"""

    async def apply_rls(
        self,
        query: ParsedQuery,
        table: TableRef,
        subject: AuthenticatedUser,
        context: RequestContext,
    ) -> list[SQLFragment]:
        """
        应用行级安全策略。

        合并逻辑:
        - 同一策略组内的条件用 OR 合并（满足任一即可）
        - 不同策略组之间用 AND 合并（必须同时满足）
        - ALLOW_ALL 策略可跳过后续 FILTER 策略
        """
        applicable_policies = await self._find_applicable_policies(
            table.object_type, subject, context
        )

        if not applicable_policies:
            # 没有匹配策略 → 默认拒绝所有行
            return [SQLFragment(sql="1 = 0", params=[])]

        # 检查是否有 ALLOW_ALL 策略
        for policy in applicable_policies:
            if policy.effect == "ALLOW_ALL":
                condition = self._compiler.compile_condition(
                    policy.condition, table.alias, subject, context
                )
                # 如果 ALLOW_ALL 条件满足，跳过其他过滤
                # 使用 CASE WHEN 表达
                return []  # 不需要额外过滤

        # 按策略组分组
        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，组间 AND
        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 关系的行级过滤"""

    async def compile_rebac_filter(
        self,
        table: TableRef,
        subject: SubjectRef,
        permission: str,
    ) -> SQLFragment:
        """
        将 ReBAC 关系转为 SQL 行过滤条件。

        策略：预计算 + 物化视图
        - 定期将 ReBAC 关系展开为 (user, resource) 对
        - 存入物化视图用于 JOIN
        """
        # 方案1：使用物化视图 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:
        """
        实时 ReBAC 过滤（适用于关系频繁变化的场景）
        通过 LookupResources 预查可访问的资源 ID 列表
        """
        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:
            # ID 太多时使用临时表
            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: str
    object_type: str
    action: ColumnAction  # MASK / HIDE / TRANSFORM / ALLOW

class ColumnAction(Enum):
    ALLOW = "allow"        # 完全可见
    HIDE = "hide"          # 完全隐藏（从 SELECT 中移除）
    MASK = "mask"          # 脱敏显示
    TRANSFORM = "transform"  # 自定义转换

class ColumnLevelSecurityRewriter:
    """列级安全重写器"""

    async def rewrite_columns(
        self,
        query: ParsedQuery,
        subject: AuthenticatedUser,
        context: RequestContext,
    ) -> ParsedQuery:
        """重写查询的 SELECT 列表"""
        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

                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:
        """生成脱敏 SQL 表达式"""
        full_name = f"{col.table_alias}.{col.name}"

        match mask_type:
            case "partial":
                # 部分显示：保留前N后M位
                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":
                # 置 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:
    """安全 SQL 构建器，强制参数化"""

    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:
        """验证 SQL 标识符（表名、列名）是否安全"""
        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:
        """安全地合并多个 WHERE 条件"""
        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:
    """查询审计器：记录所有重写操作"""

    async def audit_rewrite(
        self,
        original_query: str,
        rewritten_query: str,
        subject: AuthenticatedUser,
        applied_policies: list[str],
        context: RequestContext,
    ):
        """记录查询重写审计日志"""
        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 方言抽象基类"""

    @abstractmethod
    def quote_identifier(self, name: str) -> str:
        """引用标识符"""

    @abstractmethod
    def param_placeholder(self, index: int) -> str:
        """参数占位符"""

    @abstractmethod
    def concat_function(self, *args: str) -> str:
        """字符串连接"""

    @abstractmethod
    def date_trunc(self, precision: str, column: str) -> str:
        """日期截断"""

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


class DorisDialect(SQLDialect):
    """Apache Doris SQL 方言"""

    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 方言"""

    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 查询的策略集成"""

    async def rewrite_oql(
        self,
        oql: OQLQuery,
        subject: AuthenticatedUser,
        context: RequestContext,
    ) -> OQLQuery:
        """
        OQL 查询重写流程:
        1. OQL → AST
        2. AST 上注入策略条件
        3. AST → 目标 SQL（带策略条件）
        """
        ast = self._parser.parse(oql)

        # 在 AST 层面注入策略
        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:
    """物化权限视图管理器"""

    async def refresh(self, object_type: str):
        """刷新物化权限视图"""
        await self._db.execute(f"""
            REFRESH MATERIALIZED VIEW CONCURRENTLY
            mv_permissions_{object_type}
        """)

    async def create_materialized_view(self, object_type: str):
        """创建物化权限视图"""
        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
            -- 展开继承关系
            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:
    """增量权限更新器"""

    async def on_tuple_change(self, change: TupleChange):
        """当关系元组变更时，增量更新物化视图"""
        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]:
        """计算受影响的对象类型"""
        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:
    """策略编译结果缓存"""

    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:
        """生成缓存键"""
        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

测试环境: 100 万行 orders 表, 5 个策略规则

| 方法                  | 延迟 (p50) | 延迟 (p99) | 内存峰值  |
|----------------------|-----------|-----------|----------|
| 应用层过滤             | 2400ms    | 5800ms    | 1.2 GB   |
| 查询重写（无缓存）      | 35ms      | 120ms     | 45 MB    |
| 查询重写（策略缓存）    | 28ms      | 95ms      | 45 MB    |
| 查询重写（物化视图）    | 15ms      | 55ms      | 30 MB    |
| 数据库原生 RLS          | 12ms      | 48ms      | 28 MB    |

结论: 查询重写相比应用层过滤，延迟降低 98%，内存降低 96%

class QueryPlanAnalyzer:
    """查询计划分析器：确保重写后的查询使用索引"""

    async def analyze_rewritten_query(
        self,
        original: str,
        rewritten: str,
        params: list,
    ) -> QueryPlanReport:
        """分析重写后查询的执行计划"""
        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 感知的策略解析器"""

    async def resolve_policies(
        self,
        object_type: str,
        subject: AuthenticatedUser,
    ) -> list[CompiledPolicy]:
        """
        根据 Ontology 的类型继承关系解析策略。
        子类型自动继承父类型的策略。
        """
        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)

        # 子类型的策略可以 override 父类型的策略
        return self._merge_with_override(all_policies)

class RelationPropertyFilter:
    """基于关系属性的过滤"""

    async def compile_relation_filter(
        self,
        source_type: str,
        relation: str,
        target_type: str,
        subject: AuthenticatedUser,
    ) -> SQLFragment:
        """
        当查询涉及关系遍历时，在关系层面应用过滤。
        例如：查询 Project 的 Dataset 时，
        只返回用户有权限的 Dataset。
        """
        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:
    """策略变更管理器"""

    async def apply_policy_change(
        self,
        change: PolicyChange,
        approval: ApprovalRecord,
    ):
        """应用策略变更（需审批）"""
        # 1. 验证审批
        if not approval.is_approved:
            raise PolicyChangeNotApprovedError()

        # 2. 影响分析
        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. 灰度发布
        if impact.estimated_affected_rows > 100000:
            await self._gradual_rollout(change, impact)
        else:
            await self._immediate_apply(change)

        # 4. 刷新物化视图
        for obj_type in impact.affected_types:
            await self._materialized_view.refresh(obj_type)

        # 5. 清除编译缓存
        self._compilation_cache.clear()

POLICY_REWRITE_METRICS = {
    "pqr_compilation_total": Counter("策略编译总次数"),
    "pqr_compilation_latency": Histogram("策略编译延迟", buckets=[1, 5, 10, 50, 100]),
    "pqr_cache_hit_rate": Gauge("编译缓存命中率"),
    "pqr_rewrite_total": Counter("查询重写总次数"),
    "pqr_row_filter_count": Histogram("行过滤条件数量"),
    "pqr_column_mask_count": Histogram("列脱敏数量"),
    "pqr_query_slowdown": Histogram("重写导致的查询减速比"),
    "pqr_seq_scan_warnings": Counter("顺序扫描告警次数"),
    "pqr_policy_count": Gauge("活跃策略总数"),
}