timeseries_table_format/metadata/
schema_compat.rs

1//! Schema compatibility helpers (pure metadata).
2//!
3//! v0.1 rule: **no schema evolution**.
4//! Every appended segment must have a [`LogicalSchema`] that matches the table's
5//! canonical schema exactly.
6
7use std::collections::HashMap;
8
9use snafu::prelude::*;
10
11use crate::{
12    coverage::{EntityIdentity, EntityValue},
13    metadata::{
14        logical_schema::{LogicalDataType, LogicalField, LogicalSchema, LogicalSchemaError},
15        table_metadata::{IndexKind, IndexSpec, TableMeta},
16    },
17};
18
19/// Errors raised when a segment's schema is not compatible with the table.
20#[derive(Debug, Snafu)]
21pub enum SchemaCompatibilityError {
22    /// The table does not yet have a canonical logical schema.
23    ///
24    /// Many call sites (like append) may choose to *not* use this and
25    /// instead adopt the first segment's schema, but we keep the error
26    /// available for operations that require a fixed schema.
27    #[snafu(display("Table has no logical_schema; v0.1 cannot append without a canonical schema"))]
28    MissingTableSchema,
29
30    /// The segment is missing a column that exists in the table schema.
31    #[snafu(display("Segment schema is missing required column {column}"))]
32    MissingColumn {
33        /// The name of the missing column.
34        column: String,
35    },
36
37    /// The logical schema does not contain the registered index column.
38    #[snafu(display("Schema is missing registered index column {column}"))]
39    MissingIndexColumn {
40        /// Registered index column name.
41        column: String,
42    },
43
44    /// The logical schema does not contain a configured entity column.
45    #[snafu(display("Schema is missing configured entity column {column}"))]
46    MissingEntityColumn {
47        /// Configured entity column name.
48        column: String,
49    },
50
51    /// A configured entity column has an unsupported logical type.
52    #[snafu(display(
53        "Entity column {column} has unsupported logical type {actual}; expected utf8, int32, int64, or uint64"
54    ))]
55    UnsupportedEntityColumnType {
56        /// Configured entity column name.
57        column: String,
58        /// Unsupported logical type.
59        actual: LogicalDataType,
60    },
61
62    /// A persisted single-entity identity has the wrong component count.
63    #[snafu(display(
64        "Entity identity has {actual} components, but the table configures {expected} entity columns"
65    ))]
66    EntityIdentityArityMismatch {
67        /// Configured entity column count.
68        expected: usize,
69        /// Persisted identity component count.
70        actual: usize,
71    },
72
73    /// A persisted entity component has the wrong scalar type.
74    #[snafu(display(
75        "Entity identity component for column {column} has type {actual}; expected {expected}"
76    ))]
77    EntityIdentityTypeMismatch {
78        /// Configured entity column name.
79        column: String,
80        /// Logical type required by the table schema.
81        expected: LogicalDataType,
82        /// Persisted scalar type.
83        actual: &'static str,
84    },
85
86    /// The segment has an extra column that does not exist in the table schema.
87    #[snafu(display("Segment schema has extra column {column} not present in table schema"))]
88    ExtraColumn {
89        /// The name of the extra column.
90        column: String,
91    },
92
93    /// Column exists in both schemas, but the logical type / nullability differ.
94    #[snafu(display(
95        "Type mismatch for column {column}: table has {table_type}, segment has {segment_type}"
96    ))]
97    TypeMismatch {
98        /// The name of the column with mismatched type.
99        column: String,
100        /// The type in the table schema.
101        table_type: LogicalDataType,
102        /// The type in the segment schema.
103        segment_type: LogicalDataType,
104    },
105
106    /// Specialized version of TypeMismatch for the ordered index column.
107    #[snafu(display(
108        "Index column {column} has incompatible type: table has {table_type}, \
109         segment has {segment_type}"
110    ))]
111    IndexColumnTypeMismatch {
112        /// The name of the ordered index column.
113        column: String,
114        /// The type in the table schema.
115        table_type: LogicalDataType,
116        /// The type in the segment schema.
117        segment_type: LogicalDataType,
118    },
119
120    /// The registered index kind disagrees with the logical schema.
121    #[snafu(display(
122        "Index column {column} has incompatible logical type: expected {expected}, found {actual}"
123    ))]
124    IndexKindMismatch {
125        /// Registered index column name.
126        column: String,
127        /// Expected ordered domain.
128        expected: &'static str,
129        /// Logical type found in the schema.
130        actual: LogicalDataType,
131    },
132
133    /// Logical schema construction or validation failed.
134    #[snafu(display("Logical schema is invalid: {source}"))]
135    LogicalSchema {
136        /// The underlying logical schema error.
137        #[snafu(source)]
138        source: LogicalSchemaError,
139    },
140}
141
142/// A convenience type alias for results of schema compatibility operations.
143pub type SchemaResult<T> = Result<T, SchemaCompatibilityError>;
144
145/// Convenience helper if you want to require a schema to be present.
146pub fn require_table_schema(meta: &TableMeta) -> SchemaResult<&LogicalSchema> {
147    match &meta.logical_schema {
148        Some(schema) => Ok(schema),
149        None => MissingTableSchemaSnafu.fail(),
150    }
151}
152
153fn columns_by_name(schema: &LogicalSchema) -> HashMap<&str, &LogicalField> {
154    schema
155        .columns()
156        .iter()
157        .map(|col| (col.name.as_str(), col))
158        .collect()
159}
160
161/// Validate the registered ordered index and entity columns against a logical schema.
162///
163/// # Errors
164/// Returns [`SchemaCompatibilityError::MissingIndexColumn`] when the column is
165/// absent and [`SchemaCompatibilityError::IndexKindMismatch`] when its logical
166/// type does not match the registered domain. Missing or unsupported entity
167/// columns return their corresponding typed errors.
168pub fn ensure_index_spec_matches_schema(
169    schema: &LogicalSchema,
170    index: &IndexSpec,
171) -> SchemaResult<()> {
172    let field = schema
173        .columns()
174        .iter()
175        .find(|field| field.name == index.column)
176        .ok_or_else(|| SchemaCompatibilityError::MissingIndexColumn {
177            column: index.column.clone(),
178        })?;
179
180    let matches = matches!(
181        (&index.kind, &field.data_type),
182        (
183            IndexKind::Timestamp { .. },
184            LogicalDataType::Timestamp { .. }
185        ) | (IndexKind::Int64 { .. }, LogicalDataType::Int64)
186            | (IndexKind::UInt64 { .. }, LogicalDataType::UInt64)
187    );
188
189    if !matches {
190        return Err(SchemaCompatibilityError::IndexKindMismatch {
191            column: index.column.clone(),
192            expected: index.kind.name(),
193            actual: field.data_type.clone(),
194        });
195    }
196
197    for column in &index.entity_columns {
198        let field = schema
199            .columns()
200            .iter()
201            .find(|field| field.name == *column)
202            .ok_or_else(|| SchemaCompatibilityError::MissingEntityColumn {
203                column: column.clone(),
204            })?;
205        if !matches!(
206            field.data_type,
207            LogicalDataType::Utf8
208                | LogicalDataType::Int32
209                | LogicalDataType::Int64
210                | LogicalDataType::UInt64
211        ) {
212            return Err(SchemaCompatibilityError::UnsupportedEntityColumnType {
213                column: column.clone(),
214                actual: field.data_type.clone(),
215            });
216        }
217    }
218
219    Ok(())
220}
221
222/// Validate a persisted identity against configured entity-column types.
223///
224/// # Errors
225/// Returns an arity or component-type mismatch when the identity cannot belong
226/// to the supplied table schema and index specification.
227pub fn ensure_entity_identity_matches_schema(
228    schema: &LogicalSchema,
229    index: &IndexSpec,
230    identity: &EntityIdentity,
231) -> SchemaResult<()> {
232    if identity.components().len() != index.entity_columns.len() {
233        return Err(SchemaCompatibilityError::EntityIdentityArityMismatch {
234            expected: index.entity_columns.len(),
235            actual: identity.components().len(),
236        });
237    }
238
239    for (column, value) in index.entity_columns.iter().zip(identity.components()) {
240        let field = schema
241            .columns()
242            .iter()
243            .find(|field| field.name == *column)
244            .ok_or_else(|| SchemaCompatibilityError::MissingEntityColumn {
245                column: column.clone(),
246            })?;
247        let matches = matches!(
248            (&field.data_type, value),
249            (LogicalDataType::Utf8, EntityValue::Utf8(_))
250                | (LogicalDataType::Int32, EntityValue::Int32(_))
251                | (LogicalDataType::Int64, EntityValue::Int64(_))
252                | (LogicalDataType::UInt64, EntityValue::UInt64(_))
253        );
254        if !matches {
255            let actual = match value {
256                EntityValue::Utf8(_) => "utf8",
257                EntityValue::Int32(_) => "int32",
258                EntityValue::Int64(_) => "int64",
259                EntityValue::UInt64(_) => "uint64",
260            };
261            return Err(SchemaCompatibilityError::EntityIdentityTypeMismatch {
262                column: column.clone(),
263                expected: field.data_type.clone(),
264                actual,
265            });
266        }
267    }
268
269    Ok(())
270}
271
272/// Enforce the v0.1 "no schema evolution" rule.
273///
274/// - Every table column must appear in the segment schema.
275/// - No extra columns may appear in the segment schema.
276/// - For every column, logical type and nullability must match exactly.
277/// - If the mismatch is on the ordered index column, use a specific error.
278pub fn ensure_schema_exact_match(
279    table_schema: &LogicalSchema,
280    segment_schema: &LogicalSchema,
281    index: &IndexSpec,
282) -> SchemaResult<()> {
283    let index_col_name = index.column.as_str();
284
285    let table_cols = columns_by_name(table_schema);
286    let seg_cols = columns_by_name(segment_schema);
287
288    for (name, table_field) in &table_cols {
289        let seg_field =
290            seg_cols
291                .get(name)
292                .ok_or_else(|| SchemaCompatibilityError::MissingColumn {
293                    column: (*name).to_string(),
294                })?;
295
296        if table_field.data_type != seg_field.data_type
297            || table_field.nullable != seg_field.nullable
298        {
299            let err = if *name == index_col_name {
300                SchemaCompatibilityError::IndexColumnTypeMismatch {
301                    column: (*name).to_string(),
302                    table_type: table_field.data_type.clone(),
303                    segment_type: seg_field.data_type.clone(),
304                }
305            } else {
306                SchemaCompatibilityError::TypeMismatch {
307                    column: (*name).to_string(),
308                    table_type: table_field.data_type.clone(),
309                    segment_type: seg_field.data_type.clone(),
310                }
311            };
312            return Err(err);
313        }
314    }
315
316    for name in seg_cols.keys() {
317        if !table_cols.contains_key(name) {
318            return Err(SchemaCompatibilityError::ExtraColumn {
319                column: (*name).to_string(),
320            });
321        }
322    }
323
324    Ok(())
325}
326
327#[cfg(test)]
328mod tests {
329    use std::num::NonZeroU64;
330
331    use super::*;
332    use crate::metadata::{
333        logical_schema::{LogicalSchema, LogicalTimestampUnit},
334        table_metadata::TimeBucket,
335    };
336
337    fn schema(data_type: LogicalDataType) -> LogicalSchema {
338        LogicalSchema::new(vec![LogicalField {
339            name: "idx".to_string(),
340            data_type,
341            nullable: false,
342        }])
343        .unwrap()
344    }
345
346    fn index(kind: IndexKind) -> IndexSpec {
347        IndexSpec {
348            column: "idx".to_string(),
349            entity_columns: Vec::new(),
350            kind,
351        }
352    }
353
354    fn schema_with_entities(entity_types: Vec<LogicalDataType>) -> LogicalSchema {
355        let mut fields = vec![LogicalField {
356            name: "idx".to_string(),
357            data_type: LogicalDataType::Int64,
358            nullable: false,
359        }];
360        fields.extend(
361            entity_types
362                .into_iter()
363                .enumerate()
364                .map(|(position, data_type)| LogicalField {
365                    name: format!("entity_{position}"),
366                    data_type,
367                    nullable: false,
368                }),
369        );
370        LogicalSchema::new(fields).unwrap()
371    }
372
373    fn entity_index(count: usize) -> IndexSpec {
374        IndexSpec {
375            column: "idx".to_string(),
376            entity_columns: (0..count)
377                .map(|position| format!("entity_{position}"))
378                .collect(),
379            kind: IndexKind::Int64 {
380                bucket_width: NonZeroU64::new(1).unwrap(),
381            },
382        }
383    }
384
385    #[test]
386    fn ordered_index_schema_validation_accepts_each_exact_domain() {
387        let cases = [
388            (
389                index(IndexKind::Timestamp {
390                    bucket: TimeBucket::Seconds(1),
391                    timezone: None,
392                }),
393                schema(LogicalDataType::Timestamp {
394                    unit: LogicalTimestampUnit::Nanos,
395                    timezone: Some("UTC".to_string()),
396                }),
397            ),
398            (
399                index(IndexKind::Int64 {
400                    bucket_width: NonZeroU64::new(1).unwrap(),
401                }),
402                schema(LogicalDataType::Int64),
403            ),
404            (
405                index(IndexKind::UInt64 {
406                    bucket_width: NonZeroU64::new(1).unwrap(),
407                }),
408                schema(LogicalDataType::UInt64),
409            ),
410        ];
411
412        for (index, schema) in cases {
413            ensure_index_spec_matches_schema(&schema, &index).unwrap();
414        }
415    }
416
417    #[test]
418    fn ordered_index_schema_validation_rejects_missing_and_wrong_domains() {
419        let unsigned = index(IndexKind::UInt64 {
420            bucket_width: NonZeroU64::new(1).unwrap(),
421        });
422        let missing = LogicalSchema::new(vec![LogicalField {
423            name: "other".to_string(),
424            data_type: LogicalDataType::UInt64,
425            nullable: false,
426        }])
427        .unwrap();
428
429        assert!(matches!(
430            ensure_index_spec_matches_schema(&missing, &unsigned),
431            Err(SchemaCompatibilityError::MissingIndexColumn { .. })
432        ));
433        assert!(matches!(
434            ensure_index_spec_matches_schema(&schema(LogicalDataType::Int64), &unsigned),
435            Err(SchemaCompatibilityError::IndexKindMismatch {
436                expected: "uint64",
437                actual: LogicalDataType::Int64,
438                ..
439            })
440        ));
441    }
442
443    #[test]
444    fn entity_schema_validation_accepts_only_supported_types() {
445        let supported = vec![
446            LogicalDataType::Utf8,
447            LogicalDataType::Int32,
448            LogicalDataType::Int64,
449            LogicalDataType::UInt64,
450        ];
451        ensure_index_spec_matches_schema(&schema_with_entities(supported), &entity_index(4))
452            .unwrap();
453
454        let missing = ensure_index_spec_matches_schema(
455            &schema_with_entities(vec![LogicalDataType::Utf8]),
456            &entity_index(2),
457        )
458        .unwrap_err();
459        assert!(matches!(
460            missing,
461            SchemaCompatibilityError::MissingEntityColumn { column }
462                if column == "entity_1"
463        ));
464
465        let unsupported = ensure_index_spec_matches_schema(
466            &schema_with_entities(vec![LogicalDataType::Bool]),
467            &entity_index(1),
468        )
469        .unwrap_err();
470        assert!(matches!(
471            unsupported,
472            SchemaCompatibilityError::UnsupportedEntityColumnType {
473                column,
474                actual: LogicalDataType::Bool,
475            } if column == "entity_0"
476        ));
477    }
478
479    #[test]
480    fn persisted_entity_identity_must_match_schema_types_and_arity() {
481        let schema = schema_with_entities(vec![
482            LogicalDataType::Utf8,
483            LogicalDataType::Int32,
484            LogicalDataType::Int64,
485            LogicalDataType::UInt64,
486        ]);
487        let index = entity_index(4);
488        let identity = EntityIdentity::try_new(vec![
489            EntityValue::from("device"),
490            EntityValue::Int32(-1),
491            EntityValue::Int64(i64::MIN),
492            EntityValue::UInt64(u64::MAX),
493        ])
494        .unwrap();
495        ensure_entity_identity_matches_schema(&schema, &index, &identity).unwrap();
496
497        let wrong_type = EntityIdentity::try_new(vec![
498            EntityValue::from("device"),
499            EntityValue::UInt64(1),
500            EntityValue::Int64(2),
501            EntityValue::UInt64(3),
502        ])
503        .unwrap();
504        assert!(matches!(
505            ensure_entity_identity_matches_schema(&schema, &index, &wrong_type),
506            Err(SchemaCompatibilityError::EntityIdentityTypeMismatch {
507                column,
508                expected: LogicalDataType::Int32,
509                actual: "uint64",
510            }) if column == "entity_1"
511        ));
512
513        let too_short = EntityIdentity::try_new(vec![EntityValue::from("device")]).unwrap();
514        assert!(matches!(
515            ensure_entity_identity_matches_schema(&schema, &index, &too_short),
516            Err(SchemaCompatibilityError::EntityIdentityArityMismatch {
517                expected: 4,
518                actual: 1,
519            })
520        ));
521    }
522}
timeseries_table_format/metadata/schema_compat.rs

timeseries_table_format/metadata/
schema_compat.rs
