zeppelin_embed/meta/
columns.rs

1//! Typed, contiguous column arrays and row-aligned builders.
2
3use super::bitmap::DocBitmap;
4use super::dict::{
5    DictionaryBuilder, DictionaryCodes, DictionaryError, StringDictionary, StringStorage,
6};
7use super::{ColumnId, ColumnType, Schema, TIMESTAMP_COLUMN};
8
9/// A borrowed, typed value supplied for one metadata row.
10#[derive(Clone, Copy, Debug, PartialEq)]
11pub enum ColumnValue<'a> {
12    /// An unsigned integer.
13    U64(u64),
14    /// A full document identifier.
15    Id128(crate::ingest::DocId),
16    /// A signed integer.
17    I64(i64),
18    /// A floating-point value.
19    F64(f64),
20    /// A Boolean value.
21    Bool(bool),
22    /// A UTF-8 value for either string physical representation.
23    String(&'a str),
24}
25
26impl ColumnValue<'_> {
27    fn compatible_with(self, column_type: ColumnType) -> bool {
28        matches!(
29            (self, column_type),
30            (Self::Id128(_), ColumnType::Id128)
31                | (Self::U64(_), ColumnType::U64)
32                | (Self::I64(_), ColumnType::I64)
33                | (Self::F64(_), ColumnType::F64)
34                | (Self::Bool(_), ColumnType::Bool)
35                | (
36                    Self::String(_),
37                    ColumnType::DictionaryString | ColumnType::RawString
38                )
39        )
40    }
41
42    fn value_type(self) -> ColumnType {
43        match self {
44            Self::Id128(_) => ColumnType::Id128,
45            Self::U64(_) => ColumnType::U64,
46            Self::I64(_) => ColumnType::I64,
47            Self::F64(_) => ColumnType::F64,
48            Self::Bool(_) => ColumnType::Bool,
49            Self::String(_) => ColumnType::RawString,
50        }
51    }
52}
53
54/// One column/value pair supplied while appending a row.
55#[derive(Clone, Copy, Debug, PartialEq)]
56pub struct ColumnInput<'a> {
57    /// The schema column receiving the value.
58    pub column: ColumnId,
59    /// The typed value to append.
60    pub value: ColumnValue<'a>,
61}
62
63/// A typed failure that leaves the logical row count unchanged.
64#[derive(Clone, Debug, Eq, PartialEq)]
65pub enum BuildError {
66    /// An input referred to an identifier absent from the schema.
67    UnknownColumn(ColumnId),
68    /// The same column appeared twice in one row.
69    DuplicateColumn(ColumnId),
70    /// A non-nullable column was omitted.
71    MissingRequiredColumn(ColumnId),
72    /// The timestamp must use the dedicated append argument.
73    TimestampProvidedAsInput,
74    /// A value did not match its declared physical type.
75    TypeMismatch {
76        /// The mismatched column.
77        column: ColumnId,
78        /// The declared physical type.
79        expected: ColumnType,
80        /// The supplied value type.
81        actual: ColumnType,
82    },
83    /// The segment reached the supported u32 document-id space.
84    TooManyRows,
85    /// Dictionary or string side-array construction failed.
86    Dictionary(DictionaryError),
87}
88
89/// Narrow limits used only to prove naturally huge metadata builder guards.
90#[cfg(any(test, feature = "test-seams"))]
91#[derive(Clone, Copy, Debug, Eq, PartialEq)]
92#[doc(hidden)]
93pub struct MetadataBuildTestLimits {
94    pub max_rows: u32,
95    pub max_dictionary_entries: u64,
96    pub max_string_bytes: u64,
97}
98
99#[cfg(any(test, feature = "test-seams"))]
100impl MetadataBuildTestLimits {
101    #[must_use]
102    pub const fn new(max_rows: u32, max_dictionary_entries: u64, max_string_bytes: u64) -> Self {
103        Self {
104            max_rows,
105            max_dictionary_entries,
106            max_string_bytes,
107        }
108    }
109}
110
111impl std::fmt::Display for BuildError {
112    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
113        match self {
114            Self::UnknownColumn(column) => write!(formatter, "unknown column {}", column.get()),
115            Self::DuplicateColumn(column) => {
116                write!(formatter, "duplicate column {}", column.get())
117            }
118            Self::MissingRequiredColumn(column) => {
119                write!(formatter, "missing required column {}", column.get())
120            }
121            Self::TimestampProvidedAsInput => {
122                formatter.write_str("timestamp must use the dedicated ts argument")
123            }
124            Self::TypeMismatch {
125                column,
126                expected,
127                actual,
128            } => write!(
129                formatter,
130                "column {} expects {expected:?}, received {actual:?}",
131                column.get()
132            ),
133            Self::TooManyRows => formatter.write_str("segment row count exceeds u32"),
134            Self::Dictionary(error) => error.fmt(formatter),
135        }
136    }
137}
138
139impl std::error::Error for BuildError {}
140
141impl From<DictionaryError> for BuildError {
142    fn from(error: DictionaryError) -> Self {
143        Self::Dictionary(error)
144    }
145}
146
147/// A nullable numeric array plus its explicit presence bitmap.
148#[derive(Clone, Debug, PartialEq)]
149#[repr(C)]
150pub struct NumericColumn<T> {
151    values: Vec<T>,
152    present: DocBitmap,
153}
154
155impl<T: Copy> NumericColumn<T> {
156    /// Returns the physical row count including null placeholders.
157    #[must_use]
158    pub fn len(&self) -> usize {
159        self.values.len()
160    }
161
162    /// Returns whether the physical array is empty.
163    #[must_use]
164    pub fn is_empty(&self) -> bool {
165        self.values.is_empty()
166    }
167
168    /// Reads a non-null value, returning `None` for null or out-of-range rows.
169    #[must_use]
170    pub fn get(&self, row: u32) -> Option<T> {
171        if !self.present.contains(row) {
172            return None;
173        }
174        let position = usize::try_from(row).ok()?;
175        self.values.get(position).copied()
176    }
177
178    /// Returns the presence bitmap used by `exists` and `null` predicates.
179    #[must_use]
180    pub fn present(&self) -> &DocBitmap {
181        &self.present
182    }
183
184    pub(crate) fn values(&self) -> &[T] {
185        &self.values
186    }
187}
188
189/// A nullable Boolean array stored as contiguous zero/one bytes.
190#[derive(Clone, Debug, Eq, PartialEq)]
191#[repr(C)]
192pub struct BoolColumn {
193    values: Vec<u8>,
194    present: DocBitmap,
195}
196
197impl BoolColumn {
198    /// Returns the physical row count including null placeholders.
199    #[must_use]
200    pub fn len(&self) -> usize {
201        self.values.len()
202    }
203
204    /// Returns whether the physical array is empty.
205    #[must_use]
206    pub fn is_empty(&self) -> bool {
207        self.values.is_empty()
208    }
209
210    /// Reads a non-null value, returning `None` for null or out-of-range rows.
211    #[must_use]
212    pub fn get(&self, row: u32) -> Option<bool> {
213        if !self.present.contains(row) {
214            return None;
215        }
216        let position = usize::try_from(row).ok()?;
217        self.values.get(position).map(|value| *value != 0)
218    }
219
220    /// Returns the presence bitmap used by `exists` and `null` predicates.
221    #[must_use]
222    pub fn present(&self) -> &DocBitmap {
223        &self.present
224    }
225
226    pub(crate) fn values(&self) -> &[u8] {
227        &self.values
228    }
229}
230
231/// A nullable, low-cardinality dictionary string column.
232#[derive(Clone, Debug, Eq, PartialEq)]
233#[repr(C)]
234pub struct DictionaryColumn {
235    codes: DictionaryCodes,
236    dictionary: StringDictionary,
237    present: DocBitmap,
238}
239
240impl DictionaryColumn {
241    /// Returns the physical row count including null placeholders.
242    #[must_use]
243    pub fn len(&self) -> usize {
244        self.codes.len()
245    }
246
247    /// Returns whether the physical array is empty.
248    #[must_use]
249    pub fn is_empty(&self) -> bool {
250        self.codes.is_empty()
251    }
252
253    /// Decodes a non-null row value.
254    #[must_use]
255    pub fn get(&self, row: u32) -> Option<&str> {
256        if !self.present.contains(row) {
257            return None;
258        }
259        let position = usize::try_from(row).ok()?;
260        let code = self.codes.get(position)?;
261        self.dictionary.get(code)
262    }
263
264    /// Returns the selected physical code array.
265    #[must_use]
266    pub fn codes(&self) -> &DictionaryCodes {
267        &self.codes
268    }
269
270    /// Returns the immutable string dictionary.
271    #[must_use]
272    pub fn dictionary(&self) -> &StringDictionary {
273        &self.dictionary
274    }
275
276    /// Returns the presence bitmap used by `exists` and `null` predicates.
277    #[must_use]
278    pub fn present(&self) -> &DocBitmap {
279        &self.present
280    }
281}
282
283/// A nullable raw UTF-8 column backed by contiguous offsets and bytes.
284#[derive(Clone, Debug, Eq, PartialEq)]
285#[repr(C)]
286pub struct RawStringColumn {
287    values: StringStorage,
288    present: DocBitmap,
289}
290
291impl RawStringColumn {
292    /// Returns the physical row count including null placeholders.
293    #[must_use]
294    pub fn len(&self) -> usize {
295        self.values.len()
296    }
297
298    /// Returns whether the physical array is empty.
299    #[must_use]
300    pub fn is_empty(&self) -> bool {
301        self.values.len() == 0
302    }
303
304    /// Reads a non-null row value.
305    #[must_use]
306    pub fn get(&self, row: u32) -> Option<&str> {
307        if !self.present.contains(row) {
308            return None;
309        }
310        let position = usize::try_from(row).ok()?;
311        self.values.get(position)
312    }
313
314    /// Returns the presence bitmap used by `exists` and `null` predicates.
315    #[must_use]
316    pub fn present(&self) -> &DocBitmap {
317        &self.present
318    }
319
320    pub(crate) fn physical_get(&self, position: usize) -> Option<&str> {
321        self.values.get(position)
322    }
323}
324
325/// One immutable typed physical column.
326#[derive(Clone, Debug, PartialEq)]
327pub enum Column {
328    /// An unsigned-integer array.
329    U64(NumericColumn<u64>),
330    /// A full document-identifier array.
331    Id128(NumericColumn<crate::ingest::DocId>),
332    /// A signed-integer array, including `ts`.
333    I64(NumericColumn<i64>),
334    /// A floating-point array.
335    F64(NumericColumn<f64>),
336    /// A Boolean byte array.
337    Bool(BoolColumn),
338    /// A dictionary string array.
339    DictionaryString(DictionaryColumn),
340    /// A raw UTF-8 array.
341    RawString(RawStringColumn),
342}
343
344impl Column {
345    /// Returns the physical column type.
346    #[must_use]
347    pub const fn column_type(&self) -> ColumnType {
348        match self {
349            Self::Id128(_) => ColumnType::Id128,
350            Self::U64(_) => ColumnType::U64,
351            Self::I64(_) => ColumnType::I64,
352            Self::F64(_) => ColumnType::F64,
353            Self::Bool(_) => ColumnType::Bool,
354            Self::DictionaryString(_) => ColumnType::DictionaryString,
355            Self::RawString(_) => ColumnType::RawString,
356        }
357    }
358
359    /// Returns the physical row count including null placeholders.
360    #[must_use]
361    pub fn len(&self) -> usize {
362        match self {
363            Self::Id128(column) => column.len(),
364            Self::U64(column) => column.len(),
365            Self::I64(column) => column.len(),
366            Self::F64(column) => column.len(),
367            Self::Bool(column) => column.len(),
368            Self::DictionaryString(column) => column.len(),
369            Self::RawString(column) => column.len(),
370        }
371    }
372
373    /// Returns whether the column contains no physical rows.
374    #[must_use]
375    pub fn is_empty(&self) -> bool {
376        self.len() == 0
377    }
378
379    /// Returns the non-null row bitmap.
380    #[must_use]
381    pub fn present(&self) -> &DocBitmap {
382        match self {
383            Self::Id128(column) => column.present(),
384            Self::U64(column) => column.present(),
385            Self::I64(column) => column.present(),
386            Self::F64(column) => column.present(),
387            Self::Bool(column) => column.present(),
388            Self::DictionaryString(column) => column.present(),
389            Self::RawString(column) => column.present(),
390        }
391    }
392}
393
394/// Immutable metadata columns aligned to one segment's document-id space.
395#[derive(Clone, Debug, PartialEq)]
396pub struct ColumnStore {
397    schema: Schema,
398    columns: Vec<Column>,
399    row_count: u32,
400}
401
402impl ColumnStore {
403    pub(crate) fn compact_for_cache(&mut self) {
404        for column in &mut self.columns {
405            match column {
406                Column::Id128(column) => column.present.compact_for_cache(),
407                Column::U64(column) => column.present.compact_for_cache(),
408                Column::I64(column) => column.present.compact_for_cache(),
409                Column::F64(column) => column.present.compact_for_cache(),
410                Column::Bool(column) => column.present.compact_for_cache(),
411                Column::DictionaryString(column) => column.present.compact_for_cache(),
412                Column::RawString(column) => column.present.compact_for_cache(),
413            }
414        }
415    }
416
417    pub(crate) fn resident_bytes(&self) -> Option<usize> {
418        let direct = self.schema.resident_bytes()?.checked_add(
419            self.columns
420                .capacity()
421                .checked_mul(std::mem::size_of::<Column>())?,
422        )?;
423        self.columns.iter().try_fold(direct, |total, column| {
424            total.checked_add(column_resident_bytes(column)?)
425        })
426    }
427
428    /// Returns the schema that defines these physical arrays.
429    #[must_use]
430    pub fn schema(&self) -> &Schema {
431        &self.schema
432    }
433
434    /// Returns the shared segment row count.
435    #[must_use]
436    pub const fn row_count(&self) -> u32 {
437        self.row_count
438    }
439
440    /// Finds an immutable physical column by typed identifier.
441    #[must_use]
442    pub fn column(&self, id: ColumnId) -> Option<&Column> {
443        self.schema
444            .columns()
445            .iter()
446            .zip(&self.columns)
447            .find_map(|(definition, column)| (definition.id() == id).then_some(column))
448    }
449
450    /// Reads one required timestamp.
451    #[must_use]
452    pub fn timestamp(&self, row: u32) -> Option<i64> {
453        match self.column(TIMESTAMP_COLUMN)? {
454            Column::I64(column) => column.get(row),
455            _ => None,
456        }
457    }
458
459    /// Returns the complete required timestamp side-array.
460    #[must_use]
461    pub fn timestamps(&self) -> &[i64] {
462        match self.column(TIMESTAMP_COLUMN) {
463            Some(Column::I64(column)) => column.values(),
464            _ => &[],
465        }
466    }
467}
468
469fn column_resident_bytes(column: &Column) -> Option<usize> {
470    match column {
471        Column::Id128(column) => column
472            .values
473            .capacity()
474            .checked_mul(std::mem::size_of::<crate::ingest::DocId>())?
475            .checked_add(column.present.resident_bytes()?),
476        Column::U64(column) => column
477            .values
478            .capacity()
479            .checked_mul(std::mem::size_of::<u64>())?
480            .checked_add(column.present.resident_bytes()?),
481        Column::I64(column) => column
482            .values
483            .capacity()
484            .checked_mul(std::mem::size_of::<i64>())?
485            .checked_add(column.present.resident_bytes()?),
486        Column::F64(column) => column
487            .values
488            .capacity()
489            .checked_mul(std::mem::size_of::<f64>())?
490            .checked_add(column.present.resident_bytes()?),
491        Column::Bool(column) => column
492            .values
493            .capacity()
494            .checked_add(column.present.resident_bytes()?),
495        Column::DictionaryString(column) => column
496            .codes
497            .resident_bytes()?
498            .checked_add(column.dictionary.resident_bytes()?)?
499            .checked_add(column.present.resident_bytes()?),
500        Column::RawString(column) => column
501            .values
502            .resident_bytes()?
503            .checked_add(column.present.resident_bytes()?),
504    }
505}
506
507enum BuilderColumn {
508    U64(Vec<u64>, DocBitmap),
509    Id128(Vec<crate::ingest::DocId>, DocBitmap),
510    I64(Vec<i64>, DocBitmap),
511    F64(Vec<f64>, DocBitmap),
512    Bool(Vec<u8>, DocBitmap),
513    Dictionary {
514        codes: Vec<u32>,
515        dictionary: DictionaryBuilder,
516        present: DocBitmap,
517    },
518    RawString(StringStorage, DocBitmap),
519}
520
521impl BuilderColumn {
522    fn new(column_type: ColumnType) -> Self {
523        match column_type {
524            ColumnType::Id128 => Self::Id128(Vec::new(), DocBitmap::new()),
525            ColumnType::U64 => Self::U64(Vec::new(), DocBitmap::new()),
526            ColumnType::I64 => Self::I64(Vec::new(), DocBitmap::new()),
527            ColumnType::F64 => Self::F64(Vec::new(), DocBitmap::new()),
528            ColumnType::Bool => Self::Bool(Vec::new(), DocBitmap::new()),
529            ColumnType::DictionaryString => Self::Dictionary {
530                codes: Vec::new(),
531                dictionary: DictionaryBuilder::new(),
532                present: DocBitmap::new(),
533            },
534            ColumnType::RawString => Self::RawString(StringStorage::new(), DocBitmap::new()),
535        }
536    }
537
538    fn preflight(&self, value: Option<ColumnValue<'_>>) -> Result<(), DictionaryError> {
539        match (self, value) {
540            (Self::Dictionary { dictionary, .. }, Some(ColumnValue::String(value))) => {
541                dictionary.can_intern(value)
542            }
543            (Self::RawString(values, _), Some(ColumnValue::String(value))) => {
544                values.can_push(value)
545            }
546            (Self::RawString(values, _), None) => values.can_push(""),
547            _ => Ok(()),
548        }
549    }
550
551    #[cfg(any(test, feature = "test-seams"))]
552    fn check_test_limits(
553        &self,
554        value: Option<ColumnValue<'_>>,
555        limits: MetadataBuildTestLimits,
556    ) -> Result<(), DictionaryError> {
557        match (self, value) {
558            (Self::Dictionary { dictionary, .. }, Some(ColumnValue::String(value))) => {
559                if !dictionary.contains(value)
560                    && dictionary.cardinality() >= limits.max_dictionary_entries
561                {
562                    return Err(DictionaryError::CardinalityOverflow);
563                }
564                let bytes = u64::try_from(dictionary.string_bytes())
565                    .unwrap_or(u64::MAX)
566                    .checked_add(u64::try_from(value.len()).unwrap_or(u64::MAX))
567                    .ok_or(DictionaryError::StringStorageOverflow)?;
568                if !dictionary.contains(value) && bytes > limits.max_string_bytes {
569                    return Err(DictionaryError::StringStorageOverflow);
570                }
571                Ok(())
572            }
573            (Self::RawString(values, _), Some(ColumnValue::String(value))) => {
574                let bytes = u64::try_from(values.byte_len())
575                    .unwrap_or(u64::MAX)
576                    .checked_add(u64::try_from(value.len()).unwrap_or(u64::MAX))
577                    .ok_or(DictionaryError::StringStorageOverflow)?;
578                if bytes > limits.max_string_bytes {
579                    Err(DictionaryError::StringStorageOverflow)
580                } else {
581                    Ok(())
582                }
583            }
584            (Self::RawString(values, _), None) => {
585                if u64::try_from(values.byte_len()).unwrap_or(u64::MAX) > limits.max_string_bytes {
586                    Err(DictionaryError::StringStorageOverflow)
587                } else {
588                    Ok(())
589                }
590            }
591            _ => Ok(()),
592        }
593    }
594
595    fn push(
596        &mut self,
597        row: u32,
598        column: ColumnId,
599        value: Option<ColumnValue<'_>>,
600    ) -> Result<(), BuildError> {
601        let expected = self.column_type();
602        match (self, value) {
603            (Self::Id128(values, present), Some(ColumnValue::Id128(value))) => {
604                values.push(value);
605                present.insert(row);
606            }
607            (Self::U64(values, present), Some(ColumnValue::U64(value))) => {
608                values.push(value);
609                present.insert(row);
610            }
611            (Self::Id128(values, _), None) => values.push(crate::ingest::DocId::new(0)),
612            (Self::U64(values, _), None) => values.push(0),
613            (Self::I64(values, present), Some(ColumnValue::I64(value))) => {
614                values.push(value);
615                present.insert(row);
616            }
617            (Self::I64(values, _), None) => values.push(0),
618            (Self::F64(values, present), Some(ColumnValue::F64(value))) => {
619                values.push(value);
620                present.insert(row);
621            }
622            (Self::F64(values, _), None) => values.push(0.0),
623            (Self::Bool(values, present), Some(ColumnValue::Bool(value))) => {
624                values.push(u8::from(value));
625                present.insert(row);
626            }
627            (Self::Bool(values, _), None) => values.push(0),
628            (
629                Self::Dictionary {
630                    codes,
631                    dictionary,
632                    present,
633                },
634                Some(ColumnValue::String(value)),
635            ) => {
636                codes.push(dictionary.intern(value)?);
637                present.insert(row);
638            }
639            (Self::Dictionary { codes, .. }, None) => codes.push(0),
640            (Self::RawString(values, present), Some(ColumnValue::String(value))) => {
641                values.push(value)?;
642                present.insert(row);
643            }
644            (Self::RawString(values, _), None) => values.push("")?,
645            (_, Some(value)) => {
646                return Err(BuildError::TypeMismatch {
647                    column,
648                    expected,
649                    actual: value.value_type(),
650                });
651            }
652        }
653        Ok(())
654    }
655
656    fn column_type(&self) -> ColumnType {
657        match self {
658            Self::Id128(_, _) => ColumnType::Id128,
659            Self::U64(_, _) => ColumnType::U64,
660            Self::I64(_, _) => ColumnType::I64,
661            Self::F64(_, _) => ColumnType::F64,
662            Self::Bool(_, _) => ColumnType::Bool,
663            Self::Dictionary { .. } => ColumnType::DictionaryString,
664            Self::RawString(_, _) => ColumnType::RawString,
665        }
666    }
667
668    fn finish(self) -> Result<Column, DictionaryError> {
669        match self {
670            Self::Id128(values, present) => Ok(Column::Id128(NumericColumn { values, present })),
671            Self::U64(values, present) => Ok(Column::U64(NumericColumn { values, present })),
672            Self::I64(values, present) => Ok(Column::I64(NumericColumn { values, present })),
673            Self::F64(values, present) => Ok(Column::F64(NumericColumn { values, present })),
674            Self::Bool(values, present) => Ok(Column::Bool(BoolColumn { values, present })),
675            Self::Dictionary {
676                codes,
677                dictionary,
678                present,
679            } => {
680                let cardinality = dictionary.cardinality();
681                Ok(Column::DictionaryString(DictionaryColumn {
682                    codes: DictionaryCodes::from_u32(codes, cardinality)?,
683                    dictionary: dictionary.finish(),
684                    present,
685                }))
686            }
687            Self::RawString(values, present) => {
688                Ok(Column::RawString(RawStringColumn { values, present }))
689            }
690        }
691    }
692}
693
694/// Mutable row builder that keeps every physical column exactly aligned.
695pub struct ColumnStoreBuilder {
696    schema: Schema,
697    columns: Vec<BuilderColumn>,
698    row_count: u32,
699    #[cfg(any(test, feature = "test-seams"))]
700    test_limits: Option<MetadataBuildTestLimits>,
701}
702
703impl ColumnStoreBuilder {
704    /// Creates empty physical arrays for a validated schema.
705    #[must_use]
706    pub fn new(schema: Schema) -> Self {
707        let columns = schema
708            .columns()
709            .iter()
710            .map(|definition| BuilderColumn::new(definition.column_type()))
711            .collect();
712        Self {
713            schema,
714            columns,
715            row_count: 0,
716            #[cfg(any(test, feature = "test-seams"))]
717            test_limits: None,
718        }
719    }
720
721    /// Creates a builder with lower limits at the production guard sites.
722    #[cfg(any(test, feature = "test-seams"))]
723    #[doc(hidden)]
724    #[must_use]
725    pub fn new_with_test_limits(schema: Schema, limits: MetadataBuildTestLimits) -> Self {
726        let mut builder = Self::new(schema);
727        builder.test_limits = Some(limits);
728        builder
729    }
730
731    /// Appends one timestamp and its typed user-column values atomically.
732    pub fn push_row(
733        &mut self,
734        timestamp: i64,
735        inputs: &[ColumnInput<'_>],
736    ) -> Result<u32, BuildError> {
737        if self.row_count == u32::MAX {
738            return Err(BuildError::TooManyRows);
739        }
740        #[cfg(any(test, feature = "test-seams"))]
741        if self
742            .test_limits
743            .is_some_and(|limits| self.row_count >= limits.max_rows)
744        {
745            return Err(BuildError::TooManyRows);
746        }
747        self.validate_inputs(inputs)?;
748
749        for (definition, builder) in self.schema.columns().iter().zip(&self.columns) {
750            let value = if definition.id() == TIMESTAMP_COLUMN {
751                Some(ColumnValue::I64(timestamp))
752            } else {
753                find_input(inputs, definition.id()).map(|input| input.value)
754            };
755            #[cfg(any(test, feature = "test-seams"))]
756            if let Some(limits) = self.test_limits {
757                builder.check_test_limits(value, limits)?;
758            }
759            builder.preflight(value)?;
760        }
761
762        let row = self.row_count;
763        for (definition, builder) in self.schema.columns().iter().zip(&mut self.columns) {
764            let value = if definition.id() == TIMESTAMP_COLUMN {
765                Some(ColumnValue::I64(timestamp))
766            } else {
767                find_input(inputs, definition.id()).map(|input| input.value)
768            };
769            builder.push(row, definition.id(), value)?;
770        }
771        self.row_count = self.row_count.saturating_add(1);
772        Ok(row)
773    }
774
775    /// Seals the aligned arrays into an immutable column store.
776    pub fn finish(self) -> Result<ColumnStore, BuildError> {
777        let columns = self
778            .columns
779            .into_iter()
780            .map(BuilderColumn::finish)
781            .collect::<Result<Vec<_>, _>>()?;
782        Ok(ColumnStore {
783            schema: self.schema,
784            columns,
785            row_count: self.row_count,
786        })
787    }
788
789    fn validate_inputs(&self, inputs: &[ColumnInput<'_>]) -> Result<(), BuildError> {
790        for (position, input) in inputs.iter().enumerate() {
791            if input.column == TIMESTAMP_COLUMN {
792                return Err(BuildError::TimestampProvidedAsInput);
793            }
794            if inputs
795                .iter()
796                .take(position)
797                .any(|candidate| candidate.column == input.column)
798            {
799                return Err(BuildError::DuplicateColumn(input.column));
800            }
801            let definition = self
802                .schema
803                .column(input.column)
804                .ok_or(BuildError::UnknownColumn(input.column))?;
805            if !input.value.compatible_with(definition.column_type()) {
806                return Err(BuildError::TypeMismatch {
807                    column: input.column,
808                    expected: definition.column_type(),
809                    actual: input.value.value_type(),
810                });
811            }
812        }
813
814        for definition in self
815            .schema
816            .columns()
817            .iter()
818            .filter(|definition| definition.id() != TIMESTAMP_COLUMN)
819        {
820            if !definition.is_nullable() && find_input(inputs, definition.id()).is_none() {
821                return Err(BuildError::MissingRequiredColumn(definition.id()));
822            }
823        }
824        Ok(())
825    }
826}
827
828fn find_input<'a, 'value>(
829    inputs: &'a [ColumnInput<'value>],
830    column: ColumnId,
831) -> Option<&'a ColumnInput<'value>> {
832    inputs.iter().find(|input| input.column == column)
833}
834
835#[allow(
836    clippy::expect_used,
837    clippy::indexing_slicing,
838    clippy::panic,
839    clippy::unwrap_used
840)]
841#[cfg(test)]
842mod tests {
843    use super::*;
844    use crate::meta::{CodeWidth, ColumnDefinition};
845    use proptest::prelude::*;
846    use proptest::test_runner::{Config, RngSeed, TestRunner};
847    use rand::RngCore;
848
849    #[derive(Clone, Debug)]
850    struct ArbitraryRow {
851        timestamp: i64,
852        unsigned: Option<u64>,
853        signed: Option<i64>,
854        float: Option<f64>,
855        boolean: Option<bool>,
856        dictionary: Option<String>,
857        raw: Option<String>,
858    }
859
860    impl Arbitrary for ArbitraryRow {
861        type Parameters = ();
862        type Strategy = BoxedStrategy<Self>;
863
864        fn arbitrary_with((): Self::Parameters) -> Self::Strategy {
865            (
866                any::<i64>(),
867                proptest::option::of(any::<u64>()),
868                proptest::option::of(any::<i64>()),
869                proptest::option::of(any::<i32>().prop_map(f64::from)),
870                proptest::option::of(any::<bool>()),
871                proptest::option::of("[a-z]{0,12}"),
872                proptest::option::of("[ -~]{0,24}"),
873            )
874                .prop_map(
875                    |(timestamp, unsigned, signed, float, boolean, dictionary, raw)| Self {
876                        timestamp,
877                        unsigned,
878                        signed,
879                        float,
880                        boolean,
881                        dictionary,
882                        raw,
883                    },
884                )
885                .boxed()
886        }
887    }
888
889    #[test]
890    fn id128_columns_preserve_all_bits() {
891        use crate::ingest::DocId;
892        let id = ColumnId::new(1);
893        let schema = Schema::new(vec![ColumnDefinition::new(
894            id,
895            "parent",
896            ColumnType::Id128,
897            true,
898        )])
899        .unwrap();
900        let mut builder = ColumnStoreBuilder::new(schema);
901        for value in [0, 1, (1_u128 << 127) + 1, u128::MAX] {
902            builder
903                .push_row(
904                    0,
905                    &[ColumnInput {
906                        column: id,
907                        value: ColumnValue::Id128(DocId::new(value)),
908                    }],
909                )
910                .unwrap();
911        }
912        builder.push_row(0, &[]).unwrap();
913        let store = builder.finish().unwrap();
914        let Some(Column::Id128(column)) = store.column(id) else {
915            panic!("id column missing")
916        };
917        assert_eq!(column.get(0), Some(DocId::new(0)));
918        assert_eq!(column.get(2), Some(DocId::new((1_u128 << 127) + 1)));
919        assert_eq!(column.get(3), Some(DocId::new(u128::MAX)));
920        assert_eq!(column.get(4), None);
921        use crate::meta::{AliveSet, Predicate, PredicateValue, evaluate};
922        let alive = AliveSet::new(5);
923        let result = evaluate(
924            &Predicate::Eq {
925                column: id,
926                value: PredicateValue::Id128(DocId::new((1_u128 << 127) + 1)),
927            },
928            &store,
929            &alive,
930        )
931        .unwrap();
932        assert_eq!(result.iter().collect::<Vec<_>>(), vec![2]);
933    }
934
935    fn complete_schema(nullable: bool) -> Schema {
936        Schema::new(vec![
937            ColumnDefinition::new(ColumnId::new(1), "u", ColumnType::U64, nullable),
938            ColumnDefinition::new(ColumnId::new(2), "i", ColumnType::I64, nullable),
939            ColumnDefinition::new(ColumnId::new(3), "f", ColumnType::F64, nullable),
940            ColumnDefinition::new(ColumnId::new(4), "b", ColumnType::Bool, nullable),
941            ColumnDefinition::new(
942                ColumnId::new(5),
943                "dictionary",
944                ColumnType::DictionaryString,
945                nullable,
946            ),
947            ColumnDefinition::new(ColumnId::new(6), "raw", ColumnType::RawString, nullable),
948        ])
949        .expect("schema is valid")
950    }
951
952    fn build_arbitrary_store(rows: &[ArbitraryRow]) -> ColumnStore {
953        let mut builder = ColumnStoreBuilder::new(complete_schema(true));
954        for row in rows {
955            let mut inputs = Vec::new();
956            if let Some(value) = row.unsigned {
957                inputs.push(ColumnInput {
958                    column: ColumnId::new(1),
959                    value: ColumnValue::U64(value),
960                });
961            }
962            if let Some(value) = row.signed {
963                inputs.push(ColumnInput {
964                    column: ColumnId::new(2),
965                    value: ColumnValue::I64(value),
966                });
967            }
968            if let Some(value) = row.float {
969                inputs.push(ColumnInput {
970                    column: ColumnId::new(3),
971                    value: ColumnValue::F64(value),
972                });
973            }
974            if let Some(value) = row.boolean {
975                inputs.push(ColumnInput {
976                    column: ColumnId::new(4),
977                    value: ColumnValue::Bool(value),
978                });
979            }
980            if let Some(value) = row.dictionary.as_deref() {
981                inputs.push(ColumnInput {
982                    column: ColumnId::new(5),
983                    value: ColumnValue::String(value),
984                });
985            }
986            if let Some(value) = row.raw.as_deref() {
987                inputs.push(ColumnInput {
988                    column: ColumnId::new(6),
989                    value: ColumnValue::String(value),
990                });
991            }
992            builder
993                .push_row(row.timestamp, &inputs)
994                .expect("arbitrary row is schema-compatible");
995        }
996        builder.finish().expect("bounded store finishes")
997    }
998
999    #[test]
1000    fn prop_column_store_builders_and_accessors_roundtrip() {
1001        let name = "meta::columns::tests::prop_column_store_builders_and_accessors_roundtrip";
1002        let mut seeded = crate::test_support::seeded_rng(name);
1003        let config = Config {
1004            rng_seed: RngSeed::Fixed(seeded.next_u64()),
1005            cases: 256,
1006            ..Config::default()
1007        };
1008        let mut runner = TestRunner::new(config);
1009        let rows = prop::collection::vec(any::<ArbitraryRow>(), 0..=32);
1010        let result = runner.run(&rows, |rows| {
1011            let store = build_arbitrary_store(&rows);
1012            prop_assert_eq!(store.row_count(), rows.len() as u32);
1013            prop_assert_eq!(store.schema(), &complete_schema(true));
1014            prop_assert_eq!(
1015                store.timestamps(),
1016                rows.iter().map(|row| row.timestamp).collect::<Vec<_>>()
1017            );
1018
1019            for (offset, expected_type) in [
1020                ColumnType::I64,
1021                ColumnType::U64,
1022                ColumnType::I64,
1023                ColumnType::F64,
1024                ColumnType::Bool,
1025                ColumnType::DictionaryString,
1026                ColumnType::RawString,
1027            ]
1028            .into_iter()
1029            .enumerate()
1030            {
1031                let column = store
1032                    .column(ColumnId::new(offset as u32))
1033                    .expect("defined column");
1034                prop_assert_eq!(column.column_type(), expected_type);
1035                prop_assert_eq!(column.len(), rows.len());
1036                prop_assert_eq!(column.is_empty(), rows.is_empty());
1037            }
1038
1039            for (position, row) in rows.iter().enumerate() {
1040                let row_id = position as u32;
1041                prop_assert_eq!(store.timestamp(row_id), Some(row.timestamp));
1042                match store.column(ColumnId::new(1)) {
1043                    Some(Column::U64(column)) => prop_assert_eq!(column.get(row_id), row.unsigned),
1044                    actual => prop_assert!(false, "unexpected u64 column: {actual:?}"),
1045                }
1046                match store.column(ColumnId::new(2)) {
1047                    Some(Column::I64(column)) => prop_assert_eq!(column.get(row_id), row.signed),
1048                    actual => prop_assert!(false, "unexpected i64 column: {actual:?}"),
1049                }
1050                match store.column(ColumnId::new(3)) {
1051                    Some(Column::F64(column)) => prop_assert_eq!(column.get(row_id), row.float),
1052                    actual => prop_assert!(false, "unexpected f64 column: {actual:?}"),
1053                }
1054                match store.column(ColumnId::new(4)) {
1055                    Some(Column::Bool(column)) => {
1056                        prop_assert_eq!(column.get(row_id), row.boolean)
1057                    }
1058                    actual => prop_assert!(false, "unexpected bool column: {actual:?}"),
1059                }
1060                match store.column(ColumnId::new(5)) {
1061                    Some(Column::DictionaryString(column)) => {
1062                        prop_assert_eq!(column.get(row_id), row.dictionary.as_deref());
1063                    }
1064                    actual => prop_assert!(false, "unexpected dictionary column: {actual:?}"),
1065                }
1066                match store.column(ColumnId::new(6)) {
1067                    Some(Column::RawString(column)) => {
1068                        prop_assert_eq!(column.get(row_id), row.raw.as_deref());
1069                    }
1070                    actual => prop_assert!(false, "unexpected raw column: {actual:?}"),
1071                }
1072            }
1073            let outside = rows.len() as u32;
1074            prop_assert_eq!(store.timestamp(outside), None);
1075            prop_assert_eq!(store.column(ColumnId::new(99)), None);
1076            Ok(())
1077        });
1078        assert!(result.is_ok(), "property result: {result:?}");
1079    }
1080
1081    #[test]
1082    fn builder_errors_are_typed_and_atomic_at_capacity_edges() {
1083        let mut builder = ColumnStoreBuilder::new(complete_schema(false));
1084        let error = builder
1085            .push_row(
1086                7,
1087                &[ColumnInput {
1088                    column: ColumnId::new(1),
1089                    value: ColumnValue::Bool(true),
1090                }],
1091            )
1092            .expect_err("wrong physical type must fail");
1093        assert_eq!(
1094            error,
1095            BuildError::TypeMismatch {
1096                column: ColumnId::new(1),
1097                expected: ColumnType::U64,
1098                actual: ColumnType::Bool,
1099            }
1100        );
1101        assert_eq!(builder.row_count, 0);
1102
1103        assert_eq!(
1104            builder.push_row(7, &[]),
1105            Err(BuildError::MissingRequiredColumn(ColumnId::new(1)))
1106        );
1107        assert_eq!(builder.row_count, 0, "length mismatch must be atomic");
1108
1109        assert_eq!(
1110            builder.push_row(
1111                7,
1112                &[ColumnInput {
1113                    column: ColumnId::new(99),
1114                    value: ColumnValue::U64(1),
1115                }]
1116            ),
1117            Err(BuildError::UnknownColumn(ColumnId::new(99)))
1118        );
1119        assert_eq!(
1120            builder.push_row(
1121                7,
1122                &[
1123                    ColumnInput {
1124                        column: ColumnId::new(1),
1125                        value: ColumnValue::U64(1),
1126                    },
1127                    ColumnInput {
1128                        column: ColumnId::new(1),
1129                        value: ColumnValue::U64(2),
1130                    },
1131                ]
1132            ),
1133            Err(BuildError::DuplicateColumn(ColumnId::new(1)))
1134        );
1135        assert_eq!(
1136            builder.push_row(
1137                7,
1138                &[ColumnInput {
1139                    column: TIMESTAMP_COLUMN,
1140                    value: ColumnValue::I64(7),
1141                }]
1142            ),
1143            Err(BuildError::TimestampProvidedAsInput)
1144        );
1145        assert_eq!(builder.row_count, 0);
1146
1147        let mut at_capacity = ColumnStoreBuilder::new(complete_schema(true));
1148        at_capacity.row_count = u32::MAX;
1149        assert_eq!(at_capacity.push_row(8, &[]), Err(BuildError::TooManyRows));
1150        assert_eq!(at_capacity.row_count, u32::MAX);
1151
1152        let mut nullable = ColumnStoreBuilder::new(complete_schema(true));
1153        assert_eq!(nullable.push_row(11, &[]), Ok(0));
1154        let store = nullable.finish().expect("null row finishes");
1155        assert_eq!(store.timestamps(), &[11]);
1156        for id in 1..=6 {
1157            let column = store.column(ColumnId::new(id)).expect("nullable column");
1158            assert_eq!(column.len(), 1);
1159            assert_eq!(column.present().cardinality(), 0);
1160        }
1161        match store.column(ColumnId::new(1)) {
1162            Some(Column::U64(column)) => assert_eq!(column.get(0), None),
1163            actual => panic!("unexpected column: {actual:?}"),
1164        }
1165        match store.column(ColumnId::new(4)) {
1166            Some(Column::Bool(column)) => assert_eq!(column.get(0), None),
1167            actual => panic!("unexpected column: {actual:?}"),
1168        }
1169        match store.column(ColumnId::new(5)) {
1170            Some(Column::DictionaryString(column)) => assert_eq!(column.get(0), None),
1171            actual => panic!("unexpected column: {actual:?}"),
1172        }
1173        match store.column(ColumnId::new(6)) {
1174            Some(Column::RawString(column)) => assert_eq!(column.get(0), None),
1175            actual => panic!("unexpected column: {actual:?}"),
1176        }
1177    }
1178
1179    fn dictionary_store(cardinality: u32) -> ColumnStore {
1180        let schema = Schema::new(vec![ColumnDefinition::new(
1181            ColumnId::new(1),
1182            "dictionary",
1183            ColumnType::DictionaryString,
1184            false,
1185        )])
1186        .expect("schema is valid");
1187        let mut builder = ColumnStoreBuilder::new(schema);
1188        for value in 0..cardinality {
1189            let text = value.to_string();
1190            builder
1191                .push_row(
1192                    i64::from(value),
1193                    &[ColumnInput {
1194                        column: ColumnId::new(1),
1195                        value: ColumnValue::String(&text),
1196                    }],
1197                )
1198                .expect("bounded dictionary row");
1199        }
1200        builder.finish().expect("bounded dictionary finishes")
1201    }
1202
1203    #[test]
1204    fn column_builder_widens_dictionary_at_u16_boundary() {
1205        let narrow = dictionary_store(u32::from(u16::MAX));
1206        match narrow.column(ColumnId::new(1)) {
1207            Some(Column::DictionaryString(column)) => {
1208                assert_eq!(column.codes().width(), CodeWidth::U16);
1209                assert_eq!(column.get(u32::from(u16::MAX) - 1), Some("65534"));
1210            }
1211            actual => panic!("unexpected dictionary column: {actual:?}"),
1212        }
1213
1214        let wide = dictionary_store(u32::from(u16::MAX).saturating_add(1));
1215        match wide.column(ColumnId::new(1)) {
1216            Some(Column::DictionaryString(column)) => {
1217                assert_eq!(column.codes().width(), CodeWidth::U32);
1218                assert_eq!(column.get(u32::from(u16::MAX)), Some("65535"));
1219            }
1220            actual => panic!("unexpected dictionary column: {actual:?}"),
1221        }
1222    }
1223}
zeppelin_embed/meta/columns.rs

zeppelin_embed/meta/
columns.rs
