laser_wire/
destination.rs

1use crate::agent::wire_id;
2use crate::control::ProjectionId;
3use crate::error::InvalidError;
4use crate::schema::{LogicalSchemaRef, UuidValue};
5use crate::source::{SourceIncarnation, SourceScope};
6use crate::validate::Validate;
7use serde::{Deserialize, Serialize};
8use std::collections::BTreeSet;
9
10pub const MAX_DESTINATION_NAME_BYTES: usize = 255;
11pub const MAX_NAMESPACE_PARTS: usize = 16;
12pub const MAX_TABLE_NAME_BYTES: usize = 255;
13pub const MAX_EXPLICIT_PARTITION_STARTS: usize = 65_536;
14
15wire_id!(
16    /// Immutable identity of one materialization destination across generations.
17    DestinationId
18);
19
20wire_id!(
21    /// Stable identity of one supervisor-owned backend resource.
22    BackendResourceId
23);
24
25wire_id!(
26    /// Immutable identity of one logical query route across generations.
27    QueryRouteId
28);
29
30wire_id!(
31    /// Identity of one accepted asynchronous destination operation.
32    DestinationOperationId
33);
34
35#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
36pub struct ProjectionRef {
37    pub id: ProjectionId,
38    pub version: u32,
39}
40
41impl Validate for ProjectionRef {
42    fn validate(&self) -> Result<(), InvalidError> {
43        if self.id.as_str().is_empty() {
44            return Err(InvalidError::new("projection id must not be empty"));
45        }
46        if self.version == 0 {
47            return Err(InvalidError::new("projection version must be nonzero"));
48        }
49        Ok(())
50    }
51}
52
53#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
54pub struct BackendBinding {
55    pub resource_id: BackendResourceId,
56    pub generation: u64,
57}
58
59impl Validate for BackendBinding {
60    fn validate(&self) -> Result<(), InvalidError> {
61        if self.resource_id.as_u128() == 0 {
62            return Err(InvalidError::new("backend resource id must be nonzero"));
63        }
64        if self.generation == 0 {
65            return Err(InvalidError::new("backend generation must be nonzero"));
66        }
67        Ok(())
68    }
69}
70
71#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
72pub struct PhysicalTable {
73    pub namespace: Vec<String>,
74    pub table: String,
75    #[serde(default, skip_serializing_if = "Option::is_none")]
76    pub expected_table_uuid: Option<UuidValue>,
77}
78
79impl Validate for PhysicalTable {
80    fn validate(&self) -> Result<(), InvalidError> {
81        if self.namespace.is_empty() || self.namespace.len() > MAX_NAMESPACE_PARTS {
82            return Err(InvalidError::new(format!(
83                "table namespace must contain 1..={MAX_NAMESPACE_PARTS} parts"
84            )));
85        }
86        for part in &self.namespace {
87            validate_table_name("namespace part", part)?;
88        }
89        validate_table_name("table", &self.table)?;
90        if let Some(table_uuid) = &self.expected_table_uuid
91            && table_uuid.as_bytes().len() != UuidValue::BYTES
92        {
93            return Err(InvalidError::new(format!(
94                "table UUID must be {} bytes",
95                UuidValue::BYTES
96            )));
97        }
98        Ok(())
99    }
100}
101
102#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
103#[serde(rename_all = "snake_case")]
104#[non_exhaustive]
105pub enum FileFormat {
106    Parquet,
107}
108
109#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
110#[serde(rename_all = "snake_case")]
111#[non_exhaustive]
112pub enum TableFormat {
113    IcebergV2,
114}
115
116#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
117#[serde(tag = "kind", rename_all = "snake_case")]
118#[non_exhaustive]
119pub enum StartPolicy {
120    Beginning,
121    CapturedLatest,
122    Explicit { partitions: Vec<PartitionStart> },
123}
124
125#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
126pub struct PartitionStart {
127    pub incarnation: SourceIncarnation,
128    pub next_offset: u64,
129}
130
131#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
132#[serde(rename_all = "snake_case")]
133#[non_exhaustive]
134pub enum NewPartitionPolicy {
135    Beginning,
136    CapturedLatest,
137    Reject,
138}
139
140#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
141#[serde(rename_all = "snake_case")]
142#[non_exhaustive]
143pub enum RecreatedPartitionPolicy {
144    Reject,
145}
146
147#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
148#[serde(rename_all = "snake_case")]
149#[non_exhaustive]
150pub enum DestinationErrorPolicy {
151    Block,
152}
153
154#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
155#[serde(rename_all = "snake_case")]
156#[non_exhaustive]
157pub enum DestinationDesiredState {
158    Disabled,
159    Enabled,
160}
161
162#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
163pub struct MaterializationDestination {
164    pub id: DestinationId,
165    pub generation: u64,
166    pub definition_revision: u64,
167    pub name: String,
168    pub source: SourceScope,
169    pub recreated_partition_policy: RecreatedPartitionPolicy,
170    pub projection: ProjectionRef,
171    pub schema: LogicalSchemaRef,
172    pub backend: BackendBinding,
173    pub table: PhysicalTable,
174    pub file_format: FileFormat,
175    pub table_format: TableFormat,
176    pub start_policy: StartPolicy,
177    pub new_partition_policy: NewPartitionPolicy,
178    pub error_policy: DestinationErrorPolicy,
179    pub desired_state: DestinationDesiredState,
180}
181
182#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
183pub struct QueryRoute {
184    pub id: QueryRouteId,
185    pub generation: u64,
186    pub definition_revision: u64,
187    pub name: String,
188    pub target: QueryRouteTarget,
189    pub desired_state: DestinationDesiredState,
190}
191
192#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
193#[serde(tag = "kind", rename_all = "snake_case")]
194#[non_exhaustive]
195pub enum QueryRouteTarget {
196    Operational {
197        index: String,
198    },
199    Lakehouse {
200        destination_id: DestinationId,
201        destination_generation: u64,
202    },
203}
204
205impl Validate for QueryRoute {
206    fn validate(&self) -> Result<(), InvalidError> {
207        if self.id.as_u128() == 0 || self.generation == 0 || self.definition_revision == 0 {
208            return Err(InvalidError::new(
209                "query route identity, generation, and definition revision must be nonzero",
210            ));
211        }
212        validate_destination_name(&self.name)?;
213        match &self.target {
214            QueryRouteTarget::Operational { index } => validate_table_name("index", index),
215            QueryRouteTarget::Lakehouse {
216                destination_id,
217                destination_generation,
218            } => {
219                if destination_id.as_u128() == 0 || *destination_generation == 0 {
220                    return Err(InvalidError::new(
221                        "query route destination identity and generation must be nonzero",
222                    ));
223                }
224                Ok(())
225            }
226        }
227    }
228}
229
230impl Validate for MaterializationDestination {
231    fn validate(&self) -> Result<(), InvalidError> {
232        if self.id.as_u128() == 0 {
233            return Err(InvalidError::new("destination id must be nonzero"));
234        }
235        if self.generation == 0 {
236            return Err(InvalidError::new("destination generation must be nonzero"));
237        }
238        if self.definition_revision == 0 {
239            return Err(InvalidError::new(
240                "destination definition revision must be nonzero",
241            ));
242        }
243        validate_destination_name(&self.name)?;
244        self.source.validate()?;
245        self.projection.validate()?;
246        self.schema.validate()?;
247        self.backend.validate()?;
248        self.table.validate()?;
249        validate_start_policy(&self.start_policy)
250    }
251}
252
253fn validate_start_policy(policy: &StartPolicy) -> Result<(), InvalidError> {
254    let StartPolicy::Explicit { partitions } = policy else {
255        return Ok(());
256    };
257    if partitions.is_empty() || partitions.len() > MAX_EXPLICIT_PARTITION_STARTS {
258        return Err(InvalidError::new(format!(
259            "explicit start must contain 1..={MAX_EXPLICIT_PARTITION_STARTS} partitions"
260        )));
261    }
262    let mut partition_ids = BTreeSet::new();
263    let mut previous_partition = None;
264    let mut namespace = None;
265    for partition in partitions {
266        partition.incarnation.validate()?;
267        let current = (
268            partition.incarnation.cluster,
269            partition.incarnation.stream_id,
270            partition.incarnation.topic_id,
271        );
272        if namespace.is_some_and(|expected| expected != current) {
273            return Err(InvalidError::new(
274                "explicit start partitions must share one cluster, stream, and topic",
275            ));
276        }
277        namespace = Some(current);
278        if !partition_ids.insert(partition.incarnation.partition_id) {
279            return Err(InvalidError::new(format!(
280                "explicit start repeats partition {}",
281                partition.incarnation.partition_id
282            )));
283        }
284        if previous_partition.is_some_and(|previous| previous >= partition.incarnation.partition_id)
285        {
286            return Err(InvalidError::new(
287                "explicit start partitions must be ordered by ascending partition id",
288            ));
289        }
290        previous_partition = Some(partition.incarnation.partition_id);
291    }
292    Ok(())
293}
294
295fn validate_destination_name(value: &str) -> Result<(), InvalidError> {
296    if value.is_empty() || value.len() > MAX_DESTINATION_NAME_BYTES {
297        return Err(InvalidError::new(format!(
298            "destination name must contain 1..={MAX_DESTINATION_NAME_BYTES} bytes"
299        )));
300    }
301    if value.chars().any(char::is_control) {
302        return Err(InvalidError::new(
303            "destination name contains a control character",
304        ));
305    }
306    Ok(())
307}
308
309fn validate_table_name(label: &str, value: &str) -> Result<(), InvalidError> {
310    if value.is_empty() || value.len() > MAX_TABLE_NAME_BYTES {
311        return Err(InvalidError::new(format!(
312            "{label} must contain 1..={MAX_TABLE_NAME_BYTES} bytes"
313        )));
314    }
315    if value.chars().any(char::is_control) {
316        return Err(InvalidError::new(format!(
317            "{label} contains a control character"
318        )));
319    }
320    Ok(())
321}
322
323#[cfg(test)]
324mod tests {
325    use super::*;
326    use crate::schema::{LogicalSchemaId, SchemaFingerprint};
327    use crate::source::PhysicalClusterIncarnation;
328
329    fn incarnation(partition_id: u32) -> SourceIncarnation {
330        SourceIncarnation {
331            cluster: PhysicalClusterIncarnation::from_u128(1),
332            stream_id: 10,
333            topic_id: 20,
334            partition_id,
335            partition_created_revision: 7,
336        }
337    }
338
339    fn destination() -> MaterializationDestination {
340        MaterializationDestination {
341            id: DestinationId::from_u128(1),
342            generation: 1,
343            definition_revision: 1,
344            name: "readings-lakehouse".to_owned(),
345            source: SourceScope::new("fleet", "readings"),
346            recreated_partition_policy: RecreatedPartitionPolicy::Reject,
347            projection: ProjectionRef {
348                id: ProjectionId::new("reading.v1"),
349                version: 1,
350            },
351            schema: LogicalSchemaRef {
352                id: LogicalSchemaId::from_u128(2),
353                version: 1,
354                fingerprint: SchemaFingerprint::new([3; 32]),
355            },
356            backend: BackendBinding {
357                resource_id: BackendResourceId::from_u128(4),
358                generation: 2,
359            },
360            table: PhysicalTable {
361                namespace: vec!["fleet".to_owned()],
362                table: "readings".to_owned(),
363                expected_table_uuid: None,
364            },
365            file_format: FileFormat::Parquet,
366            table_format: TableFormat::IcebergV2,
367            start_policy: StartPolicy::CapturedLatest,
368            new_partition_policy: NewPartitionPolicy::CapturedLatest,
369            error_policy: DestinationErrorPolicy::Block,
370            desired_state: DestinationDesiredState::Disabled,
371        }
372    }
373
374    #[test]
375    fn given_a_destination_when_validated_then_should_bind_schema_projection_backend_and_table() {
376        destination().validate().expect("destination is valid");
377        let mut invalid = destination();
378        invalid.backend.generation = 0;
379        assert!(invalid.validate().is_err());
380    }
381
382    #[test]
383    fn given_an_explicit_start_when_a_partition_repeats_then_should_reject() {
384        let mut invalid = destination();
385        invalid.start_policy = StartPolicy::Explicit {
386            partitions: vec![
387                PartitionStart {
388                    incarnation: incarnation(1),
389                    next_offset: 0,
390                },
391                PartitionStart {
392                    incarnation: incarnation(1),
393                    next_offset: 1,
394                },
395            ],
396        };
397        assert!(invalid.validate().is_err());
398    }
399
400    #[test]
401    fn given_a_lakehouse_route_when_validated_then_should_require_a_specific_destination_generation()
402     {
403        let route = QueryRoute {
404            id: QueryRouteId::from_u128(5),
405            generation: 1,
406            definition_revision: 1,
407            name: "readings".to_owned(),
408            target: QueryRouteTarget::Lakehouse {
409                destination_id: destination().id,
410                destination_generation: 1,
411            },
412            desired_state: DestinationDesiredState::Enabled,
413        };
414        route.validate().expect("route is valid");
415    }
416}
laser_wire/destination.rs

laser_wire/
destination.rs
