Add support for multivalues

Author: fulmicoton

columnar/src/column/dictionary_encoded.rs

@@ -8,24 +8,35 @@ use crate::column::Column;
 use crate::RowId;
 
 /// Dictionary encoded column.
+///
+/// The column simply gives access to a regular u64-column that, in
+/// which the values are term-ordinals.
+///
+/// These ordinals are ids uniquely identify the bytes that are stored in
+/// the column. These ordinals are small, and sorted in the same order
+/// as the term_ord_column.
 #[derive(Clone)]
 pub struct BytesColumn {
     pub(crate) dictionary: Arc<Dictionary<VoidSSTable>>,
     pub(crate) term_ord_column: Column<u64>,
 }
 
 impl BytesColumn {
+    /// Fills the given `output` buffer with the term associated to the ordinal `ord`.
+    ///
     /// Returns `false` if the term does not exist (e.g. `term_ord` is greater or equal to the
     /// overll number of terms).
-    pub fn ord_to_bytes(&self, term_ord: u64, output: &mut Vec<u8>) -> io::Result<bool> {
-        self.dictionary.ord_to_term(term_ord, output)
+    pub fn ord_to_bytes(&self, ord: u64, output: &mut Vec<u8>) -> io::Result<bool> {
+        self.dictionary.ord_to_term(ord, output)
     }
 
+    /// Returns the number of rows in the column.
     pub fn num_rows(&self) -> RowId {
         self.term_ord_column.num_rows()
     }
 
-    pub fn term_ords(&self) -> &Column<u64> {
+    /// Returns the column of ordinals
+    pub fn ords(&self) -> &Column<u64> {
         &self.term_ord_column
     }
 }
@@ -40,6 +51,7 @@ impl From<BytesColumn> for StrColumn {
 }
 
 impl StrColumn {
+    /// Fills the buffer
     pub fn ord_to_str(&self, term_ord: u64, output: &mut String) -> io::Result<bool> {
         unsafe {
             let buf = output.as_mut_vec();
@@ -55,14 +67,6 @@ impl StrColumn {
         }
         Ok(true)
     }
-
-    pub fn num_rows(&self) -> RowId {
-        self.term_ord_column.num_rows()
-    }
-
-    pub fn ordinal_dictionary(&self) -> &Column<u64> {
-        &self.0.term_ord_column
-    }
 }
 
 impl Deref for StrColumn {

columnar/src/column/mod.rs

@@ -5,11 +5,11 @@ use std::ops::Deref;
 use std::sync::Arc;
 
 use common::BinarySerializable;
+pub use dictionary_encoded::{BytesColumn, StrColumn};
 pub use serialize::{
-    open_column_bytes, open_column_u128, open_column_u64, serialize_column_u128,
+    open_column_bytes, open_column_u128, open_column_u64, serialize_column_mappable_to_u128,
     serialize_column_u64,
 };
-pub use dictionary_encoded::{BytesColumn, StrColumn};
 
 use crate::column_index::ColumnIndex;
 use crate::column_values::ColumnValues;
@@ -21,23 +21,31 @@ pub struct Column<T> {
     pub values: Arc<dyn ColumnValues<T>>,
 }
 
-use crate::column_index::Set;
-
 impl<T: PartialOrd> Column<T> {
-    pub fn first(&self, row_id: RowId) -> Option<T> {
+    pub fn num_rows(&self) -> RowId {
         match &self.idx {
-            ColumnIndex::Full => Some(self.values.get_val(row_id)),
-            ColumnIndex::Optional(opt_idx) => {
-                let value_row_idx = opt_idx.rank_if_exists(row_id)?;
-                Some(self.values.get_val(value_row_idx))
-            }
-            ColumnIndex::Multivalued(_multivalued_index) => {
-                todo!();
+            ColumnIndex::Full => self.values.num_vals() as u32,
+            ColumnIndex::Optional(optional_index) => optional_index.num_rows(),
+            ColumnIndex::Multivalued(col_index) => {
+                // The multivalued index contains all value start row_id,
+                // and one extra value at the end with the overall number of rows.
+                col_index.num_vals() - 1
             }
         }
     }
 }
 
+impl<T: PartialOrd> Column<T> {
+    pub fn first(&self, row_id: RowId) -> Option<T> {
+        self.values(row_id).next()
+    }
+
+    pub fn values(&self, row_id: RowId) -> impl Iterator<Item = T> + '_ {
+        self.value_row_ids(row_id)
+            .map(|value_row_id: RowId| self.values.get_val(value_row_id))
+    }
+}
+
 impl<T> Deref for Column<T> {
     type Target = ColumnIndex<'static>;
 

columnar/src/column/serialize.rs

@@ -13,7 +13,7 @@ use crate::column_values::{
     ALL_CODEC_TYPES,
 };
 
-pub fn serialize_column_u128<
+pub fn serialize_column_mappable_to_u128<
     F: Fn() -> I,
     I: Iterator<Item = T>,
     T: MonotonicallyMappableToU128,

columnar/src/column_index/mod.rs

@@ -2,6 +2,7 @@ mod multivalued_index;
 mod optional_index;
 mod serialize;
 
+use std::ops::Range;
 use std::sync::Arc;
 
 pub use optional_index::{OptionalIndex, SerializableOptionalIndex, Set};
@@ -14,8 +15,12 @@ use crate::{Cardinality, RowId};
 pub enum ColumnIndex<'a> {
     Full,
     Optional(OptionalIndex),
-    // TODO remove the Arc<dyn> apart from serialization this is not
-    // dynamic at all.
+    // TODO Remove the static by fixing the codec if possible.
+    /// The column values enclosed contains for all row_id,
+    /// the value start_index.
+    ///
+    /// In addition, at index num_rows, an extra value is added
+    /// containing the overal number of values.
     Multivalued(Arc<dyn ColumnValues<RowId> + 'a>),
 }
 
@@ -28,13 +33,21 @@ impl<'a> ColumnIndex<'a> {
         }
     }
 
-    pub fn num_rows(&self) -> RowId {
+    pub fn value_row_ids(&self, row_id: RowId) -> Range<RowId> {
         match self {
-            ColumnIndex::Full => {
-                todo!()
+            ColumnIndex::Full => row_id..row_id + 1,
+            ColumnIndex::Optional(optional_index) => {
+                if let Some(val) = optional_index.rank_if_exists(row_id) {
+                    val..val + 1
+                } else {
+                    0..0
+                }
+            }
+            ColumnIndex::Multivalued(multivalued_index) => {
+                let start = multivalued_index.get_val(row_id);
+                let end = multivalued_index.get_val(row_id + 1);
+                start..end
             }
-            ColumnIndex::Optional(optional_index) => optional_index.num_rows(),
-            ColumnIndex::Multivalued(multivalued_index) => multivalued_index.num_vals() - 1,
         }
     }
 }

columnar/src/column_index/multivalued_index.rs

@@ -11,17 +11,19 @@ use crate::RowId;
 pub struct MultivaluedIndex(Arc<dyn ColumnValues<RowId>>);
 
 pub fn serialize_multivalued_index(
-    multivalued_index: MultivaluedIndex,
+    multivalued_index: &dyn ColumnValues<RowId>,
     output: &mut impl Write,
 ) -> io::Result<()> {
     crate::column_values::serialize_column_values(
-        &*multivalued_index.0,
+        &*multivalued_index,
         &[FastFieldCodecType::Bitpacked, FastFieldCodecType::Linear],
         output,
     )?;
     Ok(())
 }
 
 pub fn open_multivalued_index(bytes: OwnedBytes) -> io::Result<Arc<dyn ColumnValues<RowId>>> {
-    todo!();
+    let start_index_column: Arc<dyn ColumnValues<RowId>> =
+        crate::column_values::open_u64_mapped(bytes)?;
+    Ok(start_index_column)
 }

columnar/src/column_index/serialize.rs

@@ -3,17 +3,18 @@ use std::io::Write;
 
 use common::{CountingWriter, OwnedBytes};
 
-use crate::column_index::multivalued_index::{serialize_multivalued_index, MultivaluedIndex};
+use crate::column_index::multivalued_index::serialize_multivalued_index;
 use crate::column_index::optional_index::serialize_optional_index;
 use crate::column_index::{ColumnIndex, SerializableOptionalIndex};
-use crate::Cardinality;
+use crate::column_values::ColumnValues;
+use crate::{Cardinality, RowId};
 
 pub enum SerializableColumnIndex<'a> {
     Full,
     Optional(Box<dyn SerializableOptionalIndex<'a> + 'a>),
     // TODO remove the Arc<dyn> apart from serialization this is not
     // dynamic at all.
-    Multivalued(MultivaluedIndex),
+    Multivalued(Box<dyn ColumnValues<RowId> + 'a>),
 }
 
 impl<'a> SerializableColumnIndex<'a> {
@@ -39,7 +40,7 @@ pub fn serialize_column_index(
             serialize_optional_index(&*optional_index, &mut output)?
         }
         SerializableColumnIndex::Multivalued(multivalued_index) => {
-            serialize_multivalued_index(multivalued_index, &mut output)?
+            serialize_multivalued_index(&*multivalued_index, &mut output)?
         }
     }
     let column_index_num_bytes = output.written_bytes() as u32;

columnar/src/column_values/mod.rs

@@ -121,21 +121,6 @@ impl U128FastFieldCodecType {
     }
 }
 
-/// Returns the correct codec reader wrapped in the `Arc` for the data.
-// pub fn open_u128<Item: MonotonicallyMappableToU128>(
-//     bytes: OwnedBytes,
-// ) -> io::Result<Arc<dyn Column<Item>>> {
-//     todo!();
-//     // let (bytes, _format_version) = read_format_version(bytes)?;
-//     // let (mut bytes, _null_index_footer) = read_null_index_footer(bytes)?;
-//     // let header = U128Header::deserialize(&mut bytes)?;
-//     // assert_eq!(header.codec_type, U128FastFieldCodecType::CompactSpace);
-//     // let reader = CompactSpaceDecompressor::open(bytes)?;
-//     // let inverted: StrictlyMonotonicMappingInverter<StrictlyMonotonicMappingToInternal<Item>> =
-//     //     StrictlyMonotonicMappingToInternal::<Item>::new().into();
-//     // Ok(Arc::new(monotonic_map_column(reader, inverted)))
-// }
-
 /// Returns the correct codec reader wrapped in the `Arc` for the data.
 pub fn open_u128_mapped<T: MonotonicallyMappableToU128>(
     mut bytes: OwnedBytes,

columnar/src/columnar/merge.rs

@@ -15,10 +15,10 @@ pub enum MergeDocOrder {
     Complex(()),
 }
 
-pub fn merge(
-    columnar_readers: &[ColumnarReader],
+pub fn merge_columnar(
+    _columnar_readers: &[ColumnarReader],
     mapping: MergeDocOrder,
-    output: &mut impl io::Write,
+    _output: &mut impl io::Write,
 ) -> io::Result<()> {
     match mapping {
         MergeDocOrder::Stack => {

columnar/src/columnar/mod.rs

@@ -5,5 +5,6 @@ mod reader;
 mod writer;
 
 pub use column_type::ColumnType;
+pub use merge::{merge_columnar, MergeDocOrder};
 pub use reader::ColumnarReader;
 pub use writer::ColumnarWriter;

columnar/src/columnar/writer/mod.rs

@@ -22,7 +22,7 @@ use crate::columnar::writer::column_writers::{
 use crate::columnar::writer::value_index::{IndexBuilder, PreallocatedIndexBuilders};
 use crate::dictionary::{DictionaryBuilder, TermIdMapping, UnorderedId};
 use crate::value::{Coerce, NumericalType, NumericalValue};
-use crate::{column, Cardinality, RowId};
+use crate::{Cardinality, RowId};
 
 /// This is a set of buffers that are used to temporarily write the values into before passing them
 /// to the fast field codecs.
@@ -310,7 +310,7 @@ fn serialize_bytes_or_str_column(
             ColumnOperation::NewDoc(doc) => ColumnOperation::NewDoc(doc),
         }
     });
-    serialize_column(
+    serialize_column_mappable_to_u64(
         operation_iterator,
         cardinality,
         num_docs,
@@ -339,7 +339,7 @@ fn serialize_numerical_column(
     } = buffers;
     match numerical_type {
         NumericalType::I64 => {
-            serialize_column(
+            serialize_column_mappable_to_u64(
                 coerce_numerical_symbol::<i64>(op_iterator),
                 cardinality,
                 num_docs,
@@ -349,7 +349,7 @@ fn serialize_numerical_column(
             )?;
         }
         NumericalType::U64 => {
-            serialize_column(
+            serialize_column_mappable_to_u64(
                 coerce_numerical_symbol::<u64>(op_iterator),
                 cardinality,
                 num_docs,
@@ -359,7 +359,7 @@ fn serialize_numerical_column(
             )?;
         }
         NumericalType::F64 => {
-            serialize_column(
+            serialize_column_mappable_to_u64(
                 coerce_numerical_symbol::<f64>(op_iterator),
                 cardinality,
                 num_docs,
@@ -384,7 +384,7 @@ fn serialize_bool_column(
         bool_values,
         ..
     } = buffers;
-    serialize_column(
+    serialize_column_mappable_to_u64(
         column_operations_it,
         cardinality,
         num_docs,
@@ -451,12 +451,11 @@ where
         Cardinality::Multivalued => {
             let multivalued_index_builder = value_index_builders.borrow_multivalued_index_builder();
             consume_operation_iterator(op_iterator, multivalued_index_builder, values);
-            let _multivalued_index = multivalued_index_builder.finish(num_docs);
-            todo!();
-            // SerializableColumnIndex::Multivalued(Box::new(multivalued_index))
+            let multivalued_index = multivalued_index_builder.finish(num_docs);
+            SerializableColumnIndex::Multivalued(Box::new(multivalued_index))
         }
     };
-    crate::column::serialize_column_u128(
+    crate::column::serialize_column_mappable_to_u128(
         serializable_column_index,
         || values.iter().cloned(),
         values.len() as u32,
@@ -465,7 +464,7 @@ where
     Ok(())
 }
 
-fn serialize_column<
+fn serialize_column_mappable_to_u64<
     T: Copy + Default + std::fmt::Debug + Send + Sync + MonotonicallyMappableToU64 + PartialOrd,
 >(
     op_iterator: impl Iterator<Item = ColumnOperation<T>>,
@@ -497,9 +496,8 @@ where
         Cardinality::Multivalued => {
             let multivalued_index_builder = value_index_builders.borrow_multivalued_index_builder();
             consume_operation_iterator(op_iterator, multivalued_index_builder, values);
-            let _multivalued_index = multivalued_index_builder.finish(num_docs);
-            todo!();
-            // SerializableColumnIndex::Multivalued(Box::new(multivalued_index))
+            let multivalued_index = multivalued_index_builder.finish(num_docs);
+            SerializableColumnIndex::Multivalued(Box::new(multivalued_index))
         }
     };
     crate::column::serialize_column_u64(

columnar/src/columnar/writer/serializer.rs

@@ -5,7 +5,6 @@ use common::CountingWriter;
 use sstable::value::RangeValueWriter;
 use sstable::RangeSSTable;
 
-use crate::column;
 use crate::columnar::ColumnType;
 
 pub struct ColumnarSerializer<W: io::Write> {