联邦查询与虚拟图：跨源映射、下推与结果合并

数据不搬的理由：
1. 体量：源数据 TB 级，全量入图成本与时间都不可接受
2. 时效：源数据分钟级更新，ETL 跟不上（或不需要历史）
3. 合规：数据不能出域（跨境、隐私、第三方授权限制）
4. 权限：源系统有自己的访问控制，拷贝即绕过
5. 归属：数据由别的团队/公司维护，只开放查询接口
→ 联邦的价值 = 「零拷贝的关联视图」

- 需要跑图算法（PageRank/社区发现/多跳遍历）：联邦无法下推算法
- 查询模式固定且高频：ETL 一次、查询万次更划算
- 源端查询能力极弱（只有全表导出接口）：联邦会退化成「拉全量」
- 跨源 JOIN 的选择性很差：网络传输量会失控
→ 经验法则：联邦做「按需关联检索」，ETL 做「分析与高频查询」
→ 混合架构最常见：热数据入图 + 冷/受限数据联邦

R2RML（RDB to RDF Mapping Language）：把关系表映射为 RDF 三元组，W3C 标准
  LogicalTable —— 一张表或一条 SQL 视图
  SubjectMap —— 主键 → 主语 IRI 的模板，可加 class
  PredicateObjectMap —— 列 → 谓语 + 宾语（字面量或 IRI 引用）
  特点：SQL 表达能力强（视图承载复杂逻辑），但只能映射关系库

RML：R2RML 的泛化版，源可以是 CSV/JSON/XML/Parquet
  用「迭代器 + 引用」表达嵌套结构（如 JSON 数组展开）
  特点：覆盖非关系源，但实现生态比 R2RML 弱

属性图映射（LPG View）：面向 Neo4j 这类属性图
  节点视图 = 一条返回 (id, label, props) 的 SQL/查询
  关系视图 = 一条返回 (startId, endId, type, props) 的查询
  特点：直接产出属性图，适合 Cypher 联邦

@prefix rr: <http://www.w3.org/ns/r2rml#> .
@prefix ex: <http://example.com/> .

<#OrderMap> a rr:TriplesMap ;
  rr:logicalTable [ rr:tableName "orders" ] ;
  rr:subjectMap [
    rr:template "http://example.com/order/{order_id}" ;
    rr:class ex:Order ] ;
  rr:predicateObjectMap [
    rr:predicate ex:placedBy ;
    rr:objectMap [ rr:template "http://example.com/customer/{customer_id}" ] ] ;
  rr:predicateObjectMap [
    rr:predicate ex:amount ;
    rr:objectMap [ rr:column "amount" ; rr:datatype xsd:decimal ] ] .

PREFIX ex: <http://example.com/>
SELECT ?customer ?riskLevel WHERE {
  ?customer ex:hasOrder ?order .              # 本地端点
  ?order ex:amount ?amount .
  FILTER(?amount > 10000)
  SERVICE <https://risk.example.com/sparql> {  # 远端端点
    ?customer ex:riskLevel ?riskLevel .
  }
}

普通 SERVICE：强制把模式送到远端执行，远端不可达则整个查询失败
SERVICE SILENT：远端失败时该模式当作「无解」继续，
                适合「查得到就补上、查不到也不影响主结果」
→ 生产里几乎总该用 SILENT，把跨源失败降级为「该字段缺失」

1. 远端模式里没有绑定变量 → 变成远端全量扫描
   错误：SERVICE <远端> { ?s ex:p ?o }   ← 无约束，拉全表
   正确：让本地先求出绑定集，用 VALUES 或把变量带进去
2. 绑定集太大 → 远端收到上万个 IRI，查询超时或被限流
   应对：分批（把绑定集切块，逐批查远端再合并）
3. 远端不支持某个函数/模式 → 静默返回空
   应对：查询前做能力探测，文档化支持的子集

SELECT ?customer ?riskLevel WHERE {
  VALUES ?customer { ex:c1 ex:c2 ex:c3 }        # 显式给定绑定集
  SERVICE SILENT <https://risk.example.com/sparql> {
    ?customer ex:riskLevel ?riskLevel .
  }
}

该下推的：过滤谓词（等值/范围）、投影（只取需要的列）、
          聚合（COUNT/SUM 在源端算完再传）、排序 + LIMIT（源端 Top-K）
不该下推的：需要跨源数据才能算的谓词、源端不支持的函数、
            会破坏语义的等价改写（NULL 语义、字符集排序）

1. 先拉全量再本地过滤：模式里对远端节点无约束 →
   优化器可能「先把远端全拉回来」再本地 JOIN，
   网络流量与内存占用与远端表大小成正比
2. 跨源 JOIN 顺序错误：先查大表，再拿大结果集去查小表（绑定集爆炸）；
   正确顺序是先在选择性高的源上过滤，缩小结果集再查另一个源
3. 下推了但没生效：映射层用了视图，优化器无法穿透视图做谓词下推
   → 把高频过滤条件显式写进视图，或提供带参数的视图

至少需要：每个源上该模式的基数、谓词的选择性、
          源端网络往返延迟与吞吐、本地与远端算子的单位代价
难点：远端通常不暴露统计信息，只能靠「采样 + 历史执行反馈」
实用做法：维护一张「模式 → 实测基数」的统计表，定期刷新，
          优化器用它决策，并对偏差过大的估算回退到保守计划

1. 优先在「有索引、选择性高」的源上做第一轮过滤
2. 跨源 JOIN 的驱动端选「过滤后基数最小」的那个
3. 单次跨源调用的绑定集控制在百级（超过就分批）
4. 聚合与 Top-K 一律下推，避免中间结果回传
5. 远端模式必须带绑定，绝不允许无约束的远端扫描
→ 这五条能解决 80% 的联邦性能问题，剩下的靠实测调优

并集（UNION）：默认。同一实体在两个源都有记录 → 两行结果
合并（JOIN）：按共享标识连接 → 跨源属性拼在一行
覆盖（COALESCE）：按优先级取值（主数据源优先，缺失才取备用源）
→ 必须先明确要哪种，联邦查询的「正确答案」完全取决于语义选择

结果级去重：同一实体被多个源返回 → DISTINCT 或按 ID 折叠
  坑：DISTINCT 比较整行，若两源某属性值不同
      （一个写 "Beijing"、一个写 "北京市"）就折叠不掉
实体级去重：先对齐标识，再按标识聚合属性
  坑：属性冲突怎么取？（取最新、取非空、取优先级高的源）
→ 生产做法：联邦层只做结果级折叠 + 标注来源，
  真正的实体级合并交给下游的实体解析流程

溯源标注的价值：出问题能立刻定位「这条脏数据来自哪个源」；
下游可按来源做置信度加权（主数据源权重高）；
合规审计要求「数据出处可追溯」
→ 联邦结果里，来源标记应该是一等公民而不是附属字段

L1 源端查询缓存：按「源 + 规范化查询 + 参数」缓存源返回的原始行
  收益最大（省掉远端往返）；风险：源数据变了不知道
L2 映射/图模式缓存：缓存翻译后的源查询结果
  收益：省掉翻译与下推决策；风险：映射改了要清空
L3 结果缓存：缓存最终合并结果
  收益：最快；风险：跨源数据变化后结果失真最严重
→ 生产建议：重点做 L1，慎做 L3；L3 只用于明确可容忍陈旧的场景并配短 TTL

强一致（读时校验源版本）：延迟高，几乎失去缓存意义
有界陈旧（TTL ≤ N 秒）：最常用，把「陈旧」变成可量化承诺
最终一致（异步刷新）：延迟最低，需要业务能接受陈旧
→ 关键：把「可接受的陈旧时长」写进接口契约，
  而不是让缓存策略隐式决定数据新鲜度

-- 客户节点视图
CREATE VIEW v_customer AS
SELECT id, name, 'Customer' AS label, credit_code FROM customer;

-- 订单关系视图（Customer -[:PLACED]-> Order）
CREATE VIEW v_order AS
SELECT order_id AS id, customer_id AS start_id, order_id AS end_id,
       'PLACED' AS rel_type, amount, created_at
FROM   orders WHERE status <> 'CANCELLED';

BATCH = 100

def enrich_with_risk(candidates, risk_endpoint, client):
    """分批调用远端，绑定集控制在百级；远端失败降级为未知"""
    out = {}
    for i in range(0, len(candidates), BATCH):
        chunk = candidates[i:i + BATCH]
        query = """
        SELECT ?customer ?level ?listName WHERE {
          VALUES ?customer { %s }
          SERVICE SILENT <%s> {
            ?customer ex:riskLevel ?level .
            OPTIONAL { ?customer ex:onList ?listName }
          }
        }
        """ % (" ".join("<%s>" % c for c in chunk), risk_endpoint)
        try:
            for row in client.select(query):
                out[row["customer"]] = {"level": row.get("level"),
                                        "list": row.get("listName"),
                                        "source": "risk-endpoint"}
        except Exception as e:                     # 部分失败不中断整体
            for c in chunk:
                out.setdefault(c, {"level": None, "source": "unavailable",
                                   "error": str(e)})
    return out

def merge(candidates, risk_map):
    return [{
        "customerId": cid, "name": name, "total": total,
        "riskLevel": risk_map.get(cid, {}).get("level"),
        "sources": ["warehouse", risk_map.get(cid, {}).get("source", "missing")],
        "merged": cid in risk_map,
    } for cid, name, total in candidates]

源端不可达（网络/宕机/限流）；源端慢（无索引的远端查询拖垮整体）；
源端返回错误结构（schema 漂移）；部分源成功、部分源失败（最常见）；
结果不一致（不同源在同一时刻的快照不同）

1. 整体失败（fail-fast）：任何源失败即抛错，不返回部分结果
   适合：结果必须完整才有意义（合规报表）
2. 降级返回（best-effort）：缺失字段标为未知
   实现 = SERVICE SILENT + 异常兜底 + 标注 unavailable
   适合：检索类场景
3. 重试 + 退避：指数退避 + 抖动，限制次数，避免重试风暴
   适合：瞬时故障（网络抖动、限流）
4. 降级到快照：查不到时回退到「最近一次成功的物化副本」并标注陈旧
   适合：源端不可用但有历史快照
→ 生产里通常是 2 + 3 组合，对关键字段再叠加 4

1. 映射治理：规则版本化、变更走评审与回归测试
   （映射错了会静默返回错误结果，比报错更危险）
2. 权限治理：联邦层不能绕过源端权限，要透传用户身份，
   或在联邦层做「最小可见字段」控制
3. 语义治理：单位/枚举/时间语义的统一定义写进数据字典
4. 变更治理：源端 schema 变更要有通知机制，否则映射会静默失效

1. 远端模式缺绑定 → 全量扫描。看远端日志/流量，加 VALUES 或分批
2. 下推未生效 → 网络传输量异常大。检查是否被视图挡住
3. 绑定集过大 → 远端超时/限流。分批并降低驱动端基数
4. 标识对不上 → 结果比预期少很多。核对 IRI 模板与规范化规则
5. 缓存未失效 → 数据更新后仍返回旧值。检查缓存键是否含映射版本
6. 语义冲突 → 单位/枚举/时间不一致导致结果荒谬。回到映射层修
7. 部分失败未标注 → 用户以为「没有风险」实际是「查不到」
   → 最危险的一类，必须强制标注 unavailable