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

timeseries_table_format/metadata/
schema_compat.rs
