┌─────────────────────────────────────────────────┐
│                   Control Layer (Control)              │
│              Spring Boot 3.x + gRPC              │
└──────────────┬──────────────────┬────────────────┘
               │ gRPC             │ gRPC
    ┌──────────▼──────────┐  ┌───▼────────────────┐
    │   Data Layer (Data)     │  │  Reasoning & Decision Layer (Reasoning)│
    │   Quarkus 3.x        │  │  Python + FastAPI   │
    │   ┌──────────────┐   │  └────────────────────┘
    │   │ Apache Doris  │   │
    │   │ (OLAP Engine) │   │
    │   └──────┬───────┘   │
    │          │            │
    │   ┌──────▼───────┐   │
    │   │ Apache Iceberg│   │
    │   │ + Nessie      │   │
    │   └──────────────┘   │
    └──────────────────────┘

# deployment-Layer/docker-compose/doris-cluster.yml
version: '3.8'
services:
  doris-fe-1:
    image: apache/doris:2.1.4-fe
    hostname: doris-fe-1
    environment:
      - FE_SERVERS=doris-fe-1:9010,doris-fe-2:9010,doris-fe-3:9010
      - FE_ID=1
    ports:
      - "8030:8030"   # HTTP API
      - "9030:9030"   # MySQL Protocol
      - "9010:9010"   # Edit Log Port
    volumes:
      - doris-fe-1-meta:/opt/apache-doris/fe/doris-meta
    deploy:
      resources:
        limits:
          memory: 8G
          cpus: '4'
    networks:
      - coomia-dip-net

  doris-fe-2:
    image: apache/doris:2.1.4-fe
    hostname: doris-fe-2
    environment:
      - FE_SERVERS=doris-fe-1:9010,doris-fe-2:9010,doris-fe-3:9010
      - FE_ID=2
    volumes:
      - doris-fe-2-meta:/opt/apache-doris/fe/doris-meta
    deploy:
      resources:
        limits:
          memory: 8G
          cpus: '4'
    networks:
      - coomia-dip-net

  doris-fe-3:
    image: apache/doris:2.1.4-fe
    hostname: doris-fe-3
    environment:
      - FE_SERVERS=doris-fe-1:9010,doris-fe-2:9010,doris-fe-3:9010
      - FE_ID=3
    volumes:
      - doris-fe-3-meta:/opt/apache-doris/fe/doris-meta
    deploy:
      resources:
        limits:
          memory: 8G
          cpus: '4'
    networks:
      - coomia-dip-net

  doris-be-1:
    image: apache/doris:2.1.4-be
    hostname: doris-be-1
    environment:
      - FE_SERVERS=doris-fe-1:9010,doris-fe-2:9010,doris-fe-3:9010
      - BE_ADDR=doris-be-1:9050
    volumes:
      - doris-be-1-data:/opt/apache-doris/be/storage
    deploy:
      resources:
        limits:
          memory: 32G
          cpus: '16'
    networks:
      - coomia-dip-net

  doris-be-2:
    image: apache/doris:2.1.4-be
    hostname: doris-be-2
    environment:
      - FE_SERVERS=doris-fe-1:9010,doris-fe-2:9010,doris-fe-3:9010
      - BE_ADDR=doris-be-2:9050
    volumes:
      - doris-be-2-data:/opt/apache-doris/be/storage
    deploy:
      resources:
        limits:
          memory: 32G
          cpus: '16'
    networks:
      - coomia-dip-net

  doris-be-3:
    image: apache/doris:2.1.4-be
    hostname: doris-be-3
    environment:
      - FE_SERVERS=doris-fe-1:9010,doris-fe-2:9010,doris-fe-3:9010
      - BE_ADDR=doris-be-3:9050
    volumes:
      - doris-be-3-data:/opt/apache-doris/be/storage
    deploy:
      resources:
        limits:
          memory: 32G
          cpus: '16'
    networks:
      - coomia-dip-net

volumes:
  doris-fe-1-meta:
  doris-fe-2-meta:
  doris-fe-3-meta:
  doris-be-1-data:
  doris-be-2-data:
  doris-be-3-data:

networks:
  coomia-dip-net:
    external: true

# 元数据目录
meta_dir = /opt/apache-doris/fe/doris-meta

# JVM 堆内存（建议物理内存的 50-70%）
JAVA_OPTS="-Xmx6g -Xms6g -XX:+UseG1GC -XX:MaxGCPauseMillis=200"

# 查询超时（秒）
query_timeout = 300

# 最大连接数
qe_max_connection = 4096

# 并行执行实例数
parallel_fragment_exec_instance_num = 8

# 开启 Pipeline 执行引擎
enable_pipeline_engine = true

# Catalog 相关
enable_multi_catalog = true

# 审计日志
audit_log_dir = /opt/apache-doris/fe/log
audit_log_roll_num = 90

# 存储目录（多盘配置提高 IO 吞吐）
storage_root_path = /data1;/data2;/data3

# 内存限制（建议物理内存的 80%）
mem_limit = 80%

# Compaction 线程数
compaction_task_num_per_disk = 4

# 向量化执行引擎
enable_vectorized_engine = true

# Page Cache
disable_storage_page_cache = false
storage_page_cache_limit = 20%

# Chunk 大小（影响查询内存使用）
chunk_reserved_bytes_limit = 2147483648

# 网络缓冲区
brpc_num_threads = 256
thrift_server_max_worker_threads = 4096

-- coomia-dip 中的 Ontology 操作审计日志表
CREATE TABLE ontology_audit_log (
    event_time     DATETIME       NOT NULL COMMENT '事件时间',
    tenant_id      VARCHAR(64)    NOT NULL COMMENT '租户ID',
    user_id        VARCHAR(64)    NOT NULL COMMENT '用户ID',
    object_type    VARCHAR(128)   NOT NULL COMMENT '对象类型',
    object_rid     VARCHAR(256)   NOT NULL COMMENT '对象RID',
    action         VARCHAR(32)    NOT NULL COMMENT '操作类型',
    property_name  VARCHAR(128)   COMMENT '属性名',
    old_value      TEXT           COMMENT '旧值',
    new_value      TEXT           COMMENT '新值',
    source_ip      VARCHAR(45)    COMMENT '来源IP',
    request_id     VARCHAR(64)    COMMENT '请求ID',
    duration_ms    INT            COMMENT '处理耗时(ms)'
)
DUPLICATE KEY(event_time, tenant_id, user_id)
PARTITION BY RANGE(event_time) (
    FROM ("2024-01-01") TO ("2026-12-31") INTERVAL 1 MONTH
)
DISTRIBUTED BY HASH(tenant_id) BUCKETS 16
PROPERTIES (
    "replication_num" = "3",
    "dynamic_partition.enable" = "true",
    "dynamic_partition.time_unit" = "MONTH",
    "dynamic_partition.start" = "-12",
    "dynamic_partition.end" = "3",
    "dynamic_partition.prefix" = "p",
    "dynamic_partition.buckets" = "16",
    "compaction_policy" = "time_series"
);

-- coomia-dip 中的对象属性变更统计表
CREATE TABLE ontology_property_stats (
    stat_date      DATE           NOT NULL COMMENT '统计日期',
    tenant_id      VARCHAR(64)    NOT NULL COMMENT '租户ID',
    object_type    VARCHAR(128)   NOT NULL COMMENT '对象类型',
    property_name  VARCHAR(128)   NOT NULL COMMENT '属性名',
    change_count   BIGINT         SUM      COMMENT '变更次数',
    unique_objects HLL            HLL_UNION COMMENT '去重对象数',
    last_changed   DATETIME       MAX      COMMENT '最后变更时间',
    avg_duration   DOUBLE         AVG      COMMENT '平均处理耗时'
)
AGGREGATE KEY(stat_date, tenant_id, object_type, property_name)
PARTITION BY RANGE(stat_date) (
    FROM ("2024-01-01") TO ("2026-12-31") INTERVAL 1 MONTH
)
DISTRIBUTED BY HASH(tenant_id) BUCKETS 8
PROPERTIES (
    "replication_num" = "3"
);

-- coomia-dip 中的 Ontology 对象当前状态表
CREATE TABLE ontology_object_current (
    tenant_id      VARCHAR(64)    NOT NULL COMMENT '租户ID',
    object_type    VARCHAR(128)   NOT NULL COMMENT '对象类型',
    object_rid     VARCHAR(256)   NOT NULL COMMENT '对象RID',
    display_name   VARCHAR(512)   COMMENT '显示名称',
    status         VARCHAR(32)    COMMENT '状态',
    properties     JSON           COMMENT '属性JSON',
    created_at     DATETIME       COMMENT '创建时间',
    updated_at     DATETIME       COMMENT '更新时间',
    version        BIGINT         COMMENT '版本号'
)
UNIQUE KEY(tenant_id, object_type, object_rid)
DISTRIBUTED BY HASH(tenant_id) BUCKETS 16
PROPERTIES (
    "replication_num" = "3",
    "enable_unique_key_merge_on_write" = "true",
    "store_row_column" = "true"
);

需要保留所有明细？
├─ 是 → Duplicate 模型
│      （审计日志、事件流、行为记录）
└─ 否 → 数据有唯一业务键？
         ├─ 是 → 需要部分列更新？
         │       ├─ 是 → Unique 模型 (Merge-on-Write)
         │       │      （维度表、状态表、实时更新表）
         │       └─ 否 → 只需预聚合指标？
         │               ├─ 是 → Aggregate 模型
         │               │      （指标统计、计数器、HLL 去重）
         │               └─ 否 → Unique 模型
         └─ 否 → Duplicate 模型

-- 按月分区，适用于审计日志、指标数据
PARTITION BY RANGE(event_date) (
    PARTITION p202401 VALUES [('2024-01-01'), ('2024-02-01')),
    PARTITION p202402 VALUES [('2024-02-01'), ('2024-03-01')),
    -- 使用动态分区自动管理
)

PROPERTIES (
    "dynamic_partition.enable" = "true",
    "dynamic_partition.time_unit" = "DAY",
    "dynamic_partition.start" = "-30",    -- 保留30天历史
    "dynamic_partition.end" = "3",        -- 预创建3天分区
    "dynamic_partition.prefix" = "p",
    "dynamic_partition.buckets" = "16",
    "dynamic_partition.create_history_partition" = "true"
);

-- 按租户分区，实现物理隔离
PARTITION BY LIST(tenant_id) (
    PARTITION p_tenant_a VALUES IN ("tenant-001"),
    PARTITION p_tenant_b VALUES IN ("tenant-002"),
    PARTITION p_tenant_c VALUES IN ("tenant-003")
)

-- Hash 分桶：适用于点查和 Join 场景
DISTRIBUTED BY HASH(tenant_id, object_rid) BUCKETS 16

-- Random 分桶：适用于无明显分布键的场景
DISTRIBUTED BY RANDOM BUCKETS 32

推荐桶数 = max(
    BE节点数 * CPU核数 / 2,
    单分区数据量(GB) * 1024 / 256   -- 每桶约 256MB
)

EXPLAIN VERBOSE
SELECT
    t.object_type,
    COUNT(*) as total_changes,
    COUNT(DISTINCT object_rid) as unique_objects,
    AVG(duration_ms) as avg_duration
FROM ontology_audit_log t
WHERE t.tenant_id = 'tenant-001'
  AND t.event_time >= '2025-01-01'
  AND t.event_time < '2025-02-01'
GROUP BY t.object_type
ORDER BY total_changes DESC
LIMIT 20;

-- 为审计日志创建按对象类型的聚合物化视图
CREATE MATERIALIZED VIEW mv_audit_by_object_type AS
SELECT
    tenant_id,
    DATE_TRUNC('day', event_time) as event_date,
    object_type,
    action,
    COUNT(*) as action_count,
    COUNT(DISTINCT user_id) as unique_users,
    AVG(duration_ms) as avg_duration
FROM ontology_audit_log
GROUP BY tenant_id, DATE_TRUNC('day', event_time), object_type, action;

-- 查询时 Doris 自动路由到物化视图
SELECT object_type, SUM(action_count)
FROM ontology_audit_log
WHERE tenant_id = 'tenant-001'
GROUP BY object_type;
-- [自动命中 mv_audit_by_object_type]

-- 将经常 Join 的表放入同一个 Colocation Group
CREATE TABLE ontology_objects (
    tenant_id      VARCHAR(64)   NOT NULL,
    object_rid     VARCHAR(256)  NOT NULL,
    object_type    VARCHAR(128)  NOT NULL,
    display_name   VARCHAR(512),
    created_at     DATETIME
)
UNIQUE KEY(tenant_id, object_rid)
DISTRIBUTED BY HASH(tenant_id, object_rid) BUCKETS 16
PROPERTIES (
    "colocate_with" = "ontology_group"
);

CREATE TABLE ontology_relations (
    tenant_id      VARCHAR(64)   NOT NULL,
    source_rid     VARCHAR(256)  NOT NULL,
    target_rid     VARCHAR(256)  NOT NULL,
    relation_type  VARCHAR(128)  NOT NULL,
    created_at     DATETIME
)
DUPLICATE KEY(tenant_id, source_rid, target_rid)
DISTRIBUTED BY HASH(tenant_id, source_rid) BUCKETS 16
PROPERTIES (
    "colocate_with" = "ontology_group"
);

-- 全局 Session 变量设置
SET runtime_filter_mode = "GLOBAL";
SET runtime_filter_type = "IN_OR_BLOOM_FILTER,MIN_MAX";
SET runtime_filter_max_in_num = 4096;
SET runtime_filter_wait_time_ms = 2000;

CREATE ROUTINE LOAD ontology_db.load_audit_events
ON ontology_audit_log
COLUMNS(
    event_time, tenant_id, user_id, object_type,
    object_rid, action, property_name,
    old_value, new_value, source_ip,
    request_id, duration_ms
),
COLUMNS TERMINATED BY "|"
PROPERTIES (
    "desired_concurrent_number" = "3",
    "max_batch_interval" = "20",
    "max_batch_rows" = "200000",
    "max_batch_size" = "104857600",
    "strict_mode" = "false",
    "format" = "json",
    "jsonpaths" = "[
        \"$.event_time\", \"$.tenant_id\", \"$.user_id\",
        \"$.object_type\", \"$.object_rid\", \"$.action\",
        \"$.property_name\", \"$.old_value\", \"$.new_value\",
        \"$.source_ip\", \"$.request_id\", \"$.duration_ms\"
    ]"
)
FROM KAFKA (
    "kafka_broker_list" = "kafka-1:9092,kafka-2:9092,kafka-3:9092",
    "kafka_topic" = "ontology-audit-events",
    "kafka_partitions" = "0,1,2,3,4,5",
    "kafka_offsets" = "OFFSET_BEGINNING",
    "property.group.id" = "doris_audit_consumer",
    "property.kafka_default_offsets" = "OFFSET_END"
);

# data-Layer/src/main/python/doris_stream_load.py
import requests
import json
from typing import List, Dict

class DorisStreamLoader:
    """Doris Stream Load 客户端封装"""

    def __init__(self, fe_host: str, port: int = 8030,
                 user: str = "root", password: str = ""):
        self.base_url = f"http://{fe_host}:{port}"
        self.auth = (user, password)

    def load_json_data(
        self,
        database: str,
        table: str,
        data: List[Dict],
        label: str = None
    ) -> Dict:
        """通过 Stream Load 导入 JSON 数据"""
        url = f"{self.base_url}/api/{database}/{table}/_stream_load"

        headers = {
            "Content-Type": "application/json",
            "Expect": "100-continue",
            "format": "json",
            "strip_outer_array": "true",
            "max_filter_ratio": "0.1"
        }

        if label:
            headers["label"] = label

        payload = json.dumps(data)

        response = requests.put(
            url,
            headers=headers,
            data=payload,
            auth=self.auth
        )

        result = response.json()
        if result.get("Status") != "Success":
            raise Exception(f"Stream Load failed: {result}")

        return result

-- 创建 Iceberg Catalog（使用 Nessie 作为元数据后端）
CREATE CATALOG iceberg_catalog PROPERTIES (
    "type" = "iceberg",
    "iceberg.catalog.type" = "rest",
    "uri" = "http://nessie-server:19120/api/v2",
    "warehouse" = "s3://coomia-dip-lakehouse/warehouse",
    "s3.endpoint" = "http://minio:9000",
    "s3.access-key" = "${S3_ACCESS_KEY}",
    "s3.secret-key" = "${S3_SECRET_KEY}"
);

-- 跨 Catalog 查询：Doris 内部表 JOIN Iceberg 外表
SELECT
    d.tenant_id,
    d.object_type,
    d.action,
    COUNT(*) as recent_count,
    i.total_historical_count
FROM internal.ontology_db.ontology_audit_log d
JOIN iceberg_catalog.lakehouse.historical_audit_stats i
    ON d.tenant_id = i.tenant_id
    AND d.object_type = i.object_type
WHERE d.event_time >= CURRENT_DATE - INTERVAL 7 DAY
GROUP BY d.tenant_id, d.object_type, d.action, i.total_historical_count;

-- 以下查询的 WHERE 条件会被下推到 Iceberg 层
-- 利用 Iceberg 的分区裁剪和 Min/Max 统计信息
SELECT * FROM iceberg_catalog.lakehouse.ontology_snapshots
WHERE snapshot_date = '2025-03-01'  -- 分区裁剪
  AND tenant_id = 'tenant-001'     -- 谓词下推
  AND status = 'ACTIVE';           -- 谓词下推

-- 查看集群状态
SHOW BACKENDS;
SHOW FRONTENDS;

-- 查看表的数据分布
SHOW DATA FROM ontology_audit_log;

-- 查看正在运行的查询
SHOW PROCESSLIST;

-- 查看 Compaction 状态
SHOW TABLET STORAGE FORMAT;

-- 查看 Routine Load 状态
SHOW ROUTINE LOAD FOR load_audit_events;

# deployment-Layer/monitoring/prometheus/doris-targets.yml
- targets:
    - doris-fe-1:8030
    - doris-fe-2:8030
    - doris-fe-3:8030
  labels:
    component: doris-fe

- targets:
    - doris-be-1:8040
    - doris-be-2:8040
    - doris-be-3:8040
  labels:
    component: doris-be

-- 1. 启动新 BE 节点
-- 2. 注册到 FE
ALTER SYSTEM ADD BACKEND "new-be-host:9050";

-- 3. 检查均衡状态
SHOW PROC '/cluster_balance/cluster_load_stat';

-- 4. 手动触发数据均衡（如需）
ADMIN SET FRONTEND CONFIG ("tablet_rebalancer_type" = "BeLoad");

-- 添加列（Light Schema Change，秒级完成）
ALTER TABLE ontology_audit_log
ADD COLUMN correlation_id VARCHAR(128) COMMENT '关联ID';

-- 修改列类型
ALTER TABLE ontology_audit_log
MODIFY COLUMN old_value VARCHAR(4096);

-- 使用复合分桶键分散数据
DISTRIBUTED BY HASH(tenant_id, object_rid) BUCKETS 32

-- 对超大租户使用单独分区
PARTITION BY LIST(tenant_id) (
    PARTITION p_big_tenant VALUES IN ("tenant-big-001"),
    PARTITION p_others VALUES IN ("tenant-002", "tenant-003", ...)
)

# be.conf 调整
compaction_task_num_per_disk = 8
cumulative_compaction_rounds_for_each_base_compaction_round = 15
compaction_policy = time_series  # 时序数据专用策略

-- 设置单查询内存限制
SET exec_mem_limit = 4294967296;  -- 4GB

-- 开启溢写磁盘
SET enable_spill = true;
SET spill_storage_limit = "50GB";

-- 使用资源组隔离不同负载
CREATE WORKLOAD GROUP 'analytics' PROPERTIES (
    "cpu_share" = "20",
    "memory_limit" = "50%",
    "enable_memory_overcommit" = "true"
);

-- 1. 检查是否命中分区裁剪
EXPLAIN SELECT * FROM ontology_audit_log
WHERE event_time > '2025-01-01';

-- 2. 添加物化视图预计算
-- 3. 调整超时参数
SET query_timeout = 600;

-- 4. 使用异步查询（大查询场景）
-- 提交异步查询，返回 query_id
-- 后台执行，客户端轮询结果