use std::collections::HashSet; use anyhow::{Context, Result, anyhow, bail}; use serde::{Deserialize, Serialize, Serializer}; use tantivy::{ schema::{ BytesOptions, FAST, INDEXED, IndexRecordOption, NumericOptions, STORED, STRING, Schema, TextFieldIndexing, TextOptions, }, tokenizer::{TextAnalyzer, TokenizerManager}, }; use tici_common::index::IndexKey; use tici_grpc::{ ColumnInfo, DEFAULT_PARSER_INFO, FullTextIndexInfo, HybridIndexInfo, HybridInverted, HybridSharding, HybridSort, IndexInfo, TableInfo, }; use crate::{ analysis::analyzer::{AnalyzerDef, create_analyzer_manager}, reader::query_info::{DocField, TantivySchemaExt}, }; pub const HANDLE_STORE_NAME: &str = "handle"; pub const VERSION_STORE_NAME: &str = "column_-1024"; // The column_id of _tidb_rowid is -1. For non-clustered index, store // _tidb_rowid as pk columns. pub const TIDB_ROW_ID_STORE_NAME: &str = "column_-1"; fn column_store_name(column_id: i64) -> String { format!("column_{}", column_id) } pub fn parse_column_id(name: &str) -> Result { name.strip_prefix("column_") .and_then(|s| s.parse::().ok()) .ok_or_else(|| anyhow!("Parse {} failed", name)) } // Helper trait for TiCi specific extensions to ColumnInfo pub trait ColumnInfoExt { fn store_name(&self) -> String; } impl ColumnInfoExt for ColumnInfo { fn store_name(&self) -> String { column_store_name(self.column_id) } } #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] #[serde(transparent)] pub struct IndexMetaVersion(u32); impl IndexMetaVersion { pub const V0: Self = Self(0); pub const LATEST: Self = Self::V0; } #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct SortColumn { pub col_id: i64, pub is_asc: bool, } #[derive(Clone, Debug)] pub struct SortField { pub field: DocField, pub is_asc: bool, } #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct InvertedColumn { pub col_id: i64, } #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct FulltextField { col_ids: Vec, analyzer_name: String, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct AnalyzedToken { pub text: String, pub position: usize, } #[derive(Clone, Debug, PartialEq, Eq)] pub enum PhraseAnalysis { NoMatch, Tokens(Vec), } impl FulltextField { pub fn new(mut col_ids: Vec, analyzer_name: String) -> FulltextField { col_ids.sort_unstable(); FulltextField { col_ids, analyzer_name, } } pub fn field_name(&self) -> String { // ft_id0-id1-.._analyzer let ids: String = self .col_ids .iter() .map(|id| id.to_string()) .collect::>() .join("-"); format!("ft_{}_{}", ids, self.analyzer_name) } pub fn analyzer_name(&self) -> &str { &self.analyzer_name } fn get_ft_options(&self) -> TextOptions { let text_field_indexing = TextFieldIndexing::default() .set_tokenizer(&self.analyzer_name) .set_index_option(IndexRecordOption::WithFreqsAndPositions); TextOptions::default().set_indexing_options(text_field_indexing) } /// Get the column infos for this fulltext field from the given column list. pub fn get_columns<'a>(&self, columns: &'a [ColumnInfo]) -> Result> { self.col_ids .iter() .map(|col_id| { columns .iter() .find(|c| c.column_id == *col_id) .with_context(|| format!("Column {} not found for fulltext field", col_id)) }) .collect() } /// Build fulltext content from stored column fields in the document. /// Returns a Vec of texts, one per column, to be added as multi-value /// field. This prevents phrase matching across column boundaries. pub fn build_texts_from_doc( &self, doc: &tantivy::TantivyDocument, schema: &tantivy::schema::Schema, columns: &[ColumnInfo], ) -> Result> { use tantivy::schema::Value; let cols = self.get_columns(columns)?; let mut texts = Vec::with_capacity(cols.len()); for col in cols { let col_field = schema .get_field(&col.store_name()) .with_context(|| format!("Field {} not found in schema", col.store_name()))?; if let Some(value) = doc.get_first(col_field) { if let Some(text) = value.as_str() { texts.push(text.to_string()); } } } Ok(texts) } } #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct ShardingColumn { pub col_id: i64, } #[derive(Clone, PartialEq, Debug, Serialize, Deserialize)] pub struct IndexMetaStorage { pub version: IndexMetaVersion, pub index_key: IndexKey, pub created_ts: u64, pub columns: Vec, // Columns in index + pk columns // columns[0..index_column_count] are index columns pub index_column_count: usize, // TODO: convert these columns into column infos in `IndexMetaRuntime` pub sort_columns: Vec, pub inverted_columns: Vec, pub fulltext_fields: Vec, pub sharding_columns: Vec, pub analyzer_defs: Vec, pub is_partition: bool, } #[derive(Clone)] pub struct IndexMetaRuntime { sharding_columns: Vec, tantivy_schema: Schema, sort_fields: Vec, analyzer_manager: TokenizerManager, is_int_handle: bool, } // TODO: TokenizerManager doesn't implement Debug. We can wrap it in a struct // and implement Debug for it. impl std::fmt::Debug for IndexMetaRuntime { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("IndexMetaRuntime") .field("sharding_columns", &self.sharding_columns) .field("tantivy_schema", &self.tantivy_schema) .field("sort_fields", &self.sort_fields) .field("is_int_handle", &self.is_int_handle) .finish() } } #[derive(Clone, Debug, Deserialize)] #[serde(try_from = "IndexMetaStorage")] pub struct IndexMeta { storage: IndexMetaStorage, #[serde(skip)] runtime: IndexMetaRuntime, } impl Serialize for IndexMeta { fn serialize(&self, serializer: S) -> Result where S: Serializer, { self.storage.serialize(serializer) } } impl TryFrom for IndexMeta { type Error = anyhow::Error; fn try_from(storage: IndexMetaStorage) -> Result { let sharding_columns = storage.build_sharding_columns()?; let tantivy_schema = storage.build_tantivy_schema()?; let sort_fields = storage.build_sort_fields(&tantivy_schema)?; let analyzer_manager = storage.build_analyzer_manager()?; let is_int_handle = storage.is_int_handle(); let runtime = IndexMetaRuntime { sharding_columns, tantivy_schema, sort_fields, analyzer_manager, is_int_handle, }; Ok(IndexMeta { storage, runtime }) } } impl IndexMeta { pub fn new(table_info: &TableInfo, index_info: &IndexInfo) -> Result { table_info.validate()?; index_info.validate(table_info)?; let mut columns = index_info.columns.clone(); let index_column_count = columns.len(); for pk_col in table_info.pk_columns() { if !columns.iter().any(|col| col.column_id == pk_col.column_id) { columns.push(pk_col.clone()); } } let index_key = IndexKey::new( table_info.keyspace_id, table_info.table_id, index_info.index_id, ); let is_partition = table_info.is_partition; let storage = if index_info.index_type == tici_grpc::IndexType::Fulltext as i32 { // create fulltext index ... // TODO: make parser_info required in index_info when tidb supports it. let parser_info = if let Some(parser_info) = index_info.parser_info.as_ref() { parser_info } else { &DEFAULT_PARSER_INFO }; Self::parse_fulltext_index( index_key, index_info.created_ts, columns, index_column_count, index_info.fulltext_index.as_str(), parser_info, is_partition, ) } else if index_info.index_type == tici_grpc::IndexType::Hybrid as i32 { // create hybrid index ... Self::parse_hybrid_index( index_key, index_info.created_ts, columns, index_column_count, index_info.hybrid_index.as_str(), is_partition, ) } else { Err(anyhow!("Unsupported index type: {}", index_info.index_type)) }?; Self::try_from(storage) } pub fn created_ts(&self) -> u64 { self.storage.created_ts } fn parse_fulltext_index( index_key: IndexKey, created_ts: u64, columns: Vec, index_column_count: usize, ft_info: &str, parser_info: &tici_grpc::ParserInfo, is_partition: bool, ) -> Result { let ft_info: FullTextIndexInfo = serde_json::from_str(ft_info) .with_context(|| format!("Failed to parse fulltext index info: {}", ft_info))?; let analyzer_def = AnalyzerDef::from_fulltext_index_info(&ft_info, parser_info)?; let col_ids = columns[0..index_column_count] .iter() .map(|col| col.column_id) .collect(); let fulltext_fields = vec![FulltextField::new(col_ids, analyzer_def.name().to_string())]; Ok(IndexMetaStorage { version: IndexMetaVersion::LATEST, index_key, created_ts, columns, index_column_count, sort_columns: vec![], inverted_columns: vec![], fulltext_fields, sharding_columns: vec![], analyzer_defs: vec![analyzer_def], is_partition, }) } fn parse_inverted( columns: &[ColumnInfo], inverted: &HybridInverted, ) -> Result> { let column_names: Vec = inverted .columns .iter() .map(|col| col.name.lower_case.clone()) .collect(); if Self::has_duplicates(&column_names) { bail!("Duplicate columns in `inverted` field: {:?}", column_names); } let mut inverted_columns = Vec::with_capacity(column_names.len()); for col_name in column_names { let col_id = columns .iter() .find(|c| c.column_name == *col_name) .ok_or_else(|| anyhow!("Column {} not found in hybrid index", col_name))? .column_id; inverted_columns.push(InvertedColumn { col_id }); } Ok(inverted_columns) } fn parse_sort(columns: &[ColumnInfo], sort: &HybridSort) -> Result> { let column_names: Vec = sort .columns .iter() .map(|col| col.name.lower_case.clone()) .collect(); if Self::has_duplicates(&column_names) { bail!("Duplicate columns in `sort` field: {:?}", column_names); } let is_asc = &sort.is_asc; if column_names.len() != is_asc.len() { bail!( "Mismatched number of columns and order in hybrid index: {} != {}", column_names.len(), is_asc.len() ); } let mut sort_columns = Vec::with_capacity(column_names.len()); for (col, is_asc) in column_names.iter().zip(is_asc.iter()) { let col_id = columns .iter() .find(|c| c.column_name == *col) .ok_or_else(|| anyhow!("Column {} not found in hybrid index", col))? .column_id; sort_columns.push(SortColumn { col_id, is_asc: *is_asc, }) } Ok(sort_columns) } fn parse_sharding( columns: &[ColumnInfo], sharding: &HybridSharding, ) -> Result> { let column_names: Vec = sharding .columns .iter() .map(|col| col.name.lower_case.clone()) .collect(); if Self::has_duplicates(&column_names) { bail!( "Duplicate columns in `sharding_key` field: {:?}", column_names ); } let mut sharding_columns = Vec::with_capacity(column_names.len()); for col_name in column_names { let col_id = columns .iter() .find(|c| c.column_name == *col_name) .ok_or_else(|| anyhow!("Column {} not found in hybrid index", col_name))? .column_id; sharding_columns.push(ShardingColumn { col_id }); } Ok(sharding_columns) } fn parse_hybrid_index( index_key: IndexKey, created_ts: u64, columns: Vec, index_column_count: usize, hybrid_index: &str, is_partition: bool, ) -> Result { let hybrid_index_info = if hybrid_index.is_empty() { HybridIndexInfo::default() } else { serde_json::from_str(hybrid_index) .with_context(|| format!("Failed to parse hybrid index: {}", hybrid_index))? }; let sort_columns = if let Some(sort) = hybrid_index_info.sort.as_ref() { Self::parse_sort(&columns, sort)? } else { vec![] }; let inverted_columns = if let Some(inverted) = hybrid_index_info.inverted.as_ref() { let inverted = inverted .first() .ok_or_else(|| anyhow!("Empty `inverted` field in hybrid index"))?; Self::parse_inverted(&columns, inverted)? } else { vec![] }; // TODO: parse fulltext let sharding_columns = if let Some(sharding) = hybrid_index_info.sharding_key.as_ref() { Self::parse_sharding(&columns, sharding)? } else { vec![] }; Ok(IndexMetaStorage { version: IndexMetaVersion::LATEST, index_key, created_ts, columns, index_column_count, sort_columns, inverted_columns, fulltext_fields: vec![], sharding_columns, analyzer_defs: vec![], is_partition, }) } fn has_duplicates(v: &[String]) -> bool { let mut seen = HashSet::with_capacity(v.len()); for item in v { if !seen.insert(item) { return true; } } false } pub fn index_key(&self) -> &IndexKey { &self.storage.index_key } pub fn is_partition(&self) -> bool { self.storage.is_partition } pub fn get_analyzer_manager(&self) -> &TokenizerManager { &self.runtime.analyzer_manager } pub fn get_tantivy_schema(&self) -> &Schema { &self.runtime.tantivy_schema } pub fn get_sort_fields(&self) -> &[SortField] { &self.runtime.sort_fields } pub fn get_sharding_columns(&self) -> &[ColumnInfo] { &self.runtime.sharding_columns } pub fn get_column_infos(&self) -> &[ColumnInfo] { &self.storage.columns } pub fn get_fulltext_fields(&self) -> &[FulltextField] { &self.storage.fulltext_fields } pub fn get_fulltext_field(&self, mut col_ids: Vec) -> Result<&FulltextField> { col_ids.sort_unstable(); self.storage .fulltext_fields .iter() .find(|f| f.col_ids == col_ids) .ok_or_else(|| anyhow!("Fulltext field of column {:?} not found", col_ids)) } /// Build and add fulltext fields to the document from stored column fields. /// Each column's text is added as a separate value (multi-value field), /// which prevents phrase matching across column boundaries. /// Returns the total bytes added. pub fn build_fulltext_fields(&self, doc: &mut tantivy::TantivyDocument) -> Result { let schema = self.get_tantivy_schema(); let columns = self.get_column_infos(); let mut total_bytes = 0; for ft_field in self.get_fulltext_fields() { let field = schema.get_field(&ft_field.field_name()).with_context(|| { format!( "Fulltext field {} not found in schema", ft_field.field_name() ) })?; let texts = ft_field.build_texts_from_doc(doc, schema, columns)?; for text in texts { total_bytes += text.len(); doc.add_text(field, &text); } } Ok(total_bytes) } pub fn is_int_handle(&self) -> bool { self.runtime.is_int_handle } fn get_analyzer_def(&self, analyzer_name: &str) -> Result { if let Some(analyzer_def) = self .storage .analyzer_defs .iter() .find(|analyzer_def| analyzer_def.name() == analyzer_name) { return Ok(analyzer_def.clone()); } AnalyzerDef::builtin_by_name(analyzer_name)? .ok_or_else(|| anyhow!("Analyzer {} not found", analyzer_name)) } pub fn analyze_phrase_tokens(&self, analyzer_name: &str, text: &str) -> Result { let analyzer_def = self.get_analyzer_def(analyzer_name)?; let source_tokens = analyzer_def.get_source_tokens_with_positions(text)?; let analyzer_has_ngram_filter = analyzer_def.has_ngram_filter(); let has_multiple_source_tokens = source_tokens.len() > 1; let mut analyzed_tokens = Vec::new(); let mut next_position_base = 0; let mut analyzer = self.get_analyzer(analyzer_name)?; for (source_position, source_text) in source_tokens { let base_position = source_position.max(next_position_base); let mut token_group = Self::collect_analyzed_tokens_with_positions(&mut analyzer, &source_text) .into_iter() .map(|token| AnalyzedToken { text: token.text, position: token.position + base_position, }) .collect::>(); if token_group.is_empty() { if analyzer_has_ngram_filter && has_multiple_source_tokens { return Ok(PhraseAnalysis::NoMatch); } next_position_base = next_position_base.max(source_position + 1); continue; } next_position_base = token_group .iter() .map(|token| token.position + 1) .max() .unwrap_or(next_position_base); analyzed_tokens.append(&mut token_group); } if analyzed_tokens.is_empty() { Ok(PhraseAnalysis::NoMatch) } else { Ok(PhraseAnalysis::Tokens(analyzed_tokens)) } } /// Analyze `text` for a single-token prefix query term. /// /// This runs in two stages: tokenizer-only /// (`get_source_tokens_with_positions`), then /// `build_prefix_query_analyzer` on that single source span. The first /// stage is required so we reject inputs that the tokenizer splits into /// multiple tokens, even if later filters would leave only one token. For /// example, with `Length { max: N, .. }` the prefix path still applies /// `max` (`build_for_prefix_query` only relaxes `min`). Two tokenizer /// tokens such as `foo` and a second word longer than `N` characters would /// become a single analyzed token if we ran the prefix analyzer on the /// whole string—the long token is dropped—so skipping the tokenizer-only /// step would wrongly accept multi-token input as a prefix. /// /// Refer to the fts_match_prefix tests in /// tests/test-cases/default-only/mysql_standard.test. pub fn analyze_prefix_token(&self, analyzer_name: &str, text: &str) -> Result> { let analyzer_def = self.get_analyzer_def(analyzer_name)?; let mut source_tokens = analyzer_def .get_source_tokens_with_positions(text)? .into_iter(); let Some((_, source_text)) = source_tokens.next() else { return Ok(None); }; if source_tokens.next().is_some() { return Ok(None); } let mut analyzer = analyzer_def.build_prefix_query_analyzer()?; let mut analyzed_tokens = Self::collect_analyzed_tokens_with_positions(&mut analyzer, &source_text).into_iter(); let Some(token) = analyzed_tokens.next() else { return Ok(None); }; if analyzed_tokens.next().is_some() { return Ok(None); } Ok(Some(token.text)) } fn get_analyzer(&self, analyzer_name: &str) -> Result { self.get_analyzer_manager() .get(analyzer_name) .ok_or_else(|| anyhow!("Analyzer {} not found", analyzer_name)) } fn collect_analyzed_tokens_with_positions( analyzer: &mut TextAnalyzer, text: &str, ) -> Vec { let mut stream = analyzer.token_stream(text); let mut tokens = Vec::new(); stream.process(&mut |token| { tokens.push(AnalyzedToken { text: token.text.clone(), position: token.position, }); }); tokens } pub fn get_analyzed_tokens_with_positions( &self, analyzer_name: &str, text: &str, ) -> Result> { let mut analyzer = self.get_analyzer(analyzer_name)?; Ok(Self::collect_analyzed_tokens_with_positions( &mut analyzer, text, )) } // These functions below are tests and benchmarks helper. pub fn index_columns(&self) -> &[ColumnInfo] { &self.storage.columns[..self.storage.index_column_count] } pub fn pk_column_count(&self) -> usize { self.storage .columns .iter() .filter(|c| c.is_primary_key) .count() } pub fn pk_columns(&self) -> Vec { self.storage .columns .iter() .filter(|c| c.is_primary_key) .cloned() .collect() } } impl IndexMetaStorage { fn is_inverted_column(&self, col_id: i64) -> bool { self.inverted_columns.iter().any(|c| c.col_id == col_id) } fn get_text_options(&self, col_id: i64) -> TextOptions { if self.is_inverted_column(col_id) { STRING | FAST | STORED // TODO: trim tailing space for collations } else { STORED.into() } } fn get_bytes_options(&self, col_id: i64) -> BytesOptions { if self.is_inverted_column(col_id) { (INDEXED | FAST | STORED).into() } else { STORED.into() } } fn get_numeric_options(&self, col_id: i64) -> NumericOptions { if self.is_inverted_column(col_id) { (INDEXED | FAST | STORED).into() } else { STORED.into() } } /// Each index builds a Tantivy schema with the following field types: /// /// ### Reserved Fields /// - `handle`: Binary field (INDEXED|STORED|FAST) - Row handle for delete /// operations and read range filter. /// - `column_-1024`: Version field (STORED|FAST) - MVCC version tracking. /// /// ### Column Fields (`column_{id}`) /// Column fields are named by their column ID (e.g., `column_1`, /// `column_-1`). Field options depend on whether the column is in the /// inverted index: /// /// | Column Type | Inverted Column | Non-Inverted Column | /// |-----------------|------------------------|---------------------| /// | Signed Int | INDEXED|FAST|STORED | STORED | /// | Unsigned Int | INDEXED|FAST|STORED | STORED | /// | Float | INDEXED|FAST|STORED | STORED | /// | DateTime | INDEXED|FAST|STORED | STORED | /// | Text | STRING|FAST|STORED | STORED | /// | Decimal/Binary | INDEXED|FAST|STORED | STORED | /// /// ### Fulltext Fields (`ft_{col_ids}_{analyzer}`) /// Fulltext fields are separate from column storage fields: /// - Named pattern: `ft_{col_id1}-{col_id2}-..._{analyzer_name}` /// - Options: Indexed with `WithFreqsAndPositions`, NOT stored (rebuilt /// from columns) /// - Example: `ft_4_default` for single column, `ft_1-3_jieba` for /// multi-column /// - Supports M:N mapping between columns and fulltext index fields fn build_tantivy_schema(&self) -> Result { let mut schema_builder = Schema::builder(); // HANDLE_STORE_NAME: binary format(Vec), row-store, with inverted index to // support delete_term. schema_builder.add_bytes_field(HANDLE_STORE_NAME, INDEXED | STORED | FAST); schema_builder.add_u64_field(VERSION_STORE_NAME, STORED | FAST); for col in &self.columns { let store_name = col.store_name(); if col.is_signed_int_type() { let opts = self.get_numeric_options(col.column_id); schema_builder.add_i64_field(&store_name, opts); } else if col.is_unsigned_int_type() { let opts = self.get_numeric_options(col.column_id); schema_builder.add_u64_field(&store_name, opts); } else if col.is_text_type() { let opts = self.get_text_options(col.column_id); schema_builder.add_text_field(&store_name, opts); } else if col.is_float_type() { let opts = self.get_numeric_options(col.column_id); schema_builder.add_f64_field(&store_name, opts); } else if col.is_date_time_type() { let opts = self.get_numeric_options(col.column_id); schema_builder.add_u64_field(&store_name, opts); } else if col.is_decimal_type() || col.is_binary_type() { let opts = self.get_bytes_options(col.column_id); schema_builder.add_bytes_field(&store_name, opts); } else { bail!("Unsupported column={:?}", col); } } for ft_field in &self.fulltext_fields { schema_builder.add_text_field(&ft_field.field_name(), ft_field.get_ft_options()); } Ok(schema_builder.build()) } pub fn build_analyzer_manager(&self) -> Result { create_analyzer_manager(&self.analyzer_defs) } pub fn build_sort_fields(&self, schema: &Schema) -> Result> { let mut sort_fields = Vec::with_capacity(self.sort_columns.len()); for sort_col in &self.sort_columns { let field_name = column_store_name(sort_col.col_id); let field = schema .doc_field(&field_name) .with_context(|| format!("Failed to get field for sort column {}", field_name))?; sort_fields.push(SortField { field, is_asc: sort_col.is_asc, }); } Ok(sort_fields) } fn build_sharding_columns(&self) -> Result> { let mut sharding_cols = Vec::with_capacity(self.sharding_columns.len()); for sharding_col in &self.sharding_columns { let col = self .columns .iter() .find(|c| c.column_id == sharding_col.col_id) .ok_or_else(|| anyhow!("Failed to get sharding column {}", sharding_col.col_id))?; sharding_cols.push(col.clone()); } Ok(sharding_cols) } // Check if the table has a single int primary key column. fn is_int_handle(&self) -> bool { let mut pk_iter = self.columns.iter().filter(|col| col.is_primary_key); match (pk_iter.next(), pk_iter.next()) { (Some(col), None) => col.is_signed_int_type() || col.is_unsigned_int_type(), _ => false, } } } #[cfg(test)] mod tests { use tantivy::schema::Type; use tici_grpc::{ ColumnInfo, IndexInfo, TableInfo, tidb_def::{ BINARY_FLAG, COLTYPE_DATETIME, COLTYPE_DOUBLE, COLTYPE_LONGLONG, COLTYPE_NEWDECIMAL, COLTYPE_VARCHAR, UNSIGNED_FLAG, }, }; use super::*; use crate::{ analysis::{ analyzer::{AnalyzerDef, DEFAULT_ANALYZER_NAME}, token_filter::TokenFilterType, tokenizer::TokenizerType, }, test_utils::{TEST_INDEX_KEY, index_meta::TEST_FT_INDEX_INFO_STD_STR}, }; fn mock_table_info() -> TableInfo { TableInfo { columns: vec![ ColumnInfo { column_id: 1, column_name: "id".to_string(), r#type: COLTYPE_LONGLONG, is_primary_key: true, ..Default::default() }, ColumnInfo { column_id: 2, column_name: "tenant_id".to_string(), r#type: COLTYPE_LONGLONG, is_primary_key: false, ..Default::default() }, ColumnInfo { column_id: 3, column_name: "timestamp".to_string(), r#type: COLTYPE_LONGLONG, is_primary_key: false, ..Default::default() }, ColumnInfo { column_id: 4, column_name: "content".to_string(), r#type: COLTYPE_VARCHAR, collation: "utf8mb4_bin".to_string(), is_primary_key: false, ..Default::default() }, ], is_clustered: true, ..Default::default() } } const TEST_NGRAM_ANALYZER_NAME: &str = "test_ngram"; fn build_test_default_fulltext_meta() -> IndexMeta { IndexMeta::try_from(IndexMetaStorage { version: IndexMetaVersion::LATEST, index_key: TEST_INDEX_KEY.clone(), created_ts: 0, columns: vec![ ColumnInfo { column_id: 1, column_name: "title".to_string(), r#type: COLTYPE_VARCHAR, collation: "utf8mb4_bin".to_string(), ..Default::default() }, ColumnInfo { column_id: 2, column_name: "body".to_string(), r#type: COLTYPE_VARCHAR, collation: "utf8mb4_bin".to_string(), ..Default::default() }, ], index_column_count: 2, sort_columns: vec![], inverted_columns: vec![], fulltext_fields: vec![FulltextField::new( vec![1, 2], DEFAULT_ANALYZER_NAME.to_string(), )], sharding_columns: vec![], analyzer_defs: vec![], is_partition: false, }) .unwrap() } fn build_test_ngram_fulltext_meta() -> IndexMeta { IndexMeta::try_from(IndexMetaStorage { version: IndexMetaVersion::LATEST, index_key: TEST_INDEX_KEY.clone(), created_ts: 0, columns: vec![ColumnInfo { column_id: 1, column_name: "title".to_string(), r#type: COLTYPE_VARCHAR, collation: "utf8mb4_bin".to_string(), ..Default::default() }], index_column_count: 1, sort_columns: vec![], inverted_columns: vec![], fulltext_fields: vec![FulltextField::new( vec![1], TEST_NGRAM_ANALYZER_NAME.to_string(), )], sharding_columns: vec![], analyzer_defs: vec![AnalyzerDef::new( TEST_NGRAM_ANALYZER_NAME.to_string(), TokenizerType::PreserveUnderscore, vec![ TokenFilterType::Ngram { min: 2, max: 2 }, TokenFilterType::LowerCaser, ], )], is_partition: false, }) .unwrap() } #[test] fn test_analyze_phrase_tokens_standard_keeps_gap_for_filtered_source_token() { let index_meta = build_test_default_fulltext_meta(); match index_meta .analyze_phrase_tokens(DEFAULT_ANALYZER_NAME, "hello the world") .unwrap() { PhraseAnalysis::NoMatch => panic!("expected gap-preserving tokens"), PhraseAnalysis::Tokens(tokens) => assert_eq!( tokens, vec![ AnalyzedToken { text: "hello".to_string(), position: 0, }, AnalyzedToken { text: "world".to_string(), position: 2, }, ] ), } } #[test] fn test_analyze_phrase_tokens_ngram_rejects_incomplete_source_group() { let index_meta = build_test_ngram_fulltext_meta(); assert_eq!( index_meta .analyze_phrase_tokens(TEST_NGRAM_ANALYZER_NAME, "lo w") .unwrap(), PhraseAnalysis::NoMatch ); } #[test] fn test_parse_hybrid_index_basic() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "inverted":[ { "columns":[ { "name":{ "O":"id", "L":"id" }, "offset":0, "length":-1 }, { "name":{ "O":"tenant_id", "L":"tenant_id" }, "offset":1, "length":-1 }, { "name":{ "O":"timestamp", "L":"timestamp" }, "offset":2, "length":-1 } ] } ], "sort":{ "columns":[ { "name":{ "O":"id", "L":"id" }, "offset":0, "length":-1 }, { "name":{ "O":"tenant_id", "L":"tenant_id" }, "offset":1, "length":-1 }, { "name":{ "O":"timestamp", "L":"timestamp" }, "offset":2, "length":-1 } ], "is_asc":[ true, false, true ] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let meta = IndexMeta::new(&table_info, &index_info).unwrap(); let inverted_ids: Vec = meta .storage .inverted_columns .iter() .map(|c| c.col_id) .collect(); assert_eq!(inverted_ids, vec![1, 2, 3]); let sort_pairs: Vec<(i64, bool)> = meta .storage .sort_columns .iter() .map(|c| (c.col_id, c.is_asc)) .collect(); assert_eq!( sort_pairs, vec![ (1, true), // id asc (2, false), // tenant_id desc (3, true), // timestamp asc ] ); assert_eq!(meta.storage.index_column_count, 4); assert!(meta.storage.fulltext_fields.is_empty()); } #[test] fn test_parse_hybrid_index_duplicate_sort_columns_should_fail() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "inverted": [ { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 }, { "name": { "O": "tenant_id", "L": "tenant_id" }, "offset": 1, "length": -1 } ] } ], "sort": { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 }, { "name": { "O": "id", "L": "id" }, "offset": 1, "length": -1 } ], "is_asc": [true, false] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let err = IndexMeta::new(&table_info, &index_info).unwrap_err(); assert!( err.to_string() .contains("Duplicate columns in `sort` field") ); } #[test] fn test_parse_hybrid_index_sort_order_mismatch_should_fail() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "inverted": [ { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 }, { "name": { "O": "tenant_id", "L": "tenant_id" }, "offset": 1, "length": -1 } ] } ], "sort": { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 }, { "name": { "O": "tenant_id", "L": "tenant_id" }, "offset": 1, "length": -1 } ], "is_asc": [true] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let err = IndexMeta::new(&table_info, &index_info).unwrap_err(); assert!( err.to_string() .contains("Mismatched number of columns and order") ); } #[test] fn test_parse_hybrid_index_invalid_column_should_fail() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "inverted": [ { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 }, { "name": { "O": "non_exist_col", "L": "non_exist_col" }, "offset": 1, "length": -1 } ] } ], "sort": { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 }, { "name": { "O": "tenant_id", "L": "tenant_id" }, "offset": 1, "length": -1 } ], "is_asc": [true, false] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let err = IndexMeta::new(&table_info, &index_info).unwrap_err(); assert!( err.to_string().contains("Column non_exist_col not found"), "unexpected error: {}", err ); } #[test] fn test_parse_hybrid_index_empty_should_return_default() { let table_info = mock_table_info(); let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: "".to_string(), ..Default::default() }; let meta = IndexMeta::new(&table_info, &index_info).unwrap(); assert!(meta.storage.inverted_columns.is_empty()); assert!(meta.storage.sort_columns.is_empty()); assert!(meta.storage.fulltext_fields.is_empty()); } #[test] fn test_index_meta_preserves_created_ts() { let table_info = mock_table_info(); let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: "".to_string(), created_ts: 123, ..Default::default() }; let meta = IndexMeta::new(&table_info, &index_info).unwrap(); assert_eq!(meta.created_ts(), 123); assert_eq!(meta.storage.created_ts, 123); } #[test] fn test_parse_hybrid_index_only_sort_no_inverted() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "sort": { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 }, { "name": { "O": "tenant_id", "L": "tenant_id" }, "offset": 1, "length": -1 } ], "is_asc": [true, false] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let meta = IndexMeta::new(&table_info, &index_info).unwrap(); assert_eq!( meta.storage .sort_columns .iter() .map(|c| (c.col_id, c.is_asc)) .collect::>(), vec![(1, true), (2, false)] ); assert!(meta.storage.inverted_columns.is_empty()); } #[test] fn test_get_tantivy_schema_with_inverted_and_fulltext() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "inverted": [ { "columns": [ { "name": { "O": "tenant_id", "L": "tenant_id" }, "offset": 0, "length": -1 } ] } ], "sort": { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 } ], "is_asc": [true] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let mut meta = IndexMeta::new(&table_info, &index_info).unwrap(); // Manual add content column as fulltext meta.storage.fulltext_fields.push(FulltextField::new( vec![4], DEFAULT_ANALYZER_NAME.to_string(), )); let schema = meta.storage.build_tantivy_schema().unwrap(); for field in [HANDLE_STORE_NAME, VERSION_STORE_NAME] { assert!(schema.get_field(field).is_ok(), "missing field: {}", field); } // tenant_id column is inverted let tenant_field = schema.get_field("column_2").unwrap(); let field_entry = schema.get_field_entry(tenant_field); assert!(field_entry.is_fast()); assert!(field_entry.is_stored()); assert!(field_entry.is_indexed()); // content column (column_4) is store only, fulltext field is separate let content_field = schema.get_field("column_4").unwrap(); let field_entry = schema.get_field_entry(content_field); assert!(!field_entry.is_fast()); assert!(field_entry.is_stored()); assert!(!field_entry.is_indexed()); // fulltext field is ft_4_default (ft_{col_ids}_{analyzer_name}) // Fulltext fields are NOT stored to save space - they are rebuilt from // column values during compaction. let ft_field = schema.get_field("ft_4_default").unwrap(); let field_entry = schema.get_field_entry(ft_field); assert!(!field_entry.is_fast()); assert!(!field_entry.is_stored()); assert!(field_entry.is_indexed()); assert!( matches!( field_entry.field_type().get_index_record_option(), Some(IndexRecordOption::WithFreqsAndPositions) ), "fulltext field should use WithFreqsAndPositions" ); // id/timestamp column is store only let id_field = schema.get_field("column_1").unwrap(); let field_entry = schema.get_field_entry(id_field); assert!(field_entry.is_stored()); assert!( !field_entry.is_fast() || !field_entry.is_indexed(), "non-inverted column should not be indexed+fast" ); let ts_field = schema.get_field("column_3").unwrap(); let field_entry = schema.get_field_entry(ts_field); assert!(field_entry.is_stored()); assert!( !field_entry.is_fast() || !field_entry.is_indexed(), "non-inverted column should not be indexed+fast" ); } #[test] fn test_serde_index_meta() -> Result<()> { let table_info = mock_table_info(); let hybrid_index_json = r#" { "inverted":[ { "columns":[ { "name":{ "O":"id", "L":"id" }, "offset":0, "length":-1 }, { "name":{ "O":"tenant_id", "L":"tenant_id" }, "offset":1, "length":-1 }, { "name":{ "O":"timestamp", "L":"timestamp" }, "offset":2, "length":-1 } ] } ], "sort":{ "columns":[ { "name":{ "O":"id", "L":"id" }, "offset":0, "length":-1 }, { "name":{ "O":"tenant_id", "L":"tenant_id" }, "offset":1, "length":-1 }, { "name":{ "O":"timestamp", "L":"timestamp" }, "offset":2, "length":-1 } ], "is_asc":[ true, false, true ] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let index_meta = IndexMeta::new(&table_info, &index_info)?; let s = serde_json::to_string(&index_meta)?; let index_meta: IndexMeta = serde_json::from_str(&s)?; let inverted_ids: Vec = index_meta .storage .inverted_columns .iter() .map(|c| c.col_id) .collect(); assert_eq!(inverted_ids, vec![1, 2, 3]); let sort_pairs: Vec<(i64, bool)> = index_meta .storage .sort_columns .iter() .map(|c| (c.col_id, c.is_asc)) .collect(); assert_eq!( sort_pairs, vec![ (1, true), // id asc (2, false), // tenant_id desc (3, true), // timestamp asc ] ); assert_eq!(index_meta.storage.index_column_count, 4); assert!(index_meta.storage.fulltext_fields.is_empty()); Ok(()) } #[test] fn test_column_store_name() { assert_eq!(column_store_name(1), "column_1"); assert_eq!(column_store_name(-1), "column_-1"); assert_eq!(column_store_name(1024), "column_1024"); assert_eq!(column_store_name(-1024), "column_-1024"); } #[test] fn test_column_info_ext_store_name() { let col = ColumnInfo { column_id: 42, column_name: "test_col".to_string(), ..Default::default() }; assert_eq!(col.store_name(), "column_42"); let col_negative = ColumnInfo { column_id: -1, column_name: "tidb_rowid".to_string(), ..Default::default() }; assert_eq!(col_negative.store_name(), "column_-1"); } #[test] fn test_index_meta_version() { assert_eq!(IndexMetaVersion::V0, IndexMetaVersion(0)); assert_eq!(IndexMetaVersion::LATEST, IndexMetaVersion::V0); assert!(IndexMetaVersion::V0 <= IndexMetaVersion::LATEST); } #[test] fn test_has_duplicates() { assert!(!IndexMeta::has_duplicates(&[])); assert!(!IndexMeta::has_duplicates(&["a".to_string()])); assert!(!IndexMeta::has_duplicates(&[ "a".to_string(), "b".to_string(), "c".to_string() ])); assert!(IndexMeta::has_duplicates(&[ "a".to_string(), "b".to_string(), "a".to_string() ])); assert!(IndexMeta::has_duplicates(&[ "x".to_string(), "y".to_string(), "z".to_string(), "x".to_string() ])); } #[test] fn test_parse_fulltext_index() { let table_info = mock_table_info(); let index_info = IndexInfo { index_type: tici_grpc::IndexType::Fulltext as i32, columns: vec![table_info.columns[3].clone()], // content column fulltext_index: TEST_FT_INDEX_INFO_STD_STR.to_string(), ..Default::default() }; let meta = IndexMeta::new(&table_info, &index_info).unwrap(); assert_eq!(meta.storage.fulltext_fields.len(), 1); assert_eq!(meta.storage.fulltext_fields[0].col_ids, vec![4]); assert!(meta.storage.inverted_columns.is_empty()); assert!(meta.storage.sort_columns.is_empty()); } #[test] fn test_parse_fulltext_index_with_non_text_column_should_fail() { let table_info = mock_table_info(); let index_info = IndexInfo { index_type: tici_grpc::IndexType::Fulltext as i32, columns: vec![table_info.columns[0].clone()], // id column is LONGLONG fulltext_index: TEST_FT_INDEX_INFO_STD_STR.to_string(), ..Default::default() }; let err = IndexMeta::new(&table_info, &index_info).unwrap_err(); assert!( err.to_string().contains("Unsupported data type"), "unexpected error: {}", err ); } #[test] fn test_parse_inverted_duplicate_columns_should_fail() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "inverted": [ { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 }, { "name": { "O": "id", "L": "id" }, "offset": 1, "length": -1 } ] } ] }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let err = IndexMeta::new(&table_info, &index_info).unwrap_err(); assert!( err.to_string() .contains("Duplicate columns in `inverted` field"), "unexpected error: {}", err ); } #[test] fn test_parse_hybrid_index_only_inverted_no_sort() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "inverted": [ { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 }, { "name": { "O": "tenant_id", "L": "tenant_id" }, "offset": 1, "length": -1 } ] } ] }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let meta = IndexMeta::new(&table_info, &index_info).unwrap(); assert_eq!( meta.storage .inverted_columns .iter() .map(|c| c.col_id) .collect::>(), vec![1, 2] ); assert!(meta.storage.sort_columns.is_empty()); } #[test] fn test_is_inverted_column() { let storage = IndexMetaStorage { version: IndexMetaVersion::LATEST, index_key: TEST_INDEX_KEY.clone(), created_ts: 0, columns: vec![], index_column_count: 0, sort_columns: vec![], inverted_columns: vec![InvertedColumn { col_id: 1 }, InvertedColumn { col_id: 3 }], fulltext_fields: vec![], sharding_columns: vec![], analyzer_defs: vec![], is_partition: false, }; assert!(storage.is_inverted_column(1)); assert!(!storage.is_inverted_column(2)); assert!(storage.is_inverted_column(3)); } #[test] fn test_get_text_options() { let storage = IndexMetaStorage { version: IndexMetaVersion::LATEST, index_key: TEST_INDEX_KEY.clone(), created_ts: 0, columns: vec![], index_column_count: 0, sort_columns: vec![], inverted_columns: vec![InvertedColumn { col_id: 1 }], fulltext_fields: vec![FulltextField::new(vec![2], "a1".to_string())], sharding_columns: vec![], analyzer_defs: vec![], is_partition: false, }; // Inverted column should have STRING | FAST | STORED let inverted_opts = storage.get_text_options(1); assert!(inverted_opts.is_fast()); assert!(inverted_opts.is_stored()); assert!(inverted_opts.get_indexing_options().is_some()); // Fulltext column should have custom indexing options let fulltext_opts = storage.get_text_options(2); assert!(fulltext_opts.is_stored()); assert!(!fulltext_opts.is_fast()); assert!(fulltext_opts.get_indexing_options().is_none()); // Regular column should be STORED only let regular_opts = storage.get_text_options(3); assert!(regular_opts.is_stored()); assert!(!regular_opts.is_fast()); assert!(regular_opts.get_indexing_options().is_none()); } #[test] fn test_get_numeric_options() { let storage = IndexMetaStorage { version: IndexMetaVersion::LATEST, index_key: TEST_INDEX_KEY.clone(), created_ts: 0, columns: vec![], index_column_count: 0, sort_columns: vec![], inverted_columns: vec![InvertedColumn { col_id: 1 }], fulltext_fields: vec![], sharding_columns: vec![], analyzer_defs: vec![], is_partition: false, }; // Inverted column should have INDEXED | FAST | STORED let inverted_opts = storage.get_numeric_options(1); assert!(inverted_opts.is_indexed()); assert!(inverted_opts.is_fast()); assert!(inverted_opts.is_stored()); // Regular column should be STORED only (not both indexed and fast) let regular_opts = storage.get_numeric_options(2); assert!(regular_opts.is_stored()); assert!(!regular_opts.is_indexed()); assert!(!regular_opts.is_fast()); } #[test] fn test_build_tantivy_schema_with_different_column_types() { let storage = IndexMetaStorage { version: IndexMetaVersion::LATEST, index_key: TEST_INDEX_KEY.clone(), created_ts: 0, columns: vec![ ColumnInfo { column_id: 1, column_name: "signed_int".to_string(), r#type: COLTYPE_LONGLONG, ..Default::default() }, ColumnInfo { column_id: 2, column_name: "unsigned_int".to_string(), r#type: COLTYPE_LONGLONG, flag: UNSIGNED_FLAG, ..Default::default() }, ColumnInfo { column_id: 3, column_name: "float_val".to_string(), r#type: COLTYPE_DOUBLE, ..Default::default() }, ColumnInfo { column_id: 4, column_name: "text_val".to_string(), r#type: COLTYPE_VARCHAR, collation: "utf8mb4_bin".to_string(), ..Default::default() }, ColumnInfo { column_id: 5, column_name: "datetime_val".to_string(), r#type: COLTYPE_DATETIME, ..Default::default() }, ColumnInfo { column_id: 6, column_name: "decimal_val".to_string(), r#type: COLTYPE_NEWDECIMAL, ..Default::default() }, ColumnInfo { column_id: 7, column_name: "binary_val".to_string(), r#type: COLTYPE_VARCHAR, flag: BINARY_FLAG, ..Default::default() }, ], index_column_count: 7, sort_columns: vec![], inverted_columns: vec![], fulltext_fields: vec![], sharding_columns: vec![], analyzer_defs: vec![], is_partition: false, }; let schema = storage.build_tantivy_schema().unwrap(); // Check all column types are added assert!(schema.get_field("column_1").is_ok()); // signed int assert!(schema.get_field("column_2").is_ok()); // unsigned int assert!(schema.get_field("column_3").is_ok()); // float assert!(schema.get_field("column_4").is_ok()); // text assert!(schema.get_field("column_5").is_ok()); // datetime assert!(schema.get_field("column_6").is_ok()); // decimal assert!(schema.get_field("column_7").is_ok()); // binary let binary_field = schema.get_field("column_7").unwrap(); let binary_field_entry = schema.get_field_entry(binary_field); assert_eq!(binary_field_entry.field_type().value_type(), Type::Bytes); // Check reserved fields assert!(schema.get_field(HANDLE_STORE_NAME).is_ok()); assert!(schema.get_field(VERSION_STORE_NAME).is_ok()); } #[test] fn test_build_sort_fields() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "sort": { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 }, { "name": { "O": "tenant_id", "L": "tenant_id" }, "offset": 1, "length": -1 } ], "is_asc": [true, false] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let meta = IndexMeta::new(&table_info, &index_info).unwrap(); let sort_fields = meta.get_sort_fields(); assert_eq!(sort_fields.len(), 2); assert!(sort_fields[0].is_asc); assert!(!sort_fields[1].is_asc); } #[test] fn test_build_sharding_columns() { let storage = IndexMetaStorage { version: IndexMetaVersion::LATEST, index_key: TEST_INDEX_KEY.clone(), created_ts: 0, columns: vec![ ColumnInfo { column_id: 1, column_name: "id".to_string(), r#type: COLTYPE_LONGLONG, ..Default::default() }, ColumnInfo { column_id: 2, column_name: "shard_key".to_string(), r#type: COLTYPE_LONGLONG, ..Default::default() }, ], index_column_count: 2, sort_columns: vec![], inverted_columns: vec![], fulltext_fields: vec![], sharding_columns: vec![ShardingColumn { col_id: 2 }], analyzer_defs: vec![], is_partition: false, }; let sharding_cols = storage.build_sharding_columns().unwrap(); assert_eq!(sharding_cols.len(), 1); assert_eq!(sharding_cols[0].column_id, 2); assert_eq!(sharding_cols[0].column_name, "shard_key"); } #[test] fn test_build_sharding_columns_invalid_col_id_should_fail() { let storage = IndexMetaStorage { version: IndexMetaVersion::LATEST, index_key: TEST_INDEX_KEY.clone(), created_ts: 0, columns: vec![ColumnInfo { column_id: 1, column_name: "id".to_string(), r#type: COLTYPE_LONGLONG, ..Default::default() }], index_column_count: 1, sort_columns: vec![], inverted_columns: vec![], fulltext_fields: vec![], sharding_columns: vec![ShardingColumn { col_id: 999 }], analyzer_defs: vec![], is_partition: false, }; let err = storage.build_sharding_columns().unwrap_err(); assert!( err.to_string().contains("Failed to get sharding column"), "unexpected error: {}", err ); } #[test] fn test_index_meta_getter_methods() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "inverted": [ { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 } ] } ], "sort": { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 } ], "is_asc": [true] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let meta = IndexMeta::new(&table_info, &index_info).unwrap(); // Test getter methods assert!( meta.get_analyzer_manager() .get(DEFAULT_ANALYZER_NAME) .is_some() ); assert!(meta.get_tantivy_schema().get_field("column_1").is_ok()); assert_eq!(meta.get_sort_fields().len(), 1); assert_eq!(meta.get_sharding_columns().len(), 0); assert_eq!(meta.get_column_infos().len(), 4); } #[test] fn test_index_columns_and_pk_methods() { let table_info = mock_table_info(); let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: "".to_string(), ..Default::default() }; let meta = IndexMeta::new(&table_info, &index_info).unwrap(); // All columns are index columns assert_eq!(meta.index_columns().len(), 4); // Only column 1 (id) is primary key assert_eq!(meta.pk_column_count(), 1); let pk_cols = meta.pk_columns(); assert_eq!(pk_cols.len(), 1); assert_eq!(pk_cols[0].column_id, 1); assert_eq!(pk_cols[0].column_name, "id"); } #[test] fn test_unsupported_index_type_should_fail() { let table_info = mock_table_info(); let index_info = IndexInfo { index_type: 999, // Invalid index type columns: table_info.columns.clone(), hybrid_index: "".to_string(), ..Default::default() }; let err = IndexMeta::new(&table_info, &index_info).unwrap_err(); assert!( err.to_string().contains("Unsupported index type"), "unexpected error: {}", err ); } #[test] fn test_sort_column_serialization() { let sort_col = SortColumn { col_id: 42, is_asc: true, }; let json = serde_json::to_string(&sort_col).unwrap(); let deserialized: SortColumn = serde_json::from_str(&json).unwrap(); assert_eq!(deserialized, sort_col); } #[test] fn test_inverted_column_serialization() { let inv_col = InvertedColumn { col_id: 123 }; let json = serde_json::to_string(&inv_col).unwrap(); let deserialized: InvertedColumn = serde_json::from_str(&json).unwrap(); assert_eq!(deserialized, inv_col); } #[test] fn test_fulltext_field_serialization() { let ft_field = FulltextField::new(vec![456], "a1".to_string()); let json = serde_json::to_string(&ft_field).unwrap(); let deserialized: FulltextField = serde_json::from_str(&json).unwrap(); assert_eq!(deserialized, ft_field); } #[test] fn test_sharding_column_serialization() { let shard_col = ShardingColumn { col_id: 789 }; let json = serde_json::to_string(&shard_col).unwrap(); let deserialized: ShardingColumn = serde_json::from_str(&json).unwrap(); assert_eq!(deserialized, shard_col); } #[test] fn test_index_meta_storage_serialization() { let storage = IndexMetaStorage { version: IndexMetaVersion::LATEST, index_key: TEST_INDEX_KEY.clone(), created_ts: 0, columns: vec![ColumnInfo { column_id: 1, column_name: "test".to_string(), r#type: COLTYPE_LONGLONG, ..Default::default() }], index_column_count: 1, sort_columns: vec![SortColumn { col_id: 1, is_asc: true, }], inverted_columns: vec![InvertedColumn { col_id: 1 }], fulltext_fields: vec![], sharding_columns: vec![], analyzer_defs: vec![], is_partition: false, }; let json = serde_json::to_string(&storage).unwrap(); let deserialized: IndexMetaStorage = serde_json::from_str(&json).unwrap(); assert_eq!(deserialized, storage); } #[test] fn test_parse_sharding_single_column() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "sharding_key": { "columns": [ { "name": { "O": "tenant_id", "L": "tenant_id" }, "offset": 0, "length": -1 } ] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let meta = IndexMeta::new(&table_info, &index_info).unwrap(); assert_eq!(meta.storage.sharding_columns.len(), 1); assert_eq!(meta.storage.sharding_columns[0].col_id, 2); } #[test] fn test_parse_sharding_multiple_columns() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "sharding_key": { "columns": [ { "name": { "O": "tenant_id", "L": "tenant_id" }, "offset": 0, "length": -1 }, { "name": { "O": "timestamp", "L": "timestamp" }, "offset": 1, "length": -1 } ] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let meta = IndexMeta::new(&table_info, &index_info).unwrap(); assert_eq!(meta.storage.sharding_columns.len(), 2); assert_eq!(meta.storage.sharding_columns[0].col_id, 2); assert_eq!(meta.storage.sharding_columns[1].col_id, 3); } #[test] fn test_parse_sharding_duplicate_columns_should_fail() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "sharding_key": { "columns": [ { "name": { "O": "tenant_id", "L": "tenant_id" }, "offset": 0, "length": -1 }, { "name": { "O": "tenant_id", "L": "tenant_id" }, "offset": 1, "length": -1 } ] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let err = IndexMeta::new(&table_info, &index_info).unwrap_err(); assert!( err.to_string() .contains("Duplicate columns in `sharding_key` field"), "unexpected error: {}", err ); } #[test] fn test_parse_sharding_invalid_column_should_fail() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "sharding_key": { "columns": [ { "name": { "O": "non_existent", "L": "non_existent" }, "offset": 0, "length": -1 } ] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let err = IndexMeta::new(&table_info, &index_info).unwrap_err(); assert!( err.to_string() .contains("Column non_existent not found in hybrid index"), "unexpected error: {}", err ); } #[test] fn test_parse_sharding_empty_columns() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "sharding_key": { "columns": [] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let meta = IndexMeta::new(&table_info, &index_info).unwrap(); assert_eq!(meta.storage.sharding_columns.len(), 0); } #[test] fn test_parse_sharding_with_other_fields() { let table_info = mock_table_info(); let hybrid_index_json = r#" { "inverted": [ { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 } ] } ], "sort": { "columns": [ { "name": { "O": "id", "L": "id" }, "offset": 0, "length": -1 } ], "is_asc": [true] }, "sharding_key": { "columns": [ { "name": { "O": "tenant_id", "L": "tenant_id" }, "offset": 0, "length": -1 } ] } }"#; let index_info = IndexInfo { index_type: tici_grpc::IndexType::Hybrid as i32, columns: table_info.columns.clone(), hybrid_index: hybrid_index_json.to_string(), ..Default::default() }; let meta = IndexMeta::new(&table_info, &index_info).unwrap(); // Verify sharding columns assert_eq!(meta.storage.sharding_columns.len(), 1); assert_eq!(meta.storage.sharding_columns[0].col_id, 2); // Verify other fields are also parsed correctly assert_eq!(meta.storage.inverted_columns.len(), 1); assert_eq!(meta.storage.sort_columns.len(), 1); } #[test] fn test_get_fulltext_field() { let storage = IndexMetaStorage { version: IndexMetaVersion::LATEST, index_key: TEST_INDEX_KEY.clone(), created_ts: 0, columns: vec![], index_column_count: 0, sort_columns: vec![], inverted_columns: vec![], fulltext_fields: vec![ FulltextField::new(vec![2], DEFAULT_ANALYZER_NAME.to_string()), FulltextField::new(vec![4], DEFAULT_ANALYZER_NAME.to_string()), FulltextField::new(vec![1, 3], DEFAULT_ANALYZER_NAME.to_string()), ], sharding_columns: vec![], analyzer_defs: vec![], is_partition: false, }; let meta: IndexMeta = storage.try_into().unwrap(); // Single column fulltext fields assert!(meta.get_fulltext_field(vec![1]).is_err()); assert!(meta.get_fulltext_field(vec![2]).is_ok()); assert!(meta.get_fulltext_field(vec![3]).is_err()); assert!(meta.get_fulltext_field(vec![4]).is_ok()); // Multi-column fulltext field assert!(meta.get_fulltext_field(vec![1, 3]).is_ok()); // Order doesn't matter - input is sorted before comparison assert!(meta.get_fulltext_field(vec![3, 1]).is_ok()); } }