// Copyright 2024 PingCAP, Inc. use std::{ iter::FromIterator, rc::Rc, sync::{Arc, Mutex, atomic::Ordering}, thread, time::Duration, }; use bytes::{Buf, Bytes}; use collections::HashSet; use futures::executor::block_on; use rand::prelude::*; use schema::schema::StorageClassSpec; use tidb_query_datatype::{ FieldTypeAccessor, FieldTypeTp, codec::{ data_type::VectorFloat32, mysql::VectorFloat32Encoder, table::{encode_row_key, encode_row_key_prefix}, }, expr::EvalContext, }; use crate::{ DEL_PREFIXES_KEY, DeletePrefixes, EXTRA_CF, IdVer, LOCK_CF, LevelHandler, SnapAccess, WRITE_CF, WriteBatch, compaction::CompactionPriority, context::IaCtx, dfs, dfs::FileType, read::{CloudColumnarReader, CloudColumnarReaders, TableCtx}, shard::{ShardCf, ShardDataBuilder}, table, table::{ ChecksumType, SnapVersion, columnar::{ Block, ColumnarFile, ColumnarFilterReader, ColumnarLevels, ColumnarMetaCache, ColumnarReader, ColumnarRowTableReader, HANDLE_COL_ID, MinMaxIndex, VERSION_COL_ID, filter::TableScanCtx, new_int_handle_column_info, new_version_column_info, tests::{ build_table, i_to_common_handle, merge_refs, new_schema, new_schema_with_nullable, verify_with_ref_rows, }, }, file::{File, InMemFile}, schema_file::{Schema, SchemaBuf, SchemaFile, build_schema_file}, sstable::{BlockCache, SsTable}, vector_index::{VectorIndexBuilder, VectorIndexFile, VectorIndexes}, }, tests::{ 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, write_data, }, }; const COLUMNAR_COMPACTION_WAIT_TIME: usize = 10; const COLUMNAR_COMPACTION_IDLE_WAIT_TIME: usize = 30; fn wait_columnar_compaction_idle(engine: &crate::Engine, shard: &crate::Shard) -> bool { try_wait( || { let compacting = engine.is_shard_compacting(shard.id); let priority = shard.compaction_priority.read().unwrap().clone(); info!( "wait columnar compaction idle {} compacting: {:?}, priority: {:?}", shard.tag(), compacting, priority ); compacting == Some(false) && priority.is_none() }, COLUMNAR_COMPACTION_IDLE_WAIT_TIME, ) } #[test] fn test_columnar_l0_compaction() { ::test_util::init_log_for_test(); let keyspace_id = 1; let table_id = 30; let table_id2 = 31; let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, ""); let allocate_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 = new_schema(table_id, true); let schema2 = new_schema(table_id2, true); let schemas = vec![schema.clone(), schema2.clone()]; let schema_version = 10; let schema_file_data = build_schema_file(keyspace_id, schema_version, schemas, 0); let fs = engine.fs.clone(); let schema_raw_file = Arc::new(InMemFile::new(allocate_id(), Bytes::from(schema_file_data))); fs.get_runtime() .block_on( fs.create( schema_raw_file.id(), schema_raw_file .read(0, schema_raw_file.size() as usize) .unwrap(), dfs::Options::default().with_type(FileType::Schema), ), ) .unwrap(); let schema_file = SchemaFile::open(schema_raw_file).unwrap(); let opts = dfs::Options::default().with_type(FileType::Columnar); let (l0_tbl_0, l0_tbl_0_ref) = build_table(allocate_id(), &schema, 0, 600, 300); let (l0_tbl_1, l0_tbl_1_ref) = build_table(allocate_id(), &schema, 300, 900, 400); let (l0_tbl_2, l0_tbl_2_ref) = build_table(allocate_id(), &schema, 1000, 2000, 400); let (l0_tbl_3, l0_tbl_3_ref) = build_table(allocate_id(), &schema2, 0, 600, 300); let (l1_tbl_0, l1_tbl_0_ref) = build_table(allocate_id(), &schema, 0, 100, 100); let (l1_tbl_1, l1_tbl_1_ref) = build_table(allocate_id(), &schema2, 0, 100, 100); for file in [ &l0_tbl_0, &l0_tbl_1, &l0_tbl_2, &l0_tbl_3, &l1_tbl_0, &l1_tbl_1, ] { info!("build file_id: {}, file size: {}", file.id(), file.size()); fs.get_runtime() .block_on(fs.create(file.id(), file.read(0, file.size() as usize).unwrap(), opts)) .unwrap() } let columnar_meta_cache = ColumnarMetaCache::default(); let mut col_levels = ColumnarLevels::new(); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_0, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_1, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_2, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_3, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_0, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_1, None, columnar_meta_cache.clone()).unwrap(), ); 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, table_id2]); shard.set_data(builder.build()); shard.initial_flushed.store(true, Ordering::SeqCst); let id_ver = shard.id_ver(); *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::ColumnarL0 { score: 2.0 }); engine.trigger_compact(id_ver); info!("trigger columnar l0 compaction {}", shard.tag()); let ok = try_wait( || { info!( "wait columnar l0 compaction {} l0 files: {}, l1 files: {}", shard.tag(), shard.get_data().col_levels.levels[0].files.len(), shard.get_data().col_levels.levels[1].files.len() ); shard.get_data().col_levels.levels[0].files.is_empty() }, COLUMNAR_COMPACTION_WAIT_TIME, ); assert!(ok, "columnar l0 compaction failed"); let snap = shard.new_snap_access(); let mut mvcc_reader = snap .new_columnar_mvcc_reader_without_pack_clean_read( table_id, &schema.columns, None, 500, None, ) .unwrap() .unwrap(); block_on(mvcc_reader.set_handle_range(&i_to_common_handle(0), &i_to_common_handle(2100))) .unwrap(); let mut block = Block::new(&schema); block_on(mvcc_reader.read_block(&mut block, usize::MAX)).unwrap(); let tbl_refs = merge_refs( vec![l0_tbl_0_ref, l0_tbl_1_ref, l0_tbl_2_ref, l1_tbl_0_ref], 1, Some(500), None, Some((i_to_common_handle(0), i_to_common_handle(2100))), ); assert_eq!(block.length(), tbl_refs.len()); verify_with_ref_rows(&block, &tbl_refs); let mut mvcc_reader2 = snap .new_columnar_mvcc_reader_without_pack_clean_read( table_id2, &schema2.columns, None, 500, None, ) .unwrap() .unwrap(); block_on(mvcc_reader2.set_handle_range(&i_to_common_handle(0), &i_to_common_handle(2100))) .unwrap(); let mut block2 = Block::new(&schema2); block_on(mvcc_reader2.read_block(&mut block2, usize::MAX)).unwrap(); let tbl_refs2 = merge_refs( vec![l0_tbl_3_ref, l1_tbl_1_ref], 1, Some(500), None, Some((i_to_common_handle(0), i_to_common_handle(2100))), ); assert_eq!(block2.length(), tbl_refs2.len()); verify_with_ref_rows(&block2, &tbl_refs2); } #[test] fn test_columnar_l1_compaction() { ::test_util::init_log_for_test(); let keyspace_id = 1; let table_id = 30; let table_id2 = 31; let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, ""); let allocate_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 = new_schema(table_id, true); let schema2 = new_schema(table_id2, true); let schemas = vec![schema.clone(), schema2.clone()]; let schema_version = 10; let schema_file_data = build_schema_file(keyspace_id, schema_version, schemas, 0); let fs = engine.fs.clone(); let schema_raw_file = Arc::new(InMemFile::new(allocate_id(), Bytes::from(schema_file_data))); fs.get_runtime() .block_on( fs.create( schema_raw_file.id(), schema_raw_file .read(0, schema_raw_file.size() as usize) .unwrap(), dfs::Options::default().with_type(FileType::Schema), ), ) .unwrap(); let schema_file = SchemaFile::open(schema_raw_file).unwrap(); let opts = dfs::Options::default().with_type(FileType::Columnar); let (l1_tbl_0, l1_tbl_0_ref) = build_table(allocate_id(), &schema, 0, 600, 300); let (l1_tbl_1, l1_tbl_1_ref) = build_table(allocate_id(), &schema, 300, 900, 400); let (l1_tbl_2, l1_tbl_2_ref) = build_table(allocate_id(), &schema, 1000, 1500, 400); let (l1_tbl_3, l1_tbl_3_ref) = build_table(allocate_id(), &schema2, 300, 1200, 500); let (l2_tbl_0, l2_tbl_0_ref) = build_table(allocate_id(), &schema, 0, 1000, 100); let (l2_tbl_1, l2_tbl_1_ref) = build_table(allocate_id(), &schema, 1000, 2000, 100); let (l2_tbl_2, l2_tbl_2_ref) = build_table(allocate_id(), &schema2, 0, 1000, 200); for file in [ &l1_tbl_0, &l1_tbl_1, &l1_tbl_2, &l1_tbl_3, &l2_tbl_0, &l2_tbl_1, &l2_tbl_2, ] { info!("build file_id: {}, file size: {}", file.id(), file.size()); fs.get_runtime() .block_on(fs.create(file.id(), file.read(0, file.size() as usize).unwrap(), opts)) .unwrap() } let columnar_meta_cache = ColumnarMetaCache::default(); let mut col_levels = ColumnarLevels::new(); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_0, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_1, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_2, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_3, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 2, ColumnarFile::open(l2_tbl_0, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 2, ColumnarFile::open(l2_tbl_1, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 2, ColumnarFile::open(l2_tbl_2, None, columnar_meta_cache.clone()).unwrap(), ); 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, table_id2]); shard.set_data(builder.build()); shard.initial_flushed.store(true, Ordering::SeqCst); let id_ver = shard.id_ver(); *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::ColumnarL1 { score: 2.0 }); engine.trigger_compact(id_ver); info!("trigger columnar l1 compaction {}", shard.tag()); let ok = try_wait( || { info!( "wait columnar l1 compaction {} l1 files: {}, l2 files: {}", shard.tag(), shard.get_data().col_levels.levels[1].files.len(), shard.get_data().col_levels.levels[2].files.len() ); shard.get_data().col_levels.levels[1].files.is_empty() }, COLUMNAR_COMPACTION_WAIT_TIME, ); assert!(ok, "columnar l1 compaction failed"); let snap = shard.new_snap_access(); let mut mvcc_reader = snap .new_columnar_mvcc_reader_without_pack_clean_read( table_id, &schema.columns, None, 500, None, ) .unwrap() .unwrap(); block_on(mvcc_reader.set_handle_range(&i_to_common_handle(0), &i_to_common_handle(2100))) .unwrap(); let mut block = Block::new(&schema); block_on(mvcc_reader.read_block(&mut block, usize::MAX)).unwrap(); let tbl_refs = merge_refs( vec![ l1_tbl_0_ref, l1_tbl_1_ref, l1_tbl_2_ref, l2_tbl_0_ref, l2_tbl_1_ref, ], 2, Some(500), None, Some((i_to_common_handle(0), i_to_common_handle(2100))), ); assert_eq!(block.length(), tbl_refs.len()); verify_with_ref_rows(&block, &tbl_refs); let mut mvcc_reader2 = snap .new_columnar_mvcc_reader_without_pack_clean_read( table_id2, &schema2.columns, None, 600, None, ) .unwrap() .unwrap(); block_on(mvcc_reader2.set_handle_range(&i_to_common_handle(0), &i_to_common_handle(2100))) .unwrap(); let mut block2 = Block::new(&schema2); block_on(mvcc_reader2.read_block(&mut block2, usize::MAX)).unwrap(); let tbl_refs2 = merge_refs( vec![l1_tbl_3_ref, l2_tbl_2_ref], 2, Some(600), None, Some((i_to_common_handle(0), i_to_common_handle(2100))), ); assert_eq!(block2.length(), tbl_refs2.len()); verify_with_ref_rows(&block2, &tbl_refs2); } #[test] fn test_columnar_major_compaction() { use crate::table::columnar::{ColumnarMergeReader, ColumnarMvccReader}; ::test_util::init_log_for_test(); let keyspace_id = KEYSPACE_ID; let table_id = 30; let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, ""); let allocate_id = || engine.id_allocator.alloc_id(1).unwrap()[0]; engine.opts.set_build_columnar(true); let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id); let shard = engine.get_shard(shard_id).unwrap(); let schema = new_schema(table_id, false); let schemas = vec![schema.clone()]; let schema_version = 10; let schema_file_data = build_schema_file(keyspace_id, schema_version, schemas, 0); let fs = engine.fs.clone(); let schema_raw_file = Arc::new(InMemFile::new(allocate_id(), Bytes::from(schema_file_data))); fs.get_runtime() .block_on( fs.create( schema_raw_file.id(), schema_raw_file .read(0, schema_raw_file.size() as usize) .unwrap(), dfs::Options::default().with_type(FileType::Schema), ), ) .unwrap(); let schema_file = SchemaFile::open(schema_raw_file).unwrap(); let mut saved_vals = vec![]; let l1_tbl_0 = new_sst_table_for_columnar( &engine, allocate_id(), table_id, 300, 900, 350, &mut saved_vals, ); let l1_tbl_1 = new_sst_table_for_columnar( &engine, allocate_id(), table_id, 1000, 1500, 350, &mut saved_vals, ); let l2_tbl_0 = new_sst_table_for_columnar( &engine, allocate_id(), table_id, 0, 1000, 100, &mut saved_vals, ); let l2_tbl_1 = new_sst_table_for_columnar( &engine, allocate_id(), table_id, 1000, 2000, 100, &mut saved_vals, ); let level_handler_1 = LevelHandler::new(1, vec![l1_tbl_0.clone(), l1_tbl_1.clone()]); let level_handler_2 = LevelHandler::new(2, vec![l2_tbl_0.clone(), l2_tbl_1.clone()]); let mut write_cf = ShardCf::new(WRITE_CF); write_cf.set_level(level_handler_1); write_cf.set_level(level_handler_2); let mut builder = ShardDataBuilder::new(shard.get_data()); builder.set_cfs([write_cf, ShardCf::new(LOCK_CF), ShardCf::new(EXTRA_CF)]); builder.set_schema(schema_file.get_version(), 0, Some(schema_file.clone())); builder.set_persisted_version(SnapVersion::new(0, 1)); shard.set_data(builder.build()); shard.initial_flushed.store(true, Ordering::SeqCst); let id_ver = shard.id_ver(); *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::ColumnarMajor { table_ids_to_add: vec![table_id], table_ids_to_clear: vec![], is_manual: false, schema_version: schema_file.get_version(), }); engine.trigger_compact(id_ver); info!("trigger columnar major compaction {}", shard.tag()); let ok = try_wait( || { info!( "wait columnar major compaction {} l0 files: {}, l1 files: {}, l2 files: {}", shard.tag(), shard.get_data().col_levels.levels[0].files.len(), shard.get_data().col_levels.levels[1].files.len(), shard.get_data().col_levels.levels[2].files.len() ); !shard.get_data().col_levels.levels[2].files.is_empty() }, COLUMNAR_COMPACTION_WAIT_TIME, ); assert!(ok, "columnar major compaction failed"); let snap = shard.new_snap_access(); let mut mvcc_reader = snap .new_columnar_mvcc_reader_without_pack_clean_read( table_id, &schema.columns, None, 500, None, ) .unwrap() .unwrap(); // Use a random end int handle let end_handle = thread_rng().gen_range(500..2100); block_on(mvcc_reader.set_int_handle_range(0, Some(end_handle))).unwrap(); let mut block = Block::new(&schema); let mvcc_reader_counts = block_on(mvcc_reader.read_block(&mut block, usize::MAX)).unwrap(); info!("mvcc_reader read block with {} rows", mvcc_reader_counts); let mut columnar_readers: Vec> = vec![]; for tbl in [l1_tbl_0, l1_tbl_1, l2_tbl_0, l2_tbl_1] { let iter = tbl.new_iterator(false, false); let reader = ColumnarRowTableReader::new(schema.clone(), iter, None, false, None).unwrap(); columnar_readers.push(Box::new(reader)); } let merged_reader = ColumnarMergeReader::new(schema.clone(), columnar_readers); let mut mvcc_row_reader = ColumnarMvccReader::new(Box::new(merged_reader), &schema, 500); block_on(mvcc_row_reader.set_int_handle_range(0, Some(end_handle))).unwrap(); let mut row_block = Block::new(&schema); let merge_reader_counts = block_on(mvcc_row_reader.read_block(&mut row_block, usize::MAX)).unwrap(); info!("merge_reader read block with {} rows", merge_reader_counts); verify_columnar_with_blocks(&row_block, &block); let ok = wait_columnar_compaction_idle(&engine, &shard); assert!(ok, "wait other compaction failed"); assert!(shard.get_data().schema_file.is_some()); // Remove columnar compaction. let data = shard.get_data(); let mut builder = ShardDataBuilder::new(data); // Remove schema file to avoid columnar major compaction to be re-triggered. builder.set_schema(0, 0, None); shard.set_data(builder.build()); engine.trigger_compact(id_ver); info!("trigger remove columnar compaction {}", shard.tag()); let ok = try_wait( || { info!( "wait remove columnar compaction {} l0 files: {}, l1 files: {}, l2 files: {}", shard.tag(), shard.get_data().col_levels.levels[0].files.len(), shard.get_data().col_levels.levels[1].files.len(), shard.get_data().col_levels.levels[2].files.len() ); shard.get_data().col_levels.levels[2].files.is_empty() }, COLUMNAR_COMPACTION_WAIT_TIME, ); assert!(ok); assert_eq!(shard.get_columnar_snap_version(), SnapVersion::zero()); } #[test] fn test_flush_tracks_unconverted_l0_when_schema_missing_but_columnar_active() { ::test_util::init_log_for_test(); let keyspace_id = KEYSPACE_ID; let table_id = 30; let (engine, apply_tx) = new_test_engine_opt_with_custom_options(true, DEF_BLOCK_SIZE, "", |opts| { // Disable background compaction to avoid racing with this test, while keeping // the normal flush apply path. opts.for_restore = true; }); 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, Ordering::SeqCst); // Simulate schema-file gap during schema refresh / non-overlap handling. // Columnar state is still considered active as long as `columnar_table_ids` is // non-empty. let data = shard.get_data(); let mut builder = ShardDataBuilder::new(data); builder.set_schema(0, 0, None); builder.set_columnar_table_ids(vec![table_id]); shard.set_data(builder.build()); assert!(shard.get_data().schema_file.is_none()); assert_eq!(shard.get_data().columnar_table_ids, vec![table_id]); // Write some data, then force a flush. let mut wb = WriteBatch::new(shard_id); for handle in 0_i64..10 { let key = [ keyspace_prefix(keyspace_id), encode_row_key(table_id, handle), ] .concat(); wb.put(WRITE_CF, &key, b"v", 0, &[], 10); } write_data(wb, &apply_tx); let mut wb = WriteBatch::new(shard_id); wb.set_switch_mem_table(); write_data(wb, &apply_tx); let ok = try_wait( || { let data = shard.get_data(); if data.l0_tbls.is_empty() { return false; } let l0_ids: HashSet = HashSet::from_iter(data.l0_tbls.iter().map(|t| t.id())); let unconverted_ids: HashSet = HashSet::from_iter(data.col_levels.unconverted_l0s.iter().map(|t| t.id())); l0_ids == unconverted_ids }, 5, ); assert!( ok, "flush did not append newly flushed L0s into unconverted_l0s when schema_file is None" ); } #[test] fn test_columnar_major_compaction_multiple_tables() { use crate::{ table::{ columnar::{ColumnarMergeReader, ColumnarMvccReader}, table::BoundedDataSet, }, table_id::get_table_id_from_data_bound, }; ::test_util::init_log_for_test(); let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, ""); engine.opts.set_build_columnar(true); let keyspace_id = KEYSPACE_ID; let table_ids = vec![1, 2, 3, 4, 5]; let mut schemas = vec![]; // Split to keep tables in same region. let split_key_1 = [keyspace_prefix(keyspace_id), encode_row_key_prefix(1)].concat(); let split_key_2 = [keyspace_prefix(keyspace_id), encode_row_key_prefix(6)].concat(); let mut splitter = Splitter::new( vec![split_key_1, split_key_2], IdVer::new(1, 1), 5, apply_tx.clone(), ); let handle = thread::spawn(move || { splitter.run(); }); handle.join().unwrap(); for &table_id in &table_ids { let schema = new_schema(table_id, false); schemas.push(schema); } let allocate_id = || engine.id_allocator.alloc_id(1).unwrap()[0]; let schema_version = 10; // Build schema file for the first 3 tables. let schemas_1 = schemas[0..3].to_vec(); let schema_file_data_1 = build_schema_file(keyspace_id, schema_version, schemas_1, 0); let fs = engine.fs.clone(); let schema_raw_file_1 = Arc::new(InMemFile::new( allocate_id(), Bytes::from(schema_file_data_1), )); fs.get_runtime() .block_on( fs.create( schema_raw_file_1.id(), schema_raw_file_1 .read(0, schema_raw_file_1.size() as usize) .unwrap(), dfs::Options::default().with_type(FileType::Schema), ), ) .unwrap(); let schema_file = SchemaFile::open(schema_raw_file_1).unwrap(); let mut saved_vals = vec![]; let mut l1_tbls = vec![]; let mut l2_tbls = vec![]; for (idx, &table_id) in table_ids.iter().enumerate().rev() { let l1_tbl_0 = new_sst_table_for_columnar( &engine, allocate_id(), table_id, 300, 900, 350 + idx as u64, &mut saved_vals, ); l1_tbls.push(l1_tbl_0); let l1_tbl_1 = new_sst_table_for_columnar( &engine, allocate_id(), table_id, 1000, 1500, 350 + idx as u64, &mut saved_vals, ); l1_tbls.push(l1_tbl_1); let l2_tbl_0 = new_sst_table_for_columnar( &engine, allocate_id(), table_id, 0, 1000, 100 + idx as u64, &mut saved_vals, ); l2_tbls.push(l2_tbl_0); let l2_tbl_1 = new_sst_table_for_columnar( &engine, allocate_id(), table_id, 1000, 2000, 100 + idx as u64, &mut saved_vals, ); l2_tbls.push(l2_tbl_1); } l1_tbls.sort_by_key(|tbl| tbl.smallest().to_vec()); l2_tbls.sort_by_key(|tbl| tbl.smallest().to_vec()); let level_handler_1 = LevelHandler::new(1, l1_tbls.clone()); let level_handler_2 = LevelHandler::new(2, l2_tbls.clone()); let mut write_cf = ShardCf::new(WRITE_CF); write_cf.set_level(level_handler_1); write_cf.set_level(level_handler_2); let shard = engine.get_shard(7).unwrap(); let mut builder = ShardDataBuilder::new(shard.get_data()); builder.set_cfs([write_cf, ShardCf::new(LOCK_CF), ShardCf::new(EXTRA_CF)]); builder.set_schema(schema_file.get_version(), 0, Some(schema_file.clone())); shard.set_data(builder.build()); shard.initial_flushed.store(true, Ordering::SeqCst); let id_ver = shard.id_ver(); *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::ColumnarMajor { table_ids_to_add: schema_file.export_backend_table_ids(), table_ids_to_clear: vec![], is_manual: false, schema_version: schema_file.get_version(), }); engine.trigger_compact(id_ver); info!("trigger columnar major compaction {}", shard.tag()); let ok = try_wait( || { info!( "wait columnar major compaction {} l0 files: {}, l1 files: {}, l2 files: {}", shard.tag(), shard.get_data().col_levels.levels[0].files.len(), shard.get_data().col_levels.levels[1].files.len(), shard.get_data().col_levels.levels[2].files.len() ); !shard.get_data().col_levels.levels[2].files.is_empty() }, COLUMNAR_COMPACTION_WAIT_TIME, ); assert!(ok, "columnar major compaction failed"); let verify_columnar_for_table = |schema_file: &SchemaFile, snap: &SnapAccess, table_id: i64| { let schema = schema_file.get_table(table_id).unwrap(); let mut mvcc_reader = snap .new_columnar_mvcc_reader_without_pack_clean_read( table_id, &schema.columns, None, 500, None, ) .unwrap() .unwrap(); // Use a random end int handle let end_handle = thread_rng().gen_range(500..2100); block_on(mvcc_reader.set_int_handle_range(0, Some(end_handle))).unwrap(); let mut block = Block::new(schema); let mvcc_reader_counts = block_on(mvcc_reader.read_block(&mut block, usize::MAX)).unwrap(); info!("mvcc_reader read block with {} rows", mvcc_reader_counts); let mut columnar_readers: Vec> = vec![]; for tbl in l1_tbls.iter().chain(l2_tbls.iter()) { let data_bound = tbl.data_bound(); let (min_table_id, max_table_id) = get_table_id_from_data_bound(data_bound); if table_id < min_table_id || table_id > max_table_id { continue; } let iter = tbl.new_iterator(false, false); let reader = ColumnarRowTableReader::new(schema.clone(), iter, None, false, None).unwrap(); columnar_readers.push(Box::new(reader)); } let merged_reader = ColumnarMergeReader::new(schema.clone(), columnar_readers); let mut mvcc_row_reader = ColumnarMvccReader::new(Box::new(merged_reader), schema, 500); block_on(mvcc_row_reader.set_int_handle_range(0, Some(end_handle))).unwrap(); let mut row_block = Block::new(schema); let merge_reader_counts = block_on(mvcc_row_reader.read_block(&mut row_block, usize::MAX)).unwrap(); info!("merge_reader read block with {} rows", merge_reader_counts); verify_columnar_with_blocks(&row_block, &block); }; let snap = shard.new_snap_access(); for table_id in schema_file.export_backend_table_ids() { verify_columnar_for_table(&schema_file, &snap, table_id); } // Add table id 4 & 5 to schema file and remove table id 3 from schema file. let schemas_2 = vec![ schemas[0].clone(), schemas[1].clone(), schemas[3].clone(), schemas[4].clone(), ]; let schema_file_data_2 = build_schema_file(keyspace_id, schema_version + 1, schemas_2, 0); let schema_raw_file_2 = Arc::new(InMemFile::new( allocate_id(), Bytes::from(schema_file_data_2), )); fs.get_runtime() .block_on( fs.create( schema_raw_file_2.id(), schema_raw_file_2 .read(0, schema_raw_file_2.size() as usize) .unwrap(), dfs::Options::default().with_type(FileType::Schema), ), ) .unwrap(); let schema_file = SchemaFile::open(schema_raw_file_2).unwrap(); let ok = wait_columnar_compaction_idle(&engine, &shard); assert!(ok, "wait other compaction failed"); let mut builder = ShardDataBuilder::new(shard.get_data()); builder.set_schema(schema_file.get_version(), 0, Some(schema_file.clone())); shard.set_data(builder.build()); *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::ColumnarMajor { table_ids_to_add: vec![4, 5], table_ids_to_clear: vec![3], is_manual: false, schema_version: schema_file.get_version(), }); engine.trigger_compact(id_ver); info!("trigger columnar major compaction {}", shard.tag()); let ok = try_wait( || { info!( "wait columnar major compaction {} l0 files: {}, l1 files: {}, l2 files: {}, columnar_table_ids: {:?}", shard.tag(), shard.get_data().col_levels.levels[0].files.len(), shard.get_data().col_levels.levels[1].files.len(), shard.get_data().col_levels.levels[2].files.len(), shard.get_data().columnar_table_ids ); !shard.get_data().columnar_table_ids.contains(&3) && shard.get_data().columnar_table_ids.contains(&4) && shard.get_data().columnar_table_ids.contains(&5) }, COLUMNAR_COMPACTION_WAIT_TIME, ); assert!(ok, "columnar major compaction failed"); let snap = shard.new_snap_access(); let table_ids = schema_file.export_backend_table_ids(); assert!(table_ids.contains(&4)); assert!(table_ids.contains(&5)); assert!(!table_ids.contains(&3)); for table_id in table_ids { verify_columnar_for_table(&schema_file, &snap, table_id); } let ok = wait_columnar_compaction_idle(&engine, &shard); assert!(ok, "wait other compaction failed"); // Test manual columnar major compaction. let data = shard.get_data(); shard.set_data(ShardDataBuilder::new(data).build()); *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::ColumnarMajor { table_ids_to_add: vec![1, 2, 4, 5], table_ids_to_clear: vec![], is_manual: true, schema_version: schema_file.get_version(), }); engine.trigger_compact(id_ver); info!("trigger manual columnar major compaction {}", shard.tag()); let old_file_ids = HashSet::from_iter(shard.get_all_col_files().iter().cloned()); let ok = wait_columnar_compaction_idle(&engine, &shard); assert!(ok, "manual columnar major compaction failed"); let new_file_ids = HashSet::from_iter(shard.get_all_col_files().iter().cloned()); assert_ne!(old_file_ids, new_file_ids); let table_ids = shard.get_data().columnar_table_ids.clone(); let snap = shard.new_snap_access(); assert_eq!(table_ids, vec![1, 2, 4, 5]); for table_id in table_ids { verify_columnar_for_table(&schema_file, &snap, table_id); } // Remove columnar compaction. let data = shard.get_data(); shard.set_data(ShardDataBuilder::new(data).build()); *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::ColumnarClear); engine.trigger_compact(id_ver); info!("trigger remove columnar compaction {}", shard.tag()); let ok = try_wait( || { info!( "wait remove columnar compaction {} l0 files: {}, l1 files: {}, l2 files: {}", shard.tag(), shard.get_data().col_levels.levels[0].files.len(), shard.get_data().col_levels.levels[1].files.len(), shard.get_data().col_levels.levels[2].files.len() ); shard.get_data().col_levels.levels[2].files.is_empty() }, COLUMNAR_COMPACTION_WAIT_TIME, ); assert!(ok); assert_eq!(shard.get_columnar_snap_version(), SnapVersion::zero()); assert!(shard.get_data().schema_file.is_some()); } #[test] fn test_columnar_destroy_range() { ::test_util::init_log_for_test(); let keyspace_id = 1; let table_id = 30; let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, ""); let allocate_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 = new_schema(table_id, true); let schemas = vec![schema.clone()]; let schema_version = 10; let schema_file_data = build_schema_file(keyspace_id, schema_version, schemas, 0); let fs = engine.fs.clone(); let schema_raw_file = Arc::new(InMemFile::new(allocate_id(), Bytes::from(schema_file_data))); fs.get_runtime() .block_on( fs.create( schema_raw_file.id(), schema_raw_file .read(0, schema_raw_file.size() as usize) .unwrap(), dfs::Options::default().with_type(FileType::Schema), ), ) .unwrap(); let schema_file = SchemaFile::open(schema_raw_file).unwrap(); let opts = dfs::Options::default().with_type(FileType::Columnar); let (l0_tbl_0, _) = build_table(allocate_id(), &schema, 0, 600, 500); let (l0_tbl_1, _) = build_table(allocate_id(), &schema, 300, 900, 500); let (l0_tbl_2, _) = build_table(allocate_id(), &schema, 1000, 2000, 500); let (l1_tbl_0, _) = build_table(allocate_id(), &schema, 0, 600, 300); let (l1_tbl_1, _) = build_table(allocate_id(), &schema, 300, 900, 400); let (l1_tbl_2, _) = build_table(allocate_id(), &schema, 1000, 1500, 400); let (l2_tbl_0, _) = build_table(allocate_id(), &schema, 0, 1000, 100); let (l2_tbl_1, _) = build_table(allocate_id(), &schema, 1000, 2000, 100); for file in [ &l0_tbl_0, &l0_tbl_1, &l0_tbl_2, &l1_tbl_0, &l1_tbl_1, &l1_tbl_2, &l2_tbl_0, &l2_tbl_1, ] { info!("build file_id: {}, file size: {}", file.id(), file.size()); fs.get_runtime() .block_on(fs.create(file.id(), file.read(0, file.size() as usize).unwrap(), opts)) .unwrap() } let columnar_meta_cache = ColumnarMetaCache::default(); let mut col_levels = ColumnarLevels::new(); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_0, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_1, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_2, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_0, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_1, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_2, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 2, ColumnarFile::open(l2_tbl_0, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 2, ColumnarFile::open(l2_tbl_1, None, columnar_meta_cache.clone()).unwrap(), ); 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]); shard.set_data(builder.build()); let mut del_prefixes = DeletePrefixes::new_with_keyspace_id(shard.keyspace_id); let mut table_prefix = api_version::ApiV2::get_txn_keyspace_prefix(keyspace_id); table_prefix.extend_from_slice(b"t"); del_prefixes.merge_prefix_in_place(&table_prefix); shard.set_property(DEL_PREFIXES_KEY, &del_prefixes.marshal()); shard.initial_flushed.store(true, Ordering::SeqCst); let id_ver = shard.id_ver(); *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::DestroyRange); engine.trigger_compact(id_ver); info!("trigger columnar destroy range compaction {}", shard.tag()); let ok = try_wait( || { shard .get_data() .col_levels .levels .iter() .any(|l| l.files.is_empty()) }, COLUMNAR_COMPACTION_WAIT_TIME, ); assert!(ok, "columnar destroy range compaction failed"); } #[test] fn test_columnar_truncate_ts() { ::test_util::init_log_for_test(); let keyspace_id = KEYSPACE_ID; let table_id = 30; let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, ""); let allocate_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 = new_schema(table_id, false); let schemas = vec![schema.clone()]; let schema_version = 10; let schema_file_data = build_schema_file(keyspace_id, schema_version, schemas, 0); let fs = engine.fs.clone(); let schema_raw_file = Arc::new(InMemFile::new(allocate_id(), Bytes::from(schema_file_data))); fs.get_runtime() .block_on( fs.create( schema_raw_file.id(), schema_raw_file .read(0, schema_raw_file.size() as usize) .unwrap(), dfs::Options::default().with_type(FileType::Schema), ), ) .unwrap(); let schema_file = SchemaFile::open(schema_raw_file).unwrap(); let opts = dfs::Options::default().with_type(FileType::Columnar); let (l0_tbl_0, _) = build_table(allocate_id(), &schema, 0, 600, 500); let (l0_tbl_1, _) = build_table(allocate_id(), &schema, 300, 900, 500); let (l0_tbl_2, _) = build_table(allocate_id(), &schema, 1000, 2000, 500); let (l1_tbl_0, _) = build_table(allocate_id(), &schema, 0, 600, 300); let (l1_tbl_1, _) = build_table(allocate_id(), &schema, 300, 900, 400); let (l1_tbl_2, _) = build_table(allocate_id(), &schema, 1000, 1500, 400); let (l2_tbl_0, _) = build_table(allocate_id(), &schema, 0, 1000, 100); let (l2_tbl_1, _) = build_table(allocate_id(), &schema, 1000, 2000, 100); for file in [ &l0_tbl_0, &l0_tbl_1, &l0_tbl_2, &l1_tbl_0, &l1_tbl_1, &l1_tbl_2, &l2_tbl_0, &l2_tbl_1, ] { info!("build file_id: {}, file size: {}", file.id(), file.size()); fs.get_runtime() .block_on(fs.create(file.id(), file.read(0, file.size() as usize).unwrap(), opts)) .unwrap() } let columnar_meta_cache = ColumnarMetaCache::default(); let mut col_levels = ColumnarLevels::new(); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_0, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_1, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_2, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_0, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_1, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_2, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 2, ColumnarFile::open(l2_tbl_0, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 2, ColumnarFile::open(l2_tbl_1, None, columnar_meta_cache.clone()).unwrap(), ); 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]); shard.set_data(builder.build()); shard.initial_flushed.store(true, Ordering::SeqCst); let id_ver = shard.id_ver(); shard.pending_ops.write().unwrap().truncate_ts = Some(200); *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::TruncateTs); engine.trigger_compact(id_ver); info!("trigger columnar truncate ts compaction {}", shard.tag()); let ok = try_wait( || { shard.get_data().col_levels.levels[0].files.is_empty() && shard.get_data().col_levels.levels[1].files.is_empty() && shard.get_data().col_levels.levels[2].files.len() == 2 }, COLUMNAR_COMPACTION_WAIT_TIME, ); assert!(ok, "columnar truncate_ts compaction failed"); let snap = shard.new_snap_access(); let mut mvcc_reader = snap .new_columnar_mvcc_reader_without_pack_clean_read( table_id, &schema.columns, None, u64::MAX, None, ) .unwrap() .unwrap(); block_on(mvcc_reader.set_int_handle_range(0, Some(3000))).unwrap(); let mut block = Block::new(&schema); let counts = block_on(mvcc_reader.read_block(&mut block, usize::MAX)).unwrap(); for i in 0..counts { assert_eq!(100, block.versions.get_version(i)); } } #[test] fn test_columnar_trim_over_bound() { ::test_util::init_log_for_test(); let keyspace_id = KEYSPACE_ID; let table_id = 30; let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, ""); let allocate_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 = new_schema(table_id, false); let schemas = vec![schema.clone()]; let schema_version = 10; let schema_file_data = build_schema_file(keyspace_id, schema_version, schemas, 0); let fs = engine.fs.clone(); let schema_raw_file = Arc::new(InMemFile::new(allocate_id(), Bytes::from(schema_file_data))); fs.get_runtime() .block_on( fs.create( schema_raw_file.id(), schema_raw_file .read(0, schema_raw_file.size() as usize) .unwrap(), dfs::Options::default().with_type(FileType::Schema), ), ) .unwrap(); let schema_file = SchemaFile::open(schema_raw_file).unwrap(); let opts = dfs::Options::default().with_type(FileType::Columnar); let (l0_tbl_0, _) = build_table(allocate_id(), &schema, 0, 600, 500); let (l0_tbl_1, _) = build_table(allocate_id(), &schema, 300, 900, 600); let (l0_tbl_2, _) = build_table(allocate_id(), &schema, 1000, 2000, 510); let (l1_tbl_0, _) = build_table(allocate_id(), &schema, 0, 600, 300); let (l1_tbl_1, _) = build_table(allocate_id(), &schema, 300, 900, 410); let (l1_tbl_2, _) = build_table(allocate_id(), &schema, 1000, 1500, 420); let (l2_tbl_0, _) = build_table(allocate_id(), &schema, 0, 1000, 100); let (l2_tbl_1, _) = build_table(allocate_id(), &schema, 1000, 2000, 110); for file in [ &l0_tbl_0, &l0_tbl_1, &l0_tbl_2, &l1_tbl_0, &l1_tbl_1, &l1_tbl_2, &l2_tbl_0, &l2_tbl_1, ] { info!("build file_id: {}, file size: {}", file.id(), file.size()); fs.get_runtime() .block_on(fs.create(file.id(), file.read(0, file.size() as usize).unwrap(), opts)) .unwrap() } let columnar_meta_cache = ColumnarMetaCache::default(); let mut col_levels = ColumnarLevels::new(); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_0, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_1, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_2, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_0, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_1, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 1, ColumnarFile::open(l1_tbl_2, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 2, ColumnarFile::open(l2_tbl_0, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 2, ColumnarFile::open(l2_tbl_1, None, columnar_meta_cache.clone()).unwrap(), ); 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]); shard.set_data(builder.build()); // Split shard by handle key 800 let split_key = [keyspace_prefix(keyspace_id), encode_row_key(table_id, 800)].concat(); let mut splitter = Splitter::new(vec![split_key], IdVer::new(4, 4), 5, apply_tx); let handle = thread::spawn(move || { splitter.run(); }); let ok = try_wait(|| engine.shards.len() == 6, 2); assert!(ok, "split failed"); handle.join().unwrap(); let shard = engine.get_shard(shard_id).unwrap(); // Read before trim over bound data. shard.initial_flushed.store(true, Ordering::SeqCst); let snap = shard.new_snap_access(); let mut mvcc_reader = snap .new_columnar_mvcc_reader_without_pack_clean_read( table_id, &schema.columns, None, u64::MAX, None, ) .unwrap() .unwrap(); block_on(mvcc_reader.set_int_handle_range(0, Some(3000))).unwrap(); let mut block = Block::new(&schema); let read_before_trim = block_on(mvcc_reader.read_block(&mut block, usize::MAX)).unwrap(); let id_ver = shard.id_ver(); shard.pending_ops.write().unwrap().trim_over_bound = true; *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::TrimOverBound); engine.trigger_compact(id_ver); info!( "trigger columnar trim over bound compaction {}", shard.tag() ); let ok = try_wait( || shard.get_data().get_col_table_counts(0) == 2, COLUMNAR_COMPACTION_WAIT_TIME, ); assert!(ok, "columnar trim over bound compaction failed"); std::thread::sleep(Duration::from_secs(2)); let snap = shard.new_snap_access(); let mut mvcc_reader = snap .new_columnar_mvcc_reader_without_pack_clean_read( table_id, &schema.columns, None, u64::MAX, None, ) .unwrap() .unwrap(); block_on(mvcc_reader.set_int_handle_range(0, Some(3000))).unwrap(); let mut block = Block::new(&schema); let read_after_trim = block_on(mvcc_reader.read_block(&mut block, usize::MAX)).unwrap(); info!( "read_before_trim: {}, read_after_trim: {}", read_before_trim, read_after_trim ); assert!(read_before_trim > read_after_trim); for i in 0..read_after_trim { let handle = i64::from_le_bytes( std::convert::TryInto::try_into(block.handles.get_not_null_value(i)).unwrap(), ); assert!(handle >= 800); let version = block.versions.get_version(i); if handle == 800 { assert!(version == 600 || version == 599 || version == 598); } } // Check old version exists let mut mvcc_reader = snap .new_columnar_mvcc_reader_without_pack_clean_read( table_id, &schema.columns, None, 300, None, ) .unwrap() .unwrap(); block_on(mvcc_reader.set_int_handle_range(0, Some(3000))).unwrap(); let mut block = Block::new(&schema); let read_old_version = block_on(mvcc_reader.read_block(&mut block, usize::MAX)).unwrap(); for i in 0..read_old_version { let handle = i64::from_le_bytes( std::convert::TryInto::try_into(block.handles.get_not_null_value(i)).unwrap(), ); if handle == 800 { assert!(block.versions.get_version(i) <= 100); } } } #[test] fn test_min_max_index_not_in() { // not-in condition_values, data, has_null, expected let cases: Vec<(Vec, Vec, bool, bool)> = vec![ (vec![2, 3, 5], vec![1, 3, 10], false, true), (vec![2, 3, 5], vec![1, 4, 6, 10], false, true), (vec![2, 3, 5], vec![0, 3, 15], false, true), (vec![2, 3, 5], vec![1, 1, 1], false, true), (vec![2, 3, 5], vec![3, 3, 3], false, false), (vec![2, 3, 5], vec![3, 3, 3], true, true), ]; for (condition_values, data, has_null, expected) in cases { let mut min_max_index = MinMaxIndex::new(1, 8, true); let min = *data.iter().min().unwrap(); let max = *data.iter().max().unwrap(); min_max_index.push_value(&min.to_le_bytes()); min_max_index.push_value(&max.to_le_bytes()); let values = condition_values .iter() .map(|v| v.to_le_bytes().to_vec()) .collect::>>(); min_max_index.push_has_null(has_null); min_max_index.push_has_value(true); assert_eq!( min_max_index.check_not_in( 0, &values, tidb_query_datatype::FieldTypeTp::LongLong, false ), expected ); } } fn verify_columnar_with_blocks(b1: &Block, b2: &Block) { assert_eq!(b1.length(), b2.length()); let len = b1.length(); for i in 0..len { assert_eq!( b1.handles.get_not_null_value(i), b2.handles.get_not_null_value(i), "{}", i ); assert_eq!(b1.versions.get_version(i), b2.versions.get_version(i)); assert_eq!(b1.versions.is_null(i), b2.versions.is_null(i)); assert_eq!(b1.columns[0].is_null(i), b2.columns[0].is_null(i)); assert_eq!(b1.columns[1].is_null(i), b2.columns[1].is_null(i)); assert_eq!( b1.columns[0].get_value(i).unwrap().get_i64_le(), b2.columns[0].get_value(i).unwrap().get_i64_le() ); assert_eq!( b1.columns[1].get_value(i).unwrap(), b2.columns[1].get_value(i).unwrap() ); } } fn new_sst_table_for_columnar( engine: &TestEngine, id: u64, table_id: i64, begin: usize, end: usize, version: u64, saved_vals: &mut Vec>>, ) -> SsTable { let block_size = engine.opts.table_builder_options.block_size; let comp_tp = engine.opts.table_builder_options.compression_tps[0]; let comp_lvl = engine.opts.table_builder_options.compression_lvl; let fs = engine.fs.clone(); let mut builder = table::sstable::builder::Builder::new( id, block_size, comp_tp, comp_lvl, ChecksumType::Crc32, None, ); let ctx = Mutex::new(EvalContext::default()); for i in begin..end { let key = engine.key_builder.gen_row_inner_key(table_id, i); let val = Rc::new(engine.key_builder.gen_row_val(&ctx, i)); // Save the val with Rc, make sure the val_rc stay valid during the iterator // lifecycle. saved_vals.push(val.clone()); let val = table::Value::new_with_meta_version(0, version, 0, &val); builder.add(key.as_ref(), &val, None).unwrap(); } let mut data_buf = Vec::new(); builder.finish(0, &mut data_buf).unwrap(); let data = Bytes::from(data_buf); let opts = dfs::Options::default(); let runtime = fs.get_runtime(); runtime.block_on(fs.create(id, data.clone(), opts)).unwrap(); let file = InMemFile::new(id, data); SsTable::new( Arc::new(file), BlockCache::None, cloud_encryption::EncryptionState::Disabled, ) .unwrap() } #[test] fn test_columnar_not_nullable_to_nullable() { ::test_util::init_log_for_test(); let keyspace_id = 1; let table_id = 30; let table_id2 = 31; let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, ""); let allocate_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 = new_schema_with_nullable(table_id, true, false); let schemas = vec![schema.clone()]; let schema_version = 10; let schema_file_data = build_schema_file(keyspace_id, schema_version, schemas, 0); let fs = engine.fs.clone(); let schema_raw_file = Arc::new(InMemFile::new(allocate_id(), Bytes::from(schema_file_data))); fs.get_runtime() .block_on( fs.create( schema_raw_file.id(), schema_raw_file .read(0, schema_raw_file.size() as usize) .unwrap(), dfs::Options::default().with_type(FileType::Schema), ), ) .unwrap(); let schema_file = SchemaFile::open(schema_raw_file).unwrap(); let opts = dfs::Options::default().with_type(FileType::Columnar); let (l0_tbl_0, _) = build_table(allocate_id(), &schema, 0, 600, 300); let (l0_tbl_1, _) = build_table(allocate_id(), &schema, 300, 900, 400); for file in [&l0_tbl_0, &l0_tbl_1] { info!("build file_id: {}, file size: {}", file.id(), file.size()); fs.get_runtime() .block_on(fs.create(file.id(), file.read(0, file.size() as usize).unwrap(), opts)) .unwrap() } let columnar_meta_cache = ColumnarMetaCache::default(); let mut col_levels = ColumnarLevels::new(); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_0, None, columnar_meta_cache.clone()).unwrap(), ); col_levels.add_file( 0, ColumnarFile::open(l0_tbl_1, None, columnar_meta_cache.clone()).unwrap(), ); 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, table_id2]); shard.set_data(builder.build()); shard.initial_flushed.store(true, Ordering::SeqCst); let id_ver = shard.id_ver(); *shard.compaction_priority.write().unwrap() = Some(CompactionPriority::ColumnarL0 { score: 2.0 }); engine.trigger_compact(id_ver); info!("trigger columnar l0 compaction {}", shard.tag()); let ok = try_wait( || { info!( "wait columnar l0 compaction {} l0 files: {}, l1 files: {}", shard.tag(), shard.get_data().col_levels.levels[0].files.len(), shard.get_data().col_levels.levels[1].files.len() ); shard.get_data().col_levels.levels[0].files.is_empty() }, COLUMNAR_COMPACTION_WAIT_TIME, ); assert!(ok, "columnar l0 compaction failed"); let snap = shard.new_snap_access(); // Read with schema with nullable columns. let new_schema = new_schema_with_nullable(table_id, true, true); let mut mvcc_reader = snap .new_columnar_mvcc_reader_without_pack_clean_read( table_id, &new_schema.columns, None, 500, None, ) .unwrap() .unwrap(); block_on(mvcc_reader.set_handle_range(&i_to_common_handle(0), &i_to_common_handle(2100))) .unwrap(); let mut block = Block::new(&schema); block_on(mvcc_reader.read_block(&mut block, usize::MAX)).unwrap(); } #[test] fn test_store_usage_stats_excludes_columnar_from_row_size() { ::test_util::init_log_for_test(); let keyspace_id = 1; let table_id = 40; let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, ""); let allocate_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 = new_schema(table_id, true); let opts = dfs::Options::default().with_type(FileType::Columnar); let (col_file, _) = build_table(allocate_id(), &schema, 0, 100, 50); let columnar_size = col_file.size(); engine .fs .get_runtime() .block_on(engine.fs.create( col_file.id(), col_file.read(0, col_file.size() as usize).unwrap(), opts, )) .unwrap(); let before = engine.collect_store_usage_stats(); let mut col_levels = ColumnarLevels::new(); col_levels.add_file( 0, ColumnarFile::open(col_file, None, ColumnarMetaCache::default()).unwrap(), ); let mut builder = ShardDataBuilder::new(shard.get_data()); builder.set_columnar_levels(col_levels); builder.set_columnar_table_ids(vec![table_id]); 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 columnar files are added" ); assert_eq!( after.leader_columnar_size, before.leader_columnar_size + columnar_size ); assert_eq!(after.leader_fts_size, before.leader_fts_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::() + engine_stats .fts_levels .iter() .map(|l| l.data_size) .sum::() + engine_stats.vector_indexes.data_size, engine_stats.total_data_size ); assert_eq!( after.leader_columnar_size, engine_stats .columnar_levels .iter() .map(|l| l.data_size) .sum::() ); 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_columnar_size, 0); assert_eq!(inactive.leader_fts_size, 0); assert_eq!(inactive.leader_vector_size, 0); // total_data_size still includes inactive-shard format sizes; do not sum // leader_*. assert_eq!( inactive.row_size + engine_stats .columnar_levels .iter() .map(|l| l.data_size) .sum::() + engine_stats .fts_levels .iter() .map(|l| l.data_size) .sum::() + engine_stats.vector_indexes.data_size, engine_stats.total_data_size ); } fn build_test_vector_index_file(file_id: u64, table_id: i64) -> (VectorIndexFile, Bytes) { const DIMENSION: usize = 3; let snap_version = SnapVersion::from(1); let mut builder = VectorIndexBuilder::new(DIMENSION, "cosine", snap_version, table_id, 1, 1, false).unwrap(); let mut vec_col_info = tipb::ColumnInfo::new(); vec_col_info.set_column_id(1); vec_col_info.set_tp(FieldTypeTp::TiDbVectorFloat32 as i32); vec_col_info.set_flen(DIMENSION as isize); let schema = Schema::new(SchemaBuf::new( table_id, new_int_handle_column_info(), new_version_column_info(), vec![vec_col_info], vec![], 0, vec![], vec![], StorageClassSpec::default(), None, )); let mut block = Block::new(&schema); for i in 0i64..4 { block.handles.push_value(&i.to_le_bytes()); block .versions .push_version(snap_version.into_inner(), false); let vec_f32 = VectorFloat32::copy_from_f32(&[i as f32, (i + 1) as f32, (i + 2) as f32]); let mut buf = vec![]; buf.write_vector_float32(vec_f32.as_ref()).unwrap(); block.columns[0].push_value(&buf); } builder.add_block(&block, 0).unwrap(); let data = Bytes::from(builder.build().unwrap()); let vec_file = VectorIndexFile::new(Arc::new(InMemFile::new(file_id, data.clone())), 0, None).unwrap(); (vec_file, data) } #[test] fn test_store_usage_stats_excludes_vector_from_row_size() { ::test_util::init_log_for_test(); let keyspace_id = 1; let table_id = 42; let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, ""); let allocate_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 file_id = allocate_id(); let (vec_file, data) = build_test_vector_index_file(file_id, table_id); let vector_size = data.len() as u64; engine .fs .get_runtime() .block_on(engine.fs.create( file_id, data, dfs::Options::default().with_type(FileType::VectorIndex), )) .unwrap(); let before = engine.collect_store_usage_stats(); let mut vector_indexes = VectorIndexes::default(); vector_indexes.add_index_file(vec_file); let mut builder = ShardDataBuilder::new(shard.get_data()); builder.set_vector_indexes(vector_indexes); 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 vector index files are added" ); assert_eq!( after.leader_vector_size, before.leader_vector_size + vector_size ); assert_eq!(after.leader_columnar_size, before.leader_columnar_size); assert_eq!(after.leader_fts_size, before.leader_fts_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::() + engine_stats .fts_levels .iter() .map(|l| l.data_size) .sum::() + engine_stats.vector_indexes.data_size, engine_stats.total_data_size ); assert_eq!( after.leader_vector_size, engine_stats.vector_indexes.data_size ); 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_vector_size, 0); assert_eq!( inactive.row_size + engine_stats .columnar_levels .iter() .map(|l| l.data_size) .sum::() + engine_stats .fts_levels .iter() .map(|l| l.data_size) .sum::() + engine_stats.vector_indexes.data_size, engine_stats.total_data_size ); } #[test] fn test_lm_is_supported_does_not_advance_hybrid_cursor() { ::test_util::init_log_for_test(); let keyspace_id = 1; let table_id = 42; let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, ""); let allocate_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 = new_schema(table_id, false); let schema_version = 10; let schema_file_data = build_schema_file(keyspace_id, schema_version, vec![schema.clone()], 0); let fs = engine.fs.clone(); let schema_raw_file = Arc::new(InMemFile::new(allocate_id(), Bytes::from(schema_file_data))); fs.get_runtime() .block_on( fs.create( schema_raw_file.id(), schema_raw_file .read(0, schema_raw_file.size() as usize) .unwrap(), dfs::Options::default().with_type(FileType::Schema), ), ) .unwrap(); let schema_file = SchemaFile::open(schema_raw_file).unwrap(); let (l2_file, _) = build_table(allocate_id(), &schema, 0, 32, 100); fs.get_runtime() .block_on(fs.create( l2_file.id(), l2_file.read(0, l2_file.size() as usize).unwrap(), dfs::Options::default().with_type(FileType::Columnar), )) .unwrap(); let mut col_levels = ColumnarLevels::new(); col_levels.add_file( 2, ColumnarFile::open(l2_file, None, ColumnarMetaCache::default()).unwrap(), ); 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]); shard.set_data(builder.build()); let mut range_a = tipb::KeyRange::default(); let mut low_a = keyspace_prefix(keyspace_id); low_a.extend_from_slice(&encode_row_key(table_id, i64::MIN)); let mut high_a = keyspace_prefix(keyspace_id); high_a.extend_from_slice(&encode_row_key(table_id, 16)); range_a.set_low(low_a); range_a.set_high(high_a); let mut range_b = tipb::KeyRange::default(); let mut low_b = keyspace_prefix(keyspace_id); low_b.extend_from_slice(&encode_row_key(table_id, 16)); let mut high_b = keyspace_prefix(keyspace_id); high_b.extend_from_slice(&encode_row_key(table_id, i64::MAX)); range_b.set_low(low_b); range_b.set_high(high_b); let ranges = vec![range_a, range_b]; let mut tbl_scan = tipb::TableScan::default(); tbl_scan.set_table_id(table_id); let mut executor = tipb::Executor::default(); executor.set_tbl_scan(tbl_scan); let scan_ctx = TableScanCtx::new(executor, vec![]); let runtime = tokio::runtime::Builder::new_multi_thread() .thread_name("test-lm-is-supported") .enable_all() .worker_threads(2) .build() .unwrap(); let mut reader = CloudColumnarReader::new( runtime.handle().clone(), shard.new_snap_access(), IaCtx::disabled(&crate::ia::util::IaConfig::default()), table_id, ranges, &schema.columns, &scan_ctx, None, None, 200, ) .unwrap(); // handle + version + one filter column; column 2 remains late. let early_column_ids = [HANDLE_COL_ID as i64, VERSION_COL_ID as i64, 1i64]; assert!( reader .ffi_is_late_materialization_supported(&early_column_ids) .unwrap() ); assert!( !reader.hybrid_reader_is_some_for_test(), "is_supported must not construct a hybrid reader" ); assert!( !reader.hybrid_early_column_ids_is_some_for_test(), "is_supported must not pin the early projection" ); assert!( !reader.hybrid_inited_for_test(), "is_supported must not seek the hybrid cursor" ); assert_eq!( reader.hybrid_ranges_len_for_test(), 0, "is_supported must not clone/consume hybrid ranges" ); assert_eq!(reader.ranges_len_for_test(), 2); assert_eq!( reader.runtime_stats().remote_segments, 0, "is_supported must not record segment stats" ); assert_eq!(reader.runtime_stats().total_segments, 0); let early = reader .ffi_read_early_block(64, &early_column_ids) .unwrap() .expect("expected a non-empty early batch"); assert!(early.1 > 0); assert!( reader.hybrid_inited_for_test(), "first early read should initialize the hybrid cursor" ); assert_eq!( reader.hybrid_ranges_len_for_test(), 1, "first early read should consume exactly one hybrid range" ); } #[test] fn test_lm_is_supported_does_not_pin_early_projection() { ::test_util::init_log_for_test(); let keyspace_id = 1; let table_id = 42; let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, ""); let allocate_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 = new_schema(table_id, false); let schema_version = 10; let schema_file_data = build_schema_file(keyspace_id, schema_version, vec![schema.clone()], 0); let fs = engine.fs.clone(); let schema_raw_file = Arc::new(InMemFile::new(allocate_id(), Bytes::from(schema_file_data))); fs.get_runtime() .block_on( fs.create( schema_raw_file.id(), schema_raw_file .read(0, schema_raw_file.size() as usize) .unwrap(), dfs::Options::default().with_type(FileType::Schema), ), ) .unwrap(); let schema_file = SchemaFile::open(schema_raw_file).unwrap(); let (l2_file, _) = build_table(allocate_id(), &schema, 0, 32, 100); fs.get_runtime() .block_on(fs.create( l2_file.id(), l2_file.read(0, l2_file.size() as usize).unwrap(), dfs::Options::default().with_type(FileType::Columnar), )) .unwrap(); let mut col_levels = ColumnarLevels::new(); col_levels.add_file( 2, ColumnarFile::open(l2_file, None, ColumnarMetaCache::default()).unwrap(), ); 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]); shard.set_data(builder.build()); let mut range = tipb::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); let mut tbl_scan = tipb::TableScan::default(); tbl_scan.set_table_id(table_id); let mut executor = tipb::Executor::default(); executor.set_tbl_scan(tbl_scan); let scan_ctx = TableScanCtx::new(executor, vec![]); let runtime = tokio::runtime::Builder::new_multi_thread() .thread_name("test-lm-probe-projection") .enable_all() .worker_threads(2) .build() .unwrap(); let mut reader = CloudColumnarReader::new( runtime.handle().clone(), shard.new_snap_access(), IaCtx::disabled(&crate::ia::util::IaConfig::default()), table_id, vec![range], &schema.columns, &scan_ctx, None, None, 200, ) .unwrap(); let probe_early = [HANDLE_COL_ID as i64, VERSION_COL_ID as i64, 1i64]; let read_early = [HANDLE_COL_ID as i64, VERSION_COL_ID as i64, 2i64]; assert!( reader .ffi_is_late_materialization_supported(&probe_early) .unwrap() ); assert!(!reader.hybrid_reader_is_some_for_test()); let early = reader .ffi_read_early_block(64, &read_early) .unwrap() .expect("read with a different early projection should succeed"); assert!(early.1 > 0); assert!(reader.hybrid_reader_is_some_for_test()); } #[test] fn test_lm_read_early_hard_rejects_sequential_multi_table() { ::test_util::init_log_for_test(); let keyspace_id = 1; let table_id_a = 42; let table_id_b = 43; let (engine, apply_tx) = new_test_engine_opt(true, DEF_BLOCK_SIZE, ""); let allocate_id = || engine.id_allocator.alloc_id(1).unwrap()[0]; // prepare_table_region returns a shard covering table_id_a; add schema/files // for both tables. let shard_id = prepare_table_region(&engine, &apply_tx, keyspace_id, table_id_a); let shard = engine.get_shard(shard_id).unwrap(); let schema_a = new_schema(table_id_a, false); let schema_b = new_schema(table_id_b, false); let schema_version = 10; let schema_file_data = build_schema_file( keyspace_id, schema_version, vec![schema_a.clone(), schema_b.clone()], 0, ); let fs = engine.fs.clone(); let schema_raw_file = Arc::new(InMemFile::new(allocate_id(), Bytes::from(schema_file_data))); fs.get_runtime() .block_on( fs.create( schema_raw_file.id(), schema_raw_file .read(0, schema_raw_file.size() as usize) .unwrap(), dfs::Options::default().with_type(FileType::Schema), ), ) .unwrap(); let schema_file = SchemaFile::open(schema_raw_file).unwrap(); let (l2_a, _) = build_table(allocate_id(), &schema_a, 0, 16, 100); let (l2_b, _) = build_table(allocate_id(), &schema_b, 0, 16, 100); for file in [&l2_a, &l2_b] { fs.get_runtime() .block_on(fs.create( file.id(), file.read(0, file.size() as usize).unwrap(), dfs::Options::default().with_type(FileType::Columnar), )) .unwrap(); } let mut col_levels = ColumnarLevels::new(); col_levels.add_file( 2, ColumnarFile::open(l2_a, None, ColumnarMetaCache::default()).unwrap(), ); col_levels.add_file( 2, ColumnarFile::open(l2_b, None, ColumnarMetaCache::default()).unwrap(), ); 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_a, table_id_b]); shard.set_data(builder.build()); let make_range = |table_id: i64| { let mut range = tipb::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); range }; let mut tbl_scan = tipb::TableScan::default(); tbl_scan.set_table_id(table_id_a); let mut executor = tipb::Executor::default(); executor.set_tbl_scan(tbl_scan); let scan_ctx = TableScanCtx::new(executor, vec![]); let mut readers = CloudColumnarReaders::new( engine.fs.get_runtime().handle().clone(), shard.new_snap_access(), vec![ TableCtx { table_id: table_id_a, ranges: vec![make_range(table_id_a)], }, TableCtx { table_id: table_id_b, ranges: vec![make_range(table_id_b)], }, ], schema_a.columns.clone(), scan_ctx, None, None, IaCtx::disabled(&crate::ia::util::IaConfig::default()), 200, 0, // sequential multi-table (no worker aggregation) ) .unwrap(); let early_column_ids = [HANDLE_COL_ID as i64, VERSION_COL_ID as i64, 1i64]; assert!( !readers .ffi_is_late_materialization_supported(&early_column_ids) .unwrap(), "capability probe should soft-reject sequential multi-table" ); let err = readers .ffi_read_early_block(64, &early_column_ids) .expect_err("early read must hard-reject sequential multi-table"); assert!( err.to_string().contains("sequential multi-table readers"), "unexpected error: {}", err ); }