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::{Backtrace, 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, LogicalToArrowSchemaError> {
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, LogicalToArrowSchemaError> {
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 FixedBinaryInvalidWidthSnafu {
164                        column,
165                        byte_width: *byte_width,
166                    }
167                    .fail();
168                }
169                DataType::FixedSizeBinary(*byte_width)
170            }
171
172            LogicalDataType::Timestamp { unit, timezone } => {
173                let tz: Option<Arc<str>> = timezone.as_ref().map(|s| Arc::<str>::from(s.as_str()));
174                DataType::Timestamp(unit.to_arrow_time_unit(), tz)
175            }
176
177            LogicalDataType::Int96 => {
178                return Int96UnsupportedSnafu { column }.fail();
179            }
180
181            LogicalDataType::Decimal { precision, scale } => {
182                let precision = *precision;
183                let scale = *scale;
184                if precision <= 0 {
185                    return DecimalInvalidSnafu {
186                        column,
187                        precision,
188                        scale,
189                        details: "precision must be > 0",
190                    }
191                    .fail();
192                }
193                if scale < 0 {
194                    return DecimalInvalidSnafu {
195                        column,
196                        precision,
197                        scale,
198                        details: "scale must be >= 0",
199                    }
200                    .fail();
201                }
202                if scale > precision {
203                    return DecimalInvalidSnafu {
204                        column,
205                        precision,
206                        scale,
207                        details: "scale must be <= precision",
208                    }
209                    .fail();
210                }
211
212                if precision <= 38 {
213                    DataType::Decimal128(precision as u8, scale as i8)
214                } else if precision <= 76 {
215                    DataType::Decimal256(precision as u8, scale as i8)
216                } else {
217                    return DecimalInvalidSnafu {
218                        column,
219                        precision,
220                        scale,
221                        details: "precision exceeds Arrow maximum (76 digits)",
222                    }
223                    .fail();
224                }
225            }
226
227            LogicalDataType::Struct { fields } => {
228                let mut arrow_children: Vec<FieldRef> = Vec::with_capacity(fields.len());
229                for f in fields {
230                    let child_path = join_path(column, &f.name);
231                    arrow_children.push(f.to_arrow_field_ref(&child_path)?);
232                }
233                DataType::Struct(Fields::from(arrow_children))
234            }
235
236            LogicalDataType::List { elements } => {
237                let child_path = join_path(column, &elements.name);
238                let element_field = elements.to_arrow_field_ref(&child_path)?;
239                DataType::List(element_field)
240            }
241
242            LogicalDataType::Map {
243                key,
244                value,
245                keys_sorted,
246            } => {
247                if key.nullable {
248                    return MapKeyMustBeNonNullSnafu { column }.fail();
249                }
250
251                // Canonical Arrow Map field names are "entries", "key", "value"
252                let key_path = format!("{column}.key");
253                let val_path = format!("{column}.value");
254
255                let key_dt = key.data_type.to_arrow_datatype(&key_path)?;
256
257                let (val_dt, val_nullable) = match value.as_deref() {
258                    Some(v) => (v.data_type.to_arrow_datatype(&val_path)?, v.nullable),
259                    None => (DataType::Null, true),
260                };
261
262                let key_field: FieldRef = Arc::new(Field::new("key", key_dt, false));
263                let val_field: FieldRef = Arc::new(Field::new("value", val_dt, val_nullable));
264
265                let entries_dt = DataType::Struct(Fields::from(vec![key_field, val_field]));
266                let entries_field: FieldRef = Arc::new(Field::new("entries", entries_dt, false));
267
268                DataType::Map(entries_field, *keys_sorted)
269            }
270
271            LogicalDataType::Other(name) => {
272                return OtherTypeUnsupportedSnafu { column, name }.fail();
273            }
274        })
275    }
276}
277
278impl fmt::Display for LogicalDataType {
279    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
280        match self {
281            LogicalDataType::Bool => write!(f, "bool"),
282            LogicalDataType::Int32 => write!(f, "int32"),
283            LogicalDataType::Int64 => write!(f, "int64"),
284            LogicalDataType::UInt64 => write!(f, "uint64"),
285            LogicalDataType::Float32 => write!(f, "float32"),
286            LogicalDataType::Float64 => write!(f, "float64"),
287            LogicalDataType::Binary => write!(f, "binary"),
288            LogicalDataType::FixedBinary { byte_width } => write!(f, "fixed_binary[{byte_width}]"),
289            LogicalDataType::Utf8 => write!(f, "utf8"),
290            LogicalDataType::Int96 => write!(f, "int96"),
291
292            LogicalDataType::Timestamp { unit, timezone } => match timezone {
293                Some(tz) => write!(f, "timestamp[{}]({})", unit, tz),
294                None => write!(f, "timestamp[{}]", unit),
295            },
296
297            LogicalDataType::Decimal { precision, scale } => {
298                write!(f, "decimal(precision={precision}, scale={scale})")
299            }
300
301            LogicalDataType::Struct { fields } => {
302                write!(f, "Struct{{")?;
303                for (i, field) in fields.iter().enumerate() {
304                    if i > 0 {
305                        write!(f, ", ")?;
306                    }
307                    write!(f, "{}", field)?;
308                }
309                write!(f, "}}")
310            }
311
312            LogicalDataType::List { elements } => {
313                write!(f, "List<{}>", elements)
314            }
315
316            LogicalDataType::Map {
317                key,
318                value,
319                keys_sorted,
320            } => match value.as_deref() {
321                Some(v) => write!(f, "Map<{}, {}, keys_sorted={}>", key, v, keys_sorted),
322                None => write!(
323                    f,
324                    "Map<{}, value=omitted, keys_sorted={}>",
325                    key, keys_sorted
326                ),
327            },
328
329            LogicalDataType::Other(s) => write!(f, "{s}"),
330        }
331    }
332}
333
334/// Logical schema metadata describing the ordered collection of logical columns.
335#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
336pub struct LogicalSchema {
337    /// All logical columns that compose the schema in their defined order.
338    columns: Vec<LogicalField>,
339}
340
341impl LogicalSchema {
342    /// Convert an Arrow schema into this table's exact logical schema model.
343    pub(crate) fn try_from_arrow_schema(
344        schema: &Schema,
345    ) -> Result<Self, ArrowToLogicalSchemaError> {
346        let fields = schema
347            .fields()
348            .iter()
349            .map(|field| logical_field_from_arrow(field, field.name()))
350            .collect::<Result<Vec<_>, _>>()?;
351        Self::new(fields).context(InvalidArrowLogicalSchemaSnafu)
352    }
353
354    /// Convert this logical schema to an owned Arrow [`Schema`].
355    ///
356    /// Fails if any column uses a logical type that cannot be represented in
357    /// Arrow (see [`LogicalToArrowSchemaError`]).
358    pub fn to_arrow_schema(&self) -> Result<Schema, LogicalToArrowSchemaError> {
359        let mut fields = Vec::with_capacity(self.columns.len());
360        for c in &self.columns {
361            let fref = c.to_arrow_field_ref(&c.name)?;
362            fields.push(fref.as_ref().clone());
363        }
364
365        Ok(Schema::new(fields))
366    }
367
368    /// Convert this logical schema to a shared Arrow [`SchemaRef`].
369    ///
370    /// This is a convenience wrapper around [`Self::to_arrow_schema`].
371    pub fn to_arrow_schema_ref(&self) -> Result<SchemaRef, LogicalToArrowSchemaError> {
372        Ok(Arc::new(self.to_arrow_schema()?))
373    }
374}
375
376/// Errors that can occur while constructing or validating a logical schema.
377#[derive(Debug, Clone, Snafu, PartialEq, Eq)]
378#[non_exhaustive]
379pub enum LogicalSchemaValidationError {
380    /// Duplicate column names are not allowed.
381    #[snafu(display("Duplicate column name: {column}"))]
382    DuplicateColumn {
383        /// The duplicate column name.
384        column: String,
385    },
386
387    /// FixedBinary columns must include a positive byte width.
388    #[snafu(display(
389        "invalid FixedBinary byte_width for column '{column}': {byte_width} (must be > 0)"
390    ))]
391    FixedBinaryInvalidWidthInSchema {
392        /// Column name that failed validation.
393        column: String,
394        /// Declared byte width.
395        byte_width: i32,
396    },
397
398    /// Parquet FIXED_LEN_BYTE_ARRAY columns must include a type_length.
399    #[snafu(display(
400        "FIXED_LEN_BYTE_ARRAY column '{column}' missing type_length in Parquet schema"
401    ))]
402    FixedBinaryMissingLength {
403        /// Column name that failed validation.
404        column: String,
405    },
406
407    /// Duplicate field names within a struct are not allowed.
408    #[snafu(display("Duplicate field name: column={column_path}, field={field}"))]
409    DuplicateFieldName {
410        /// Column path for the struct that contains the duplicate field.
411        column_path: String,
412        /// Duplicate field name.
413        field: String,
414    },
415
416    /// Map key fields must be non-nullable in schema validation.
417    #[snafu(display("Invalid map key for column '{column_path}': keys must be non-nullable"))]
418    InvalidMapKeyNullability {
419        /// Column path for the map with an invalid key nullability.
420        column_path: String,
421    },
422
423    /// Struct fields must be non-empty.
424    #[snafu(display("Struct must have at least one field: column={column_path}"))]
425    EmptyStruct {
426        /// Column path for the empty struct.
427        column_path: String,
428    },
429
430    /// List element fields must have a non-empty name.
431    #[snafu(display("List element field name must be non-empty: column={column_path}"))]
432    ListElementNameEmpty {
433        /// Column path for the list with an empty element name.
434        column_path: String,
435    },
436
437    /// Struct fields must have a non-empty name.
438    #[snafu(display("Struct field name must be non-empty: column={column_path}, field={field}"))]
439    StructFieldNameEmpty {
440        /// Column path for the struct with an empty field name.
441        column_path: String,
442        /// Empty field name.
443        field: String,
444    },
445
446    /// Parquet LIST encoding does not match the supported layout.
447    #[snafu(display("Unsupported Parquet LIST encoding: column={column_path}, details={details}"))]
448    UnsupportedParquetListEncoding {
449        /// Column path for the list with unsupported encoding.
450        column_path: String,
451        /// Details describing why the LIST encoding is unsupported.
452        details: String,
453    },
454
455    /// Parquet MAP encoding does not match the supported layout.
456    #[snafu(display("Unsupported Parquet MAP encoding: column={column_path}, details={details}"))]
457    UnsupportedParquetMapEncoding {
458        /// Column path for the map with unsupported encoding.
459        column_path: String,
460        /// Details describing why the MAP encoding is unsupported.
461        details: String,
462    },
463}
464
465impl LogicalSchema {
466    /// Construct a validated logical schema (rejects duplicate column names).
467    pub fn new(columns: Vec<LogicalField>) -> Result<Self, LogicalSchemaValidationError> {
468        let mut seen = HashSet::new();
469        for col in &columns {
470            if !seen.insert(col.name.clone()) {
471                return DuplicateColumnSnafu {
472                    column: col.name.clone(),
473                }
474                .fail();
475            }
476            validate_field(col, &col.name)?;
477        }
478
479        Ok(Self { columns })
480    }
481
482    /// Borrow the logical columns.
483    pub fn columns(&self) -> &[LogicalField] {
484        &self.columns
485    }
486}
487
488fn validate_field(field: &LogicalField, path: &str) -> Result<(), LogicalSchemaValidationError> {
489    validate_dtype(&field.data_type, path)
490}
491
492fn validate_dtype(dt: &LogicalDataType, path: &str) -> Result<(), LogicalSchemaValidationError> {
493    match dt {
494        LogicalDataType::FixedBinary { byte_width } => {
495            if *byte_width <= 0 {
496                return Err(
497                    LogicalSchemaValidationError::FixedBinaryInvalidWidthInSchema {
498                        column: path.to_string(),
499                        byte_width: *byte_width,
500                    },
501                );
502            }
503            Ok(())
504        }
505
506        LogicalDataType::Struct { fields } => {
507            if fields.is_empty() {
508                return Err(LogicalSchemaValidationError::EmptyStruct {
509                    column_path: path.to_string(),
510                });
511            }
512
513            let mut seen = HashSet::with_capacity(fields.len());
514            for child in fields {
515                if child.name.trim().is_empty() {
516                    return Err(LogicalSchemaValidationError::StructFieldNameEmpty {
517                        column_path: path.to_string(),
518                        field: child.name.clone(),
519                    });
520                }
521
522                if !seen.insert(child.name.clone()) {
523                    return Err(LogicalSchemaValidationError::DuplicateFieldName {
524                        column_path: path.to_string(),
525                        field: child.name.clone(),
526                    });
527                }
528                let child_path = format!("{}.{}", path, child.name);
529                validate_field(child, &child_path)?;
530            }
531            Ok(())
532        }
533
534        LogicalDataType::List { elements } => {
535            if elements.name.trim().is_empty() {
536                return Err(LogicalSchemaValidationError::ListElementNameEmpty {
537                    column_path: path.to_string(),
538                });
539            }
540            let child_path = format!("{}.{}", path, elements.name);
541            validate_field(elements, &child_path)
542        }
543
544        LogicalDataType::Map { key, value, .. } => {
545            if key.nullable {
546                return Err(LogicalSchemaValidationError::InvalidMapKeyNullability {
547                    column_path: path.to_string(),
548                });
549            }
550            validate_field(key, &format!("{}.key", path))?;
551            if let Some(v) = value.as_deref() {
552                validate_field(v, &format!("{}.value", path))?;
553            }
554
555            Ok(())
556        }
557
558        _ => Ok(()),
559    }
560}
561
562/// Errors encountered while converting a logical schema to Arrow.
563#[derive(Debug, Snafu)]
564#[non_exhaustive]
565pub enum LogicalToArrowSchemaError {
566    /// FixedBinary fields must declare a positive byte width.
567    #[snafu(display(
568        "invalid FixedBinary byte_width for column '{column}': {byte_width} (must be > 0)"
569    ))]
570    FixedBinaryInvalidWidth {
571        /// Column name that failed validation.
572        column: String,
573        /// Declared byte width.
574        byte_width: i32,
575        /// Backtrace captured at the conversion boundary.
576        backtrace: Backtrace,
577    },
578
579    /// Int96 cannot be represented without legacy timestamp ambiguity.
580    #[snafu(display("Int96 cannot be converted to Arrow for column '{column}'"))]
581    Int96Unsupported {
582        /// Column name that failed conversion.
583        column: String,
584        /// Backtrace captured at the conversion boundary.
585        backtrace: Backtrace,
586    },
587
588    /// A named catch-all logical type cannot be represented in Arrow.
589    #[snafu(display("Logical type '{name}' cannot be converted to Arrow for column '{column}'"))]
590    OtherTypeUnsupported {
591        /// Column name that failed conversion.
592        column: String,
593        /// Type name reported by the source.
594        name: String,
595        /// Backtrace captured at the conversion boundary.
596        backtrace: Backtrace,
597    },
598
599    /// Decimal precision/scale is out of supported bounds for Arrow conversion.
600    #[snafu(display(
601        "invalid decimal definition for column '{column}': precision={precision}, scale={scale} ({details})"
602    ))]
603    DecimalInvalid {
604        /// Column name that failed conversion.
605        column: String,
606        /// Declared total precision.
607        precision: i32,
608        /// Declared scale (digits to the right of the decimal point).
609        scale: i32,
610        /// Human-readable details describing the constraint violation.
611        details: String,
612        /// Backtrace captured at the conversion boundary.
613        backtrace: Backtrace,
614    },
615
616    /// Map keys must be non-nullable when converting to Arrow.
617    #[snafu(display("map key must be non-nullable for column '{column}'"))]
618    MapKeyMustBeNonNull {
619        /// Column name that failed conversion.
620        column: String,
621        /// Backtrace captured at the conversion boundary.
622        backtrace: Backtrace,
623    },
624}
625
626/// Errors converting an Arrow schema into the table logical schema model.
627#[derive(Debug, Snafu)]
628#[non_exhaustive]
629pub enum ArrowToLogicalSchemaError {
630    /// An Arrow type cannot be represented exactly by the logical schema model.
631    #[snafu(display(
632        "Arrow type cannot be represented exactly in the table logical schema at '{column}': {data_type:?}"
633    ))]
634    Unsupported {
635        /// Dotted field path containing the unsupported type.
636        column: String,
637        /// Unsupported Arrow data type.
638        data_type: DataType,
639    },
640
641    /// The converted fields do not form a valid logical schema.
642    #[snafu(display("invalid logical schema derived from Arrow: {source}"))]
643    InvalidArrowLogicalSchema {
644        /// Logical schema validation failure.
645        source: LogicalSchemaValidationError,
646    },
647}
648
649fn logical_field_from_arrow(
650    field: &Field,
651    path: &str,
652) -> Result<LogicalField, ArrowToLogicalSchemaError> {
653    Ok(LogicalField {
654        name: field.name().clone(),
655        data_type: logical_data_type_from_arrow(field.data_type(), path)?,
656        nullable: field.is_nullable(),
657    })
658}
659
660fn logical_data_type_from_arrow(
661    data_type: &DataType,
662    path: &str,
663) -> Result<LogicalDataType, ArrowToLogicalSchemaError> {
664    let unsupported = || ArrowToLogicalSchemaError::Unsupported {
665        column: path.to_string(),
666        data_type: data_type.clone(),
667    };
668
669    Ok(match data_type {
670        DataType::Boolean => LogicalDataType::Bool,
671        DataType::Int32 => LogicalDataType::Int32,
672        DataType::Int64 => LogicalDataType::Int64,
673        DataType::UInt64 => LogicalDataType::UInt64,
674        DataType::Float32 => LogicalDataType::Float32,
675        DataType::Float64 => LogicalDataType::Float64,
676        DataType::Binary => LogicalDataType::Binary,
677        DataType::FixedSizeBinary(byte_width) if *byte_width > 0 => LogicalDataType::FixedBinary {
678            byte_width: *byte_width,
679        },
680        DataType::Utf8 => LogicalDataType::Utf8,
681        DataType::Timestamp(unit, timezone) => LogicalDataType::Timestamp {
682            unit: match unit {
683                TimeUnit::Millisecond => LogicalTimestampUnit::Millis,
684                TimeUnit::Microsecond => LogicalTimestampUnit::Micros,
685                TimeUnit::Nanosecond => LogicalTimestampUnit::Nanos,
686                TimeUnit::Second => return Err(unsupported()),
687            },
688            timezone: timezone.as_ref().map(ToString::to_string),
689        },
690        DataType::Decimal128(precision, scale)
691            if *precision > 0 && *precision <= 38 && *scale >= 0 && *scale <= *precision as i8 =>
692        {
693            LogicalDataType::Decimal {
694                precision: i32::from(*precision),
695                scale: i32::from(*scale),
696            }
697        }
698        DataType::Decimal256(precision, scale)
699            if *precision > 38 && *precision <= 76 && *scale >= 0 && *scale <= *precision as i8 =>
700        {
701            LogicalDataType::Decimal {
702                precision: i32::from(*precision),
703                scale: i32::from(*scale),
704            }
705        }
706        DataType::Struct(fields) => LogicalDataType::Struct {
707            fields: fields
708                .iter()
709                .map(|field| logical_field_from_arrow(field, &join_path(path, field.name())))
710                .collect::<Result<Vec<_>, _>>()?,
711        },
712        DataType::List(elements) => LogicalDataType::List {
713            elements: Box::new(logical_field_from_arrow(
714                elements,
715                &join_path(path, elements.name()),
716            )?),
717        },
718        DataType::Map(entries, keys_sorted) => {
719            let DataType::Struct(fields) = entries.data_type() else {
720                return Err(unsupported());
721            };
722            if entries.name() != "entries"
723                || entries.is_nullable()
724                || fields.len() != 2
725                || fields[0].name() != "key"
726                || fields[0].is_nullable()
727                || fields[1].name() != "value"
728            {
729                return Err(unsupported());
730            }
731
732            let key = logical_field_from_arrow(&fields[0], &join_path(path, "key"))?;
733            let value = if matches!(fields[1].data_type(), DataType::Null) {
734                if !fields[1].is_nullable() {
735                    return Err(unsupported());
736                }
737                None
738            } else {
739                Some(Box::new(logical_field_from_arrow(
740                    &fields[1],
741                    &join_path(path, "value"),
742                )?))
743            };
744            LogicalDataType::Map {
745                key: Box::new(key),
746                value,
747                keys_sorted: *keys_sorted,
748            }
749        }
750        _ => return Err(unsupported()),
751    })
752}
753
754#[cfg(test)]
755mod tests {
756    use super::*;
757    use std::collections::HashMap;
758
759    fn sample_logical_schema_all_supported() -> LogicalSchema {
760        LogicalSchema::new(vec![
761            LogicalField {
762                name: "flag".to_string(),
763                data_type: LogicalDataType::Bool,
764                nullable: false,
765            },
766            LogicalField {
767                name: "i32".to_string(),
768                data_type: LogicalDataType::Int32,
769                nullable: false,
770            },
771            LogicalField {
772                name: "i64".to_string(),
773                data_type: LogicalDataType::Int64,
774                nullable: true,
775            },
776            LogicalField {
777                name: "f32".to_string(),
778                data_type: LogicalDataType::Float32,
779                nullable: false,
780            },
781            LogicalField {
782                name: "f64".to_string(),
783                data_type: LogicalDataType::Float64,
784                nullable: true,
785            },
786            LogicalField {
787                name: "text".to_string(),
788                data_type: LogicalDataType::Utf8,
789                nullable: true,
790            },
791            LogicalField {
792                name: "bytes".to_string(),
793                data_type: LogicalDataType::Binary,
794                nullable: true,
795            },
796            LogicalField {
797                name: "fixed".to_string(),
798                data_type: LogicalDataType::FixedBinary { byte_width: 16 },
799                nullable: false,
800            },
801            LogicalField {
802                name: "ts".to_string(),
803                data_type: LogicalDataType::Timestamp {
804                    unit: LogicalTimestampUnit::Micros,
805                    timezone: Some("UTC".to_string()),
806                },
807                nullable: false,
808            },
809        ])
810        .expect("valid logical schema")
811    }
812
813    #[test]
814    fn arrow_schema_conversion_preserves_supported_logical_types() {
815        let expected = LogicalSchema::new(vec![
816            LogicalField {
817                name: "ts".to_string(),
818                data_type: LogicalDataType::Timestamp {
819                    unit: LogicalTimestampUnit::Nanos,
820                    timezone: Some("America/Phoenix".to_string()),
821                },
822                nullable: false,
823            },
824            LogicalField {
825                name: "decimal".to_string(),
826                data_type: LogicalDataType::Decimal {
827                    precision: 40,
828                    scale: 2,
829                },
830                nullable: true,
831            },
832            LogicalField {
833                name: "items".to_string(),
834                data_type: LogicalDataType::List {
835                    elements: Box::new(LogicalField {
836                        name: "item".to_string(),
837                        data_type: LogicalDataType::Struct {
838                            fields: vec![LogicalField {
839                                name: "value".to_string(),
840                                data_type: LogicalDataType::UInt64,
841                                nullable: false,
842                            }],
843                        },
844                        nullable: true,
845                    }),
846                },
847                nullable: true,
848            },
849            LogicalField {
850                name: "attrs".to_string(),
851                data_type: LogicalDataType::Map {
852                    key: Box::new(LogicalField {
853                        name: "key".to_string(),
854                        data_type: LogicalDataType::Utf8,
855                        nullable: false,
856                    }),
857                    value: Some(Box::new(LogicalField {
858                        name: "value".to_string(),
859                        data_type: LogicalDataType::Binary,
860                        nullable: true,
861                    })),
862                    keys_sorted: true,
863                },
864                nullable: true,
865            },
866        ])
867        .expect("valid logical schema");
868        let arrow = expected.to_arrow_schema().expect("Arrow schema");
869        let fields = arrow
870            .fields()
871            .iter()
872            .map(|field| {
873                field
874                    .as_ref()
875                    .clone()
876                    .with_metadata(HashMap::from([("ignored".to_string(), "yes".to_string())]))
877            })
878            .collect::<Vec<_>>();
879        let arrow = Schema::new_with_metadata(
880            fields,
881            HashMap::from([("schema-metadata".to_string(), "ignored".to_string())]),
882        );
883
884        assert_eq!(
885            LogicalSchema::try_from_arrow_schema(&arrow).expect("logical schema"),
886            expected
887        );
888    }
889
890    #[test]
891    fn arrow_schema_conversion_rejects_lossy_types() {
892        let cases = [
893            DataType::Int8,
894            DataType::Int16,
895            DataType::UInt8,
896            DataType::UInt16,
897            DataType::UInt32,
898            DataType::LargeUtf8,
899            DataType::Utf8View,
900            DataType::LargeBinary,
901            DataType::BinaryView,
902            DataType::Date32,
903            DataType::Timestamp(TimeUnit::Second, None),
904            DataType::Decimal128(39, 2),
905            DataType::Decimal256(10, 2),
906            DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8)),
907            DataType::LargeList(Arc::new(Field::new("item", DataType::Int64, true))),
908            DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Int64, true)), 2),
909        ];
910
911        for data_type in cases {
912            let schema = Schema::new(vec![Field::new("value", data_type.clone(), true)]);
913            assert!(matches!(
914                LogicalSchema::try_from_arrow_schema(&schema),
915                Err(ArrowToLogicalSchemaError::Unsupported {
916                    column,
917                    data_type: actual,
918                }) if column == "value" && actual == data_type
919            ));
920        }
921    }
922
923    #[test]
924    fn logical_schema_to_arrow_schema_happy_path() {
925        let logical = sample_logical_schema_all_supported();
926        let schema = logical.to_arrow_schema().expect("arrow schema conversion");
927
928        let expected = Schema::new(vec![
929            Field::new("flag", DataType::Boolean, false),
930            Field::new("i32", DataType::Int32, false),
931            Field::new("i64", DataType::Int64, true),
932            Field::new("f32", DataType::Float32, false),
933            Field::new("f64", DataType::Float64, true),
934            Field::new("text", DataType::Utf8, true),
935            Field::new("bytes", DataType::Binary, true),
936            Field::new("fixed", DataType::FixedSizeBinary(16), false),
937            Field::new(
938                "ts",
939                DataType::Timestamp(TimeUnit::Microsecond, Some(Arc::<str>::from("UTC"))),
940                false,
941            ),
942        ]);
943
944        assert_eq!(schema, expected);
945    }
946
947    #[test]
948    fn logical_schema_rejects_fixed_binary_invalid_width() {
949        for width in [0, -1] {
950            let err = LogicalSchema::new(vec![LogicalField {
951                name: "bad_fixed".to_string(),
952                data_type: LogicalDataType::FixedBinary { byte_width: width },
953                nullable: false,
954            }])
955            .expect_err("expected invalid schema to be rejected");
956
957            assert!(
958                matches!(
959                    &err,
960                    LogicalSchemaValidationError::FixedBinaryInvalidWidthInSchema {
961                        column,
962                        byte_width
963                    } if column == "bad_fixed" && *byte_width == width
964                ),
965                "unexpected error: {err:?}"
966            );
967        }
968    }
969
970    #[test]
971    fn logical_schema_rejects_int96() {
972        let logical = LogicalSchema::new(vec![LogicalField {
973            name: "legacy_ts".to_string(),
974            data_type: LogicalDataType::Int96,
975            nullable: false,
976        }])
977        .expect("valid schema structure");
978
979        let err = logical.to_arrow_schema().unwrap_err();
980        assert!(
981            matches!(
982                &err,
983                LogicalToArrowSchemaError::Int96Unsupported { column, .. }
984                    if column == "legacy_ts"
985            ),
986            "unexpected error: {err:?}"
987        );
988    }
989
990    #[test]
991    fn logical_schema_map_entries_field_is_non_nullable() {
992        let logical = LogicalSchema::new(vec![LogicalField {
993            name: "attrs".to_string(),
994            data_type: LogicalDataType::Map {
995                key: Box::new(LogicalField {
996                    name: "key".to_string(),
997                    data_type: LogicalDataType::Utf8,
998                    nullable: false,
999                }),
1000                value: Some(Box::new(LogicalField {
1001                    name: "value".to_string(),
1002                    data_type: LogicalDataType::Int64,
1003                    nullable: true,
1004                })),
1005                keys_sorted: false,
1006            },
1007            nullable: true,
1008        }])
1009        .expect("valid schema");
1010
1011        let schema = logical.to_arrow_schema().expect("arrow schema conversion");
1012        let field = schema.field(0);
1013        let DataType::Map(entries_field, _) = field.data_type() else {
1014            panic!("expected map type, got {:?}", field.data_type());
1015        };
1016        assert!(
1017            !entries_field.is_nullable(),
1018            "map entries field should be non-nullable"
1019        );
1020    }
1021
1022    #[test]
1023    fn logical_schema_map_value_none_maps_to_null_field() {
1024        let logical = LogicalSchema::new(vec![LogicalField {
1025            name: "attrs".to_string(),
1026            data_type: LogicalDataType::Map {
1027                key: Box::new(LogicalField {
1028                    name: "key".to_string(),
1029                    data_type: LogicalDataType::Utf8,
1030                    nullable: false,
1031                }),
1032                value: None,
1033                keys_sorted: false,
1034            },
1035            nullable: false,
1036        }])
1037        .expect("valid schema");
1038
1039        let schema = logical.to_arrow_schema().expect("arrow schema conversion");
1040        let field = schema.field(0);
1041        let DataType::Map(entries_field, _) = field.data_type() else {
1042            panic!("expected map type, got {:?}", field.data_type());
1043        };
1044        let DataType::Struct(fields) = entries_field.data_type() else {
1045            panic!(
1046                "expected entries struct, got {:?}",
1047                entries_field.data_type()
1048            );
1049        };
1050        let value_field = fields
1051            .iter()
1052            .find(|f| f.name() == "value")
1053            .expect("value field");
1054        assert!(
1055            matches!(value_field.data_type(), DataType::Null) && value_field.is_nullable(),
1056            "value field should be Null and nullable"
1057        );
1058    }
1059
1060    #[test]
1061    fn logical_schema_rejects_empty_struct_field_name() {
1062        let err = LogicalSchema::new(vec![LogicalField {
1063            name: "root".to_string(),
1064            data_type: LogicalDataType::Struct {
1065                fields: vec![LogicalField {
1066                    name: "".to_string(),
1067                    data_type: LogicalDataType::Int32,
1068                    nullable: false,
1069                }],
1070            },
1071            nullable: false,
1072        }])
1073        .expect_err("expected invalid schema");
1074
1075        assert!(
1076            matches!(
1077                &err,
1078                LogicalSchemaValidationError::StructFieldNameEmpty { column_path, field }
1079                if column_path == "root" && field.is_empty()
1080            ),
1081            "unexpected error: {err:?}"
1082        );
1083    }
1084
1085    #[test]
1086    fn logical_schema_rejects_other_type() {
1087        let logical = LogicalSchema::new(vec![LogicalField {
1088            name: "opaque".to_string(),
1089            data_type: LogicalDataType::Other("parquet::Map".to_string()),
1090            nullable: true,
1091        }])
1092        .expect("valid schema structure");
1093
1094        let err = logical.to_arrow_schema().unwrap_err();
1095        assert!(
1096            matches!(
1097                &err,
1098                LogicalToArrowSchemaError::OtherTypeUnsupported { column, name, .. }
1099                    if column == "opaque" && name == "parquet::Map"
1100            ),
1101            "unexpected error: {err:?}"
1102        );
1103    }
1104
1105    #[test]
1106    fn logical_schema_timestamp_without_timezone() {
1107        let logical = LogicalSchema::new(vec![LogicalField {
1108            name: "ts".to_string(),
1109            data_type: LogicalDataType::Timestamp {
1110                unit: LogicalTimestampUnit::Millis,
1111                timezone: None,
1112            },
1113            nullable: false,
1114        }])
1115        .expect("valid schema structure");
1116
1117        let schema = logical.to_arrow_schema().expect("arrow schema conversion");
1118        let expected = Schema::new(vec![Field::new(
1119            "ts",
1120            DataType::Timestamp(TimeUnit::Millisecond, None),
1121            false,
1122        )]);
1123        assert_eq!(schema, expected);
1124    }
1125
1126    #[test]
1127    fn logical_schema_decimal_conversion_bounds() {
1128        let valid_128 = LogicalSchema::new(vec![LogicalField {
1129            name: "dec128".to_string(),
1130            data_type: LogicalDataType::Decimal {
1131                precision: 38,
1132                scale: 10,
1133            },
1134            nullable: false,
1135        }])
1136        .expect("valid schema structure");
1137        let schema = valid_128
1138            .to_arrow_schema()
1139            .expect("arrow schema conversion");
1140        assert_eq!(
1141            schema,
1142            Schema::new(vec![Field::new(
1143                "dec128",
1144                DataType::Decimal128(38, 10),
1145                false
1146            )])
1147        );
1148
1149        let valid_256 = LogicalSchema::new(vec![LogicalField {
1150            name: "dec256".to_string(),
1151            data_type: LogicalDataType::Decimal {
1152                precision: 76,
1153                scale: 5,
1154            },
1155            nullable: false,
1156        }])
1157        .expect("valid schema structure");
1158        let schema = valid_256
1159            .to_arrow_schema()
1160            .expect("arrow schema conversion");
1161        assert_eq!(
1162            schema,
1163            Schema::new(vec![Field::new(
1164                "dec256",
1165                DataType::Decimal256(76, 5),
1166                false
1167            )])
1168        );
1169
1170        let invalid = LogicalSchema::new(vec![LogicalField {
1171            name: "dec_too_large".to_string(),
1172            data_type: LogicalDataType::Decimal {
1173                precision: 77,
1174                scale: 0,
1175            },
1176            nullable: false,
1177        }])
1178        .expect("valid schema structure");
1179        let err = invalid.to_arrow_schema().unwrap_err();
1180        assert!(
1181            matches!(
1182                &err,
1183                LogicalToArrowSchemaError::DecimalInvalid { column, precision, scale, .. }
1184                    if column == "dec_too_large" && *precision == 77 && *scale == 0
1185            ),
1186            "unexpected error: {err:?}"
1187        );
1188    }
1189
1190    #[test]
1191    fn logical_schema_decimal_validation_errors() {
1192        let cases = vec![
1193            ("dec_precision_zero", 0, 0, "precision must be > 0"),
1194            ("dec_scale_negative", 10, -1, "scale must be >= 0"),
1195            ("dec_scale_gt_precision", 4, 5, "scale must be <= precision"),
1196        ];
1197
1198        for (name, precision, scale, details_substr) in cases {
1199            let logical = LogicalSchema::new(vec![LogicalField {
1200                name: name.to_string(),
1201                data_type: LogicalDataType::Decimal { precision, scale },
1202                nullable: false,
1203            }])
1204            .expect("valid schema structure");
1205
1206            let err = logical.to_arrow_schema().unwrap_err();
1207            assert!(
1208                matches!(
1209                    &err,
1210                    LogicalToArrowSchemaError::DecimalInvalid {
1211                        column,
1212                        precision: p,
1213                        scale: s,
1214                        details,
1215                        ..
1216                    }
1217                        if column == name && *p == precision && *s == scale && details.contains(details_substr)
1218                ),
1219                "unexpected error: {err:?}"
1220            );
1221        }
1222    }
1223
1224    #[test]
1225    fn logical_schema_fixed_binary_json_roundtrip() {
1226        let logical = LogicalSchema::new(vec![LogicalField {
1227            name: "fixed".to_string(),
1228            data_type: LogicalDataType::FixedBinary { byte_width: 8 },
1229            nullable: false,
1230        }])
1231        .expect("valid schema structure");
1232
1233        let json = serde_json::to_string(&logical).unwrap();
1234        let back: LogicalSchema = serde_json::from_str(&json).unwrap();
1235        assert_eq!(back, logical);
1236    }
1237
1238    #[test]
1239    fn logical_schema_decimal_json_roundtrip() {
1240        let logical = LogicalSchema::new(vec![LogicalField {
1241            name: "amount".to_string(),
1242            data_type: LogicalDataType::Decimal {
1243                precision: 18,
1244                scale: 4,
1245            },
1246            nullable: true,
1247        }])
1248        .expect("valid schema structure");
1249
1250        let json = serde_json::to_string(&logical).unwrap();
1251        let back: LogicalSchema = serde_json::from_str(&json).unwrap();
1252        assert_eq!(back, logical);
1253    }
1254
1255    #[test]
1256    fn logical_schema_uint64_roundtrips_json_and_maps_exactly_to_arrow() {
1257        let schema = LogicalSchema::new(vec![LogicalField {
1258            name: "offset".to_string(),
1259            data_type: LogicalDataType::UInt64,
1260            nullable: false,
1261        }])
1262        .unwrap();
1263
1264        let json = serde_json::to_string(&schema).unwrap();
1265        let restored: LogicalSchema = serde_json::from_str(&json).unwrap();
1266        assert_eq!(restored, schema);
1267        assert_eq!(
1268            restored.to_arrow_schema().unwrap().field(0).data_type(),
1269            &DataType::UInt64
1270        );
1271    }
1272}
timeseries_table_format/metadata/logical_schema.rs

timeseries_table_format/metadata/
logical_schema.rs
