sql: LIR schema registry to support stable LIR

.gitattributes

@@ -1 +1,2 @@
 **/*.svg linguist-generated=true
+src/compute-types/tests/snapshots/*.json linguist-generated=true

Cargo.toml

@@ -502,6 +502,7 @@ sentry-panic = "0.48.2"
 sentry-tracing = "0.48.2"
 serde = { version = "1.0.219", features = ["derive"] }
 serde-aux = "4.7.0"
+serde-reflection = "0.5.0"
 serde_bytes = "0.11.19"
 serde_json = "1.0.150"
 serde_plain = "1.0.2"

src/compute-types/Cargo.toml

@@ -23,5 +23,12 @@ serde.workspace = true
 timely.workspace = true
 tracing.workspace = true
 
+[dev-dependencies]
+chrono.workspace = true
+chrono-tz.workspace = true
+mz-pgtz = { path = "../pgtz" }
+serde-reflection.workspace = true
+serde_json.workspace = true
+
 [features]
 default = []

src/compute-types/src/plan.rs

@@ -203,6 +203,66 @@ impl std::fmt::Display for LirId {
     }
 }
 
+/// Version of the stable LIR serialization format.
+///
+/// Bump this when the serialized representation of [`LirRelationExpr`] or
+/// anything it transitively contains changes. The schema snapshot test in
+/// `tests/lir_schema.rs` enforces that the traced schema matches the
+/// checked-in `tests/snapshots/lir_v{LIR_VERSION}.yaml`.
+pub const LIR_VERSION: u64 = 1;
+
+pub use constant_rows_serde::ConstantRows;
+
+/// Serializes `LirRelationNode::Constant`'s rows through the named
+/// [`ConstantRows`] mirror enum instead of std `Result`.
+///
+/// The stable LIR schema registry maps each container name to a single
+/// format, and `Result` would clash with the differently instantiated
+/// `Result` in `LirScalarExpr::Literal`. The mirror has the same variant
+/// order as `Result`, so the encoded bytes are unchanged.
+mod constant_rows_serde {
+    use mz_expr::EvalError;
+    use mz_repr::{Diff, Row, Timestamp};
+    use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+    /// The serialized form of `LirRelationNode::Constant`'s rows.
+    #[derive(Debug, Serialize, Deserialize)]
+    pub enum ConstantRows {
+        /// See `Result::Ok`.
+        Ok(Vec<(Row, Timestamp, Diff)>),
+        /// See `Result::Err`.
+        Err(EvalError),
+    }
+
+    /// Borrowing mirror of [`ConstantRows`], to serialize without cloning.
+    #[derive(Serialize)]
+    #[serde(rename = "ConstantRows")]
+    enum ConstantRowsRef<'a> {
+        Ok(&'a Vec<(Row, Timestamp, Diff)>),
+        Err(&'a EvalError),
+    }
+
+    pub fn serialize<S: Serializer>(
+        rows: &Result<Vec<(Row, Timestamp, Diff)>, EvalError>,
+        serializer: S,
+    ) -> Result<S::Ok, S::Error> {
+        let mirror = match rows {
+            Ok(rows) => ConstantRowsRef::Ok(rows),
+            Err(err) => ConstantRowsRef::Err(err),
+        };
+        mirror.serialize(serializer)
+    }
+
+    pub fn deserialize<'de, D: Deserializer<'de>>(
+        deserializer: D,
+    ) -> Result<Result<Vec<(Row, Timestamp, Diff)>, EvalError>, D::Error> {
+        Ok(match ConstantRows::deserialize(deserializer)? {
+            ConstantRows::Ok(rows) => Ok(rows),
+            ConstantRows::Err(err) => Err(err),
+        })
+    }
+}
+
 /// A rendering plan with as much conditional logic as possible removed.
 #[derive(Clone, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
 pub struct LirRelationExpr {
@@ -218,6 +278,7 @@ pub enum LirRelationNode {
     /// A collection containing a pre-determined collection.
     Constant {
         /// Explicit update triples for the collection.
+        #[serde(with = "constant_rows_serde")]
         rows: Result<Vec<(Row, Timestamp, Diff)>, EvalError>,
     },
     /// A reference to a bound collection.

src/compute-types/src/plan/scalar.rs

@@ -35,7 +35,10 @@ pub enum LirScalarExpr {
     Column(usize, TreatAsEqual<Option<Arc<str>>>),
     /// A literal value.
     /// (Stored as a row, because we can't own a Datum)
-    Literal(Result<Row, EvalError>, ReprColumnType),
+    Literal(
+        #[serde(with = "literal_value_serde")] Result<Row, EvalError>,
+        ReprColumnType,
+    ),
     /// A function call that takes one expression as an argument.
     CallUnary {
         /// Function
@@ -75,6 +78,58 @@ pub enum LirScalarExpr {
     },
 }
 
+pub use literal_value_serde::LiteralValue;
+
+/// Serializes `LirScalarExpr::Literal`'s value through the named
+/// [`LiteralValue`] mirror enum instead of std `Result`.
+///
+/// The stable LIR schema registry maps each container name to a single
+/// format, and `Result` would clash with the differently instantiated
+/// `Result` in `LirRelationNode::Constant`. The mirror has the same variant
+/// order as `Result`, so the encoded bytes are unchanged.
+mod literal_value_serde {
+    use mz_expr::EvalError;
+    use mz_repr::Row;
+    use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+    /// The serialized form of `LirScalarExpr::Literal`'s value.
+    #[derive(Debug, Serialize, Deserialize)]
+    pub enum LiteralValue {
+        /// See `Result::Ok`.
+        Ok(Row),
+        /// See `Result::Err`.
+        Err(EvalError),
+    }
+
+    /// Borrowing mirror of [`LiteralValue`], to serialize without cloning.
+    #[derive(Serialize)]
+    #[serde(rename = "LiteralValue")]
+    enum LiteralValueRef<'a> {
+        Ok(&'a Row),
+        Err(&'a EvalError),
+    }
+
+    pub fn serialize<S: Serializer>(
+        value: &Result<Row, EvalError>,
+        serializer: S,
+    ) -> Result<S::Ok, S::Error> {
+        let mirror = match value {
+            Ok(row) => LiteralValueRef::Ok(row),
+            Err(err) => LiteralValueRef::Err(err),
+        };
+        mirror.serialize(serializer)
+    }
+
+    pub fn deserialize<'de, D: Deserializer<'de>>(
+        deserializer: D,
+    ) -> Result<Result<Row, EvalError>, D::Error> {
+        Ok(match LiteralValue::deserialize(deserializer)? {
+            LiteralValue::Ok(row) => Ok(row),
+            LiteralValue::Err(err) => Err(err),
+        })
+    }
+}
+
 impl LirScalarExpr {
     /// Generates an LSE representing the given column reference.
     pub fn column(c: usize) -> Self {