timeseries_table_format/metadata/
logical_schema.rs

1//! Logical schema definitions and validation for table metadata.
2//!
3//! This module models logical fields and data types stored in the transaction
4//! log, along with validation and conversion to Arrow schemas.
5use std::{collections::HashSet, fmt, sync::Arc};
6
7use arrow::datatypes::{DataType, Field, FieldRef, Fields, Schema, SchemaRef, TimeUnit};
8
9use serde::{Deserialize, Serialize};
10use snafu::prelude::*;
11
12/// Units for logical timestamps recorded in the table metadata.
13#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
14pub enum LogicalTimestampUnit {
15    /// Millisecond precision timestamps.
16    Millis,
17    /// Microsecond precision timestamps.
18    Micros,
19    /// Nanosecond precision timestamps.
20    Nanos,
21}
22
23impl LogicalTimestampUnit {
24    fn to_arrow_time_unit(self) -> TimeUnit {
25        match self {
26            LogicalTimestampUnit::Millis => TimeUnit::Millisecond,
27            LogicalTimestampUnit::Micros => TimeUnit::Microsecond,
28            LogicalTimestampUnit::Nanos => TimeUnit::Nanosecond,
29        }
30    }
31}
32
33impl fmt::Display for LogicalTimestampUnit {
34    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
35        match self {
36            LogicalTimestampUnit::Millis => write!(f, "ms"),
37            LogicalTimestampUnit::Micros => write!(f, "us"),
38            LogicalTimestampUnit::Nanos => write!(f, "ns"),
39        }
40    }
41}
42
43/// Logical column definition in a schema.
44#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
45pub struct LogicalField {
46    /// Column name as stored in the schema.
47    pub name: String,
48    /// Logical data type for the column.
49    pub data_type: LogicalDataType,
50    /// Whether the column allows null values.
51    pub nullable: bool,
52}
53
54impl LogicalField {
55    fn to_arrow_field_ref(&self, path: &str) -> Result<FieldRef, SchemaConvertError> {
56        let dt = self.data_type.to_arrow_datatype(path)?;
57        Ok(Arc::new(Field::new(self.name.clone(), dt, self.nullable)))
58    }
59}
60
61impl fmt::Display for LogicalField {
62    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
63        if self.nullable {
64            write!(f, "{}?: {}", self.name, self.data_type)
65        } else {
66            write!(f, "{}: {}", self.name, self.data_type)
67        }
68    }
69}
70
71fn join_path(parent: &str, child: &str) -> String {
72    if parent.is_empty() {
73        child.to_string()
74    } else {
75        format!("{parent}.{child}")
76    }
77}
78
79/// Logical data types that can be stored in the table schema metadata.
80#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
81pub enum LogicalDataType {
82    /// Boolean value.
83    Bool,
84    /// 32-bit signed integer.
85    Int32,
86    /// 64-bit signed integer.
87    Int64,
88    /// 64-bit unsigned integer.
89    UInt64,
90    /// 32-bit floating point.
91    Float32,
92    /// 64-bit floating point.
93    Float64,
94    /// Variable-length binary data.
95    Binary,
96    /// Fixed-length binary data.
97    FixedBinary {
98        /// Fixed byte width for each value (in bytes).
99        byte_width: i32,
100    },
101    /// UTF-8 encoded string.
102    Utf8,
103    /// Legacy 96-bit integer (primarily for Parquet compatibility).
104    Int96,
105
106    /// Timestamp value with a precision unit and optional timezone.
107    Timestamp {
108        /// Timestamp precision unit (millis, micros, nanos).
109        unit: LogicalTimestampUnit,
110        /// Optional IANA timezone identifier.
111        timezone: Option<String>, // keep Option for future TZ support
112    },
113
114    /// Fixed-precision decimal value with declared precision and scale.
115    Decimal {
116        /// Total number of decimal digits (both sides of the decimal point).
117        precision: i32,
118        /// Number of digits to the right of the decimal point.
119        scale: i32,
120    },
121
122    /// Struct with named child fields.
123    Struct {
124        /// Ordered set of child fields for the struct.
125        fields: Vec<LogicalField>,
126    },
127
128    /// List (array) with a single element field definition.
129    List {
130        /// Element field definition for list items.
131        elements: Box<LogicalField>,
132    },
133
134    /// Map with key/value field definitions.
135    /// If `value` is None, this represents Parquet MAP "keys-only" semantics (set of keys).
136    Map {
137        /// Key field definition (must be non-nullable for Arrow compatibility).
138        key: Box<LogicalField>,
139        /// Value field definition.
140        value: Option<Box<LogicalField>>,
141        /// Whether entries are sorted by key.
142        keys_sorted: bool,
143    },
144
145    /// Catch-all logical data type referenced by name.
146    Other(String),
147}
148
149impl LogicalDataType {
150    fn to_arrow_datatype(&self, column: &str) -> Result<DataType, SchemaConvertError> {
151        Ok(match self {
152            LogicalDataType::Bool => DataType::Boolean,
153            LogicalDataType::Int32 => DataType::Int32,
154            LogicalDataType::Int64 => DataType::Int64,
155            LogicalDataType::UInt64 => DataType::UInt64,
156            LogicalDataType::Float32 => DataType::Float32,
157            LogicalDataType::Float64 => DataType::Float64,
158            LogicalDataType::Binary => DataType::Binary,
159            LogicalDataType::Utf8 => DataType::Utf8,
160
161            LogicalDataType::FixedBinary { byte_width } => {
162                if *byte_width <= 0 {
163                    return Err(SchemaConvertError::FixedBinaryInvalidWidth {
164                        column: column.to_string(),
165                        byte_width: *byte_width,
166                    });
167                }
168                DataType::FixedSizeBinary(*byte_width)
169            }
170
171            LogicalDataType::Timestamp { unit, timezone } => {
172                let tz: Option<Arc<str>> = timezone.as_ref().map(|s| Arc::<str>::from(s.as_str()));
173                DataType::Timestamp(unit.to_arrow_time_unit(), tz)
174            }
175
176            LogicalDataType::Int96 => {
177                return Err(SchemaConvertError::Int96Unsupported {
178                    column: column.to_string(),
179                });
180            }
181
182            LogicalDataType::Decimal { precision, scale } => {
183                let precision = *precision;
184                let scale = *scale;
185                if precision <= 0 {
186                    return Err(SchemaConvertError::DecimalInvalid {
187                        column: column.to_string(),
188                        precision,
189                        scale,
190                        details: "precision must be > 0".to_string(),
191                    });
192                }
193                if scale < 0 {
194                    return Err(SchemaConvertError::DecimalInvalid {
195                        column: column.to_string(),
196                        precision,
197                        scale,
198                        details: "scale must be >= 0".to_string(),
199                    });
200                }
201                if scale > precision {
202                    return Err(SchemaConvertError::DecimalInvalid {
203                        column: column.to_string(),
204                        precision,
205                        scale,
206                        details: "scale must be <= precision".to_string(),
207                    });
208                }
209
210                if precision <= 38 {
211                    DataType::Decimal128(precision as u8, scale as i8)
212                } else if precision <= 76 {
213                    DataType::Decimal256(precision as u8, scale as i8)
214                } else {
215                    return Err(SchemaConvertError::DecimalInvalid {
216                        column: column.to_string(),
217                        precision,
218                        scale,
219                        details: "precision exceeds Arrow maximum (76 digits)".to_string(),
220                    });
221                }
222            }
223
224            LogicalDataType::Struct { fields } => {
225                let mut arrow_children: Vec<FieldRef> = Vec::with_capacity(fields.len());
226                for f in fields {
227                    let child_path = join_path(column, &f.name);
228                    arrow_children.push(f.to_arrow_field_ref(&child_path)?);
229                }
230                DataType::Struct(Fields::from(arrow_children))
231            }
232
233            LogicalDataType::List { elements } => {
234                let child_path = join_path(column, &elements.name);
235                let element_field = elements.to_arrow_field_ref(&child_path)?;
236                DataType::List(element_field)
237            }
238
239            LogicalDataType::Map {
240                key,
241                value,
242                keys_sorted,
243            } => {
244                if key.nullable {
245                    return Err(SchemaConvertError::MapKeyMustBeNonNull {
246                        column: column.to_string(),
247                    });
248                }
249
250                // Canonical Arrow Map field names are "entries", "key", "value"
251                let key_path = format!("{column}.key");
252                let val_path = format!("{column}.value");
253
254                let key_dt = key.data_type.to_arrow_datatype(&key_path)?;
255
256                let (val_dt, val_nullable) = match value.as_deref() {
257                    Some(v) => (v.data_type.to_arrow_datatype(&val_path)?, v.nullable),
258                    None => (DataType::Null, true),
259                };
260
261                let key_field: FieldRef = Arc::new(Field::new("key", key_dt, false));
262                let val_field: FieldRef = Arc::new(Field::new("value", val_dt, val_nullable));
263
264                let entries_dt = DataType::Struct(Fields::from(vec![key_field, val_field]));
265                let entries_field: FieldRef = Arc::new(Field::new("entries", entries_dt, false));
266
267                DataType::Map(entries_field, *keys_sorted)
268            }
269
270            LogicalDataType::Other(name) => {
271                return Err(SchemaConvertError::OtherTypeUnsupported {
272                    column: column.to_string(),
273                    name: name.clone(),
274                });
275            }
276        })
277    }
278}
279
280impl fmt::Display for LogicalDataType {
281    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
282        match self {
283            LogicalDataType::Bool => write!(f, "bool"),
284            LogicalDataType::Int32 => write!(f, "int32"),
285            LogicalDataType::Int64 => write!(f, "int64"),
286            LogicalDataType::UInt64 => write!(f, "uint64"),
287            LogicalDataType::Float32 => write!(f, "float32"),
288            LogicalDataType::Float64 => write!(f, "float64"),
289            LogicalDataType::Binary => write!(f, "binary"),
290            LogicalDataType::FixedBinary { byte_width } => write!(f, "fixed_binary[{byte_width}]"),
291            LogicalDataType::Utf8 => write!(f, "utf8"),
292            LogicalDataType::Int96 => write!(f, "int96"),
293
294            LogicalDataType::Timestamp { unit, timezone } => match timezone {
295                Some(tz) => write!(f, "timestamp[{}]({})", unit, tz),
296                None => write!(f, "timestamp[{}]", unit),
297            },
298
299            LogicalDataType::Decimal { precision, scale } => {
300                write!(f, "decimal(precision={precision}, scale={scale})")
301            }
302
303            LogicalDataType::Struct { fields } => {
304                write!(f, "Struct{{")?;
305                for (i, field) in fields.iter().enumerate() {
306                    if i > 0 {
307                        write!(f, ", ")?;
308                    }
309                    write!(f, "{}", field)?;
310                }
311                write!(f, "}}")
312            }
313
314            LogicalDataType::List { elements } => {
315                write!(f, "List<{}>", elements)
316            }
317
318            LogicalDataType::Map {
319                key,
320                value,
321                keys_sorted,
322            } => match value.as_deref() {
323                Some(v) => write!(f, "Map<{}, {}, keys_sorted={}>", key, v, keys_sorted),
324                None => write!(
325                    f,
326                    "Map<{}, value=omitted, keys_sorted={}>",
327                    key, keys_sorted
328                ),
329            },
330
331            LogicalDataType::Other(s) => write!(f, "{s}"),
332        }
333    }
334}
335
336/// Logical schema metadata describing the ordered collection of logical columns.
337#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
338pub struct LogicalSchema {
339    /// All logical columns that compose the schema in their defined order.
340    columns: Vec<LogicalField>,
341}
342
343impl LogicalSchema {
344    /// Convert this logical schema to an owned Arrow [`Schema`].
345    ///
346    /// Fails if any column uses a logical type that cannot be represented in
347    /// Arrow (see [`SchemaConvertError`]).
348    pub fn to_arrow_schema(&self) -> Result<Schema, SchemaConvertError> {
349        let mut fields = Vec::with_capacity(self.columns.len());
350        for c in &self.columns {
351            let fref = c.to_arrow_field_ref(&c.name)?;
352            fields.push(fref.as_ref().clone());
353        }
354
355        Ok(Schema::new(fields))
356    }
357
358    /// Convert this logical schema to a shared Arrow [`SchemaRef`].
359    ///
360    /// This is a convenience wrapper around [`Self::to_arrow_schema`].
361    pub fn to_arrow_schema_ref(&self) -> Result<SchemaRef, SchemaConvertError> {
362        Ok(Arc::new(self.to_arrow_schema()?))
363    }
364}
365
366/// Errors that can occur while constructing or validating a logical schema.
367#[derive(Debug, Clone, Snafu, PartialEq, Eq)]
368pub enum LogicalSchemaError {
369    /// Duplicate column names are not allowed.
370    #[snafu(display("Duplicate column name: {column}"))]
371    DuplicateColumn {
372        /// The duplicate column name.
373        column: String,
374    },
375
376    /// FixedBinary columns must include a positive byte width.
377    #[snafu(display(
378        "invalid FixedBinary byte_width for column '{column}': {byte_width} (must be > 0)"
379    ))]
380    FixedBinaryInvalidWidthInSchema {
381        /// Column name that failed validation.
382        column: String,
383        /// Declared byte width.
384        byte_width: i32,
385    },
386
387    /// Parquet FIXED_LEN_BYTE_ARRAY columns must include a type_length.
388    #[snafu(display(
389        "FIXED_LEN_BYTE_ARRAY column '{column}' missing type_length in Parquet schema"
390    ))]
391    FixedBinaryMissingLength {
392        /// Column name that failed validation.
393        column: String,
394    },
395
396    /// Duplicate field names within a struct are not allowed.
397    #[snafu(display("Duplicate field name: column={column_path}, field={field}"))]
398    DuplicatedFieldName {
399        /// Column path for the struct that contains the duplicate field.
400        column_path: String,
401        /// Duplicate field name.
402        field: String,
403    },
404
405    /// Map key fields must be non-nullable in schema validation.
406    #[snafu(display("Invalid Map Key: map key should not be null for column={column_path}"))]
407    InvalidMapKeyNullability {
408        /// Column path for the map with an invalid key nullability.
409        column_path: String,
410    },
411
412    /// Struct fields must be non-empty.
413    #[snafu(display("Struct must have at least one field: column={column_path}"))]
414    EmptyStruct {
415        /// Column path for the empty struct.
416        column_path: String,
417    },
418
419    /// List element fields must have a non-empty name.
420    #[snafu(display("List element field name must be non-empty: column={column_path}"))]
421    ListElementNameEmpty {
422        /// Column path for the list with an empty element name.
423        column_path: String,
424    },
425
426    /// Struct fields must have a non-empty name.
427    #[snafu(display("Struct field name must be non-empty: column={column_path}, field={field}"))]
428    StructFieldNameEmpty {
429        /// Column path for the struct with an empty field name.
430        column_path: String,
431        /// Empty field name.
432        field: String,
433    },
434
435    /// Parquet LIST encoding does not match the supported layout.
436    #[snafu(display("Unsupported Parquet LIST encoding: column={column_path}, details={details}"))]
437    UnsupportedParquetListEncoding {
438        /// Column path for the list with unsupported encoding.
439        column_path: String,
440        /// Details describing why the LIST encoding is unsupported.
441        details: String,
442    },
443
444    /// Parquet MAP encoding does not match the supported layout.
445    #[snafu(display("Unsupported Parquet MAP encoding: column={column_path}, details={details}"))]
446    UnsupportedParquetMapEncoding {
447        /// Column path for the map with unsupported encoding.
448        column_path: String,
449        /// Details describing why the MAP encoding is unsupported.
450        details: String,
451    },
452}
453
454impl LogicalSchema {
455    /// Construct a validated logical schema (rejects duplicate column names).
456    pub fn new(columns: Vec<LogicalField>) -> Result<Self, LogicalSchemaError> {
457        let mut seen = HashSet::new();
458        for col in &columns {
459            if !seen.insert(col.name.clone()) {
460                return DuplicateColumnSnafu {
461                    column: col.name.clone(),
462                }
463                .fail();
464            }
465            validate_field(col, &col.name)?;
466        }
467
468        Ok(Self { columns })
469    }
470
471    /// Borrow the logical columns.
472    pub fn columns(&self) -> &[LogicalField] {
473        &self.columns
474    }
475}
476
477fn validate_field(field: &LogicalField, path: &str) -> Result<(), LogicalSchemaError> {
478    validate_dtype(&field.data_type, path)
479}
480
481fn validate_dtype(dt: &LogicalDataType, path: &str) -> Result<(), LogicalSchemaError> {
482    match dt {
483        LogicalDataType::FixedBinary { byte_width } => {
484            if *byte_width <= 0 {
485                return Err(LogicalSchemaError::FixedBinaryInvalidWidthInSchema {
486                    column: path.to_string(),
487                    byte_width: *byte_width,
488                });
489            }
490            Ok(())
491        }
492
493        LogicalDataType::Struct { fields } => {
494            if fields.is_empty() {
495                return Err(LogicalSchemaError::EmptyStruct {
496                    column_path: path.to_string(),
497                });
498            }
499
500            let mut seen = HashSet::with_capacity(fields.len());
501            for child in fields {
502                if child.name.trim().is_empty() {
503                    return Err(LogicalSchemaError::StructFieldNameEmpty {
504                        column_path: path.to_string(),
505                        field: child.name.clone(),
506                    });
507                }
508
509                if !seen.insert(child.name.clone()) {
510                    return Err(LogicalSchemaError::DuplicatedFieldName {
511                        column_path: path.to_string(),
512                        field: child.name.clone(),
513                    });
514                }
515                let child_path = format!("{}.{}", path, child.name);
516                validate_field(child, &child_path)?;
517            }
518            Ok(())
519        }
520
521        LogicalDataType::List { elements } => {
522            if elements.name.trim().is_empty() {
523                return Err(LogicalSchemaError::ListElementNameEmpty {
524                    column_path: path.to_string(),
525                });
526            }
527            let child_path = format!("{}.{}", path, elements.name);
528            validate_field(elements, &child_path)
529        }
530
531        LogicalDataType::Map { key, value, .. } => {
532            if key.nullable {
533                return Err(LogicalSchemaError::InvalidMapKeyNullability {
534                    column_path: path.to_string(),
535                });
536            }
537            validate_field(key, &format!("{}.key", path))?;
538            if let Some(v) = value.as_deref() {
539                validate_field(v, &format!("{}.value", path))?;
540            }
541
542            Ok(())
543        }
544
545        _ => Ok(()),
546    }
547}
548
549/// Errors encountered while converting between logical schema representations.
550#[derive(Debug, Snafu)]
551pub enum SchemaConvertError {
552    /// The logical type is not supported by the target representation.
553    #[snafu(display("unsupported logical type for column '{column}': {type_name} ({details})"))]
554    UnsupportedLogicalType {
555        /// Column name that failed conversion.
556        column: String,
557        /// High-level type name (for diagnostics).
558        type_name: String,
559        /// Additional details describing why it is unsupported.
560        details: String,
561    },
562
563    /// FixedBinary fields must declare a positive byte width.
564    #[snafu(display(
565        "invalid FixedBinary byte_width for column '{column}': {byte_width} (must be > 0)"
566    ))]
567    FixedBinaryInvalidWidth {
568        /// Column name that failed validation.
569        column: String,
570        /// Declared byte width.
571        byte_width: i32,
572    },
573
574    /// Int96 is rejected for now to avoid legacy timestamp ambiguity.
575    #[snafu(display("Int96 is not supported in v0.1 for column '{column}'"))]
576    Int96Unsupported {
577        /// Column name that failed conversion.
578        column: String,
579    },
580
581    /// Catch-all "Other" types are not accepted in v0.1.
582    #[snafu(display("Other type '{name}' is not supported in v0.1 for column '{column}'"))]
583    OtherTypeUnsupported {
584        /// Column name that failed conversion.
585        column: String,
586        /// Type name reported by the source.
587        name: String,
588    },
589
590    /// Decimal precision/scale is out of supported bounds for Arrow conversion.
591    #[snafu(display(
592        "invalid decimal definition for column '{column}': precision={precision}, scale={scale} ({details})"
593    ))]
594    DecimalInvalid {
595        /// Column name that failed conversion.
596        column: String,
597        /// Declared total precision.
598        precision: i32,
599        /// Declared scale (digits to the right of the decimal point).
600        scale: i32,
601        /// Human-readable details describing the constraint violation.
602        details: String,
603    },
604
605    /// Map keys must be non-nullable when converting to Arrow.
606    #[snafu(display("map key must be non-nullable for column '{column}'"))]
607    MapKeyMustBeNonNull {
608        /// Column name that failed conversion.
609        column: String,
610    },
611}
612
613#[cfg(test)]
614mod tests {
615    use super::*;
616
617    fn sample_logical_schema_all_supported() -> LogicalSchema {
618        LogicalSchema::new(vec![
619            LogicalField {
620                name: "flag".to_string(),
621                data_type: LogicalDataType::Bool,
622                nullable: false,
623            },
624            LogicalField {
625                name: "i32".to_string(),
626                data_type: LogicalDataType::Int32,
627                nullable: false,
628            },
629            LogicalField {
630                name: "i64".to_string(),
631                data_type: LogicalDataType::Int64,
632                nullable: true,
633            },
634            LogicalField {
635                name: "f32".to_string(),
636                data_type: LogicalDataType::Float32,
637                nullable: false,
638            },
639            LogicalField {
640                name: "f64".to_string(),
641                data_type: LogicalDataType::Float64,
642                nullable: true,
643            },
644            LogicalField {
645                name: "text".to_string(),
646                data_type: LogicalDataType::Utf8,
647                nullable: true,
648            },
649            LogicalField {
650                name: "bytes".to_string(),
651                data_type: LogicalDataType::Binary,
652                nullable: true,
653            },
654            LogicalField {
655                name: "fixed".to_string(),
656                data_type: LogicalDataType::FixedBinary { byte_width: 16 },
657                nullable: false,
658            },
659            LogicalField {
660                name: "ts".to_string(),
661                data_type: LogicalDataType::Timestamp {
662                    unit: LogicalTimestampUnit::Micros,
663                    timezone: Some("UTC".to_string()),
664                },
665                nullable: false,
666            },
667        ])
668        .expect("valid logical schema")
669    }
670
671    #[test]
672    fn logical_schema_to_arrow_schema_happy_path() {
673        let logical = sample_logical_schema_all_supported();
674        let schema = logical.to_arrow_schema().expect("arrow schema conversion");
675
676        let expected = Schema::new(vec![
677            Field::new("flag", DataType::Boolean, false),
678            Field::new("i32", DataType::Int32, false),
679            Field::new("i64", DataType::Int64, true),
680            Field::new("f32", DataType::Float32, false),
681            Field::new("f64", DataType::Float64, true),
682            Field::new("text", DataType::Utf8, true),
683            Field::new("bytes", DataType::Binary, true),
684            Field::new("fixed", DataType::FixedSizeBinary(16), false),
685            Field::new(
686                "ts",
687                DataType::Timestamp(TimeUnit::Microsecond, Some(Arc::<str>::from("UTC"))),
688                false,
689            ),
690        ]);
691
692        assert_eq!(schema, expected);
693    }
694
695    #[test]
696    fn logical_schema_rejects_fixed_binary_invalid_width() {
697        for width in [0, -1] {
698            let err = LogicalSchema::new(vec![LogicalField {
699                name: "bad_fixed".to_string(),
700                data_type: LogicalDataType::FixedBinary { byte_width: width },
701                nullable: false,
702            }])
703            .expect_err("expected invalid schema to be rejected");
704
705            assert!(
706                matches!(
707                    &err,
708                    LogicalSchemaError::FixedBinaryInvalidWidthInSchema {
709                        column,
710                        byte_width
711                    } if column == "bad_fixed" && *byte_width == width
712                ),
713                "unexpected error: {err:?}"
714            );
715        }
716    }
717
718    #[test]
719    fn logical_schema_rejects_int96() {
720        let logical = LogicalSchema::new(vec![LogicalField {
721            name: "legacy_ts".to_string(),
722            data_type: LogicalDataType::Int96,
723            nullable: false,
724        }])
725        .expect("valid schema structure");
726
727        let err = logical.to_arrow_schema().unwrap_err();
728        assert!(
729            matches!(
730                &err,
731                SchemaConvertError::Int96Unsupported { column } if column == "legacy_ts"
732            ),
733            "unexpected error: {err:?}"
734        );
735    }
736
737    #[test]
738    fn logical_schema_map_entries_field_is_non_nullable() {
739        let logical = LogicalSchema::new(vec![LogicalField {
740            name: "attrs".to_string(),
741            data_type: LogicalDataType::Map {
742                key: Box::new(LogicalField {
743                    name: "key".to_string(),
744                    data_type: LogicalDataType::Utf8,
745                    nullable: false,
746                }),
747                value: Some(Box::new(LogicalField {
748                    name: "value".to_string(),
749                    data_type: LogicalDataType::Int64,
750                    nullable: true,
751                })),
752                keys_sorted: false,
753            },
754            nullable: true,
755        }])
756        .expect("valid schema");
757
758        let schema = logical.to_arrow_schema().expect("arrow schema conversion");
759        let field = schema.field(0);
760        let DataType::Map(entries_field, _) = field.data_type() else {
761            panic!("expected map type, got {:?}", field.data_type());
762        };
763        assert!(
764            !entries_field.is_nullable(),
765            "map entries field should be non-nullable"
766        );
767    }
768
769    #[test]
770    fn logical_schema_map_value_none_maps_to_null_field() {
771        let logical = LogicalSchema::new(vec![LogicalField {
772            name: "attrs".to_string(),
773            data_type: LogicalDataType::Map {
774                key: Box::new(LogicalField {
775                    name: "key".to_string(),
776                    data_type: LogicalDataType::Utf8,
777                    nullable: false,
778                }),
779                value: None,
780                keys_sorted: false,
781            },
782            nullable: false,
783        }])
784        .expect("valid schema");
785
786        let schema = logical.to_arrow_schema().expect("arrow schema conversion");
787        let field = schema.field(0);
788        let DataType::Map(entries_field, _) = field.data_type() else {
789            panic!("expected map type, got {:?}", field.data_type());
790        };
791        let DataType::Struct(fields) = entries_field.data_type() else {
792            panic!(
793                "expected entries struct, got {:?}",
794                entries_field.data_type()
795            );
796        };
797        let value_field = fields
798            .iter()
799            .find(|f| f.name() == "value")
800            .expect("value field");
801        assert!(
802            matches!(value_field.data_type(), DataType::Null) && value_field.is_nullable(),
803            "value field should be Null and nullable"
804        );
805    }
806
807    #[test]
808    fn logical_schema_rejects_empty_struct_field_name() {
809        let err = LogicalSchema::new(vec![LogicalField {
810            name: "root".to_string(),
811            data_type: LogicalDataType::Struct {
812                fields: vec![LogicalField {
813                    name: "".to_string(),
814                    data_type: LogicalDataType::Int32,
815                    nullable: false,
816                }],
817            },
818            nullable: false,
819        }])
820        .expect_err("expected invalid schema");
821
822        assert!(
823            matches!(
824                &err,
825                LogicalSchemaError::StructFieldNameEmpty { column_path, field }
826                if column_path == "root" && field.is_empty()
827            ),
828            "unexpected error: {err:?}"
829        );
830    }
831
832    #[test]
833    fn logical_schema_rejects_other_type() {
834        let logical = LogicalSchema::new(vec![LogicalField {
835            name: "opaque".to_string(),
836            data_type: LogicalDataType::Other("parquet::Map".to_string()),
837            nullable: true,
838        }])
839        .expect("valid schema structure");
840
841        let err = logical.to_arrow_schema().unwrap_err();
842        assert!(
843            matches!(
844                &err,
845                SchemaConvertError::OtherTypeUnsupported { column, name }
846                    if column == "opaque" && name == "parquet::Map"
847            ),
848            "unexpected error: {err:?}"
849        );
850    }
851
852    #[test]
853    fn logical_schema_timestamp_without_timezone() {
854        let logical = LogicalSchema::new(vec![LogicalField {
855            name: "ts".to_string(),
856            data_type: LogicalDataType::Timestamp {
857                unit: LogicalTimestampUnit::Millis,
858                timezone: None,
859            },
860            nullable: false,
861        }])
862        .expect("valid schema structure");
863
864        let schema = logical.to_arrow_schema().expect("arrow schema conversion");
865        let expected = Schema::new(vec![Field::new(
866            "ts",
867            DataType::Timestamp(TimeUnit::Millisecond, None),
868            false,
869        )]);
870        assert_eq!(schema, expected);
871    }
872
873    #[test]
874    fn logical_schema_decimal_conversion_bounds() {
875        let valid_128 = LogicalSchema::new(vec![LogicalField {
876            name: "dec128".to_string(),
877            data_type: LogicalDataType::Decimal {
878                precision: 38,
879                scale: 10,
880            },
881            nullable: false,
882        }])
883        .expect("valid schema structure");
884        let schema = valid_128
885            .to_arrow_schema()
886            .expect("arrow schema conversion");
887        assert_eq!(
888            schema,
889            Schema::new(vec![Field::new(
890                "dec128",
891                DataType::Decimal128(38, 10),
892                false
893            )])
894        );
895
896        let valid_256 = LogicalSchema::new(vec![LogicalField {
897            name: "dec256".to_string(),
898            data_type: LogicalDataType::Decimal {
899                precision: 76,
900                scale: 5,
901            },
902            nullable: false,
903        }])
904        .expect("valid schema structure");
905        let schema = valid_256
906            .to_arrow_schema()
907            .expect("arrow schema conversion");
908        assert_eq!(
909            schema,
910            Schema::new(vec![Field::new(
911                "dec256",
912                DataType::Decimal256(76, 5),
913                false
914            )])
915        );
916
917        let invalid = LogicalSchema::new(vec![LogicalField {
918            name: "dec_too_large".to_string(),
919            data_type: LogicalDataType::Decimal {
920                precision: 77,
921                scale: 0,
922            },
923            nullable: false,
924        }])
925        .expect("valid schema structure");
926        let err = invalid.to_arrow_schema().unwrap_err();
927        assert!(
928            matches!(
929                &err,
930                SchemaConvertError::DecimalInvalid { column, precision, scale, .. }
931                    if column == "dec_too_large" && *precision == 77 && *scale == 0
932            ),
933            "unexpected error: {err:?}"
934        );
935    }
936
937    #[test]
938    fn logical_schema_decimal_validation_errors() {
939        let cases = vec![
940            ("dec_precision_zero", 0, 0, "precision must be > 0"),
941            ("dec_scale_negative", 10, -1, "scale must be >= 0"),
942            ("dec_scale_gt_precision", 4, 5, "scale must be <= precision"),
943        ];
944
945        for (name, precision, scale, details_substr) in cases {
946            let logical = LogicalSchema::new(vec![LogicalField {
947                name: name.to_string(),
948                data_type: LogicalDataType::Decimal { precision, scale },
949                nullable: false,
950            }])
951            .expect("valid schema structure");
952
953            let err = logical.to_arrow_schema().unwrap_err();
954            assert!(
955                matches!(
956                    &err,
957                    SchemaConvertError::DecimalInvalid { column, precision: p, scale: s, details }
958                        if column == name && *p == precision && *s == scale && details.contains(details_substr)
959                ),
960                "unexpected error: {err:?}"
961            );
962        }
963    }
964
965    #[test]
966    fn logical_schema_fixed_binary_json_roundtrip() {
967        let logical = LogicalSchema::new(vec![LogicalField {
968            name: "fixed".to_string(),
969            data_type: LogicalDataType::FixedBinary { byte_width: 8 },
970            nullable: false,
971        }])
972        .expect("valid schema structure");
973
974        let json = serde_json::to_string(&logical).unwrap();
975        let back: LogicalSchema = serde_json::from_str(&json).unwrap();
976        assert_eq!(back, logical);
977    }
978
979    #[test]
980    fn logical_schema_decimal_json_roundtrip() {
981        let logical = LogicalSchema::new(vec![LogicalField {
982            name: "amount".to_string(),
983            data_type: LogicalDataType::Decimal {
984                precision: 18,
985                scale: 4,
986            },
987            nullable: true,
988        }])
989        .expect("valid schema structure");
990
991        let json = serde_json::to_string(&logical).unwrap();
992        let back: LogicalSchema = serde_json::from_str(&json).unwrap();
993        assert_eq!(back, logical);
994    }
995
996    #[test]
997    fn logical_schema_uint64_roundtrips_json_and_maps_exactly_to_arrow() {
998        let schema = LogicalSchema::new(vec![LogicalField {
999            name: "offset".to_string(),
1000            data_type: LogicalDataType::UInt64,
1001            nullable: false,
1002        }])
1003        .unwrap();
1004
1005        let json = serde_json::to_string(&schema).unwrap();
1006        let restored: LogicalSchema = serde_json::from_str(&json).unwrap();
1007        assert_eq!(restored, schema);
1008        assert_eq!(
1009            restored.to_arrow_schema().unwrap().field(0).data_type(),
1010            &DataType::UInt64
1011        );
1012    }
1013}
timeseries_table_format/metadata/logical_schema.rs

timeseries_table_format/metadata/
logical_schema.rs
