OceanBase2025 决赛回顾

SELECT `docid_col`, MATCH(`fulltext_col`) AGAINST(`text`) as _score
FROM `items1`
WHERE MATCH(fulltext_col) AGAINST(`text`)
  AND base_id IN ('base_id_1', 'base_id_2', 'base_id_3')
  AND id < 1000
ORDER BY _score DESC
LIMIT 10

CREATE TABLE `items1` (
  `id` BIGINT AUTO_INCREMENT,
  `base_id` VARCHAR(255) NOT NULL,
  `docid_col` VARCHAR(255) NOT NULL,
  `fulltext_col` LONGTEXT
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

-- 普通索引
CREATE INDEX idx_base_id ON `items1`(`base_id`);
CREATE INDEX idx_docid ON `items1`(`docid_col`);

-- 全文索引
CREATE /*+ parallel(90) */ FULLTEXT INDEX ft_fulltext ON `items1`(`fulltext_col`);

初始状态:  ~5分   (无任何优化，基本等同于全表扫描)
TopN下推:  ~100分  (强制走全文本索引算子 + 开启TopN谓词下推)
IDF缓存:   ~500分  (初版全局hashmap缓存，后迁移到ObKVCache)
结果缓存:  ~1300分 (缓存倒排链扫描结果，避免重复IO)
最终状态:  未能提交有效成绩（Bitmap方案未完成，导致违规内容被取消）

用户SQL查询
    │
    ▼
[SQL层] 查询解析 → 逻辑计划 → 优化器 → 物理计划
    │
    ▼
[DAS层] 分布式访问服务 (Data Access Service)
    │  - 管理倒排索引扫描
    │  - 管理正排索引聚合
    │  - 多token合并
    ▼
[存储层] Token迭代器 → 合并算法(TAAT/DaaT/BMW) → TopK结果
    │
    ▼
[索引层] SSTable倒排索引 + 正排索引 (LSM-Tree存储)

// ob_log_plan.cpp（简化）
// 判断是否可以将TopN下推到全文检索扫描
int ObLogPlan::try_push_topn_into_text_retrieval_scan(ObLogTopk *topk_op, ...) {
  // 检查是否有标量过滤条件（原始代码在有filter时会放弃下推）
  // [优化]：移除这个检查，允许带filter的TopN下推
  // 这样TEXT_RETRIEVAL_SCAN可以知道只需要返回Top-N个结果
}

ObDASTextRetrievalIter          — 单token检索（DAS入口）
ObDASTRCacheIter                — 带结果缓存的单token检索
ObDASTextRetrievalMergeIter     — 多token合并检索（DAS层）
ObDAsTrMergeIter                — 稀疏检索合并迭代器

ObTextRetrievalTokenIter        — 单token的倒排链迭代
ObTextRetrievalDaaTTokenIter    — 包装为DaaT接口的Token迭代器
ObTextRetrievalBlockMaxIter     — 带块最大分数的Token迭代器（供BMW使用）

ObISparseRetrievalDimIter           (基础接口)
├── get_next_row()
├── get_next_batch()
└── advance_to(id_datum)
    │
    └── ObISRDaaTDimIter            (DaaT接口)
        ├── get_curr_score()
        ├── get_curr_id()
        ├── get_dim_max_score()
        │
        └── ObISRDimBlockMaxIter    (Block-Max接口)
            ├── advance_shallow()   — 浅推进（不移动游标，只更新块信息）
            ├── get_curr_block_max_info()
            └── in_shallow_status()

BM25(D, Q) = Σ_{t∈Q} IDF(t) × (tf(t,D) × (k1+1)) / (tf(t,D) + k1×(1 - b + b×|D|/avgdl))

IDF(t) = ln((N - df(t) + 0.5) / (df(t) + 0.5) + 1)

// BM25评分计算（简化示意）
double calc_bm25_score(
    int64_t token_count,     // tf(t,D)：token在文档中的出现次数
    int64_t doc_length,      // |D|：文档长度
    int64_t token_doc_cnt,   // df(t)：token的文档频率
    int64_t total_doc_cnt,   // N：总文档数
    double avg_doc_length)   // avgdl：平均文档长度
{
    const double k1 = 1.2;
    const double b = 0.75;

    // IDF计算
    double idf = std::log((total_doc_cnt - token_doc_cnt + 0.5) /
                          (token_doc_cnt + 0.5) + 1.0);

    // TF归一化
    double tf_norm = (token_count * (k1 + 1)) /
                     (token_count + k1 * (1 - b + b * doc_length / avg_doc_length));

    return idf * tf_norm;
}

Token1的倒排链: doc1→score1.1, doc2→score1.2, doc5→score1.5
Token2的倒排链: doc1→score2.1, doc3→score2.3, doc5→score2.5

处理Token1: acc[doc1]+=score1.1, acc[doc2]+=score1.2, acc[doc5]+=score1.5
处理Token2: acc[doc1]+=score2.1, acc[doc3]+=score2.3, acc[doc5]+=score2.5
最终: 对acc数组取Top-K

Token1: [doc1, doc2, doc5, ...]
Token2: [doc1, doc3, doc5, ...]
Token3: [doc2, doc5, doc7, ...]

合并堆: 按doc_id排序
→ 取doc1: 合并Token1和Token2的分数
→ 取doc2: 合并Token1和Token3的分数
→ 取doc3: 只有Token2的分数
→ 取doc5: 合并Token1、Token2、Token3的分数
...

// 败者树（Loser Tree）用于多路归并
ObSRMergeLoserTree merge_heap_;

// 合并项
struct ObSRMergeItem {
    int64_t iter_idx_;    // 对应哪个token的迭代器
    double relevance_;     // 当前doc_id在这个token下的分数
};

// 迭代器域ID数组（每个token当前的doc_id）
ObFixedArray<const ObDatum *, ObIAllocator> iter_domain_ids_;

FIND_NEXT_PIVOT
    │ 找到pivot
    ▼
EVALUATE_PIVOT_RANGE ←─────┐
    │ 块级上界 ≤ θ           │
    ├──────────────────► FIND_NEXT_PIVOT_RANGE
    │ 块级上界 > θ           │
    ▼                       │
EVALUATE_PIVOT              │ 未找到候选范围
    │ 精确评估，更新Top-K    │
    ▼                       │
FIND_NEXT_PIVOT ────────────┘

struct ObMaxScoreTuple {
    const ObDatum *min_domain_id_;   // 块的起始doc_id
    const ObDatum *max_domain_id_;   // 块的结束doc_id
    double max_score_;               // 块内最大可能BM25分数
};

// 原始代码：有scalar filter时不做TopN下推
int ObLogPlan::try_push_topn_into_text_retrieval_scan(...) {
    // 检查是否有过滤条件
    if (has_scalar_filter) {
        // 放弃下推
        return OB_SUCCESS;
    }
    // ...做下推
}

// 优化后：移除这个限制，允许带filter的TopN下推
int ObLogPlan::try_push_topn_into_text_retrieval_scan(...) {
    // 直接做下推，不检查scalar filter
    // 全文检索算子会在内部处理filter
}

// 将全文索引的代价估算置为0，强制优化器选择全文索引路径
// 原始代码有复杂的代价计算，导致有时选择不走全文索引
fulltext_scan_cost = 0;  // 强制走全文索引

IDF(t) = ln((N - df(t) + 0.5) / (df(t) + 0.5) + 1)

// 第一版（提交fd702ae8）：简单全局hashmap
static std::unordered_map<std::string, int64_t> g_token_df_cache;
static oceanbase::common::SpinRWLock g_df_cache_lock;

// 查询时
int ObTextRetrievalTokenIter::estimate_token_doc_cnt() {
    if (!curr_token_text_.empty()) {
        oceanbase::common::SpinRLockGuard guard(g_df_cache_lock);
        auto it = g_token_df_cache.find(curr_token_text_);
        if (it != g_token_df_cache.end()) {
            token_doc_cnt_ = it->second;
            token_doc_cnt_calculated_ = true;
            return OB_SUCCESS;  // 缓存命中，直接返回
        }
    }
    // 缓存未命中：执行倒排索引扫描
    // ...

    // 计算完成后写入缓存
    if (!curr_token_text_.empty()) {
        oceanbase::common::SpinWLockGuard guard(g_df_cache_lock);
        g_token_df_cache[curr_token_text_] = token_doc_cnt_;
    }
}

class ObTokenDFCacheKey : public common::ObIKVCacheKey {
public:
    uint64_t tenant_id_;       // 租户ID（支持多租户）
    ObTabletID tablet_id_;     // 数据库对象标识符
    ObString token_;           // token文本（变长字符串）

    // 哈希计算
    int hash(uint64_t &hash_val) const {
        hash_val = murmurhash(&tenant_id_, sizeof(tenant_id_), hash_val);
        hash_val = murmurhash(&tablet_id_, sizeof(tablet_id_), hash_val);
        hash_val = murmurhash(token_.ptr(), token_.length(), hash_val);
        return OB_SUCCESS;
    }

    // 深拷贝（缓存框架要求）
    int deep_copy(char *buf, const int64_t buf_len, ObIKVCacheKey *&key) const {
        // 将token_内容内联存储在Key对象后面
        ObTokenDFCacheKey *new_key = new (buf) ObTokenDFCacheKey(...);
        char *data_ptr = buf + sizeof(*this);
        MEMCPY(data_ptr, token_.ptr(), token_.length());
        new_key->token_.assign_ptr(data_ptr, token_.length());
        key = new_key;
        return OB_SUCCESS;
    }

    int64_t size() const {
        return sizeof(*this) + token_.length();  // 变长Key
    }
};

class ObTokenDFCacheValue : public common::ObIKVCacheValue {
public:
    int64_t token_doc_cnt_;       // df(t)：含有该token的文档数
    double max_token_relevance_;   // 该token能贡献的最大BM25分数（用于BMW剪枝）

    int64_t size() const { return sizeof(*this); }

    int deep_copy(char *buf, ...) const {
        // 简单的固定大小值，直接拷贝即可
        ObTokenDFCacheValue *new_val = new (buf) ObTokenDFCacheValue();
        new_val->token_doc_cnt_ = token_doc_cnt_;
        new_val->max_token_relevance_ = max_token_relevance_;
        return OB_SUCCESS;
    }
};

class ObStorageCacheSuite {
public:
    ObTokenDFCache token_df_cache_;     // IDF缓存
    ObTokenResultCache token_result_cache_;  // 结果缓存

    ObTokenDFCache &get_token_df_cache() { return token_df_cache_; }
    ObTokenResultCache &get_token_result_cache() { return token_result_cache_; }
};

// 初始化时尝试从缓存读取
int ObTextRetrievalTokenIter::init(const ObTextRetrievalScanIterParam &iter_param) {
    // ...初始化代码...

    if (!curr_token_text_.empty()) {
        const uint64_t tenant_id = MTL_ID();
        blocksstable::ObTokenDFCacheKey key(
            tenant_id,
            inv_idx_agg_param_->tablet_id_,
            ObString(curr_token_text_.length(), curr_token_text_.c_str()));
        blocksstable::ObTokenDFValueHandle handle;
        if (OB_SUCCESS == OB_STORE_CACHE.get_token_df_cache().get_token_df(key, handle)) {
            // 缓存命中：直接使用缓存的IDF值
            token_doc_cnt_ = handle.value_->get_token_doc_cnt();
            max_token_relevance_ = handle.value_->get_max_token_relevance();
            token_doc_cnt_calculated_ = true;  // 标记已计算，跳过倒排索引扫描
        }
    }
}

// 计算完IDF后写入缓存
int ObTextRetrievalTokenIter::update_max_token_relevance(const double max_token_relevance) {
    max_token_relevance_ = max_token_relevance;
    if (!curr_token_text_.empty()) {
        const uint64_t tenant_id = MTL_ID();
        blocksstable::ObTokenDFCacheKey key(...);
        blocksstable::ObTokenDFCacheValue token_value;
        if (OB_SUCC(token_value.init(token_doc_cnt_, max_token_relevance_))) {
            OB_STORE_CACHE.get_token_df_cache().put_token_df(key, token_value);
        }
    }
    return ret;
}

class ObTokenResultCacheKey : public common::ObIKVCacheKey {
public:
    uint64_t tenant_id_;
    uint64_t tablet_id_;       // 注意：这里用uint64_t而非ObTabletID
    ObString token_;
    int64_t start_doc_id_;     // 分页起始位置，支持大结果集分页缓存

    int64_t size() const {
        return sizeof(*this) + token_.length();
    }
};

class ObTokenResultCacheValue : public common::ObIKVCacheValue {
public:
    int64_t count_;            // 本次缓存的结果数
    const ObDatum *doc_ids_;   // doc_id数组（指向内部buffer）
    const double *scores_;     // score数组（指向内部buffer）

    // 计算总大小（包含内联存储的数组数据）
    int64_t size() const {
        int64_t total_size = sizeof(*this);
        total_size += count_ * sizeof(double);    // scores数组
        total_size += count_ * sizeof(ObDatum);   // doc_ids数组
        // doc_ids的payload数据（变长字符串）
        for (int64_t i = 0; i < count_; ++i) {
            if (!doc_ids_[i].is_null() && doc_ids_[i].len_ > 0) {
                total_size = upper_align(total_size, 8);  // 8字节对齐
                total_size += doc_ids_[i].len_;
            }
        }
        return total_size;
    }

    // 深拷贝：将所有数据内联存储在缓存buffer中
    int deep_copy(char *buf, const int64_t buf_len, ObIKVCacheValue *&value) const {
        value = new (buf) ObTokenResultCacheValue();
        auto *result_value = static_cast<ObTokenResultCacheValue *>(value);
        result_value->count_ = count_;

        int64_t offset = sizeof(*this);

        // 1. 拷贝scores数组
        result_value->scores_ = reinterpret_cast<double*>(buf + offset);
        MEMCPY(buf + offset, scores_, count_ * sizeof(double));
        offset += count_ * sizeof(double);

        // 2. 拷贝ObDatum数组
        ObDatum *new_doc_ids = reinterpret_cast<ObDatum*>(buf + offset);
        MEMCPY(new_doc_ids, doc_ids_, count_ * sizeof(ObDatum));
        result_value->doc_ids_ = new_doc_ids;
        offset += count_ * sizeof(ObDatum);

        // 3. 深拷贝每个Datum的payload（变长字符串内容）
        for (int64_t i = 0; i < count_; ++i) {
            if (!new_doc_ids[i].is_null() && new_doc_ids[i].len_ > 0) {
                offset = upper_align(offset, 8);   // 对齐
                char *payload_ptr = buf + offset;
                MEMCPY(payload_ptr, doc_ids_[i].ptr_, doc_ids_[i].len_);
                new_doc_ids[i].ptr_ = payload_ptr;  // 重定向指针
                offset += doc_ids_[i].len_;
            }
        }
        return OB_SUCCESS;
    }
};

[ObTokenResultCacheValue对象]
[scores_数组: count_个double]
[doc_ids_数组: count_个ObDatum结构体]
[doc_id[0]的字符串内容]   ← 8字节对齐
[doc_id[1]的字符串内容]   ← 8字节对齐
...
[doc_id[n-1]的字符串内容] ← 8字节对齐

// 修改前：默认页面大小（通常4KB或8KB）
ObSRBMWIterImpl() : allocator_("BMWAlloc"), ...

// 修改后：128KB页面，减少malloc次数
ObSRBMWIterImpl() : allocator_("BMWAlloc", 128 * 1024), ...

void ObSRBMWIterImpl::reuse(const bool switch_tablet) {
    while (!top_k_heap_.empty()) {
        top_k_heap_.pop();
    }
    status_ = BMWStatus::MAX_STATUS;
    allocator_.reuse();  // 释放内存给"缓存池"，不实际归还给OS
    ObSRDaaTIterImpl::reuse(switch_tablet);
}

// 在ObSRBMWIterImpl::top_k_search()中的改动思路（伪代码）：
int top_k_search() {
    // 对于 id < 1000 的场景，预分配密集得分数组（约8KB，完全在L1缓存中）
    double scores[1000] = {0.0};

    // TAAT模式：外层循环按 term（dim_iters_ 是各 token 的迭代器）
    for (auto iter : dim_iters_) {
        // 对每个 term，扫描其完整倒排链
        while (iter->has_next()) {
            int64_t doc_id = iter->curr_id();
            if (doc_id < 1000) {
                scores[doc_id] += iter->curr_score();  // 累加到密集数组
            }
            iter->next();
        }
    }

    // 处理完所有 term 后，从密集数组中取 Top-K
    // ...（partial_sort 或堆选取）
}

// 参考resolve_column_index的实现（为UNIQUE列自动建索引）
// 仿照此逻辑，为AUTO_INCREMENT列自动建索引

ObTableSchema &table_schema = create_table_stmt->get_create_table_arg().schema_;
ObColumnSchemaV2 *autoinc_col = table_schema.get_column_schema(autoinc_column_id);

// 设置索引列
obrpc::ObColumnSortItem sort_item;
sort_item.column_name_ = autoinc_col->get_column_name_str();
sort_item.order_type_ = common::ObOrderType::ASC;
sort_item.prefix_len_ = 0;

// 生成索引参数并添加到建表语句中
generate_index_arg(...);
create_table_stmt->get_index_arg_list().push_back(index_arg);

Step 1: 标量索引过滤
  使用idx_base_id索引，找到所有 base_id IN (...) 的rowkey
  使用主键范围 id < 1000，进一步过滤
  结果：满足条件的rowkey集合（例如：doc_ids = {5, 23, 78, 156, ...}）

Step 2: 将rowkey集合转化为BitSet
  分配一个1000位的BitSet（因为id < 1000，最多1000个文档）
  对每个满足条件的id: bitset.set(id)

Step 3: 将BitSet注入BMW算子
  BMW在处理每个pivot doc_id时，查BitSet
  若bitset[doc_id] == 0，跳过（不在候选集中）
  只对BitSet中为1的文档计算BM25

// 提交d1b1ceb9中的HACK实现
// 在BMW的next_pivot()中直接过滤id >= 1000的文档
int ObSRBMWIterImpl::next_pivot(int64_t &pivot_iter_idx) {
    // ...
    // [HACK] 比赛专用：标量过滤下推
    const ObDatum *curr_id_datum = nullptr;
    if (OB_FAIL(get_iter(iter_idx)->get_curr_id(curr_id_datum))) {
        LOG_WARN("failed to get current id", K(ret));
    } else {
        int64_t doc_id = curr_id_datum->get_int();
        if (doc_id >= 1000) {
            ret = OB_ITER_END;  // 直接终止，因为doc_id是有序的
            break;
        }
    }
    // ...
}

// src/sql/das/ob_das_ir_define.h
struct ObDASIRDefine {
    // ... 原有字段 ...
    ObBitmap *scalar_filter_bitmap_;  // 新增：标量过滤BitMap
};

// 在ObSRBMWIterImpl中
if (OB_NOT_NULL(bitmap_) && !bitmap_->test(doc_id)) {
    // 不在候选集中，跳过
    continue;
}

ObIndexMergeNode
├── INTERSECT（AND操作）
│   ├── SCAN（idx_base_id）
│   └── SCAN（idx_id）
└── UNION（OR操作）

// generate_candi_index_merge_trees()
// 为给定的filters生成候选的索引合并树
int ObJoinOrder::generate_candi_index_merge_trees(
    const ObIArray<ObRawExpr *> &filters,
    ObIndexMergeNode *&candi_index_tree)
{
    // 1. 创建INTERSECT根节点
    candi_index_tree->node_type_ = INDEX_MERGE_INTERSECT;

    // 2. 遍历所有filters
    for (int64_t i = 0; i < filters.count(); ++i) {
        // 3. 尝试合并到已有子节点
        // 4. 或生成新的索引合并节点
    }
}

// 优化思路：利用位运算检测ASCII字符批
uint64_t *s_ptr64 = (uint64_t *)s;
while ((uint64_t)(s_end - s_ptr) >= 8) {
    // 检测8字节是否都是ASCII（ASCII字符高位为0）
    if ((*s_ptr64 & 0x8080808080808080ULL) == 0) {
        // 批量处理8个ASCII字符
        // 直接查weight表，无需UTF-8解码
        // weights[0] + code * lengths[0]
    }
    // ...
}

// 使用256位SIMD一次处理256个bit
__m256i *bitmap_ptr = (__m256i *)bitmap;
__m256i *filter_ptr = (__m256i *)filter_bitmap;
for (int i = 0; i < bitmap_size / 32; ++i) {
    // 一次AND 256个位
    bitmap_ptr[i] = _mm256_and_si256(bitmap_ptr[i], filter_ptr[i]);
}

策略A（Pre-Filter，前置过滤）：
  适用：标量条件选择率低（过滤掉大部分文档）
  步骤：先标量过滤→构建候选BitSet→在候选集上做全文检索

策略B（Post-Filter，后置过滤）：
  适用：标量条件选择率高（过滤掉少量文档）
  步骤：先全文检索→再标量过滤

策略C（Index Merge，索引合并）：
  适用：多个标量条件都有索引
  步骤：多索引求交集→与全文检索结果合并

// 对多个token的倒排链扫描并行化
// Token1的倒排链扫描 和 Token2的倒排链扫描 并行执行
// 最后合并结果

tf_norm = tf * (k1+1) / (tf + k1*(...))

维护：
  - 每个token i 的最大分数 Ui（= token i 的最大BM25分数）
  - 当前Top-K堆的最低分数θ
  - 所有token的当前指针（指向当前doc_id）

主循环：
  1. 按当前指针的doc_id对所有token排序（doc_id递增）
  2. 从第一个token开始，累计Ui，直到累计值 > θ
     → 此时的doc_id称为"pivot"
  3. 若所有指针都指向同一doc_id（pivot）：
     → 精确计算该doc_id的BM25分数
     → 若分数 > θ，更新Top-K，θ上升
     → 将所有指针前进
  4. 若指针不一致：
     → 将pivot之前的所有指针直接跳到pivot位置
     → 重新排序

关键性质：WAND保证不遗漏任何实际分数 > θ 的文档

当pivot落在某个块内：
  若该块的BlockMax ≤ θ → 跳过整块
  否则 → 进入该块精确查找

ObKVCacheInst (全局缓存实例)
├── cache_id (唯一标识)
├── priority (优先级，影响LRU淘汰)
└── mem_limit_pct (内存配额百分比)
    │
    └── ObKVCacheStore (缓存存储)
        ├── ObKVCacheWorkingSet (工作集，NUMA感知)
        ├── ObKVCache<K,V> (类型化接口)
        └── ObKVCacheHandle (访问句柄，RAII)

// 1. 定义缓存类（继承ObKVCache）
class ObTokenDFCache : public common::ObKVCache<ObTokenDFCacheKey, ObTokenDFCacheValue> {
public:
    int get_token_df(const ObTokenDFCacheKey &key, ObTokenDFValueHandle &handle);
    int put_token_df(const ObTokenDFCacheKey &key, ObTokenDFCacheValue &value);
};

// 2. 注册缓存
int ObStorageCacheSuite::init() {
    token_df_cache_.init("TokenDFCache", OB_SYS_TENANT_ID, 10 /*priority*/);
}

// 3. 读取缓存
ObTokenDFValueHandle handle;  // RAII句柄，handle销毁时自动释放引用
if (OB_SUCCESS == cache_.get(key, value_ptr, handle.handle_)) {
    // 缓存命中，value_ptr有效直到handle销毁
    use(value_ptr);
}

// 4. 写入缓存（触发deep_copy）
ObTokenDFCacheValue value;
value.init(doc_cnt, max_relevance);
cache_.put(key, value, true /*overwrite*/);

MemTable (内存，可写)
    │
    │ Flush（达到阈值时）
    ▼
L0 SSTable (磁盘，无序)
    │
    │ Compaction
    ▼
L1 SSTable (磁盘，有序)
    │
    │ Compaction
    ▼
...

ALTER SYSTEM MAJOR FREEZE;

SQL解析 → 语义分析 → 逻辑计划 → 查询优化 → 物理计划 → 执行

不下推的代价：
  - 全文检索返回N_ft个文档（可能很多）
  - 回表获取每个文档的标量列
  - Sort + Limit

下推后的代价：
  - 全文检索算子内部维护Top-K堆
  - 只返回K个文档
  - 无需Sort（已有序）

收益 = (N_ft - K) * (IO + BM25计算) 的节省

// 原始代价估算（考虑IO、CPU等）
double fulltext_scan_cost = calc_fulltext_cost(token_cnt, doc_cnt, ...);

// 优化后：强制走全文索引
double fulltext_scan_cost = 0.0;

class ObArenaAllocator {
    // 预分配大块内存（页）
    // 分配时从当前页线性增长
    // 释放整个arena时，一次性释放所有内存
    // 不支持单个对象的释放
};