OLAP 分析     → ClickHouse / Doris
全文检索       → Elasticsearch
向量语义搜索   → Milvus / Pinecone

┌──────────────────────────────────────┐
│            Doris Segment             │
│  ┌─────────────┬──────────────────┐  │
│  │ Column Data  │ Inverted Index   │  │
│  │ (Columnar)   │ (CLucene)        │  │
│  │              │ ┌──────────────┐ │  │
│  │ tenant_id    │ │ Term Dict    │ │  │
│  │ object_type  │ │ Posting List │ │  │
│  │ display_name │ │ Doc Values   │ │  │
│  │ description  │ │ Norms        │ │  │
│  │ properties   │ └──────────────┘ │  │
│  └─────────────┴──────────────────┘  │
└──────────────────────────────────────┘

CREATE TABLE ontology_object_searchable (
    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)   NOT NULL COMMENT '显示名称',
    description    TEXT           COMMENT '描述',
    tags           ARRAY<VARCHAR(64)> COMMENT '标签数组',
    properties_json TEXT          COMMENT '属性JSON',
    status         VARCHAR(32)    COMMENT '状态',
    created_at     DATETIME       COMMENT '创建时间',
    updated_at     DATETIME       COMMENT '更新时间',

    -- 倒排索引定义
    INDEX idx_display_name (display_name)
        USING INVERTED
        PROPERTIES("parser" = "unicode", "support_phrase" = "true"),

    INDEX idx_description (description)
        USING INVERTED
        PROPERTIES(
            "parser" = "unicode",
            "support_phrase" = "true",
            "lower_case" = "true"
        ),

    INDEX idx_tags (tags)
        USING INVERTED
        PROPERTIES("parser" = "unicode"),

    INDEX idx_properties (properties_json)
        USING INVERTED
        PROPERTIES("parser" = "unicode", "support_phrase" = "true"),

    INDEX idx_status (status)
        USING INVERTED,

    INDEX idx_object_type (object_type)
        USING INVERTED
)
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",
    "inverted_index_storage_format" = "V2",
    "store_row_column" = "true"
);

-- 基础全文搜索：MATCH_ANY（OR 语义）
SELECT object_rid, display_name, description
FROM ontology_object_searchable
WHERE tenant_id = 'tenant-001'
  AND display_name MATCH_ANY '供应链 风险'
ORDER BY updated_at DESC
LIMIT 20;

-- 短语搜索：MATCH_PHRASE（要求词序连续）
SELECT object_rid, display_name, description
FROM ontology_object_searchable
WHERE tenant_id = 'tenant-001'
  AND description MATCH_PHRASE '供应链风险评估'
LIMIT 20;

-- 全匹配搜索：MATCH_ALL（AND 语义）
SELECT object_rid, display_name, description
FROM ontology_object_searchable
WHERE tenant_id = 'tenant-001'
  AND description MATCH_ALL '供应链 风险 评估'
LIMIT 20;

-- 前缀搜索：MATCH_PHRASE_PREFIX
SELECT object_rid, display_name
FROM ontology_object_searchable
WHERE tenant_id = 'tenant-001'
  AND display_name MATCH_PHRASE_PREFIX '供应链'
LIMIT 20;

-- 组合条件：全文搜索 + 精确过滤
SELECT object_rid, display_name, description, status, created_at
FROM ontology_object_searchable
WHERE tenant_id = 'tenant-001'
  AND description MATCH_ANY '风险 评估 预警'
  AND status = 'ACTIVE'
  AND created_at >= '2025-01-01'
ORDER BY created_at DESC
LIMIT 50;

Layer 3:  [A] ────────────────── [F]
           |                      |
Layer 2:  [A] ──── [C] ──── [F] ──── [H]
           |        |        |        |
Layer 1:  [A] ─ [B] ─ [C] ─ [D] ─ [F] ─ [G] ─ [H]
           |    |    |    |    |    |    |    |
Layer 0:  [A]-[B]-[C]-[D]-[E]-[F]-[G]-[H]-[I]-[J]

CREATE TABLE ontology_embeddings (
    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 '显示名称',
    description     TEXT           COMMENT '描述文本',
    embedding       ARRAY<FLOAT>   NOT NULL COMMENT 'Embedding 向量 (768维)',
    model_version   VARCHAR(32)    COMMENT '模型版本',
    created_at      DATETIME       COMMENT '创建时间',

    -- 向量索引
    INDEX idx_embedding (embedding)
        USING INVERTED
        PROPERTIES(
            "index_type" = "HNSW",
            "metric_type" = "L2",
            "dim" = "768",
            "M" = "32",
            "efConstruction" = "400"
        ),

    -- 文本倒排索引（用于混合查询）
    INDEX idx_display_name (display_name)
        USING INVERTED
        PROPERTIES("parser" = "unicode"),

    INDEX idx_description (description)
        USING INVERTED
        PROPERTIES("parser" = "unicode", "support_phrase" = "true")
)
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"
);

# intelligence-Layer/embedding_indexer.py
import numpy as np
from typing import List, Dict
from sentence_transformers import SentenceTransformer
from doris_stream_loader import DorisStreamLoader

class OntologyEmbeddingIndexer:
    """Ontology 对象 Embedding 索引器"""

    def __init__(
        self,
        model_name: str = "BAAI/bge-base-zh-v1.5",
        doris_host: str = "doris-fe-1",
        doris_port: int = 8030
    ):
        self.model = SentenceTransformer(model_name)
        self.loader = DorisStreamLoader(doris_host, doris_port)
        self.model_version = model_name.split("/")[-1]

    def index_objects(
        self,
        tenant_id: str,
        objects: List[Dict]
    ) -> int:
        """批量索引 Ontology 对象"""
        records = []
        texts = []

        for obj in objects:
            text = f"{obj['display_name']}. {obj.get('description', '')}"
            texts.append(text)

        # 批量编码
        embeddings = self.model.encode(
            texts,
            batch_size=64,
            show_progress_bar=True,
            normalize_embeddings=True
        )

        for obj, embedding in zip(objects, embeddings):
            records.append({
                "tenant_id": tenant_id,
                "object_type": obj["object_type"],
                "object_rid": obj["object_rid"],
                "display_name": obj["display_name"],
                "description": obj.get("description", ""),
                "embedding": embedding.tolist(),
                "model_version": self.model_version,
                "created_at": obj.get("created_at")
            })

        # 批量导入
        result = self.loader.load_json_data(
            database="ontology_db",
            table="ontology_embeddings",
            data=records
        )

        return len(records)

    def search_similar(
        self,
        tenant_id: str,
        query_text: str,
        top_k: int = 10,
        object_type: str = None
    ) -> List[Dict]:
        """语义相似搜索"""
        query_embedding = self.model.encode(
            query_text,
            normalize_embeddings=True
        )

        # 构建 SQL
        embedding_str = ", ".join(str(x) for x in query_embedding.tolist())

        sql = f"""
        SELECT
            object_rid,
            display_name,
            description,
            object_type,
            L2_DISTANCE(embedding, ARRAY[{embedding_str}]) as distance
        FROM ontology_embeddings
        WHERE tenant_id = '{tenant_id}'
        """

        if object_type:
            sql += f" AND object_type = '{object_type}'"

        sql += f"""
        ORDER BY distance ASC
        LIMIT {top_k}
        """

        return self._execute_query(sql)

-- 基础向量相似度搜索（L2 距离）
SELECT
    object_rid,
    display_name,
    L2_DISTANCE(embedding, ARRAY[0.1, 0.2, ...]) as distance
FROM ontology_embeddings
WHERE tenant_id = 'tenant-001'
ORDER BY distance ASC
LIMIT 10;

-- 余弦相似度搜索
SELECT
    object_rid,
    display_name,
    COSINE_DISTANCE(embedding, ARRAY[0.1, 0.2, ...]) as distance
FROM ontology_embeddings
WHERE tenant_id = 'tenant-001'
ORDER BY distance ASC
LIMIT 10;

-- 内积搜索
SELECT
    object_rid,
    display_name,
    INNER_PRODUCT(embedding, ARRAY[0.1, 0.2, ...]) as score
FROM ontology_embeddings
WHERE tenant_id = 'tenant-001'
ORDER BY score DESC
LIMIT 10;

用户查询："找到最近30天创建的、与'供应链风险评估'语义相似的活跃对象"
                    │
        ┌───────────┼───────────┐
        ▼           ▼           ▼
   向量检索     全文检索     OLAP过滤
   (语义相似)   (关键词匹配)  (时间+状态)
        │           │           │
        └───────────┼───────────┘
                    ▼
              结果融合排序
              (加权评分)

-- 混合查询：向量相似度 + 全文匹配 + OLAP 过滤
SELECT
    e.object_rid,
    e.display_name,
    e.description,
    e.object_type,
    L2_DISTANCE(e.embedding, ARRAY[/* query embedding */]) as vector_distance,
    e.created_at
FROM ontology_embeddings e
WHERE e.tenant_id = 'tenant-001'
  -- OLAP 过滤
  AND e.created_at >= DATE_SUB(NOW(), INTERVAL 30 DAY)
  -- 全文检索
  AND e.description MATCH_ANY '供应链 风险'
ORDER BY vector_distance ASC
LIMIT 20;

-- 加权混合评分查询
WITH vector_results AS (
    SELECT
        object_rid,
        display_name,
        description,
        L2_DISTANCE(embedding, ARRAY[/* query embedding */]) as v_dist,
        ROW_NUMBER() OVER (ORDER BY L2_DISTANCE(embedding, ARRAY[/* query embedding */])) as v_rank
    FROM ontology_embeddings
    WHERE tenant_id = 'tenant-001'
    ORDER BY v_dist ASC
    LIMIT 100
),
text_results AS (
    SELECT
        object_rid,
        display_name,
        description,
        ROW_NUMBER() OVER (ORDER BY updated_at DESC) as t_rank
    FROM ontology_object_searchable
    WHERE tenant_id = 'tenant-001'
      AND description MATCH_ANY '供应链 风险 评估'
    LIMIT 100
)
SELECT
    COALESCE(v.object_rid, t.object_rid) as object_rid,
    COALESCE(v.display_name, t.display_name) as display_name,
    -- 加权评分：向量相似度权重 0.6，全文匹配权重 0.4
    (COALESCE(1.0 / v.v_rank, 0) * 0.6 +
     COALESCE(1.0 / t.t_rank, 0) * 0.4) as hybrid_score
FROM vector_results v
FULL OUTER JOIN text_results t ON v.object_rid = t.object_rid
ORDER BY hybrid_score DESC
LIMIT 20;

# intelligence-Layer/search/hybrid_search.py
from dataclasses import dataclass, field
from typing import List, Optional
from enum import Enum

class SearchMode(Enum):
    KEYWORD = "keyword"
    SEMANTIC = "semantic"
    HYBRID = "hybrid"

@dataclass
class SearchRequest:
    tenant_id: str
    query: str
    mode: SearchMode = SearchMode.HYBRID
    object_types: Optional[List[str]] = None
    status_filter: Optional[str] = None
    date_from: Optional[str] = None
    date_to: Optional[str] = None
    top_k: int = 20
    vector_weight: float = 0.6
    text_weight: float = 0.4

@dataclass
class SearchResult:
    object_rid: str
    display_name: str
    description: str
    object_type: str
    score: float
    highlights: List[str] = field(default_factory=list)

class HybridSearchEngine:
    """混合搜索引擎 — 统一向量 + 全文 + OLAP"""

    def __init__(self, doris_connection, embedding_model):
        self.conn = doris_connection
        self.model = embedding_model

    async def search(self, request: SearchRequest) -> List[SearchResult]:
        if request.mode == SearchMode.KEYWORD:
            return await self._keyword_search(request)
        elif request.mode == SearchMode.SEMANTIC:
            return await self._semantic_search(request)
        else:
            return await self._hybrid_search(request)

    async def _hybrid_search(self, request: SearchRequest) -> List[SearchResult]:
        # 1. 生成查询向量
        query_embedding = self.model.encode(
            request.query,
            normalize_embeddings=True
        )

        # 2. 构建混合查询 SQL
        embedding_str = ", ".join(str(x) for x in query_embedding.tolist())

        filters = [f"tenant_id = '{request.tenant_id}'"]
        if request.object_types:
            types_str = ", ".join(f"'{t}'" for t in request.object_types)
            filters.append(f"object_type IN ({types_str})")
        if request.status_filter:
            filters.append(f"status = '{request.status_filter}'")
        if request.date_from:
            filters.append(f"created_at >= '{request.date_from}'")
        if request.date_to:
            filters.append(f"created_at <= '{request.date_to}'")

        where_clause = " AND ".join(filters)

        sql = f"""
        WITH vector_scores AS (
            SELECT object_rid, display_name, description, object_type,
                   1.0 / (1.0 + L2_DISTANCE(embedding, ARRAY[{embedding_str}])) as v_score
            FROM ontology_embeddings
            WHERE {where_clause}
            ORDER BY L2_DISTANCE(embedding, ARRAY[{embedding_str}]) ASC
            LIMIT {request.top_k * 3}
        ),
        text_scores AS (
            SELECT object_rid, display_name, description, object_type,
                   1.0 as t_score
            FROM ontology_object_searchable
            WHERE {where_clause}
              AND (display_name MATCH_ANY '{request.query}'
                   OR description MATCH_ANY '{request.query}')
            LIMIT {request.top_k * 3}
        )
        SELECT
            COALESCE(v.object_rid, t.object_rid) as object_rid,
            COALESCE(v.display_name, t.display_name) as display_name,
            COALESCE(v.description, t.description) as description,
            COALESCE(v.object_type, t.object_type) as object_type,
            (COALESCE(v.v_score, 0) * {request.vector_weight} +
             COALESCE(t.t_score, 0) * {request.text_weight}) as score
        FROM vector_scores v
        FULL OUTER JOIN text_scores t ON v.object_rid = t.object_rid
        ORDER BY score DESC
        LIMIT {request.top_k}
        """

        rows = await self.conn.execute(sql)
        return [
            SearchResult(
                object_rid=row["object_rid"],
                display_name=row["display_name"],
                description=row["description"],
                object_type=row["object_type"],
                score=row["score"]
            )
            for row in rows
        ]

-- 查看索引构建进度
SHOW BUILD INDEX FROM ontology_db;

-- 查看表的索引信息
SHOW INDEX FROM ontology_embeddings;

-- 查看 Segment 级别的索引状态
SHOW TABLET FROM ontology_embeddings;

-- 删除并重建倒排索引
DROP INDEX idx_description ON ontology_object_searchable;

CREATE INDEX idx_description ON ontology_object_searchable(description)
    USING INVERTED
    PROPERTIES("parser" = "unicode", "support_phrase" = "true");

-- 触发索引重建
BUILD INDEX idx_description ON ontology_object_searchable;

-- 查看索引存储占用
SHOW DATA FROM ontology_embeddings;

-- 向量索引压缩（PQ 量化）
ALTER TABLE ontology_embeddings
MODIFY INDEX idx_embedding
PROPERTIES("index_type" = "HNSW", "pq_enable" = "true", "pq_subvector_num" = "48");

# be.conf — 向量检索相关
enable_vectorized_engine = true
inverted_index_ram_dir_enable = true  # 索引缓存到内存

# 向量索引搜索线程
vector_index_search_thread_pool_size = 16

# HNSW 搜索参数（全局默认）
default_hnsw_ef_search = 200

# be.conf — 倒排索引相关
inverted_index_cache_stale_sweep_time_sec = 600
inverted_index_max_buffered_docs = 65536
inverted_index_compaction_enable = true

# 分词器缓存
inverted_index_searcher_cache_limit = 10%

总内存 64GB 分配方案：
├── 查询执行引擎    30GB (47%)
├── 向量索引缓存    10GB (16%)
├── 倒排索引缓存     8GB (12%)
├── 页缓存           8GB (12%)
├── Compaction        5GB (8%)
└── 系统预留          3GB (5%)

-- 使用 model_version 字段隔离不同模型的向量
SELECT * FROM ontology_embeddings
WHERE tenant_id = 'tenant-001'
  AND model_version = 'bge-base-zh-v1.5'
ORDER BY L2_DISTANCE(embedding, ARRAY[...]) ASC
LIMIT 10;

-- 对长文本字段限制索引 token 数
ALTER TABLE ontology_object_searchable
MODIFY INDEX idx_description
PROPERTIES("max_token_per_doc" = "10000");

-- 策略1：先用标量过滤缩小范围，再做向量检索
-- 策略2：使用分区键 + 向量索引，利用分区裁剪
-- 策略3：两阶段查询——先粗筛再精排