zeppelin_embed/meta/
mod.rs

1//! Typed, segment-aligned metadata columns and bitmap predicate evaluation.
2//!
3//! Metadata predicates are a closed data model. They identify schema columns
4//! by [`crate::meta::ColumnId`]; this module contains no parser and accepts no
5//! filter text.
6
7mod alive;
8mod bitmap;
9#[cfg(any(test, feature = "test-seams"))]
10#[doc(hidden)]
11pub mod bitmap_observer;
12mod columns;
13mod dict;
14mod predicate;
15
16pub use alive::{AliveError, AliveSet};
17pub use bitmap::DocBitmap;
18#[cfg(any(test, feature = "test-seams"))]
19#[doc(hidden)]
20pub use columns::MetadataBuildTestLimits;
21pub use columns::{
22    BoolColumn, BuildError, Column, ColumnInput, ColumnStore, ColumnStoreBuilder, ColumnValue,
23    DictionaryColumn, NumericColumn, RawStringColumn,
24};
25pub use dict::{CodeWidth, DictionaryCodes, DictionaryError, StringDictionary};
26pub use predicate::{EvalError, Predicate, PredicateValue, RangeBound, RangePredicate, evaluate};
27
28/// The reserved identifier of the required per-row timestamp column.
29pub const TIMESTAMP_COLUMN: ColumnId = ColumnId(0);
30
31/// A stable, schema-local identifier used by structured predicates.
32#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
33#[repr(transparent)]
34pub struct ColumnId(u32);
35
36impl ColumnId {
37    /// Creates an identifier from its schema-local integer representation.
38    #[must_use]
39    pub const fn new(value: u32) -> Self {
40        Self(value)
41    }
42
43    /// Returns the schema-local integer representation.
44    #[must_use]
45    pub const fn get(self) -> u32 {
46        self.0
47    }
48}
49
50/// The physical type of a segment-aligned metadata column.
51#[derive(Clone, Copy, Debug, Eq, PartialEq)]
52pub enum ColumnType {
53    /// Unsigned 64-bit integers.
54    U64,
55    /// Full 128-bit document identifiers.
56    Id128,
57    /// Signed 64-bit integers.
58    I64,
59    /// IEEE-754 64-bit floating-point values.
60    F64,
61    /// Boolean values stored as contiguous bytes.
62    Bool,
63    /// Low-cardinality UTF-8 strings encoded through a segment dictionary.
64    DictionaryString,
65    /// Raw UTF-8 strings retained as stored-field values.
66    RawString,
67}
68
69/// One typed user-column declaration in a collection schema.
70#[derive(Clone, Debug, Eq, PartialEq)]
71pub struct ColumnDefinition {
72    id: ColumnId,
73    name: String,
74    column_type: ColumnType,
75    nullable: bool,
76}
77
78impl ColumnDefinition {
79    /// Creates a typed column declaration.
80    #[must_use]
81    pub fn new(
82        id: ColumnId,
83        name: impl Into<String>,
84        column_type: ColumnType,
85        nullable: bool,
86    ) -> Self {
87        Self {
88            id,
89            name: name.into(),
90            column_type,
91            nullable,
92        }
93    }
94
95    /// Returns the column identifier.
96    #[must_use]
97    pub const fn id(&self) -> ColumnId {
98        self.id
99    }
100
101    /// Returns the binding-facing column name.
102    #[must_use]
103    pub fn name(&self) -> &str {
104        &self.name
105    }
106
107    /// Returns the physical column type.
108    #[must_use]
109    pub const fn column_type(&self) -> ColumnType {
110        self.column_type
111    }
112
113    /// Returns whether a row may omit this column.
114    #[must_use]
115    pub const fn is_nullable(&self) -> bool {
116        self.nullable
117    }
118}
119
120/// A schema-construction failure detected before any rows are accepted.
121#[derive(Clone, Debug, Eq, PartialEq)]
122pub enum SchemaError {
123    /// Column zero is reserved for the required timestamp array.
124    ReservedTimestampId,
125    /// The name `ts` is reserved for the required timestamp array.
126    ReservedTimestampName,
127    /// Two declarations used the same identifier.
128    DuplicateColumnId(ColumnId),
129    /// Two declarations used the same binding-facing name.
130    DuplicateColumnName(String),
131}
132
133impl std::fmt::Display for SchemaError {
134    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
135        match self {
136            Self::ReservedTimestampId => formatter.write_str("column id 0 is reserved for ts"),
137            Self::ReservedTimestampName => formatter.write_str("column name ts is reserved"),
138            Self::DuplicateColumnId(id) => write!(formatter, "duplicate column id {}", id.get()),
139            Self::DuplicateColumnName(name) => write!(formatter, "duplicate column name {name}"),
140        }
141    }
142}
143
144impl std::error::Error for SchemaError {}
145
146/// Why a declared schema is not an additive extension of a persisted one.
147#[derive(Clone, Debug, Eq, PartialEq)]
148pub enum SchemaConflict {
149    /// A persisted attribute is absent from the declaration.
150    Removed(ColumnDefinition),
151    /// A persisted attribute is declared with a different name, type, or
152    /// nullability.
153    Changed {
154        /// Definition already persisted.
155        persisted: ColumnDefinition,
156        /// Definition declared with the same identifier.
157        declared: ColumnDefinition,
158    },
159    /// A new attribute is declared non-nullable, so rows written before it
160    /// existed could not read it as null.
161    AddedNotNullable(ColumnDefinition),
162}
163
164struct DescribedColumn<'a>(&'a ColumnDefinition);
165
166impl std::fmt::Display for DescribedColumn<'_> {
167    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
168        let definition = self.0;
169        write!(
170            formatter,
171            "'{}' (id {}, {:?}, {})",
172            definition.name,
173            definition.id.get(),
174            definition.column_type,
175            if definition.nullable {
176                "nullable"
177            } else {
178                "not nullable"
179            }
180        )
181    }
182}
183
184impl std::fmt::Display for SchemaConflict {
185    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
186        match self {
187            Self::Removed(persisted) => write!(
188                formatter,
189                "attribute {} is persisted but not declared; attributes cannot be removed",
190                DescribedColumn(persisted)
191            ),
192            Self::Changed {
193                persisted,
194                declared,
195            } => write!(
196                formatter,
197                "attribute {} is declared as {}; persisted attributes cannot change",
198                DescribedColumn(persisted),
199                DescribedColumn(declared)
200            ),
201            Self::AddedNotNullable(declared) => write!(
202                formatter,
203                "added attribute {} must be nullable so existing documents read it as null",
204                DescribedColumn(declared)
205            ),
206        }
207    }
208}
209
210impl std::error::Error for SchemaConflict {}
211
212/// An immutable collection schema including the required `ts: i64` column.
213#[derive(Clone, Debug, Eq, PartialEq)]
214pub struct Schema {
215    columns: Vec<ColumnDefinition>,
216}
217
218impl Schema {
219    pub(crate) fn resident_bytes(&self) -> Option<usize> {
220        let definitions = self
221            .columns
222            .capacity()
223            .checked_mul(std::mem::size_of::<ColumnDefinition>())?;
224        self.columns
225            .iter()
226            .try_fold(definitions, |total, definition| {
227                total.checked_add(definition.name.capacity())
228            })
229    }
230
231    /// Creates the mandatory timestamp-only schema used by public ingest today.
232    #[must_use]
233    pub fn timestamp_only() -> Self {
234        Self {
235            columns: vec![ColumnDefinition::new(
236                TIMESTAMP_COLUMN,
237                "ts",
238                ColumnType::I64,
239                false,
240            )],
241        }
242    }
243
244    /// Validates user columns and prepends the reserved timestamp definition.
245    pub fn new(user_columns: Vec<ColumnDefinition>) -> Result<Self, SchemaError> {
246        for (position, definition) in user_columns.iter().enumerate() {
247            if definition.id == TIMESTAMP_COLUMN {
248                return Err(SchemaError::ReservedTimestampId);
249            }
250            if definition.name == "ts" {
251                return Err(SchemaError::ReservedTimestampName);
252            }
253            if let Some(duplicate) = user_columns.iter().take(position).find(|candidate| {
254                candidate.id == definition.id || candidate.name == definition.name
255            }) {
256                if duplicate.id == definition.id {
257                    return Err(SchemaError::DuplicateColumnId(definition.id));
258                }
259                return Err(SchemaError::DuplicateColumnName(definition.name.clone()));
260            }
261        }
262
263        let mut columns = Self::timestamp_only().columns;
264        columns.reserve(user_columns.len());
265        columns.extend(user_columns);
266        Ok(Self { columns })
267    }
268
269    /// Reconciles a declared schema with this persisted one by column id.
270    ///
271    /// Every persisted column must be declared unchanged, and every declared
272    /// column this schema lacks must be nullable; declaration order does not
273    /// matter. Returns `None` when both name the same columns, otherwise the
274    /// evolved schema: this schema's columns in their order, then the added
275    /// columns in declaration order.
276    pub fn additive_evolution(&self, declared: &Self) -> Result<Option<Self>, SchemaConflict> {
277        for persisted in &self.columns {
278            match declared.column(persisted.id) {
279                None => return Err(SchemaConflict::Removed(persisted.clone())),
280                Some(candidate) if candidate != persisted => {
281                    return Err(SchemaConflict::Changed {
282                        persisted: persisted.clone(),
283                        declared: candidate.clone(),
284                    });
285                }
286                Some(_) => {}
287            }
288        }
289        let mut added = Vec::new();
290        for candidate in &declared.columns {
291            if self.column(candidate.id).is_none() {
292                if !candidate.nullable {
293                    return Err(SchemaConflict::AddedNotNullable(candidate.clone()));
294                }
295                added.push(candidate.clone());
296            }
297        }
298        if added.is_empty() {
299            return Ok(None);
300        }
301        let mut columns = self.columns.clone();
302        columns.extend(added);
303        Ok(Some(Self { columns }))
304    }
305
306    /// Returns all declarations, beginning with the required timestamp column.
307    #[must_use]
308    pub fn columns(&self) -> &[ColumnDefinition] {
309        &self.columns
310    }
311
312    /// Finds a declaration by its typed identifier.
313    #[must_use]
314    pub fn column(&self, id: ColumnId) -> Option<&ColumnDefinition> {
315        self.columns.iter().find(|definition| definition.id == id)
316    }
317
318    /// Returns the number of user-declared columns, excluding `ts`.
319    #[must_use]
320    pub fn user_column_count(&self) -> usize {
321        self.columns.len().saturating_sub(1)
322    }
323}
zeppelin_embed/meta/mod.rs

zeppelin_embed/meta/
mod.rs
