laminar_connectors/schema/
resolution.rs

1//! Shared schema-resolution policy. Discovery never consumes records or creates a target.
2
3use arrow_schema::SchemaRef;
4use serde::Serialize;
5
6use crate::config::ConnectorConfig;
7use crate::error::ConnectorError;
8
9pub use laminar_core::schema_binding::{
10    NativeSchema, NativeSchemaArtifact, SchemaBinding, SchemaDirection, SchemaFieldMapping,
11    SchemaOrigin,
12};
13
14/// Available source of schema authority, independently of transport support.
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
16#[serde(rename_all = "snake_case")]
17pub enum SchemaDiscovery {
18    /// Explicit fields are required; no authoritative discovery exists.
19    Explicit,
20    /// A connector has a fixed protocol schema.
21    BuiltIn,
22    /// Read-only authoritative metadata is available for the listed formats.
23    Metadata,
24    /// Writer fields come from the bound query, without a remote schema authority.
25    Query,
26}
27
28/// External preparation is independent of metadata discovery.
29#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
30#[serde(rename_all = "snake_case")]
31pub enum SchemaPreparation {
32    /// No schema/table mutation is supported.
33    None,
34    /// Table creation requires an explicit connector setting.
35    ExplicitTableCreation,
36    /// Schema registration requires an explicit connector setting.
37    ExplicitRegistration,
38}
39
40/// A factory must declare its direction's schema behavior in its connector metadata.
41#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
42pub struct SchemaCapabilities {
43    /// Schema authority for this registered direction.
44    pub discovery: SchemaDiscovery,
45    /// Formats with authoritative metadata; empty means a native protocol schema.
46    pub metadata_formats: Vec<String>,
47    /// Formats with a fixed decoder schema when this transport also supports explicit formats.
48    pub built_in_formats: Vec<String>,
49    /// Whether writer fields are derived from a bound input schema.
50    pub query_derived: bool,
51    /// Optional sampling is implemented, separately from metadata resolution.
52    pub bounded_sampling: bool,
53    /// Explicit external preparation capability.
54    pub preparation: SchemaPreparation,
55}
56
57impl SchemaCapabilities {
58    pub(crate) fn validate_native_format(
59        &self,
60        config: &ConnectorConfig,
61    ) -> Result<(), ConnectorError> {
62        let native = match self.discovery {
63            SchemaDiscovery::BuiltIn => self.built_in_formats.is_empty(),
64            SchemaDiscovery::Metadata => self.metadata_formats.is_empty(),
65            SchemaDiscovery::Explicit | SchemaDiscovery::Query => false,
66        };
67        if native && config.get("format").is_some() {
68            return Err(ConnectorError::FeatureUnsupported(format!(
69                "{} uses a fixed native protocol; omit FORMAT instead of selecting a serialization codec",
70                config.connector_type()
71            )));
72        }
73        Ok(())
74    }
75
76    /// Conservative declaration for custom connectors with user-supplied source fields
77    /// or query-derived sink fields and no discovery or external preparation.
78    #[must_use]
79    pub fn declared(is_sink: bool) -> Self {
80        Self {
81            discovery: if is_sink {
82                SchemaDiscovery::Query
83            } else {
84                SchemaDiscovery::Explicit
85            },
86            metadata_formats: Vec::new(),
87            built_in_formats: Vec::new(),
88            query_derived: is_sink,
89            bounded_sampling: false,
90            preparation: SchemaPreparation::None,
91        }
92    }
93
94    /// A fixed source protocol schema.
95    #[must_use]
96    pub fn built_in() -> Self {
97        Self {
98            discovery: SchemaDiscovery::BuiltIn,
99            ..Self::declared(false)
100        }
101    }
102
103    /// Authoritative metadata, scoped to the actual codec formats.
104    #[must_use]
105    pub fn metadata(formats: &[&str], is_sink: bool, preparation: SchemaPreparation) -> Self {
106        Self {
107            discovery: SchemaDiscovery::Metadata,
108            metadata_formats: formats.iter().map(|format| (*format).into()).collect(),
109            preparation,
110            ..Self::declared(is_sink)
111        }
112    }
113}
114
115/// Create a validated logical-only reader or writer binding.
116///
117/// # Errors
118/// Rejects empty or incomplete schemas; no fallback to strings or sampling occurs.
119pub fn logical_binding(
120    config: &ConnectorConfig,
121    direction: SchemaDirection,
122    origin: SchemaOrigin,
123    schema: &SchemaRef,
124) -> Result<SchemaBinding, ConnectorError> {
125    SchemaBinding::logical(
126        config.connector_type(),
127        direction,
128        origin,
129        schema.as_ref().clone(),
130    )
131    .map_err(binding_error)
132}
133
134/// Attach an external schema under the default exact-name/type policy.
135///
136/// # Errors
137/// Rejects directional type/nullability mismatches and unmapped sink fields.
138pub fn bind_external(
139    binding: &mut SchemaBinding,
140    schema: &SchemaRef,
141) -> Result<(), ConnectorError> {
142    binding
143        .bind_external(schema.as_ref().clone())
144        .map_err(binding_error)
145}
146
147/// Preserve the existing connector schema-error taxonomy.
148#[must_use]
149#[allow(clippy::needless_pass_by_value)] // Adapter for owned Result::map_err errors.
150pub fn binding_error(error: laminar_core::schema_binding::SchemaBindingError) -> ConnectorError {
151    ConnectorError::SchemaMismatch(error.to_string())
152}
153
154/// Validate that authorized external preparation retains the resolved query contract.
155/// Native identity, writer metadata and mappings may be finalized by the preparation hook.
156///
157/// # Errors
158/// Rejects a changed connector, direction, logical schema or changelog policy, and malformed
159/// native content. Call before catalog publication or connector activation.
160pub fn validate_prepared_writer(
161    config: &ConnectorConfig,
162    resolved: &SchemaBinding,
163    prepared: &SchemaBinding,
164) -> Result<(), ConnectorError> {
165    if prepared.connector != config.connector_type()
166        || prepared.direction != SchemaDirection::Sink
167        || prepared.logical != resolved.logical
168        || prepared.control_fields != resolved.control_fields
169    {
170        return Err(ConnectorError::SchemaMismatch(format!(
171            "{} sink preparation changed the resolved query schema or writer policy",
172            config.connector_type()
173        )));
174    }
175    prepared.canonical_bytes().map_err(binding_error)?;
176    Ok(())
177}
178
179/// Bind a fixed protocol whose decoder emits the complete declared field list.
180///
181/// # Errors
182/// Rejects unsupported projections, ordering and field types; explicit nullability may be weaker.
183pub fn fixed_binding(
184    config: &ConnectorConfig,
185    explicit: Option<SchemaRef>,
186    protocol: &SchemaRef,
187) -> Result<SchemaBinding, ConnectorError> {
188    let origin = if explicit.is_some() {
189        SchemaOrigin::Explicit
190    } else {
191        SchemaOrigin::BuiltIn
192    };
193    let logical = explicit.unwrap_or_else(|| std::sync::Arc::clone(protocol));
194    if logical.fields().len() != protocol.fields().len()
195        || logical
196            .fields()
197            .iter()
198            .zip(protocol.fields())
199            .any(|(logical, native)| logical.name() != native.name())
200    {
201        return Err(ConnectorError::FeatureUnsupported("fixed protocol fields must be declared in their complete protocol order; project in CREATE STREAM".into()));
202    }
203    let mut binding = logical_binding(config, SchemaDirection::Source, origin, &logical)?;
204    bind_external(&mut binding, protocol)?;
205    Ok(binding)
206}
207
208/// Apply a prepared exact-name mapping without casts or per-row conversion.
209///
210/// # Errors
211/// Rejects changed types and actual nulls in non-nullable logical fields.
212pub fn project_batch(
213    batch: &arrow_array::RecordBatch,
214    logical: &SchemaRef,
215) -> Result<arrow_array::RecordBatch, ConnectorError> {
216    let mut columns = Vec::with_capacity(logical.fields().len());
217    for field in logical.fields() {
218        let index = batch.schema().index_of(field.name()).map_err(|_| {
219            ConnectorError::SchemaMismatch(format!("native batch is missing '{}'", field.name()))
220        })?;
221        let column = batch.column(index);
222        if !laminar_core::schema_binding::same_logical_type(column.data_type(), field.data_type())
223            || (!field.is_nullable() && column.null_count() > 0)
224        {
225            return Err(ConnectorError::SchemaMismatch(format!(
226                "native field '{}' changed type or nullability",
227                field.name()
228            )));
229        }
230        let column = if column.data_type() == field.data_type() {
231            std::sync::Arc::clone(column)
232        } else {
233            // Only native metadata differs; Arrow retains the value buffers in this retyping.
234            arrow_cast::cast(column, field.data_type())
235                .map_err(|error| ConnectorError::SchemaMismatch(error.to_string()))?
236        };
237        columns.push(column);
238    }
239    arrow_array::RecordBatch::try_new(std::sync::Arc::clone(logical), columns)
240        .map_err(|error| ConnectorError::SchemaMismatch(error.to_string()))
241}
laminar_connectors/schema/resolution.rs

laminar_connectors/schema/
resolution.rs
