// Copyright 2026 PingCAP, Inc. use std::collections::{BTreeMap, HashMap, HashSet}; use anyhow::{Result, anyhow, bail}; use clara_fts::{IndexLayout, SingleFieldLayout}; use kvenginepb::fts::{ FtsFieldLayout, FtsTrackedIndexMeta, FullTextAnalyzerType, FullTextIndexDef, FullTextIndexFieldDef, FullTextIndexFieldOption, }; use tantivy::schema::Field; use tipb::ColumnInfo; use super::{dedicated_file::EDedicatedFile, packed_file::PackedFile}; use crate::table::SnapVersion; pub type TrackedFtsIndexMap = HashMap>; fn legacy_parser_type_to_analyzer_type(parser_type: &str) -> FullTextAnalyzerType { match parser_type { // only the first three parsers(which has no extra parameters) are support for legacy meta. "EXACT_VALUE_V1" => FullTextAnalyzerType::FtsAnalyzerExactValueV1, "STANDARD_V1" => FullTextAnalyzerType::FtsAnalyzerStandardV1, "MULTILINGUAL_V1" => FullTextAnalyzerType::FtsAnalyzerMultilingualV1, "NGRAM_V1" | "PATH_HIERARCHY_V1" => unreachable!(), _ => FullTextAnalyzerType::FtsAnalyzerUnspecified, } } fn tracked_index_field_from_legacy_col_id(col_id: i64, parser_type: &str) -> FullTextIndexFieldDef { let mut index_field = FullTextIndexFieldDef::new(); index_field.set_col_id(col_id); let mut index_field_option = FullTextIndexFieldOption::new(); index_field_option.set_analyzer_type(legacy_parser_type_to_analyzer_type(parser_type)); index_field_option.set_enable_bm25(true); index_field.set_index_field_option(index_field_option); index_field } pub fn fulltext_index_fields(index_def: &FullTextIndexDef) -> Vec { if !index_def.get_index_fields().is_empty() { index_def.get_index_fields().to_vec() } else if index_def.get_col_id() != 0 { vec![tracked_index_field_from_legacy_col_id( index_def.get_col_id(), index_def.get_parser_type(), )] } else { Vec::new() } } fn is_legacy_fulltext_index_def(index_def: &FullTextIndexDef) -> bool { index_def.get_index_fields().is_empty() && index_def.get_col_id() != 0 } pub fn fulltext_index_col_ids(index_def: &FullTextIndexDef) -> Vec { fulltext_index_fields(index_def) .into_iter() .map(|field| field.get_col_id()) .collect() } pub fn tracked_index_fields_for_layout( index_def: &FullTextIndexDef, field_layout: FtsFieldLayout, ) -> Vec { let mut index_fields = fulltext_index_fields(index_def); match field_layout { FtsFieldLayout::FtsFieldLayoutLegacyBody | FtsFieldLayout::FtsFieldLayoutUnspecified => { index_fields.truncate(1); } FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString => {} } verify_tracked_index_fields(&index_fields, field_layout); index_fields } pub fn tracked_index_col_ids_for_layout( index_def: &FullTextIndexDef, field_layout: FtsFieldLayout, ) -> Vec { tracked_index_fields_for_layout(index_def, field_layout) .into_iter() .map(|field| field.get_col_id()) .collect() } fn tracked_index_field_col_ids(index_fields: &[FullTextIndexFieldDef]) -> Vec { index_fields .iter() .map(|field| field.get_col_id()) .collect() } fn tracked_index_field_def_matches( tracked_field: &FullTextIndexFieldDef, schema_field: &FullTextIndexFieldDef, ) -> bool { if tracked_field.get_col_id() != schema_field.get_col_id() || tracked_field.has_index_field_option() != schema_field.has_index_field_option() { return false; } if !tracked_field.has_index_field_option() { return true; } let tracked_option = tracked_field.get_index_field_option(); let schema_option = schema_field.get_index_field_option(); if tracked_option.get_analyzer_type() != schema_option.get_analyzer_type() || tracked_option.get_enable_bm25() != schema_option.get_enable_bm25() || tracked_option.has_ngram() != schema_option.has_ngram() || tracked_option.has_path_hierarchy() != schema_option.has_path_hierarchy() { return false; } if tracked_option.has_ngram() { let tracked_ngram = tracked_option.get_ngram(); let schema_ngram = schema_option.get_ngram(); if tracked_ngram.get_min_gram() != schema_ngram.get_min_gram() || tracked_ngram.get_max_gram() != schema_ngram.get_max_gram() || tracked_ngram.get_granularity() != schema_ngram.get_granularity() || tracked_ngram.get_lower_case() != schema_ngram.get_lower_case() { return false; } } if tracked_option.has_path_hierarchy() && tracked_option.get_path_hierarchy().get_delimiter() != schema_option.get_path_hierarchy().get_delimiter() { return false; } true } fn verify_tracked_index_fields( index_fields: &[FullTextIndexFieldDef], field_layout: FtsFieldLayout, ) { match field_layout { FtsFieldLayout::FtsFieldLayoutLegacyBody | FtsFieldLayout::FtsFieldLayoutUnspecified => { assert!( index_fields.len() <= 1, "legacy-body tracked index must contain at most one field, got {:?}", tracked_index_field_col_ids(index_fields) ); } FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString => { assert!( !index_fields.is_empty(), "ordinal-string tracked index must contain at least one field" ); } } } fn tracked_index_fields_from_meta(meta: &FtsTrackedIndexMeta) -> Vec { meta.get_index_fields().to_vec() } fn validate_tracked_index_fields( meta: &FtsTrackedIndexMeta, field_layout: FtsFieldLayout, index_fields: &[FullTextIndexFieldDef], ) -> Result<()> { match field_layout { FtsFieldLayout::FtsFieldLayoutLegacyBody | FtsFieldLayout::FtsFieldLayoutUnspecified => { if index_fields.len() > 1 { bail!( "legacy-body tracked index ({}, {}) has {} fields", meta.get_table_id(), meta.get_index_id(), index_fields.len() ); } } FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString => { if index_fields.is_empty() { bail!( "ordinal-string tracked index ({}, {}) requires at least 1 field", meta.get_table_id(), meta.get_index_id(), ); } } } Ok(()) } pub fn new_tracked_index_meta( table_id: i64, index_id: i64, index_fields: Vec, mut field_layout: FtsFieldLayout, ) -> FtsTrackedIndexMeta { if field_layout == FtsFieldLayout::FtsFieldLayoutUnspecified { assert!(index_fields.len() <= 1); field_layout = FtsFieldLayout::FtsFieldLayoutLegacyBody; } verify_tracked_index_fields(&index_fields, field_layout); let mut meta = FtsTrackedIndexMeta::new(); meta.set_table_id(table_id); meta.set_index_id(index_id); meta.set_field_layout(field_layout); meta.set_index_fields(index_fields.into()); meta } pub fn new_legacy_tracked_index_meta(table_id: i64, index_id: i64) -> FtsTrackedIndexMeta { new_tracked_index_meta( table_id, index_id, Vec::new(), FtsFieldLayout::FtsFieldLayoutLegacyBody, ) } pub fn tracked_index_meta_from_index_def( table_id: i64, index_def: &FullTextIndexDef, field_layout: FtsFieldLayout, ) -> FtsTrackedIndexMeta { new_tracked_index_meta( table_id, index_def.get_index_id(), tracked_index_fields_for_layout(index_def, field_layout), field_layout, ) } pub fn tracked_index_fields_from_meta_or_index_def( meta: &FtsTrackedIndexMeta, index_def: &FullTextIndexDef, ) -> Vec { if is_legacy_fulltext_index_def(index_def) { return tracked_index_fields_for_layout(index_def, meta.get_field_layout()); } let meta_index_fields = tracked_index_fields_from_meta(meta); if meta_index_fields.is_empty() { tracked_index_fields_for_layout(index_def, meta.get_field_layout()) } else { meta_index_fields } } pub fn tracked_index_fields_from_layout_or_index_def( tracked_index_layout: &TrackedIndexLayout, index_def: &FullTextIndexDef, ) -> Result> { let schema_index_fields = tracked_index_fields_for_layout(index_def, tracked_index_layout.field_layout); if tracked_index_layout.index_fields.is_empty() { return Ok(schema_index_fields); } if tracked_index_layout.index_fields.len() != schema_index_fields.len() || tracked_index_layout .index_fields .iter() .zip(&schema_index_fields) .any(|(tracked_field, schema_field)| { !tracked_index_field_def_matches(tracked_field, schema_field) }) { bail!( "tracked index field definitions {:?} do not match schema field definitions {:?}", tracked_index_layout.index_fields, schema_index_fields, ); } Ok(tracked_index_layout.index_fields.clone()) } #[derive(Clone, Debug, PartialEq)] pub struct TrackedIndexLayout { pub index_fields: Vec, pub field_layout: FtsFieldLayout, pub index_layout: IndexLayout, } impl TrackedIndexLayout { pub fn physical_col_ids(&self) -> Vec { tracked_index_field_col_ids(&self.index_fields) } pub fn select_field_layout(&self, query_col_id: i64) -> Result { match self.field_layout { FtsFieldLayout::FtsFieldLayoutLegacyBody | FtsFieldLayout::FtsFieldLayoutUnspecified => { Ok(self.index_layout.first_field().clone()) } FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString => { let mut matched_field_indexes = self.index_fields .iter() .enumerate() .filter_map(|(field_idx, field)| { (field.get_col_id() == query_col_id).then_some(field_idx) }); let Some(field_idx) = matched_field_indexes.next() else { bail!( "query column {} is not tracked by index columns {:?}", query_col_id, self.physical_col_ids() ); }; if matched_field_indexes.next().is_some() { bail!( "query column {} is ambiguously tracked by index columns {:?}", query_col_id, self.physical_col_ids() ); } self.index_layout.field_by_id(field_idx).ok_or_else(|| { anyhow!( "tracked field {} is missing from index layout {:?}", field_idx, self.index_layout ) }) } } } } fn build_index_layout( index_fields: &[FullTextIndexFieldDef], field_layout: FtsFieldLayout, ) -> Result { match field_layout { FtsFieldLayout::FtsFieldLayoutLegacyBody | FtsFieldLayout::FtsFieldLayoutUnspecified => { Ok(IndexLayout::legacy_body()) } FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString => { let mut fields = Vec::with_capacity(index_fields.len()); for (field_idx, index_field) in index_fields.iter().enumerate() { if !index_field.has_index_field_option() { bail!( "FTS index field {} for column {} is missing its index option", field_idx, index_field.get_col_id() ); } let option = index_field.get_index_field_option(); let analyzer_type = option.get_analyzer_type(); let enable_bm25 = option.get_enable_bm25(); if analyzer_type == FullTextAnalyzerType::FtsAnalyzerUnspecified { bail!( "FTS index field {} for column {} requires a specified analyzer", field_idx, index_field.get_col_id() ); } if analyzer_type == FullTextAnalyzerType::FtsAnalyzerPathHierarchyV1 && option.get_path_hierarchy().get_delimiter().is_empty() { bail!( "path hierarchy FTS field {} for column {} requires a non-empty delimiter", field_idx, index_field.get_col_id() ); } let field = Field::from_field_id(field_idx as u32); let field_name = field_idx.to_string(); // This first pass fixes field id/name and BM25 semantics. The // physical value type is resolved from ColumnInfo below. let field_layout = SingleFieldLayout::new_with_field_and_bm25(field, field_name, enable_bm25); fields.push(field_layout); } Ok(IndexLayout::new(fields)) } } } pub fn tracked_index_layout_from_tracked_meta( meta: &FtsTrackedIndexMeta, columns: &[ColumnInfo], ) -> Result { let index_fields = tracked_index_fields_from_meta(meta); validate_tracked_index_fields(meta, meta.get_field_layout(), &index_fields)?; let unresolved_layout = build_index_layout(&index_fields, meta.get_field_layout())?; let index_layout = if index_fields.is_empty() || index_fields .iter() .any(|index_field| !index_field.has_index_field_option()) { // Legacy-body metadata may only carry column ids. Such indexes keep // their historical text layout; typed v1 fields always carry options. unresolved_layout } else { clara_fts::resolve_physical_index_layout(&index_fields, columns, &unresolved_layout)? }; Ok(TrackedIndexLayout { index_fields, field_layout: meta.get_field_layout(), index_layout, }) } /// Management structure for FTS files across all levels #[derive(Clone, Default)] pub struct FtsLevels { tracked_indexes: TrackedFtsIndexMap, /// Runtime-only physical layouts resolved from the current table schema. /// /// This map is rebuilt whenever `ShardData` synchronizes its SchemaFile; /// it is deliberately not persisted in FTS metadata. tracked_index_layouts: HashMap<(i64, i64), TrackedIndexLayout>, /// Incremental processing boundary: all L0 data <= this version has been /// processed. This enables clean separation between historical and /// incremental processing. This is a watermark and never moves backwards. pub l0_snap_version: SnapVersion, /// Columnar L0 ids that still need incremental processing. This is mainly /// used after region merge when `l0_snap_version` has been advanced to the /// max of source/target, but some L0s are still unprocessed. pending_columnar_l0_ids: Vec, l0: Vec, // Order by snap_version ascending l1: Vec, // Order by smallest_lp_key ascending l2: BTreeMap, Vec>, // Grouped by LP key } impl std::fmt::Debug for FtsLevels { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("FtsLevels") .field("tracked_indexes", &self.tracked_indexes) .field("l0_snap_version", &self.l0_snap_version) .field("pending_columnar_l0_ids", &self.pending_columnar_l0_ids) .field_with("l0", |f| { let mut list = f.debug_list(); for file in &self.l0 { list.entry_with(|f| { write!(f, "{} @{}", file.id(), file.props().get_snap_version()) }); } list.finish() }) .field_with("l1", |f| { let mut list = f.debug_list(); for file in &self.l1 { list.entry_with(|f| { write!( f, "{} [{}, {}]", file.id(), hex::encode(file.props().get_smallest_lp_key()), hex::encode(file.props().get_largest_lp_key()) ) }); } list.finish() }) .field_with("l2", |f| { let mut map = f.debug_map(); for (lp_key, files) in &self.l2 { map.entry( &hex::encode(lp_key), &files.iter().map(|file| file.id()).collect::>(), ); } map.finish() }) .finish() } } impl FtsLevels { /// Use this function to modify the L0 files. /// After modification, snap_version will be updated accordingly /// (advance only). pub fn mut_l0(&mut self, f: impl FnOnce(&mut Vec)) { f(&mut self.l0); self.l0.sort_by(|a, b| { a.props() .get_snap_version() .cmp(&b.props().get_snap_version()) }); // We only advance l0_snap_version and never move it backwards, because // l0_snap_version is a watermark for incremental indexing - what has // already been indexed should not be indexed again. // When new L0 files are added, l0_snap_version will be advanced. // When L0 files are removed, l0_snap_version remains unchanged. if let Some(new_version) = self .l0 .last() .map(|file| file.props().get_snap_version().into()) { if new_version > self.l0_snap_version { self.l0_snap_version = new_version; } } } /// Use this function to set the pending L0 ids that need additional /// incremental processing. Normally these are L0 files with /// snap_version <= l0_snap_version. pub fn mut_pending_columnar_l0_ids(&mut self, f: impl FnOnce(&mut Vec)) { f(&mut self.pending_columnar_l0_ids); self.pending_columnar_l0_ids.sort_unstable(); self.pending_columnar_l0_ids.dedup(); } /// Get the pending L0 ids that need additional incremental processing. pub fn pending_columnar_l0_ids(&self) -> &[u64] { &self.pending_columnar_l0_ids } /// Use this function to modify the L1 files. pub fn mut_l1(&mut self, f: impl FnOnce(&mut Vec)) { f(&mut self.l1); // Sort by smallest_lp_key first, then by smallest_key when lp_key is the same self.l1.sort_by(|a, b| { ( a.props().get_smallest_lp_key(), a.props().get_smallest_key(), ) .cmp(&( b.props().get_smallest_lp_key(), b.props().get_smallest_key(), )) }); // Check that L1 files do not have invalid overlaps after modification. // LP key overlap is allowed temporarily, but key ranges must be ordered // properly. The l1 files with overlapping lp key will be compacted // later. for w in self.l1.windows(2) { let (prev, curr) = (&w[0], &w[1]); let (prev_props, curr_props) = (prev.props(), curr.props()); let has_invalid_overlap = ( prev_props.get_largest_lp_key(), prev_props.get_biggest_key(), ) > ( curr_props.get_smallest_lp_key(), curr_props.get_smallest_key(), ); debug_assert!( !has_invalid_overlap, "Overlapping L1 files after modification: id={} [lp: {}, {}] [key: {}, {}] \ overlap id={} [lp: {}, {}] [key: {}, {}]", prev.id(), hex::encode(prev_props.get_smallest_lp_key()), hex::encode(prev_props.get_largest_lp_key()), hex::encode(prev_props.get_smallest_key()), hex::encode(prev_props.get_biggest_key()), curr.id(), hex::encode(curr_props.get_smallest_lp_key()), hex::encode(curr_props.get_largest_lp_key()), hex::encode(curr_props.get_smallest_key()), hex::encode(curr_props.get_biggest_key()), ); } } pub fn l0(&self) -> &[PackedFile] { &self.l0 } pub fn l1(&self) -> &[PackedFile] { &self.l1 } pub fn l2(&self) -> &BTreeMap, Vec> { &self.l2 } /// Inserts multiple new L2 files into the levels structure. pub fn insert_l2_files(&mut self, files: impl IntoIterator) { let mut touched_lps = HashSet::new(); for file in files.into_iter() { let lp_key = file.props().get_lp_key(); touched_lps.insert(lp_key.to_vec()); let files_in_lp = self.l2.entry(lp_key.to_vec()).or_default(); files_in_lp.push(file); } for lp_key in touched_lps { self.l2.get_mut(&lp_key).unwrap().sort_by_key(|f| f.id()); } } /// Inserts a new L2 file into the levels structure. pub fn insert_l2_file(&mut self, file: EDedicatedFile) { self.insert_l2_files(std::iter::once(file)); } /// Removes multiple L2 files from the levels structure. pub fn remove_l2_files(&mut self, file_ids: impl IntoIterator) { let id_set = file_ids.into_iter().collect::>(); for files in self.l2.values_mut() { files.retain(|file| !id_set.contains(&file.id())); } self.l2.retain(|_, files| !files.is_empty()); } /// Removes a specific L2 file from the levels structure. pub fn remove_l2_file(&mut self, file_id: u64) { self.remove_l2_files(std::iter::once(file_id)); } /// Check if a specific index is tracked. pub fn has_tracked_index(&self, table_id: i64, index_id: i64) -> bool { self.tracked_indexes .get(&table_id) .is_some_and(|indexes| indexes.contains_key(&index_id)) } /// Add a new FTS index for tracking. pub fn track_index(&mut self, table_id: i64, index_id: i64) { self.track_legacy_index(table_id, index_id); } /// Add a legacy single-field FTS index for tracking. pub fn track_legacy_index(&mut self, table_id: i64, index_id: i64) { if self.has_tracked_index(table_id, index_id) { return; } self.track_index_meta( table_id, index_id, new_legacy_tracked_index_meta(table_id, index_id), ); } /// Add a new FTS index metadata entry for tracking. pub fn track_index_meta(&mut self, table_id: i64, index_id: i64, meta: FtsTrackedIndexMeta) { // The metadata can only become a usable layout after the next schema // synchronization resolves its ColumnInfo. self.tracked_index_layouts.remove(&(table_id, index_id)); let mut meta = meta; meta.set_table_id(table_id); meta.set_index_id(index_id); let indexes = self.tracked_indexes.entry(table_id).or_default(); match indexes.get_mut(&index_id) { Some(existing) => { if existing.get_field_layout() == FtsFieldLayout::FtsFieldLayoutUnspecified || (existing.get_index_fields().is_empty() && !meta.get_index_fields().is_empty()) { existing.set_field_layout(meta.get_field_layout()); } if existing.get_index_fields().is_empty() && !meta.get_index_fields().is_empty() { existing.set_index_fields(meta.get_index_fields().to_vec().into()); } else if existing.get_index_fields() != meta.get_index_fields() || existing.get_field_layout() != meta.get_field_layout() { debug_assert!( false, "tracked FTS metadata changed in place for ({}, {}): {:?} -> {:?}", table_id, index_id, existing, meta, ); *existing = meta; } } None => { indexes.insert(index_id, meta); } } } /// Synchronize tracked FTS metadata with the latest table schema and build /// the single physical layout used by read and write paths. /// /// `ColumnInfo` remains the only source of SQL type truth. The resolved /// layouts live only in `ShardData` and are rebuilt after restart. pub fn sync_tracked_index_layouts( &mut self, schema_file: Option<&crate::table::schema_file::SchemaFile>, ) { let mut tracked_index_layouts = HashMap::new(); let Some(schema_file) = schema_file else { self.tracked_index_layouts = tracked_index_layouts; return; }; for (&table_id, indexes) in &mut self.tracked_indexes { let Some(schema) = schema_file.get_table(table_id) else { continue; }; for (&index_id, meta) in indexes.iter_mut() { if meta.get_index_fields().is_empty() { if let Some(index_def) = schema .fulltext_indexes .iter() .find(|index_def| index_def.get_index_id() == meta.get_index_id()) { meta.set_index_fields( tracked_index_fields_for_layout(index_def, meta.get_field_layout()) .into(), ); } } match tracked_index_layout_from_tracked_meta(meta, &schema.columns) { Ok(layout) => { tracked_index_layouts.insert((table_id, index_id), layout); } Err(err) => { warn!( "failed to resolve tracked FTS layout for ({}, {}): {}", table_id, index_id, err ); } } } } self.tracked_index_layouts = tracked_index_layouts; } /// Get tracked metadata for a specific index. pub fn tracked_index_meta(&self, table_id: i64, index_id: i64) -> Option<&FtsTrackedIndexMeta> { self.tracked_indexes.get(&table_id)?.get(&index_id) } /// Get the physical layout resolved during the latest schema sync. pub fn tracked_index_layout( &self, table_id: i64, index_id: i64, ) -> Option<&TrackedIndexLayout> { self.tracked_index_layouts.get(&(table_id, index_id)) } /// Remove an FTS index from tracking. pub fn untrack_index(&mut self, table_id: i64, index_id: i64) { self.tracked_index_layouts.remove(&(table_id, index_id)); let Some(indexes) = self.tracked_indexes.get_mut(&table_id) else { return; }; indexes.remove(&index_id); if indexes.is_empty() { self.tracked_indexes.remove(&table_id); } if self.tracked_indexes.is_empty() { self.pending_columnar_l0_ids.clear(); } } /// Iterator over tracked indexes. pub fn iter_tracked_indexes( &self, ) -> impl Iterator)> { self.tracked_indexes.iter() } /// Iterator over tracked index metadata. pub fn iter_tracked_index_metas( &self, ) -> impl Iterator { self.tracked_indexes.iter().flat_map(|(table_id, indexes)| { indexes .iter() .map(move |(index_id, meta)| (table_id, index_id, meta)) }) } /// Reference to tracked indexes. pub fn tracked_indexes_ref(&self) -> &TrackedFtsIndexMap { &self.tracked_indexes } /// Check if incremental indexing is needed. pub fn needs_delta_indexing(&self, max_columnar_l0_version: SnapVersion) -> bool { !self.tracked_indexes.is_empty() && (max_columnar_l0_version > self.l0_snap_version || !self.pending_columnar_l0_ids.is_empty()) } /// Check if the FTS levels are empty. pub fn is_empty(&self) -> bool { self.tracked_indexes.is_empty() } /// Returns whether a columnar L0 file is already covered by the current FTS /// index. /// /// A columnar L0 is considered tracked when: /// - Its `snap_version` is **not newer** than the incremental watermark /// `l0_snap_version`; and /// - Its `file_id` is **not** listed in `pending_l0_ids` (used after region /// merge when the watermark may have advanced beyond some unprocessed /// files). pub fn is_columnar_l0_tracked(&self, file_id: u64, snap_version: SnapVersion) -> bool { if snap_version > self.l0_snap_version { return false; } if self.pending_columnar_l0_ids.is_empty() { return true; } self.pending_columnar_l0_ids .binary_search(&file_id) .is_err() } /// Returns all tracked FTS file IDs grouped by level. /// The first entry contains L0 IDs, the second contains L1 IDs. pub fn get_all_file_ids(&self) -> [Vec; 3] { [ self.l0.iter().map(|file| file.id()).collect(), self.l1.iter().map(|file| file.id()).collect(), self.l2 .values() .flat_map(|files| files.iter()) .map(|file| file.id()) .collect(), ] } /// Retain only FTS indexes for tables that are still active. /// This cleans up index tracking for dropped tables and removes FTS files /// that no longer have active data. pub fn retain_for_tables(&mut self, active_table_ids: &[i64], columnar_l0_file_ids: &[u64]) { let active_set: HashSet = active_table_ids.iter().copied().collect(); // Clean up tracked indexes self.tracked_indexes .retain(|table_id, _| active_set.contains(table_id)); self.tracked_index_layouts .retain(|(table_id, _), _| active_set.contains(table_id)); // Clean up L0/L1/L2 files that belong to dropped tables self.l0.retain(|l0_file| { active_table_ids .iter() .any(|&table_id| l0_file.has_table(table_id)) }); self.l1.retain(|file| { active_table_ids .iter() .any(|&table_id| file.has_table(table_id)) }); self.l2.retain(|_, files| { files.retain(|file| active_table_ids.contains(&file.props().get_table_id())); !files.is_empty() }); // Cleanup pending columnar L0 IDs according to latest columnar L0 files if active_set.is_empty() || self.tracked_indexes.is_empty() { self.pending_columnar_l0_ids.clear(); } else { let l0_id_set: HashSet = columnar_l0_file_ids.iter().copied().collect(); self.mut_pending_columnar_l0_ids(|ids| ids.retain(|id| l0_id_set.contains(id))); } } /// Returns file IDs for L1/L2 files that are no longer covered by any /// tracked index. pub fn collect_cleanup_candidates(&self) -> (Vec, Vec) { let tracked: HashSet<(i64, i64)> = self .iter_tracked_indexes() .flat_map(|(table_id, indexes)| { indexes.keys().map(move |index_id| (*table_id, *index_id)) }) .collect(); if tracked.is_empty() { // No indexes are tracked (e.g. all FTS indexes were dropped). In this case, // all remaining FTS files are obsolete. let l1_remove = self.l1.iter().map(|file| file.id()).collect(); let l2_remove = self .l2 .values() .flat_map(|files| files.iter().map(|file| file.id())) .collect(); return (l1_remove, l2_remove); } let mut l1_remove = Vec::new(); for file in &self.l1 { if should_remove_l1(file, &tracked) { l1_remove.push(file.id()); } } let mut l2_remove = Vec::new(); for files in self.l2.values() { for file in files { let props = file.props(); let pair = (props.get_table_id(), props.get_index_id()); if !tracked.contains(&pair) { l2_remove.push(file.id()); } } } (l1_remove, l2_remove) } } fn should_remove_l1(file: &PackedFile, tracked: &HashSet<(i64, i64)>) -> bool { let props = file.props(); debug_assert!( props.has_smallest_table_index() && props.has_largest_table_index(), "L1 file {} missing table/index bounds", file.id() ); let smallest = props.get_smallest_table_index(); let largest = props.get_largest_table_index(); let min_pair = (smallest.get_table_id(), smallest.get_index_id()); let max_pair = (largest.get_table_id(), largest.get_index_id()); !tracked .iter() .any(|pair| *pair >= min_pair && *pair <= max_pair) } #[cfg(test)] mod tests { use kvenginepb::fts::FullTextIndexFieldDef; use tidb_query_datatype::FieldTypeTp; use super::*; use crate::table::fts::{ IntPk, test_util::{SchemaBuilder, new_schema_file}, }; fn tracked_field(col_id: i64) -> FullTextIndexFieldDef { let mut field = FullTextIndexFieldDef::new(); field.set_col_id(col_id); field .mut_index_field_option() .set_analyzer_type(FullTextAnalyzerType::FtsAnalyzerStandardV1); field.mut_index_field_option().set_enable_bm25(true); field } fn tracked_field_with_option( col_id: i64, analyzer_type: FullTextAnalyzerType, enable_bm25: bool, ) -> FullTextIndexFieldDef { let mut field = tracked_field(col_id); field .mut_index_field_option() .set_analyzer_type(analyzer_type); field.mut_index_field_option().set_enable_bm25(enable_bm25); field } fn column(col_id: i64, field_type: FieldTypeTp) -> ColumnInfo { let mut column = ColumnInfo::new(); column.set_column_id(col_id); column.set_tp(field_type as i32); column } #[test] fn test_tracked_index_layout_from_tracked_meta_accepts_legacy_and_new_layouts() { let legacy_meta = new_legacy_tracked_index_meta(11, 7); let legacy_layout = tracked_index_layout_from_tracked_meta(&legacy_meta, &[]).unwrap(); assert_eq!( legacy_layout.index_layout.first_field().field_name(), "body" ); assert_eq!( legacy_layout.select_field_layout(101).unwrap().field_name(), "body" ); let multi_col_meta = new_tracked_index_meta( 11, 9, vec![tracked_field(101), tracked_field(202)], FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString, ); assert_eq!( tracked_index_field_col_ids(multi_col_meta.get_index_fields()), vec![101, 202] ); let multi_col_layout = tracked_index_layout_from_tracked_meta(&multi_col_meta, &[]).unwrap(); assert_eq!( multi_col_layout .select_field_layout(202) .unwrap() .field_name(), "1" ); assert_eq!( multi_col_layout .select_field_layout(202) .unwrap() .field() .field_id(), 1 ); multi_col_layout.select_field_layout(303).unwrap_err(); } #[test] fn test_tracked_index_layout_rejects_ambiguous_query_column() { let duplicate_col_meta = new_tracked_index_meta( 11, 9, vec![tracked_field(202), tracked_field(202)], FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString, ); let duplicate_col_layout = tracked_index_layout_from_tracked_meta(&duplicate_col_meta, &[]).unwrap(); duplicate_col_layout.select_field_layout(202).unwrap_err(); } #[test] fn test_schema_sync_builds_runtime_physical_layout() { let table_id = 11; let index_id = 9; let index_field = tracked_field_with_option(101, FullTextAnalyzerType::FtsAnalyzerExactValueV1, false); let mut index_def = FullTextIndexDef::new(); index_def.set_index_id(index_id); index_def.set_index_fields(vec![index_field.clone()].into()); let schema = SchemaBuilder::::new(table_id) .column(101, FieldTypeTp::LongLong) .fts_index(index_def) .schema(); let schema_file = new_schema_file(&[&schema]); let mut levels = FtsLevels::default(); levels.track_index_meta( table_id, index_id, new_tracked_index_meta( table_id, index_id, vec![index_field], FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString, ), ); assert!(levels.tracked_index_layout(table_id, index_id).is_none()); levels.sync_tracked_index_layouts(Some(&schema_file)); let layout = levels.tracked_index_layout(table_id, index_id).unwrap(); assert_eq!(layout.index_fields.len(), 1); assert_eq!( layout.index_layout.first_field().field_type(), clara_fts::PhysicalFieldType::I64 ); levels.sync_tracked_index_layouts(None); assert!(levels.tracked_index_layout(table_id, index_id).is_none()); } #[test] fn test_ordinal_layout_maps_field_options_to_bm25_layout() { let mut path_field = tracked_field_with_option(101, FullTextAnalyzerType::FtsAnalyzerPathHierarchyV1, false); path_field .mut_index_field_option() .mut_path_hierarchy() .set_delimiter("/".to_string()); let meta = new_tracked_index_meta( 11, 9, vec![ tracked_field_with_option(101, FullTextAnalyzerType::FtsAnalyzerStandardV1, true), tracked_field_with_option( 101, FullTextAnalyzerType::FtsAnalyzerExactValueV1, false, ), path_field, tracked_field_with_option(202, FullTextAnalyzerType::FtsAnalyzerStandardV1, false), ], FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString, ); let layout = tracked_index_layout_from_tracked_meta(&meta, &[column(101, FieldTypeTp::LongLong)]) .unwrap(); assert_eq!(layout.physical_col_ids(), vec![101, 101, 101, 202]); assert_eq!(layout.index_layout.fields()[0].field_name(), "0"); assert!(layout.index_layout.fields()[0].enable_bm25()); assert!(!layout.index_layout.fields()[1].enable_bm25()); assert!(!layout.index_layout.fields()[2].enable_bm25()); assert!(!layout.index_layout.fields()[3].enable_bm25()); } #[test] fn test_ordinal_layout_accepts_bm25_path_hierarchy() { let mut path_field = tracked_field_with_option(101, FullTextAnalyzerType::FtsAnalyzerPathHierarchyV1, true); path_field .mut_index_field_option() .mut_path_hierarchy() .set_delimiter("/".to_string()); let meta = new_tracked_index_meta( 11, 9, vec![path_field], FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString, ); let layout = tracked_index_layout_from_tracked_meta(&meta, &[]).unwrap(); assert!(layout.index_layout.first_field().enable_bm25()); assert_eq!(layout.index_layout.first_field().field().field_id(), 0); } #[test] fn test_fulltext_index_col_ids_uses_index_fields() { let mut index_def = FullTextIndexDef::new(); index_def.set_col_id(7); let mut first_field = FullTextIndexFieldDef::new(); first_field.set_col_id(11); index_def.mut_index_fields().push(first_field); let mut second_field = FullTextIndexFieldDef::new(); second_field.set_col_id(13); index_def.mut_index_fields().push(second_field); assert_eq!(fulltext_index_col_ids(&index_def), vec![11, 13]); } #[test] fn test_fulltext_index_col_ids_falls_back_to_legacy_col_id() { let mut index_def = FullTextIndexDef::new(); index_def.set_col_id(7); assert_eq!(fulltext_index_col_ids(&index_def), vec![7]); } #[test] fn test_fulltext_index_fields_hydrates_legacy_enable_bm25() { let mut index_def = FullTextIndexDef::new(); index_def.set_col_id(7); index_def.set_parser_type("STANDARD_V1".to_string()); let index_fields = fulltext_index_fields(&index_def); assert_eq!(index_fields.len(), 1); let index_field_option = index_fields[0].get_index_field_option(); assert_eq!( index_field_option.get_analyzer_type(), FullTextAnalyzerType::FtsAnalyzerStandardV1 ); assert!(index_field_option.get_enable_bm25()); } #[test] fn test_tracked_index_fields_from_layout_uses_tracked_field_options() { let tracked_meta = new_tracked_index_meta( 11, 9, vec![ tracked_field_with_option(101, FullTextAnalyzerType::FtsAnalyzerStandardV1, false), tracked_field_with_option(202, FullTextAnalyzerType::FtsAnalyzerExactValueV1, true), ], FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString, ); let tracked_layout = tracked_index_layout_from_tracked_meta( &tracked_meta, &[column(202, FieldTypeTp::LongLong)], ) .unwrap(); let mut index_def = FullTextIndexDef::new(); index_def.set_index_id(9); index_def.mut_index_fields().push(tracked_field_with_option( 101, FullTextAnalyzerType::FtsAnalyzerStandardV1, false, )); index_def.mut_index_fields().push(tracked_field_with_option( 202, FullTextAnalyzerType::FtsAnalyzerExactValueV1, true, )); let index_fields = tracked_index_fields_from_layout_or_index_def(&tracked_layout, &index_def).unwrap(); assert_eq!(tracked_index_field_col_ids(&index_fields), vec![101, 202]); assert!(!index_fields[0].get_index_field_option().get_enable_bm25()); assert_eq!( index_fields[0].get_index_field_option().get_analyzer_type(), FullTextAnalyzerType::FtsAnalyzerStandardV1 ); assert!(index_fields[1].get_index_field_option().get_enable_bm25()); assert_eq!( index_fields[1].get_index_field_option().get_analyzer_type(), FullTextAnalyzerType::FtsAnalyzerExactValueV1 ); } #[test] fn test_tracked_index_fields_from_layout_rejects_schema_option_drift() { let tracked_meta = new_tracked_index_meta( 11, 9, vec![tracked_field_with_option( 101, FullTextAnalyzerType::FtsAnalyzerStandardV1, false, )], FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString, ); let tracked_layout = tracked_index_layout_from_tracked_meta(&tracked_meta, &[]).unwrap(); let mut index_def = FullTextIndexDef::new(); index_def.set_index_id(9); index_def.mut_index_fields().push(tracked_field_with_option( 101, FullTextAnalyzerType::FtsAnalyzerStandardV1, true, )); let error = tracked_index_fields_from_layout_or_index_def(&tracked_layout, &index_def).unwrap_err(); assert!( error .to_string() .contains("tracked index field definitions") ); } #[test] fn test_tracked_index_fields_from_layout_hydrates_legacy_placeholder() { let tracked_meta = new_legacy_tracked_index_meta(11, 9); let tracked_layout = tracked_index_layout_from_tracked_meta(&tracked_meta, &[]).unwrap(); let mut index_def = FullTextIndexDef::new(); index_def.set_index_id(9); index_def.set_col_id(101); index_def.set_parser_type("STANDARD_V1".to_string()); let index_fields = tracked_index_fields_from_layout_or_index_def(&tracked_layout, &index_def).unwrap(); assert_eq!(index_fields.len(), 1); assert_eq!(index_fields[0].get_col_id(), 101); assert_eq!( index_fields[0].get_index_field_option().get_analyzer_type(), FullTextAnalyzerType::FtsAnalyzerStandardV1 ); assert!(index_fields[0].get_index_field_option().get_enable_bm25()); let mut match_expr = tipb::Expr::new(); match_expr.set_tp(tipb::ExprType::ScalarFunc); match_expr.set_sig(tipb::ScalarFuncSig::FtsMatchWord); match_expr.set_val( serde_json::to_vec(&serde_json::json!({ "version": clara_fts::FTS_MATCH_EXPR_METADATA_VERSION, "query": "rust", "fields": [{ "col_id": 101, "index_field": "STANDARD_V1", }], })) .unwrap(), ); let mut query_info = tipb::FtsQueryInfo::new(); query_info.set_query_type(tipb::FtsQueryType::FtsQueryTypeNoScore); query_info.set_index_id(9); query_info .mut_columns() .push(column(101, FieldTypeTp::String)); query_info.mut_match_expr().push(match_expr); clara_fts::Query::new_with_index_layout( query_info, &index_fields, &tracked_layout.index_layout, ) .unwrap(); } }