🛡️ 技術調査 - Cloudflare Town Lake / Skipper（監査可能なNL-SQLエージェント）

# docker-compose.yml (実装案) 参考: https://trino.io/docs/current/installation/deployment.html
version: "3.9"
services:
  trino:
    image: trinodb/trino:latest
    ports:
      - "8080:8080"
    volumes:
      - ./trino/etc:/etc/trino
      - ./trino/catalog:/etc/trino/catalog
  iceberg-rest:
    image: tabulario/iceberg-rest:latest
    ports:
      - "8181:8181"
    environment:
      CATALOG_WAREHOUSE: s3://your-bucket/warehouse
      CATALOG_IO__IMPL: org.apache.iceberg.aws.s3.S3FileIO
  datahub-gms:
    image: linkedin/datahub-gms:head
    ports:
      - "8081:8080"

# etc/catalog/iceberg.properties (実装案) 参考: https://trino.io/docs/current/connector/iceberg.html
connector.name=iceberg
iceberg.catalog.type=rest
iceberg.rest-catalog.uri=http://iceberg-rest:8181
iceberg.rest-catalog.warehouse=s3://your-bucket/warehouse
iceberg.file-format=PARQUET
fs.native-s3.enabled=true

# etc/catalog/postgres.properties (実装案) 参考: https://trino.io/docs/current/connector/postgresql.html
connector.name=postgresql
connection-url=jdbc:postgresql://postgres-host:5432/mydb
connection-user=${ENV:POSTGRES_USER}
connection-password=${ENV:POSTGRES_PASSWORD}

-- クロスソース結合の例: Postgres の orders と Iceberg の enriched_customers を join
SELECT o.order_id, o.amount, c.segment, c.region
FROM postgres.mydb.orders o
JOIN iceberg.warehouse.enriched_customers c ON o.customer_id = c.customer_id
WHERE o.created_at >= DATE '2026-01-01'

# datahub_recipe.yml (実装案) 参考: https://datahubproject.io/docs/metadata-ingestion/
source:
  type: trino
  config:
    host_port: "trino-host:8080"
    database: "iceberg"
    include_tables: true
    profiling:
      enabled: true
sink:
  type: datahub-rest
  config:
    server: "http://datahub-gms:8080"

# 実装案: .meta.json emit に相当する処理 (コード由来メタを DataHub へ)
import datahub.emitter.mce_builder as builder
from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.schema_classes import EditableDatasetPropertiesClass

emitter = DatahubRestEmitter(gms_server="http://datahub-gms:8080")
dataset_urn = builder.make_dataset_urn(platform="trino", name="iceberg.warehouse.billing_events")
mce = builder.make_mce_builder(
    entityUrn=dataset_urn,
    aspect=EditableDatasetPropertiesClass(
        description=(
            "billing イベントテーブル。alloc_amount は年払いなら billed_amount / 12、"
            "月払いなら billed_amount。Source: transformer/billing_events.sql (commit abc1234)"
        )
    ),
)
emitter.emit_mce(mce)

# models/billing/schema.yml (実装案) 参考: https://docs.getdbt.com/docs/build/semantic-models
version: 2
models:
  - name: billing_events
    description: >
      billing イベントテーブル。年払いプランの alloc_amount は billed_amount を 12 で除算。
    columns:
      - name: customer_id
        description: "顧客の一意識別子 (accounts テーブルの FK)"
        data_tests: [not_null]
      - name: alloc_amount
        description: "plan_type=annual なら billed_amount/12、monthly なら billed_amount"
semantic_models:
  - name: billing_semantic
    model: ref('billing_events')
    entities:
      - {name: billing_event, type: primary, expr: event_id}
      - {name: customer, type: foreign, expr: customer_id}
    measures:
      - {name: total_alloc_amount, agg: sum, expr: alloc_amount}

# policy/data_access.rego (実装案) 参考: https://www.openpolicyagent.org/docs/latest/
package data.access
import rego.v1

default allow := false

allow if {
    input.table.status == "reviewed"
    user_has_permission
    not column_is_restricted
}
user_has_permission if {
    some group in data.groups[input.user.id]
    group in input.table.allowed_groups
}
column_is_restricted if {
    input.column.pii == true
    not "pii_viewer" in data.groups[input.user.id]
}

# etc/access-control.properties (実装案) 参考: https://trino.io/docs/current/security/opa-access-control.html
access-control.name=opa
opa.policy.uri=http://opa-server:8181/v1/data/data/access/allow
opa.policy.row-filters-enabled=true
opa.policy.column-masking-enabled=true

{
  "result": {
    "allow": true,
    "rowFilters": [
      {"expression": "region = 'apac'"}
    ],
    "columnMasks": [
      {"column": "email", "expression": "'***REDACTED***'"}
    ]
  }
}

# 実装案: DataHub の table description を embedding 化して検索する
from qdrant_client import QdrantClient

qdrant = QdrantClient(url="http://qdrant:6333")

def search_tables_by_embedding(nlq: str, top_k: int = 5) -> list:
    # クエリを埋め込み、テーブル説明の事前インデックスと近傍検索する
    vector = embed(nlq)  # 任意の embedding モデル
    hits = qdrant.search(collection_name="table_descriptions", query_vector=vector, limit=top_k)
    return [h.payload["table_name"] for h in hits]

# 実装案: 5 層コンテキスト供給 + closed-loop 検証エージェント
# 参考: https://blog.cloudflare.com/our-unified-data-platform/
import anthropic, requests, trino

client = anthropic.Anthropic()
trino_conn = trino.dbapi.connect(host="trino-host", port=8080, user="skipper-agent")

def get_metadata_context(table_name: str) -> str:
    """DataHub から 1-4 層コンテキストを取得する"""
    resp = requests.get(
        f"http://datahub-gms:8080/entities/{table_name}",
        headers={"Authorization": f"Bearer {DATAHUB_TOKEN}"},
    )
    info = resp.json().get("aspects", {})
    schema = info.get("schemaMetadata", {})              # 層1: Schema/Usage
    annotations = info.get("editableSchemaMetadata", {}) # 層2: Human Annotations
    code_derived = info.get("datasetProperties", {})     # 層3: Code-Derived
    curated = load_curated_model(table_name)             # 層4: Curated Model
    return f"Schema:{schema}\nAnnotations:{annotations}\nCode:{code_derived}\nCurated:{curated}"

def runtime_introspection(table: str, col: str) -> str:
    """層5: DESCRIBE / SELECT DISTINCT でリアルタイム検証"""
    cur = trino_conn.cursor()
    cur.execute(f"DESCRIBE {table}")
    describe = cur.fetchall()
    cur.execute(f"SELECT DISTINCT {col} FROM {table} LIMIT 20")
    distinct = [r[0] for r in cur.fetchall()]
    return f"DESCRIBE:{describe}\nDISTINCT {col}:{distinct}"

def nl_to_sql_agent(nlq: str, user_id: str, user_groups: list) -> dict:
    """closed-loop で最大 3 回まで検証・修正する"""
    candidates = search_tables_by_embedding(nlq)[:5]
    context = "\n".join(get_metadata_context(t) for t in candidates)
    system = (
        "あなたは社内データ分析の SQL 生成エージェントです。\n"
        "- 生成 SQL は Trino 構文に従う\n"
        "- 権限のないテーブル・列にアクセスしない\n"
        "- 結果が 0 件なら join 条件や絞り込みを見直して再試行する\n"
        "- PII 列 (is_pii=true) は SELECT に含めない\n"
        f"## テーブル情報\n{context}\n## 権限\nuser_id:{user_id} groups:{user_groups}"
    )
    messages = [{"role": "user", "content": nlq}]
    for attempt in range(3):
        resp = client.messages.create(
            model="claude-sonnet-4-5", max_tokens=2048, system=system, messages=messages
        )
        sql = extract_sql(resp.content[0].text)
        try:
            cur = trino_conn.cursor()
            cur.execute(sql)
            rows = cur.fetchall()
            if len(rows) == 0 and attempt < 2:
                messages += [
                    {"role": "assistant", "content": resp.content[0].text},
                    {"role": "user", "content": f"結果0件。join/フィルタを見直して再生成。{runtime_introspection(candidates[0], 'created_at')}"},
                ]
                continue
            return {"sql": sql, "rows": rows, "attempts": attempt + 1}
        except Exception as e:
            messages += [
                {"role": "assistant", "content": resp.content[0].text},
                {"role": "user", "content": f"SQL エラー:{e}。修正して再生成。"},
            ]
    return {"error": "max attempts reached"}

# 実装案: search/execute 2 プリミティブ
# 参考: https://blog.cloudflare.com/code-mode/ , https://www.anthropic.com/engineering/code-execution-with-mcp
TOOLS = [
    {
        "name": "search_tools",
        "description": "利用可能な MCP ツールを検索。定義を全件ロードせず必要なものだけ取得する",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "detail_level": {"type": "string", "enum": ["name_only", "summary", "full_schema"]},
            },
            "required": ["query"],
        },
    },
    {
        "name": "execute_code",
        "description": "コードをサンドボックス実行。外部アクセスは MCP binding 経由のみ。コードはログに残り監査証跡になる",
        "input_schema": {
            "type": "object",
            "properties": {"code": {"type": "string"}, "timeout_seconds": {"type": "integer"}},
            "required": ["code"],
        },
    },
]

クエリ保存 → 権限は未評価 (save-time check なし)
      ↓
閲覧 → 閲覧者の現在のグループ情報で権限再評価
      ↓
権限なし → アクセス権申請へ誘導 / 権限あり → 結果表示

生成コード → 直接トークン参照 → 不可 (throw)
生成コード → binding を呼ぶ → supervisor がトークンを代理注入 → MCP サーバー

# .meta.json の例 (Transformer が emit)
alloc_amount:
  computed_as: "billed_amount / 12 for annual; billed_amount for monthly"
  source_table: billing_events
  transform_sql: "CASE WHEN plan_type='annual' THEN billed_amount/12 ELSE billed_amount END"