feat: Extensible segment components via plugin trait

columnar/src/column_values/u128_based/compact_space/mod.rs

@@ -529,6 +529,7 @@ impl CompactSpaceDecompressor {
 #[cfg(test)]
 mod tests {
 
+    use common::file_slice::FileSlice;
     use itertools::Itertools;
 
     use super::*;

src/directory/mmap_directory/mod.rs

@@ -675,9 +675,10 @@ mod tests {
             reader.reload().unwrap();
             let num_segments = reader.searcher().segment_readers().len();
             assert!(num_segments <= 4);
-            let num_components_except_deletes_and_tempstore =
-                crate::index::SegmentComponent::iterator().len() - 2;
-            let max_num_mmapped = num_components_except_deletes_and_tempstore * num_segments;
+            // Built-in segment components excluding the delete file (no deletes here):
+            // postings, positions, fast fields, field norms, terms, store, temp store.
+            let max_components_per_segment = 7;
+            let max_num_mmapped = max_components_per_segment * num_segments;
             assert_eventually(|| {
                 let num_mmapped = mmap_directory.get_cache_info().mmapped.len();
                 if num_mmapped > max_num_mmapped {

src/error.rs

@@ -110,6 +110,26 @@ pub enum TantivyError {
     #[error("Deserialize error: {0}")]
     /// An error occurred while attempting to deserialize a document.
     DeserializeError(DeserializeError),
+    /// A segment plugin required by the index is not registered.
+    #[error(
+        "Missing required segment plugin(s) for extension(s): {0}. Re-register via \
+         Index::register_plugin before writing, merging, or garbage collecting."
+    )]
+    MissingPlugin(String),
+    /// Multiple registered segment plugins claim ownership of the same extension(s).
+    #[error(
+        "Conflicting segment plugins claim the same extension(s): {0}. Each extension must be \
+         owned by exactly one registered plugin."
+    )]
+    ConflictingPlugins(String),
+    /// A segment plugin was registered on a non-empty index whose existing segments do not
+    /// contain its extension(s).
+    #[error(
+        "Segment plugin(s) for extension(s) {0} were registered on a non-empty index whose \
+         existing segments do not contain them. The plugin set is fixed once an index has \
+         segments; create a new index to change it."
+    )]
+    UnexpectedPlugin(String),
 }
 
 impl From<io::Error> for TantivyError {

src/fastfield/mod.rs

@@ -25,6 +25,8 @@ use columnar::MonotonicallyMappableToU64;
 pub use self::alive_bitset::{intersect_alive_bitsets, write_alive_bitset, AliveBitSet};
 pub use self::error::{FastFieldNotAvailableError, Result};
 pub use self::facet_reader::FacetReader;
+pub mod plugin;
+pub use self::plugin::{FastFieldsPlugin, FastFieldsPluginWriter};
 pub use self::readers::FastFieldReaders;
 pub use self::writer::FastFieldsWriter;
 use crate::schema::Type;

src/fastfield/plugin.rs

@@ -0,0 +1,182 @@
+//! FastFields as a [`SegmentPlugin`] implementation.
+//!
+//! This wraps the existing `FastFieldsWriter` and `FastFieldReaders` types behind
+//! the plugin interface so that fast fields participate in the unified plugin lifecycle.
+
+use std::any::Any;
+use std::collections::BTreeMap;
+
+use columnar::{
+    ColumnType, ColumnarReader, MergeRowOrder, RowAddr, ShuffleMergeOrder, StackMergeOrder,
+};
+use common::TerminatingWrite;
+use measure_time::debug_time;
+
+use crate::directory::{Directory, WritePtr};
+use crate::fastfield::{FastFieldReaders, FastFieldsWriter};
+use crate::index::{SegmentComponent, SegmentReader};
+use crate::indexer::doc_id_mapping::{DocIdMapping, MappingType, SegmentDocIdMapping};
+use crate::plugin::{PluginMergeContext, PluginWriter, PluginWriterContext, SegmentPlugin};
+use crate::schema::document::Document;
+use crate::schema::{value_type_to_column_type, Schema};
+use crate::space_usage::{ComponentSpaceUsage, FAST_FIELDS};
+use crate::Segment;
+
+pub struct FastFieldsPlugin;
+
+impl SegmentPlugin for FastFieldsPlugin {
+    fn extensions(&self) -> &[&str] {
+        &["fast"]
+    }
+
+    fn create_writer(&self, _ctx: &PluginWriterContext) -> crate::Result<Box<dyn PluginWriter>> {
+        unimplemented!("FastFieldsPlugin is a built-in; use FastFieldsPluginWriter::new")
+    }
+
+    fn merge(&self, ctx: PluginMergeContext) -> crate::Result<()> {
+        debug_time!("write-fast-fields");
+        let path = ctx
+            .target_segment
+            .relative_path(SegmentComponent::FastFields);
+        let mut fast_field_wrt: WritePtr =
+            ctx.target_segment.index().directory().open_write(&path)?;
+
+        let required_columns = extract_fast_field_required_columns(ctx.schema);
+        let columnars: Vec<&ColumnarReader> = ctx
+            .readers
+            .iter()
+            .map(|reader| reader.fast_fields().columnar())
+            .collect();
+
+        // Clone the doc_id_mapping since convert_to_merge_order consumes it by value
+        let doc_id_mapping = ctx.doc_id_mapping.clone();
+        let merge_row_order = convert_to_merge_order(&columnars[..], doc_id_mapping);
+
+        columnar::merge_columnar(
+            &columnars[..],
+            &required_columns,
+            merge_row_order,
+            &mut fast_field_wrt,
+        )?;
+
+        fast_field_wrt.terminate()?;
+        Ok(())
+    }
+
+    fn space_usage(
+        &self,
+        segment_reader: &SegmentReader,
+    ) -> crate::Result<BTreeMap<String, ComponentSpaceUsage>> {
+        let file = segment_reader.open_read(SegmentComponent::FastFields)?;
+        let readers = FastFieldReaders::open(file, segment_reader.schema().clone())
+            .map_err(|e| crate::TantivyError::InternalError(e.to_string()))?;
+        let usage = readers.space_usage()?;
+        Ok(BTreeMap::from([(
+            FAST_FIELDS.to_string(),
+            ComponentSpaceUsage::PerField(usage),
+        )]))
+    }
+}
+
+pub struct FastFieldsPluginWriter {
+    pub writer: FastFieldsWriter,
+    fast_field_write: WritePtr,
+}
+
+impl FastFieldsPluginWriter {
+    pub(crate) fn new(ctx: &PluginWriterContext) -> crate::Result<Self> {
+        let index = ctx.segment.index();
+        let tokenizer_manager = index.fast_field_tokenizer().clone();
+        let writer = FastFieldsWriter::from_schema_and_tokenizer_manager(
+            &ctx.segment.schema(),
+            tokenizer_manager,
+        )?;
+
+        let path = ctx.segment.relative_path(SegmentComponent::FastFields);
+        let fast_field_write = index.directory().open_write(&path)?;
+
+        Ok(FastFieldsPluginWriter {
+            writer,
+            fast_field_write,
+        })
+    }
+
+    /// Generic, zero-copy document ingestion. Called directly by the `SegmentWriter` for
+    /// this built-in, so a custom `Document` is indexed in place without materializing a
+    /// `TantivyDocument`.
+    pub(crate) fn index_document<D: Document>(&mut self, doc: &D) -> crate::Result<()> {
+        self.writer.add_document(doc)
+    }
+
+    pub fn writer_mut(&mut self) -> &mut FastFieldsWriter {
+        &mut self.writer
+    }
+
+    pub fn writer(&self) -> &FastFieldsWriter {
+        &self.writer
+    }
+}
+
+impl PluginWriter for FastFieldsPluginWriter {
+    fn serialize(
+        mut self: Box<Self>,
+        _segment: &Segment,
+        doc_id_map: Option<&DocIdMapping>,
+    ) -> crate::Result<()> {
+        self.writer
+            .serialize(&mut self.fast_field_write, doc_id_map)
+            .map_err(|e| crate::TantivyError::InternalError(e.to_string()))?;
+        self.fast_field_write.terminate()?;
+        Ok(())
+    }
+
+    fn mem_usage(&self) -> usize {
+        self.writer.mem_usage()
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn as_any_mut(&mut self) -> &mut dyn Any {
+        self
+    }
+}
+
+// --- Helper functions moved from merger.rs ---
+
+fn convert_to_merge_order(
+    columnars: &[&ColumnarReader],
+    doc_id_mapping: SegmentDocIdMapping,
+) -> MergeRowOrder {
+    match doc_id_mapping.mapping_type() {
+        MappingType::Stacked => MergeRowOrder::Stack(StackMergeOrder::stack(columnars)),
+        MappingType::StackedWithDeletes | MappingType::Shuffled => {
+            let new_row_id_to_old_row_id: Vec<RowAddr> = doc_id_mapping
+                .new_doc_id_to_old_doc_addr
+                .into_iter()
+                .map(|doc_addr| RowAddr {
+                    segment_ord: doc_addr.segment_ord,
+                    row_id: doc_addr.doc_id,
+                })
+                .collect();
+            MergeRowOrder::Shuffled(ShuffleMergeOrder {
+                new_row_id_to_old_row_id,
+                alive_bitsets: doc_id_mapping.alive_bitsets,
+            })
+        }
+    }
+}
+
+fn extract_fast_field_required_columns(schema: &Schema) -> Vec<(String, ColumnType)> {
+    schema
+        .fields()
+        .map(|(_, field_entry)| field_entry)
+        .filter(|field_entry| field_entry.is_fast())
+        .filter_map(|field_entry| {
+            let column_name = field_entry.name().to_string();
+            let column_type = value_type_to_column_type(field_entry.field_type().value_type())?;
+            Some((column_name, column_type))
+        })
+        .collect()
+}

src/fastfield/writer.rs

@@ -234,7 +234,7 @@ impl FastFieldsWriter {
     /// Serializes all of the `FastFieldWriter`s by pushing them in
     /// order to the fast field serializer.
     pub fn serialize(
-        mut self,
+        &mut self,
         wrt: &mut dyn io::Write,
         doc_id_map_opt: Option<&DocIdMapping>,
     ) -> io::Result<()> {