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

laser_wire/
destination.rs
