// Copyright 2026 PingCAP, Inc.

use std::{
    convert::TryInto,
    sync::{Arc, atomic::Ordering},
};

use bytes::Bytes;
use clara_fts::{
    FTS_MATCH_EXPR_METADATA_VERSION,
    test_util::{PlainFtsQueryInfo, make_scored_query, make_unscored_query},
};
use collections::HashMap;
use futures::executor::block_on;
use kvenginepb::fts::{
    FtsFieldLayout, FullTextAnalyzerType, FullTextIndexDef, FullTextIndexFieldDef,
};
use rstest::rstest;
use tidb_query_datatype::{FieldTypeTp, codec::table::encode_row_key};
use tikv_util::{config::ReadableSize, time::Instant};
use tipb::{Executor, KeyRange, TableScan};
use txn_types::{ClusterGcStates, GcState};

use crate::{
    DEL_PREFIXES_KEY, DeletePrefixes, FilePrepareType, IdVer, PrepareOpts,
    apply::ChangeSet,
    compaction::CompactionPriority,
    context::IaCtx,
    dfs,
    dfs::FileType,
    read::{CloudColumnarReader, CloudColumnarReaders, SnapAccess, TableCtx},
    shard::{MajorCompactionType, ShardDataBuilder},
    table::{
        SnapVersion,
        columnar::{ColumnarFile, ColumnarLevels, ColumnarMetaCache, filter::TableScanCtx},
        file::{File, InMemFile},
        fts::{
            ColumnarToFtsL0Opts, EDedicatedFile, FtsCache, FtsDeltaCache, FtsDeltaKind, FtsLevels,
            IntPk, OrderedPkIterator, PackedFile, PkReader, PkType, columnar_to_fts_l0, lp_key,
            new_tracked_index_meta,
            test_util::{
                SchemaBuilder, TestColumnarFileBuilder, TestDedicatedFileBuilder,
                TestPackedFileBuilder, new_block, new_columnar, new_ded, new_packed,
                new_write_batch,
            },
            tracked_index_layout_from_tracked_meta, tracked_index_meta_from_index_def,
        },
        schema_file::{Schema, SchemaFile, build_schema_file},
    },
    tests::{
        ApplyTask, DEF_BLOCK_SIZE, KEYSPACE_ID, Splitter, TestEngine, keyspace_prefix,
        new_test_engine_opt, new_test_engine_opt_with_custom_options, prepare_table_region,
        try_wait,
    },
};

#[test]
fn test_fts_trim_overbound() {
    ::test_util::init_log_for_test();
    let keyspace_id = KEYSPACE_ID;
    let table_id = 30;
    let index_id = 1;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);
    shard
        .active
        .store(true, std::sync::atomic::Ordering::SeqCst);
    assert!(shard.ready_to_compact());

    // Build a schema file so shard metadata looks consistent.
    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    let l0_id = id();
    let l0_file = put_packed(
        &engine,
        l0_id,
        new_packed(l0_id, 500)
            .lp(table_id, index_id, |d| {
                d(8, 220, false, "trim doc");
                d(11, 210, false, "trim doc");
                d(15, 205, true, "trim doc");
                d(20, 150, false, "trim doc");
                d(35, 120, false, "trim doc");
            })
            .lp(table_id + 1, index_id, |d| {
                d(8, 220, false, "trim doc");
                d(11, 210, false, "trim doc");
                d(15, 205, true, "trim doc");
                d(20, 150, false, "trim doc");
                d(35, 120, false, "trim doc");
            }),
    );
    let ded_id = id();
    let l2_file = put_ded(
        &engine,
        ded_id,
        new_ded(ded_id).lp(table_id, index_id, |d| {
            d(5, 300, false, "trim doc");
            d(6, 290, false, "trim doc");
        }),
    );
    let ded_id_1 = id();
    let l2_file_1 = put_ded(
        &engine,
        ded_id_1,
        new_ded(ded_id_1).lp(table_id + 1, index_id, |d| {
            d(5, 300, false, "trim doc");
            d(6, 290, false, "trim doc");
        }),
    );
    let mut levels = FtsLevels::default();
    levels.track_index(table_id, index_id);
    levels.mut_l0(|files| files.push(l0_file));
    levels.insert_l2_file(l2_file);
    levels.insert_l2_file(l2_file_1); // File is out of range and should be removed.
    // TODO: add more cases after suupport split & merge.

    let base_data = shard.get_data();
    let mut builder = ShardDataBuilder::new(base_data);
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(levels);
    shard.set_data(builder.build());
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    let shard = engine.get_shard(shard_id).unwrap();
    shard.pending_ops.write().unwrap().trim_over_bound = true;
    *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::TrimOverBound);
    let id_ver = shard.id_ver();
    engine.trigger_compact(id_ver);

    let ok = try_wait(
        || {
            let data = shard.get_data();
            !data.fts_levels.l0().is_empty() && data.fts_levels.l0()[0].id() != l0_id
        },
        10,
    );
    assert!(ok, "fts trim over bound compaction failed");

    let levels = shard.get_data().fts_levels.clone();
    assert_eq!(levels.l2().len(), 1, "l2 files should be removed");
    let rewritten = &levels.l0()[0];
    assert!(
        rewritten.id() != l0_id,
        "packed file should be rewritten with new id"
    );
}

#[test]
fn test_fts_truncate_ts() {
    ::test_util::init_log_for_test();
    let keyspace_id = KEYSPACE_ID;
    let table_id = 40;
    let index_id = 1;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);
    shard
        .active
        .store(true, std::sync::atomic::Ordering::SeqCst);
    assert!(shard.ready_to_compact());

    // Build a schema file so shard metadata looks consistent.
    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    let l0_id = id();
    let l0_file = put_packed(
        &engine,
        l0_id,
        new_packed(l0_id, 1000).lp(table_id, index_id, |d| {
            d(10, 300, false, "truncate doc");
            d(10, 100, false, "truncate doc");
            d(11, 220, true, "truncate doc");
            d(11, 150, false, "truncate doc");
            d(12, 90, false, "truncate doc");
        }),
    );

    let mut levels = FtsLevels::default();
    levels.track_index(table_id, index_id);
    levels.mut_l0(|files| files.push(l0_file));

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(levels);
    shard.set_data(builder.build());

    shard.pending_ops.write().unwrap().truncate_ts = Some(175);
    *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::TruncateTs);
    let id_ver = shard.id_ver();
    engine.trigger_compact(id_ver);

    let ok = try_wait(
        || {
            let data = shard.get_data();
            !data.fts_levels.l0().is_empty() && data.fts_levels.l0()[0].id() != l0_id
        },
        10,
    );
    assert!(ok, "fts truncate ts compaction failed");

    let rewritten = shard.get_data().fts_levels.l0()[0].clone();
    let handles = collect_packed_handles(&rewritten, table_id);
    assert_eq!(handles, vec![10, 11, 12]);
}

#[test]
fn test_fts_destroy_range() {
    ::test_util::init_log_for_test();
    let keyspace_id = KEYSPACE_ID;
    let table_id = 50;
    let index_id = 1;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    // Build a schema file so shard metadata looks consistent.
    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    let l0_id = id();
    let l0_file = put_packed(
        &engine,
        l0_id,
        new_packed(l0_id, 600).lp(table_id, index_id, |d| {
            d(10, 200, false, "destroy doc");
            d(11, 180, true, "destroy doc");
            d(12, 170, false, "destroy doc");
            d(13, 160, false, "destroy doc");
        }),
    );
    let l2_id = id();
    let l2_file = put_ded(
        &engine,
        l2_id,
        new_ded(l2_id).lp(table_id, index_id, |d| {
            d(20, 200, false, "destroy doc");
            d(21, 180, true, "destroy doc");
            d(22, 170, false, "destroy doc");
            d(23, 160, false, "destroy doc");
        }),
    );
    let mut levels = FtsLevels::default();
    levels.track_index(table_id, index_id);
    levels.mut_l0(|files| files.push(l0_file));
    levels.insert_l2_file(l2_file);
    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(levels);
    shard.set_data(builder.build());

    let mut del_prefixes = DeletePrefixes::new_with_keyspace_id(keyspace_id);
    del_prefixes.merge_prefix_in_place(&outer_row_key(keyspace_id, table_id, 12));
    del_prefixes.merge_prefix_in_place(&outer_row_key(keyspace_id, table_id, 22));
    shard.pending_ops.write().unwrap().del_prefixes = Arc::new(del_prefixes.clone());
    shard.set_property(DEL_PREFIXES_KEY, &del_prefixes.marshal());
    *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::DestroyRange);
    let id_ver = shard.id_ver();
    engine.trigger_compact(id_ver);

    let ok = try_wait(
        || {
            let data = shard.get_data();
            !data.fts_levels.l0().is_empty()
                && data.fts_levels.l0()[0].id() != l0_id
                && data
                    .fts_levels
                    .l2()
                    .iter()
                    .any(|(_, files)| files.iter().any(|f| f.id() != l2_id))
        },
        10,
    );
    assert!(ok, "fts destroy range compaction failed");

    let rewritten_l0 = shard.get_data().fts_levels.l0()[0].clone();
    let handles = collect_packed_handles(&rewritten_l0, table_id);
    assert_eq!(handles, vec![10, 11, 13]);

    let rewritten_l2 = shard
        .get_data()
        .fts_levels
        .l2()
        .iter()
        .next()
        .unwrap()
        .1
        .first()
        .unwrap()
        .clone();
    let handles = collect_dedicated_handles(&rewritten_l2);
    assert_eq!(handles, vec![20, 21, 23]);
}

#[test]
fn test_fts_intra_l2() {
    ::test_util::init_log_for_test();
    let keyspace_id = KEYSPACE_ID;
    let table_id = 60;
    let index_id = 5;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    // Build a schema file so shard metadata looks consistent.
    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    let lp_key_bytes = lp_key(table_id, index_id).to_vec();

    // Prepare multiple dedicated files for the same logical partition.
    let expected_docs = 5;

    let file_id_a = id();
    let file_a = put_ded(
        &engine,
        file_id_a,
        new_ded(file_id_a).lp(table_id, index_id, |d| {
            d(3, 120, false, "doc_3_put");
            d(5, 80, false, "doc_5_put");
        }),
    );

    let file_id_b = id();
    let file_b = put_ded(
        &engine,
        file_id_b,
        new_ded(file_id_b).lp(table_id, index_id, |d| {
            d(3, 110, true, "doc_3_del");
            d(6, 90, false, "doc_6_put");
        }),
    );

    let file_id_c = id();
    let file_c = put_ded(
        &engine,
        file_id_c,
        new_ded(file_id_c).lp(table_id, index_id, |d| {
            d(7, 70, false, "doc_7_put");
        }),
    );

    let mut levels = FtsLevels::default();
    levels.track_index(table_id, index_id);
    levels.insert_l2_file(file_a);
    levels.insert_l2_file(file_b);
    levels.insert_l2_file(file_c);

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(levels);
    shard.set_data(builder.build());

    let mut expected_ids = vec![file_id_a, file_id_b, file_id_c];
    expected_ids.sort_unstable();
    let priority = shard
        .test_get_fts_l2_compact_priority()
        .expect("l2 compaction priority must exist");
    let source_ids = match &priority {
        CompactionPriority::FtsCompactL2 { lp_key, file_ids } => {
            assert_eq!(lp_key.as_slice(), lp_key_bytes.as_slice());
            let mut picked = file_ids.clone();
            picked.sort_unstable();
            assert_eq!(picked, expected_ids);
            file_ids.clone()
        }
        other => panic!("unexpected priority {:?}", other),
    };
    *shard.compaction_priority.write().unwrap() = Some(priority);

    let id_ver = shard.id_ver();
    engine.trigger_compact(id_ver);

    let merged_ready = try_wait(
        || {
            let data = shard.get_data();
            let Some(files) = data.fts_levels.l2().get(&lp_key_bytes) else {
                return false;
            };
            if files.len() != 1 {
                return false;
            }
            let new_id = files[0].id();
            !source_ids.contains(&new_id) && files[0].props().get_pk_total() == expected_docs
        },
        20,
    );
    assert!(
        merged_ready,
        "fts intra-l2 compaction should output a single merged file"
    );

    let data = shard.get_data();
    let merged_file = data
        .fts_levels
        .l2()
        .get(&lp_key_bytes)
        .and_then(|files| files.first())
        .cloned()
        .expect("merged file should exist");

    let rows = collect_dedicated_rows(&merged_file);
    assert_eq!(
        rows,
        vec![
            (3, 120, false),
            (3, 110, true),
            (5, 80, false),
            (6, 90, false),
            (7, 70, false),
        ]
    );

    let doc3_hits = search_l2(&merged_file, "doc_3_put");
    assert!(
        doc3_hits
            .iter()
            .all(|doc_id| *doc_id < expected_docs as u32),
        "merged L2 should contain searchable docs from both source files"
    );

    let missing_hits = search_l2(&merged_file, "nonexistent");
    assert!(missing_hits.is_empty(), "unrelated terms should not match");
}

#[test]
fn test_fts_full_l2_compact_drops_tombstone_below_safe_ts() {
    ::test_util::init_log_for_test();
    let table_id = 61;
    let index_id = 5;
    let safe_ts = 200;

    let (engine, _apply_tx) =
        new_test_engine_opt_with_custom_options(true, DEF_BLOCK_SIZE, "", |opts| {
            // Make the full-L2 trigger easy to hit for unit tests.
            opts.fts_build_options.l2_full_compact_dead_ratio = 0.1;
            opts.fts_build_options.l2_full_compact_dead_bytes = ReadableSize(1);
        });
    let now = Instant::now();
    let mut cluster_gc_states = ClusterGcStates::new(HashMap::default(), now);
    cluster_gc_states.keyspace_gc_states.insert(
        KEYSPACE_ID,
        GcState::new(
            KEYSPACE_ID,
            true,
            safe_ts.into(),
            safe_ts.into(),
            vec![],
            now,
        ),
    );
    engine.update_cluster_gc_states_cache(cluster_gc_states);

    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard = engine.get_shard(1).unwrap();
    assert!(shard.ready_to_compact());

    // Build a schema file so shard metadata looks consistent.
    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    let lp_key_bytes = lp_key(table_id, index_id).to_vec();

    // Prepare two dedicated files for the same LP. The newer one contains a
    // tombstone below `safe_ts` that can be dropped in a *full* merge.
    let old_id = id();
    let old_file = put_ded(
        &engine,
        old_id,
        new_ded(old_id).lp(table_id, index_id, |d| {
            d(1, 120, false, "alpha");
            d(2, 140, false, "beta");
        }),
    );

    let tombstone_id = id();
    let tombstone_file = put_ded(
        &engine,
        tombstone_id,
        new_ded(tombstone_id).lp(table_id, index_id, |d| {
            d(1, 150, true, "alpha");
            d(3, 160, false, "gamma");
        }),
    );

    let mut levels = FtsLevels::default();
    levels.track_index(table_id, index_id);
    levels.insert_l2_file(old_file);
    levels.insert_l2_file(tombstone_file);

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(levels);
    shard.set_data(builder.build());

    engine.refresh_shard_states(&shard);
    match shard.get_compaction_priority() {
        Some(CompactionPriority::FtsFullCompactL2 { lp_key, .. }) => {
            assert_eq!(lp_key.as_slice(), lp_key_bytes.as_slice());
        }
        other => panic!("unexpected priority {:?}", other),
    }
    engine.trigger_compact(shard.id_ver());
    engine
        .get_shard_with_ver(shard.id, shard.ver)
        .expect("shard must still match id/ver after refresh");

    let merged_ready = try_wait(
        || {
            let data = shard.get_data();
            let Some(files) = data.fts_levels.l2().get(&lp_key_bytes) else {
                return false;
            };
            if files.len() != 1 {
                return false;
            }
            let new_file = &files[0];
            if new_file.id() == old_id || new_file.id() == tombstone_id {
                return false;
            }
            collect_dedicated_handles(new_file) == vec![2, 3]
        },
        20,
    );
    assert!(
        merged_ready,
        "fts full L2 compaction should drop tombstones below safe_ts"
    );
}

#[test]
fn test_fts_create_l0() {
    ::test_util::init_log_for_test();
    let keyspace_id = KEYSPACE_ID;
    let table_id = 80;
    let index_id = 6;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    let snap_version = 500;
    let columnar_file_id = id();
    let columnar_file = put_columnar(
        &engine,
        new_columnar(columnar_file_id, snap_version).table::<IntPk>(&schema, |row| {
            row(1, 100, false, |datum| {
                datum(1);
                datum("Who knew the people would adore me so much?");
            });
            row(2, 200, false, |datum| {
                datum(2);
                datum("Being too popular can be such a hassle");
            });
            row(3, 300, false, |datum| {
                datum(3);
                datum("The world is but a stage.");
            });
        }),
    );
    let mut col_levels = ColumnarLevels::new();
    col_levels.add_file(0, columnar_file);

    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index(table_id, index_id);

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_levels(col_levels);
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    let priority = match shard
        .test_get_fts_incremental_update_priority()
        .expect("incremental FTS priority should exist")
    {
        CompactionPriority::FtsCreateL0 {
            col_file_ids,
            active_tracked_indexes,
            snap_version: ver,
        } => {
            assert_eq!(col_file_ids, vec![columnar_file_id]);
            assert_eq!(active_tracked_indexes, vec![(table_id, index_id)]);
            assert_eq!(ver, snap_version.into());
            CompactionPriority::FtsCreateL0 {
                col_file_ids,
                active_tracked_indexes,
                snap_version: ver,
            }
        }
        other => panic!("unexpected priority {:?}", other),
    };
    *shard.compaction_priority.write().unwrap() = Some(priority);
    let id_ver = shard.id_ver();
    engine.trigger_compact(id_ver);

    let ok = try_wait(|| !shard.get_data().fts_levels.l0().is_empty(), 20);
    assert!(ok, "fts create L0 compaction failed");

    let data = shard.get_data();
    assert_eq!(data.fts_levels.l1().len(), 0);
    assert_eq!(data.fts_levels.l0().len(), 1);
    assert_eq!(
        data.fts_levels.l0_snap_version,
        SnapVersion::from(snap_version)
    );

    let l0_file = data.fts_levels.l0()[0].clone();
    let mut hits = search_packed(&l0_file, table_id, index_id, "the");
    hits.sort_unstable();
    assert_eq!(hits, vec![0, 2], "unexpected doc hits for 'the'");

    let none_hits = search_packed(&l0_file, table_id, index_id, "nonexistent");
    assert!(none_hits.is_empty(), "unexpected hits for missing term");
}

#[test]
fn test_fts_l0_compaction() {
    ::test_util::init_log_for_test();
    let keyspace_id = KEYSPACE_ID;
    let table_id = 90;
    let index_id = 7;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    let batches: Vec<Vec<&str>> = vec![
        vec!["alpha apple", "alpha banana"],
        vec!["beta cherry", "beta date"],
        vec!["gamma elderberry", "gamma fig"],
        vec!["delta grape", "delta honey"],
    ];
    let mut l0_files = Vec::new();
    let mut expected_ids = Vec::new();
    for (idx, docs) in batches.into_iter().enumerate() {
        let file_id = id();
        let file = put_packed(
            &engine,
            file_id,
            new_packed(file_id, 1000 + idx as u64).lp(table_id, index_id, |d| {
                for (doc_idx, body) in docs.into_iter().enumerate() {
                    d((idx * 100 + doc_idx) as i64, 1000, false, body);
                }
            }),
        );
        expected_ids.push(file_id);
        l0_files.push(file);
    }

    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index(table_id, index_id);
    fts_levels.mut_l0(|files| files.extend(l0_files));

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::FtsCompactL0);
    engine.trigger_compact(shard.id_ver());

    let ok = try_wait(
        || {
            let data = shard.get_data();
            data.fts_levels.l0().is_empty() && data.fts_levels.l1().len() == 1
        },
        30,
    );
    assert!(ok, "fts l0->l1 compaction failed");

    let data = shard.get_data();
    assert!(
        data.fts_levels.l0().is_empty(),
        "fts l0 compaction should clear all L0 files"
    );
    assert_eq!(
        data.fts_levels.l1().len(),
        1,
        "fts l0 compaction should output exactly one L1 file"
    );
    let l1_file = data.fts_levels.l1()[0].clone();
    assert!(
        !expected_ids.contains(&l1_file.id()),
        "compaction should output new file id"
    );

    let search = |term: &str| {
        let mut hits = search_packed(&l1_file, table_id, index_id, term);
        hits.sort_unstable();
        hits
    };

    assert_eq!(search("alpha").len(), 2, "alpha should hit two docs");
    assert_eq!(search("beta").len(), 2, "beta should hit two docs");
    assert_eq!(search("banana").len(), 1, "banana should hit once");
    assert_eq!(search("cherry").len(), 1, "cherry should hit once");
    assert!(
        search("nonexistent").is_empty(),
        "unexpected hits for missing term"
    );
}

#[test]
fn test_fts_l0_compaction_skips_untracked_indexes() {
    ::test_util::init_log_for_test();
    let keyspace_id = KEYSPACE_ID;
    let table_id = 92;
    let tracked_index = 11;
    let untracked_index = 12;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: tracked_index,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .fts_index(FullTextIndexDef {
            index_id: untracked_index,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    let file_id = id();
    let file = put_packed(
        &engine,
        file_id,
        new_packed(file_id, 600)
            .lp(table_id, tracked_index, |d| {
                d(1, 500, false, "tracked alpha");
                d(2, 500, false, "tracked beta");
            })
            .lp(table_id, untracked_index, |d| {
                d(100, 500, false, "untracked apple");
                d(200, 500, false, "untracked grape");
            }),
    );
    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index(table_id, tracked_index);
    fts_levels.mut_l0(|files| files.push(file));

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::FtsCompactL0);
    engine.trigger_compact(shard.id_ver());

    let ok = try_wait(
        || {
            let data = shard.get_data();
            data.fts_levels.l0().is_empty() && data.fts_levels.l1().len() == 1
        },
        30,
    );
    assert!(ok, "fts l0 compaction failed");

    let data = shard.get_data();
    let l1_file = data.fts_levels.l1()[0].clone();
    let mut tracked_hits = search_packed(&l1_file, table_id, tracked_index, "tracked");
    tracked_hits.sort_unstable();
    assert_eq!(tracked_hits.len(), 2, "tracked docs should remain");

    let untracked_hits = search_packed(&l1_file, table_id, untracked_index, "untracked");
    assert!(
        untracked_hits.is_empty(),
        "untracked index should be dropped from L1 output"
    );

    // Ensure no remaining L0 files
    assert!(data.fts_levels.l0().is_empty());
}

#[test]
fn test_fts_schema_drop_makes_l0_compaction_drop_data() {
    ::test_util::init_log_for_test();
    let keyspace_id = KEYSPACE_ID;
    let table_id = 93;
    let kept_index = 16;
    let dropped_index = 17;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let schema_with_two = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: kept_index,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .fts_index(FullTextIndexDef {
            index_id: dropped_index,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file_with_two = put_schema_file(&engine, id(), &[&schema_with_two]);

    let l0_id = id();
    let l0 = put_packed(
        &engine,
        l0_id,
        new_packed(l0_id, 600)
            .lp(table_id, kept_index, |d| {
                d(1, 500, false, "kept alpha");
                d(2, 500, false, "kept beta");
            })
            .lp(table_id, dropped_index, |d| {
                d(100, 500, false, "dropped apple");
                d(200, 500, false, "dropped grape");
            }),
    );

    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index(table_id, kept_index);
    fts_levels.track_index(table_id, dropped_index);
    fts_levels.mut_l0(|files| files.push(l0));

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(
        schema_file_with_two.get_version(),
        0,
        Some(schema_file_with_two),
    );
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    // Drop one index from schema; it should become untracked and thus be dropped by
    // compaction.
    let schema_with_one = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: kept_index,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file_with_one = put_schema_file(&engine, id(), &[&schema_with_one]);

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(
        schema_file_with_one.get_version(),
        0,
        Some(schema_file_with_one),
    );
    shard.set_data(builder.build());

    assert!(
        shard
            .get_data()
            .fts_levels
            .has_tracked_index(table_id, kept_index),
        "kept index should remain tracked"
    );
    assert!(
        shard
            .get_data()
            .fts_levels
            .has_tracked_index(table_id, dropped_index),
        "dropped index is still tracked until FtsDropIndex is applied"
    );

    shard.refresh_states();
    let priority = shard
        .get_compaction_priority()
        .expect("fts drop index priority expected");
    match &priority {
        CompactionPriority::FtsDropIndex { dropped_indexes } => {
            assert_eq!(dropped_indexes, &vec![(table_id, dropped_index)]);
        }
        other => panic!("unexpected priority {:?}", other),
    }
    engine.trigger_compact(shard.id_ver());

    let ok = try_wait(
        || {
            let data = shard.get_data();
            data.fts_levels.has_tracked_index(table_id, kept_index)
                && !data.fts_levels.has_tracked_index(table_id, dropped_index)
        },
        20,
    );
    assert!(ok, "fts drop index did not finish in time");

    let ok = try_wait(|| engine.is_shard_compacting(shard_id) == Some(false), 20);
    assert!(ok, "fts drop-index compaction did not become idle in time");

    let ok = try_wait(
        || {
            let data = shard.get_data();
            if data.fts_levels.l0().is_empty() && data.fts_levels.l1().len() == 1 {
                return true;
            }
            drop(data);

            if engine.is_shard_compacting(shard_id) == Some(false) {
                *shard.compaction_priority.write().unwrap() =
                    Some(CompactionPriority::FtsCompactL0);
                engine.trigger_compact(shard.id_ver());
            }
            false
        },
        30,
    );
    assert!(ok, "fts l0 compaction failed");

    let l1_file = shard.get_data().fts_levels.l1()[0].clone();
    let kept_hits = search_packed(&l1_file, table_id, kept_index, "kept");
    assert!(!kept_hits.is_empty(), "kept index data should remain");

    let dropped_hits = search_packed(&l1_file, table_id, dropped_index, "dropped");
    assert!(
        dropped_hits.is_empty(),
        "dropped index data should be removed by compaction"
    );
}

#[test]
fn test_fts_cleanup_compaction_removes_orphans() {
    ::test_util::init_log_for_test();
    let keyspace_id = KEYSPACE_ID;
    let table_id = 95;
    let tracked_index = 13;
    let orphan_index = 14;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: tracked_index,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    let tracked_l1_id = id();
    let tracked_l1 = put_packed(
        &engine,
        tracked_l1_id,
        new_packed(tracked_l1_id, 0).lp(table_id, tracked_index, |d| {
            d(1, 800, false, "tracked apple");
            d(2, 800, false, "tracked berry");
        }),
    );
    let orphan_l1_id = id();
    let orphan_l1 = put_packed(
        &engine,
        orphan_l1_id,
        new_packed(orphan_l1_id, 0).lp(table_id, orphan_index, |d| {
            d(100, 800, false, "orphan citrus");
            d(200, 800, false, "orphan date");
        }),
    );
    let tracked_l2_id = id();
    let tracked_l2 = put_ded(
        &engine,
        tracked_l2_id,
        new_ded(tracked_l2_id).lp(table_id, tracked_index, |d| {
            d(1, 800, false, "tracked apple");
            d(2, 800, false, "tracked berry");
        }),
    );
    let orphan_l2_id = id();
    let orphan_l2 = put_ded(
        &engine,
        orphan_l2_id,
        new_ded(orphan_l2_id).lp(table_id, orphan_index, |d| {
            d(100, 800, false, "orphan citrus");
            d(200, 800, false, "orphan date");
        }),
    );

    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index(table_id, tracked_index);
    fts_levels.mut_l1(|files| {
        files.push(tracked_l1);
        files.push(orphan_l1);
    });
    fts_levels.insert_l2_file(tracked_l2);
    fts_levels.insert_l2_file(orphan_l2);

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    let priority = shard
        .test_get_fts_cleanup_priority()
        .expect("cleanup priority expected");
    let (cleanup_l1, cleanup_l2) = match &priority {
        CompactionPriority::FtsCleanup {
            l1_file_ids,
            l2_file_ids,
        } => (l1_file_ids.clone(), l2_file_ids.clone()),
        _ => panic!("unexpected priority {:?}", priority),
    };
    assert!(
        cleanup_l1.contains(&orphan_l1_id),
        "orphan L1 file should be scheduled for cleanup"
    );
    assert!(
        cleanup_l2.contains(&orphan_l2_id),
        "orphan L2 file should be scheduled for cleanup"
    );
    assert!(
        !cleanup_l1.contains(&tracked_l1_id),
        "tracked L1 file should not be removed"
    );

    *shard.compaction_priority.write().unwrap() = Some(priority);
    engine.trigger_compact(shard.id_ver());

    let ok = try_wait(
        || {
            let data = shard.get_data();
            let l1_ids: Vec<u64> = data.fts_levels.l1().iter().map(|f| f.id()).collect();
            let l2_ids: Vec<u64> = data
                .fts_levels
                .l2()
                .values()
                .flat_map(|files| files.iter().map(|f| f.id()))
                .collect();
            l1_ids == vec![tracked_l1_id] && l2_ids == vec![tracked_l2_id]
        },
        20,
    );
    assert!(ok, "fts cleanup compaction did not finish in time");

    let data = shard.get_data();
    assert_eq!(
        data.fts_levels.l1().len(),
        1,
        "cleanup should keep exactly one L1 file"
    );
    let remaining_l1 = data.fts_levels.l1()[0].clone();
    let mut hits = search_packed(&remaining_l1, table_id, tracked_index, "tracked");
    hits.sort_unstable();
    assert_eq!(hits.len(), 2, "tracked index data should remain searchable");
    let l2_ids: Vec<u64> = data
        .fts_levels
        .l2()
        .values()
        .flat_map(|files| files.iter().map(|f| f.id()))
        .collect();
    assert_eq!(
        l2_ids,
        vec![tracked_l2_id],
        "cleanup should retain only tracked L2 file"
    );
}

#[test]
fn test_fts_schema_drop_untracks_and_cleans_files() {
    ::test_util::init_log_for_test();
    let keyspace_id = KEYSPACE_ID;
    let table_id = 96;
    let index_id = 15;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let schema_with_fts = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file_with_fts = put_schema_file(&engine, id(), &[&schema_with_fts]);

    let l1_id = id();
    let l1 = put_packed(
        &engine,
        l1_id,
        new_packed(l1_id, 0).lp(table_id, index_id, |d| {
            d(1, 800, false, "tracked apple");
            d(2, 800, false, "tracked berry");
        }),
    );
    let l2_id = id();
    let l2 = put_ded(
        &engine,
        l2_id,
        new_ded(l2_id).lp(table_id, index_id, |d| {
            d(1, 800, false, "tracked apple");
            d(2, 800, false, "tracked berry");
        }),
    );

    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index(table_id, index_id);
    fts_levels.mut_l1(|files| files.push(l1));
    fts_levels.insert_l2_file(l2);

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(
        schema_file_with_fts.get_version(),
        0,
        Some(schema_file_with_fts),
    );
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    // Drop the fulltext index from schema.
    let schema_without_fts = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .schema();
    let schema_file_without_fts = put_schema_file(&engine, id(), &[&schema_without_fts]);

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(
        schema_file_without_fts.get_version(),
        0,
        Some(schema_file_without_fts),
    );
    shard.set_data(builder.build());

    assert!(
        shard
            .get_data()
            .fts_levels
            .has_tracked_index(table_id, index_id),
        "tracked index is still tracked until FtsDropIndex is applied"
    );

    shard.refresh_states();
    let priority = shard
        .get_compaction_priority()
        .expect("fts drop index priority expected");
    match &priority {
        CompactionPriority::FtsDropIndex { dropped_indexes } => {
            assert_eq!(dropped_indexes, &vec![(table_id, index_id)]);
        }
        other => panic!("unexpected priority {:?}", other),
    }
    engine.trigger_compact(shard.id_ver());

    let ok = try_wait(
        || {
            let data = shard.get_data();
            !data.fts_levels.has_tracked_index(table_id, index_id)
        },
        20,
    );
    assert!(ok, "fts drop index did not finish in time");

    if let Some(priority) = shard.test_get_fts_cleanup_priority() {
        let (cleanup_l1, cleanup_l2) = match &priority {
            CompactionPriority::FtsCleanup {
                l1_file_ids,
                l2_file_ids,
            } => (l1_file_ids.clone(), l2_file_ids.clone()),
            _ => panic!("unexpected priority {:?}", priority),
        };
        assert!(
            cleanup_l1.contains(&l1_id),
            "L1 file for dropped index should be scheduled for cleanup"
        );
        assert!(
            cleanup_l2.contains(&l2_id),
            "L2 file for dropped index should be scheduled for cleanup"
        );
    }

    shard.refresh_states();
    engine.trigger_compact(shard.id_ver());

    let ok = try_wait(
        || {
            let data = shard.get_data();
            data.fts_levels.l1().is_empty() && data.fts_levels.l2().is_empty()
        },
        20,
    );
    assert!(ok, "fts cleanup after schema drop did not finish in time");
}

#[test]
fn test_fts_add_index() {
    ::test_util::init_log_for_test();
    let keyspace_id = KEYSPACE_ID;
    let table_id = 91;
    let index_id = 8;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file_id = engine.id_allocator.alloc_id(1).unwrap()[0];
    let schema_file = put_schema_file(&engine, schema_file_id, &[&schema]);

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(FtsLevels::default());
    shard.set_data(builder.build());
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    let priority = shard
        .test_get_fts_add_index_priority()
        .expect("fts add index priority should exist");
    match &priority {
        CompactionPriority::FtsAddIndex { new_indexes } => {
            assert_eq!(new_indexes, &vec![(table_id, index_id)]);
        }
        other => panic!("unexpected priority {:?}", other),
    }

    // Simulate compaction: track the index and ensure the state reflects it.
    *shard.compaction_priority.write().unwrap() = Some(priority);
    engine.trigger_compact(shard.id_ver());

    let ok = try_wait(
        || {
            shard
                .get_data()
                .fts_levels
                .has_tracked_index(table_id, index_id)
        },
        20,
    );
    assert!(ok, "fts add index should track the new index");
    assert!(
        shard.test_get_fts_add_index_priority().is_none(),
        "priority should disappear after index is tracked"
    );
}

#[test]
fn test_fts_l0_compaction_promote_l2() {
    ::test_util::init_log_for_test();
    let keyspace_id = KEYSPACE_ID;
    let table_id = 95;
    let index_id = 9;
    let (engine, apply_tx) =
        new_test_engine_opt_with_custom_options(true, DEF_BLOCK_SIZE, "", |opts| {
            opts.fts_build_options.max_l0_files = 1;
            // Ensure L1->L2 promotion relies on LP serialized size (regression coverage for
            // EPackedFileLp::serialized_size).
            opts.fts_build_options.min_l2_lp_size = ReadableSize(100);
        });
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    let l0_id = id();
    let l0_file = put_packed(
        &engine,
        l0_id,
        new_packed(l0_id, 2000).lp(table_id, index_id, |d| {
            for i in 0..16 {
                d(i, 2000, false, format!("promote payload {i}"));
            }
        }),
    );
    let l0_lp_size = block_on(async {
        let mut iter = l0_file.lp_iter().unwrap();
        iter.next_lp().await.unwrap().unwrap().serialized_size()
    });
    assert!(
        l0_lp_size >= 100,
        "test setup: expected LP size >= 100, got {}",
        l0_lp_size
    );

    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index(table_id, index_id);
    fts_levels.mut_l0(|files| files.push(l0_file));

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file.clone()));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::FtsCompactL0);
    engine.trigger_compact(shard.id_ver());

    let ok = try_wait(
        || {
            let data = shard.get_data();
            data.fts_levels.l0().is_empty()
                && (!data.fts_levels.l1().is_empty()
                    || data.fts_levels.l2().values().any(|files| !files.is_empty()))
        },
        30,
    );
    assert!(ok, "fts L0 compaction did not finish");

    let mut current_data = shard.get_data();
    assert!(
        current_data.fts_levels.l1().is_empty(),
        "expected L0 compaction to promote to L2, but got L1 files"
    );
    let initial_l2_files: Vec<EDedicatedFile> = current_data
        .fts_levels
        .l2()
        .values()
        .flat_map(|files| files.iter().cloned())
        .collect();
    assert_eq!(initial_l2_files.len(), 1);
    let promote_hits: usize = initial_l2_files
        .iter()
        .map(|file| search_l2(file, "promote").len())
        .sum();
    assert_eq!(promote_hits, 16, "promote docs should move to L2");

    let follow_id = id();
    let follow_file = put_packed(
        &engine,
        follow_id,
        new_packed(follow_id, 3000).lp(table_id, index_id, |d| {
            for i in 0..8 {
                d(
                    1_000 + i as i64,
                    3000,
                    false,
                    format!("followup payload {i}"),
                );
            }
        }),
    );

    let mut levels = (*current_data.fts_levels).clone();
    levels.mut_l0(|files| files.push(follow_file));
    let mut builder = ShardDataBuilder::new(current_data);
    builder.set_fts_levels(levels);
    shard.set_data(builder.build());

    *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::FtsCompactL0);
    engine.trigger_compact(shard.id_ver());

    let ok = try_wait(
        || {
            let data = shard.get_data();
            data.fts_levels.l0().is_empty()
                && (!data.fts_levels.l1().is_empty()
                    || data.fts_levels.l2().values().any(|files| files.len() >= 2))
        },
        30,
    );
    assert!(ok, "fts follow-up compaction failed");

    current_data = shard.get_data();
    assert!(current_data.fts_levels.l1().is_empty());

    let all_l2_files: Vec<EDedicatedFile> = current_data
        .fts_levels
        .l2()
        .values()
        .flat_map(|files| files.iter().cloned())
        .collect();
    assert!(
        all_l2_files.len() >= 2,
        "expected multiple L2 files after follow-up"
    );
    for prev in &initial_l2_files {
        assert!(
            all_l2_files.iter().any(|file| file.id() == prev.id()),
            "original promoted file should remain"
        );
    }

    let search_all = |term: &str| -> usize {
        all_l2_files
            .iter()
            .map(|file| search_l2(file, term).len())
            .sum()
    };

    assert_eq!(search_all("promote"), 16, "promote docs should persist");
    assert_eq!(
        search_all("followup"),
        8,
        "followup docs should be searchable"
    );
}

#[test]
fn test_fts_l0_compaction_no_promotion_below_l2_threshold() {
    ::test_util::init_log_for_test();
    let keyspace_id = KEYSPACE_ID;
    let table_id = 96;
    let index_id = 10;
    let min_l2_lp_size = ReadableSize::mb(64);
    let (engine, apply_tx) =
        new_test_engine_opt_with_custom_options(true, DEF_BLOCK_SIZE, "", |opts| {
            opts.fts_build_options.max_l0_files = 1;
            opts.fts_build_options.min_l2_lp_size = min_l2_lp_size;
        });
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    let l0_id = id();
    let l0_file = put_packed(
        &engine,
        l0_id,
        new_packed(l0_id, 4000).lp(table_id, index_id, |d| {
            for i in 0..2 {
                d(i, 4000, false, format!("stayl1 payload {i}"));
            }
        }),
    );
    let l0_lp_size = block_on(async {
        let mut iter = l0_file.lp_iter().unwrap();
        iter.next_lp().await.unwrap().unwrap().serialized_size()
    });
    let min_l2_lp_size_usize: usize = min_l2_lp_size.0.try_into().unwrap();
    assert!(
        l0_lp_size < min_l2_lp_size_usize,
        "test setup: expected LP size < {}, got {l0_lp_size}",
        min_l2_lp_size.0
    );

    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index(table_id, index_id);
    fts_levels.mut_l0(|files| files.push(l0_file));

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file.clone()));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::FtsCompactL0);
    engine.trigger_compact(shard.id_ver());

    let ok = try_wait(
        || {
            let data = shard.get_data();
            data.fts_levels.l0().is_empty()
                && (!data.fts_levels.l1().is_empty()
                    || data.fts_levels.l2().values().any(|files| !files.is_empty()))
        },
        30,
    );
    assert!(ok, "fts L0 compaction did not finish");

    let data = shard.get_data();
    assert!(
        data.fts_levels.l2().values().all(|files| files.is_empty()),
        "expected no L2 promotion when below threshold"
    );
    assert!(
        !data.fts_levels.l1().is_empty(),
        "expected L1 outputs when not promoted to L2"
    );
    let stayl1_hits: usize = data
        .fts_levels
        .l1()
        .iter()
        .map(|file| search_packed(file, table_id, index_id, "stayl1").len())
        .sum();
    assert_eq!(stayl1_hits, 2, "stayl1 docs should be searchable in L1");
}

#[test]
fn test_fts_split_merge() {
    ::test_util::init_log_for_test();

    let keyspace_id = KEYSPACE_ID;
    let table_id = 60;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    // Build a schema file so shard metadata looks consistent.
    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id: 1,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    // Create packed file with entries spanning a range (handles 10-90)
    // L0 file covering low range (handles 10-40)
    let l0_id_low = id();
    let l0_file_low = put_packed(
        &engine,
        l0_id_low,
        new_packed(l0_id_low, 500).lp(table_id, 1, |d| {
            d(10, 200, false, "split doc");
            d(20, 190, false, "split doc");
            d(30, 180, false, "split doc");
            d(40, 170, false, "split doc");
        }),
    );
    let l0_id_overlap = id();
    let l0_file_overlap = put_packed(
        &engine,
        l0_id_overlap,
        new_packed(l0_id_overlap, 500).lp(table_id, 1, |d| {
            d(35, 280, false, "split doc");
            d(60, 270, false, "split doc");
        }),
    );
    // L0 file covering high range (handles 60-90)
    let l0_id_high = id();
    let l0_file_high = put_packed(
        &engine,
        l0_id_high,
        new_packed(l0_id_high, 500).lp(table_id, 1, |d| {
            d(60, 160, false, "split doc");
            d(70, 150, false, "split doc");
            d(80, 140, false, "split doc");
            d(90, 130, false, "split doc");
        }),
    );

    // Create L2 dedicated files with different key ranges
    let l2_id_low = id();
    let l2_file_low = put_ded(
        &engine,
        l2_id_low,
        new_ded(l2_id_low).lp(table_id, 1, |d| {
            d(15, 300, false, "split doc");
            d(25, 290, false, "split doc");
        }),
    );
    let l2_id_overlap = id();
    let l2_file_overlap = put_ded(
        &engine,
        l2_id_overlap,
        new_ded(l2_id_overlap).lp(table_id, 1, |d| {
            d(35, 280, false, "split doc");
            d(60, 270, false, "split doc");
        }),
    );
    let l2_id_high = id();
    let l2_file_high = put_ded(
        &engine,
        l2_id_high,
        new_ded(l2_id_high).lp(table_id, 1, |d| {
            d(65, 280, false, "split doc");
            d(75, 270, false, "split doc");
        }),
    );

    let mut levels = FtsLevels::default();
    levels.track_index(table_id, 1);
    levels.l0_snap_version = SnapVersion::from(123);
    levels.mut_l0(|files| {
        files.push(l0_file_low);
        files.push(l0_file_overlap);
        files.push(l0_file_high);
    });
    levels.insert_l2_file(l2_file_low);
    levels.insert_l2_file(l2_file_overlap);
    levels.insert_l2_file(l2_file_high);

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(levels.clone());
    shard.set_data(builder.build());
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    // Verify initial state
    let data_before = shard.get_data();
    assert_eq!(
        data_before.fts_levels.l0().len(),
        3,
        "should have 3 L0 files before split"
    );
    assert_eq!(
        data_before
            .fts_levels
            .l2()
            .values()
            .map(|v| v.len())
            .sum::<usize>(),
        3,
        "should have 3 L2 files before split"
    );
    let l0_snap_version_before = data_before.fts_levels.l0_snap_version;

    // Split shard at handle 50 (middle of range), similar to
    // test_columnar_trim_over_bound
    let split_key = [keyspace_prefix(keyspace_id), encode_row_key(table_id, 50)].concat();
    let shard_ver = shard.ver;
    let mut splitter = Splitter::new(
        vec![split_key],
        IdVer::new(shard_id, shard_ver),
        100,
        apply_tx,
    );
    let handle = std::thread::spawn(move || {
        splitter.run();
    });

    // Wait for split to complete (expect 6 shards now: initial 5 + 1 new)
    let ok = try_wait(|| engine.shards.len() == 6, 5);
    assert!(
        ok,
        "fts split failed: expected 6 shards, got {}",
        engine.shards.len()
    );
    handle.join().unwrap();

    // The original shard (shard_id) should now cover higher range (>=50)
    // New shard (101) should cover lower range (<50)
    let new_shard_id = 101;
    let shard_high = engine.get_shard(shard_id).unwrap();
    let shard_low = engine.get_shard(new_shard_id).unwrap();

    // Check FTS levels in the high range shard
    let levels_high = &shard_high.get_data().fts_levels;
    assert_eq!(
        levels_high.l0().len(),
        2,
        "high range shard should have 2 L0 files"
    );
    // High range shard should only have the L0 file that overlaps with high range
    assert!(
        levels_high.l0().iter().any(|f| f.id() == l0_id_high),
        "high range shard should have high L0 file"
    );
    assert!(
        levels_high.l0().iter().any(|f| f.id() == l0_id_overlap),
        "high range shard should have overlap L0 file"
    );

    // Check FTS levels in the low range shard
    let levels_low = &shard_low.get_data().fts_levels;
    assert_eq!(
        levels_low.l0().len(),
        2,
        "low range shard should have 2 L0 files"
    );
    // Low range shard should only have the L0 file that overlaps with low range
    assert!(
        levels_low.l0().iter().any(|f| f.id() == l0_id_low),
        "low range shard should have low L0 file"
    );
    assert!(
        levels_low.l0().iter().any(|f| f.id() == l0_id_overlap),
        "low range shard should have overlap L0 file"
    );
    // Verify L2 files are correctly distributed
    assert_eq!(
        levels_high.l2().values().map(|v| v.len()).sum::<usize>(),
        2,
        "high range shard should have 2 L2 files"
    );
    assert_eq!(
        levels_low.l2().values().map(|v| v.len()).sum::<usize>(),
        2,
        "low range shard should have 2 L2 files"
    );
    assert!(
        levels_high
            .l2()
            .values()
            .any(|files| files.iter().any(|f| f.id() == l2_id_high)),
        "high range shard should have high L2 file"
    );
    assert!(
        levels_low
            .l2()
            .values()
            .any(|files| files.iter().any(|f| f.id() == l2_id_low)),
        "low range shard should have low L2 file"
    );
    assert!(
        levels_high
            .l2()
            .values()
            .any(|files| files.iter().any(|f| f.id() == l2_id_overlap)),
        "high range shard should have overlap L2 file"
    );
    assert!(
        levels_low
            .l2()
            .values()
            .any(|files| files.iter().any(|f| f.id() == l2_id_overlap)),
        "low range shard should have overlap L2 file"
    );

    // Verify l0_snap_version is preserved in both shards
    assert_eq!(
        levels_high.l0_snap_version, l0_snap_version_before,
        "high shard should preserve l0_snap_version"
    );
    assert_eq!(
        levels_low.l0_snap_version, l0_snap_version_before,
        "low shard should preserve l0_snap_version"
    );

    // Verify tracked indexes are preserved in both shards
    assert!(
        levels_high
            .iter_tracked_indexes()
            .any(|(tid, _)| *tid == table_id),
        "high range shard should track table_id"
    );
    assert!(
        levels_low
            .iter_tracked_indexes()
            .any(|(tid, _)| *tid == table_id),
        "low range shard should track table_id"
    );

    // Merge shard_low into shard_high.

    let shard_low_ver = shard_low.ver;
    engine
        .prepare_merge(
            shard_low.id,
            shard_low_ver,
            shard_low.get_write_sequence() + 1,
        )
        .unwrap();
    // Generate changeset for source shard.
    let mut source_cs = kvenginepb::ChangeSet::new();
    source_cs.set_shard_id(shard_low.id);
    source_cs.set_shard_ver(shard_low_ver + 1);
    let snap = source_cs.mut_snapshot();
    snap.set_outer_start(shard_low.get_data().range.outer_start.to_vec());
    snap.set_outer_end(shard_low.get_data().range.outer_end.to_vec());
    snap.set_base_version(shard_low.get_base_version());
    snap.set_data_sequence(shard_low.get_write_sequence());
    snap.set_columnar_table_ids(shard_low.get_data().columnar_table_ids.clone());
    snap.set_fts_l0_snap_version(levels_low.l0_snap_version.into_inner());
    snap.set_fts_indexes(
        levels_low
            .iter_tracked_indexes()
            .map(|(tid, _)| {
                let mut idx = kvenginepb::fts::TableIndexId::default();
                idx.set_table_id(*tid);
                idx.set_index_id(1);
                idx
            })
            .collect::<Vec<_>>()
            .into(),
    );
    snap.set_fts_index_metas(
        levels_low
            .iter_tracked_index_metas()
            .map(|(_, _, m)| m.clone())
            .collect::<Vec<_>>()
            .into(),
    );
    snap.set_fts_l0_files(
        levels_low
            .l0()
            .iter()
            .map(|file| file.build_info())
            .collect(),
    );
    snap.set_fts_l1_files(
        levels_low
            .l1()
            .iter()
            .map(|file| file.build_info())
            .collect(),
    );
    let fts_l2_files = levels_low
        .l2()
        .values()
        .map(|files| {
            files
                .iter()
                .map(|file| file.build_info())
                .collect::<Vec<_>>()
        })
        .collect::<Vec<_>>();
    snap.set_fts_l2_files(
        fts_l2_files
            .into_iter()
            .flatten()
            .collect::<Vec<_>>()
            .into(),
    );

    let prepared_cs = engine
        .prepare_change_set(
            source_cs,
            PrepareOpts {
                prepare_type: FilePrepareType::Local {
                    transit_from_ia: false,
                },
                ..Default::default()
            },
        )
        .unwrap();
    engine
        .commit_merge(
            shard_high.id,
            shard_high.ver,
            &prepared_cs,
            shard_high.get_write_sequence() + 1,
        )
        .unwrap();
    let merged_data = engine.get_shard(shard_high.id).unwrap().get_data();
    assert_eq!(
        merged_data.fts_levels.l0().len(),
        3,
        "should have 3 L0 files after merge"
    );
    assert_eq!(
        merged_data
            .fts_levels
            .l2()
            .values()
            .map(|v| v.len())
            .sum::<usize>(),
        3,
        "should have 3 L2 files after merge"
    );
}

#[test]
fn test_columnar_l0_compaction_skips_fts_pending_columnar_l0() {
    ::test_util::init_log_for_test();

    let keyspace_id = KEYSPACE_ID;
    let table_id = 61;
    let index_id = 1;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    // Two columnar L0 files; one is explicitly pending for FTS.
    let pending_col_id = id();
    let pending_col = put_columnar(
        &engine,
        new_columnar(pending_col_id, 7).table::<IntPk>(&schema, |row| {
            row(1, 7, false, |datum| {
                datum(10);
                datum("bluetooth pending");
            });
        }),
    );

    let tracked_col_id = id();
    let tracked_col = put_columnar(
        &engine,
        new_columnar(tracked_col_id, 5).table::<IntPk>(&schema, |row| {
            row(1, 5, false, |datum| {
                datum(20);
                datum("bluetooth tracked");
            });
        }),
    );

    let mut col_levels = ColumnarLevels::new();
    col_levels.add_file(0, pending_col);
    col_levels.add_file(0, tracked_col);

    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index(table_id, index_id);
    // Make pending_col_id "covered by watermark" but explicitly pending.
    fts_levels.l0_snap_version = SnapVersion::from(10);
    fts_levels.mut_pending_columnar_l0_ids(|ids| ids.push(pending_col_id));

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_columnar_levels(col_levels);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());
    shard.initial_flushed.store(true, Ordering::SeqCst);

    let cs = block_on(async { engine.trigger_columnar_l0_compaction(&shard).await })
        .expect("should trigger columnar L0 compaction")
        .expect("columnar L0 compaction should succeed");

    let deletes = cs
        .get_columnar_compaction()
        .get_columnar_change()
        .get_columnar_deletes()
        .iter()
        .map(|d| d.get_id())
        .collect::<Vec<_>>();
    assert!(
        !deletes.contains(&pending_col_id),
        "pending columnar L0 must not be compacted away: pending_col_id={}, deletes={:?}",
        pending_col_id,
        deletes
    );

    let (tx, rx) = tikv_util::mpsc::bounded(1);
    apply_tx.send(ApplyTask::new_cs(cs, tx)).unwrap();
    rx.recv().unwrap().unwrap();

    let data = shard.get_data();
    assert!(
        data.col_levels.levels[0]
            .files
            .iter()
            .any(|f| f.id() == pending_col_id),
        "pending columnar L0 must remain at level 0 after compaction"
    );
}

#[test]
fn test_fts_merge_keeps_source_unindexed_columnar_l0_pending() {
    ::test_util::init_log_for_test();

    let keyspace_id = KEYSPACE_ID;
    let table_id = 61;
    let index_id = 1;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    // Make split metadata consistent: schema + tracked index + some watermark.
    let mut init_levels = FtsLevels::default();
    init_levels.track_index(table_id, index_id);
    init_levels.l0_snap_version = SnapVersion::from(10);

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file.clone()));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(init_levels);
    shard.set_data(builder.build());
    shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    // Split at handle 50, original shard becomes the higher range shard.
    let split_key = [keyspace_prefix(keyspace_id), encode_row_key(table_id, 50)].concat();
    let shard_ver = shard.ver;
    let mut splitter = Splitter::new(
        vec![split_key],
        IdVer::new(shard_id, shard_ver),
        100,
        apply_tx,
    );
    let handle = std::thread::spawn(move || {
        splitter.run();
    });
    let ok = try_wait(|| engine.shards.len() == 6, 5);
    assert!(
        ok,
        "fts split failed: expected 6 shards, got {}",
        engine.shards.len()
    );
    handle.join().unwrap();

    let target_shard_id = 101;
    let target_shard = engine.get_shard(target_shard_id).unwrap();
    let source_shard = engine.get_shard(shard_id).unwrap();

    // Target has a higher FTS watermark, source has an unindexed columnar L0 with
    // snap_version between the two watermarks.
    let mut target_levels = FtsLevels::default();
    target_levels.track_index(table_id, index_id);
    target_levels.l0_snap_version = SnapVersion::from(10);

    let mut builder = ShardDataBuilder::new(target_shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file.clone()));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(target_levels);
    target_shard.set_data(builder.build());
    target_shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    let col_file_id = id();
    let source_col_file = put_columnar(
        &engine,
        new_columnar(col_file_id, 7).table::<IntPk>(&schema, |row| {
            row(1, 7, false, |datum| {
                datum(60);
                datum("bluetooth");
            });
        }),
    );
    let mut source_col_levels = ColumnarLevels::new();
    source_col_levels.add_file(0, source_col_file.clone());

    let mut source_levels = FtsLevels::default();
    source_levels.track_index(table_id, index_id);
    source_levels.l0_snap_version = SnapVersion::from(5);

    let mut builder = ShardDataBuilder::new(source_shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file.clone()));
    builder.set_columnar_levels(source_col_levels);
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(source_levels);
    source_shard.set_data(builder.build());
    source_shard
        .initial_flushed
        .store(true, std::sync::atomic::Ordering::SeqCst);

    // Prepare merge on source.
    let source_ver = source_shard.ver;
    engine
        .prepare_merge(
            source_shard.id,
            source_ver,
            source_shard.get_write_sequence() + 1,
        )
        .unwrap();

    // Build source changeset. Note we intentionally do NOT include any FTS files,
    // to simulate "unindexed columnar L0 exists".
    let mut source_cs = kvenginepb::ChangeSet::new();
    source_cs.set_shard_id(source_shard.id);
    source_cs.set_shard_ver(source_ver + 1);
    let snap = source_cs.mut_snapshot();
    snap.set_outer_start(source_shard.get_data().range.outer_start.to_vec());
    snap.set_outer_end(source_shard.get_data().range.outer_end.to_vec());
    snap.set_base_version(source_shard.get_base_version());
    snap.set_data_sequence(source_shard.get_write_sequence());
    snap.set_columnar_table_ids(source_shard.get_data().columnar_table_ids.clone());
    snap.set_fts_l0_snap_version(5);
    let mut tbl_idx = kvenginepb::fts::TableIndexId::default();
    tbl_idx.set_table_id(table_id);
    tbl_idx.set_index_id(index_id);
    snap.mut_fts_indexes().push(tbl_idx);

    snap.mut_columnar_creates()
        .push(source_col_file.to_columnar_create(0));

    let prepared_cs = engine
        .prepare_change_set(
            source_cs,
            PrepareOpts {
                prepare_type: FilePrepareType::Local {
                    transit_from_ia: false,
                },
                ..Default::default()
            },
        )
        .unwrap();

    let target_ver = target_shard.ver;
    engine
        .commit_merge(
            target_shard.id,
            target_ver,
            &prepared_cs,
            target_shard.get_write_sequence() + 1,
        )
        .unwrap();

    let merged_data = engine.get_shard(target_shard.id).unwrap().get_data();
    assert!(
        merged_data.col_levels.levels[0]
            .files
            .iter()
            .any(|f| f.id() == col_file_id),
        "merged shard should keep source columnar L0"
    );
    assert_eq!(
        merged_data.fts_levels.l0_snap_version,
        SnapVersion::from(10),
        "merged shard should keep max watermark"
    );
    assert!(
        merged_data
            .fts_levels
            .pending_columnar_l0_ids()
            .contains(&col_file_id),
        "source-unindexed columnar L0 must be pending after merge"
    );
}

#[test]
fn test_fts_reader_fallback_end_to_end_scored_and_unscored() {
    ::test_util::init_log_for_test();

    let keyspace_id = KEYSPACE_ID;
    let table_id = 150;
    let index_id = 1;
    let text_col_id = 2;

    let (engine, apply_tx) =
        new_test_engine_opt_with_custom_options(true, DEF_BLOCK_SIZE, "", |opts| {
            opts.read_columnar = false;
        });
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    // The tracked layout remains required even though columnar reads are disabled
    // and execution must use the whole-table brute-force reader.
    let index_def = FullTextIndexDef {
        index_id,
        col_id: text_col_id,
        parser_type: "STANDARD_V1".to_string(),
        ..Default::default()
    };
    let table_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(text_col_id, FieldTypeTp::String)
        .fts_index(index_def.clone())
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&table_schema]);

    // Columnar L0 data (converted).
    let col_file_id = id();
    let col_file = put_columnar(
        &engine,
        new_columnar(col_file_id, 5).table::<IntPk>(&table_schema, |row| {
            row(1, 10, false, |datum| {
                datum(101);
                datum("alpha from columnar");
            });
            row(2, 12, false, |datum| {
                datum(202);
                datum("alpha from columnar");
            });
            row(3, 11, false, |datum| {
                datum(303);
                datum("beta from columnar");
            });
        }),
    );
    let mut col_levels = ColumnarLevels::new();
    col_levels.add_file(0, col_file);

    // Memtable row data (unconverted).
    let mem = new_write_batch()
        .table::<IntPk>(&table_schema, |row| {
            row.v2(4, 13, |datum| {
                datum(404);
                datum("alpha from memtable");
            });
        })
        .finish_as_cf();

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_levels(col_levels);
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_mem_tbls(vec![mem]);
    builder.set_fts_levels(FtsLevels::default());
    shard.set_data(builder.build());

    let snap = SnapAccess::new(&shard);
    let read_ts = 20;
    let runtime = engine.fs.get_runtime().handle().clone();

    // ========================
    // WithScore (output schema excludes text column).
    // ========================
    let scored_query = make_scored_query(&PlainFtsQueryInfo {
        query: "alpha".to_string(),
        top_k: 10,
        column_id: text_col_id,
        index_id,
        ..Default::default()
    });
    let scored_schema_no_text = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .fts_score_column(false)
        .schema();
    crate::table::fts::validate_schema(&scored_schema_no_text, &scored_query).unwrap();
    let error = match snap.new_fts_reader(
        &runtime,
        table_id,
        scored_query.info(),
        scored_schema_no_text.clone(),
        read_ts,
        None,
        None,
    ) {
        Ok(_) => panic!("legacy query unexpectedly executed without a tracked layout"),
        Err(error) => error,
    };
    assert!(
        error
            .to_string()
            .contains("unresolved tracked FTS layout for query")
    );

    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index_meta(
        table_id,
        index_id,
        tracked_index_meta_from_index_def(
            table_id,
            &index_def,
            FtsFieldLayout::FtsFieldLayoutLegacyBody,
        ),
    );
    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());
    let snap = SnapAccess::new(&shard);

    let mut reader = snap
        .new_fts_reader(
            &runtime,
            table_id,
            scored_query.info(),
            scored_schema_no_text.clone(),
            read_ts,
            None,
            None,
        )
        .unwrap();
    let mut out = block_on(async {
        reader.set_unbounded_handle_range().await.unwrap();
        reader.read_all().await
    });
    out.sort();
    let expected = new_block::<IntPk>(&scored_schema_no_text, |row| {
        row(1, 10, false, |datum| {
            datum(101);
            datum(0.0);
        });
        row(2, 12, false, |datum| {
            datum(202);
            datum(0.0);
        });
        row(4, 13, false, |datum| {
            datum(404);
            datum(0.0);
        });
    });
    assert!(out.cols_eq(&expected, true, &[1]));
    let score_col = out.get_column(crate::table::fts::VIRTUAL_SCORE_COLUMN_ID);
    for row in 0..out.length() {
        let score = f64::from_le_bytes(score_col.get_value(row).unwrap().try_into().unwrap());
        assert!(score > 0.0, "scores must be positive for matched rows");
    }

    // ========================
    // WithScore (output schema includes text column).
    // ========================
    let scored_schema_with_text = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(text_col_id, FieldTypeTp::String)
        .fts_score_column(false)
        .schema();
    crate::table::fts::validate_schema(&scored_schema_with_text, &scored_query).unwrap();
    let mut reader = snap
        .new_fts_reader(
            &runtime,
            table_id,
            scored_query.info(),
            scored_schema_with_text.clone(),
            read_ts,
            None,
            None,
        )
        .unwrap();
    let mut out = block_on(async {
        reader.set_unbounded_handle_range().await.unwrap();
        reader.read_all().await
    });
    out.sort();
    let expected = new_block::<IntPk>(&scored_schema_with_text, |row| {
        row(1, 10, false, |datum| {
            datum(101);
            datum("alpha from columnar");
            datum(0.0);
        });
        row(2, 12, false, |datum| {
            datum(202);
            datum("alpha from columnar");
            datum(0.0);
        });
        row(4, 13, false, |datum| {
            datum(404);
            datum("alpha from memtable");
            datum(0.0);
        });
    });
    assert!(out.cols_eq(&expected, true, &[1, text_col_id]));
    let score_col = out.get_column(crate::table::fts::VIRTUAL_SCORE_COLUMN_ID);
    for row in 0..out.length() {
        let score = f64::from_le_bytes(score_col.get_value(row).unwrap().try_into().unwrap());
        assert!(score > 0.0, "scores must be positive for matched rows");
    }

    // ========================
    // NoScore (output schema excludes both text and score).
    // ========================
    let noscore_query = make_unscored_query(&PlainFtsQueryInfo {
        query: "alpha".to_string(),
        top_k: 0,
        column_id: text_col_id,
        index_id,
        ..Default::default()
    });
    let noscore_schema_no_text = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .schema();
    crate::table::fts::validate_schema(&noscore_schema_no_text, &noscore_query).unwrap();
    let mut reader = snap
        .new_fts_reader(
            &runtime,
            table_id,
            noscore_query.info(),
            noscore_schema_no_text.clone(),
            read_ts,
            None,
            None,
        )
        .unwrap();
    let mut out = block_on(async {
        reader.set_unbounded_handle_range().await.unwrap();
        reader.read_all().await
    });
    out.sort();
    let expected = new_block::<IntPk>(&noscore_schema_no_text, |row| {
        row(1, 10, false, |datum| {
            datum(101);
        });
        row(2, 12, false, |datum| {
            datum(202);
        });
        row(4, 13, false, |datum| {
            datum(404);
        });
    });
    assert!(out.eq(&expected));

    // ========================
    // NoScore (output schema includes text column).
    // ========================
    let noscore_schema_with_text = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(text_col_id, FieldTypeTp::String)
        .schema();
    crate::table::fts::validate_schema(&noscore_schema_with_text, &noscore_query).unwrap();
    let mut reader = snap
        .new_fts_reader(
            &runtime,
            table_id,
            noscore_query.info(),
            noscore_schema_with_text.clone(),
            read_ts,
            None,
            None,
        )
        .unwrap();
    let mut out = block_on(async {
        reader.set_unbounded_handle_range().await.unwrap();
        reader.read_all().await
    });
    out.sort();
    let expected = new_block::<IntPk>(&noscore_schema_with_text, |row| {
        row(1, 10, false, |datum| {
            datum(101);
            datum("alpha from columnar");
        });
        row(2, 12, false, |datum| {
            datum(202);
            datum("alpha from columnar");
        });
        row(4, 13, false, |datum| {
            datum(404);
            datum("alpha from memtable");
        });
    });
    assert!(out.eq(&expected));
}

fn exact_index_field(column_id: i64) -> FullTextIndexFieldDef {
    let mut index_field = FullTextIndexFieldDef::new();
    index_field.set_col_id(column_id);
    let option = index_field.mut_index_field_option();
    option.set_analyzer_type(FullTextAnalyzerType::FtsAnalyzerExactValueV1);
    option.set_enable_bm25(false);
    index_field
}

fn match_expr_query_info(
    query: impl Into<serde_json::Value>,
    column_id: i64,
    index_id: i64,
    index_field: &str,
) -> tipb::FtsQueryInfo {
    let mut info = PlainFtsQueryInfo {
        column_id,
        index_id,
        top_k: 0,
        ..Default::default()
    }
    .as_query_info();
    info.set_query_type(tipb::FtsQueryType::FtsQueryTypeNoScore);
    let mut expr = tipb::Expr::new();
    expr.set_tp(tipb::ExprType::ScalarFunc);
    expr.set_sig(tipb::ScalarFuncSig::FtsMatchWord);
    expr.set_val(
        serde_json::to_vec(&serde_json::json!({
            "version": FTS_MATCH_EXPR_METADATA_VERSION,
            "score_mode": "filter",
            "query": query.into(),
            "fields": [{
                "col_id": column_id,
                "index_field": index_field,
            }],
        }))
        .unwrap(),
    );
    info.mut_match_expr().push(expr);
    // TiDB match-expression requests carry source fields in expression
    // metadata and intentionally leave the legacy `columns` field empty.
    info.clear_columns();
    info
}

#[test]
fn test_fts_reader_fallback_exact_source_omitted_from_projection() {
    ::test_util::init_log_for_test();

    let table_id = 151;
    let index_id = 1;
    let payload_column_id = 1;
    let source_column_id = 2;

    let mut index_def = FullTextIndexDef::new();
    index_def.set_index_id(index_id);
    index_def.set_index_fields(vec![exact_index_field(source_column_id)].into());

    let table_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(payload_column_id, FieldTypeTp::LongLong)
        .column(source_column_id, FieldTypeTp::LongLong)
        .fts_index(index_def.clone())
        .schema();

    let (engine, apply_tx) =
        new_test_engine_opt_with_custom_options(true, DEF_BLOCK_SIZE, "", |opts| {
            opts.read_columnar = false;
        });
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, KEYSPACE_ID, table_id);
    let shard = engine.get_shard(shard_id).unwrap();
    let schema_file = put_schema_file(&engine, id(), &[&table_schema]);
    let mem = new_write_batch()
        .table::<IntPk>(&table_schema, |row| {
            row.v2(1, 10, |datum| {
                datum(101);
                datum(7);
            });
            row.v2(2, 10, |datum| {
                datum(202);
                datum(9);
            });
        })
        .finish_as_cf();

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_mem_tbls(vec![mem]);
    builder.set_fts_levels(FtsLevels::default());
    shard.set_data(builder.build());

    let query_info = match_expr_query_info(7, source_column_id, index_id, "EXACT_VALUE_V1");

    // The exact source column is not projected, so document construction must use
    // the table schema. Request metadata must not guess the physical layout when
    // the tracked layout is absent.
    let output_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(payload_column_id, FieldTypeTp::LongLong)
        .fts_index(index_def.clone())
        .schema();
    let snap = SnapAccess::new(&shard);
    let runtime = engine.fs.get_runtime().handle().clone();
    // The coordinator validates and retains only the request-level info. It
    // does not resolve a physical query before constructing a table reader.
    CloudColumnarReaders::new(
        runtime.clone(),
        snap.clone(),
        vec![TableCtx {
            table_id,
            ranges: Vec::new(),
        }],
        output_schema.columns.clone(),
        TableScanCtx::default(),
        None,
        Some(query_info.clone()),
        IaCtx::disabled(&crate::ia::util::IaConfig::default()),
        20,
        0,
    )
    .unwrap();
    let error = match snap.new_fts_reader(
        &runtime,
        table_id,
        &query_info,
        output_schema.clone(),
        20,
        None,
        None,
    ) {
        Ok(_) => panic!("fallback match_expr unexpectedly resolved without tracked layout"),
        Err(error) => error,
    };
    assert!(
        error
            .to_string()
            .contains("unresolved tracked FTS layout for query")
    );

    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index_meta(
        table_id,
        index_id,
        tracked_index_meta_from_index_def(
            table_id,
            &index_def,
            FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString,
        ),
    );
    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());

    let snap = SnapAccess::new(&shard);
    let mut reader = snap
        .new_fts_reader(
            &runtime,
            table_id,
            &query_info,
            output_schema.clone(),
            20,
            None,
            None,
        )
        .unwrap();
    let output = block_on(async {
        reader.set_unbounded_handle_range().await.unwrap();
        reader.read_all().await
    });
    let expected = new_block::<IntPk>(&output_schema, |row| {
        row(1, 10, false, |datum| {
            datum(101);
        });
    });
    assert!(output.eq(&expected));
}

#[test]
fn test_fts_reader_fallback_resolves_tracked_legacy_body_layout() {
    ::test_util::init_log_for_test();

    let table_id = 153;
    let index_id = 1;
    let payload_column_id = 1;
    let source_column_id = 2;
    // A legacy layout is exactly one text source mapped to the physical
    // `body` field. Multi-field and typed exact layouts are ordinal-only.
    let mut index_def = FullTextIndexDef::new();
    index_def.set_index_id(index_id);
    index_def.set_col_id(source_column_id);
    index_def.set_parser_type("STANDARD_V1".to_string());
    let table_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(payload_column_id, FieldTypeTp::LongLong)
        .column(source_column_id, FieldTypeTp::String)
        .fts_index(index_def.clone())
        .schema();

    let (engine, apply_tx) =
        new_test_engine_opt_with_custom_options(true, DEF_BLOCK_SIZE, "", |opts| {
            opts.read_columnar = false;
        });
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, KEYSPACE_ID, table_id);
    let shard = engine.get_shard(shard_id).unwrap();
    let schema_file = put_schema_file(&engine, id(), &[&table_schema]);
    let mem = new_write_batch()
        .table::<IntPk>(&table_schema, |row| {
            row.v2(1, 10, |datum| {
                datum(101);
                datum("rust language");
            });
            row.v2(2, 10, |datum| {
                datum(202);
                datum("search engine");
            });
        })
        .finish_as_cf();
    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index_meta(
        table_id,
        index_id,
        tracked_index_meta_from_index_def(
            table_id,
            &index_def,
            FtsFieldLayout::FtsFieldLayoutLegacyBody,
        ),
    );
    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_mem_tbls(vec![mem]);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());

    let query_info = match_expr_query_info("rust", source_column_id, index_id, "STANDARD_V1");
    let output_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(payload_column_id, FieldTypeTp::LongLong)
        .fts_index(index_def)
        .schema();
    let snap = SnapAccess::new(&shard);
    let runtime = engine.fs.get_runtime().handle().clone();
    let mut reader = snap
        .new_fts_reader(
            &runtime,
            table_id,
            &query_info,
            output_schema.clone(),
            20,
            None,
            None,
        )
        .unwrap();
    let output = block_on(async {
        reader.set_unbounded_handle_range().await.unwrap();
        reader.read_all().await
    });
    let expected = new_block::<IntPk>(&output_schema, |row| {
        row(1, 10, false, |datum| {
            datum(101);
        });
    });
    assert!(output.eq(&expected));
}

#[test]
fn test_cloud_columnar_reader_fts_scored_and_unscored_multi_range() {
    ::test_util::init_log_for_test();

    let keyspace_id = KEYSPACE_ID;
    let table_id = 152;
    let index_id = 1;
    let text_col_id = 2;

    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    // The tracked layout lets each cloud reader build its own complete query.
    // With no FTS index files, source rows are evaluated by the conditional
    // brute-force path.
    let index_def = FullTextIndexDef {
        index_id,
        col_id: text_col_id,
        parser_type: "STANDARD_V1".to_string(),
        ..Default::default()
    };
    let table_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(text_col_id, FieldTypeTp::String)
        .fts_index(index_def.clone())
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&table_schema]);

    // Row storage backing for the converted columnar rows. The conditional FTS
    // reader consumes query source columns, then FtsJoinReader loads the final
    // projection from this row L0 by exact handle/version lookup.
    let row_l0 = new_write_batch()
        .table::<IntPk>(&table_schema, |row| {
            row.v2(1, 10, |datum| {
                datum(101);
                datum("alpha from columnar");
            });
            row.v2(2, 12, |datum| {
                datum(202);
                datum("alpha from columnar");
            });
            row.v2(3, 11, |datum| {
                datum(303);
                datum("beta from columnar");
            });
        })
        .finish_as_sst_l0(id());

    // Columnar L0 data (converted).
    let col_file_id = id();
    let col_file = put_columnar(
        &engine,
        new_columnar(col_file_id, 5).table::<IntPk>(&table_schema, |row| {
            row(1, 10, false, |datum| {
                datum(101);
                datum("alpha from columnar");
            });
            row(2, 12, false, |datum| {
                datum(202);
                datum("alpha from columnar");
            });
            row(3, 11, false, |datum| {
                datum(303);
                datum("beta from columnar");
            });
        }),
    );
    let mut col_levels = ColumnarLevels::new();
    col_levels.add_file(0, col_file);

    // Memtable row data (unconverted).
    let mem = new_write_batch()
        .table::<IntPk>(&table_schema, |row| {
            row.v2(4, 13, |datum| {
                datum(404);
                datum("alpha from memtable");
            });
        })
        .finish_as_cf();

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_l0_tbls(vec![row_l0]);
    builder.set_columnar_levels(col_levels);
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_mem_tbls(vec![mem]);
    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index_meta(
        table_id,
        index_id,
        tracked_index_meta_from_index_def(
            table_id,
            &index_def,
            FtsFieldLayout::FtsFieldLayoutLegacyBody,
        ),
    );
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());

    let snap = SnapAccess::new(&shard);
    let read_ts = 20;

    // Multi-range scan: range 1 contains matched rows 1 & 2; range 2 contains
    // matched row 4, ensuring range switching is exercised.
    let ranges_for_cloud = {
        let mut r1 = KeyRange::default();
        let mut low1 = keyspace_prefix(keyspace_id);
        low1.extend_from_slice(&encode_row_key(table_id, i64::MIN));
        let mut high1 = keyspace_prefix(keyspace_id);
        high1.extend_from_slice(&encode_row_key(table_id, 3));
        r1.set_low(low1);
        r1.set_high(high1);

        let mut r2 = KeyRange::default();
        let mut low2 = keyspace_prefix(keyspace_id);
        low2.extend_from_slice(&encode_row_key(table_id, 3));
        let mut high2 = keyspace_prefix(keyspace_id);
        high2.extend_from_slice(&encode_row_key(table_id, i64::MAX));
        r2.set_low(low2);
        r2.set_high(high2);
        vec![r1, r2]
    };
    let scan_ctx = {
        let mut tbl_scan = TableScan::default();
        tbl_scan.set_table_id(table_id);
        let mut executor = Executor::default();
        executor.set_tbl_scan(tbl_scan);
        TableScanCtx::new(executor, vec![])
    };
    // ========================
    // WithScore + text
    // ========================
    let scored_query = make_scored_query(&PlainFtsQueryInfo {
        query: "alpha".to_string(),
        top_k: 10,
        column_id: text_col_id,
        index_id,
        ..Default::default()
    });
    let scored_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(text_col_id, FieldTypeTp::String)
        .fts_score_column(false)
        .schema();
    crate::table::fts::validate_schema(&scored_schema, &scored_query).unwrap();

    let runtime = tokio::runtime::Builder::new_multi_thread()
        .thread_name("test-fts-reader")
        .enable_all()
        .worker_threads(2)
        .build()
        .unwrap();
    let runtime_handle = runtime.handle().clone();
    let ia_ctx = IaCtx::disabled(&crate::ia::util::IaConfig::default());

    let mut reader = CloudColumnarReader::new(
        runtime_handle.clone(),
        snap.clone(),
        ia_ctx.clone(),
        table_id,
        ranges_for_cloud.clone(),
        &scored_schema.columns,
        &scan_ctx,
        None,
        Some(Arc::new(scored_query.info().clone())),
        read_ts,
    )
    .unwrap();
    let mut out = reader.read_all_for_test().unwrap();
    out.sort();
    let expected = new_block::<IntPk>(&scored_schema, |row| {
        row(1, 10, false, |datum| {
            datum(101);
            datum("alpha from columnar");
            datum(0.0);
        });
        row(2, 12, false, |datum| {
            datum(202);
            datum("alpha from columnar");
            datum(0.0);
        });
        row(4, 13, false, |datum| {
            datum(404);
            datum("alpha from memtable");
            datum(0.0);
        });
    });
    assert!(out.cols_eq(&expected, true, &[1, text_col_id]));
    let score_col = out.get_column(crate::table::fts::VIRTUAL_SCORE_COLUMN_ID);
    for row in 0..out.length() {
        let score = f64::from_le_bytes(score_col.get_value(row).unwrap().try_into().unwrap());
        assert!(score > 0.0, "scores must be positive for matched rows");
    }

    // ========================
    // NoScore + text
    // ========================
    let noscore_query = make_unscored_query(&PlainFtsQueryInfo {
        query: "alpha".to_string(),
        top_k: 0,
        column_id: text_col_id,
        index_id,
        ..Default::default()
    });
    let noscore_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(text_col_id, FieldTypeTp::String)
        .schema();
    crate::table::fts::validate_schema(&noscore_schema, &noscore_query).unwrap();

    let mut reader = CloudColumnarReader::new(
        runtime_handle.clone(),
        snap.clone(),
        ia_ctx.clone(),
        table_id,
        ranges_for_cloud.clone(),
        &noscore_schema.columns,
        &scan_ctx,
        None,
        Some(Arc::new(noscore_query.info().clone())),
        read_ts,
    )
    .unwrap();
    let mut out = reader.read_all_for_test().unwrap();
    out.sort();
    let expected = new_block::<IntPk>(&noscore_schema, |row| {
        row(1, 10, false, |datum| {
            datum(101);
            datum("alpha from columnar");
        });
        row(2, 12, false, |datum| {
            datum(202);
            datum("alpha from columnar");
        });
        row(4, 13, false, |datum| {
            datum(404);
            datum("alpha from memtable");
        });
    });
    assert!(out.eq(&expected));

    // The multi-table coordinator retains request-level query info. The
    // concrete table reader builds its own complete execution query.
    let mut readers = CloudColumnarReaders::new(
        runtime_handle,
        snap,
        vec![TableCtx {
            table_id,
            ranges: ranges_for_cloud,
        }],
        noscore_schema.columns.clone(),
        scan_ctx,
        None,
        Some(noscore_query.info().clone()),
        ia_ctx,
        read_ts,
        0,
    )
    .unwrap();
    let mut rows = 0;
    loop {
        let read = readers.ffi_read_block(1024).unwrap();
        if read == 0 {
            break;
        }
        rows += read;
    }
    assert_eq!(rows, 3);
}

#[test]
fn test_cloud_columnar_readers_isolate_bm25_state_per_table_reader() {
    ::test_util::init_log_for_test();

    let first_table_id = 154;
    let second_table_id = 155;
    let index_id = 1;
    let payload_column_id = 1;
    let text_column_id = 2;
    let index_def = FullTextIndexDef {
        index_id,
        col_id: text_column_id,
        parser_type: "STANDARD_V1".to_string(),
        ..Default::default()
    };
    let first_schema = SchemaBuilder::<IntPk>::new(first_table_id)
        .column(payload_column_id, FieldTypeTp::LongLong)
        .column(text_column_id, FieldTypeTp::String)
        .fts_index(index_def.clone())
        .schema();
    let second_schema = SchemaBuilder::<IntPk>::new(second_table_id)
        .column(payload_column_id, FieldTypeTp::LongLong)
        .column(text_column_id, FieldTypeTp::String)
        .fts_index(index_def.clone())
        .schema();

    let (engine, _apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard = engine.get_shard(1).unwrap();
    let schema_file = put_schema_file(&engine, id(), &[&first_schema, &second_schema]);

    // The two logical indexes deliberately have different BM25 corpora. If the
    // concrete readers share one QueryCore, whichever reader prepares the
    // OnceLock first will force the other reader to use the wrong statistics.
    let indexed_columnar = put_columnar(
        &engine,
        new_columnar(id(), 100)
            .table::<IntPk>(&first_schema, |row| {
                row(1, 50, false, |datum| {
                    datum(11);
                    datum("alpha");
                });
                row(2, 50, false, |datum| {
                    datum(12);
                    datum("beta");
                });
            })
            .table::<IntPk>(&second_schema, |row| {
                row(1, 50, false, |datum| {
                    datum(21);
                    datum("alpha");
                });
                for handle in 2..=10 {
                    row(handle, 50, false, |datum| {
                        datum(20 + handle);
                        datum("beta");
                    });
                }
            }),
    );
    let row_l0 = new_write_batch()
        .table::<IntPk>(&first_schema, |row| {
            row.v2(1, 50, |datum| {
                datum(11);
                datum("alpha");
            });
            row.v2(2, 50, |datum| {
                datum(12);
                datum("beta");
            });
        })
        .table::<IntPk>(&second_schema, |row| {
            row.v2(1, 50, |datum| {
                datum(21);
                datum("alpha");
            });
            for handle in 2..=10 {
                row.v2(handle, 50, |datum| {
                    datum(20 + handle);
                    datum("beta");
                });
            }
        })
        .finish_as_sst_l0(id());
    let fts_file_id = id();
    let indexed_fts = put_packed(
        &engine,
        fts_file_id,
        new_packed(fts_file_id, 100)
            .lp(first_table_id, index_id, |doc| {
                doc(1, 50, false, "alpha");
                doc(2, 50, false, "beta");
            })
            .lp(second_table_id, index_id, |doc| {
                doc(1, 50, false, "alpha");
                for handle in 2..=10 {
                    doc(handle, 50, false, "beta");
                }
            }),
    );

    let mut columnar_levels = ColumnarLevels::new();
    columnar_levels.add_file(0, indexed_columnar);
    let mut fts_levels = FtsLevels::default();
    for table_id in [first_table_id, second_table_id] {
        fts_levels.track_index_meta(
            table_id,
            index_id,
            tracked_index_meta_from_index_def(
                table_id,
                &index_def,
                FtsFieldLayout::FtsFieldLayoutLegacyBody,
            ),
        );
    }
    fts_levels.mut_l0(|files| files.push(indexed_fts));
    fts_levels.l0_snap_version = 100.into();

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_l0_tbls(vec![row_l0]);
    builder.set_columnar_levels(columnar_levels);
    builder.set_columnar_table_ids(vec![first_table_id, second_table_id]);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());

    let query_info = make_scored_query(&PlainFtsQueryInfo {
        query: "alpha".to_string(),
        top_k: 10,
        column_id: text_column_id,
        index_id,
        ..Default::default()
    })
    .info()
    .clone();
    let output_schema = |table_id| {
        SchemaBuilder::<IntPk>::new(table_id)
            .column(payload_column_id, FieldTypeTp::LongLong)
            .fts_score_column(false)
            .schema()
    };
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .thread_name("test-fts-bm25-isolation")
        .enable_all()
        .worker_threads(2)
        .build()
        .unwrap();
    let runtime_handle = runtime.handle().clone();
    let snap = SnapAccess::new(&shard);
    let score_from_independent_reader = |table_id| {
        let schema = output_schema(table_id);
        let mut reader = snap
            .new_fts_reader(
                &runtime_handle,
                table_id,
                &query_info,
                schema,
                60,
                None,
                None,
            )
            .unwrap();
        let output = block_on(async {
            reader.set_unbounded_handle_range().await.unwrap();
            reader.read_all().await
        });
        assert_eq!(output.length(), 1);
        f64::from_le_bytes(
            output
                .get_column(crate::table::fts::VIRTUAL_SCORE_COLUMN_ID)
                .get_value(0)
                .unwrap()
                .try_into()
                .unwrap(),
        )
    };
    let expected_first_score = score_from_independent_reader(first_table_id);
    let expected_second_score = score_from_independent_reader(second_table_id);
    assert!(
        (expected_first_score - expected_second_score).abs() > 1e-3,
        "test corpora must produce distinguishable BM25 scores"
    );

    let table_range = |table_id| {
        let mut range = KeyRange::default();
        let mut low = keyspace_prefix(KEYSPACE_ID);
        low.extend_from_slice(&encode_row_key(table_id, i64::MIN));
        let mut high = keyspace_prefix(KEYSPACE_ID);
        high.extend_from_slice(&encode_row_key(table_id, i64::MAX));
        range.set_low(low);
        range.set_high(high);
        vec![range]
    };
    let mut table_scan = TableScan::default();
    table_scan.set_table_id(first_table_id);
    let mut executor = Executor::default();
    executor.set_tbl_scan(table_scan);
    let scan_ctx = TableScanCtx::new(executor, vec![]);
    let mut readers = CloudColumnarReaders::new(
        runtime_handle,
        snap,
        vec![
            TableCtx {
                table_id: first_table_id,
                ranges: table_range(first_table_id),
            },
            TableCtx {
                table_id: second_table_id,
                ranges: table_range(second_table_id),
            },
        ],
        output_schema(first_table_id).columns.clone(),
        scan_ctx,
        None,
        Some(query_info),
        IaCtx::disabled(&crate::ia::util::IaConfig::default()),
        60,
        2,
    )
    .unwrap();
    let mut actual_scores = HashMap::default();
    loop {
        let rows = readers.ffi_read_block(1024).unwrap();
        if rows == 0 {
            break;
        }
        assert_eq!(rows, 1);
        let table_id = readers.ffi_physical_table_id();
        let score = f32::from_le_bytes(
            readers
                .ffi_read_column(crate::table::fts::VIRTUAL_SCORE_COLUMN_ID)
                .try_into()
                .unwrap(),
        ) as f64;
        assert!(actual_scores.insert(table_id, score).is_none());
    }

    assert!((actual_scores[&first_table_id] - expected_first_score).abs() < 1e-5);
    assert!((actual_scores[&second_table_id] - expected_second_score).abs() < 1e-5);
}

#[test]
fn test_fts_reader_fallback_mvcc_scored_and_noscore_with_text() {
    ::test_util::init_log_for_test();

    let keyspace_id = KEYSPACE_ID;
    let table_id = 151;
    let index_id = 1;
    let text_col_id = 2;

    let (engine, apply_tx) =
        new_test_engine_opt_with_custom_options(true, DEF_BLOCK_SIZE, "", |opts| {
            opts.read_columnar = false;
        });
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    // The tracked layout remains required even though columnar reads are disabled
    // and execution must use the whole-table brute-force reader.
    let index_def = FullTextIndexDef {
        index_id,
        col_id: text_col_id,
        parser_type: "STANDARD_V1".to_string(),
        ..Default::default()
    };
    let table_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(text_col_id, FieldTypeTp::String)
        .fts_index(index_def.clone())
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&table_schema]);

    let mem = crate::table::memtable::CfTable::new();
    let mem_old = crate::table::memtable::CfTable::new();
    let col_levels = ColumnarLevels::new();

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_levels(col_levels);
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_mem_tbls(vec![mem.clone(), mem_old.clone()]);
    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index_meta(
        table_id,
        index_id,
        tracked_index_meta_from_index_def(
            table_id,
            &index_def,
            FtsFieldLayout::FtsFieldLayoutLegacyBody,
        ),
    );
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());

    let snap = SnapAccess::new(&shard);
    let runtime = engine.fs.get_runtime().handle().clone();

    // Keep an older version in an "old" memtable so tombstones are preserved and
    // can shadow it later.
    new_write_batch()
        .table::<IntPk>(&table_schema, |row| {
            row.v2(3, 15, |datum| {
                datum(33);
                datum("alpha row 3 v15");
            });
        })
        .apply_to_cf(&mem_old, Some(&snap));

    // MVCC setup:
    // - 1: updated at ts=30 but still matches query.
    // - 2: updated at ts=30 and no longer matches query.
    // - 3: deleted at ts=25.
    // - 4: inserted at ts=32.
    new_write_batch()
        .table::<IntPk>(&table_schema, |row| {
            row.v2(1, 10, |datum| {
                datum(11);
                datum("alpha row 1 v10");
            });
            row.v2(1, 30, |datum| {
                datum(31);
                datum("alpha row 1 v30");
            });
            row.v2(2, 12, |datum| {
                datum(22);
                datum("alpha row 2 v12");
            });
            row.v2(2, 30, |datum| {
                datum(30);
                datum("beta row 2 v30");
            });
            row.delete(3, 25);
            row.v2(4, 32, |datum| {
                datum(44);
                datum("alpha row 4 v32");
            });
        })
        .apply_to_cf(&mem, Some(&snap));

    let read_ts_before = 20;
    let read_ts_after = 35;

    // ========================
    // WithScore + text
    // ========================
    let scored_query = make_scored_query(&PlainFtsQueryInfo {
        query: "alpha".to_string(),
        top_k: 10,
        column_id: text_col_id,
        index_id,
        ..Default::default()
    });
    let scored_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(text_col_id, FieldTypeTp::String)
        .fts_score_column(false)
        .schema();
    crate::table::fts::validate_schema(&scored_schema, &scored_query).unwrap();
    let read_scored = |read_ts: u64| -> crate::table::columnar::Block {
        let mut reader = snap
            .new_fts_reader(
                &runtime,
                table_id,
                scored_query.info(),
                scored_schema.clone(),
                read_ts,
                None,
                None,
            )
            .unwrap();
        block_on(async {
            reader.set_unbounded_handle_range().await.unwrap();
            reader.read_all().await
        })
    };

    let mut scored_before = read_scored(read_ts_before);
    scored_before.sort();
    let expected_before = new_block::<IntPk>(&scored_schema, |row| {
        row(1, 10, false, |datum| {
            datum(11);
            datum("alpha row 1 v10");
            datum(0.0);
        });
        row(2, 12, false, |datum| {
            datum(22);
            datum("alpha row 2 v12");
            datum(0.0);
        });
        row(3, 15, false, |datum| {
            datum(33);
            datum("alpha row 3 v15");
            datum(0.0);
        });
    });
    assert!(scored_before.cols_eq(&expected_before, true, &[1, text_col_id]));
    let score_col_before = scored_before.get_column(crate::table::fts::VIRTUAL_SCORE_COLUMN_ID);
    for row in 0..scored_before.length() {
        let score =
            f64::from_le_bytes(score_col_before.get_value(row).unwrap().try_into().unwrap());
        assert!(score > 0.0, "scores must be positive for matched rows");
    }

    let mut scored_after = read_scored(read_ts_after);
    scored_after.sort();
    let expected_after = new_block::<IntPk>(&scored_schema, |row| {
        row(1, 30, false, |datum| {
            datum(31);
            datum("alpha row 1 v30");
            datum(0.0);
        });
        row(4, 32, false, |datum| {
            datum(44);
            datum("alpha row 4 v32");
            datum(0.0);
        });
    });
    assert!(scored_after.cols_eq(&expected_after, true, &[1, text_col_id]));
    let score_col_after = scored_after.get_column(crate::table::fts::VIRTUAL_SCORE_COLUMN_ID);
    for row in 0..scored_after.length() {
        let score = f64::from_le_bytes(score_col_after.get_value(row).unwrap().try_into().unwrap());
        assert!(score > 0.0, "scores must be positive for matched rows");
    }

    // ========================
    // NoScore + text
    // ========================
    let noscore_query = make_unscored_query(&PlainFtsQueryInfo {
        query: "alpha".to_string(),
        top_k: 0,
        column_id: text_col_id,
        index_id,
        ..Default::default()
    });
    let noscore_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(text_col_id, FieldTypeTp::String)
        .schema();
    crate::table::fts::validate_schema(&noscore_schema, &noscore_query).unwrap();
    let read_noscore = |read_ts: u64| -> crate::table::columnar::Block {
        let mut reader = snap
            .new_fts_reader(
                &runtime,
                table_id,
                noscore_query.info(),
                noscore_schema.clone(),
                read_ts,
                None,
                None,
            )
            .unwrap();
        block_on(async {
            reader.set_unbounded_handle_range().await.unwrap();
            reader.read_all().await
        })
    };

    let mut noscore_before = read_noscore(read_ts_before);
    noscore_before.sort();
    let expected_before = new_block::<IntPk>(&noscore_schema, |row| {
        row(1, 10, false, |datum| {
            datum(11);
            datum("alpha row 1 v10");
        });
        row(2, 12, false, |datum| {
            datum(22);
            datum("alpha row 2 v12");
        });
        row(3, 15, false, |datum| {
            datum(33);
            datum("alpha row 3 v15");
        });
    });
    assert!(noscore_before.eq(&expected_before));

    let mut noscore_after = read_noscore(read_ts_after);
    noscore_after.sort();
    let expected_after = new_block::<IntPk>(&noscore_schema, |row| {
        row(1, 30, false, |datum| {
            datum(31);
            datum("alpha row 1 v30");
        });
        row(4, 32, false, |datum| {
            datum(44);
            datum("alpha row 4 v32");
        });
    });
    assert!(noscore_after.eq(&expected_after));
}

#[rstest]
#[case::delta_cache_off(false)]
#[case::delta_cache_on(true)]
fn test_fts_reader_indexed_end_to_end_scored_and_unscored(#[case] delta_cache_enabled: bool) {
    ::test_util::init_log_for_test();

    let keyspace_id = KEYSPACE_ID;
    let table_id = 160;
    let index_id = 2;
    let text_col_id = 2;

    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let table_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column_not_null(text_col_id, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: text_col_id,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&table_schema]);

    // Build a row-based L0 table that serves as the backing store for join.
    //
    // These rows must be visible for exact (handle, version) lookups.
    let row_l0_id = id();
    let row_l0 = new_write_batch()
        .table::<IntPk>(&table_schema, |row| {
            row.v2(1001, 50, |datum| {
                datum(11);
                datum("alpha row 1001");
            });
            row.v2(1002, 40, |datum| {
                datum(22);
                datum("alpha row 1002 v40");
            });
            row.v2(1003, 40, |datum| {
                datum(33);
                datum("alpha row 1003 v40");
            });
            row.v2(1005, 40, |datum| {
                datum(55);
                datum("alpha row 1005 v40");
            });
            row.v2(1006, 52, |datum| {
                datum(66);
                datum("alpha col untracked");
            });
        })
        .finish_as_sst_l0(row_l0_id);

    // Memtable holds unindexed updates / deletes and new data.
    //
    // For MVCC / visibility:
    // - 1002: newer *indexed* version exists but does NOT match query; old indexed
    //   hit must disappear at a larger `read_ts` (indexed-only visibility).
    // - 1003: newer *unindexed* update overlaps with indexed hit; old indexed hit
    //   must disappear at a larger `read_ts` (indexed+unindexed visibility).
    // - 1005: newer *unindexed* delete overlaps with indexed hit; old indexed hit
    //   must disappear at a larger `read_ts` (indexed+unindexed visibility).
    let mem = new_write_batch()
        .table::<IntPk>(&table_schema, |row| {
            // Newer unindexed update that should shadow the indexed hit for 1003.
            row.v2(1003, 60, |datum| {
                datum(333);
                datum("beta row 1003");
            });
            // Unindexed row that should be matched by brute-force after MVCC.
            row.v2(1004, 55, |datum| {
                datum(14);
                datum("alpha mem 1004");
            });
        })
        .finish_as_cf();

    // Unconverted row L0 (unindexed).
    // - DeltaCache disabled: brute-force input.
    // - DeltaCache enabled: searched via delta index.
    let unconv_l0_id = id();
    let unconv_l0 = new_write_batch()
        .table::<IntPk>(&table_schema, |row| {
            row.v2(1007, 53, |datum| {
                datum(77);
                datum("alpha unconverted l0 1007");
            });
        })
        .finish_as_sst_l0(unconv_l0_id);

    // Columnar L0:
    // - tracked_col_file: already covered by FTS and should not be read as
    //   unindexed source.
    // - untracked_col_file: unindexed source.
    //   - DeltaCache disabled: brute-force input.
    //   - DeltaCache enabled: searched via delta index.
    let tracked_col_file_id = id();
    let tracked_col_file = put_columnar(
        &engine,
        new_columnar(tracked_col_file_id, 10).table::<IntPk>(&table_schema, |row| {
            row(1001, 50, false, |datum| {
                datum(111);
                datum("alpha col tracked");
            });
        }),
    );
    let untracked_col_file_id = id();
    let untracked_col_file = put_columnar(
        &engine,
        new_columnar(untracked_col_file_id, 200).table::<IntPk>(&table_schema, |row| {
            row(1006, 52, false, |datum| {
                datum(666);
                datum("alpha col untracked");
            });
        }),
    );
    let mut col_levels = ColumnarLevels::new();
    col_levels.unconverted_l0s.push(unconv_l0.clone());
    col_levels.add_file(0, tracked_col_file);
    col_levels.add_file(0, untracked_col_file);

    // FTS index (packed L0):
    // - "old" file: contains hits that match the query.
    // - "new" file: only contains a newer version for 1002 that does NOT match the
    //   query, to test indexed-only visibility.
    let fts_old_id = id();
    let fts_old = put_packed(
        &engine,
        fts_old_id,
        new_packed(fts_old_id, 100).lp(table_id, index_id, |d| {
            d(1001, 50, false, "alpha indexed 1001");
            d(1002, 40, false, "alpha indexed 1002 v40");
            d(1003, 40, false, "alpha indexed 1003 v40");
            d(1005, 40, false, "alpha indexed 1005 v40");
        }),
    );
    let fts_new_id = id();
    let fts_new = put_packed(
        &engine,
        fts_new_id,
        new_packed(fts_new_id, 150).lp(table_id, index_id, |d| {
            d(1002, 60, false, "beta indexed 1002 v60");
        }),
    );
    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index(table_id, index_id);
    fts_levels.mut_l0(|files| {
        files.push(fts_old);
        files.push(fts_new);
    });

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_l0_tbls(vec![row_l0, unconv_l0]);
    builder.set_mem_tbls(vec![mem.clone()]);
    builder.set_columnar_levels(col_levels);
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());

    let fts_delta_cache = if delta_cache_enabled {
        FtsDeltaCache::new(ReadableSize::mb(64).0)
    } else {
        FtsDeltaCache::disabled()
    };
    let snap = SnapAccess::new_with_fts_delta_cache(&shard, fts_delta_cache.clone());
    let runtime = engine.fs.get_runtime().handle().clone();

    // Newer unindexed delete that should shadow the indexed hit for 1005.
    // `CfTable::put_batch` needs a snapshot to decide whether to keep a tombstone
    // (vs dropping it when no older versions exist).
    new_write_batch()
        .table::<IntPk>(&table_schema, |row| {
            row.delete(1005, 60);
        })
        .apply_to_cf(&mem, Some(&snap));

    let read_ts_before = 55;
    let read_ts_after = 70;
    let start = 1000i64;
    let end = Some(1008i64);
    // For `SnapAccessCore::new_fts_reader` bounds, int handles are encoded in TiDB
    // row-key format (memcomparable i64), not little-endian.
    let start_bytes = Bytes::copy_from_slice(&IntPk::encode(start));
    let end_bytes = end.map(|e| Bytes::copy_from_slice(&IntPk::encode(e)));

    // Full Data layout:
    //   ShardData(table_id=160)
    //   |- Row storage (WRITE_CF)
    //   |  |- l0_tbls: [row_l0, unconv_l0]
    //   |  |  |- row_l0: 1001@50, 1002@40, 1003@40, 1005@40, 1006@52
    //   |  |  `- unconv_l0: 1007@53
    //   |  `- mem_tbls: [mem]
    //   |     |- 1003@60 = "beta row 1003" (unindexed update)
    //   |     |- 1004@55 = "alpha mem 1004" (unindexed new row)
    //   |     `- 1005@60 = DELETE tombstone (unindexed delete)
    //   |
    //   |- Columnar storage (L0 files)
    //   |  `- col_levels.level0.files:
    //   |     |- tracked_col_file  snap_version=10  (<= watermark)
    //   |     |    - 1001@50 text="alpha col tracked"
    //   |     `- untracked_col_file snap_version=200 (> watermark)
    //   |          - 1006@52 text="alpha col untracked"
    //   |
    //   `- FTS index storage
    //       `- fts_levels (track_index + l0 files)
    //           |- fts_old snap_version=100: 1001@50,1002@40,1003@40,1005@40
    //           `- fts_new snap_version=150: 1002@60 ("beta ...", not matching)
    //           => l0_snap_version watermark = 150

    // ========================
    // WithScore
    // ========================
    let scored_query = make_scored_query(&PlainFtsQueryInfo {
        query: "alpha".to_string(),
        top_k: 10,
        column_id: text_col_id,
        index_id,
        ..Default::default()
    });
    let scored_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column_not_null(text_col_id, FieldTypeTp::String)
        .fts_score_column(false)
        .schema();
    crate::table::fts::validate_schema(&scored_schema, &scored_query).unwrap();
    let read_scored = |read_ts: u64| -> crate::table::columnar::Block {
        let mut reader = snap
            .new_fts_reader(
                &runtime,
                table_id,
                scored_query.info(),
                scored_schema.clone(),
                read_ts,
                Some(start_bytes.clone()),
                end_bytes.clone(),
            )
            .unwrap();
        block_on(async {
            reader.set_int_handle_range(start, end).await.unwrap();
            reader.read_all().await
        })
    };

    // MVCC / visibility assertions for WithScore.
    let mut scored_before = read_scored(read_ts_before);
    scored_before.sort();
    let expected_before = new_block::<IntPk>(&scored_schema, |row| {
        row(1001, 50, false, |datum| {
            datum(11);
            datum("alpha row 1001");
            datum(0.0);
        });
        row(1002, 40, false, |datum| {
            datum(22);
            datum("alpha row 1002 v40");
            datum(0.0);
        });
        row(1003, 40, false, |datum| {
            datum(33);
            datum("alpha row 1003 v40");
            datum(0.0);
        });
        row(1004, 55, false, |datum| {
            datum(14);
            datum("alpha mem 1004");
            datum(0.0);
        });
        row(1005, 40, false, |datum| {
            datum(55);
            datum("alpha row 1005 v40");
            datum(0.0);
        });
        row(1006, 52, false, |datum| {
            datum(66);
            datum("alpha col untracked");
            datum(0.0);
        });
        row(1007, 53, false, |datum| {
            datum(77);
            datum("alpha unconverted l0 1007");
            datum(0.0);
        });
    });
    assert!(scored_before.cols_eq(&expected_before, true, &[1, text_col_id]));

    let mut scored_after = read_scored(read_ts_after);
    scored_after.sort();
    let expected_after = new_block::<IntPk>(&scored_schema, |row| {
        row(1001, 50, false, |datum| {
            datum(11);
            datum("alpha row 1001");
            datum(0.0);
        });
        row(1004, 55, false, |datum| {
            datum(14);
            datum("alpha mem 1004");
            datum(0.0);
        });
        row(1006, 52, false, |datum| {
            datum(66);
            datum("alpha col untracked");
            datum(0.0);
        });
        row(1007, 53, false, |datum| {
            datum(77);
            datum("alpha unconverted l0 1007");
            datum(0.0);
        });
    });
    assert!(scored_after.cols_eq(&expected_after, true, &[1, text_col_id]));

    let score_col_before = scored_before.get_column(crate::table::fts::VIRTUAL_SCORE_COLUMN_ID);
    for row in 0..scored_before.length() {
        let score =
            f64::from_le_bytes(score_col_before.get_value(row).unwrap().try_into().unwrap());
        assert!(score > 0.0, "scores must be positive for matched rows");
    }
    let score_col_after = scored_after.get_column(crate::table::fts::VIRTUAL_SCORE_COLUMN_ID);
    for row in 0..scored_after.length() {
        let score = f64::from_le_bytes(score_col_after.get_value(row).unwrap().try_into().unwrap());
        assert!(score > 0.0, "scores must be positive for matched rows");
    }

    // ========================
    // NoScore
    // ========================
    let noscore_query = make_unscored_query(&PlainFtsQueryInfo {
        query: "alpha".to_string(),
        top_k: 0,
        column_id: text_col_id,
        index_id,
        ..Default::default()
    });
    let noscore_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column_not_null(text_col_id, FieldTypeTp::String)
        .schema();
    crate::table::fts::validate_schema(&noscore_schema, &noscore_query).unwrap();
    let read_noscore = |read_ts: u64| -> crate::table::columnar::Block {
        let mut reader = snap
            .new_fts_reader(
                &runtime,
                table_id,
                noscore_query.info(),
                noscore_schema.clone(),
                read_ts,
                Some(start_bytes.clone()),
                end_bytes.clone(),
            )
            .unwrap();
        block_on(async {
            reader.set_int_handle_range(start, end).await.unwrap();
            reader.read_all().await
        })
    };

    // MVCC / visibility assertions for NoScore.
    let mut noscore_before = read_noscore(read_ts_before);
    noscore_before.sort();
    let expected_before = new_block::<IntPk>(&noscore_schema, |row| {
        row(1001, 50, false, |datum| {
            datum(11);
            datum("alpha row 1001");
        });
        row(1002, 40, false, |datum| {
            datum(22);
            datum("alpha row 1002 v40");
        });
        row(1003, 40, false, |datum| {
            datum(33);
            datum("alpha row 1003 v40");
        });
        row(1004, 55, false, |datum| {
            datum(14);
            datum("alpha mem 1004");
        });
        row(1005, 40, false, |datum| {
            datum(55);
            datum("alpha row 1005 v40");
        });
        row(1006, 52, false, |datum| {
            datum(66);
            datum("alpha col untracked");
        });
        row(1007, 53, false, |datum| {
            datum(77);
            datum("alpha unconverted l0 1007");
        });
    });
    assert!(noscore_before.eq(&expected_before));

    let mut noscore_after = read_noscore(read_ts_after);
    noscore_after.sort();
    let expected_after = new_block::<IntPk>(&noscore_schema, |row| {
        row(1001, 50, false, |datum| {
            datum(11);
            datum("alpha row 1001");
        });
        row(1004, 55, false, |datum| {
            datum(14);
            datum("alpha mem 1004");
        });
        row(1006, 52, false, |datum| {
            datum(66);
            datum("alpha col untracked");
        });
        row(1007, 53, false, |datum| {
            datum(77);
            datum("alpha unconverted l0 1007");
        });
    });
    assert!(noscore_after.eq(&expected_after));

    // Verify DeltaCache was actually exercised (or not) for immutable sources.
    if delta_cache_enabled {
        assert!(
            fts_delta_cache
                .get(FtsDeltaKind::RowL0, unconv_l0_id, table_id, index_id)
                .is_some()
        );
        assert!(
            fts_delta_cache
                .get(
                    FtsDeltaKind::ColumnarL0,
                    untracked_col_file_id,
                    table_id,
                    index_id
                )
                .is_some()
        );
    } else {
        assert!(
            fts_delta_cache
                .get(FtsDeltaKind::RowL0, unconv_l0_id, table_id, index_id)
                .is_none()
        );
        assert!(
            fts_delta_cache
                .get(
                    FtsDeltaKind::ColumnarL0,
                    untracked_col_file_id,
                    table_id,
                    index_id
                )
                .is_none()
        );
    }
}

#[test]
fn test_fts_reader_multi_column_match_expr_conditional_fallback() {
    ::test_util::init_log_for_test();

    fn standard_index_field(column_id: i64) -> FullTextIndexFieldDef {
        let mut field = FullTextIndexFieldDef::new();
        field.set_col_id(column_id);
        let option = field.mut_index_field_option();
        option.set_analyzer_type(FullTextAnalyzerType::FtsAnalyzerStandardV1);
        option.set_enable_bm25(true);
        field
    }

    fn match_word(column_id: i64, query_text: &str) -> tipb::Expr {
        let mut expr = tipb::Expr::new();
        expr.set_tp(tipb::ExprType::ScalarFunc);
        expr.set_sig(tipb::ScalarFuncSig::FtsMatchWord);
        expr.set_val(
            serde_json::to_vec(&serde_json::json!({
                "version": FTS_MATCH_EXPR_METADATA_VERSION,
                "score_mode": "inherit",
                "query": query_text,
                "fields": [{
                    "col_id": column_id,
                    "index_field": "STANDARD_V1",
                }],
            }))
            .unwrap(),
        );
        expr
    }

    let table_id = 161;
    let index_id = 3;
    let payload_column_id = 1;
    let first_source_column_id = 2;
    let second_source_column_id = 3;

    let index_fields = vec![
        standard_index_field(first_source_column_id),
        standard_index_field(second_source_column_id),
    ];
    let mut index_def = FullTextIndexDef::new();
    index_def.set_index_id(index_id);
    index_def.set_index_fields(index_fields.clone().into());
    let table_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(payload_column_id, FieldTypeTp::LongLong)
        .column(first_source_column_id, FieldTypeTp::String)
        .column(second_source_column_id, FieldTypeTp::String)
        .fts_index(index_def)
        .schema();

    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, KEYSPACE_ID, table_id);
    let shard = engine.get_shard(shard_id).unwrap();
    let schema_file = put_schema_file(&engine, id(), &[&table_schema]);

    // These rows are covered by the persisted multi-field FTS index.
    let indexed_columnar = put_columnar(
        &engine,
        new_columnar(id(), 100).table::<IntPk>(&table_schema, |row| {
            row(1, 50, false, |datum| {
                datum(11);
                datum("rust language");
                datum("search engine");
            });
            row(2, 50, false, |datum| {
                datum(22);
                datum("rust language");
                datum("search engine");
            });
        }),
    );
    let tracked_meta = new_tracked_index_meta(
        table_id,
        index_id,
        index_fields,
        FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString,
    );
    let tracked_layout =
        tracked_index_layout_from_tracked_meta(&tracked_meta, &table_schema.columns).unwrap();
    let index_layouts = std::collections::HashMap::from([((table_id, index_id), tracked_layout)]);
    let packed_outputs = block_on(columnar_to_fts_l0(ColumnarToFtsL0Opts {
        columnar_files: std::slice::from_ref(&indexed_columnar),
        schema_file: schema_file.clone(),
        index_pairs: &[(table_id, index_id)],
        index_layouts: &index_layouts,
        encryption_key: None,
        snap_version: 100.into(),
        l0_file_max_size: 0,
    }))
    .unwrap();
    assert_eq!(packed_outputs.len(), 1);
    let indexed_fts = PackedFile::from_buffer(&packed_outputs[0].data).unwrap();

    // Row storage supplies projected columns for indexed hits during the final
    // join. Row 1 is superseded below by a newer unindexed non-match.
    let row_l0 = new_write_batch()
        .table::<IntPk>(&table_schema, |row| {
            row.v2(1, 50, |datum| {
                datum(11);
                datum("rust language");
                datum("search engine");
            });
            row.v2(2, 50, |datum| {
                datum(22);
                datum("rust language");
                datum("search engine");
            });
        })
        .finish_as_sst_l0(id());
    let mem = new_write_batch()
        .table::<IntPk>(&table_schema, |row| {
            row.v2(1, 60, |datum| {
                datum(111);
                datum("rust language");
                datum("database");
            });
            row.v2(3, 60, |datum| {
                datum(33);
                datum("rust language");
                datum("search engine");
            });
            row.v2(4, 60, |datum| {
                datum(44);
                datum("go language");
                datum("search engine");
            });
        })
        .finish_as_cf();

    let mut columnar_levels = ColumnarLevels::new();
    columnar_levels.add_file(0, indexed_columnar);
    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index_meta(table_id, index_id, tracked_meta);
    fts_levels.mut_l0(|files| files.push(indexed_fts));

    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_l0_tbls(vec![row_l0]);
    builder.set_mem_tbls(vec![mem]);
    builder.set_columnar_levels(columnar_levels);
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());

    let mut info = PlainFtsQueryInfo {
        column_id: first_source_column_id,
        index_id,
        top_k: 0,
        ..Default::default()
    }
    .as_query_info();
    info.set_query_type(tipb::FtsQueryType::FtsQueryTypeWithScore);
    info.mut_match_expr()
        .push(match_word(first_source_column_id, "rust"));
    info.mut_match_expr()
        .push(match_word(second_source_column_id, "search"));
    // The TableScan projection intentionally omits both source columns. The
    // conditional fallback must still read them, merge indexed and unindexed
    // rows, apply MVCC, and join the projected payload column.
    let output_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(payload_column_id, FieldTypeTp::LongLong)
        .fts_score_column(false)
        .schema();
    let snap = SnapAccess::new(&shard);
    let runtime = engine.fs.get_runtime().handle().clone();
    let mut reader = snap
        .new_fts_reader(
            &runtime,
            table_id,
            &info,
            output_schema.clone(),
            70,
            None,
            None,
        )
        .unwrap();
    let mut output = block_on(async {
        reader.set_unbounded_handle_range().await.unwrap();
        reader.read_all().await
    });
    output.sort();

    let expected = new_block::<IntPk>(&output_schema, |row| {
        row(2, 50, false, |datum| {
            datum(22);
            datum(0.0);
        });
        row(3, 60, false, |datum| {
            datum(33);
            datum(0.0);
        });
    });
    assert!(output.cols_eq(&expected, true, &[payload_column_id]));

    let score_column = output.get_column(crate::table::fts::VIRTUAL_SCORE_COLUMN_ID);
    let indexed_score = f64::from_le_bytes(score_column.get_value(0).unwrap().try_into().unwrap());
    let fallback_score = f64::from_le_bytes(score_column.get_value(1).unwrap().try_into().unwrap());
    assert!(indexed_score > 0.0);
    assert_eq!(indexed_score, fallback_score);
}

#[test]
fn test_fts_index_ready_tracked_index() {
    ::test_util::init_log_for_test();
    let keyspace_id = KEYSPACE_ID;
    let table_id = 70;
    let index_id = 3;
    let missing_index_id = index_id + 1;
    let (engine, apply_tx) =
        new_test_engine_opt_with_custom_options(true, DEF_BLOCK_SIZE, "", |opts| {
            opts.set_build_fts_index(true);
        });
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard_ver = engine.get_shard(shard_id).unwrap().ver;
    let split_key = [keyspace_prefix(keyspace_id), encode_row_key(table_id, 50)].concat();
    let mut splitter = Splitter::new(
        vec![split_key],
        IdVer::new(shard_id, shard_ver),
        100,
        apply_tx,
    );
    let handle = std::thread::spawn(move || {
        splitter.run();
    });
    let ok = try_wait(|| engine.shards.len() == 6, 5);
    assert!(
        ok,
        "fts index ready split failed: expected 6 shards, got {}",
        engine.shards.len()
    );
    handle.join().unwrap();
    let shard_high = engine.get_shard(shard_id).unwrap();
    let shard_low_id = 101;
    let shard_low = engine.get_shard(shard_low_id).unwrap();

    // Expect false before schema is installed.
    assert!(!shard_high.fts_index_ready(table_id, index_id));

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    let install_shard = |shard: &Arc<crate::Shard>, fts_levels: FtsLevels| {
        let mut builder = ShardDataBuilder::new(shard.get_data());
        builder.set_schema(schema_file.get_version(), 0, Some(schema_file.clone()));
        builder.set_columnar_table_ids(vec![table_id]);
        builder.set_fts_levels(fts_levels);
        shard.set_data(builder.build());
    };

    install_shard(&shard_high, FtsLevels::default());
    install_shard(&shard_low, FtsLevels::default());

    let status = engine.collect_columnar_status(
        keyspace_id,
        table_id,
        Some(index_id),
        vec![shard_id, shard_low_id],
    );
    assert_eq!(status.total, 2);
    assert_eq!(status.ready, 2);
    assert_eq!(status.vector_index_ready, 0);
    assert_eq!(status.fts_index_ready, 0);

    let mut stale_fts_levels = FtsLevels::default();
    stale_fts_levels.track_index(table_id, missing_index_id);
    install_shard(&shard_high, stale_fts_levels.clone());
    install_shard(&shard_low, stale_fts_levels);
    assert!(!shard_high.fts_index_ready(table_id, missing_index_id));

    let status = engine.collect_columnar_status(
        keyspace_id,
        table_id,
        Some(missing_index_id),
        vec![shard_id, shard_low_id],
    );
    assert_eq!(status.total, 2);
    assert_eq!(status.ready, 2);
    assert_eq!(status.vector_index_ready, 0);
    assert_eq!(status.fts_index_ready, 0);

    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index(table_id, index_id);

    install_shard(&shard_high, fts_levels.clone());
    install_shard(&shard_low, fts_levels);

    let status = engine.collect_columnar_status(
        keyspace_id,
        table_id,
        Some(index_id),
        vec![shard_id, shard_low_id],
    );
    assert_eq!(status.total, 2);
    assert_eq!(status.ready, 2);
    assert_eq!(status.vector_index_ready, 0);
    assert_eq!(status.fts_index_ready, 2);
}

#[test]
fn test_fts_update_does_not_decrease_l0_snap_version() {
    ::test_util::init_log_for_test();
    let table_id = 60;
    let index_id = 1;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, KEYSPACE_ID, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    let mut levels = FtsLevels::default();
    levels.track_index(table_id, index_id);
    levels.l0_snap_version = 200.into();

    let base_data = shard.get_data();
    let mut builder = ShardDataBuilder::new(base_data);
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_fts_levels(levels);
    shard.set_data(builder.build());

    let current_seq = crate::shard::load_u64(&shard.meta_seq);
    let mut cs_pb = kvenginepb::ChangeSet::default();
    cs_pb.set_shard_id(shard_id);
    cs_pb.set_shard_ver(shard.ver);
    cs_pb.set_sequence(current_seq + 10);
    cs_pb.mut_fts_update().set_set_snap_version(100);
    let cs = ChangeSet::new(cs_pb);
    engine.apply_change_set(&cs).unwrap();

    assert_eq!(shard.get_data().fts_levels.l0_snap_version, 200.into());
}

#[test]
fn test_fts_incremental_update_priority_snap_version_step_back() {
    ::test_util::init_log_for_test();
    let table_id = 70;
    let index_id = 1;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, KEYSPACE_ID, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    let col_file_id = id();
    let col_file = put_columnar(
        &engine,
        new_columnar(col_file_id, 100).table::<IntPk>(&schema, |row| {
            row(1, 10, false, |datum| {
                datum(10);
                datum("pending doc");
            });
        }),
    );
    let mut col_levels = ColumnarLevels::new();
    col_levels.add_file(0, col_file);
    col_levels.sort();

    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index(table_id, index_id);
    // Make the columnar L0 look "covered" by watermark, but explicitly pending
    // (hole after merge).
    fts_levels.l0_snap_version = 200.into();
    fts_levels.mut_pending_columnar_l0_ids(|ids| ids.push(col_file_id));

    let base_data = shard.get_data();
    let mut builder = ShardDataBuilder::new(base_data);
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_columnar_levels(col_levels);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());

    let priority = shard.test_get_fts_incremental_update_priority().unwrap();
    match priority {
        CompactionPriority::FtsCreateL0 { snap_version, .. } => {
            assert_eq!(snap_version, 200.into());
        }
        other => panic!("unexpected priority: {:?}", other),
    }
}

#[test]
fn test_manual_columnar_major_compaction_is_deferred_by_fts_pending_l0() {
    ::test_util::init_log_for_test();
    let table_id = 80;
    let index_id = 1;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, KEYSPACE_ID, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(FullTextIndexDef {
            index_id,
            col_id: 2,
            parser_type: "STANDARD_V1".to_string(),
            ..Default::default()
        })
        .schema();
    let schema_file = put_schema_file(&engine, id(), &[&schema]);

    let col_file_id = id();
    let col_file = put_columnar(
        &engine,
        new_columnar(col_file_id, 100).table::<IntPk>(&schema, |row| {
            row(1, 10, false, |datum| {
                datum(10);
                datum("pending doc");
            });
        }),
    );
    let mut col_levels = ColumnarLevels::new();
    col_levels.add_file(0, col_file);
    col_levels.sort();

    let mut fts_levels = FtsLevels::default();
    fts_levels.track_index(table_id, index_id);
    fts_levels.l0_snap_version = 200.into();
    fts_levels.mut_pending_columnar_l0_ids(|ids| ids.push(col_file_id));

    let base_data = shard.get_data();
    let mut builder = ShardDataBuilder::new(base_data);
    builder.set_schema(schema_file.get_version(), 0, Some(schema_file));
    builder.set_columnar_table_ids(vec![table_id]);
    builder.set_columnar_levels(col_levels);
    builder.set_fts_levels(fts_levels);
    shard.set_data(builder.build());

    shard.pending_ops.write().unwrap().manual_major_compaction =
        MajorCompactionType::EnableColumnar;
    shard.refresh_states();
    let priority = shard.compaction_priority.read().unwrap().clone();
    assert!(
        matches!(priority, Some(CompactionPriority::FtsCreateL0 { .. })),
        "unexpected priority: {:?}",
        priority
    );

    {
        let data = shard.get_data();
        let mut new_levels = (*data.fts_levels).clone();
        new_levels.mut_pending_columnar_l0_ids(|ids| ids.clear());
        let mut builder = ShardDataBuilder::new(data);
        builder.set_fts_levels(new_levels);
        shard.set_data(builder.build());
    }

    shard.refresh_states();
    let priority = shard.compaction_priority.read().unwrap().clone();
    assert!(
        matches!(
            priority,
            Some(CompactionPriority::ColumnarMajor {
                is_manual: true,
                ..
            })
        ),
        "unexpected priority after clearing pending: {:?}",
        priority
    );
}

fn collect_packed_handles(file: &PackedFile, table_id: i64) -> Vec<i64> {
    block_on(async {
        let mut iter = file.lp_iter().unwrap();
        let mut handles = Vec::new();
        while let Some(lp) = iter.next_lp().await.unwrap() {
            let props = lp.props();
            if props.get_table_id() != table_id || !props.get_is_int_handle() {
                continue;
            }
            let lp_int = lp.as_int_lp().unwrap();
            let mut pk_iter = lp_int.pk_iter().unwrap();
            while let Some((_doc_id, pk, ..)) = pk_iter.next().await.unwrap() {
                handles.push(IntPk::decode(pk.as_ref()).unwrap());
            }
        }
        handles.sort_unstable();
        handles
    })
}

fn collect_dedicated_handles(file: &EDedicatedFile) -> Vec<i64> {
    block_on(async {
        let file = file.as_int().unwrap();
        let mut iter = file.pk_iter().unwrap();
        let mut handles = Vec::new();
        while let Some((_doc_id, pk, ..)) = iter.next().await.unwrap() {
            handles.push(IntPk::decode(pk.as_ref()).unwrap());
        }
        handles.sort_unstable();
        handles
    })
}

fn collect_dedicated_rows(file: &EDedicatedFile) -> Vec<(i64, u64, bool)> {
    block_on(async {
        let file = file.as_int().unwrap();
        let mut iter = file.pk_iter().unwrap();
        let mut rows = Vec::new();
        while let Some((_doc_id, pk, version, deleted)) = iter.next().await.unwrap() {
            rows.push((IntPk::decode(pk.as_ref()).unwrap(), version, deleted != 0));
        }
        rows
    })
}

fn put_schema_file(engine: &TestEngine, file_id: u64, schemas: &[&Schema]) -> SchemaFile {
    let schema_version = engine.id_allocator.alloc_id(1).unwrap()[0] as i64;
    let schema_vec: Vec<Schema> = schemas.iter().map(|schema| (*schema).clone()).collect();
    let schema_file_data = build_schema_file(KEYSPACE_ID, schema_version, schema_vec, 0);
    let schema_raw = Arc::new(InMemFile::new(file_id, Bytes::from(schema_file_data)));
    let fs = engine.fs.clone();
    fs.get_runtime()
        .block_on(fs.create(
            schema_raw.id(),
            schema_raw.read(0, schema_raw.size() as usize).unwrap(),
            dfs::Options::default().with_type(FileType::Schema),
        ))
        .unwrap();
    SchemaFile::open(schema_raw).unwrap()
}

fn put_columnar(engine: &TestEngine, builder: TestColumnarFileBuilder) -> ColumnarFile {
    let file = builder.finish_as_file();
    let backing = file.get_file();
    let bytes = backing.read(0, backing.size() as usize).unwrap();
    engine
        .fs
        .get_runtime()
        .block_on(engine.fs.clone().create(
            backing.id(),
            bytes.clone(),
            dfs::Options::default().with_type(FileType::Columnar),
        ))
        .unwrap();
    ColumnarFile::open(
        Arc::new(InMemFile::new(backing.id(), bytes)),
        None,
        ColumnarMetaCache::default(),
    )
    .unwrap()
}

fn put_packed(engine: &TestEngine, file_id: u64, builder: TestPackedFileBuilder) -> PackedFile {
    let file_bytes = builder.finish_as_bytes();
    engine
        .fs
        .get_runtime()
        .block_on(engine.fs.clone().create(
            file_id,
            file_bytes.clone(),
            dfs::Options::default().with_type(FileType::FtsPackedFile),
        ))
        .unwrap();
    PackedFile::new(
        Arc::new(InMemFile::new(file_id, file_bytes)),
        FtsCache::disabled(),
    )
    .unwrap()
}

fn put_ded(engine: &TestEngine, file_id: u64, builder: TestDedicatedFileBuilder) -> EDedicatedFile {
    let file_bytes = builder.finish_as_bytes();
    engine
        .fs
        .get_runtime()
        .block_on(engine.fs.clone().create(
            file_id,
            file_bytes.clone(),
            dfs::Options::default().with_type(FileType::FtsDedicatedFile),
        ))
        .unwrap();
    EDedicatedFile::new(
        Arc::new(InMemFile::new(file_id, file_bytes)),
        FtsCache::disabled(),
    )
    .unwrap()
}

fn search_packed(file: &PackedFile, table_id: i64, index_id: i64, term: &str) -> Vec<u32> {
    block_on(async {
        let mut iter = file.lp_iter().unwrap();
        while let Some(lp) = iter.next_lp().await.unwrap() {
            if lp.props().get_table_id() != table_id || lp.props().get_index_id() != index_id {
                continue;
            }
            if lp.props().get_n_pk() == 0 {
                return Vec::new();
            }
            if lp.props().get_tantivy_layout().get_meta().get_size() == 0 {
                return Vec::new();
            }
            let reader = lp.read_tantivy_index().unwrap();
            let mut results = Vec::new();
            let query = make_unscored_query(&PlainFtsQueryInfo {
                query: term.to_string(),
                ..Default::default()
            });
            reader.search(&query, &mut results).unwrap();
            return results.into_iter().map(|r| r.doc_id).collect();
        }
        Vec::new()
    })
}

#[test]
fn test_store_usage_stats_excludes_fts_from_row_size() {
    ::test_util::init_log_for_test();
    let keyspace_id = 1;
    let table_id = 41;
    let index_id = 1;
    let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, "");
    let id = || engine.id_allocator.alloc_id(1).unwrap()[0];
    let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id);
    let shard = engine.get_shard(shard_id).unwrap();

    let l0_id = id();
    let l0_file = put_packed(
        &engine,
        l0_id,
        new_packed(l0_id, 500).lp(table_id, index_id, |d| {
            d(8, 220, false, "store usage doc");
            d(11, 210, false, "store usage doc");
        }),
    );
    let fts_size = l0_file.file().size();

    let before = engine.collect_store_usage_stats();
    let mut levels = FtsLevels::default();
    levels.track_index(table_id, index_id);
    levels.mut_l0(|files| files.push(l0_file));
    let mut builder = ShardDataBuilder::new(shard.get_data());
    builder.set_fts_levels(levels);
    shard.set_data(builder.build());
    shard.set_active(true);
    shard.refresh_estimated_size_and_entries();

    let after = engine.collect_store_usage_stats();
    assert_eq!(
        after.row_size, before.row_size,
        "row_size must stay row-only when FTS files are added"
    );
    assert_eq!(after.leader_fts_size, before.leader_fts_size + fts_size);
    assert_eq!(after.leader_columnar_size, before.leader_columnar_size);
    assert_eq!(after.leader_vector_size, before.leader_vector_size);
    assert_eq!(after.leader_row_size, after.row_size);

    let engine_stats = engine.get_engine_stats(engine.get_all_shard_stats());
    assert_eq!(
        after.row_size
            + engine_stats
                .columnar_levels
                .iter()
                .map(|l| l.data_size)
                .sum::<u64>()
            + engine_stats
                .fts_levels
                .iter()
                .map(|l| l.data_size)
                .sum::<u64>()
            + engine_stats.vector_indexes.data_size,
        engine_stats.total_data_size
    );
    assert_eq!(
        after.leader_fts_size,
        engine_stats
            .fts_levels
            .iter()
            .map(|l| l.data_size)
            .sum::<u64>()
    );

    shard.set_active(false);
    let inactive = engine.collect_store_usage_stats();
    assert_eq!(inactive.row_size, after.row_size);
    assert_eq!(inactive.leader_row_size, 0);
    assert_eq!(inactive.leader_fts_size, 0);
    assert_eq!(
        inactive.row_size
            + engine_stats
                .columnar_levels
                .iter()
                .map(|l| l.data_size)
                .sum::<u64>()
            + engine_stats
                .fts_levels
                .iter()
                .map(|l| l.data_size)
                .sum::<u64>()
            + engine_stats.vector_indexes.data_size,
        engine_stats.total_data_size
    );
}

fn search_l2(file: &EDedicatedFile, term: &str) -> Vec<u32> {
    block_on(async {
        let dedicated = file.as_int().unwrap();
        let reader = dedicated.cached_read_index().await.unwrap();
        let mut results = Vec::new();
        let query = make_unscored_query(&PlainFtsQueryInfo {
            query: term.to_string(),
            ..Default::default()
        });
        reader.search(&query, &mut results).unwrap();
        results.into_iter().map(|r| r.doc_id).collect()
    })
}

fn outer_row_key(keyspace_id: u32, table_id: i64, handle: i64) -> Vec<u8> {
    let mut key = keyspace_prefix(keyspace_id);
    key.extend_from_slice(&encode_row_key(table_id, handle));
    key
}