laminar_core/schema_binding/
mod.rs

1//! Durable connector contracts. Native schemas remain separate from Arrow read/input schemas.
2
3use std::collections::BTreeMap;
4
5use arrow_schema::Schema;
6use serde::{Deserialize, Serialize};
7use sha2::{Digest, Sha256};
8use thiserror::Error;
9
10/// Current schema contract and declarative mapping representation.
11pub const SCHEMA_BINDING_VERSION: u16 = 1;
12/// Maximum canonical size of one connector contract, including native references.
13pub const MAX_SCHEMA_BINDING_BYTES: usize = 4 * 1024 * 1024;
14/// Maximum number of logical or external fields in one binding.
15pub const MAX_SCHEMA_FIELDS: usize = 4096;
16/// Maximum transitive native schema artifacts in one binding.
17pub const MAX_SCHEMA_REFERENCES: usize = 64;
18
19/// Direction in which a connector applies its schema contract.
20#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
21#[serde(rename_all = "snake_case")]
22pub enum SchemaDirection {
23    /// External data is projected into a logical read contract.
24    Source,
25    /// Bound query output is mapped into a writer contract.
26    Sink,
27}
28
29/// Authority that established the immutable logical schema.
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
31#[serde(rename_all = "snake_case")]
32pub enum SchemaOrigin {
33    /// User-supplied fields, including legacy catalog reconstruction.
34    Explicit,
35    /// A fixed connector protocol or generator definition.
36    BuiltIn,
37    /// Authoritative registry, database, table, or file metadata.
38    Metadata,
39    /// The output of a query bound against the catalog.
40    Query,
41    /// A separately authorized, bounded sample; never an authoritative constraint.
42    Sample,
43}
44
45/// One immutable native schema, including the semantics Arrow cannot express.
46#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
47#[serde(deny_unknown_fields)]
48pub struct NativeSchema {
49    /// Codec or native metadata representation (for example `avro` or `iceberg`).
50    pub format: String,
51    /// Stable resource identity and concrete schema selection, without credentials.
52    pub identity: BTreeMap<String, String>,
53    /// Complete native definition, preserving ordered fields, unions, defaults and IDs.
54    pub definition: serde_json::Value,
55    /// Transitive definitions in deterministic order.
56    #[serde(default, skip_serializing_if = "Vec::is_empty")]
57    pub references: Vec<NativeSchemaArtifact>,
58}
59
60/// A transitive native definition and its scoped identity.
61#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
62#[serde(deny_unknown_fields)]
63pub struct NativeSchemaArtifact {
64    /// Native name used by the referring definition.
65    pub name: String,
66    /// Scoped concrete version/identifier.
67    pub identity: BTreeMap<String, String>,
68    /// Complete definition, including its own reference declarations.
69    pub definition: serde_json::Value,
70}
71
72/// Declarative mapping compiled once by the connector when it activates.
73#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
74#[serde(deny_unknown_fields)]
75pub struct SchemaFieldMapping {
76    /// Field in the read or query-input contract.
77    pub logical: String,
78    /// Field in the external or serialization contract.
79    pub external: String,
80}
81
82/// Immutable connector binding for one catalog generation.
83#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
84#[serde(deny_unknown_fields)]
85pub struct SchemaBinding {
86    /// Version of this encoding and mapping semantics.
87    pub version: u16,
88    /// Connector type, independent of credentials and connection handles.
89    pub connector: String,
90    /// Source or sink binding.
91    pub direction: SchemaDirection,
92    /// Origin of the logical contract.
93    pub origin: SchemaOrigin,
94    /// Complete logical reader or bound query-output schema.
95    pub logical: Schema,
96    /// Native reader/writer fields, when an external authority exists.
97    pub external: Option<Schema>,
98    /// Value/table/protocol contract, distinct from observed message writers.
99    pub value: Option<NativeSchema>,
100    /// Independently selected key contract, when the codec uses one.
101    pub key: Option<NativeSchema>,
102    /// Field mapping in logical field order.
103    pub mapping: Vec<SchemaFieldMapping>,
104    /// Engine changelog fields consumed by a sink and excluded from business writes.
105    #[serde(default, skip_serializing_if = "Vec::is_empty")]
106    pub control_fields: Vec<String>,
107}
108
109/// Structural or directional incompatibility in a resolved schema contract.
110#[derive(Debug, Error)]
111pub enum SchemaBindingError {
112    /// Invalid contract, unsupported representation, or resource limit.
113    #[error("invalid schema contract: {0}")]
114    Invalid(String),
115    /// A field cannot be read or written under the declared policy.
116    #[error("incompatible schema contract: {0}")]
117    Incompatible(String),
118    /// Encoding failed.
119    #[error("schema contract encoding: {0}")]
120    Encoding(#[from] serde_json::Error),
121}
122
123impl SchemaBinding {
124    /// Construct a logical-only contract; no external mutation or inference occurs.
125    ///
126    /// # Errors
127    /// Rejects empty/duplicate fields and contracts exceeding the representation bounds.
128    pub fn logical(
129        connector: impl Into<String>,
130        direction: SchemaDirection,
131        origin: SchemaOrigin,
132        logical: Schema,
133    ) -> Result<Self, SchemaBindingError> {
134        let mapping = logical
135            .fields()
136            .iter()
137            .map(|field| SchemaFieldMapping {
138                logical: field.name().clone(),
139                external: field.name().clone(),
140            })
141            .collect();
142        let binding = Self {
143            version: SCHEMA_BINDING_VERSION,
144            connector: connector.into(),
145            direction,
146            origin,
147            logical,
148            external: None,
149            value: None,
150            key: None,
151            mapping,
152            control_fields: Vec::new(),
153        };
154        binding.canonical_bytes()?;
155        Ok(binding)
156    }
157
158    /// Bind fields by exact name and type. Sources may project external fields;
159    /// sinks must supply every writer field and may not discard extra input fields.
160    /// Casts, defaults and nullable fills require a destination-specific policy.
161    ///
162    /// # Errors
163    /// Rejects missing fields, differing types or unsafe directional nullability.
164    pub fn bind_external(&mut self, external: Schema) -> Result<(), SchemaBindingError> {
165        validation::validate_schema(&external)?;
166        let mapping = validation::mapping(
167            self.direction,
168            &self.logical,
169            &external,
170            &self.control_fields,
171        )?;
172        self.external = Some(external);
173        self.mapping = mapping;
174        Ok(())
175    }
176
177    /// Validate structure and produce deterministic JSON without reordering fields or unions.
178    ///
179    /// # Errors
180    /// Rejects invalid versions, incomplete mappings, and oversized contracts.
181    pub fn canonical_bytes(&self) -> Result<Vec<u8>, SchemaBindingError> {
182        validation::validate(self)?;
183        let mut value = serde_json::to_value(self)?;
184        value.sort_all_objects();
185        let bytes = serde_json::to_vec(&value)?;
186        if bytes.len() > MAX_SCHEMA_BINDING_BYTES {
187            return Err(SchemaBindingError::Invalid(format!(
188                "contract exceeds {MAX_SCHEMA_BINDING_BYTES} bytes"
189            )));
190        }
191        Ok(bytes)
192    }
193
194    /// SHA-256 of the versioned canonical encoding; native identity is also retained.
195    ///
196    /// # Errors
197    /// Returns structural or encoding errors for an invalid contract.
198    pub fn fingerprint(&self) -> Result<String, SchemaBindingError> {
199        const HEX: &[u8; 16] = b"0123456789abcdef";
200        let digest = Sha256::digest(self.canonical_bytes()?);
201        let mut fingerprint = String::with_capacity(64);
202        for byte in digest {
203            fingerprint.push(char::from(HEX[usize::from(byte >> 4)]));
204            fingerprint.push(char::from(HEX[usize::from(byte & 15)]));
205        }
206        Ok(fingerprint)
207    }
208
209    /// Decode and validate a persisted contract without contacting its external authority.
210    ///
211    /// # Errors
212    /// Rejects malformed, unsupported, or oversized representations.
213    pub fn decode(bytes: &[u8]) -> Result<Self, SchemaBindingError> {
214        if bytes.len() > MAX_SCHEMA_BINDING_BYTES {
215            return Err(SchemaBindingError::Invalid(
216                "encoded contract is oversized".into(),
217            ));
218        }
219        let binding: Self = serde_json::from_slice(bytes)?;
220        binding.canonical_bytes()?;
221        Ok(binding)
222    }
223}
224
225#[cfg(test)]
226mod tests;
227mod validation;
228pub use validation::same_logical_type;
laminar_core/schema_binding/mod.rs

laminar_core/schema_binding/
mod.rs
