Refactoring dyn Column

remove Column impl on Vec

examples/custom_collector.rs

@@ -7,11 +7,12 @@
 // Of course, you can have a look at the tantivy's built-in collectors
 // such as the `CountCollector` for more examples.
 
+use std::sync::Arc;
+
 use fastfield_codecs::Column;
 // ---
 // Importing tantivy...
 use tantivy::collector::{Collector, SegmentCollector};
-use tantivy::fastfield::DynamicFastFieldReader;
 use tantivy::query::QueryParser;
 use tantivy::schema::{Field, Schema, FAST, INDEXED, TEXT};
 use tantivy::{doc, Index, Score, SegmentReader};
@@ -96,7 +97,7 @@ impl Collector for StatsCollector {
 }
 
 struct StatsSegmentCollector {
-    fast_field_reader: DynamicFastFieldReader<u64>,
+    fast_field_reader: Arc<dyn Column<u64>>,
     stats: Stats,
 }
 

examples/warmer.rs

@@ -2,7 +2,6 @@ use std::cmp::Reverse;
 use std::collections::{HashMap, HashSet};
 use std::sync::{Arc, RwLock, Weak};
 
-use fastfield_codecs::Column;
 use tantivy::collector::TopDocs;
 use tantivy::query::QueryParser;
 use tantivy::schema::{Field, Schema, FAST, TEXT};

fastfield_codecs/src/blockwise_linear.rs

@@ -300,9 +300,13 @@ impl FastFieldCodec for BlockwiseLinearCodec {
         // If this doesn't overflow the algorithm should be fine
         let theorethical_maximum_offset =
             fastfield_accessor.max_value() - fastfield_accessor.min_value();
-        fastfield_accessor
+        if fastfield_accessor
             .max_value()
-            .checked_add(theorethical_maximum_offset)?;
+            .checked_add(theorethical_maximum_offset)
+            .is_none()
+        {
+            return None;
+        }
 
         let first_val_in_first_block = fastfield_accessor.get_val(0);
         let last_elem_in_first_chunk = CHUNK_SIZE.min(fastfield_accessor.num_vals());

fastfield_codecs/src/column.rs

@@ -1,6 +1,4 @@
-use std::ops::Range;
-
-use crate::ColumnIter;
+use std::marker::PhantomData;
 
 pub trait Column<T = u64> {
     /// Return the value associated to the given idx.
@@ -12,18 +10,23 @@ pub trait Column<T = u64> {
     /// May panic if `idx` is greater than the column length.
     fn get_val(&self, idx: u64) -> T;
 
-    /// Returns an iterator over given doc range.
+    /// Fills an output buffer with the fast field values
+    /// associated with the `DocId` going from
+    /// `start` to `start + output.len()`.
+    ///
+    /// Regardless of the type of `Item`, this method works
+    /// - transmuting the output array
+    /// - extracting the `Item`s as if they were `u64`
+    /// - possibly converting the `u64` value to the right type.
     ///
     /// # Panics
     ///
-    /// May panic if `range.end()` is greater than
+    /// May panic if `start + output.len()` is greater than
     /// the segment's `maxdoc`.
-    #[inline]
-    fn get_range(&self, range: Range<u64>) -> ColumnIter<'_, Self, T>
-    where
-        Self: Sized,
-    {
-        ColumnIter::new(self, range)
+    fn get_range(&self, start: u64, output: &mut [T]) {
+        for (out, idx) in output.iter_mut().zip(start..) {
+            *out = self.get_val(idx);
+        }
     }
 
     /// Returns the minimum value for this fast field.
@@ -41,8 +44,103 @@ pub trait Column<T = u64> {
     fn max_value(&self) -> T;
 
     fn num_vals(&self) -> u64;
+
     /// Returns a iterator over the data
     fn iter<'a>(&'a self) -> Box<dyn Iterator<Item = T> + 'a> {
         Box::new((0..self.num_vals()).map(|idx| self.get_val(idx)))
     }
 }
+
+struct VecColumn<'a>(&'a [u64]);
+impl<'a> Column for VecColumn<'a> {
+    fn get_val(&self, position: u64) -> u64 {
+        self.0[position as usize]
+    }
+
+    fn iter<'b>(&'b self) -> Box<dyn Iterator<Item = u64> + 'b> {
+        Box::new(self.0.iter().cloned())
+    }
+
+    fn min_value(&self) -> u64 {
+        self.0.iter().min().cloned().unwrap_or(0)
+    }
+
+    fn max_value(&self) -> u64 {
+        self.0.iter().max().cloned().unwrap_or(0)
+    }
+
+    fn num_vals(&self) -> u64 {
+        self.0.len() as u64
+    }
+}
+
+impl<'a> From<&'a [u64]> for VecColumn<'a> {
+    fn from(data: &'a [u64]) -> Self {
+        Self(data)
+    }
+}
+
+struct MonotonicMappingColumn<C, T, Input> {
+    from_column: C,
+    monotonic_mapping: T,
+    _phantom: PhantomData<Input>,
+}
+
+/// Creates a view of a column transformed by a monotonic mapping.
+pub fn monotonic_map_column<C, T, Input, Output>(
+    from_column: C,
+    monotonic_mapping: T,
+) -> impl Column<Output>
+where
+    C: Column<Input>,
+    T: Fn(Input) -> Output,
+{
+    MonotonicMappingColumn {
+        from_column,
+        monotonic_mapping,
+        _phantom: PhantomData,
+    }
+}
+
+impl<C, T, Input, Output> Column<Output> for MonotonicMappingColumn<C, T, Input>
+where
+    C: Column<Input>,
+    T: Fn(Input) -> Output,
+{
+    fn get_val(&self, idx: u64) -> Output {
+        let from_val = self.from_column.get_val(idx);
+        (self.monotonic_mapping)(from_val)
+    }
+
+    fn min_value(&self) -> Output {
+        let from_min_value = self.from_column.min_value();
+        (self.monotonic_mapping)(from_min_value)
+    }
+
+    fn max_value(&self) -> Output {
+        let from_max_value = self.from_column.max_value();
+        (self.monotonic_mapping)(from_max_value)
+    }
+
+    fn num_vals(&self) -> u64 {
+        self.from_column.num_vals()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_monotonic_mapping() {
+        let vals = &[1u64, 3u64][..];
+        let col = VecColumn::from(vals);
+        let mapped = monotonic_map_column(col, |el| el + 4);
+        assert_eq!(mapped.min_value(), 5u64);
+        assert_eq!(mapped.max_value(), 7u64);
+        assert_eq!(mapped.num_vals(), 2);
+        assert_eq!(mapped.num_vals(), 2);
+        assert_eq!(mapped.get_val(0), 5);
+        assert_eq!(mapped.get_val(0), 7);
+    }
+}

fastfield_codecs/src/lib.rs

@@ -4,9 +4,6 @@ extern crate more_asserts;
 
 use std::io;
 use std::io::Write;
-use std::iter::FusedIterator;
-use std::marker::PhantomData;
-use std::ops::Range;
 
 use common::BinarySerializable;
 use ownedbytes::OwnedBytes;
@@ -17,7 +14,7 @@ pub mod linear;
 
 mod column;
 
-pub use self::column::Column;
+pub use self::column::{monotonic_map_column, Column};
 
 #[derive(PartialEq, Eq, PartialOrd, Ord, Debug, Clone, Copy)]
 #[repr(u8)]
@@ -59,12 +56,12 @@ impl FastFieldCodecType {
 
 /// The FastFieldSerializerEstimate trait is required on all variants
 /// of fast field compressions, to decide which one to choose.
-pub trait FastFieldCodec {
+pub trait FastFieldCodec: 'static {
     /// A codex needs to provide a unique name and id, which is
     /// used for debugging and de/serialization.
     const CODEC_TYPE: FastFieldCodecType;
 
-    type Reader: Column<u64>;
+    type Reader: Column<u64> + 'static;
 
     /// Reads the metadata and returns the CodecReader
     fn open_from_bytes(bytes: OwnedBytes) -> io::Result<Self::Reader>;
@@ -93,98 +90,6 @@ pub struct FastFieldStats {
     pub num_vals: u64,
 }
 
-impl<'a> Column for &'a [u64] {
-    fn get_val(&self, position: u64) -> u64 {
-        self[position as usize]
-    }
-
-    fn iter<'b>(&'b self) -> Box<dyn Iterator<Item = u64> + 'b> {
-        Box::new((self as &[u64]).iter().cloned())
-    }
-
-    fn min_value(&self) -> u64 {
-        self.iter().min().unwrap_or(0)
-    }
-
-    fn max_value(&self) -> u64 {
-        self.iter().max().unwrap_or(0)
-    }
-
-    fn num_vals(&self) -> u64 {
-        self.len() as u64
-    }
-}
-
-pub struct ColumnIter<'a, C: Column<I>, I> {
-    column: &'a C,
-    range: Range<u64>,
-    _phantom: PhantomData<I>,
-}
-
-impl<'a, C: Column<I>, I> ColumnIter<'a, C, I> {
-    #[inline]
-    pub fn new(col: &'a C, range: Range<u64>) -> Self {
-        Self {
-            column: col,
-            range,
-            _phantom: PhantomData,
-        }
-    }
-}
-
-impl<'a, C: Column<I>, I> Iterator for ColumnIter<'a, C, I> {
-    type Item = I;
-
-    #[inline]
-    fn next(&mut self) -> Option<Self::Item> {
-        Some(self.column.get_val(self.range.next()?))
-    }
-
-    #[inline]
-    fn fold<Acc, G>(self, init: Acc, mut g: G) -> Acc
-    where
-        G: FnMut(Acc, Self::Item) -> Acc,
-    {
-        self.range
-            .fold(init, move |acc, idx| g(acc, self.column.get_val(idx)))
-    }
-
-    #[inline]
-    fn size_hint(&self) -> (usize, Option<usize>) {
-        let size = (self.range.end - self.range.start) as usize;
-        (size, Some(size))
-    }
-}
-
-impl<'a, C: Column<I>, I> ExactSizeIterator for ColumnIter<'a, C, I> {
-    #[inline]
-    fn len(&self) -> usize {
-        let size = (self.range.end - self.range.start) as usize;
-        size as usize
-    }
-}
-impl<'a, C: Column<I>, I> FusedIterator for ColumnIter<'a, C, I> {}
-
-impl Column for Vec<u64> {
-    fn get_val(&self, position: u64) -> u64 {
-        self[position as usize]
-    }
-    fn iter<'b>(&'b self) -> Box<dyn Iterator<Item = u64> + 'b> {
-        Box::new((self as &[u64]).iter().cloned())
-    }
-    fn min_value(&self) -> u64 {
-        self.iter().min().unwrap_or(0)
-    }
-
-    fn max_value(&self) -> u64 {
-        self.iter().max().unwrap_or(0)
-    }
-
-    fn num_vals(&self) -> u64 {
-        self.len() as u64
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use proptest::arbitrary::any;
@@ -198,10 +103,10 @@ mod tests {
         data: &[u64],
         name: &str,
     ) -> Option<(f32, f32)> {
-        let estimation = Codec::estimate(&data)?;
+        let estimation = Codec::estimate(&VecColum::from(data))?;
 
         let mut out: Vec<u8> = Vec::new();
-        Codec::serialize(&mut out, &data).unwrap();
+        Codec::serialize(&mut out, &VecColum::from(data)).unwrap();
 
         let actual_compression = out.len() as f32 / (data.len() as f32 * 8.0);
 
@@ -287,6 +192,7 @@ mod tests {
     #[test]
     fn estimation_good_interpolation_case() {
         let data = (10..=20000_u64).collect::<Vec<_>>();
+        let data: VecColum = data.as_slice().into();
 
         let linear_interpol_estimation = LinearCodec::estimate(&data).unwrap();
         assert_le!(linear_interpol_estimation, 0.01);
@@ -300,8 +206,9 @@ mod tests {
     }
     #[test]
     fn estimation_test_bad_interpolation_case() {
-        let data = vec![200, 10, 10, 10, 10, 1000, 20];
+        let data: &[u64] = &[200, 10, 10, 10, 10, 1000, 20];
 
+        let data: VecColum = data.into();
         let linear_interpol_estimation = LinearCodec::estimate(&data).unwrap();
         assert_le!(linear_interpol_estimation, 0.32);
 
@@ -312,6 +219,7 @@ mod tests {
     fn estimation_test_bad_interpolation_case_monotonically_increasing() {
         let mut data: Vec<u64> = (200..=20000_u64).collect();
         data.push(1_000_000);
+        let data: VecColum = data.as_slice().into();
 
         // in this case the linear interpolation can't in fact not be worse than bitpacking,
         // but the estimator adds some threshold, which leads to estimated worse behavior

fastfield_codecs/src/linear.rs

@@ -204,9 +204,13 @@ impl FastFieldCodec for LinearCodec {
         // If this doesn't overflow the algorithm should be fine
         let theorethical_maximum_offset =
             fastfield_accessor.max_value() - fastfield_accessor.min_value();
-        fastfield_accessor
+        if fastfield_accessor
             .max_value()
-            .checked_add(theorethical_maximum_offset)?;
+            .checked_add(theorethical_maximum_offset)
+            .is_none()
+        {
+            return None;
+        }
 
         let first_val = fastfield_accessor.get_val(0);
         let last_val = fastfield_accessor.get_val(fastfield_accessor.num_vals() as u64 - 1);

fastfield_codecs/src/main.rs

@@ -3,9 +3,33 @@ extern crate prettytable;
 use fastfield_codecs::bitpacked::BitpackedCodec;
 use fastfield_codecs::blockwise_linear::BlockwiseLinearCodec;
 use fastfield_codecs::linear::LinearCodec;
-use fastfield_codecs::{FastFieldCodec, FastFieldCodecType, FastFieldStats};
+use fastfield_codecs::{Column, FastFieldCodec, FastFieldCodecType, FastFieldStats};
 use prettytable::{Cell, Row, Table};
 
+struct Data<'a>(&'a [u64]);
+
+impl<'a> Column for Data<'a> {
+    fn get_val(&self, position: u64) -> u64 {
+        self.0[position as usize]
+    }
+
+    fn iter<'b>(&'b self) -> Box<dyn Iterator<Item = u64> + 'b> {
+        Box::new(self.0.iter().cloned())
+    }
+
+    fn min_value(&self) -> u64 {
+        *self.0.iter().min().unwrap_or(&0)
+    }
+
+    fn max_value(&self) -> u64 {
+        *self.0.iter().max().unwrap_or(&0)
+    }
+
+    fn num_vals(&self) -> u64 {
+        self.0.len() as u64
+    }
+}
+
 fn main() {
     let mut table = Table::new();
 
@@ -86,10 +110,11 @@ pub fn get_codec_test_data_sets() -> Vec<(Vec<u64>, &'static str)> {
 pub fn serialize_with_codec<C: FastFieldCodec>(
     data: &[u64],
 ) -> Option<(f32, f32, FastFieldCodecType)> {
+    let data = Data(data);
     let estimation = C::estimate(&data)?;
     let mut out = Vec::new();
     C::serialize(&mut out, &data).unwrap();
-    let actual_compression = out.len() as f32 / (data.len() * 8) as f32;
+    let actual_compression = out.len() as f32 / (data.num_vals() * 8) as f32;
     Some((estimation, actual_compression, C::CODEC_TYPE))
 }
 

src/aggregation/agg_req_with_accessor.rs

@@ -4,14 +4,14 @@ use std::rc::Rc;
 use std::sync::atomic::AtomicU32;
 use std::sync::Arc;
 
+use fastfield_codecs::Column;
+
 use super::agg_req::{Aggregation, Aggregations, BucketAggregationType, MetricAggregation};
 use super::bucket::{HistogramAggregation, RangeAggregation, TermsAggregation};
 use super::metric::{AverageAggregation, StatsAggregation};
 use super::segment_agg_result::BucketCount;
 use super::VecWithNames;
-use crate::fastfield::{
-    type_and_cardinality, DynamicFastFieldReader, FastType, MultiValuedFastFieldReader,
-};
+use crate::fastfield::{type_and_cardinality, FastType, MultiValuedFastFieldReader};
 use crate::schema::{Cardinality, Type};
 use crate::{InvertedIndexReader, SegmentReader, TantivyError};
 
@@ -37,10 +37,16 @@ impl AggregationsWithAccessor {
 #[derive(Clone)]
 pub(crate) enum FastFieldAccessor {
     Multi(MultiValuedFastFieldReader<u64>),
-    Single(DynamicFastFieldReader<u64>),
+    Single(Arc<dyn Column<u64>>),
 }
 impl FastFieldAccessor {
-    pub fn as_single(&self) -> Option<&DynamicFastFieldReader<u64>> {
+    pub fn as_single(&self) -> Option<&dyn Column<u64>> {
+        match self {
+            FastFieldAccessor::Multi(_) => None,
+            FastFieldAccessor::Single(reader) => Some(&**reader),
+        }
+    }
+    pub fn into_single(self) -> Option<Arc<dyn Column<u64>>> {
         match self {
             FastFieldAccessor::Multi(_) => None,
             FastFieldAccessor::Single(reader) => Some(reader),
@@ -118,7 +124,7 @@ impl BucketAggregationWithAccessor {
 pub struct MetricAggregationWithAccessor {
     pub metric: MetricAggregation,
     pub field_type: Type,
-    pub accessor: DynamicFastFieldReader<u64>,
+    pub accessor: Arc<dyn Column>,
 }
 
 impl MetricAggregationWithAccessor {
@@ -134,9 +140,8 @@ impl MetricAggregationWithAccessor {
 
                 Ok(MetricAggregationWithAccessor {
                     accessor: accessor
-                        .as_single()
-                        .expect("unexpected fast field cardinality")
-                        .clone(),
+                        .into_single()
+                        .expect("unexpected fast field cardinality"),
                     field_type,
                     metric: metric.clone(),
                 })

src/aggregation/bucket/histogram/histogram.rs

@@ -15,7 +15,6 @@ use crate::aggregation::intermediate_agg_result::{
     IntermediateAggregationResults, IntermediateBucketResult, IntermediateHistogramBucketEntry,
 };
 use crate::aggregation::segment_agg_result::SegmentAggregationResultsCollector;
-use crate::fastfield::DynamicFastFieldReader;
 use crate::schema::Type;
 use crate::{DocId, TantivyError};
 
@@ -264,7 +263,7 @@ impl SegmentHistogramCollector {
         req: &HistogramAggregation,
         sub_aggregation: &AggregationsWithAccessor,
         field_type: Type,
-        accessor: &DynamicFastFieldReader<u64>,
+        accessor: &dyn Column<u64>,
     ) -> crate::Result<Self> {
         req.validate()?;
         let min = f64_from_fastfield_u64(accessor.min_value(), &field_type);

src/aggregation/bucket/range.rs

@@ -1,7 +1,6 @@
 use std::fmt::Debug;
 use std::ops::Range;
 
-use fastfield_codecs::Column;
 use fnv::FnvHashMap;
 use serde::{Deserialize, Serialize};
 

src/aggregation/bucket/term_agg.rs

@@ -242,13 +242,13 @@ impl TermBuckets {
 
     fn increment_bucket(
         &mut self,
-        term_ids: impl Iterator<Item = u64>,
+        term_ids: &[u64],
         doc: DocId,
         sub_aggregation: &AggregationsWithAccessor,
         bucket_count: &BucketCount,
         blueprint: &Option<SegmentAggregationResultsCollector>,
     ) -> crate::Result<()> {
-        for term_id in term_ids {
+        for &term_id in term_ids {
             let entry = self.entries.entry(term_id as u32).or_insert_with(|| {
                 bucket_count.add_count(1);
 
@@ -432,30 +432,39 @@ impl SegmentTermCollector {
             .as_multi()
             .expect("unexpected fast field cardinatility");
         let mut iter = doc.chunks_exact(4);
+        let mut vals1 = vec![];
+        let mut vals2 = vec![];
+        let mut vals3 = vec![];
+        let mut vals4 = vec![];
         for docs in iter.by_ref() {
+            accessor.get_vals(docs[0], &mut vals1);
+            accessor.get_vals(docs[1], &mut vals2);
+            accessor.get_vals(docs[2], &mut vals3);
+            accessor.get_vals(docs[3], &mut vals4);
+
             self.term_buckets.increment_bucket(
-                accessor.get_vals(docs[0]),
+                &vals1,
                 docs[0],
                 &bucket_with_accessor.sub_aggregation,
                 &bucket_with_accessor.bucket_count,
                 &self.blueprint,
             )?;
             self.term_buckets.increment_bucket(
-                accessor.get_vals(docs[1]),
+                &vals2,
                 docs[1],
                 &bucket_with_accessor.sub_aggregation,
                 &bucket_with_accessor.bucket_count,
                 &self.blueprint,
             )?;
             self.term_buckets.increment_bucket(
-                accessor.get_vals(docs[2]),
+                &vals3,
                 docs[2],
                 &bucket_with_accessor.sub_aggregation,
                 &bucket_with_accessor.bucket_count,
                 &self.blueprint,
             )?;
             self.term_buckets.increment_bucket(
-                accessor.get_vals(docs[3]),
+                &vals4,
                 docs[3],
                 &bucket_with_accessor.sub_aggregation,
                 &bucket_with_accessor.bucket_count,
@@ -463,8 +472,10 @@ impl SegmentTermCollector {
             )?;
         }
         for &doc in iter.remainder() {
+            accessor.get_vals(doc, &mut vals1);
+
             self.term_buckets.increment_bucket(
-                accessor.get_vals(doc),
+                &vals1,
                 doc,
                 &bucket_with_accessor.sub_aggregation,
                 &bucket_with_accessor.bucket_count,
@@ -1323,15 +1334,11 @@ mod bench {
             max_bucket_count: 1_000_001u32,
         };
         b.iter(|| {
-            collector
-                .increment_bucket(
-                    vals.iter().cloned(),
-                    0,
-                    &aggregations_with_accessor,
-                    &bucket_count,
-                    &None,
-                )
-                .unwrap();
+            for &val in &vals {
+                collector
+                    .increment_bucket(&[val], 0, &aggregations_with_accessor, &bucket_count, &None)
+                    .unwrap();
+            }
         })
     }
 

src/aggregation/metric/average.rs

@@ -4,7 +4,6 @@ use fastfield_codecs::Column;
 use serde::{Deserialize, Serialize};
 
 use crate::aggregation::f64_from_fastfield_u64;
-use crate::fastfield::DynamicFastFieldReader;
 use crate::schema::Type;
 use crate::DocId;
 
@@ -58,7 +57,7 @@ impl SegmentAverageCollector {
             data: Default::default(),
         }
     }
-    pub(crate) fn collect_block(&mut self, doc: &[DocId], field: &DynamicFastFieldReader<u64>) {
+    pub(crate) fn collect_block(&mut self, doc: &[DocId], field: &dyn Column<u64>) {
         let mut iter = doc.chunks_exact(4);
         for docs in iter.by_ref() {
             let val1 = field.get_val(docs[0] as u64);

src/aggregation/metric/stats.rs

@@ -2,7 +2,6 @@ use fastfield_codecs::Column;
 use serde::{Deserialize, Serialize};
 
 use crate::aggregation::f64_from_fastfield_u64;
-use crate::fastfield::DynamicFastFieldReader;
 use crate::schema::Type;
 use crate::{DocId, TantivyError};
 
@@ -164,7 +163,7 @@ impl SegmentStatsCollector {
             stats: IntermediateStats::default(),
         }
     }
-    pub(crate) fn collect_block(&mut self, doc: &[DocId], field: &DynamicFastFieldReader<u64>) {
+    pub(crate) fn collect_block(&mut self, doc: &[DocId], field: &dyn Column<u64>) {
         let mut iter = doc.chunks_exact(4);
         for docs in iter.by_ref() {
             let val1 = field.get_val(docs[0] as u64);

src/aggregation/segment_agg_result.rs

@@ -185,10 +185,10 @@ impl SegmentMetricResultCollector {
     pub(crate) fn collect_block(&mut self, doc: &[DocId], metric: &MetricAggregationWithAccessor) {
         match self {
             SegmentMetricResultCollector::Average(avg_collector) => {
-                avg_collector.collect_block(doc, &metric.accessor);
+                avg_collector.collect_block(doc, &*metric.accessor);
             }
             SegmentMetricResultCollector::Stats(stats_collector) => {
-                stats_collector.collect_block(doc, &metric.accessor);
+                stats_collector.collect_block(doc, &*metric.accessor);
             }
         }
     }

src/collector/filter_collector_wrapper.rs

@@ -10,11 +10,12 @@
 // ---
 // Importing tantivy...
 use std::marker::PhantomData;
+use std::sync::Arc;
 
 use fastfield_codecs::Column;
 
 use crate::collector::{Collector, SegmentCollector};
-use crate::fastfield::{DynamicFastFieldReader, FastValue};
+use crate::fastfield::FastValue;
 use crate::schema::Field;
 use crate::{Score, SegmentReader, TantivyError};
 
@@ -160,7 +161,7 @@ where
     TPredicate: 'static,
     TPredicateValue: FastValue,
 {
-    fast_field_reader: DynamicFastFieldReader<TPredicateValue>,
+    fast_field_reader: Arc<dyn Column<TPredicateValue>>,
     segment_collector: TSegmentCollector,
     predicate: TPredicate,
     t_predicate_value: PhantomData<TPredicateValue>,

src/collector/histogram_collector.rs

@@ -1,8 +1,10 @@
+use std::sync::Arc;
+
 use fastdivide::DividerU64;
 use fastfield_codecs::Column;
 
 use crate::collector::{Collector, SegmentCollector};
-use crate::fastfield::{DynamicFastFieldReader, FastValue};
+use crate::fastfield::FastValue;
 use crate::schema::{Field, Type};
 use crate::{DocId, Score};
 
@@ -85,7 +87,7 @@ impl HistogramComputer {
 }
 pub struct SegmentHistogramCollector {
     histogram_computer: HistogramComputer,
-    ff_reader: DynamicFastFieldReader<u64>,
+    ff_reader: Arc<dyn Column<u64>>,
 }
 
 impl SegmentCollector for SegmentHistogramCollector {

src/collector/tests.rs

@@ -1,9 +1,11 @@
+use std::sync::Arc;
+
 use fastfield_codecs::Column;
 
 use super::*;
 use crate::collector::{Count, FilterCollector, TopDocs};
 use crate::core::SegmentReader;
-use crate::fastfield::{BytesFastFieldReader, DynamicFastFieldReader};
+use crate::fastfield::BytesFastFieldReader;
 use crate::query::{AllQuery, QueryParser};
 use crate::schema::{Field, Schema, FAST, TEXT};
 use crate::time::format_description::well_known::Rfc3339;
@@ -158,7 +160,7 @@ pub struct FastFieldTestCollector {
 
 pub struct FastFieldSegmentCollector {
     vals: Vec<u64>,
-    reader: DynamicFastFieldReader<u64>,
+    reader: Arc<dyn Column<u64>>,
 }
 
 impl FastFieldTestCollector {

src/collector/top_score_collector.rs

@@ -1,6 +1,7 @@
 use std::collections::BinaryHeap;
 use std::fmt;
 use std::marker::PhantomData;
+use std::sync::Arc;
 
 use fastfield_codecs::Column;
 
@@ -11,7 +12,7 @@ use crate::collector::tweak_score_top_collector::TweakedScoreTopCollector;
 use crate::collector::{
     CustomScorer, CustomSegmentScorer, ScoreSegmentTweaker, ScoreTweaker, SegmentCollector,
 };
-use crate::fastfield::{DynamicFastFieldReader, FastValue};
+use crate::fastfield::FastValue;
 use crate::query::Weight;
 use crate::schema::Field;
 use crate::{DocAddress, DocId, Score, SegmentOrdinal, SegmentReader, TantivyError};
@@ -131,7 +132,7 @@ impl fmt::Debug for TopDocs {
 }
 
 struct ScorerByFastFieldReader {
-    ff_reader: DynamicFastFieldReader<u64>,
+    ff_reader: Arc<dyn Column<u64>>,
 }
 
 impl CustomSegmentScorer<u64> for ScorerByFastFieldReader {
@@ -409,7 +410,6 @@ impl TopDocs {
     /// # use tantivy::query::QueryParser;
     /// use tantivy::SegmentReader;
     /// use tantivy::collector::TopDocs;
-    /// use tantivy::fastfield::Column;
     /// use tantivy::schema::Field;
     ///
     /// fn create_schema() -> Schema {
@@ -517,7 +517,6 @@ impl TopDocs {
     /// use tantivy::SegmentReader;
     /// use tantivy::collector::TopDocs;
     /// use tantivy::schema::Field;
-    /// use fastfield_codecs::Column;
     ///
     /// # fn create_schema() -> Schema {
     /// #    let mut schema_builder = Schema::builder();

src/core/index.rs

@@ -16,7 +16,7 @@ use crate::directory::MmapDirectory;
 use crate::directory::{Directory, ManagedDirectory, RamDirectory, INDEX_WRITER_LOCK};
 use crate::error::{DataCorruption, TantivyError};
 use crate::indexer::index_writer::{MAX_NUM_THREAD, MEMORY_ARENA_NUM_BYTES_MIN};
-use crate::indexer::segment_updater::save_new_metas;
+use crate::indexer::segment_updater::save_metas;
 use crate::reader::{IndexReader, IndexReaderBuilder};
 use crate::schema::{Field, FieldType, Schema};
 use crate::tokenizer::{TextAnalyzer, TokenizerManager};
@@ -47,6 +47,34 @@ fn load_metas(
         .map_err(From::from)
 }
 
+/// Save the index meta file.
+/// This operation is atomic :
+/// Either
+///  - it fails, in which case an error is returned,
+/// and the `meta.json` remains untouched,
+/// - it succeeds, and `meta.json` is written
+/// and flushed.
+///
+/// This method is not part of tantivy's public API
+fn save_new_metas(
+    schema: Schema,
+    index_settings: IndexSettings,
+    directory: &dyn Directory,
+) -> crate::Result<()> {
+    save_metas(
+        &IndexMeta {
+            index_settings,
+            segments: Vec::new(),
+            schema,
+            opstamp: 0u64,
+            payload: None,
+        },
+        directory,
+    )?;
+    directory.sync_directory()?;
+    Ok(())
+}
+
 /// IndexBuilder can be used to create an index.
 ///
 /// Use in conjunction with `SchemaBuilder`. Global index settings

src/fastfield/bytes/reader.rs

@@ -1,7 +1,9 @@
+use std::sync::Arc;
+
 use fastfield_codecs::Column;
 
 use crate::directory::{FileSlice, OwnedBytes};
-use crate::fastfield::{DynamicFastFieldReader, MultiValueLength};
+use crate::fastfield::MultiValueLength;
 use crate::DocId;
 
 /// Reader for byte array fast fields
@@ -16,13 +18,13 @@ use crate::DocId;
 /// and the start index for the next document, and keeping the bytes in between.
 #[derive(Clone)]
 pub struct BytesFastFieldReader {
-    idx_reader: DynamicFastFieldReader<u64>,
+    idx_reader: Arc<dyn Column<u64>>,
     values: OwnedBytes,
 }
 
 impl BytesFastFieldReader {
     pub(crate) fn open(
-        idx_reader: DynamicFastFieldReader<u64>,
+        idx_reader: Arc<dyn Column<u64>>,
         values_file: FileSlice,
     ) -> crate::Result<BytesFastFieldReader> {
         let values = values_file.read_bytes()?;

src/fastfield/facet_reader.rs

@@ -76,8 +76,7 @@ impl FacetReader {
 
     /// Return the list of facet ordinals associated to a document.
     pub fn facet_ords(&self, doc: DocId, output: &mut Vec<u64>) {
-        output.clear();
-        output.extend(self.term_ords.get_vals(doc))
+        self.term_ords.get_vals(doc, output);
     }
 }
 

src/fastfield/gcd.rs

@@ -3,20 +3,11 @@ use std::num::NonZeroU64;
 
 use common::BinarySerializable;
 use fastdivide::DividerU64;
-use fastfield_codecs::{Column, FastFieldCodec};
+use fastfield_codecs::{monotonic_map_column, Column, FastFieldCodec};
 use ownedbytes::OwnedBytes;
 
 pub const GCD_DEFAULT: u64 = 1;
 
-/// Wrapper for accessing a fastfield.
-///
-/// Holds the data and the codec to the read the data.
-#[derive(Clone)]
-pub struct GCDReader<CodecReader: Column> {
-    gcd_params: GCDParams,
-    reader: CodecReader,
-}
-
 #[derive(Debug, Clone, Copy)]
 struct GCDParams {
     gcd: u64,
@@ -24,12 +15,6 @@ struct GCDParams {
     num_vals: u64,
 }
 
-impl GCDParams {
-    pub fn eval(&self, val: u64) -> u64 {
-        self.min_value + self.gcd * val
-    }
-}
-
 impl BinarySerializable for GCDParams {
     fn serialize<W: Write>(&self, writer: &mut W) -> io::Result<()> {
         self.gcd.serialize(writer)?;
@@ -52,31 +37,13 @@ impl BinarySerializable for GCDParams {
 
 pub fn open_gcd_from_bytes<WrappedCodec: FastFieldCodec>(
     bytes: OwnedBytes,
-) -> io::Result<GCDReader<WrappedCodec::Reader>> {
+) -> io::Result<impl Column> {
     let footer_offset = bytes.len() - 24;
     let (body, mut footer) = bytes.split(footer_offset);
     let gcd_params = GCDParams::deserialize(&mut footer)?;
+    let gcd_remap = move |val: u64| gcd_params.min_value + gcd_params.gcd * val;
     let reader: WrappedCodec::Reader = WrappedCodec::open_from_bytes(body)?;
-    Ok(GCDReader { gcd_params, reader })
-}
-
-impl<C: Column + Clone> Column for GCDReader<C> {
-    #[inline]
-    fn get_val(&self, doc: u64) -> u64 {
-        let val = self.reader.get_val(doc);
-        self.gcd_params.eval(val)
-    }
-
-    fn min_value(&self) -> u64 {
-        self.gcd_params.eval(self.reader.min_value())
-    }
-
-    fn max_value(&self) -> u64 {
-        self.gcd_params.eval(self.reader.max_value())
-    }
-    fn num_vals(&self) -> u64 {
-        self.gcd_params.num_vals
-    }
+    Ok(monotonic_map_column(reader, gcd_remap))
 }
 
 pub fn write_gcd_header<W: Write>(
@@ -134,6 +101,7 @@ mod tests {
     use std::collections::HashMap;
     use std::num::NonZeroU64;
     use std::path::Path;
+    use std::sync::Arc;
     use std::time::{Duration, SystemTime};
 
     use common::HasLen;
@@ -141,11 +109,11 @@ mod tests {
 
     use crate::directory::{CompositeFile, RamDirectory, WritePtr};
     use crate::fastfield::gcd::compute_gcd;
+    use crate::fastfield::reader::open_fast_field;
     use crate::fastfield::serializer::FastFieldCodecEnableCheck;
-    use crate::fastfield::tests::{FIELD, FIELDI64, SCHEMA, SCHEMAI64};
+    use crate::fastfield::tests::{encode_decode_fast_field, FIELD, FIELDI64, SCHEMA, SCHEMAI64};
     use crate::fastfield::{
-        find_gcd, CompositeFastFieldSerializer, DynamicFastFieldReader, FastFieldCodecType,
-        FastFieldsWriter, ALL_CODECS,
+        find_gcd, CompositeFastFieldSerializer, FastFieldCodecType, FastFieldsWriter, ALL_CODECS,
     };
     use crate::schema::{Cardinality, Schema};
     use crate::{DateOptions, DatePrecision, DateTime, Directory};
@@ -187,8 +155,7 @@ mod tests {
         let file = directory.open_read(path).unwrap();
         let composite_file = CompositeFile::open(&file)?;
         let file = composite_file.open_read(*FIELD).unwrap();
-        let fast_field_reader = DynamicFastFieldReader::<i64>::open(file)?;
-
+        let fast_field_reader: Arc<dyn Column<i64>> = open_fast_field(file.read_bytes()?)?;
         assert_eq!(fast_field_reader.get_val(0), -4000i64);
         assert_eq!(fast_field_reader.get_val(1), -3000i64);
         assert_eq!(fast_field_reader.get_val(2), -2000i64);
@@ -229,7 +196,7 @@ mod tests {
         let file = directory.open_read(path).unwrap();
         let composite_file = CompositeFile::open(&file)?;
         let file = composite_file.open_read(*FIELD).unwrap();
-        let fast_field_reader = DynamicFastFieldReader::<u64>::open(file)?;
+        let fast_field_reader = open_fast_field::<u64>(file.read_bytes()?)?;
         assert_eq!(fast_field_reader.get_val(0), 1000u64);
         assert_eq!(fast_field_reader.get_val(1), 2000u64);
         assert_eq!(fast_field_reader.get_val(2), 3000u64);
@@ -258,7 +225,7 @@ mod tests {
 
     #[test]
     pub fn test_fastfield2() {
-        let test_fastfield = DynamicFastFieldReader::<u64>::from(vec![100, 200, 300]);
+        let test_fastfield = encode_decode_fast_field(&[100u64, 200u64, 300u64]);
         assert_eq!(test_fastfield.get_val(0), 100);
         assert_eq!(test_fastfield.get_val(1), 200);
         assert_eq!(test_fastfield.get_val(2), 300);
@@ -324,7 +291,7 @@ mod tests {
         let composite_file = CompositeFile::open(&file)?;
         let file = composite_file.open_read(*FIELD).unwrap();
         let len = file.len();
-        let test_fastfield = DynamicFastFieldReader::<DateTime>::open(file)?;
+        let test_fastfield = open_fast_field::<DateTime>(file.read_bytes()?)?;
 
         assert_eq!(test_fastfield.get_val(0), time1.truncate(precision));
         assert_eq!(test_fastfield.get_val(1), time2.truncate(precision));

src/fastfield/mod.rs

@@ -26,9 +26,8 @@ pub use self::alive_bitset::{intersect_alive_bitsets, write_alive_bitset, AliveB
 pub use self::bytes::{BytesFastFieldReader, BytesFastFieldWriter};
 pub use self::error::{FastFieldNotAvailableError, Result};
 pub use self::facet_reader::FacetReader;
-pub(crate) use self::gcd::{find_gcd, GCDReader, GCD_DEFAULT};
+pub(crate) use self::gcd::{find_gcd, GCD_DEFAULT};
 pub use self::multivalued::{MultiValuedFastFieldReader, MultiValuedFastFieldWriter};
-pub use self::reader::DynamicFastFieldReader;
 pub use self::readers::FastFieldReaders;
 pub(crate) use self::readers::{type_and_cardinality, FastType};
 pub use self::serializer::{Column, CompositeFastFieldSerializer, FastFieldStats};
@@ -266,6 +265,7 @@ mod tests {
     use std::collections::HashMap;
     use std::ops::Range;
     use std::path::Path;
+    use std::sync::Arc;
 
     use common::HasLen;
     use once_cell::sync::Lazy;
@@ -275,6 +275,7 @@ mod tests {
 
     use super::*;
     use crate::directory::{CompositeFile, Directory, RamDirectory, WritePtr};
+    use crate::fastfield::reader::open_fast_field;
     use crate::merge_policy::NoMergePolicy;
     use crate::schema::{Document, Field, Schema, FAST, STRING, TEXT};
     use crate::time::OffsetDateTime;
@@ -295,9 +296,51 @@ mod tests {
     pub static FIELD: Lazy<Field> = Lazy::new(|| SCHEMA.get_field("field").unwrap());
     pub static FIELDI64: Lazy<Field> = Lazy::new(|| SCHEMAI64.get_field("field").unwrap());
 
+    /// Encode values using the most appropriate codec and and then loads it
+    /// right away.
+    ///
+    /// This is useful in tests and bench.
+    pub(crate) fn encode_decode_fast_field<Item: FastValue>(
+        vals: &[Item],
+    ) -> Arc<dyn Column<Item>> {
+        let mut schema_builder = Schema::builder();
+        let field = schema_builder.add_u64_field("field", FAST);
+        let schema = schema_builder.build();
+        let path = Path::new("__dummy__");
+        let directory: RamDirectory = RamDirectory::create();
+        {
+            let write: WritePtr = directory
+                .open_write(path)
+                .expect("With a RamDirectory, this should never fail.");
+            let mut serializer = CompositeFastFieldSerializer::from_write(write)
+                .expect("With a RamDirectory, this should never fail.");
+            let mut fast_field_writers = FastFieldsWriter::from_schema(&schema);
+            {
+                let fast_field_writer = fast_field_writers
+                    .get_field_writer_mut(field)
+                    .expect("With a RamDirectory, this should never fail.");
+                for val in vals {
+                    fast_field_writer.add_val(val.to_u64());
+                }
+            }
+            fast_field_writers
+                .serialize(&mut serializer, &HashMap::new(), None)
+                .unwrap();
+            serializer.close().unwrap();
+        }
+        let file = directory.open_read(path).expect("Failed to open the file");
+        let composite_file = CompositeFile::open(&file).expect("Failed to read the composite file");
+        let field_bytes = composite_file
+            .open_read(field)
+            .expect("File component not found")
+            .read_bytes()
+            .unwrap();
+        open_fast_field(field_bytes).unwrap()
+    }
+
     #[test]
     pub fn test_fastfield() {
-        let test_fastfield = DynamicFastFieldReader::<u64>::from(vec![100, 200, 300]);
+        let test_fastfield = encode_decode_fast_field(&[100u64, 200u64, 300u64]);
         assert_eq!(test_fastfield.get_val(0u64), 100);
         assert_eq!(test_fastfield.get_val(1u64), 200);
         assert_eq!(test_fastfield.get_val(2u64), 300);
@@ -328,8 +371,8 @@ mod tests {
         let file = directory.open_read(path).unwrap();
         assert_eq!(file.len(), 45);
         let composite_file = CompositeFile::open(&file)?;
-        let file = composite_file.open_read(*FIELD).unwrap();
-        let fast_field_reader = DynamicFastFieldReader::<u64>::open(file)?;
+        let fast_field_bytes = composite_file.open_read(*FIELD).unwrap().read_bytes()?;
+        let fast_field_reader = open_fast_field::<u64>(fast_field_bytes)?;
         assert_eq!(fast_field_reader.get_val(0), 13u64);
         assert_eq!(fast_field_reader.get_val(1), 14u64);
         assert_eq!(fast_field_reader.get_val(2), 2u64);
@@ -360,8 +403,11 @@ mod tests {
         assert_eq!(file.len(), 70);
         {
             let fast_fields_composite = CompositeFile::open(&file)?;
-            let data = fast_fields_composite.open_read(*FIELD).unwrap();
-            let fast_field_reader = DynamicFastFieldReader::<u64>::open(data)?;
+            let data = fast_fields_composite
+                .open_read(*FIELD)
+                .unwrap()
+                .read_bytes()?;
+            let fast_field_reader = open_fast_field::<u64>(data)?;
             assert_eq!(fast_field_reader.get_val(0), 4u64);
             assert_eq!(fast_field_reader.get_val(1), 14_082_001u64);
             assert_eq!(fast_field_reader.get_val(2), 3_052u64);
@@ -396,8 +442,11 @@ mod tests {
         assert_eq!(file.len(), 43);
         {
             let fast_fields_composite = CompositeFile::open(&file).unwrap();
-            let data = fast_fields_composite.open_read(*FIELD).unwrap();
-            let fast_field_reader = DynamicFastFieldReader::<u64>::open(data)?;
+            let data = fast_fields_composite
+                .open_read(*FIELD)
+                .unwrap()
+                .read_bytes()?;
+            let fast_field_reader = open_fast_field::<u64>(data)?;
             for doc in 0..10_000 {
                 assert_eq!(fast_field_reader.get_val(doc), 100_000u64);
             }
@@ -428,8 +477,11 @@ mod tests {
         assert_eq!(file.len(), 80051);
         {
             let fast_fields_composite = CompositeFile::open(&file)?;
-            let data = fast_fields_composite.open_read(*FIELD).unwrap();
-            let fast_field_reader = DynamicFastFieldReader::<u64>::open(data)?;
+            let data = fast_fields_composite
+                .open_read(*FIELD)
+                .unwrap()
+                .read_bytes()?;
+            let fast_field_reader = open_fast_field::<u64>(data)?;
             assert_eq!(fast_field_reader.get_val(0), 0u64);
             for doc in 1..10_001 {
                 assert_eq!(
@@ -469,15 +521,19 @@ mod tests {
         assert_eq!(file.len(), 75_usize); // linear interpol size after calc improvement
         {
             let fast_fields_composite = CompositeFile::open(&file)?;
-            let data = fast_fields_composite.open_read(i64_field).unwrap();
-            let fast_field_reader = DynamicFastFieldReader::<i64>::open(data)?;
+            let data = fast_fields_composite
+                .open_read(i64_field)
+                .unwrap()
+                .read_bytes()?;
+            let fast_field_reader = open_fast_field::<i64>(data)?;
 
             assert_eq!(fast_field_reader.min_value(), -100i64);
             assert_eq!(fast_field_reader.max_value(), 9_999i64);
             for (doc, i) in (-100i64..10_000i64).enumerate() {
                 assert_eq!(fast_field_reader.get_val(doc as u64), i);
             }
-            let buffer: Vec<i64> = fast_field_reader.get_range(53..154).collect();
+            let mut buffer = vec![0i64; 100];
+            fast_field_reader.get_range(53, &mut buffer[..]);
             for i in 0..100 {
                 assert_eq!(buffer[i], -100i64 + 53i64 + i as i64);
             }
@@ -508,8 +564,11 @@ mod tests {
         let file = directory.open_read(path).unwrap();
         {
             let fast_fields_composite = CompositeFile::open(&file).unwrap();
-            let data = fast_fields_composite.open_read(i64_field).unwrap();
-            let fast_field_reader = DynamicFastFieldReader::<i64>::open(data)?;
+            let data = fast_fields_composite
+                .open_read(i64_field)
+                .unwrap()
+                .read_bytes()?;
+            let fast_field_reader = open_fast_field::<i64>(data)?;
             assert_eq!(fast_field_reader.get_val(0), 0i64);
         }
         Ok(())
@@ -546,8 +605,11 @@ mod tests {
         let file = directory.open_read(path)?;
         {
             let fast_fields_composite = CompositeFile::open(&file)?;
-            let data = fast_fields_composite.open_read(*FIELD).unwrap();
-            let fast_field_reader = DynamicFastFieldReader::<u64>::open(data)?;
+            let data = fast_fields_composite
+                .open_read(*FIELD)
+                .unwrap()
+                .read_bytes()?;
+            let fast_field_reader = open_fast_field::<u64>(data)?;
 
             for a in 0..n {
                 assert_eq!(fast_field_reader.get_val(a as u64), permutation[a as usize]);
@@ -606,7 +668,9 @@ mod tests {
         let mut all = vec![];
 
         for doc in docs {
-            all.extend(ff.get_vals(doc));
+            let mut out: Vec<u64> = vec![];
+            ff.get_vals(doc, &mut out);
+            all.extend(out);
         }
         all
     }
@@ -651,7 +715,8 @@ mod tests {
                 vec![1, 0, 0, 0, 1, 2]
             );
 
-            let out = text_fast_field.get_vals(3u32).collect::<Vec<_>>();
+            let mut out = vec![];
+            text_fast_field.get_vals(3, &mut out);
             assert_eq!(out, vec![0, 1]);
 
             let inverted_index = segment_reader.inverted_index(text_field)?;
@@ -836,20 +901,22 @@ mod tests {
         let fast_fields = segment_reader.fast_fields();
         let date_fast_field = fast_fields.date(date_field).unwrap();
         let dates_fast_field = fast_fields.dates(multi_date_field).unwrap();
+        let mut dates = vec![];
         {
             assert_eq!(date_fast_field.get_val(0).into_timestamp_micros(), 1i64);
-            let dates = dates_fast_field.get_vals(0u32).collect::<Vec<_>>();
+            dates_fast_field.get_vals(0u32, &mut dates);
             assert_eq!(dates.len(), 2);
             assert_eq!(dates[0].into_timestamp_micros(), 2i64);
             assert_eq!(dates[1].into_timestamp_micros(), 3i64);
         }
         {
             assert_eq!(date_fast_field.get_val(1).into_timestamp_micros(), 4i64);
-            assert!(dates_fast_field.get_vals(1u32).next().is_none());
+            dates_fast_field.get_vals(1u32, &mut dates);
+            assert!(dates.is_empty());
         }
         {
             assert_eq!(date_fast_field.get_val(2).into_timestamp_micros(), 0i64);
-            let dates = dates_fast_field.get_vals(2u32).collect::<Vec<_>>();
+            dates_fast_field.get_vals(2u32, &mut dates);
             assert_eq!(dates.len(), 2);
             assert_eq!(dates[0].into_timestamp_micros(), 5i64);
             assert_eq!(dates[1].into_timestamp_micros(), 6i64);
@@ -859,7 +926,7 @@ mod tests {
 
     #[test]
     pub fn test_fastfield_bool() {
-        let test_fastfield = DynamicFastFieldReader::<bool>::from(vec![true, false, true, false]);
+        let test_fastfield = encode_decode_fast_field::<bool>(&[true, false, true, false]);
         assert_eq!(test_fastfield.get_val(0), true);
         assert_eq!(test_fastfield.get_val(1), false);
         assert_eq!(test_fastfield.get_val(2), true);
@@ -892,8 +959,8 @@ mod tests {
         let file = directory.open_read(path).unwrap();
         assert_eq!(file.len(), 44);
         let composite_file = CompositeFile::open(&file)?;
-        let file = composite_file.open_read(field).unwrap();
-        let fast_field_reader = DynamicFastFieldReader::<bool>::open(file)?;
+        let data = composite_file.open_read(field).unwrap().read_bytes()?;
+        let fast_field_reader = open_fast_field::<bool>(data)?;
         assert_eq!(fast_field_reader.get_val(0), true);
         assert_eq!(fast_field_reader.get_val(1), false);
         assert_eq!(fast_field_reader.get_val(2), true);
@@ -928,8 +995,8 @@ mod tests {
         let file = directory.open_read(path).unwrap();
         assert_eq!(file.len(), 56);
         let composite_file = CompositeFile::open(&file)?;
-        let file = composite_file.open_read(field).unwrap();
-        let fast_field_reader = DynamicFastFieldReader::<bool>::open(file)?;
+        let data = composite_file.open_read(field).unwrap().read_bytes()?;
+        let fast_field_reader = open_fast_field::<bool>(data)?;
         for i in 0..25 {
             assert_eq!(fast_field_reader.get_val(i * 2), true);
             assert_eq!(fast_field_reader.get_val(i * 2 + 1), false);
@@ -962,8 +1029,8 @@ mod tests {
         let file = directory.open_read(path).unwrap();
         assert_eq!(file.len(), 43);
         let composite_file = CompositeFile::open(&file)?;
-        let file = composite_file.open_read(field).unwrap();
-        let fast_field_reader = DynamicFastFieldReader::<bool>::open(file)?;
+        let data = composite_file.open_read(field).unwrap().read_bytes()?;
+        let fast_field_reader = open_fast_field::<bool>(data)?;
         assert_eq!(fast_field_reader.get_val(0), false);
 
         Ok(())
@@ -972,93 +1039,17 @@ mod tests {
 
 #[cfg(all(test, feature = "unstable"))]
 mod bench {
-    use std::collections::HashMap;
-    use std::path::Path;
+    use std::sync::Arc;
 
     use fastfield_codecs::Column;
     use test::{self, Bencher};
 
-    use super::tests::{generate_permutation, FIELD, SCHEMA};
-    use super::*;
-    use crate::directory::{CompositeFile, Directory, RamDirectory, WritePtr};
-    use crate::fastfield::tests::generate_permutation_gcd;
-    use crate::schema::{NumericOptions, Schema};
-    use crate::Document;
-
-    fn multi_values(num_docs: usize, vals_per_doc: usize) -> Vec<Vec<u64>> {
-        let mut vals = vec![];
-        for _i in 0..num_docs {
-            let mut block = vec![];
-            for j in 0..vals_per_doc {
-                block.push(j as u64);
-            }
-            vals.push(block);
-        }
-
-        vals
-    }
+    use crate::fastfield::tests::{
+        encode_decode_fast_field, generate_permutation, generate_permutation_gcd,
+    };
 
     #[bench]
-    fn bench_multi_value_fflookup(b: &mut Bencher) {
-        let num_docs = 100_000;
-
-        let path = Path::new("test");
-        let directory: RamDirectory = RamDirectory::create();
-        {
-            let options = NumericOptions::default().set_fast(Cardinality::MultiValues);
-            let mut schema_builder = Schema::builder();
-            let field = schema_builder.add_u64_field("field", options);
-            let schema = schema_builder.build();
-
-            let write: WritePtr = directory.open_write(Path::new("test")).unwrap();
-            let mut serializer = CompositeFastFieldSerializer::from_write(write).unwrap();
-            let mut fast_field_writers = FastFieldsWriter::from_schema(&schema);
-            for block in &multi_values(num_docs, 3) {
-                let mut doc = Document::new();
-                for val in block {
-                    doc.add_u64(field, *val);
-                }
-                fast_field_writers.add_document(&doc);
-            }
-            fast_field_writers
-                .serialize(&mut serializer, &HashMap::new(), None)
-                .unwrap();
-            serializer.close().unwrap();
-        }
-        let file = directory.open_read(&path).unwrap();
-        {
-            let fast_fields_composite = CompositeFile::open(&file).unwrap();
-            let data_idx = fast_fields_composite.open_read_with_idx(*FIELD, 0).unwrap();
-            let idx_reader = DynamicFastFieldReader::<u64>::open(data_idx).unwrap();
-
-            let data_vals = fast_fields_composite.open_read_with_idx(*FIELD, 1).unwrap();
-            let vals_reader = DynamicFastFieldReader::<u64>::open(data_vals).unwrap();
-            let fast_field_reader = MultiValuedFastFieldReader::open(idx_reader, vals_reader);
-            b.iter(|| {
-                let mut sum = 0u64;
-                for i in 0u32..num_docs as u32 {
-                    sum += fast_field_reader.get_vals(i).sum::<u64>();
-                }
-                sum
-            });
-        }
-    }
-
-    #[bench]
-    fn bench_intfastfield_linear_veclookup(b: &mut Bencher) {
-        let permutation = generate_permutation();
-        b.iter(|| {
-            let n = test::black_box(7000u32);
-            let mut a = 0u64;
-            for i in (0u32..n / 7).map(|v| v * 7) {
-                a ^= permutation[i as usize];
-            }
-            a
-        });
-    }
-
-    #[bench]
-    fn bench_intfastfield_veclookup(b: &mut Bencher) {
+    fn bench_intfastfield_jumpy_veclookup(b: &mut Bencher) {
         let permutation = generate_permutation();
         b.iter(|| {
             let n = test::black_box(1000u32);
@@ -1071,102 +1062,81 @@ mod bench {
     }
 
     #[bench]
-    fn bench_intfastfield_linear_fflookup(b: &mut Bencher) {
-        let path = Path::new("test");
+    fn bench_intfastfield_jumpy_fflookup(b: &mut Bencher) {
         let permutation = generate_permutation();
-        let directory: RamDirectory = RamDirectory::create();
-        {
-            let write: WritePtr = directory.open_write(Path::new("test")).unwrap();
-            let mut serializer = CompositeFastFieldSerializer::from_write(write).unwrap();
-            let mut fast_field_writers = FastFieldsWriter::from_schema(&SCHEMA);
-            for &x in &permutation {
-                fast_field_writers.add_document(&doc!(*FIELD=>x));
+        let column: Arc<dyn Column<u64>> = encode_decode_fast_field(&permutation);
+        b.iter(|| {
+            let n = test::black_box(1000u32);
+            let mut a = 0u64;
+            for _ in 0u32..n {
+                a = column.get_val(a as u64);
             }
-            fast_field_writers
-                .serialize(&mut serializer, &HashMap::new(), None)
-                .unwrap();
-            serializer.close().unwrap();
-        }
-        let file = directory.open_read(&path).unwrap();
-        {
-            let fast_fields_composite = CompositeFile::open(&file).unwrap();
-            let data = fast_fields_composite.open_read(*FIELD).unwrap();
-            let fast_field_reader = DynamicFastFieldReader::<u64>::open(data).unwrap();
+            a
+        });
+    }
 
-            b.iter(|| {
-                let n = test::black_box(7000u32);
-                let mut a = 0u64;
-                for i in (0u32..n / 7).map(|val| val * 7) {
-                    a ^= fast_field_reader.get_val(i as u64);
-                }
-                a
-            });
-        }
+    #[bench]
+    fn bench_intfastfield_linear_fflookup(b: &mut Bencher) {
+        let permutation = generate_permutation();
+        let column: Arc<dyn Column<u64>> = encode_decode_fast_field(&permutation);
+        b.iter(|| {
+            let n = test::black_box(7000u32);
+            let mut a = 0u64;
+            for i in (0..n / 7).map(|val| val * 7) {
+                a += column.get_val(i as u64);
+            }
+            a
+        });
+    }
+
+    #[bench]
+    fn bench_intfastfield_linear_vec(b: &mut Bencher) {
+        let permutation = generate_permutation();
+        b.iter(|| {
+            let n = test::black_box(7000);
+            let mut a = 0u64;
+            for i in (0..n / 7).map(|val| val * 7) {
+                a += permutation[i];
+            }
+            a
+        });
     }
 
     #[bench]
     fn bench_intfastfield_fflookup(b: &mut Bencher) {
-        let path = Path::new("test");
         let permutation = generate_permutation();
-        let directory: RamDirectory = RamDirectory::create();
-        {
-            let write: WritePtr = directory.open_write(Path::new("test")).unwrap();
-            let mut serializer = CompositeFastFieldSerializer::from_write(write).unwrap();
-            let mut fast_field_writers = FastFieldsWriter::from_schema(&SCHEMA);
-            for &x in &permutation {
-                fast_field_writers.add_document(&doc!(*FIELD=>x));
+        let column: Arc<dyn Column<u64>> = encode_decode_fast_field(&permutation);
+        b.iter(|| {
+            let mut a = 0u64;
+            for i in 0u64..permutation.len() as u64 {
+                a = column.get_val(i);
             }
-            fast_field_writers
-                .serialize(&mut serializer, &HashMap::new(), None)
-                .unwrap();
-            serializer.close().unwrap();
-        }
-        let file = directory.open_read(&path).unwrap();
-        {
-            let fast_fields_composite = CompositeFile::open(&file).unwrap();
-            let data = fast_fields_composite.open_read(*FIELD).unwrap();
-            let fast_field_reader = DynamicFastFieldReader::<u64>::open(data).unwrap();
-
-            b.iter(|| {
-                let mut a = 0u32;
-                for i in 0u64..permutation.len() as u64 {
-                    a = fast_field_reader.get_val(i) as u32;
-                }
-                a
-            });
-        }
+            a
+        });
     }
 
     #[bench]
     fn bench_intfastfield_fflookup_gcd(b: &mut Bencher) {
-        let path = Path::new("test");
         let permutation = generate_permutation_gcd();
-        let directory: RamDirectory = RamDirectory::create();
-        {
-            let write: WritePtr = directory.open_write(Path::new("test")).unwrap();
-            let mut serializer = CompositeFastFieldSerializer::from_write(write).unwrap();
-            let mut fast_field_writers = FastFieldsWriter::from_schema(&SCHEMA);
-            for &x in &permutation {
-                fast_field_writers.add_document(&doc!(*FIELD=>x));
+        let column: Arc<dyn Column<u64>> = encode_decode_fast_field(&permutation);
+        b.iter(|| {
+            let mut a = 0u64;
+            for i in 0..permutation.len() as u64 {
+                a += column.get_val(i);
             }
-            fast_field_writers
-                .serialize(&mut serializer, &HashMap::new(), None)
-                .unwrap();
-            serializer.close().unwrap();
-        }
-        let file = directory.open_read(&path).unwrap();
-        {
-            let fast_fields_composite = CompositeFile::open(&file).unwrap();
-            let data = fast_fields_composite.open_read(*FIELD).unwrap();
-            let fast_field_reader = DynamicFastFieldReader::<u64>::open(data).unwrap();
+            a
+        });
+    }
 
-            b.iter(|| {
-                let mut a = 0u32;
-                for i in 0u32..permutation.len() as u32 {
-                    a = fast_field_reader.get_val(i as u64) as u32;
-                }
-                a
-            });
-        }
+    #[bench]
+    fn bench_intfastfield_vec(b: &mut Bencher) {
+        let permutation = generate_permutation_gcd();
+        b.iter(|| {
+            let mut a = 0u64;
+            for i in 0..permutation.len() {
+                a += permutation[i as usize] as u64;
+            }
+            a
+        });
     }
 }

src/fastfield/multivalued/mod.rs

@@ -36,17 +36,19 @@ mod tests {
 
         let searcher = index.reader()?.searcher();
         let segment_reader = searcher.segment_reader(0);
+        let mut vals = Vec::new();
         let multi_value_reader = segment_reader.fast_fields().u64s(field)?;
         {
-            let vals = multi_value_reader.get_vals(2u32).collect::<Vec<_>>();
+            multi_value_reader.get_vals(2, &mut vals);
             assert_eq!(&vals, &[4u64]);
         }
         {
-            let vals = multi_value_reader.get_vals(0u32).collect::<Vec<_>>();
+            multi_value_reader.get_vals(0, &mut vals);
             assert_eq!(&vals, &[1u64, 3u64]);
         }
         {
-            assert!(multi_value_reader.get_vals(1u32).next().is_none());
+            multi_value_reader.get_vals(1, &mut vals);
+            assert!(vals.is_empty());
         }
         Ok(())
     }
@@ -211,13 +213,15 @@ mod tests {
 
         let searcher = index.reader()?.searcher();
         let segment_reader = searcher.segment_reader(0);
+        let mut vals = Vec::new();
         let multi_value_reader = segment_reader.fast_fields().i64s(field).unwrap();
-        let vals = multi_value_reader.get_vals(2u32).collect::<Vec<_>>();
+        multi_value_reader.get_vals(2, &mut vals);
         assert_eq!(&vals, &[-4i64]);
-        let vals = multi_value_reader.get_vals(0u32).collect::<Vec<_>>();
+        multi_value_reader.get_vals(0, &mut vals);
         assert_eq!(&vals, &[1i64, 3i64]);
-        assert!(multi_value_reader.get_vals(1u32).next().is_none());
-        let vals = multi_value_reader.get_vals(3u32).collect::<Vec<_>>();
+        multi_value_reader.get_vals(1, &mut vals);
+        assert!(vals.is_empty());
+        multi_value_reader.get_vals(3, &mut vals);
         assert_eq!(&vals, &[-5i64, -20i64, 1i64]);
         Ok(())
     }
@@ -241,13 +245,15 @@ mod tests {
 
         let searcher = index.reader()?.searcher();
         let segment_reader = searcher.segment_reader(0);
+        let mut vals = Vec::new();
         let multi_value_reader = segment_reader.fast_fields().bools(bool_field).unwrap();
-        let vals = multi_value_reader.get_vals(2u32).collect::<Vec<_>>();
+        multi_value_reader.get_vals(2, &mut vals);
         assert_eq!(&vals, &[false]);
-        let vals = multi_value_reader.get_vals(0u32).collect::<Vec<_>>();
+        multi_value_reader.get_vals(0, &mut vals);
         assert_eq!(&vals, &[true, false]);
-        assert!(multi_value_reader.get_vals(1u32).next().is_none());
-        let vals = multi_value_reader.get_vals(3u32).collect::<Vec<_>>();
+        multi_value_reader.get_vals(1, &mut vals);
+        assert!(vals.is_empty());
+        multi_value_reader.get_vals(3, &mut vals);
         assert_eq!(&vals, &[true, true, false]);
         Ok(())
     }
@@ -340,6 +346,7 @@ mod tests {
             assert!(test_multivalued_no_panic(&ops[..]).is_ok());
         }
     }
+
     #[test]
     fn test_multivalued_proptest_gcd() {
         use IndexingOp::*;

src/fastfield/multivalued/reader.rs

@@ -1,8 +1,9 @@
 use std::ops::Range;
+use std::sync::Arc;
 
 use fastfield_codecs::Column;
 
-use crate::fastfield::{DynamicFastFieldReader, FastValue, MultiValueLength};
+use crate::fastfield::{FastValue, MultiValueLength};
 use crate::DocId;
 
 /// Reader for a multivalued `u64` fast field.
@@ -14,16 +15,14 @@ use crate::DocId;
 /// The `idx_reader` associated, for each document, the index of its first value.
 #[derive(Clone)]
 pub struct MultiValuedFastFieldReader<Item: FastValue> {
-    idx_reader: DynamicFastFieldReader<u64>,
-    vals_reader: DynamicFastFieldReader<Item>,
+    idx_reader: Arc<dyn Column<u64>>,
+    vals_reader: Arc<dyn Column<Item>>,
 }
 
-impl<Item: FastValue> MultiValuedFastFieldReader<Item>
-where DynamicFastFieldReader<Item>: Column<Item>
-{
+impl<Item: FastValue> MultiValuedFastFieldReader<Item> {
     pub(crate) fn open(
-        idx_reader: DynamicFastFieldReader<u64>,
-        vals_reader: DynamicFastFieldReader<Item>,
+        idx_reader: Arc<dyn Column<u64>>,
+        vals_reader: Arc<dyn Column<Item>>,
     ) -> MultiValuedFastFieldReader<Item> {
         MultiValuedFastFieldReader {
             idx_reader,
@@ -43,9 +42,17 @@ where DynamicFastFieldReader<Item>: Column<Item>
 
     /// Returns the array of values associated to the given `doc`.
     #[inline]
-    pub fn get_vals(&self, doc: DocId) -> impl Iterator<Item = Item> + '_ {
+    fn get_vals_for_range(&self, range: Range<u64>, vals: &mut Vec<Item>) {
+        let len = (range.end - range.start) as usize;
+        vals.resize(len, Item::make_zero());
+        self.vals_reader.get_range(range.start, &mut vals[..]);
+    }
+
+    /// Returns the array of values associated to the given `doc`.
+    #[inline]
+    pub fn get_vals(&self, doc: DocId, vals: &mut Vec<Item>) {
         let range = self.range(doc);
-        self.vals_reader.get_range(range)
+        self.get_vals_for_range(range, vals);
     }
 
     /// Returns the minimum value for this fast field.

src/fastfield/reader.rs

@@ -1,132 +1,65 @@
-use std::collections::HashMap;
 use std::marker::PhantomData;
-use std::path::Path;
+use std::sync::Arc;
 
 use common::BinarySerializable;
-use fastfield_codecs::bitpacked::{BitpackedCodec, BitpackedReader};
-use fastfield_codecs::blockwise_linear::{BlockwiseLinearCodec, BlockwiseLinearReader};
-use fastfield_codecs::linear::{LinearCodec, LinearReader};
-use fastfield_codecs::{Column, FastFieldCodec, FastFieldCodecType};
+use fastfield_codecs::bitpacked::BitpackedCodec;
+use fastfield_codecs::blockwise_linear::BlockwiseLinearCodec;
+use fastfield_codecs::linear::LinearCodec;
+use fastfield_codecs::{monotonic_map_column, Column, FastFieldCodec, FastFieldCodecType};
 
 use super::gcd::open_gcd_from_bytes;
 use super::FastValue;
-use crate::directory::{CompositeFile, Directory, FileSlice, OwnedBytes, RamDirectory, WritePtr};
+use crate::directory::OwnedBytes;
 use crate::error::DataCorruption;
-use crate::fastfield::{CompositeFastFieldSerializer, FastFieldsWriter, GCDReader};
-use crate::schema::{Schema, FAST};
 
-#[derive(Clone)]
-/// DynamicFastFieldReader wraps different readers to access
-/// the various encoded fastfield data
-pub enum DynamicFastFieldReader<Item: FastValue> {
-    /// Bitpacked compressed fastfield data.
-    Bitpacked(FastFieldReaderCodecWrapper<Item, BitpackedReader>),
-    /// Linear interpolated values + bitpacked
-    Linear(FastFieldReaderCodecWrapper<Item, LinearReader>),
-    /// Blockwise linear interpolated values + bitpacked
-    BlockwiseLinear(FastFieldReaderCodecWrapper<Item, BlockwiseLinearReader>),
-
-    /// GCD and Bitpacked compressed fastfield data.
-    BitpackedGCD(FastFieldReaderCodecWrapper<Item, GCDReader<BitpackedReader>>),
-    /// GCD and Linear interpolated values + bitpacked
-    LinearGCD(FastFieldReaderCodecWrapper<Item, GCDReader<LinearReader>>),
-    /// GCD and Blockwise linear interpolated values + bitpacked
-    BlockwiseLinearGCD(FastFieldReaderCodecWrapper<Item, GCDReader<BlockwiseLinearReader>>),
+fn open_codec_from_bytes<C: FastFieldCodec, Item: FastValue>(
+    bytes: OwnedBytes,
+) -> crate::Result<Arc<dyn Column<Item>>> {
+    let reader = C::open_from_bytes(bytes)?;
+    Ok(Arc::new(monotonic_map_column(reader, Item::from_u64)))
 }
 
-impl<Item: FastValue> DynamicFastFieldReader<Item> {
-    /// Returns correct the reader wrapped in the `DynamicFastFieldReader` enum for the data.
-    pub fn open_from_id(
-        mut bytes: OwnedBytes,
-        codec_type: FastFieldCodecType,
-    ) -> crate::Result<DynamicFastFieldReader<Item>> {
-        let reader =
+fn open_codec_with_gcd<C: FastFieldCodec, Item: FastValue>(
+    bytes: OwnedBytes,
+) -> crate::Result<Arc<dyn Column<Item>>> {
+    let reader = open_gcd_from_bytes::<C>(bytes)?;
+    Ok(Arc::new(monotonic_map_column(reader, Item::from_u64)))
+}
+
+/// Returns correct the reader wrapped in the `DynamicFastFieldReader` enum for the data.
+fn open_from_id<Item: FastValue>(
+    mut bytes: OwnedBytes,
+    codec_type: FastFieldCodecType,
+) -> crate::Result<Arc<dyn Column<Item>>> {
+    match codec_type {
+        FastFieldCodecType::Bitpacked => open_codec_from_bytes::<BitpackedCodec, _>(bytes),
+        FastFieldCodecType::Linear => open_codec_from_bytes::<LinearCodec, _>(bytes),
+        FastFieldCodecType::BlockwiseLinear => {
+            open_codec_from_bytes::<BlockwiseLinearCodec, _>(bytes)
+        }
+        FastFieldCodecType::Gcd => {
+            let codec_type = FastFieldCodecType::deserialize(&mut bytes)?;
             match codec_type {
-                FastFieldCodecType::Bitpacked => DynamicFastFieldReader::Bitpacked(
-                    BitpackedCodec::open_from_bytes(bytes)?.into(),
-                ),
-                FastFieldCodecType::Linear => {
-                    DynamicFastFieldReader::Linear(LinearCodec::open_from_bytes(bytes)?.into())
+                FastFieldCodecType::Bitpacked => open_codec_with_gcd::<BitpackedCodec, _>(bytes),
+                FastFieldCodecType::Linear => open_codec_with_gcd::<LinearCodec, _>(bytes),
+                FastFieldCodecType::BlockwiseLinear => {
+                    open_codec_with_gcd::<BlockwiseLinearCodec, _>(bytes)
                 }
-                FastFieldCodecType::BlockwiseLinear => DynamicFastFieldReader::BlockwiseLinear(
-                    BlockwiseLinearCodec::open_from_bytes(bytes)?.into(),
-                ),
-                FastFieldCodecType::Gcd => {
-                    let codec_type = FastFieldCodecType::deserialize(&mut bytes)?;
-                    match codec_type {
-                        FastFieldCodecType::Bitpacked => DynamicFastFieldReader::BitpackedGCD(
-                            open_gcd_from_bytes::<BitpackedCodec>(bytes)?.into(),
-                        ),
-                        FastFieldCodecType::Linear => DynamicFastFieldReader::LinearGCD(
-                            open_gcd_from_bytes::<LinearCodec>(bytes)?.into(),
-                        ),
-                        FastFieldCodecType::BlockwiseLinear => {
-                            DynamicFastFieldReader::BlockwiseLinearGCD(
-                                open_gcd_from_bytes::<BlockwiseLinearCodec>(bytes)?.into(),
-                            )
-                        }
-                        FastFieldCodecType::Gcd => return Err(DataCorruption::comment_only(
-                            "Gcd codec wrapped into another gcd codec. This combination is not \
-                             allowed.",
-                        )
-                        .into()),
-                    }
-                }
-            };
-        Ok(reader)
-    }
-
-    /// Returns correct the reader wrapped in the `DynamicFastFieldReader` enum for the data.
-    pub fn open(file: FileSlice) -> crate::Result<DynamicFastFieldReader<Item>> {
-        let mut bytes = file.read_bytes()?;
-        let codec_type = FastFieldCodecType::deserialize(&mut bytes)?;
-        Self::open_from_id(bytes, codec_type)
+                FastFieldCodecType::Gcd => Err(DataCorruption::comment_only(
+                    "Gcd codec wrapped into another gcd codec. This combination is not allowed.",
+                )
+                .into()),
+            }
+        }
     }
 }
 
-impl<Item: FastValue> Column<Item> for DynamicFastFieldReader<Item> {
-    #[inline]
-    fn get_val(&self, idx: u64) -> Item {
-        match self {
-            Self::Bitpacked(reader) => reader.get_val(idx),
-            Self::Linear(reader) => reader.get_val(idx),
-            Self::BlockwiseLinear(reader) => reader.get_val(idx),
-            Self::BitpackedGCD(reader) => reader.get_val(idx),
-            Self::LinearGCD(reader) => reader.get_val(idx),
-            Self::BlockwiseLinearGCD(reader) => reader.get_val(idx),
-        }
-    }
-    fn min_value(&self) -> Item {
-        match self {
-            Self::Bitpacked(reader) => reader.min_value(),
-            Self::Linear(reader) => reader.min_value(),
-            Self::BlockwiseLinear(reader) => reader.min_value(),
-            Self::BitpackedGCD(reader) => reader.min_value(),
-            Self::LinearGCD(reader) => reader.min_value(),
-            Self::BlockwiseLinearGCD(reader) => reader.min_value(),
-        }
-    }
-    fn max_value(&self) -> Item {
-        match self {
-            Self::Bitpacked(reader) => reader.max_value(),
-            Self::Linear(reader) => reader.max_value(),
-            Self::BlockwiseLinear(reader) => reader.max_value(),
-            Self::BitpackedGCD(reader) => reader.max_value(),
-            Self::LinearGCD(reader) => reader.max_value(),
-            Self::BlockwiseLinearGCD(reader) => reader.max_value(),
-        }
-    }
-
-    fn num_vals(&self) -> u64 {
-        match self {
-            Self::Bitpacked(reader) => reader.num_vals(),
-            Self::Linear(reader) => reader.num_vals(),
-            Self::BlockwiseLinear(reader) => reader.num_vals(),
-            Self::BitpackedGCD(reader) => reader.num_vals(),
-            Self::LinearGCD(reader) => reader.num_vals(),
-            Self::BlockwiseLinearGCD(reader) => reader.num_vals(),
-        }
-    }
+/// Returns correct the reader wrapped in the `DynamicFastFieldReader` enum for the data.
+pub fn open_fast_field<Item: FastValue>(
+    mut bytes: OwnedBytes,
+) -> crate::Result<Arc<dyn Column<Item>>> {
+    let codec_type = FastFieldCodecType::deserialize(&mut bytes)?;
+    open_from_id(bytes, codec_type)
 }
 
 /// Wrapper for accessing a fastfield.
@@ -155,6 +88,24 @@ impl<Item: FastValue, D: Column> FastFieldReaderCodecWrapper<Item, D> {
         let data = self.reader.get_val(idx);
         Item::from_u64(data)
     }
+
+    /// Internally `multivalued` also use SingleValue Fast fields.
+    /// It works as follows... A first column contains the list of start index
+    /// for each document, a second column contains the actual values.
+    ///
+    /// The values associated to a given doc, are then
+    ///  `second_column[first_column.get(doc)..first_column.get(doc+1)]`.
+    ///
+    /// Which means single value fast field reader can be indexed internally with
+    /// something different from a `DocId`. For this use case, we want to use `u64`
+    /// values.
+    ///
+    /// See `get_range` for an actual documentation about this method.
+    pub(crate) fn get_range_u64(&self, start: u64, output: &mut [Item]) {
+        for (i, out) in output.iter_mut().enumerate() {
+            *out = self.get_u64(start + (i as u64));
+        }
+    }
 }
 
 impl<Item: FastValue, C: Column + Clone> Column<Item> for FastFieldReaderCodecWrapper<Item, C> {
@@ -170,6 +121,23 @@ impl<Item: FastValue, C: Column + Clone> Column<Item> for FastFieldReaderCodecWr
         self.get_u64(idx)
     }
 
+    /// Fills an output buffer with the fast field values
+    /// associated with the `DocId` going from
+    /// `start` to `start + output.len()`.
+    ///
+    /// Regardless of the type of `Item`, this method works
+    /// - transmuting the output array
+    /// - extracting the `Item`s as if they were `u64`
+    /// - possibly converting the `u64` value to the right type.
+    ///
+    /// # Panics
+    ///
+    /// May panic if `start + output.len()` is greater than
+    /// the segment's `maxdoc`.
+    fn get_range(&self, start: u64, output: &mut [Item]) {
+        self.get_range_u64(start, output);
+    }
+
     /// Returns the minimum value for this fast field.
     ///
     /// The max value does not take in account of possible
@@ -192,40 +160,3 @@ impl<Item: FastValue, C: Column + Clone> Column<Item> for FastFieldReaderCodecWr
         self.reader.num_vals()
     }
 }
-
-impl<Item: FastValue> From<Vec<Item>> for DynamicFastFieldReader<Item> {
-    fn from(vals: Vec<Item>) -> DynamicFastFieldReader<Item> {
-        let mut schema_builder = Schema::builder();
-        let field = schema_builder.add_u64_field("field", FAST);
-        let schema = schema_builder.build();
-        let path = Path::new("__dummy__");
-        let directory: RamDirectory = RamDirectory::create();
-        {
-            let write: WritePtr = directory
-                .open_write(path)
-                .expect("With a RamDirectory, this should never fail.");
-            let mut serializer = CompositeFastFieldSerializer::from_write(write)
-                .expect("With a RamDirectory, this should never fail.");
-            let mut fast_field_writers = FastFieldsWriter::from_schema(&schema);
-            {
-                let fast_field_writer = fast_field_writers
-                    .get_field_writer_mut(field)
-                    .expect("With a RamDirectory, this should never fail.");
-                for val in vals {
-                    fast_field_writer.add_val(val.to_u64());
-                }
-            }
-            fast_field_writers
-                .serialize(&mut serializer, &HashMap::new(), None)
-                .unwrap();
-            serializer.close().unwrap();
-        }
-
-        let file = directory.open_read(path).expect("Failed to open the file");
-        let composite_file = CompositeFile::open(&file).expect("Failed to read the composite file");
-        let field_file = composite_file
-            .open_read(field)
-            .expect("File component not found");
-        DynamicFastFieldReader::open(field_file).unwrap()
-    }
-}

src/fastfield/readers.rs

@@ -1,5 +1,9 @@
-use super::reader::DynamicFastFieldReader;
+use std::sync::Arc;
+
+use fastfield_codecs::Column;
+
 use crate::directory::{CompositeFile, FileSlice};
+use crate::fastfield::reader::open_fast_field;
 use crate::fastfield::{
     BytesFastFieldReader, FastFieldNotAvailableError, FastValue, MultiValuedFastFieldReader,
 };
@@ -109,14 +113,16 @@ impl FastFieldReaders {
         &self,
         field: Field,
         index: usize,
-    ) -> crate::Result<DynamicFastFieldReader<TFastValue>> {
+    ) -> crate::Result<Arc<dyn Column<TFastValue>>> {
         let fast_field_slice = self.fast_field_data(field, index)?;
-        DynamicFastFieldReader::open(fast_field_slice)
+        let bytes = fast_field_slice.read_bytes()?;
+        open_fast_field(bytes)
     }
+
     pub(crate) fn typed_fast_field_reader<TFastValue: FastValue>(
         &self,
         field: Field,
-    ) -> crate::Result<DynamicFastFieldReader<TFastValue>> {
+    ) -> crate::Result<Arc<dyn Column<TFastValue>>> {
         self.typed_fast_field_reader_with_idx(field, 0)
     }
 
@@ -132,7 +138,7 @@ impl FastFieldReaders {
     /// Returns the `u64` fast field reader reader associated to `field`.
     ///
     /// If `field` is not a u64 fast field, this method returns an Error.
-    pub fn u64(&self, field: Field) -> crate::Result<DynamicFastFieldReader<u64>> {
+    pub fn u64(&self, field: Field) -> crate::Result<Arc<dyn Column<u64>>> {
         self.check_type(field, FastType::U64, Cardinality::SingleValue)?;
         self.typed_fast_field_reader(field)
     }
@@ -142,14 +148,14 @@ impl FastFieldReaders {
     ///
     /// If not, the fastfield reader will returns the u64-value associated to the original
     /// FastValue.
-    pub fn u64_lenient(&self, field: Field) -> crate::Result<DynamicFastFieldReader<u64>> {
+    pub fn u64_lenient(&self, field: Field) -> crate::Result<Arc<dyn Column<u64>>> {
         self.typed_fast_field_reader(field)
     }
 
     /// Returns the `i64` fast field reader reader associated to `field`.
     ///
     /// If `field` is not a i64 fast field, this method returns an Error.
-    pub fn i64(&self, field: Field) -> crate::Result<DynamicFastFieldReader<i64>> {
+    pub fn i64(&self, field: Field) -> crate::Result<Arc<dyn Column<i64>>> {
         self.check_type(field, FastType::I64, Cardinality::SingleValue)?;
         self.typed_fast_field_reader(field)
     }
@@ -157,7 +163,7 @@ impl FastFieldReaders {
     /// Returns the `date` fast field reader reader associated to `field`.
     ///
     /// If `field` is not a date fast field, this method returns an Error.
-    pub fn date(&self, field: Field) -> crate::Result<DynamicFastFieldReader<DateTime>> {
+    pub fn date(&self, field: Field) -> crate::Result<Arc<dyn Column<DateTime>>> {
         self.check_type(field, FastType::Date, Cardinality::SingleValue)?;
         self.typed_fast_field_reader(field)
     }
@@ -165,7 +171,7 @@ impl FastFieldReaders {
     /// Returns the `f64` fast field reader reader associated to `field`.
     ///
     /// If `field` is not a f64 fast field, this method returns an Error.
-    pub fn f64(&self, field: Field) -> crate::Result<DynamicFastFieldReader<f64>> {
+    pub fn f64(&self, field: Field) -> crate::Result<Arc<dyn Column<f64>>> {
         self.check_type(field, FastType::F64, Cardinality::SingleValue)?;
         self.typed_fast_field_reader(field)
     }
@@ -173,7 +179,7 @@ impl FastFieldReaders {
     /// Returns the `bool` fast field reader reader associated to `field`.
     ///
     /// If `field` is not a bool fast field, this method returns an Error.
-    pub fn bool(&self, field: Field) -> crate::Result<DynamicFastFieldReader<bool>> {
+    pub fn bool(&self, field: Field) -> crate::Result<Arc<dyn Column<bool>>> {
         self.check_type(field, FastType::Bool, Cardinality::SingleValue)?;
         self.typed_fast_field_reader(field)
     }
@@ -241,7 +247,8 @@ impl FastFieldReaders {
                 )));
             }
             let fast_field_idx_file = self.fast_field_data(field, 0)?;
-            let idx_reader = DynamicFastFieldReader::open(fast_field_idx_file)?;
+            let fast_field_idx_bytes = fast_field_idx_file.read_bytes()?;
+            let idx_reader = open_fast_field(fast_field_idx_bytes)?;
             let data = self.fast_field_data(field, 1)?;
             BytesFastFieldReader::open(idx_reader, data)
         } else {

src/fastfield/serializer/mod.rs

@@ -6,7 +6,7 @@ use fastdivide::DividerU64;
 pub use fastfield_codecs::bitpacked::{BitpackedCodec, BitpackedSerializerLegacy};
 use fastfield_codecs::blockwise_linear::BlockwiseLinearCodec;
 use fastfield_codecs::linear::LinearCodec;
-use fastfield_codecs::FastFieldCodecType;
+use fastfield_codecs::{monotonic_map_column, FastFieldCodecType};
 pub use fastfield_codecs::{Column, FastFieldCodec, FastFieldStats};
 
 use super::{find_gcd, ALL_CODECS, GCD_DEFAULT};
@@ -136,56 +136,22 @@ impl CompositeFastFieldSerializer {
         }
 
         Self::write_header(field_write, FastFieldCodecType::Gcd)?;
-        struct GCDWrappedFFAccess<T: Column> {
-            fastfield_accessor: T,
-            base_value: u64,
-            max_value: u64,
-            num_vals: u64,
-            gcd: DividerU64,
-        }
 
-        impl<T: Column> Column for GCDWrappedFFAccess<T> {
-            fn get_val(&self, position: u64) -> u64 {
-                self.gcd
-                    .divide(self.fastfield_accessor.get_val(position) - self.base_value)
-            }
-            fn iter(&self) -> Box<dyn Iterator<Item = u64> + '_> {
-                Box::new(
-                    self.fastfield_accessor
-                        .iter()
-                        .map(|val| self.gcd.divide(val - self.base_value)),
-                )
-            }
-            fn min_value(&self) -> u64 {
-                0
-            }
-
-            fn max_value(&self) -> u64 {
-                self.max_value
-            }
-
-            fn num_vals(&self) -> u64 {
-                self.num_vals
-            }
-        }
-
-        let num_vals = fastfield_accessor.num_vals();
         let base_value = fastfield_accessor.min_value();
-        let max_value = (fastfield_accessor.max_value() - fastfield_accessor.min_value()) / gcd;
 
-        let fastfield_accessor = GCDWrappedFFAccess {
-            fastfield_accessor,
-            base_value,
-            max_value,
-            num_vals,
-            gcd: DividerU64::divide_by(gcd),
-        };
+        let gcd_divider = DividerU64::divide_by(gcd);
+
+        let divided_fastfield_accessor = monotonic_map_column(fastfield_accessor, |val: u64| {
+            gcd_divider.divide(val - base_value)
+        });
+
+        let num_vals = divided_fastfield_accessor.num_vals();
 
         Self::create_auto_detect_u64_fast_field_with_idx_gcd(
             self.codec_enable_checker.clone(),
             field,
             field_write,
-            fastfield_accessor,
+            divided_fastfield_accessor,
         )?;
         write_gcd_header(field_write, base_value, gcd, num_vals)?;
         Ok(())

src/indexer/doc_id_mapping.rs

@@ -143,8 +143,6 @@ pub(crate) fn get_doc_id_mapping_from_field(
 
 #[cfg(test)]
 mod tests_indexsorting {
-    use fastfield_codecs::Column;
-
     use crate::collector::TopDocs;
     use crate::indexer::doc_id_mapping::DocIdMapping;
     use crate::query::QueryParser;
@@ -471,13 +469,15 @@ mod tests_indexsorting {
 
         let multi_numbers = index.schema().get_field("multi_numbers").unwrap();
         let multifield = fast_fields.u64s(multi_numbers).unwrap();
-        let vals = multifield.get_vals(0u32).collect::<Vec<_>>();
+        let mut vals = vec![];
+        multifield.get_vals(0u32, &mut vals);
         assert_eq!(vals, &[] as &[u64]);
-
-        let vals = multifield.get_vals(1u32).collect::<Vec<_>>();
+        let mut vals = vec![];
+        multifield.get_vals(1u32, &mut vals);
         assert_eq!(vals, &[5, 6]);
 
-        let vals = multifield.get_vals(2u32).collect::<Vec<_>>();
+        let mut vals = vec![];
+        multifield.get_vals(2u32, &mut vals);
         assert_eq!(vals, &[3]);
         Ok(())
     }

src/indexer/index_writer.rs

@@ -174,9 +174,7 @@ fn index_documents(
     segment_updater: &mut SegmentUpdater,
     mut delete_cursor: DeleteCursor,
 ) -> crate::Result<()> {
-    let schema = segment.schema();
-
-    let mut segment_writer = SegmentWriter::for_segment(memory_budget, segment.clone(), schema)?;
+    let mut segment_writer = SegmentWriter::for_segment(memory_budget, segment.clone())?;
     for document_group in grouped_document_iterator {
         for doc in document_group {
             segment_writer.add_document(doc)?;
@@ -777,7 +775,6 @@ impl Drop for IndexWriter {
 mod tests {
     use std::collections::{HashMap, HashSet};
 
-    use fastfield_codecs::Column;
     use proptest::prelude::*;
     use proptest::prop_oneof;
     use proptest::strategy::Strategy;
@@ -1535,11 +1532,13 @@ mod tests {
             let ff_reader = segment_reader.fast_fields().u64s(multi_numbers).unwrap();
             let bool_ff_reader = segment_reader.fast_fields().bools(multi_bools).unwrap();
             for doc in segment_reader.doc_ids_alive() {
-                let vals = ff_reader.get_vals(doc).collect::<Vec<_>>();
+                let mut vals = vec![];
+                ff_reader.get_vals(doc, &mut vals);
                 assert_eq!(vals.len(), 2);
                 assert_eq!(vals[0], vals[1]);
 
-                let bool_vals = bool_ff_reader.get_vals(doc).collect::<Vec<_>>();
+                let mut bool_vals = vec![];
+                bool_ff_reader.get_vals(doc, &mut bool_vals);
                 assert_eq!(bool_vals.len(), 2);
                 assert_ne!(bool_vals[0], bool_vals[1]);
 

src/indexer/merger.rs

@@ -9,8 +9,8 @@ use crate::core::{Segment, SegmentReader};
 use crate::docset::{DocSet, TERMINATED};
 use crate::error::DataCorruption;
 use crate::fastfield::{
-    AliveBitSet, Column, CompositeFastFieldSerializer, DynamicFastFieldReader, FastFieldStats,
-    MultiValueLength, MultiValuedFastFieldReader,
+    AliveBitSet, Column, CompositeFastFieldSerializer, FastFieldStats, MultiValueLength,
+    MultiValuedFastFieldReader,
 };
 use crate::fieldnorm::{FieldNormReader, FieldNormReaders, FieldNormsSerializer, FieldNormsWriter};
 use crate::indexer::doc_id_mapping::{expect_field_id_for_sort_field, SegmentDocIdMapping};
@@ -87,7 +87,7 @@ pub struct IndexMerger {
 }
 
 fn compute_min_max_val(
-    u64_reader: &impl Column<u64>,
+    u64_reader: &dyn Column<u64>,
     segment_reader: &SegmentReader,
 ) -> Option<(u64, u64)> {
     if segment_reader.max_doc() == 0 {
@@ -133,7 +133,7 @@ impl TermOrdinalMapping {
     fn max_term_ord(&self) -> TermOrdinal {
         self.per_segment_new_term_ordinals
             .iter()
-            .flat_map(|term_ordinals| term_ordinals.iter().max())
+            .flat_map(|term_ordinals| term_ordinals.iter().max().cloned())
             .max()
             .unwrap_or_default()
     }
@@ -341,12 +341,12 @@ impl IndexMerger {
             .readers
             .iter()
             .filter_map(|reader| {
-                let u64_reader: DynamicFastFieldReader<u64> =
+                let u64_reader: Arc<dyn Column<u64>> =
                     reader.fast_fields().typed_fast_field_reader(field).expect(
                         "Failed to find a reader for single fast field. This is a tantivy bug and \
                          it should never happen.",
                     );
-                compute_min_max_val(&u64_reader, reader)
+                compute_min_max_val(&*u64_reader, reader)
             })
             .reduce(|a, b| (a.0.min(b.0), a.1.max(b.1)))
             .expect("Unexpected error, empty readers in IndexMerger");
@@ -355,7 +355,7 @@ impl IndexMerger {
             .readers
             .iter()
             .map(|reader| {
-                let u64_reader: DynamicFastFieldReader<u64> =
+                let u64_reader: Arc<dyn Column<u64>> =
                     reader.fast_fields().typed_fast_field_reader(field).expect(
                         "Failed to find a reader for single fast field. This is a tantivy bug and \
                          it should never happen.",
@@ -372,7 +372,7 @@ impl IndexMerger {
         #[derive(Clone)]
         struct SortedDocIdFieldAccessProvider<'a> {
             doc_id_mapping: &'a SegmentDocIdMapping,
-            fast_field_readers: &'a Vec<DynamicFastFieldReader<u64>>,
+            fast_field_readers: &'a Vec<Arc<dyn Column<u64>>>,
             stats: FastFieldStats,
         }
         impl<'a> Column for SortedDocIdFieldAccessProvider<'a> {
@@ -443,7 +443,7 @@ impl IndexMerger {
     pub(crate) fn get_sort_field_accessor(
         reader: &SegmentReader,
         sort_by_field: &IndexSortByField,
-    ) -> crate::Result<impl Column> {
+    ) -> crate::Result<Arc<dyn Column>> {
         let field_id = expect_field_id_for_sort_field(reader.schema(), sort_by_field)?; // for now expect fastfield, but not strictly required
         let value_accessor = reader.fast_fields().u64_lenient(field_id)?;
         Ok(value_accessor)
@@ -452,7 +452,7 @@ impl IndexMerger {
     pub(crate) fn get_reader_with_sort_field_accessor(
         &self,
         sort_by_field: &IndexSortByField,
-    ) -> crate::Result<Vec<(SegmentOrdinal, impl Column)>> {
+    ) -> crate::Result<Vec<(SegmentOrdinal, Arc<dyn Column>)>> {
         let reader_ordinal_and_field_accessors = self
             .readers
             .iter()
@@ -578,7 +578,6 @@ impl IndexMerger {
             stats: FastFieldStats,
         }
         impl<'a> Column for FieldIndexAccessProvider<'a> {
-            #[inline]
             fn get_val(&self, doc: u64) -> u64 {
                 self.offsets[doc as usize]
             }
@@ -619,7 +618,7 @@ impl IndexMerger {
             .map(|reader| {
                 let u64s_reader: MultiValuedFastFieldReader<u64> = reader
                     .fast_fields()
-                    .typed_fast_field_multi_reader(field)
+                    .typed_fast_field_multi_reader::<u64>(field)
                     .expect(
                         "Failed to find index for multivalued field. This is a bug in tantivy, \
                          please report.",
@@ -669,13 +668,15 @@ impl IndexMerger {
         {
             let mut serialize_vals =
                 fast_field_serializer.new_u64_fast_field_with_idx(field, 0u64, max_term_ord, 1)?;
+            let mut vals: Vec<u64> = Vec::with_capacity(100);
 
             for old_doc_addr in doc_id_mapping.iter_old_doc_addrs() {
                 let term_ordinal_mapping: &[TermOrdinal] =
                     term_ordinal_mappings.get_segment(old_doc_addr.segment_ord as usize);
 
                 let ff_reader = &fast_field_reader[old_doc_addr.segment_ord as usize];
-                for prev_term_ord in ff_reader.get_vals(old_doc_addr.doc_id) {
+                ff_reader.get_vals(old_doc_addr.doc_id, &mut vals);
+                for &prev_term_ord in &vals {
                     let new_term_ord = term_ordinal_mapping[prev_term_ord as usize];
                     serialize_vals.add_val(new_term_ord)?;
                 }
@@ -728,6 +729,8 @@ impl IndexMerger {
         let mut max_value = u64::MIN;
         let mut num_vals = 0;
 
+        let mut vals = Vec::with_capacity(100);
+
         let mut ff_readers = Vec::new();
 
         // Our values are bitpacked and we need to know what should be
@@ -739,17 +742,18 @@ impl IndexMerger {
         for reader in &self.readers {
             let ff_reader: MultiValuedFastFieldReader<u64> = reader
                 .fast_fields()
-                .typed_fast_field_multi_reader(field)
+                .typed_fast_field_multi_reader::<u64>(field)
                 .expect(
                     "Failed to find multivalued fast field reader. This is a bug in tantivy. \
                      Please report.",
                 );
             for doc in reader.doc_ids_alive() {
-                for val in ff_reader.get_vals(doc) {
+                ff_reader.get_vals(doc, &mut vals);
+                for &val in &vals {
                     min_value = cmp::min(val, min_value);
                     max_value = cmp::max(val, max_value);
-                    num_vals += 1;
                 }
+                num_vals += vals.len();
             }
             ff_readers.push(ff_reader);
             // TODO optimize when no deletes
@@ -780,7 +784,7 @@ impl IndexMerger {
                 let new_doc_id: DocId =
                     self.offsets
                         .iter()
-                        .position(|offset| offset > pos)
+                        .position(|&offset| offset > pos)
                         .expect("pos is out of bounds") as DocId
                         - 1u32;
 
@@ -792,10 +796,11 @@ impl IndexMerger {
                 let num_vals = self.fast_field_readers[old_doc_addr.segment_ord as usize]
                     .get_len(old_doc_addr.doc_id);
                 assert!(num_vals >= pos_in_values);
+                let mut vals = Vec::new();
                 self.fast_field_readers[old_doc_addr.segment_ord as usize]
-                    .get_vals(old_doc_addr.doc_id)
-                    .nth(pos_in_values as usize)
-                    .expect("computation error in SortedDocIdMultiValueAccessProvider")
+                    .get_vals(old_doc_addr.doc_id, &mut vals);
+
+                vals[pos_in_values as usize]
             }
 
             fn iter(&self) -> Box<dyn Iterator<Item = u64> + '_> {
@@ -805,7 +810,9 @@ impl IndexMerger {
                         .flat_map(|old_doc_addr| {
                             let ff_reader =
                                 &self.fast_field_readers[old_doc_addr.segment_ord as usize];
-                            ff_reader.get_vals(old_doc_addr.doc_id)
+                            let mut vals = Vec::new();
+                            ff_reader.get_vals(old_doc_addr.doc_id, &mut vals);
+                            vals.into_iter()
                         }),
                 )
             }
@@ -1192,7 +1199,6 @@ impl IndexMerger {
 #[cfg(test)]
 mod tests {
     use byteorder::{BigEndian, ReadBytesExt};
-    use fastfield_codecs::Column;
     use schema::FAST;
 
     use crate::collector::tests::{
@@ -1968,32 +1974,49 @@ mod tests {
         }
         let reader = index.reader()?;
         let searcher = reader.searcher();
+        let mut vals: Vec<u64> = Vec::new();
 
         {
             let segment = searcher.segment_reader(0u32);
             let ff_reader = segment.fast_fields().u64s(int_field).unwrap();
 
-            assert_eq!(&ff_reader.get_vals(0).collect::<Vec<_>>(), &[1, 2]);
-            assert_eq!(&ff_reader.get_vals(1).collect::<Vec<_>>(), &[1, 2, 3]);
-            assert_eq!(&ff_reader.get_vals(2).collect::<Vec<_>>(), &[4, 5]);
-            assert_eq!(&ff_reader.get_vals(3).collect::<Vec<_>>(), &[1, 2]);
-            assert_eq!(&ff_reader.get_vals(4).collect::<Vec<_>>(), &[1, 5]);
-            assert_eq!(&ff_reader.get_vals(5).collect::<Vec<_>>(), &[3]);
-            assert_eq!(&ff_reader.get_vals(6).collect::<Vec<_>>(), &[17]);
+            ff_reader.get_vals(0, &mut vals);
+            assert_eq!(&vals, &[1, 2]);
+
+            ff_reader.get_vals(1, &mut vals);
+            assert_eq!(&vals, &[1, 2, 3]);
+
+            ff_reader.get_vals(2, &mut vals);
+            assert_eq!(&vals, &[4, 5]);
+
+            ff_reader.get_vals(3, &mut vals);
+            assert_eq!(&vals, &[1, 2]);
+
+            ff_reader.get_vals(4, &mut vals);
+            assert_eq!(&vals, &[1, 5]);
+
+            ff_reader.get_vals(5, &mut vals);
+            assert_eq!(&vals, &[3]);
+
+            ff_reader.get_vals(6, &mut vals);
+            assert_eq!(&vals, &[17]);
         }
 
         {
             let segment = searcher.segment_reader(1u32);
             let ff_reader = segment.fast_fields().u64s(int_field).unwrap();
-            assert_eq!(&ff_reader.get_vals(0).collect::<Vec<_>>(), &[28, 27]);
+            ff_reader.get_vals(0, &mut vals);
+            assert_eq!(&vals, &[28, 27]);
 
-            assert_eq!(&ff_reader.get_vals(1).collect::<Vec<_>>(), &[1000]);
+            ff_reader.get_vals(1, &mut vals);
+            assert_eq!(&vals, &[1_000]);
         }
 
         {
             let segment = searcher.segment_reader(2u32);
             let ff_reader = segment.fast_fields().u64s(int_field).unwrap();
-            assert_eq!(&ff_reader.get_vals(0).collect::<Vec<_>>(), &[20]);
+            ff_reader.get_vals(0, &mut vals);
+            assert_eq!(&vals, &[20]);
         }
 
         // Merging the segments
@@ -2010,16 +2033,35 @@ mod tests {
             let segment = searcher.segment_reader(0u32);
             let ff_reader = segment.fast_fields().u64s(int_field).unwrap();
 
-            assert_eq!(&ff_reader.get_vals(0).collect::<Vec<_>>(), &[1, 2]);
-            assert_eq!(&ff_reader.get_vals(1).collect::<Vec<_>>(), &[1, 2, 3]);
-            assert_eq!(&ff_reader.get_vals(2).collect::<Vec<_>>(), &[4, 5]);
-            assert_eq!(&ff_reader.get_vals(3).collect::<Vec<_>>(), &[1, 2]);
-            assert_eq!(&ff_reader.get_vals(4).collect::<Vec<_>>(), &[1, 5]);
-            assert_eq!(&ff_reader.get_vals(5).collect::<Vec<_>>(), &[3]);
-            assert_eq!(&ff_reader.get_vals(6).collect::<Vec<_>>(), &[17]);
-            assert_eq!(&ff_reader.get_vals(7).collect::<Vec<_>>(), &[28, 27]);
-            assert_eq!(&ff_reader.get_vals(8).collect::<Vec<_>>(), &[1_000]);
-            assert_eq!(&ff_reader.get_vals(9).collect::<Vec<_>>(), &[20]);
+            ff_reader.get_vals(0, &mut vals);
+            assert_eq!(&vals, &[1, 2]);
+
+            ff_reader.get_vals(1, &mut vals);
+            assert_eq!(&vals, &[1, 2, 3]);
+
+            ff_reader.get_vals(2, &mut vals);
+            assert_eq!(&vals, &[4, 5]);
+
+            ff_reader.get_vals(3, &mut vals);
+            assert_eq!(&vals, &[1, 2]);
+
+            ff_reader.get_vals(4, &mut vals);
+            assert_eq!(&vals, &[1, 5]);
+
+            ff_reader.get_vals(5, &mut vals);
+            assert_eq!(&vals, &[3]);
+
+            ff_reader.get_vals(6, &mut vals);
+            assert_eq!(&vals, &[17]);
+
+            ff_reader.get_vals(7, &mut vals);
+            assert_eq!(&vals, &[28, 27]);
+
+            ff_reader.get_vals(8, &mut vals);
+            assert_eq!(&vals, &[1_000]);
+
+            ff_reader.get_vals(9, &mut vals);
+            assert_eq!(&vals, &[20]);
         }
         Ok(())
     }

src/indexer/merger_sorted_index_test.rs

@@ -1,7 +1,5 @@
 #[cfg(test)]
 mod tests {
-    use fastfield_codecs::Column;
-
     use crate::collector::TopDocs;
     use crate::core::Index;
     use crate::fastfield::{AliveBitSet, MultiValuedFastFieldReader};
@@ -383,7 +381,9 @@ mod tests {
         assert_eq!(fast_field.get_val(5), 1_000u64);
 
         let get_vals = |fast_field: &MultiValuedFastFieldReader<u64>, doc_id: u32| -> Vec<u64> {
-            fast_field.get_vals(doc_id).collect()
+            let mut vals = vec![];
+            fast_field.get_vals(doc_id, &mut vals);
+            vals
         };
         let fast_fields = segment_reader.fast_fields();
         let fast_field = fast_fields.u64s(multi_numbers).unwrap();
@@ -478,11 +478,12 @@ mod tests {
 #[cfg(all(test, feature = "unstable"))]
 mod bench_sorted_index_merge {
 
+    use std::sync::Arc;
+
     use fastfield_codecs::Column;
     use test::{self, Bencher};
 
     use crate::core::Index;
-    use crate::fastfield::DynamicFastFieldReader;
     use crate::indexer::merger::IndexMerger;
     use crate::schema::{Cardinality, NumericOptions, Schema};
     use crate::{IndexSettings, IndexSortByField, IndexWriter, Order};
@@ -534,7 +535,7 @@ mod bench_sorted_index_merge {
         b.iter(|| {
             let sorted_doc_ids = doc_id_mapping.iter_old_doc_addrs().map(|doc_addr| {
                 let reader = &merger.readers[doc_addr.segment_ord as usize];
-                let u64_reader: DynamicFastFieldReader<u64> =
+                let u64_reader: Arc<dyn Column<u64>> =
                     reader.fast_fields().typed_fast_field_reader(field).expect(
                         "Failed to find a reader for single fast field. This is a tantivy bug and \
                          it should never happen.",

src/indexer/segment_updater.rs

@@ -25,39 +25,10 @@ use crate::indexer::{
     DefaultMergePolicy, MergeCandidate, MergeOperation, MergePolicy, SegmentEntry,
     SegmentSerializer,
 };
-use crate::schema::Schema;
 use crate::{FutureResult, Opstamp};
 
 const NUM_MERGE_THREADS: usize = 4;
 
-/// Save the index meta file.
-/// This operation is atomic :
-/// Either
-///  - it fails, in which case an error is returned,
-/// and the `meta.json` remains untouched,
-/// - it succeeds, and `meta.json` is written
-/// and flushed.
-///
-/// This method is not part of tantivy's public API
-pub fn save_new_metas(
-    schema: Schema,
-    index_settings: IndexSettings,
-    directory: &dyn Directory,
-) -> crate::Result<()> {
-    save_metas(
-        &IndexMeta {
-            index_settings,
-            segments: Vec::new(),
-            schema,
-            opstamp: 0u64,
-            payload: None,
-        },
-        directory,
-    )?;
-    directory.sync_directory()?;
-    Ok(())
-}
-
 /// Save the index meta file.
 /// This operation is atomic:
 /// Either
@@ -67,7 +38,7 @@ pub fn save_new_metas(
 /// and flushed.
 ///
 /// This method is not part of tantivy's public API
-fn save_metas(metas: &IndexMeta, directory: &dyn Directory) -> crate::Result<()> {
+pub(crate) fn save_metas(metas: &IndexMeta, directory: &dyn Directory) -> crate::Result<()> {
     info!("save metas");
     let mut buffer = serde_json::to_vec_pretty(metas)?;
     // Just adding a new line at the end of the buffer.

src/indexer/segment_writer.rs

@@ -80,8 +80,8 @@ impl SegmentWriter {
     pub fn for_segment(
         memory_budget_in_bytes: usize,
         segment: Segment,
-        schema: Schema,
     ) -> crate::Result<SegmentWriter> {
+        let schema = segment.schema();
         let tokenizer_manager = segment.index().tokenizers().clone();
         let table_size = compute_initial_table_size(memory_budget_in_bytes)?;
         let segment_serializer = SegmentSerializer::for_segment(segment, false)?;

src/lib.rs

@@ -421,7 +421,6 @@ pub struct DocAddress {
 #[cfg(test)]
 pub mod tests {
     use common::{BinarySerializable, FixedSize};
-    use fastfield_codecs::Column;
     use rand::distributions::{Bernoulli, Uniform};
     use rand::rngs::StdRng;
     use rand::{Rng, SeedableRng};

src/postings/mod.rs

@@ -227,7 +227,7 @@ pub mod tests {
 
         {
             let mut segment_writer =
-                SegmentWriter::for_segment(3_000_000, segment.clone(), schema).unwrap();
+                SegmentWriter::for_segment(3_000_000, segment.clone()).unwrap();
             {
                 // checking that position works if the field has two values
                 let op = AddOperation {

src/snippet/mod.rs

@@ -639,7 +639,10 @@ Survey in 2016, 2017, and 2018."#;
     #[test]
     fn test_collapse_overlapped_ranges() {
         assert_eq!(&collapse_overlapped_ranges(&[0..1, 2..3,]), &[0..1, 2..3]);
-        assert_eq!(collapse_overlapped_ranges(&[0..1, 1..2,]), &[0..1, 1..2]);
+        assert_eq!(
+            collapse_overlapped_ranges(&vec![0..1, 1..2,]),
+            vec![0..1, 1..2]
+        );
         assert_eq!(collapse_overlapped_ranges(&[0..2, 1..2,]), vec![0..2]);
         assert_eq!(collapse_overlapped_ranges(&[0..2, 1..3,]), vec![0..3]);
         assert_eq!(collapse_overlapped_ranges(&[0..3, 1..2,]), vec![0..3]);

src/store/reader.rs

@@ -339,7 +339,7 @@ impl StoreReader {
     async fn read_block_async(&self, checkpoint: &Checkpoint) -> crate::AsyncIoResult<Block> {
         let cache_key = checkpoint.byte_range.start;
         if let Some(block) = self.cache.get_from_cache(checkpoint.byte_range.start) {
-            return Ok(block.clone());
+            return Ok(block);
         }
 
         let compressed_block = self

src/termdict/sstable_termdict/sstable/delta.rs

@@ -172,8 +172,7 @@ where TValueReader: value::ValueReader
     }
 
     pub fn suffix(&self) -> &[u8] {
-        &self
-            .block_reader
+        self.block_reader
             .buffer_from_to(self.suffix_start, self.suffix_end)
     }
 

src/termdict/sstable_termdict/sstable/sstable_index.rs

@@ -50,7 +50,7 @@ pub struct SSTableIndexBuilder {
 /// matches `left <= left' < right`.
 fn find_shorter_str_in_between(left: &mut Vec<u8>, right: &[u8]) {
     assert!(&left[..] < right);
-    let common_len = common_prefix_len(&left, right);
+    let common_len = common_prefix_len(left, right);
     if left.len() == common_len {
         return;
     }