// Copyright 2026 TiKV Project Authors. Licensed under Apache-2.0.

use std::{collections::BTreeMap, sync::Arc};

use api_version::{ApiV1, ApiV2};
use futures::executor::block_on;
use kvengine::{UserMeta, WRITE_CF};
use kvproto::{
    coprocessor::KeyRange,
    kvrpcpb::{ApiVersion, Op},
};
use protobuf::Message;
use rfstore::store::RegionSnapshot;
use test_cloud_server::{ServerCluster, client::TxnMutations, util::Mutation};
use test_coprocessor::{DagChunkSpliter, DagSelect};
use tidb_query_common::{
    storage::{
        IntervalRange, Storage, scanner::ScanRescheduleOptions, test_fixture::FixtureStorage,
    },
    util::convert_to_prefix_next,
};
use tidb_query_datatype::{
    FieldTypeAccessor, FieldTypeTp,
    codec::{Datum, datum, table},
    expr::{EvalConfig, EvalContext},
};
use tidb_query_executors::{
    BatchIndexScanExecutor, BatchTableScanExecutor, interface::BatchExecutor,
};
use tikv::{
    coprocessor::dag::TikvStorage,
    storage::{
        Scanner,
        mvcc::CloudReader,
        txn::{CloudStore, CloudStoreScanner},
    },
};
use tikv_kv::SEEK_BOUND;
use tipb::{AnnQueryInfo, ColumnInfo, ColumnarIndexInfo, Expr, IndexScan, TableScan};
use txn_types::{Key, LockType, TsSet, WriteType};

use super::{
    i_to_key, i_to_val, single_key_txn_mutations,
    test_remote_coprocessor::{
        DagTest, KEYSPACE_ID, ProductTable, add_txn_prefix, encode_key_range,
    },
};
use crate::alloc_node_id;

#[test]
fn test_commit_ts_column_for_interval_and_point_ranges() {
    const TABLE_ID: i64 = 7;

    let rows = [
        (1, vec![(2, Datum::I64(10)), (4, Datum::F64(5.2))]),
        (3, vec![(4, Datum::Null), (2, Datum::I64(-5))]),
        (4, vec![(2, Datum::Null)]),
        (5, vec![(4, Datum::F64(0.1))]),
        (6, vec![]),
    ];
    let mut context = EvalContext::default();
    let key_values = rows
        .iter()
        .map(|(handle, columns)| {
            let values = columns.iter().map(|(_, value)| value.clone()).collect();
            let column_ids = columns.iter().map(|(id, _)| *id).collect::<Vec<_>>();
            (
                table::encode_row_key(TABLE_ID, *handle),
                table::encode_row(&mut context, values, &column_ids).unwrap(),
            )
        })
        .collect::<Vec<_>>();
    let commit_ts_by_key = rows
        .iter()
        .map(|(handle, _)| {
            (
                table::encode_row_key(TABLE_ID, *handle),
                100 + *handle as u64,
            )
        })
        .collect::<BTreeMap<_, _>>();
    let expected_commit_ts = rows
        .iter()
        .map(|(handle, _)| Some(100 + *handle))
        .collect::<Vec<_>>();

    let mut handle_column = ColumnInfo::default();
    handle_column
        .as_mut_accessor()
        .set_tp(FieldTypeTp::LongLong);
    handle_column.set_pk_handle(true);
    handle_column.set_column_id(1);

    let mut value_column = ColumnInfo::default();
    value_column.as_mut_accessor().set_tp(FieldTypeTp::LongLong);
    value_column.set_column_id(2);

    let mut defaulted_column = ColumnInfo::default();
    defaulted_column
        .as_mut_accessor()
        .set_tp(FieldTypeTp::Double);
    defaulted_column.set_column_id(4);
    defaulted_column
        .set_default_val(datum::encode_value(&mut context, &[Datum::F64(4.5)]).unwrap());

    let mut commit_ts_column = ColumnInfo::default();
    commit_ts_column
        .as_mut_accessor()
        .set_tp(FieldTypeTp::LongLong);
    commit_ts_column.set_column_id(table::EXTRA_COMMIT_TS_COL_ID);

    let mut interval_range = KeyRange::default();
    interval_range.set_start(table::encode_row_key(TABLE_ID, i64::MIN));
    interval_range.set_end(table::encode_row_key(TABLE_ID, i64::MAX));
    let point_ranges = rows
        .iter()
        .map(|(handle, _)| {
            let mut range = KeyRange::default();
            range.set_start(table::encode_row_key(TABLE_ID, *handle));
            range.set_end(range.get_start().to_vec());
            convert_to_prefix_next(range.mut_end());
            range
        })
        .collect::<Vec<_>>();

    let table_columns = vec![handle_column, value_column, defaulted_column];
    for (commit_ts_index, ranges) in [
        (0, vec![interval_range]),
        (table_columns.len(), point_ranges),
    ] {
        let mut columns = table_columns.clone();
        columns.insert(commit_ts_index, commit_ts_column.clone());
        let mut descriptor = TableScan::default();
        descriptor.set_columns(columns.into());
        let storage =
            FixtureStorage::from(key_values.clone()).with_commit_ts(commit_ts_by_key.clone());
        let mut executor = BatchTableScanExecutor::<_, ApiV1>::new(
            storage,
            Arc::new(EvalConfig::default()),
            descriptor,
            ranges,
            None,
            None,
            true,
            ScanRescheduleOptions::default(),
        )
        .unwrap();

        let result = block_on(executor.next_batch(10));
        assert!(result.is_drained.as_ref().unwrap().stop());
        assert_eq!(result.physical_columns.rows_len(), rows.len());
        assert_eq!(
            result.physical_columns[commit_ts_index]
                .decoded()
                .to_int_vec(),
            expected_commit_ts
        );
    }
}

#[test]
fn test_tikv_storage_reused_scanner_updates_load_commit_ts() {
    let node_id = alloc_node_id();
    let cluster = ServerCluster::new(vec![node_id], |_, _| {});
    let mut client = cluster.new_client();

    let key_numbers = [10, 20, 30];
    let keyspace_id = ApiV2::get_u32_keyspace_id_by_key(&i_to_key(10)).unwrap_or_default();
    client.split_keyspace(keyspace_id);
    let commit_ts = client.put_kv(key_numbers, i_to_key, i_to_val);
    let read_ts = client.get_ts().into_inner();

    let region_id = client.get_region_id(&i_to_key(10));
    let engine = cluster.get_kvengine(node_id);
    let snap = engine.get_snap_access(region_id).unwrap();
    let region_snapshot = RegionSnapshot::from_snapshot(snap, None, None);
    let store = CloudStore::new(region_snapshot, read_ts, TsSet::Empty, true, false);
    let mut storage = TikvStorage::new(store, false);

    // Reuse the same forward scanner while toggling whether commit timestamps
    // are requested. Every reset must apply the new setting.
    for (key_number, load_commit_ts) in [(10, false), (20, true), (30, false)] {
        storage
            .begin_scan(
                false,
                false,
                load_commit_ts,
                IntervalRange {
                    lower_inclusive: i_to_key(key_number),
                    upper_exclusive: i_to_key(key_number + 1),
                },
            )
            .unwrap();
        let entry = storage.scan_next_entry().unwrap().unwrap();
        assert_eq!(entry.key, i_to_key(key_number));
        assert_eq!(
            entry.commit_ts,
            load_commit_ts.then_some(commit_ts.into_inner())
        );
    }
}

#[test]
fn test_reject_commit_ts_with_cse_columnar_features() {
    let mut commit_ts_column = ColumnInfo::default();
    commit_ts_column
        .as_mut_accessor()
        .set_tp(FieldTypeTp::LongLong);
    commit_ts_column.set_column_id(table::EXTRA_COMMIT_TS_COL_ID);

    let mut deprecated_ann_scan = TableScan::default();
    deprecated_ann_scan.set_deprecated_ann_query(AnnQueryInfo::default());

    let mut columnar_index_scan = TableScan::default();
    columnar_index_scan.set_used_columnar_indexes(vec![ColumnarIndexInfo::default()].into());

    let mut pushed_down_filter_scan = TableScan::default();
    pushed_down_filter_scan.set_pushed_down_filter_conditions(vec![Expr::default()].into());

    for (feature, mut descriptor) in [
        ("deprecated ANN query", deprecated_ann_scan),
        ("used columnar indexes", columnar_index_scan),
        ("pushed-down filter conditions", pushed_down_filter_scan),
    ] {
        descriptor.mut_columns().push(commit_ts_column.clone());

        let error = BatchTableScanExecutor::check_supported(&descriptor).unwrap_err();
        assert!(
            error.to_string().contains(
                "TableScan cannot output _tidb_commit_ts with CSE columnar-only features"
            ),
            "feature: {}, error: {}",
            feature,
            error
        );

        for allow_columnar_scanner in [false, true] {
            let result = BatchTableScanExecutor::<_, ApiV1>::new(
                FixtureStorage::from(Vec::<(Vec<u8>, Vec<u8>)>::new()),
                Arc::new(EvalConfig::default()),
                descriptor.clone(),
                vec![],
                None,
                None,
                allow_columnar_scanner,
                ScanRescheduleOptions::default(),
            );
            let error = match result {
                Err(error) => error,
                Ok(_) => panic!(
                    "TableScan unexpectedly accepted _tidb_commit_ts with {}; \
                     allow_columnar_scanner: {}",
                    feature, allow_columnar_scanner
                ),
            };
            assert!(
                error.to_string().contains(
                    "TableScan cannot output _tidb_commit_ts with CSE columnar-only features"
                ),
                "feature: {}, allow_columnar_scanner: {}, error: {}",
                feature,
                allow_columnar_scanner,
                error
            );
        }
    }
}

#[test]
fn test_reject_commit_ts_column() {
    let mut commit_ts_column = ColumnInfo::default();
    commit_ts_column
        .as_mut_accessor()
        .set_tp(FieldTypeTp::LongLong);
    commit_ts_column.set_column_id(table::EXTRA_COMMIT_TS_COL_ID);

    let mut descriptor = IndexScan::default();
    descriptor.mut_columns().push(commit_ts_column.clone());
    let error = BatchIndexScanExecutor::check_supported(&descriptor).unwrap_err();
    assert!(
        error
            .to_string()
            .contains("IndexScan cannot output _tidb_commit_ts")
    );

    let result = BatchIndexScanExecutor::<_, ApiV1>::new(
        FixtureStorage::from(Vec::<(Vec<u8>, Vec<u8>)>::new()),
        Arc::new(EvalConfig::default()),
        vec![commit_ts_column],
        vec![],
        0,
        false,
        false,
        false,
        None,
        ScanRescheduleOptions::default(),
    );
    let error = match result {
        Err(error) => error,
        Ok(_) => panic!("IndexScan unexpectedly accepted _tidb_commit_ts"),
    };
    assert!(
        error
            .to_string()
            .contains("IndexScan cannot output _tidb_commit_ts")
    );
}

#[test]
fn test_backward_cloud_store_scanner_returns_put_commit_ts_below_many_locks() {
    let node_id = alloc_node_id();
    let cluster = ServerCluster::new(vec![node_id], |_, _| {});
    let mut client = cluster.new_client();

    let key_number = 10;
    let raw_key = i_to_key(key_number);
    let expected_value = i_to_val(key_number);
    let keyspace_id = ApiV2::get_u32_keyspace_id_by_key(&raw_key).unwrap_or_default();
    client.split_keyspace(keyspace_id);

    // Write one visible Put, then stack enough newer committed Lock records on
    // the same key to exceed the cursor seek bound from the original bug shape.
    let put_commit_ts = client.put_kv([key_number], i_to_key, i_to_val);
    let lock_count = SEEK_BOUND as usize + 1;
    let mut lock_versions = Vec::with_capacity(lock_count);
    for _ in 0..lock_count {
        let start_ts = client.get_ts();
        let lock_txn = single_key_txn_mutations(&raw_key, &[], Op::Lock);
        client
            .kv_prewrite_with_retry(lock_txn.primary(), None, lock_txn.clone(), start_ts)
            .expect("prewrite lock transaction");
        let commit_ts = client.get_ts();
        client
            .kv_commit(lock_txn, start_ts, commit_ts, false)
            .expect("commit lock transaction");
        lock_versions.push((start_ts, commit_ts));
    }

    let read_ts = client.get_ts().into_inner();
    let region_id = client.get_region_id(&raw_key);
    let engine = cluster.get_kvengine(node_id);
    let snap = engine.get_snap_access(region_id).unwrap();

    // Cloud Engine stores committed Lock records in EXTRA_CF. Verify that the
    // full newer stack exists before asserting that the data scanner ignores it.
    let key = Key::from_raw(&raw_key);
    let mut reader = CloudReader::new(snap.clone(), true);
    for &(start_ts, commit_ts) in &lock_versions {
        assert_eq!(
            reader.get_txn_commit_record(&key, start_ts).unwrap().info(),
            Some((commit_ts, WriteType::Lock))
        );
        assert!(commit_ts > put_commit_ts);
    }

    // A backward data scan must return the visible Put and its commit timestamp,
    // not the timestamp of any newer committed Lock record.
    let mut scanner = CloudStoreScanner::new(
        snap,
        true,
        true,
        TsSet::Empty,
        read_ts,
        None,
        None,
        false,
        false,
        true,
    )
    .unwrap();
    let (actual_key, entry) = scanner.next_entry().unwrap().unwrap();
    assert_eq!(actual_key.into_raw().unwrap(), raw_key);
    assert_eq!(entry.value, expected_value);
    assert_eq!(entry.commit_ts, Some(put_commit_ts));
    assert_ne!(
        entry.commit_ts,
        Some(lock_versions.last().unwrap().1),
        "the returned commit ts must not come from the newest Lock record"
    );
    assert!(scanner.next_entry().unwrap().is_none());
}

#[test]
fn test_cloud_store_scanner_commit_ts_ignores_lock_and_rollback() {
    let node_id = alloc_node_id();
    let cluster = ServerCluster::new(vec![node_id], |_, _| {});
    let mut client = cluster.new_client();

    let keyspace_id = ApiV2::get_u32_keyspace_id_by_key(&i_to_key(10)).unwrap_or_default();
    client.split_keyspace(keyspace_id);

    let key_numbers = [10, 20];
    let put_commit_ts = client.put_kv(key_numbers, i_to_key, i_to_val);

    let lock_mutations = || {
        key_numbers
            .iter()
            .map(|&key_number| {
                let mut mutation = Mutation::default();
                mutation.set_key(i_to_key(key_number));
                mutation.set_op(Op::Lock);
                mutation
            })
            .collect::<Vec<_>>()
    };

    // Commit a newer Lock record for each key. Cloud Engine stores these
    // transaction-status records in EXTRA_CF rather than WRITE_CF.
    let lock_start_ts = client.get_ts();
    let lock_txn = TxnMutations::from_normal(lock_mutations());
    client
        .kv_prewrite(lock_txn.primary(), None, lock_txn.clone(), lock_start_ts)
        .expect("prewrite lock transaction");
    let lock_commit_ts = client.get_ts();
    client
        .kv_commit(lock_txn, lock_start_ts, lock_commit_ts, false)
        .expect("commit lock transaction");

    // Add a still newer rollback record for each key. It must not hide the
    // older Put value or replace its commit timestamp during a data scan.
    let rollback_start_ts = client.get_ts();
    let rollback_txn = TxnMutations::from_normal(lock_mutations());
    client
        .kv_prewrite(
            rollback_txn.primary(),
            None,
            rollback_txn.clone(),
            rollback_start_ts,
        )
        .expect("prewrite transaction to roll back");
    client
        .kv_rollback(rollback_txn, rollback_start_ts)
        .expect("rollback transaction");

    let region_id = client.get_region_id(&i_to_key(10));
    let engine = cluster.get_kvengine(node_id);
    let snap = engine.get_snap_access(region_id).unwrap();

    // Verify that both logical transaction-status records really exist before
    // checking that the data scanner ignores them.
    let first_key = Key::from_raw(&i_to_key(10));
    let mut reader = CloudReader::new(snap.clone(), true);
    assert_eq!(
        reader
            .get_txn_commit_record(&first_key, lock_start_ts)
            .unwrap()
            .info(),
        Some((lock_commit_ts, WriteType::Lock))
    );
    assert_eq!(
        reader
            .get_txn_commit_record(&first_key, rollback_start_ts)
            .unwrap()
            .info(),
        Some((rollback_start_ts, WriteType::Rollback))
    );

    let read_ts = client.get_ts().into_inner();
    for desc in [false, true] {
        let mut scanner = CloudStoreScanner::new(
            snap.clone(),
            desc,
            true,
            TsSet::Empty,
            read_ts,
            None,
            None,
            false,
            false,
            true,
        )
        .unwrap();

        let mut actual = Vec::new();
        while let Some((key, entry)) = scanner.next_entry().unwrap() {
            actual.push((
                key.into_raw().unwrap(),
                entry.value,
                entry.commit_ts.expect("commit ts should be loaded"),
            ));
        }

        let ordered_key_numbers: Vec<_> = if desc {
            key_numbers.iter().rev().copied().collect()
        } else {
            key_numbers.to_vec()
        };
        let expected = ordered_key_numbers
            .into_iter()
            .map(|key_number| (i_to_key(key_number), i_to_val(key_number), put_commit_ts))
            .collect::<Vec<_>>();
        assert_eq!(actual, expected, "desc: {desc}");
    }
}

#[test]
fn test_table_scan_with_commit_ts() {
    let rows = vec![
        (1, Some("name:2"), 2),
        (2, Some("name:1"), 3),
        (4, Some("name:1"), 1),
        (5, Some("name:3"), 4),
    ];

    let product = ProductTable::new();
    let mut dag_test = DagTest::new(&product);
    dag_test.insert_and_commit(&rows);

    // Rewrite one row so the scan must return its latest commit timestamp.
    let updated_id = 2;
    let updated_row_key = add_txn_prefix(&table::encode_row_key(
        product.table_info().get_table_id(),
        updated_id,
    ));
    let updated_row_value = table::encode_row(
        &mut EvalContext::default(),
        vec![
            Datum::I64(updated_id),
            Datum::Bytes(b"name:1".to_vec()),
            Datum::I64(3),
        ],
        &[product["id"].id, product["name"].id, product["count"].id],
    )
    .unwrap();
    let updated_row_commit_ts = dag_test.client.put_kv(
        0..1,
        |_| updated_row_key.clone(),
        |_| updated_row_value.clone(),
    );

    // Read the commit timestamp from the raw row metadata so the coprocessor
    // result is checked against the storage source of truth.
    let read_ts = dag_test.get_ts().into_inner();
    let row_key_range = dag_test.get_key_range_all();
    let snap = dag_test.get_snap_access(read_ts, vec![row_key_range]);
    let expected_commit_ts: BTreeMap<_, _> = rows
        .iter()
        .map(|(id, ..)| {
            let key = add_txn_prefix(&table::encode_row_key(
                product.table_info().get_table_id(),
                *id,
            ));
            let item = snap.get(WRITE_CF, &key, read_ts);
            assert!(item.exists());
            let commit_ts = UserMeta::from_slice(item.user_meta()).commit_ts as i64;
            (*id, commit_ts)
        })
        .collect();
    assert_eq!(
        expected_commit_ts[&updated_id],
        updated_row_commit_ts.into_inner() as i64
    );

    let mut commit_ts_col = ColumnInfo::default();
    commit_ts_col.set_column_id(table::EXTRA_COMMIT_TS_COL_ID);
    commit_ts_col
        .as_mut_accessor()
        .set_tp(FieldTypeTp::LongLong);

    // Put the virtual column before the primary-key handle. This covers the
    // layout that previously caused table scan to panic.
    let mut table_select = DagSelect::from(&product);
    table_select.execs[0]
        .mut_tbl_scan()
        .mut_columns()
        .insert(0, commit_ts_col.clone());
    table_select.output_offsets = Some(vec![0, 1, 2, 3]);
    table_select.start_ts = Some(read_ts);
    let mut table_resp = dag_test.select_all(table_select.build(), None);
    let actual_table_rows =
        DagChunkSpliter::new(table_resp.take_chunks().into(), 4).collect::<Vec<_>>();
    let expected_table_rows = rows
        .iter()
        .copied()
        .map(|(id, name, count)| {
            vec![
                Datum::I64(expected_commit_ts[&id]),
                Datum::I64(id),
                name.map(|value| value.as_bytes()).into(),
                Datum::I64(count),
            ]
        })
        .collect::<Vec<_>>();
    assert_eq!(actual_table_rows, expected_table_rows);

    // A single-row range is served by the point-get path rather than by the
    // range scanner. Keep this check separate so both metadata paths are covered.
    let point_id = 2;
    let mut point_select = DagSelect::from(&product);
    point_select.execs[0]
        .mut_tbl_scan()
        .mut_columns()
        .push(commit_ts_col);
    point_select.key_ranges = vec![product.get_record_range_one(point_id)];
    point_select.output_offsets = Some(vec![0, 1, 2, 3]);
    point_select.start_ts = Some(read_ts);
    let mut point_resp = dag_test.select_all(point_select.build(), None);
    let point_rows = DagChunkSpliter::new(point_resp.take_chunks().into(), 4).collect::<Vec<_>>();
    assert_eq!(
        point_rows,
        vec![vec![
            Datum::I64(point_id),
            Datum::Bytes(b"name:1".to_vec()),
            Datum::I64(3),
            Datum::I64(expected_commit_ts[&point_id]),
        ]]
    );
}

#[test]
fn test_table_scan_commit_ts_with_committed_lock_requires_write_cf() {
    let product = ProductTable::new();
    let mut dag_test = DagTest::new(&product);

    let row_mutation = |id: i64, name: &str, count: i64| {
        let key = add_txn_prefix(&table::encode_row_key(
            product.table_info().get_table_id(),
            id,
        ));
        let value = table::encode_row(
            &mut EvalContext::default(),
            vec![
                Datum::I64(id),
                Datum::Bytes(name.as_bytes().to_vec()),
                Datum::I64(count),
            ],
            &[product["id"].id, product["name"].id, product["count"].id],
        )
        .unwrap();
        let mut mutation = Mutation::default();
        mutation.set_op(Op::Put);
        mutation.set_key(key);
        mutation.set_value(value);
        mutation
    };

    let primary = row_mutation(1, "primary", 1);
    let secondary = row_mutation(2, "secondary", 2);
    let primary_key = primary.get_key().to_vec();
    let secondary_key = secondary.get_key().to_vec();
    let start_ts = dag_test.get_ts();
    let prewrite_mutations = TxnMutations::from_normal(vec![primary.clone(), secondary.clone()]);
    dag_test
        .client
        .kv_prewrite_with_retry(
            primary_key.clone().into(),
            None,
            prewrite_mutations,
            start_ts,
        )
        .expect("prewrite primary and secondary rows");

    // Commit only the primary. The transaction is now known to be committed,
    // while the target secondary Put remains in LOCK_CF.
    let commit_ts = dag_test
        .client
        .kv_commit(
            TxnMutations::from_normal(vec![primary]),
            start_ts,
            dag_test.get_ts(),
            false,
        )
        .expect("commit primary row");
    dag_test.client.must_locked_with_properties(
        &secondary_key,
        &primary_key,
        start_ts,
        LockType::Put,
    );

    let read_ts = dag_test.get_ts().into_inner();
    assert!(read_ts >= commit_ts.into_inner());

    let mut commit_ts_column = tipb::ColumnInfo::default();
    commit_ts_column.set_column_id(table::EXTRA_COMMIT_TS_COL_ID);
    commit_ts_column
        .as_mut_accessor()
        .set_tp(FieldTypeTp::LongLong);

    // Use a point range so this TableScan exercises incremental_get_entry.
    let mut select = DagSelect::from(&product);
    select.execs[0]
        .mut_tbl_scan()
        .mut_columns()
        .push(commit_ts_column);
    select.key_ranges = vec![product.get_record_range_one(2)];
    select.output_offsets = Some(vec![0, 1, 2, 3]);
    select.start_ts = Some(read_ts);
    let mut request = select.build();
    for range in request.mut_ranges().iter_mut() {
        encode_key_range(range);
    }
    assert_eq!(request.get_ranges()[0].get_start(), secondary_key);

    let region = dag_test.client.get_region_by_key(&secondary_key);
    let mut request_context = dag_test
        .client
        .new_rpc_ctx(region.id, &secondary_key)
        .unwrap();
    request_context.set_api_version(ApiVersion::V2);
    request_context.set_keyspace_id(KEYSPACE_ID);
    request_context.set_committed_locks(vec![start_ts.into_inner()]);
    let store_id = request_context.get_peer().get_store_id();
    request.set_context(request_context);

    // CloudStore does not consume ReqContext::access_locks. Requiring the
    // commit timestamp therefore treats the secondary Put as a conflict
    // instead of reading through its lock value.
    let response = dag_test
        .client
        .get_kv_client(store_id)
        .coprocessor(&request)
        .unwrap();
    assert!(
        response.has_locked(),
        "expected KeyIsLocked, got {:?}",
        response
    );
    let lock_info = response.get_locked();
    assert_eq!(lock_info.get_key(), secondary_key);
    assert_eq!(lock_info.get_primary_lock(), primary_key);
    assert_eq!(lock_info.get_lock_version(), start_ts.into_inner());

    // Apply the secondary with the transaction's original commit timestamp.
    // A fresh snapshot can now obtain both the row and its commit timestamp
    // from WRITE_CF, even though committed_locks remains populated.
    let secondary_commit_ts = dag_test
        .client
        .kv_commit(
            TxnMutations::from_normal(vec![secondary]),
            start_ts,
            commit_ts,
            false,
        )
        .expect("commit secondary row");
    assert_eq!(secondary_commit_ts, commit_ts);

    let mut cop_response = dag_test
        .client
        .get_kv_client(store_id)
        .coprocessor(&request)
        .expect("send TableScan after the Put reaches WRITE_CF");
    assert!(
        !cop_response.has_locked(),
        "TableScan should succeed after the Put reaches WRITE_CF: {:?}",
        cop_response
    );
    let mut select_response = tipb::SelectResponse::default();
    select_response
        .merge_from_bytes(&cop_response.take_data())
        .unwrap();
    assert!(!select_response.has_error(), "{:?}", select_response);
    let rows = DagChunkSpliter::new(select_response.take_chunks().into(), 4).collect::<Vec<_>>();
    assert_eq!(
        rows,
        vec![vec![
            Datum::I64(2),
            Datum::Bytes(b"secondary".to_vec()),
            Datum::I64(2),
            Datum::I64(commit_ts.into_inner() as i64),
        ]]
    );
}