┌─────────────────────────────────────────────────┐
│                   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

# Metadata directory
meta_dir = /opt/apache-doris/fe/doris-meta

# JVM heap memory (recommended 50-70% of physical memory)
JAVA_OPTS="-Xmx6g -Xms6g -XX:+UseG1GC -XX:MaxGCPauseMillis=200"

# Query timeout (seconds)
query_timeout = 300

# Maximum connections
qe_max_connection = 4096

# Parallel execution instance count
parallel_fragment_exec_instance_num = 8

# Enable Pipeline execution engine
enable_pipeline_engine = true

# Catalog related
enable_multi_catalog = true

# Audit log
audit_log_dir = /opt/apache-doris/fe/log
audit_log_roll_num = 90

# Storage directories (multi-disk for higher IO throughput)
storage_root_path = /data1;/data2;/data3

# Memory limit (recommended 80% of physical memory)
mem_limit = 80%

# Compaction thread count
compaction_task_num_per_disk = 4

# Vectorized execution engine
enable_vectorized_engine = true

# Page Cache
disable_storage_page_cache = false
storage_page_cache_limit = 20%

# Chunk size (affects query memory usage)
chunk_reserved_bytes_limit = 2147483648

# Network buffer
brpc_num_threads = 256
thrift_server_max_worker_threads = 4096

-- Ontology operation audit log table in coomia-dip
CREATE TABLE ontology_audit_log (
    event_time     DATETIME       NOT NULL COMMENT 'Event time',
    tenant_id      VARCHAR(64)    NOT NULL COMMENT 'Tenant ID',
    user_id        VARCHAR(64)    NOT NULL COMMENT 'User ID',
    object_type    VARCHAR(128)   NOT NULL COMMENT 'Object type',
    object_rid     VARCHAR(256)   NOT NULL COMMENT 'Object RID',
    action         VARCHAR(32)    NOT NULL COMMENT 'Action type',
    property_name  VARCHAR(128)   COMMENT 'Property name',
    old_value      TEXT           COMMENT 'Old value',
    new_value      TEXT           COMMENT 'New value',
    source_ip      VARCHAR(45)    COMMENT 'Source IP',
    request_id     VARCHAR(64)    COMMENT 'Request ID',
    duration_ms    INT            COMMENT 'Processing time (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"
);

-- Object property change statistics table in coomia-dip
CREATE TABLE ontology_property_stats (
    stat_date      DATE           NOT NULL COMMENT 'Statistics date',
    tenant_id      VARCHAR(64)    NOT NULL COMMENT 'Tenant ID',
    object_type    VARCHAR(128)   NOT NULL COMMENT 'Object type',
    property_name  VARCHAR(128)   NOT NULL COMMENT 'Property name',
    change_count   BIGINT         SUM      COMMENT 'Change count',
    unique_objects HLL            HLL_UNION COMMENT 'Distinct object count',
    last_changed   DATETIME       MAX      COMMENT 'Last change time',
    avg_duration   DOUBLE         AVG      COMMENT 'Average processing time'
)
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"
);

-- Ontology object current state table in coomia-dip
CREATE TABLE ontology_object_current (
    tenant_id      VARCHAR(64)    NOT NULL COMMENT 'Tenant ID',
    object_type    VARCHAR(128)   NOT NULL COMMENT 'Object type',
    object_rid     VARCHAR(256)   NOT NULL COMMENT 'Object RID',
    display_name   VARCHAR(512)   COMMENT 'Display name',
    status         VARCHAR(32)    COMMENT 'Status',
    properties     JSON           COMMENT 'Properties JSON',
    created_at     DATETIME       COMMENT 'Created at',
    updated_at     DATETIME       COMMENT 'Updated at',
    version        BIGINT         COMMENT 'Version'
)
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"
);

Need to retain all detail records?
├─ Yes → Duplicate Model
│      (Audit logs, event streams, behavioral records)
└─ No  → Data has a unique business key?
         ├─ Yes → Need partial column updates?
         │       ├─ Yes → Unique Model (Merge-on-Write)
         │       │      (Dimension tables, state tables, real-time update tables)
         │       └─ No  → Only need pre-aggregated metrics?
         │               ├─ Yes → Aggregate Model
         │               │      (Metric statistics, counters, HLL dedup)
         │               └─ No  → Unique Model
         └─ No  → Duplicate Model

-- Monthly partition, suitable for audit logs and metric data
PARTITION BY RANGE(event_date) (
    PARTITION p202401 VALUES [('2024-01-01'), ('2024-02-01')),
    PARTITION p202402 VALUES [('2024-02-01'), ('2024-03-01')),
    -- Use dynamic partitions for automatic management
)

PROPERTIES (
    "dynamic_partition.enable" = "true",
    "dynamic_partition.time_unit" = "DAY",
    "dynamic_partition.start" = "-30",    -- Retain 30 days of history
    "dynamic_partition.end" = "3",        -- Pre-create 3 days of partitions
    "dynamic_partition.prefix" = "p",
    "dynamic_partition.buckets" = "16",
    "dynamic_partition.create_history_partition" = "true"
);

-- Partition by tenant for physical isolation
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 bucketing: suitable for point queries and joins
DISTRIBUTED BY HASH(tenant_id, object_rid) BUCKETS 16

-- Random bucketing: suitable for scenarios without clear distribution keys
DISTRIBUTED BY RANDOM BUCKETS 32

Recommended buckets = max(
    BE_nodes * CPU_cores / 2,
    partition_data_size_GB * 1024 / 256   -- ~256MB per bucket
)

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 aggregation materialized view for audit logs by object type
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 automatically routes queries to the materialized view
SELECT object_type, SUM(action_count)
FROM ontology_audit_log
WHERE tenant_id = 'tenant-001'
GROUP BY object_type;
-- [Automatically hits mv_audit_by_object_type]

-- Place frequently joined tables in the same 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"
);

-- Global session variable settings
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 client wrapper"""

    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:
        """Import JSON data via Stream Load"""
        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

-- Create Iceberg Catalog (using Nessie as metadata backend)
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}"
);

-- Cross-catalog query: Doris internal table JOIN Iceberg external table
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;

-- The WHERE conditions in this query are pushed down to the Iceberg layer
-- Utilizing Iceberg's partition pruning and Min/Max statistics
SELECT * FROM iceberg_catalog.lakehouse.ontology_snapshots
WHERE snapshot_date = '2025-03-01'  -- Partition pruning
  AND tenant_id = 'tenant-001'     -- Predicate pushdown
  AND status = 'ACTIVE';           -- Predicate pushdown

-- Check cluster status
SHOW BACKENDS;
SHOW FRONTENDS;

-- Check table data distribution
SHOW DATA FROM ontology_audit_log;

-- Check running queries
SHOW PROCESSLIST;

-- Check Compaction status
SHOW TABLET STORAGE FORMAT;

-- Check Routine Load status
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. Start new BE node
-- 2. Register with FE
ALTER SYSTEM ADD BACKEND "new-be-host:9050";

-- 3. Check balancing status
SHOW PROC '/cluster_balance/cluster_load_stat';

-- 4. Manually trigger data rebalancing (if needed)
ADMIN SET FRONTEND CONFIG ("tablet_rebalancer_type" = "BeLoad");

-- Add column (Light Schema Change, completes in seconds)
ALTER TABLE ontology_audit_log
ADD COLUMN correlation_id VARCHAR(128) COMMENT 'Correlation ID';

-- Modify column type
ALTER TABLE ontology_audit_log
MODIFY COLUMN old_value VARCHAR(4096);

-- Use composite bucket keys to distribute data
DISTRIBUTED BY HASH(tenant_id, object_rid) BUCKETS 32

-- Use separate partitions for very large tenants
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 adjustments
compaction_task_num_per_disk = 8
cumulative_compaction_rounds_for_each_base_compaction_round = 15
compaction_policy = time_series  # Policy optimized for time-series data

-- Set per-query memory limit
SET exec_mem_limit = 4294967296;  -- 4GB

-- Enable spilling to disk
SET enable_spill = true;
SET spill_storage_limit = "50GB";

-- Use resource groups to isolate different workloads
CREATE WORKLOAD GROUP 'analytics' PROPERTIES (
    "cpu_share" = "20",
    "memory_limit" = "50%",
    "enable_memory_overcommit" = "true"
);

-- 1. Check if partition pruning is working
EXPLAIN SELECT * FROM ontology_audit_log
WHERE event_time > '2025-01-01';

-- 2. Add materialized views for pre-computation
-- 3. Adjust timeout parameters
SET query_timeout = 600;

-- 4. Use async queries (for large query scenarios)
-- Submit async query, returns query_id
-- Background execution, client polls for results