timeseries_table_format/metadata/
table.rs

1//! Table-level metadata structures recorded in the log.
2
3use std::collections::{BTreeMap, BTreeSet};
4
5use arrow::datatypes::SchemaRef;
6use chrono::{DateTime, Utc};
7use serde::{Deserialize, Serialize};
8use snafu::{Backtrace, prelude::*};
9
10use crate::metadata::{
11    index::IndexSpec,
12    logical_schema::{LogicalSchema, LogicalToArrowSchemaError},
13    protocol::{TABLE_PROTOCOL_VERSION, deserialize_required_features},
14};
15
16/// The high-level "kind" of table.
17///
18/// v0.1 supports only `TimeSeries`, but a `Generic` kind is reserved so that
19/// the log format can represent non-timeseries tables later without breaking
20/// existing JSON.
21#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
22pub enum TableKind {
23    /// A time-series table with an explicit ordered index specification.
24    TimeSeries(IndexSpec),
25
26    /// Placeholder for future basic tables that do not have a time index.
27    /// Not used in v0.1.
28    Generic,
29}
30
31/// High-level table metadata stored in the log.
32///
33/// This describes the table kind, a logical schema (optional in v0.1), and
34/// basic bookkeeping fields.
35#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
36#[serde(try_from = "RawTableMeta")]
37pub struct TableMeta {
38    /// Table kind: TimeSeries or Generic.
39    pub(crate) kind: TableKind,
40
41    /// Optional logical schema description.
42    ///
43    /// v0.1 can treat this as informational; enforcement is handled by
44    /// higher layers.
45    pub(crate) logical_schema: Option<LogicalSchema>,
46
47    /// Creation timestamp of the table, stored as RFC3339 UTC.
48    pub(crate) created_at: DateTime<Utc>,
49
50    /// Version of the core metadata and commit-log protocol.
51    ///
52    /// Writers set this to [`TABLE_PROTOCOL_VERSION`].
53    pub(crate) protocol_version: u32,
54
55    /// Features a client must support to read this table.
56    pub(crate) required_reader_features: BTreeSet<String>,
57
58    /// Features a client must support to write this table.
59    pub(crate) required_writer_features: BTreeSet<String>,
60}
61
62#[derive(Deserialize)]
63struct RawTableMeta {
64    kind: TableKind,
65    logical_schema: Option<LogicalSchema>,
66    created_at: DateTime<Utc>,
67    protocol_version: u32,
68    #[serde(deserialize_with = "deserialize_required_features")]
69    required_reader_features: BTreeSet<String>,
70    #[serde(deserialize_with = "deserialize_required_features")]
71    required_writer_features: BTreeSet<String>,
72    #[serde(flatten)]
73    extra: BTreeMap<String, serde_json::Value>,
74}
75
76impl TryFrom<RawTableMeta> for TableMeta {
77    type Error = String;
78
79    fn try_from(raw: RawTableMeta) -> Result<Self, Self::Error> {
80        if raw.extra.contains_key("format_version") {
81            return Err("legacy field 'format_version' is not valid protocol-v7 metadata".into());
82        }
83
84        Ok(Self {
85            kind: raw.kind,
86            logical_schema: raw.logical_schema,
87            created_at: raw.created_at,
88            protocol_version: raw.protocol_version,
89            required_reader_features: raw.required_reader_features,
90            required_writer_features: raw.required_writer_features,
91        })
92    }
93}
94
95/// Errors encountered while retrieving or converting a table's logical schema.
96#[derive(Debug, Snafu)]
97#[non_exhaustive]
98pub enum TableArrowSchemaError {
99    /// The table metadata has not yet recorded a canonical logical schema.
100    #[snafu(display("table has no canonical logical schema yet (logical_schema is None)"))]
101    MissingCanonicalSchema {
102        /// Backtrace captured at the table schema boundary.
103        backtrace: Backtrace,
104    },
105
106    /// Failed to convert the table's logical schema to Arrow types.
107    #[snafu(display("Failed to convert the table logical schema to Arrow: {source}"))]
108    Conversion {
109        /// Underlying conversion error.
110        #[snafu(source, backtrace)]
111        source: LogicalToArrowSchemaError,
112    },
113}
114
115impl TableMeta {
116    /// Returns the table kind (e.g. time series or generic).
117    pub fn kind(&self) -> &TableKind {
118        &self.kind
119    }
120
121    /// Returns the optional logical schema if it has been set.
122    pub fn logical_schema(&self) -> Option<&LogicalSchema> {
123        self.logical_schema.as_ref()
124    }
125
126    /// Returns the UTC timestamp when the table was created.
127    pub fn created_at(&self) -> DateTime<Utc> {
128        self.created_at
129    }
130
131    /// Convenience constructor for a time-series table.
132    ///
133    /// - Fills `created_at` with `Utc::now()`.
134    /// - Fills `protocol_version` with `TABLE_PROTOCOL_VERSION`.
135    /// - Starts with no required reader or writer features.
136    /// - Leaves `logical_schema` as `None`; it will be adopted from the
137    ///   first appended segment in v0.1.
138    pub fn new_time_series(index: IndexSpec) -> Self {
139        TableMeta {
140            kind: TableKind::TimeSeries(index),
141            logical_schema: None,
142            created_at: Utc::now(),
143            protocol_version: TABLE_PROTOCOL_VERSION,
144            required_reader_features: BTreeSet::new(),
145            required_writer_features: BTreeSet::new(),
146        }
147    }
148
149    /// Variant that lets you explicitly pass a logical schema up front.
150    pub fn new_time_series_with_schema(index: IndexSpec, logical_schema: LogicalSchema) -> Self {
151        TableMeta {
152            kind: TableKind::TimeSeries(index),
153            logical_schema: Some(logical_schema),
154            created_at: Utc::now(),
155            protocol_version: TABLE_PROTOCOL_VERSION,
156            required_reader_features: BTreeSet::new(),
157            required_writer_features: BTreeSet::new(),
158        }
159    }
160
161    /// Convert the table's logical schema to a shared Arrow [`SchemaRef`].
162    ///
163    /// Returns [`TableArrowSchemaError::MissingCanonicalSchema`] if the schema has
164    /// not yet been established for the table.
165    pub fn arrow_schema_ref(&self) -> Result<SchemaRef, TableArrowSchemaError> {
166        let logical = self
167            .logical_schema
168            .as_ref()
169            .ok_or_else(|| MissingCanonicalSchemaSnafu.build())?;
170
171        logical.to_arrow_schema_ref().context(ConversionSnafu)
172    }
173}
174
175/// For v0.1, a `TableMetaDelta` is just a full replacement of [`TableMeta`].
176///
177/// This alias keeps the wire format simple (the JSON is the same as `TableMeta`)
178/// while leaving room to evolve to more granular metadata updates in future
179/// versions (for example, partial updates or additive fields).
180pub type TableMetaDelta = TableMeta;
181
182#[cfg(test)]
183mod tests {
184    use std::error::Error as _;
185
186    use chrono::TimeZone;
187    use snafu::ErrorCompat;
188
189    use crate::metadata::{
190        index::{IndexKind, IndexSpec, TimeIndexGranularity},
191        logical_schema::{LogicalDataType, LogicalField},
192        protocol::TABLE_PROTOCOL_VERSION,
193    };
194
195    use super::*;
196
197    fn sample_time_index_spec() -> IndexSpec {
198        IndexSpec {
199            column: "ts".to_string(),
200            entity_columns: vec!["symbol".to_string()],
201            kind: IndexKind::Timestamp {
202                index_granularity: TimeIndexGranularity::Minutes(1),
203                timezone: None,
204            },
205        }
206    }
207
208    #[test]
209    fn table_meta_baseline_protocol_json_is_stable() {
210        let mut meta = TableMeta::new_time_series(sample_time_index_spec());
211        meta.created_at = Utc.with_ymd_and_hms(2025, 1, 1, 0, 0, 0).single().unwrap();
212
213        assert_eq!(
214            serde_json::to_value(meta).unwrap(),
215            serde_json::json!({
216                "kind": {
217                    "TimeSeries": {
218                        "column": "ts",
219                        "entity_columns": ["symbol"],
220                        "kind": {
221                            "type": "timestamp",
222                            "index_granularity": {"Minutes": 1}
223                        }
224                    }
225                },
226                "logical_schema": null,
227                "created_at": "2025-01-01T00:00:00Z",
228                "protocol_version": 7,
229                "required_reader_features": [],
230                "required_writer_features": []
231            })
232        );
233    }
234
235    #[test]
236    fn table_meta_protocol_fields_roundtrip_canonically() {
237        let mut meta = TableMeta::new_time_series(sample_time_index_spec());
238        meta.required_reader_features.insert("z_reader".to_string());
239        meta.required_reader_features.insert("a_reader".to_string());
240        meta.required_writer_features.insert("writer_2".to_string());
241
242        let mut json = serde_json::to_value(&meta).unwrap();
243        assert_eq!(json["protocol_version"], TABLE_PROTOCOL_VERSION);
244        assert_eq!(
245            json["required_reader_features"],
246            serde_json::json!(["a_reader", "z_reader"])
247        );
248        assert_eq!(
249            json["required_writer_features"],
250            serde_json::json!(["writer_2"])
251        );
252        assert!(json.get("format_version").is_none());
253
254        json["future_field"] = serde_json::json!({"ignored": true});
255        let decoded: TableMeta = serde_json::from_value(json).unwrap();
256        assert_eq!(decoded, meta);
257    }
258
259    #[test]
260    fn table_meta_rejects_legacy_format_version_field() {
261        let mut json =
262            serde_json::to_value(TableMeta::new_time_series(sample_time_index_spec())).unwrap();
263        json["format_version"] = serde_json::json!(6);
264
265        let error = serde_json::from_value::<TableMeta>(json).unwrap_err();
266        assert!(error.to_string().contains("legacy field 'format_version'"));
267    }
268
269    #[test]
270    fn table_meta_arrow_schema_ref_requires_logical_schema() {
271        let meta = TableMeta::new_time_series(sample_time_index_spec());
272        let err = meta.arrow_schema_ref().unwrap_err();
273        assert!(matches!(
274            &err,
275            TableArrowSchemaError::MissingCanonicalSchema { .. }
276        ));
277        assert!(ErrorCompat::backtrace(&err).is_some());
278    }
279
280    #[test]
281    fn table_meta_arrow_schema_ref_propagates_convert_error() {
282        let logical = LogicalSchema::new(vec![LogicalField {
283            name: "legacy_ts".to_string(),
284            data_type: LogicalDataType::Int96,
285            nullable: false,
286        }])
287        .expect("valid schema structure");
288        let meta = TableMeta::new_time_series_with_schema(sample_time_index_spec(), logical);
289
290        let err = meta.arrow_schema_ref().unwrap_err();
291        let table_backtrace = ErrorCompat::backtrace(&err).expect("table schema backtrace");
292        let conversion = err
293            .source()
294            .and_then(|source| source.downcast_ref::<LogicalToArrowSchemaError>())
295            .expect("Arrow conversion source");
296        let conversion_backtrace =
297            ErrorCompat::backtrace(conversion).expect("conversion backtrace");
298
299        assert!(matches!(
300            conversion,
301            LogicalToArrowSchemaError::Int96Unsupported { column, .. }
302                if column == "legacy_ts"
303        ));
304        assert!(std::ptr::eq(table_backtrace, conversion_backtrace));
305    }
306}
timeseries_table_format/metadata/table.rs

timeseries_table_format/metadata/
table.rs
