// Copyright 2026 PingCAP, Inc.

use std::{convert::TryInto, sync::Mutex, time::Duration};

use clara_fts::test_util::{PlainFtsQueryInfo, make_scored_query, make_unscored_query};
use futures::executor::block_on;
use kvengine::{
    dfs,
    dfs::FileType,
    table::{
        columnar::Block,
        fts::{
            IntPk, VIRTUAL_SCORE_COLUMN_ID,
            test_util::{SchemaBuilder, new_block},
        },
        schema_file::build_schema_file,
    },
};
use kvenginepb::fts::{FtsFieldLayout, FullTextIndexDef};
use pd_client::PdClient;
use test_cloud_server::{ServerCluster, must_wait, oss::prepare_dfs};
use test_pd_client::PdClientExt;
use tidb_query_datatype::{
    FieldTypeTp,
    codec::row::v2::encoder_for_test::{Column, RowEncoder},
    expr::EvalContext,
};
use tikv_util::{codec::bytes::encode_bytes, store::find_peer};

use crate::{
    alloc_node_id, alloc_node_id_vec,
    columnar::{create_keyspace_and_split_tables, gen_row_key, send_schema_file_request},
};

fn encode_fts_row_val(ctx: &Mutex<EvalContext>, col1: i64, text: &str) -> Vec<u8> {
    let mut row_val = vec![];
    let cols = vec![
        Column::new(1, Some(col1)),
        Column::new(2, Some(text.as_bytes().to_vec())),
    ];
    let mut guard = ctx.lock().unwrap();
    row_val.write_row(&mut guard, cols).unwrap();
    row_val
}

#[test]
fn test_fts_indexed_and_unindexed_mvcc_end_to_end_scored_and_unscored() {
    let node_id = alloc_node_id();
    let (_temp_dir, mut oss, dfs_config) =
        prepare_dfs("test_fts_indexed_and_unindexed_mvcc_end_to_end");
    let mut cluster = ServerCluster::new(vec![node_id], |_, conf| {
        conf.kvengine
            .columnar_table_build_options
            .max_columnar_table_size = 1024;
        conf.kvengine
            .columnar_table_build_options
            .pack_max_row_count = 9;
        conf.kvengine.build_columnar = true;
        conf.kvengine.read_columnar = true;
        conf.kvengine.build_fts_index = true;
        // Keep thresholds small so compaction triggers quickly in tests.
        conf.kvengine.fts_build_options.max_l0_files = 1;
        conf.kvengine.fts_build_options.min_l2_lp_size = tikv_util::config::ReadableSize::kb(0);
        conf.dfs = dfs_config.clone();
    });

    let dfs = cluster.get_dfs().unwrap();
    let (keyspace_id, table_ids) = dfs
        .get_runtime()
        .block_on(create_keyspace_and_split_tables(&mut cluster));
    let table_id = table_ids[1];
    let index_id = 1i64;
    let text_col_id = 2i64;

    let table_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(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_version = 10;
    let schema_file_data =
        build_schema_file(keyspace_id, schema_version, vec![table_schema.clone()], 0);
    let schema_file_id = 100;
    let opts = dfs::Options::default().with_type(FileType::Schema);
    dfs.get_runtime()
        .block_on(dfs.create(schema_file_id, schema_file_data.into(), opts))
        .unwrap();
    let status_addr = cluster.status_addr(node_id);

    let pd_client = cluster.get_pd_client();
    let region = pd_client
        .get_region(&encode_bytes(&gen_row_key(keyspace_id, table_id, 0)))
        .unwrap();
    let shard_id = region.id;

    let kvengine = cluster.get_kvengine(node_id);
    must_wait(
        || {
            dfs.get_runtime().block_on(send_schema_file_request(
                &status_addr,
                keyspace_id,
                schema_file_id,
            ));
            let Some(shard) = kvengine.get_shard(shard_id) else {
                return false;
            };
            shard.get_schema_file().is_some()
        },
        10,
        || "failed to sync schema file".to_string(),
    );

    let mut client = cluster.new_client();
    let eval_ctx = Mutex::new(EvalContext::default());

    // ======================================================
    // Stage 1: insert rows and flush to build a real FTS index
    // ======================================================
    // Query term: "alpha"
    // Visible at ts_before: 1, 2, 4
    // Visible at ts_after: 4 (updated), 6 (new)
    client.put_kv(
        1..6,
        |i: usize| gen_row_key(keyspace_id, table_id, i),
        |i| match i {
            1 => encode_fts_row_val(&eval_ctx, 101, "alpha alpha"),
            2 => encode_fts_row_val(&eval_ctx, 102, "alpha beta"),
            3 => encode_fts_row_val(&eval_ctx, 103, "beta"),
            4 => encode_fts_row_val(&eval_ctx, 104, "alpha"),
            5 => encode_fts_row_val(&eval_ctx, 105, "gamma"),
            _ => unreachable!(),
        },
    );

    cluster.flush_memtable(shard_id).unwrap();
    assert!(
        cluster.wait_for_memtable_flushed(shard_id, Duration::from_secs(10)),
        "memtable flush timeout"
    );

    must_wait(
        || {
            let Some(shard) = kvengine.get_shard(shard_id) else {
                return false;
            };
            shard.get_columnar_table_ids().contains(&table_id)
        },
        10,
        || "failed to build columnar files".to_string(),
    );

    must_wait(
        || {
            let Some(shard) = kvengine.get_shard(shard_id) else {
                return false;
            };
            !shard.get_all_fts_files().is_empty()
        },
        20,
        || "failed to build FTS index files".to_string(),
    );

    let ts_before = client.get_ts().into_inner();

    // ============================================================
    // Stage 2: insert/update/delete WITHOUT flushing (unindexed part)
    // ============================================================
    client.put_kv(
        vec![2usize, 4usize, 6usize],
        |i: usize| gen_row_key(keyspace_id, table_id, i),
        |i| match i {
            // Indexed hit at pk=2 becomes non-matching in the unindexed part.
            2 => encode_fts_row_val(&eval_ctx, 202, "beta"),
            // Indexed hit at pk=4 stays matching but changes content.
            4 => encode_fts_row_val(&eval_ctx, 404, "alpha updated"),
            // New unindexed row.
            6 => encode_fts_row_val(&eval_ctx, 106, "alpha new"),
            _ => unreachable!(),
        },
    );
    client.del_kv(1..2, |i: usize| gen_row_key(keyspace_id, table_id, i));

    let shard_stats = kvengine.get_shard_stat(shard_id);
    assert!(
        !shard_stats.mem_table_is_empty(),
        "expected unflushed memtable (unindexed part), got empty memtable: {:?}",
        shard_stats
    );

    let ts_after = client.get_ts().into_inner();

    // ============================================================
    // Query helpers
    // ============================================================
    let runtime = dfs.get_runtime().handle().clone();
    let scored_query = make_scored_query(&PlainFtsQueryInfo {
        query: "alpha".to_string(),
        top_k: 10,
        column_id: text_col_id,
        index_id,
        ..Default::default()
    });
    assert_eq!(
        scored_query.info().get_query_type(),
        tipb::FtsQueryType::FtsQueryTypeWithScore
    );

    let scored_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(text_col_id, FieldTypeTp::String)
        .fts_score_column(false)
        .schema();
    kvengine::table::fts::validate_schema(&scored_schema, &scored_query).unwrap();

    let read_scored = |read_ts: u64| -> Block {
        let shard = kvengine.get_shard(shard_id).unwrap();
        let snap = shard.new_snap_access();
        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_int_handle_range(0, Some(10)).await.unwrap();
            reader.read_all().await
        })
    };

    let noscore_query = make_unscored_query(&PlainFtsQueryInfo {
        query: "alpha".to_string(),
        top_k: 0,
        column_id: text_col_id,
        index_id,
        ..Default::default()
    });
    assert_eq!(
        noscore_query.info().get_query_type(),
        tipb::FtsQueryType::FtsQueryTypeNoScore
    );

    let noscore_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(text_col_id, FieldTypeTp::String)
        .schema();
    kvengine::table::fts::validate_schema(&noscore_schema, &noscore_query).unwrap();

    let read_noscore = |read_ts: u64| -> Block {
        let shard = kvengine.get_shard(shard_id).unwrap();
        let snap = shard.new_snap_access();
        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_int_handle_range(0, Some(10)).await.unwrap();
            reader.read_all().await
        })
    };

    // ============================================================
    // Scored query
    // ============================================================
    let mut scored_before = read_scored(ts_before);
    scored_before.sort();
    let expected_scored_before = new_block::<IntPk>(&scored_schema, |row| {
        row(1, 0, false, |datum| {
            datum(101);
            datum("alpha alpha");
            datum(0.0);
        });
        row(2, 0, false, |datum| {
            datum(102);
            datum("alpha beta");
            datum(0.0);
        });
        row(4, 0, false, |datum| {
            datum(104);
            datum("alpha");
            datum(0.0);
        });
    });
    assert_eq!(
        (0..scored_before.length())
            .map(|row| scored_before.get_handle_buf().get_int_handle_value(row))
            .collect::<Vec<_>>(),
        vec![1, 2, 4]
    );
    assert!(scored_before.cols_eq(&expected_scored_before, false, &[1, text_col_id]));
    let score_col = scored_before.get_column(VIRTUAL_SCORE_COLUMN_ID);
    for row in 0..scored_before.length() {
        let score = f64::from_le_bytes(score_col.get_not_null_value(row).try_into().unwrap());
        assert!(score > 0.0, "scores must be positive for matched rows");
    }

    let mut scored_after = read_scored(ts_after);
    scored_after.sort();
    let expected_scored_after = new_block::<IntPk>(&scored_schema, |row| {
        row(4, 0, false, |datum| {
            datum(404);
            datum("alpha updated");
            datum(0.0);
        });
        row(6, 0, false, |datum| {
            datum(106);
            datum("alpha new");
            datum(0.0);
        });
    });
    assert_eq!(
        (0..scored_after.length())
            .map(|row| scored_after.get_handle_buf().get_int_handle_value(row))
            .collect::<Vec<_>>(),
        vec![4, 6]
    );
    assert!(scored_after.cols_eq(&expected_scored_after, false, &[1, text_col_id]));
    let score_col = scored_after.get_column(VIRTUAL_SCORE_COLUMN_ID);
    for row in 0..scored_after.length() {
        let score = f64::from_le_bytes(score_col.get_not_null_value(row).try_into().unwrap());
        assert!(score > 0.0, "scores must be positive for matched rows");
    }

    // ============================================================
    // Unscored query
    // ============================================================
    let mut noscore_before = read_noscore(ts_before);
    noscore_before.sort();
    let expected_noscore_before = new_block::<IntPk>(&noscore_schema, |row| {
        row(1, 0, false, |datum| {
            datum(101);
            datum("alpha alpha");
        });
        row(2, 0, false, |datum| {
            datum(102);
            datum("alpha beta");
        });
        row(4, 0, false, |datum| {
            datum(104);
            datum("alpha");
        });
    });
    assert_eq!(
        (0..noscore_before.length())
            .map(|row| noscore_before.get_handle_buf().get_int_handle_value(row))
            .collect::<Vec<_>>(),
        vec![1, 2, 4]
    );
    assert!(noscore_before.cols_eq(&expected_noscore_before, false, &[1, text_col_id]));

    let mut noscore_after = read_noscore(ts_after);
    noscore_after.sort();
    let expected_noscore_after = new_block::<IntPk>(&noscore_schema, |row| {
        row(4, 0, false, |datum| {
            datum(404);
            datum("alpha updated");
        });
        row(6, 0, false, |datum| {
            datum(106);
            datum("alpha new");
        });
    });
    assert_eq!(
        (0..noscore_after.length())
            .map(|row| noscore_after.get_handle_buf().get_int_handle_value(row))
            .collect::<Vec<_>>(),
        vec![4, 6]
    );
    assert!(noscore_after.cols_eq(&expected_noscore_after, false, &[1, text_col_id]));

    cluster.stop();
    oss.shutdown();
}

fn tracked_fts_field_layout(
    engine: &kvengine::Engine,
    shard_id: u64,
    table_id: i64,
    index_id: i64,
) -> Option<FtsFieldLayout> {
    engine
        .get_shard(shard_id)?
        .get_tracked_fts_field_layout(table_id, index_id)
}

fn publish_fts_schema(
    dfs: &dyn kvengine::dfs::Dfs,
    keyspace_id: u32,
    table_id: i64,
    index_id: i64,
    schema_file_id: u64,
    schema_version: i64,
    layout: FtsFieldLayout,
) {
    let mut fulltext_index = FullTextIndexDef {
        index_id,
        col_id: 2,
        parser_type: "STANDARD_V1".to_string(),
        ..Default::default()
    };
    fulltext_index.set_initial_field_layout(layout);
    let table_schema = SchemaBuilder::<IntPk>::new(table_id)
        .column(1, FieldTypeTp::LongLong)
        .column(2, FieldTypeTp::String)
        .fts_index(fulltext_index)
        .schema();
    let schema_file_data = build_schema_file(keyspace_id, schema_version, vec![table_schema], 0);
    dfs.get_runtime()
        .block_on(dfs.create(
            schema_file_id,
            schema_file_data.into(),
            dfs::Options::default().with_type(FileType::Schema),
        ))
        .unwrap();
}

fn assert_check_merge_rejects_inconsistent_fts(
    engine: &kvengine::Engine,
    left_region_id: u64,
    right_region_id: u64,
) {
    let left_shard = engine
        .get_shard(left_region_id)
        .expect("left shard missing during check_merge");
    let right_shard = engine
        .get_shard(right_region_id)
        .expect("right shard missing during check_merge");
    match engine.check_merge(
        left_region_id,
        left_shard.ver,
        right_region_id,
        right_shard.ver,
    ) {
        Err(kvengine::Error::CheckMerge(msg)) => {
            assert!(
                msg.contains("inconsistent FTS indexes"),
                "unexpected check_merge error: {}",
                msg
            );
        }
        other => panic!(
            "mixed FTS field layouts should fail check_merge, got {:?}",
            other
        ),
    }
}

/// Region leaders on different stores must derive the same tracked layout from
/// the same schema. After tracking completes, a leader transfer must keep that
/// layout so the regions remain merge-compatible.
#[test]
fn test_schema_initial_layout_is_consistent_across_region_leaders() {
    let node_ids = alloc_node_id_vec(3);
    let left_node = node_ids[0];
    let right_node = node_ids[1];
    let (_temp_dir, mut oss, dfs_config) =
        prepare_dfs("test_schema_initial_layout_is_consistent_across_region_leaders");
    let mut cluster = ServerCluster::new(node_ids.clone(), |_, conf| {
        conf.kvengine
            .columnar_table_build_options
            .max_columnar_table_size = 1024;
        conf.kvengine
            .columnar_table_build_options
            .pack_max_row_count = 9;
        conf.kvengine.build_columnar = true;
        conf.kvengine.read_columnar = true;
        conf.kvengine.build_fts_index = true;
        conf.kvengine.fts_build_options.max_l0_files = 1;
        conf.dfs = dfs_config.clone();
    });

    cluster.wait_region_replicated(&[], 3);

    let dfs = cluster.get_dfs().unwrap();
    let pd_client = cluster.get_pd_client();
    let (keyspace_id, table_ids) = dfs
        .get_runtime()
        .block_on(create_keyspace_and_split_tables(&mut cluster));
    let table_id = table_ids[1];
    let index_id = 7i64;
    let left_key = gen_row_key(keyspace_id, table_id, 0);
    let split_key = gen_row_key(keyspace_id, table_id, 100);

    cluster.wait_region_replicated(&left_key, 3);
    pd_client.disable_default_operator();

    let mut client = cluster.new_client();
    let regions_count = pd_client.get_regions_number();
    client.split(&split_key);
    cluster.wait_pd_region_count(regions_count + 1);
    cluster.wait_region_replicated(&left_key, 3);
    cluster.wait_region_replicated(&split_key, 3);

    let left_region = pd_client.get_region(&encode_bytes(&left_key)).unwrap();
    let right_region = pd_client.get_region(&encode_bytes(&split_key)).unwrap();
    let left_peer = find_peer(&left_region, cluster.get_store_id(left_node))
        .expect("left region missing peer on left store")
        .clone();
    let right_peer = find_peer(&right_region, cluster.get_store_id(right_node))
        .expect("right region missing peer on right store")
        .clone();
    cluster.transfer_leader(left_region.id, left_peer);
    cluster.transfer_leader(right_region.id, right_peer);

    let left_status = cluster.status_addr(left_node);
    let right_status = cluster.status_addr(right_node);
    let schema_file_id = 201;
    publish_fts_schema(
        dfs.as_ref(),
        keyspace_id,
        table_id,
        index_id,
        schema_file_id,
        10,
        FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString,
    );

    // End-to-end schema request (CREATE INDEX equivalent) on both stores.
    let left_engine = cluster.get_kvengine(left_node);
    let right_engine = cluster.get_kvengine(right_node);
    must_wait(
        || {
            dfs.get_runtime().block_on(send_schema_file_request(
                &left_status,
                keyspace_id,
                schema_file_id,
            ));
            dfs.get_runtime().block_on(send_schema_file_request(
                &right_status,
                keyspace_id,
                schema_file_id,
            ));
            left_engine
                .get_shard(left_region.id)
                .is_some_and(|shard| shard.get_schema_file().is_some())
                && right_engine
                    .get_shard(right_region.id)
                    .is_some_and(|shard| shard.get_schema_file().is_some())
        },
        15,
        || "failed to sync FTS schema to both region leaders".to_string(),
    );

    let eval_ctx = Mutex::new(EvalContext::default());
    client.put_kv(
        0..50,
        |i| gen_row_key(keyspace_id, table_id, i),
        |i| encode_fts_row_val(&eval_ctx, i as i64, "leftalpha"),
    );
    client.put_kv(
        100..150,
        |i| gen_row_key(keyspace_id, table_id, i),
        |i| encode_fts_row_val(&eval_ctx, i as i64, "rightalpha"),
    );
    cluster.flush_memtable(left_region.id).unwrap();
    cluster.flush_memtable(right_region.id).unwrap();

    must_wait(
        || {
            tracked_fts_field_layout(&left_engine, left_region.id, table_id, index_id)
                == Some(FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString)
                && tracked_fts_field_layout(&right_engine, right_region.id, table_id, index_id)
                    == Some(FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString)
        },
        30,
        || {
            format!(
                "ordinal layout not recorded consistently: left={:?} right={:?}",
                tracked_fts_field_layout(&left_engine, left_region.id, table_id, index_id),
                tracked_fts_field_layout(&right_engine, right_region.id, table_id, index_id)
            )
        },
    );

    // Tracking already finished. Move the left region onto the third store and
    // confirm the new leader still has the recorded ordinal layout.
    let third_node = node_ids[2];
    let left_peer_on_third = find_peer(&left_region, cluster.get_store_id(third_node))
        .expect("left region missing peer on third store")
        .clone();
    cluster.transfer_leader(left_region.id, left_peer_on_third);
    let third_engine = cluster.get_kvengine(third_node);
    must_wait(
        || {
            tracked_fts_field_layout(&third_engine, left_region.id, table_id, index_id)
                == Some(FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString)
        },
        15,
        || {
            format!(
                "ordinal layout missing after leader transfer: {:?}",
                tracked_fts_field_layout(&third_engine, left_region.id, table_id, index_id)
            )
        },
    );

    // Identical schema-selected layouts keep the regions merge-compatible.
    assert!(
        client.try_merge_and_wait(&left_key, &split_key, 8),
        "regions with identical FTS field layouts must merge"
    );
    assert_eq!(
        pd_client.get_region(&encode_bytes(&left_key)).unwrap().id,
        pd_client.get_region(&encode_bytes(&split_key)).unwrap().id
    );

    cluster.stop();
    oss.shutdown();
}

/// Incompatible tracked FTS layouts on adjacent regions must fail
/// `check_merge`. The two layouts are installed through the schema-request path
/// so the guard is exercised on real shard metadata (interrupted upgrade or
/// restore). TiDB publishes one schema file; this test only constructs the
/// incompatible tracked state.
#[test]
fn test_mixed_schema_layout_request_blocks_region_merge() {
    let node_ids = alloc_node_id_vec(3);
    let legacy_node = node_ids[0];
    let ordinal_node = node_ids[1];
    let (_temp_dir, mut oss, dfs_config) =
        prepare_dfs("test_mixed_schema_layout_request_blocks_region_merge");
    let mut cluster = ServerCluster::new(node_ids.clone(), |_, conf| {
        conf.kvengine
            .columnar_table_build_options
            .max_columnar_table_size = 1024;
        conf.kvengine
            .columnar_table_build_options
            .pack_max_row_count = 9;
        conf.kvengine.build_columnar = true;
        conf.kvengine.read_columnar = true;
        conf.kvengine.build_fts_index = true;
        conf.kvengine.fts_build_options.max_l0_files = 1;
        conf.dfs = dfs_config.clone();
    });

    cluster.wait_region_replicated(&[], 3);

    let dfs = cluster.get_dfs().unwrap();
    let pd_client = cluster.get_pd_client();
    let (keyspace_id, table_ids) = dfs
        .get_runtime()
        .block_on(create_keyspace_and_split_tables(&mut cluster));
    let table_id = table_ids[1];
    let index_id = 7i64;
    let left_key = gen_row_key(keyspace_id, table_id, 0);
    let split_key = gen_row_key(keyspace_id, table_id, 100);

    cluster.wait_region_replicated(&left_key, 3);
    pd_client.disable_default_operator();

    let mut client = cluster.new_client();
    let regions_count = pd_client.get_regions_number();
    client.split(&split_key);
    cluster.wait_pd_region_count(regions_count + 1);
    cluster.wait_region_replicated(&left_key, 3);
    cluster.wait_region_replicated(&split_key, 3);

    let left_region = pd_client.get_region(&encode_bytes(&left_key)).unwrap();
    let right_region = pd_client.get_region(&encode_bytes(&split_key)).unwrap();
    let left_peer = find_peer(&left_region, cluster.get_store_id(legacy_node))
        .expect("left region missing peer on legacy store")
        .clone();
    let right_peer = find_peer(&right_region, cluster.get_store_id(ordinal_node))
        .expect("right region missing peer on ordinal store")
        .clone();
    cluster.transfer_leader(left_region.id, left_peer);
    cluster.transfer_leader(right_region.id, right_peer);

    let legacy_status = cluster.status_addr(legacy_node);
    let ordinal_status = cluster.status_addr(ordinal_node);
    let legacy_schema_file_id = 301;
    let ordinal_schema_file_id = 302;
    publish_fts_schema(
        dfs.as_ref(),
        keyspace_id,
        table_id,
        index_id,
        legacy_schema_file_id,
        10,
        FtsFieldLayout::FtsFieldLayoutLegacyBody,
    );
    publish_fts_schema(
        dfs.as_ref(),
        keyspace_id,
        table_id,
        index_id,
        ordinal_schema_file_id,
        11,
        FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString,
    );

    // End-to-end schema requests: each leader loads a different layout.
    let left_engine = cluster.get_kvengine(legacy_node);
    let right_engine = cluster.get_kvengine(ordinal_node);
    must_wait(
        || {
            dfs.get_runtime().block_on(send_schema_file_request(
                &legacy_status,
                keyspace_id,
                legacy_schema_file_id,
            ));
            dfs.get_runtime().block_on(send_schema_file_request(
                &ordinal_status,
                keyspace_id,
                ordinal_schema_file_id,
            ));
            left_engine
                .get_shard(left_region.id)
                .is_some_and(|shard| shard.get_schema_file().is_some())
                && right_engine
                    .get_shard(right_region.id)
                    .is_some_and(|shard| shard.get_schema_file().is_some())
        },
        15,
        || "failed to sync mixed FTS schemas to region leaders".to_string(),
    );

    let eval_ctx = Mutex::new(EvalContext::default());
    client.put_kv(
        0..50,
        |i| gen_row_key(keyspace_id, table_id, i),
        |i| encode_fts_row_val(&eval_ctx, i as i64, "legacyalpha"),
    );
    client.put_kv(
        100..150,
        |i| gen_row_key(keyspace_id, table_id, i),
        |i| encode_fts_row_val(&eval_ctx, i as i64, "ordinalalpha"),
    );
    cluster.flush_memtable(left_region.id).unwrap();
    cluster.flush_memtable(right_region.id).unwrap();

    must_wait(
        || {
            tracked_fts_field_layout(&left_engine, left_region.id, table_id, index_id)
                == Some(FtsFieldLayout::FtsFieldLayoutLegacyBody)
                && tracked_fts_field_layout(&right_engine, right_region.id, table_id, index_id)
                    == Some(FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString)
                && tracked_fts_field_layout(&left_engine, right_region.id, table_id, index_id)
                    == Some(FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString)
                && tracked_fts_field_layout(&right_engine, left_region.id, table_id, index_id)
                    == Some(FtsFieldLayout::FtsFieldLayoutLegacyBody)
        },
        30,
        || {
            format!(
                "mixed layouts not recorded: left-leader={:?} right-leader={:?} left-follower={:?} right-follower={:?}",
                tracked_fts_field_layout(&left_engine, left_region.id, table_id, index_id),
                tracked_fts_field_layout(&right_engine, right_region.id, table_id, index_id),
                tracked_fts_field_layout(&left_engine, right_region.id, table_id, index_id),
                tracked_fts_field_layout(&right_engine, left_region.id, table_id, index_id)
            )
        },
    );

    // Hard rejection: PrepareMerge's check_merge sees the mixed tracked layouts.
    assert_check_merge_rejects_inconsistent_fts(&left_engine, left_region.id, right_region.id);
    assert_check_merge_rejects_inconsistent_fts(&right_engine, left_region.id, right_region.id);

    // PD still accepts a merge operator, but the regions must stay distinct.
    assert!(
        client.try_merge(&left_key, &split_key),
        "mixed-layout regions must still be distinct when a merge is attempted"
    );
    assert_ne!(
        pd_client.get_region(&encode_bytes(&left_key)).unwrap().id,
        pd_client.get_region(&encode_bytes(&split_key)).unwrap().id
    );

    cluster.stop();
    oss.shutdown();
}