feat(mito2): support extension index providers and input reuse (#10297)

* feat(mito2): register extension index providers

Signed-off-by: WenyXu <wenymedia@gmail.com>

* feat(mito2): support index input reuse and extension budgets

Signed-off-by: WenyXu <wenymedia@gmail.com>

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>

src/cmd/src/datanode/objbench.rs

@@ -203,6 +203,7 @@ impl ObjbenchCommand {
             max_sequence: None,
             sst_write_format: FormatType::PrimaryKey,
             index_options: Default::default(),
+            index_build_inputs: None,
             index_config: mito_engine_config.index.clone(),
             inverted_index_config: MitoConfig::default().inverted_index,
             fulltext_index_config,

src/mito2/src/access_layer.rs

@@ -44,6 +44,7 @@ use crate::metrics::{COMPACTION_STAGE_ELAPSED, FLUSH_ELAPSED};
 use crate::read::FlatSource;
 use crate::region::options::IndexOptions;
 use crate::sst::file::{FileHandle, RegionFileId, RegionIndexId};
+use crate::sst::index::build::IndexBuildInputs;
 use crate::sst::index::intermediate::IntermediateManager;
 use crate::sst::index::puffin_manager::{PuffinManagerFactory, SstPuffinManager};
 use crate::sst::index::registry::IndexRegistry;
@@ -441,6 +442,7 @@ impl AccessLayer {
                 self.puffin_manager_factory.clone()
             };
             let indexer_builder = IndexerBuilderImpl {
+                index_build_inputs: request.index_build_inputs,
                 build_type: request.op_type.into(),
                 metadata: request.metadata.clone(),
                 puffin_manager: puffin_factory.build(store, path_provider.clone()),
@@ -671,6 +673,8 @@ pub enum OperationType {
 
 /// Contents to build a SST.
 pub struct SstWriteRequest {
+    /// Optional preplanned index work and physical input files.
+    pub index_build_inputs: Option<IndexBuildInputs>,
     pub op_type: OperationType,
     pub metadata: RegionMetadataRef,
     pub source: FlatSource,
@@ -707,6 +711,7 @@ impl EncodedSstIndexBuildParams {
         intermediate_manager: IntermediateManager,
     ) -> IndexerBuilderImpl {
         IndexerBuilderImpl {
+            index_build_inputs: None,
             build_type: OperationType::Flush.into(),
             metadata: self.metadata,
             puffin_manager,
@@ -1026,6 +1031,7 @@ mod tests {
             access_layer
                 .write_sst(
                     SstWriteRequest {
+                        index_build_inputs: None,
                         op_type,
                         metadata,
                         source: new_flat_source_from_record_batches(vec![

src/mito2/src/cache/write_cache.rs

@@ -301,6 +301,7 @@ impl WriteCache {
         let store = self.file_cache.local_store();
         let path_provider = WriteCachePathProvider::new(self.file_cache.clone());
         let indexer = IndexerBuilderImpl {
+            index_build_inputs: write_request.index_build_inputs,
             build_type: write_request.op_type.into(),
             metadata: write_request.metadata.clone(),
             puffin_manager: self
@@ -746,6 +747,7 @@ mod tests {
         ]);
 
         let write_request = SstWriteRequest {
+            index_build_inputs: None,
             op_type,
             metadata,
             source,
@@ -863,6 +865,7 @@ mod tests {
 
         // Write to local cache and upload sst to mock remote store
         let write_request = SstWriteRequest {
+            index_build_inputs: None,
             op_type: OperationType::Flush,
             metadata,
             source,
@@ -955,6 +958,7 @@ mod tests {
 
         // Write to local cache and upload sst to mock remote store
         let write_request = SstWriteRequest {
+            index_build_inputs: None,
             op_type: OperationType::Flush,
             metadata,
             source,

src/mito2/src/compaction/compactor.rs

@@ -653,6 +653,7 @@ impl SstMerger for DefaultSstMerger {
             .access_layer
             .write_sst(
                 SstWriteRequest {
+                    index_build_inputs: None,
                     op_type: OperationType::Compact,
                     metadata: region_metadata.clone(),
                     source,

src/mito2/src/compaction/multi_region/tests.rs

@@ -447,6 +447,7 @@ fn write_request(
     format: FormatType,
 ) -> SstWriteRequest {
     SstWriteRequest {
+        index_build_inputs: None,
         op_type: OperationType::Flush,
         metadata,
         source,

src/mito2/src/compaction/multi_region/writer.rs

@@ -105,6 +105,7 @@ impl<'a> TargetWriter<'a> {
         bloom.mem_threshold_on_create = self.index_threshold(bloom.mem_threshold_on_create());
         let source = output_projection(self.target, source, self.context.cancellation.clone())?;
         Ok(SstWriteRequest {
+            index_build_inputs: None,
             op_type: OperationType::Compact,
             metadata: self.target.metadata.clone(),
             source,

src/mito2/src/flush.rs

@@ -851,6 +851,7 @@ impl RegionFlushTask {
             .map(|f| f == FormatType::Flat)
             .unwrap_or(self.engine_config.default_flat_format);
         SstWriteRequest {
+            index_build_inputs: None,
             op_type: OperationType::Flush,
             metadata: version.metadata.clone(),
             source,

src/mito2/src/metrics.rs

@@ -38,6 +38,11 @@ pub const STAGING_TYPE: &str = "index_staging";
 /// Recycle bin type label.
 pub const RECYCLE_TYPE: &str = "recycle_bin";
 
+/// Discard reason when an operation cannot admit extension index work.
+pub(crate) const INDEX_DISCARD_ADMISSION_REJECTED: &str = "admission_rejected";
+/// Discard reason when a creator exceeds its retained-memory allowance.
+pub(crate) const INDEX_DISCARD_MEMORY_BUDGET_EXCEEDED: &str = "memory_budget_exceeded";
+
 // Write metrics.
 lazy_static! {
     /// Global write buffer size in bytes.
@@ -505,7 +510,15 @@ lazy_static! {
         &[TYPE_LABEL],
     )
     .unwrap();
-    /// Gauge of index create memory usage.
+    /// Optional index kinds discarded by a resource policy, not a build error.
+    pub static ref INDEX_CREATE_DISCARDED_TOTAL: IntCounterVec = register_int_counter_vec!(
+        "greptime_index_create_discarded_total",
+        "index kinds omitted or discarded by resource policy",
+        &[TYPE_LABEL, "reason"],
+    ).unwrap();
+    /// Gauge of index create memory usage. The `extension` type aggregates retained
+    /// memory across optional creators, not individual algorithms: one creator may
+    /// share state across several algorithms.
     pub static ref INDEX_CREATE_MEMORY_USAGE: IntGaugeVec = register_int_gauge_vec!(
         "greptime_index_create_memory_usage",
         "index create memory usage",

src/mito2/src/sst/file.rs

@@ -147,7 +147,7 @@ fn deserialize_level<'de, D: serde::Deserializer<'de>>(deserializer: D) -> Resul
 }
 
 /// Type to store index types for a column.
-pub type IndexTypes = SmallVec<[IndexType; 4]>;
+pub type IndexTypes = SmallVec<[IndexType; 6]>;
 
 /// Cross-region file id.
 ///

src/mito2/src/sst/index/build.rs

@@ -0,0 +1,46 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Shared index plans and physical inputs carried into SST writers.
+
+use std::sync::Arc;
+
+use crate::sst::file::FileHandle;
+use crate::sst::index::puffin_manager::SstPuffinManager;
+use crate::sst::index::registry::IndexBuildPlan;
+
+/// An immutable plan with operation-scoped input files and output constraints.
+/// Currently compaction and explicit rebuild supply these inputs; flush does not
+/// inject extension handlers and builds only the existing built-in indexes.
+#[derive(Clone)]
+pub struct IndexBuildInputs {
+    /// Typed plans shared by output files of this operation.
+    pub plan: Arc<IndexBuildPlan>,
+    /// Pinned physical input files; flushes have no inputs.
+    pub inputs: Arc<[FileHandle]>,
+    /// Reads persisted inputs even when output uses a local write cache.
+    /// Input availability is determined by `inputs.is_empty()`, not this manager.
+    pub input_puffin_manager: SstPuffinManager,
+    /// Allows coarse input reuse only for the complete input scope without slicing.
+    /// The caller that owns the output plan validates this scope.
+    /// This is not a promise that input rows equal live output rows: deduplication
+    /// or deletion may remove rows. Handlers must record input-derived coverage
+    /// separately from actual output counts when reusing approximate summaries.
+    pub can_merge_inputs: bool,
+    /// Existing allocation available to optional index creators; `None` sets no limit here.
+    /// Single-region compaction and rebuild currently provide no shared total limit.
+    pub memory_limit: Option<usize>,
+    /// Exact row count of an unchanged SST during rebuild, otherwise unknown.
+    pub output_rows: Option<u64>,
+}

src/mito2/src/sst/index/column_test.rs

@@ -188,6 +188,7 @@ async fn materialized_runs_match_legacy_index_bytes_and_bloom_lookups() {
     let mut options = IndexOptions::default();
     options.inverted_index.segment_row_count = 3;
     let builder = IndexerBuilderImpl {
+        index_build_inputs: None,
         build_type: IndexBuildType::Flush,
         metadata: metadata.clone(),
         puffin_manager: puffin.clone(),

src/mito2/src/sst/index/handler.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Planning and creation contracts for built-in indexes.
+//! Planning and creation contracts for built-in and injected indexes.
 
 pub(crate) mod builtin;
 
@@ -24,63 +24,128 @@ use common_error::ext::BoxedError;
 use datatypes::arrow::record_batch::RecordBatch;
 use index::target::IndexTarget;
 use store_api::metadata::RegionMetadataRef;
+use store_api::storage::ColumnId;
 
 use crate::read::Batch;
 use crate::region::options::IndexOptions;
-use crate::sst::file::RegionIndexId;
+use crate::sst::file::{ColumnIndexMetadata, FileHandle, RegionIndexId};
 use crate::sst::index::intermediate::IntermediateManager;
 use crate::sst::index::kind::IndexKind;
-use crate::sst::index::puffin_manager::SstPuffinWriter;
+use crate::sst::index::puffin_manager::{SstPuffinManager, SstPuffinWriter};
 use crate::sst::index::{CompletedEntry, IndexBuildType};
 
+/// A requested column format; typed implementation parameters stay in its plan.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PlanEntry {
+    /// Column to index.
+    pub column_id: ColumnId,
+    /// Implementation owning this entry.
+    pub kind: IndexKind,
+    /// Requested format version. The manifest records extension types, not versions.
+    /// A format upgrade requires a forced rebuild; the provider validates Puffin
+    /// properties and payload compatibility before reusing input state.
+    pub format_version: u32,
+}
+
 /// Schema and options used to prepare work without reading input files.
-pub(crate) struct PlanContext<'a> {
+pub struct PlanContext<'a> {
     /// Shared region schema.
     pub metadata: &'a RegionMetadataRef,
     /// Region index options.
     pub index_options: &'a IndexOptions,
     /// Lifecycle event requesting the build.
     pub build_type: IndexBuildType,
+    /// Existing column descriptors of the target SST during rebuild.
+    /// Empty for new outputs, including compaction; input descriptors are available
+    /// through [`BuildContext::inputs`] when constructing a creator.
+    pub existing_indexes: &'a [ColumnIndexMetadata],
 }
 
 /// Inputs for constructing all of one handler's indexes in a single output file.
-pub(crate) struct BuildContext<'a> {
+pub struct BuildContext<'a> {
     /// Shared output schema.
     pub metadata: &'a RegionMetadataRef,
     /// Physical output index identity.
     pub output: RegionIndexId,
+    /// Output manager, possibly backed by a local write cache.
+    pub puffin_manager: &'a SstPuffinManager,
     /// Existing intermediate-file facility shared by built-in creators.
     pub intermediate_manager: &'a IntermediateManager,
+    /// Pinned physical input files.
+    pub inputs: &'a [FileHandle],
+    /// Manager for reading persisted input indexes. With no inputs this falls back
+    /// to the output manager; determine input availability from `inputs.is_empty()`.
+    pub input_puffin_manager: &'a SstPuffinManager,
+    /// Permits coarse input-summary reuse only for a complete single output.
+    /// The caller that owns the output plan verifies this scope before construction.
+    pub can_merge_inputs: bool,
+    /// Retained-memory allowance for an injected creator, capped by the remaining
+    /// operation budget and its plan ceiling. `None` applies no limit at this layer.
+    /// Built-in creators use spill budgets.
+    pub memory_limit: Option<usize>,
     /// Actual output row-group size used to align inverted-index segments.
     pub row_group_size: Option<usize>,
+    /// Exact row count when rebuilding an existing SST; unknown for compaction outputs.
+    pub output_rows: Option<u64>,
 }
 
-/// Prepares one built-in index implementation.
-pub(crate) trait IndexKindHandler: Send + Sync + Debug {
-    /// Built-in implementation owned by this handler.
-    fn kind(&self) -> IndexKind;
+/// Shared plugin installed by the embedding application.
+pub type IndexKindProviderRef = Arc<dyn IndexKindProvider>;
 
-    /// Prepares typed parameters. `None` means no work.
+/// Supplies extension handlers while the engine retains the complete lifecycle.
+pub trait IndexKindProvider: Send + Sync + Debug {
+    /// Related algorithms may share one handler to reuse row traversal.
+    fn handlers(&self) -> Vec<Arc<dyn IndexKindHandler>>;
+}
+
+/// Owns one or more related kinds that can share preparation and row traversal.
+pub trait IndexKindHandler: Send + Sync + Debug {
+    /// Nonempty, unique kinds exclusively owned by this handler. A group must be
+    /// entirely built-in or entirely extension kinds.
+    fn kinds(&self) -> &[IndexKind];
+
+    /// Prepares typed parameters. `None` means no work; an empty entry list may
+    /// still request removal of obsolete descriptors during maintenance.
     fn plan(&self, context: &PlanContext<'_>) -> Result<Option<Arc<dyn PlannedIndex>>, BoxedError>;
 }
 
 /// An immutable typed plan shared by output files of an operation.
 #[async_trait]
-pub(crate) trait PlannedIndex: Send + Sync + Debug {
-    /// Built-in implementation owned by this plan.
-    fn kind(&self) -> IndexKind;
+pub trait PlannedIndex: Send + Sync + Debug {
+    /// Nonempty, unique subset of the handler's kinds, including removal plans.
+    fn kinds(&self) -> &[IndexKind];
 
-    /// Creates a per-file index creator before any output batch is consumed.
+    /// Requested column formats. Every entry kind must belong to this plan.
+    fn entries(&self) -> &[PlanEntry];
+
+    /// Prepares input reuse before any output batch is consumed.
     async fn build(
         &self,
         context: &BuildContext<'_>,
     ) -> Result<Option<Box<dyn IndexCreator>>, BoxedError>;
+
+    /// Whether building this plan can reuse persisted input summaries.
+    /// Row-only plans do not require checking input object existence.
+    fn needs_input_reuse(&self) -> bool {
+        false
+    }
+
+    /// Optional hard ceiling; built-in creators use their existing spill budgets.
+    fn memory_limit(&self) -> Option<usize> {
+        None
+    }
 }
 
 /// Accumulates rows and writes blobs into the engine-owned Puffin lifecycle.
 /// Implementations release retained state on drop, including cancellation.
 #[async_trait]
-pub(crate) trait IndexCreator: Send {
+pub trait IndexCreator: Send {
+    /// Whether rebuilding this index requires scanning SST rows after preparation.
+    /// A creator reusing persisted state must account for output rows without updates.
+    fn requires_row_scan(&self) -> bool {
+        true
+    }
+
     /// Borrows a batch and its reusable primary-key decode cache.
     async fn update(&mut self, batch: &mut Batch) -> Result<(), BoxedError>;
 

src/mito2/src/sst/index/handler/builtin.rs

@@ -27,7 +27,7 @@ use crate::config::{BloomFilterConfig, FulltextIndexConfig, InvertedIndexConfig}
 use crate::sst::index::handler::builtin::bloom::BloomPlan;
 use crate::sst::index::handler::builtin::fulltext::FulltextPlan;
 use crate::sst::index::handler::builtin::inverted::InvertedPlan;
-use crate::sst::index::handler::{IndexKindHandler, PlanContext, PlannedIndex};
+use crate::sst::index::handler::{IndexKindHandler, PlanContext, PlanEntry, PlannedIndex};
 use crate::sst::index::kind::IndexKind;
 
 /// Configuration for one of the existing file-level built-in creators.
@@ -39,11 +39,11 @@ pub(crate) enum BuiltinIndexHandler {
 }
 
 impl IndexKindHandler for BuiltinIndexHandler {
-    fn kind(&self) -> IndexKind {
+    fn kinds(&self) -> &[IndexKind] {
         match self {
-            Self::Inverted(_) => IndexKind::Inverted,
-            Self::Fulltext(_) => IndexKind::Fulltext,
-            Self::Bloom(_) => IndexKind::Bloom,
+            Self::Inverted(_) => &[IndexKind::Inverted],
+            Self::Fulltext(_) => &[IndexKind::Fulltext],
+            Self::Bloom(_) => &[IndexKind::Bloom],
         }
     }
 
@@ -57,6 +57,13 @@ impl IndexKindHandler for BuiltinIndexHandler {
     }
 }
 
+fn target_column_id(target: &IndexTarget) -> store_api::storage::ColumnId {
+    match target {
+        IndexTarget::ColumnId(id) => *id,
+        IndexTarget::JsonPath(path) => path.column_id(),
+    }
+}
+
 fn column_target_id(target: &IndexTarget) -> Option<store_api::storage::ColumnId> {
     match target {
         IndexTarget::ColumnId(id) => Some(*id),
@@ -64,6 +71,18 @@ fn column_target_id(target: &IndexTarget) -> Option<store_api::storage::ColumnId
     }
 }
 
+fn normalize_entries(entries: &mut Vec<PlanEntry>) {
+    entries.sort_by_key(|entry| (entry.column_id, entry.format_version));
+    entries.dedup_by(|a, b| {
+        if a.column_id == b.column_id {
+            b.format_version = b.format_version.max(a.format_version);
+            true
+        } else {
+            false
+        }
+    });
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -103,6 +122,7 @@ mod tests {
                 metadata: &metadata,
                 index_options: &options,
                 build_type: IndexBuildType::Flush,
+                existing_indexes: &[],
             };
             assert!(handler.plan(&context).is_err());
         }

src/mito2/src/sst/index/handler/builtin/bloom.rs

@@ -25,8 +25,10 @@ use index::target::IndexTarget;
 use crate::config::BloomFilterConfig;
 use crate::read::Batch;
 use crate::sst::index::bloom_filter::creator::BloomFilterIndexer;
-use crate::sst::index::handler::builtin::column_target_id;
-use crate::sst::index::handler::{BuildContext, IndexCreator, PlanContext, PlannedIndex};
+use crate::sst::index::handler::builtin::{column_target_id, normalize_entries, target_column_id};
+use crate::sst::index::handler::{
+    BuildContext, IndexCreator, PlanContext, PlanEntry, PlannedIndex,
+};
 use crate::sst::index::kind::IndexKind;
 use crate::sst::index::puffin_manager::SstPuffinWriter;
 use crate::sst::index::{CompletedEntry, IndexBuildType};
@@ -35,6 +37,7 @@ use crate::sst::index::{CompletedEntry, IndexBuildType};
 #[derive(Debug)]
 pub(crate) struct BloomPlan {
     config: BloomFilterConfig,
+    entries: Vec<PlanEntry>,
     targets: Vec<(IndexTarget, SkippingIndexOptions)>,
 }
 
@@ -53,18 +56,26 @@ impl BloomPlan {
             return Ok(None);
         }
         let mut targets = Vec::new();
+        let mut entries = Vec::new();
         for column in &context.metadata.column_metadatas {
             for (target, options) in
                 BloomFilterIndexer::targets(column, &context.build_type).map_err(BoxedError::new)?
             {
+                entries.push(PlanEntry {
+                    column_id: target_column_id(&target),
+                    kind: IndexKind::Bloom,
+                    format_version: 0,
+                });
                 targets.push((target, options));
             }
         }
-        if targets.is_empty() {
+        normalize_entries(&mut entries);
+        if entries.is_empty() {
             return Ok(None);
         }
         Ok(Some(Arc::new(Self {
             config: config.clone(),
+            entries,
             targets,
         })))
     }
@@ -72,8 +83,12 @@ impl BloomPlan {
 
 #[async_trait]
 impl PlannedIndex for BloomPlan {
-    fn kind(&self) -> IndexKind {
-        IndexKind::Bloom
+    fn kinds(&self) -> &[IndexKind] {
+        &[IndexKind::Bloom]
+    }
+
+    fn entries(&self) -> &[PlanEntry] {
+        &self.entries
     }
 
     async fn build(

src/mito2/src/sst/index/handler/builtin/fulltext.rs

@@ -19,14 +19,17 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use common_error::ext::BoxedError;
 use datatypes::arrow::record_batch::RecordBatch;
+use datatypes::schema::FulltextBackend;
 use index::target::IndexTarget;
 
 use crate::config::FulltextIndexConfig;
 use crate::read::Batch;
 use crate::sst::index::fulltext_index;
 use crate::sst::index::fulltext_index::creator::{FulltextIndexer, FulltextTarget};
-use crate::sst::index::handler::builtin::column_target_id;
-use crate::sst::index::handler::{BuildContext, IndexCreator, PlanContext, PlannedIndex};
+use crate::sst::index::handler::builtin::{column_target_id, normalize_entries, target_column_id};
+use crate::sst::index::handler::{
+    BuildContext, IndexCreator, PlanContext, PlanEntry, PlannedIndex,
+};
 use crate::sst::index::kind::IndexKind;
 use crate::sst::index::puffin_manager::SstPuffinWriter;
 use crate::sst::index::{CompletedEntry, IndexBuildType};
@@ -35,6 +38,7 @@ use crate::sst::index::{CompletedEntry, IndexBuildType};
 #[derive(Debug)]
 pub(crate) struct FulltextPlan {
     config: FulltextIndexConfig,
+    entries: Vec<PlanEntry>,
     targets: Vec<FulltextTarget>,
 }
 
@@ -53,10 +57,20 @@ impl FulltextPlan {
             return Ok(None);
         }
         let mut targets = Vec::new();
+        let mut entries = Vec::new();
         for column in &context.metadata.column_metadatas {
             for (target, options) in
                 FulltextIndexer::targets(column, &context.build_type).map_err(BoxedError::new)?
             {
+                entries.push(PlanEntry {
+                    column_id: target_column_id(&target),
+                    kind: IndexKind::Fulltext,
+                    format_version: if options.backend == FulltextBackend::Bloom {
+                        u32::from(fulltext_index::FULLTEXT_BLOOM_VERSION)
+                    } else {
+                        0
+                    },
+                });
                 targets.push((
                     column.column_id,
                     column.column_schema.name.clone(),
@@ -65,11 +79,13 @@ impl FulltextPlan {
                 ));
             }
         }
-        if targets.is_empty() {
+        normalize_entries(&mut entries);
+        if entries.is_empty() {
             return Ok(None);
         }
         Ok(Some(Arc::new(Self {
             config: config.clone(),
+            entries,
             targets,
         })))
     }
@@ -77,8 +93,12 @@ impl FulltextPlan {
 
 #[async_trait]
 impl PlannedIndex for FulltextPlan {
-    fn kind(&self) -> IndexKind {
-        IndexKind::Fulltext
+    fn kinds(&self) -> &[IndexKind] {
+        &[IndexKind::Fulltext]
+    }
+
+    fn entries(&self) -> &[PlanEntry] {
+        &self.entries
     }
 
     async fn build(

src/mito2/src/sst/index/handler/builtin/inverted.rs

@@ -27,8 +27,10 @@ use index::target::IndexTarget;
 
 use crate::config::InvertedIndexConfig;
 use crate::read::Batch;
-use crate::sst::index::handler::builtin::column_target_id;
-use crate::sst::index::handler::{BuildContext, IndexCreator, PlanContext, PlannedIndex};
+use crate::sst::index::handler::builtin::{column_target_id, normalize_entries, target_column_id};
+use crate::sst::index::handler::{
+    BuildContext, IndexCreator, PlanContext, PlanEntry, PlannedIndex,
+};
 use crate::sst::index::inverted_index::creator::InvertedIndexer;
 use crate::sst::index::kind::IndexKind;
 use crate::sst::index::puffin_manager::SstPuffinWriter;
@@ -38,6 +40,7 @@ use crate::sst::index::{CompletedEntry, IndexBuildType};
 #[derive(Debug)]
 pub(crate) struct InvertedPlan {
     config: InvertedIndexConfig,
+    entries: Vec<PlanEntry>,
     targets: Vec<IndexTarget>,
     segment_row_count: NonZeroUsize,
 }
@@ -78,11 +81,21 @@ impl InvertedPlan {
             );
             return Ok(None);
         };
-        if targets.is_empty() {
+        let mut entries: Vec<_> = targets
+            .iter()
+            .map(|target| PlanEntry {
+                column_id: target_column_id(target),
+                kind: IndexKind::Inverted,
+                format_version: 0,
+            })
+            .collect();
+        normalize_entries(&mut entries);
+        if entries.is_empty() {
             return Ok(None);
         }
         Ok(Some(Arc::new(Self {
             config: config.clone(),
+            entries,
             targets,
             segment_row_count,
         })))
@@ -91,8 +104,12 @@ impl InvertedPlan {
 
 #[async_trait]
 impl PlannedIndex for InvertedPlan {
-    fn kind(&self) -> IndexKind {
-        IndexKind::Inverted
+    fn kinds(&self) -> &[IndexKind] {
+        &[IndexKind::Inverted]
+    }
+
+    fn entries(&self) -> &[PlanEntry] {
+        &self.entries
     }
 
     async fn build(
@@ -250,6 +267,7 @@ mod tests {
                 metadata: &metadata,
                 index_options: &options,
                 build_type: build_type.clone(),
+                existing_indexes: &[],
             };
             let plan = InvertedPlan::plan(&config, &context).unwrap().unwrap();
             let columns: Vec<_> = plan

src/mito2/src/sst/index/indexer/abort.rs

@@ -22,10 +22,10 @@ impl Indexer {
     pub(crate) async fn abort_creator(&mut self, position: usize) {
         let mut active = self.creators.remove(position);
         if let Err(err) = active.creator.abort().await {
-            if cfg!(any(test, feature = "test")) {
+            if active.is_builtin() && cfg!(any(test, feature = "test")) {
                 panic!("Failed to abort index: {err:?}");
             }
-            warn!(err; "Failed to abort index, region_id: {}, file_id: {}, kind: {:?}", self.region_id, self.file_id, active.kind);
+            warn!(err; "Failed to abort index, region_id: {}, file_id: {}, kinds: {:?}", self.region_id, self.file_id, active.kinds);
         }
     }
 

src/mito2/src/sst/index/indexer/finish.rs

@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
+use common_error::ext::BoxedError;
 use common_telemetry::warn;
 use puffin::puffin_manager::{PuffinManager, PuffinWriter};
 
@@ -37,6 +40,7 @@ impl Indexer {
                 Ok(entries) => {
                     if let Err(err) = entries.iter().try_for_each(|entry| entry.validate()) {
                         warn!(err; "Invalid index completion, region_id: {}, file_id: {}", self.region_id, self.file_id);
+                        self.failure.get_or_insert_with(|| Arc::new(err));
                         drop(active);
                         self.do_abort().await;
                         return IndexOutput::default();
@@ -44,10 +48,11 @@ impl Indexer {
                     output.entries.extend(entries);
                 }
                 Err(err) => {
-                    if cfg!(any(test, feature = "test")) {
+                    if active.is_builtin() && cfg!(any(test, feature = "test")) {
                         panic!("Failed to finish index: {err:?}");
                     }
-                    warn!(err; "Failed to finish index, region_id: {}, file_id: {}, kind: {:?}", self.region_id, self.file_id, active.kind);
+                    warn!(err; "Failed to finish index, region_id: {}, file_id: {}, kinds: {:?}", self.region_id, self.file_id, active.kinds);
+                    self.record_failure(err);
                     drop(active);
                     self.do_abort().await;
                     return IndexOutput::default();
@@ -89,6 +94,7 @@ impl Indexer {
             );
         }
 
+        self.record_failure(BoxedError::new(err));
         None
     }
 
@@ -105,6 +111,7 @@ impl Indexer {
             self.region_id, self.file_id,
         );
 
+        self.record_failure(BoxedError::new(err));
         0
     }
 

src/mito2/src/sst/index/indexer/update.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
 use common_telemetry::warn;
 use datatypes::arrow::record_batch::RecordBatch;
 
@@ -28,20 +30,30 @@ impl Indexer {
                 panic!("Failed to decode primary key for indexes: {err:?}");
             }
             warn!(err; "Failed to decode primary key for indexes, region_id: {}, file_id: {}", self.region_id, self.file_id);
+            self.failure.get_or_insert_with(|| Arc::new(err));
             self.do_abort().await;
             return;
         }
         let mut position = 0;
         while position < self.creators.len() {
+            if !self.creators[position].creator.requires_row_scan() {
+                position += 1;
+                continue;
+            }
             match self.creators[position].creator.update(batch).await {
                 Ok(()) => position += 1,
                 Err(err) => {
-                    if cfg!(any(test, feature = "test")) {
+                    let builtin = self.creators[position].is_builtin();
+                    if builtin && cfg!(any(test, feature = "test")) {
                         panic!("Failed to update index: {err:?}");
                     }
-                    warn!(err; "Failed to update index, region_id: {}, file_id: {}, kind: {:?}", self.region_id, self.file_id, self.creators[position].kind);
-                    self.do_abort().await;
-                    return;
+                    warn!(err; "Failed to update index, region_id: {}, file_id: {}, kinds: {:?}", self.region_id, self.file_id, self.creators[position].kinds);
+                    self.record_failure(err);
+                    if builtin {
+                        self.do_abort().await;
+                        return;
+                    }
+                    self.abort_creator(position).await;
                 }
             }
         }
@@ -53,15 +65,24 @@ impl Indexer {
         }
         let mut position = 0;
         while position < self.creators.len() {
+            if !self.creators[position].creator.requires_row_scan() {
+                position += 1;
+                continue;
+            }
             match self.creators[position].creator.update_flat(batch).await {
                 Ok(()) => position += 1,
                 Err(err) => {
-                    if cfg!(any(test, feature = "test")) {
+                    let builtin = self.creators[position].is_builtin();
+                    if builtin && cfg!(any(test, feature = "test")) {
                         panic!("Failed to update flat index: {err:?}");
                     }
-                    warn!(err; "Failed to update flat index, region_id: {}, file_id: {}, kind: {:?}", self.region_id, self.file_id, self.creators[position].kind);
-                    self.do_abort().await;
-                    return;
+                    warn!(err; "Failed to update flat index, region_id: {}, file_id: {}, kinds: {:?}", self.region_id, self.file_id, self.creators[position].kinds);
+                    self.record_failure(err);
+                    if builtin {
+                        self.do_abort().await;
+                        return;
+                    }
+                    self.abort_creator(position).await;
                 }
             }
         }

src/mito2/src/sst/index/kind.rs

@@ -16,30 +16,78 @@
 
 use crate::sst::file::IndexType;
 
-/// Index implementations supplied by the storage engine.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
-pub(crate) enum IndexKind {
+/// An index implementation known to the storage engine or supplied by a plugin.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum IndexKind {
+    /// Inverted index.
     Inverted,
+    /// Full-text index.
     Fulltext,
+    /// Bloom filter index.
     Bloom,
+    /// An index whose implementation is supplied by the embedding application.
+    Extension(ExtensionIndexKind),
+}
+
+/// Extension algorithm identities; implementations remain outside mito2.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum ExtensionIndexKind {
+    /// HyperLogLog distinct-value estimate.
+    Hll,
+    /// KLL rank estimate.
+    Kll,
+    /// Misra–Gries frequency estimate.
+    Mg,
+}
+
+impl ExtensionIndexKind {
+    /// Known extension algorithms checked by operation-level provider validation.
+    pub const ALL: [Self; 3] = [Self::Hll, Self::Kll, Self::Mg];
+
+    /// Column metadata key that requires this algorithm's provider.
+    pub const fn required_metadata_key(self) -> &'static str {
+        match self {
+            Self::Hll => datatypes::schema::HLL_KEY,
+            Self::Kll => datatypes::schema::KLL_KEY,
+            Self::Mg => datatypes::schema::MG_KEY,
+        }
+    }
+
+    /// Stable label used by index diagnostics and metrics.
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Hll => "hll_index",
+            Self::Kll => "kll_index",
+            Self::Mg => "mg_index",
+        }
+    }
 }
 
 impl IndexKind {
     /// Stable label used by the existing index metrics.
-    pub(crate) const fn as_str(self) -> &'static str {
+    pub const fn as_str(self) -> &'static str {
         match self {
             Self::Inverted => "inverted_index",
             Self::Fulltext => "fulltext_index",
             Self::Bloom => "bloom_filter_index",
+            Self::Extension(kind) => kind.as_str(),
         }
     }
 
     /// Returns the identity recorded in file and column index metadata.
-    pub(crate) fn index_type(self) -> IndexType {
+    pub fn index_type(self) -> IndexType {
         match self {
             Self::Inverted => IndexType::InvertedIndex,
             Self::Fulltext => IndexType::FulltextIndex,
             Self::Bloom => IndexType::BloomFilterIndex,
+            Self::Extension(ExtensionIndexKind::Hll) => IndexType::HllIndex,
+            Self::Extension(ExtensionIndexKind::Kll) => IndexType::KllIndex,
+            Self::Extension(ExtensionIndexKind::Mg) => IndexType::MgIndex,
         }
     }
+
+    /// Whether this kind is implemented by the storage engine.
+    pub fn is_builtin(self) -> bool {
+        !matches!(self, Self::Extension(_))
+    }
 }

src/mito2/src/sst/index/registry.rs

@@ -12,29 +12,40 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Ordered planning of built-in index implementations.
+//! Ordered registration and planning of index implementations.
 
+use std::collections::HashSet;
 use std::sync::Arc;
 
 use common_error::ext::BoxedError;
 use common_telemetry::warn;
+use snafu::IntoError;
+use store_api::metadata::RegionMetadataRef;
+use store_api::storage::RegionId;
 
 use crate::config::{BloomFilterConfig, FulltextIndexConfig, InvertedIndexConfig};
+use crate::error::{Error, ExternalSnafu, InvalidConfigSnafu};
+use crate::metrics::{INDEX_CREATE_DISCARDED_TOTAL, INDEX_DISCARD_ADMISSION_REJECTED};
+use crate::sst::file::{FileMeta, IndexType};
+use crate::sst::index::IndexBuildType;
 use crate::sst::index::handler::builtin::BuiltinIndexHandler;
-use crate::sst::index::handler::{IndexKindHandler, PlanContext, PlannedIndex};
+use crate::sst::index::handler::{
+    IndexKindHandler, IndexKindProviderRef, PlanContext, PlanEntry, PlannedIndex,
+};
+use crate::sst::index::kind::{ExtensionIndexKind, IndexKind};
 
-/// Shared registry installed by the engine.
-pub(crate) type IndexRegistryRef = Arc<IndexRegistry>;
+/// Shared registry installed by the engine and its embedding application.
+pub type IndexRegistryRef = Arc<IndexRegistry>;
 
-/// Built-in handlers in index creation order.
+/// One handler per owner; related algorithms may share one creator and row traversal.
 #[derive(Debug, Clone)]
-pub(crate) struct IndexRegistry {
+pub struct IndexRegistry {
     handlers: Vec<Arc<dyn IndexKindHandler>>,
 }
 
 impl IndexRegistry {
     /// Registers the existing built-in creators in their established order.
-    pub(crate) fn new(
+    pub fn new(
         inverted: InvertedIndexConfig,
         fulltext: FulltextIndexConfig,
         bloom: BloomFilterConfig,
@@ -48,46 +59,452 @@ impl IndexRegistry {
         }
     }
 
-    /// Prepares each handler once, retaining the existing test panic and
-    /// production log-and-skip policy for construction failures.
-    pub(crate) fn plan(&self, context: &PlanContext<'_>) -> IndexBuildPlan {
+    /// Adds the embedding application's handlers after the built-in configuration.
+    pub fn with_provider(
+        mut self,
+        provider: Option<IndexKindProviderRef>,
+    ) -> Result<Self, BoxedError> {
+        if let Some(provider) = provider {
+            for handler in provider.handlers() {
+                self.register(handler)?;
+            }
+        }
+        Ok(self)
+    }
+
+    /// Installs a handler, rejecting empty or overlapping kind ownership.
+    pub fn register(&mut self, handler: Arc<dyn IndexKindHandler>) -> Result<(), BoxedError> {
+        self.validate_registration(handler.kinds())?;
+        self.handlers.push(handler);
+        self.handlers
+            .sort_by_key(|handler| handler.kinds().iter().min().copied());
+        Ok(())
+    }
+
+    /// Rejects required extension indexes whose implementation is unavailable.
+    /// Checks current schema options and both file-level and column-level metadata.
+    pub fn validate_required_kinds<'a>(
+        &self,
+        metadata: &RegionMetadataRef,
+        files: impl IntoIterator<Item = &'a FileMeta>,
+    ) -> Result<(), BoxedError> {
+        self.validate_schema_kinds(metadata)?;
+        for file in files {
+            for kind in file.available_indexes.iter().chain(
+                file.indexes
+                    .iter()
+                    .flat_map(|column| &column.created_indexes),
+            ) {
+                self.validate_required_kind(*kind)?;
+            }
+        }
+        Ok(())
+    }
+
+    /// Prepares each handler once. Construction failures retain the existing
+    /// built-in test panic and production skip policy. Other failures are retained
+    /// for the caller to check before publishing rebuilt indexes. Required-provider
+    /// validation belongs to the operation entry point, before calling this method.
+    pub fn plan(&self, context: &PlanContext<'_>) -> IndexBuildPlan {
         let mut planned = Vec::new();
+        let mut failure = None;
         for handler in &self.handlers {
-            match handler.plan(context) {
+            if context.build_type == IndexBuildType::Flush
+                && !handler.kinds().iter().all(|kind| kind.is_builtin())
+            {
+                continue;
+            }
+            let result = handler.plan(context).and_then(|plan| {
+                if let Some(plan) = &plan {
+                    Self::validate_plan(handler.as_ref(), plan.as_ref())?;
+                }
+                Ok(plan)
+            });
+            match result {
                 Ok(Some(plan)) => planned.push(plan),
                 Ok(None) => {}
                 Err(err) => {
-                    Self::record_planning_error(handler.as_ref(), context, err);
+                    Self::record_planning_error(handler.as_ref(), context, err, &mut failure);
                 }
             }
         }
-        IndexBuildPlan { planned }
+        IndexBuildPlan { planned, failure }
+    }
+
+    fn validate_registration(&self, kinds: &[IndexKind]) -> Result<(), BoxedError> {
+        if kinds.is_empty()
+            || kinds
+                .iter()
+                .enumerate()
+                .any(|(i, kind)| kinds[..i].contains(kind))
+        {
+            return Err(BoxedError::new(
+                InvalidConfigSnafu {
+                    reason: "index handler kinds must be nonempty and unique",
+                }
+                .build(),
+            ));
+        }
+        if kinds.iter().any(|kind| kind.is_builtin()) && kinds.iter().any(|kind| !kind.is_builtin())
+        {
+            return Err(BoxedError::new(
+                InvalidConfigSnafu {
+                    reason: "index handler cannot mix built-in and extension kinds",
+                }
+                .build(),
+            ));
+        }
+        if kinds.iter().any(|kind| self.has_handler(kind.index_type())) {
+            return Err(BoxedError::new(
+                InvalidConfigSnafu {
+                    reason: "index handler kind is already registered",
+                }
+                .build(),
+            ));
+        }
+        Ok(())
+    }
+
+    fn has_handler(&self, kind: IndexType) -> bool {
+        self.handlers.iter().any(|handler| {
+            handler
+                .kinds()
+                .iter()
+                .any(|owned| owned.index_type() == kind)
+        })
+    }
+
+    fn validate_plan(
+        handler: &dyn IndexKindHandler,
+        plan: &dyn PlannedIndex,
+    ) -> Result<(), BoxedError> {
+        let kinds = plan.kinds();
+        if kinds.is_empty()
+            || kinds
+                .iter()
+                .enumerate()
+                .any(|(i, kind)| kinds[..i].contains(kind) || !handler.kinds().contains(kind))
+            || plan
+                .entries()
+                .iter()
+                .any(|entry| !kinds.contains(&entry.kind))
+        {
+            return Err(BoxedError::new(
+                InvalidConfigSnafu {
+                    reason: "index plan kinds must be nonempty, unique, owned by the handler, and include every entry kind",
+                }
+                .build(),
+            ));
+        }
+        Ok(())
     }
 
     fn record_planning_error(
         handler: &dyn IndexKindHandler,
         context: &PlanContext<'_>,
         err: BoxedError,
+        failure: &mut Option<Arc<Error>>,
     ) {
-        if cfg!(any(test, feature = "test")) {
+        if handler.kinds().iter().all(|kind| kind.is_builtin()) && cfg!(any(test, feature = "test"))
+        {
             panic!(
                 "Failed to prepare built-in index, region_id: {}, err: {err}",
                 context.metadata.region_id
             );
         }
-        warn!(err; "Failed to prepare index, region_id: {}, kind: {:?}", context.metadata.region_id, handler.kind());
+        warn!(err; "Failed to prepare index, region_id: {}, kinds: {:?}", context.metadata.region_id, handler.kinds());
+        failure.get_or_insert_with(|| {
+            Arc::new(
+                ExternalSnafu {
+                    context: "index planning".to_string(),
+                }
+                .into_error(err),
+            )
+        });
+    }
+
+    fn validate_schema_kinds(&self, metadata: &RegionMetadataRef) -> Result<(), BoxedError> {
+        for column in &metadata.column_metadatas {
+            let schema = &column.column_schema;
+            for kind in ExtensionIndexKind::ALL {
+                if schema.metadata().contains_key(kind.required_metadata_key()) {
+                    self.validate_required_kind(IndexKind::Extension(kind).index_type())?;
+                }
+            }
+        }
+        Ok(())
+    }
+
+    fn validate_required_kind(&self, kind: IndexType) -> Result<(), BoxedError> {
+        if ExtensionIndexKind::ALL
+            .iter()
+            .any(|extension| IndexKind::Extension(*extension).index_type() == kind)
+            && !self.has_handler(kind)
+        {
+            return Err(BoxedError::new(
+                InvalidConfigSnafu {
+                    reason: format!("required index {kind:?} has no registered handler"),
+                }
+                .build(),
+            ));
+        }
+        Ok(())
     }
 }
 
 /// Typed plans retained across output files of one operation.
 #[derive(Debug, Clone, Default)]
-pub(crate) struct IndexBuildPlan {
+pub struct IndexBuildPlan {
     planned: Vec<Arc<dyn PlannedIndex>>,
+    failure: Option<Arc<Error>>,
 }
 
 impl IndexBuildPlan {
-    /// Plans in index creation order.
-    pub(crate) fn planned(&self) -> &[Arc<dyn PlannedIndex>] {
+    /// Reports preparation failure separately from a successful empty plan.
+    pub fn check_success(&self) -> Result<(), Arc<Error>> {
+        self.failure.clone().map_or(Ok(()), Err)
+    }
+
+    /// Plans in execution order, grouped by owner rather than by column.
+    pub fn planned(&self) -> &[Arc<dyn PlannedIndex>] {
         &self.planned
     }
+
+    /// All requested column formats.
+    pub fn entries(&self) -> impl Iterator<Item = &PlanEntry> {
+        self.planned.iter().flat_map(|plan| plan.entries())
+    }
+
+    /// Whether any planned implementation can reuse persisted input indexes.
+    pub fn needs_input_reuse(&self) -> bool {
+        self.planned.iter().any(|plan| plan.needs_input_reuse())
+    }
+
+    /// Omits optional extension work when the caller cannot admit its memory budget.
+    /// Records each requested kind once per target plan, rather than once per column.
+    pub fn omit_extensions_for_budget(&mut self, region_id: RegionId) {
+        let omitted: HashSet<_> = self
+            .entries()
+            .filter(|entry| !entry.kind.is_builtin())
+            .map(|entry| entry.kind)
+            .collect();
+        for kind in omitted {
+            warn!(
+                "Omitting index before construction, region_id: {}, kind: {}, reason: admission_rejected",
+                region_id,
+                kind.as_str()
+            );
+            INDEX_CREATE_DISCARDED_TOTAL
+                .with_label_values(&[kind.as_str(), INDEX_DISCARD_ADMISSION_REJECTED])
+                .inc();
+        }
+        self.planned
+            .retain(|plan| plan.kinds().iter().all(|kind| kind.is_builtin()));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::sst::index::kind::ExtensionIndexKind;
+
+    #[derive(Debug)]
+    struct TestHandler(Vec<IndexKind>);
+
+    impl IndexKindHandler for TestHandler {
+        fn kinds(&self) -> &[IndexKind] {
+            &self.0
+        }
+        fn plan(&self, _: &PlanContext<'_>) -> Result<Option<Arc<dyn PlannedIndex>>, BoxedError> {
+            Ok(None)
+        }
+    }
+
+    #[derive(Debug)]
+    struct TestPlan {
+        kinds: Vec<IndexKind>,
+        entries: Vec<PlanEntry>,
+    }
+
+    #[async_trait::async_trait]
+    impl PlannedIndex for TestPlan {
+        fn kinds(&self) -> &[IndexKind] {
+            &self.kinds
+        }
+        fn entries(&self) -> &[PlanEntry] {
+            &self.entries
+        }
+        async fn build(
+            &self,
+            _: &crate::sst::index::handler::BuildContext<'_>,
+        ) -> Result<Option<Box<dyn crate::sst::index::handler::IndexCreator>>, BoxedError> {
+            unreachable!("admission must not construct creators")
+        }
+    }
+
+    #[test]
+    fn test_admission_omission_records_each_requested_kind_once() {
+        let builtin = IndexKind::Bloom;
+        let enabled = IndexKind::Extension(ExtensionIndexKind::Mg);
+        let disabled = IndexKind::Extension(ExtensionIndexKind::Kll);
+        let entry = |kind, column_id| PlanEntry {
+            kind,
+            column_id,
+            format_version: 1,
+        };
+        let mut plan = IndexBuildPlan {
+            planned: vec![
+                Arc::new(TestPlan {
+                    kinds: vec![builtin],
+                    entries: vec![entry(builtin, 0)],
+                }),
+                Arc::new(TestPlan {
+                    kinds: vec![enabled, disabled],
+                    entries: vec![entry(enabled, 0), entry(enabled, 1)],
+                }),
+            ],
+            failure: None,
+        };
+        let counter = INDEX_CREATE_DISCARDED_TOTAL
+            .with_label_values(&[enabled.as_str(), "admission_rejected"]);
+        let disabled_counter = INDEX_CREATE_DISCARDED_TOTAL
+            .with_label_values(&[disabled.as_str(), "admission_rejected"]);
+        let before = counter.get();
+        let disabled_before = disabled_counter.get();
+        plan.omit_extensions_for_budget(RegionId::new(1, 1));
+        assert_eq!(
+            plan.entries().cloned().collect::<Vec<_>>(),
+            vec![entry(builtin, 0)]
+        );
+        assert_eq!(counter.get(), before + 1);
+        assert_eq!(disabled_counter.get(), disabled_before);
+        assert!(plan.check_success().is_ok());
+        plan.omit_extensions_for_budget(RegionId::new(1, 1));
+        assert_eq!(counter.get(), before + 1);
+    }
+
+    #[test]
+    fn test_registration_preserves_grouped_owner_and_rejects_overlap() {
+        let mut registry =
+            IndexRegistry::new(Default::default(), Default::default(), Default::default());
+        let kinds = vec![
+            IndexKind::Extension(ExtensionIndexKind::Hll),
+            IndexKind::Extension(ExtensionIndexKind::Kll),
+            IndexKind::Extension(ExtensionIndexKind::Mg),
+        ];
+        registry
+            .register(Arc::new(TestHandler(kinds.clone())))
+            .unwrap();
+        assert_eq!(registry.handlers.len(), 4);
+        assert_eq!(registry.handlers[3].kinds(), kinds);
+        for invalid in [vec![], vec![kinds[0]], vec![IndexKind::Fulltext]] {
+            assert!(registry.register(Arc::new(TestHandler(invalid))).is_err());
+            assert_eq!(registry.handlers.len(), 4);
+        }
+    }
+
+    #[test]
+    fn test_registration_rejects_duplicates_within_one_owner() {
+        let mut registry =
+            IndexRegistry::new(Default::default(), Default::default(), Default::default());
+        let kind = IndexKind::Extension(ExtensionIndexKind::Hll);
+        assert!(
+            registry
+                .register(Arc::new(TestHandler(vec![kind, kind])))
+                .is_err()
+        );
+        assert_eq!(registry.handlers.len(), 3);
+    }
+
+    #[derive(Debug)]
+    struct PlannedHandler {
+        owned: Vec<IndexKind>,
+        plan: Arc<TestPlan>,
+    }
+
+    impl IndexKindHandler for PlannedHandler {
+        fn kinds(&self) -> &[IndexKind] {
+            &self.owned
+        }
+
+        fn plan(&self, _: &PlanContext<'_>) -> Result<Option<Arc<dyn PlannedIndex>>, BoxedError> {
+            Ok(Some(self.plan.clone()))
+        }
+    }
+
+    #[test]
+    fn test_plans_validate_ownership_without_removing_valid_plans() {
+        let hll = IndexKind::Extension(ExtensionIndexKind::Hll);
+        let kll = IndexKind::Extension(ExtensionIndexKind::Kll);
+        let metadata = Arc::new(crate::test_util::sst_util::sst_region_metadata());
+        let options = Default::default();
+        let context = PlanContext {
+            metadata: &metadata,
+            index_options: &options,
+            build_type: crate::sst::index::IndexBuildType::Compact,
+            existing_indexes: &[],
+        };
+        for (kinds, entry_kind, invalid) in [
+            (vec![], hll, true),
+            (vec![hll, hll], hll, true),
+            (vec![kll], kll, true),
+            (vec![hll], kll, true),
+            (vec![hll], hll, false),
+        ] {
+            let mut registry =
+                IndexRegistry::new(Default::default(), Default::default(), Default::default());
+            registry
+                .register(Arc::new(PlannedHandler {
+                    owned: vec![hll, IndexKind::Extension(ExtensionIndexKind::Mg)],
+                    plan: Arc::new(TestPlan {
+                        kinds,
+                        entries: vec![PlanEntry {
+                            column_id: 0,
+                            kind: entry_kind,
+                            format_version: 1,
+                        }],
+                    }),
+                }))
+                .unwrap();
+            registry
+                .register(Arc::new(PlannedHandler {
+                    owned: vec![kll],
+                    plan: Arc::new(TestPlan {
+                        kinds: vec![kll],
+                        entries: vec![PlanEntry {
+                            column_id: 0,
+                            kind: kll,
+                            format_version: 1,
+                        }],
+                    }),
+                }))
+                .unwrap();
+            let plan = registry.plan(&context);
+            assert_eq!(plan.check_success().is_err(), invalid);
+            assert_eq!(
+                plan.entries()
+                    .filter(|entry| !entry.kind.is_builtin())
+                    .map(|entry| entry.kind)
+                    .collect::<Vec<_>>(),
+                if invalid { vec![kll] } else { vec![hll, kll] }
+            );
+            assert_eq!(registry.handlers.len(), 5);
+        }
+    }
+
+    #[test]
+    fn test_registration_rejects_mixed_builtin_and_extension_owner() {
+        let mut registry = IndexRegistry {
+            handlers: Vec::new(),
+        };
+        assert!(
+            registry
+                .register(Arc::new(TestHandler(vec![
+                    IndexKind::Inverted,
+                    IndexKind::Extension(ExtensionIndexKind::Hll),
+                ])))
+                .is_err()
+        );
+        assert!(registry.handlers.is_empty());
+    }
 }

src/mito2/src/sst/index/sparse_test.rs

@@ -130,6 +130,7 @@ async fn sparse_and_materialized_tags_produce_identical_indexes() {
     let mut options = IndexOptions::default();
     options.inverted_index.segment_row_count = 3;
     let builder = IndexerBuilderImpl {
+        index_build_inputs: None,
         build_type: IndexBuildType::Flush,
         metadata: metadata.clone(),
         puffin_manager: puffin.clone(),

src/mito2/src/sst/parquet.rs

@@ -1560,6 +1560,7 @@ mod tests {
         let intermediate_manager = env.get_intermediate_manager();
 
         let indexer_builder = IndexerBuilderImpl {
+            index_build_inputs: None,
             build_type: IndexBuildType::Flush,
             metadata: metadata.clone(),
             puffin_manager,
@@ -2106,6 +2107,7 @@ mod tests {
         let intermediate_manager = env.get_intermediate_manager();
 
         IndexerBuilderImpl {
+            index_build_inputs: None,
             build_type: IndexBuildType::Flush,
             metadata,
             puffin_manager,
@@ -2227,6 +2229,7 @@ mod tests {
         let intermediate_manager = env.get_intermediate_manager();
 
         let indexer_builder = IndexerBuilderImpl {
+            index_build_inputs: None,
             build_type: IndexBuildType::Flush,
             metadata: metadata.clone(),
             puffin_manager,

src/mito2/src/worker/handle_rebuild_index.rs

@@ -76,6 +76,7 @@ impl<S> RegionWorkerLoop<S> {
         let indexer_builder_ref = Arc::new(IndexerBuilderImpl {
             build_type: build_type.clone(),
             metadata: snapshot.metadata.clone(),
+            index_build_inputs: None,
             registry: Arc::new(crate::sst::index::registry::IndexRegistry::new(
                 self.config.inverted_index.clone(),
                 self.config.fulltext_index.clone(),