distributed/table/
registry.rs

1//! Registry and schema-management adapter surface for table schemas.
2
3use std::collections::{BTreeMap, BTreeSet};
4
5use crate::read_model::RelationalReadModel;
6
7use super::{RelationshipDef, RelationshipKind, TableSchema, TableStoreError};
8
9/// Registry of table schemas an adapter should manage.
10#[derive(Clone, Debug, Default, PartialEq, Eq)]
11pub struct TableSchemaRegistry {
12    schemas_by_table: BTreeMap<String, TableSchema>,
13    tables_by_model: BTreeMap<String, String>,
14}
15
16impl TableSchemaRegistry {
17    pub fn new() -> Self {
18        Self::default()
19    }
20
21    pub fn register<M>(&mut self) -> Result<&mut Self, TableStoreError>
22    where
23        M: RelationalReadModel,
24    {
25        self.register_schema(M::schema().clone())
26    }
27
28    pub fn register_schema(&mut self, schema: TableSchema) -> Result<&mut Self, TableStoreError> {
29        schema.validate()?;
30
31        if self.schemas_by_table.contains_key(&schema.table_name) {
32            return Err(TableStoreError::Metadata(format!(
33                "table schema registry already contains table `{}`",
34                schema.table_name
35            )));
36        }
37        if self.tables_by_model.contains_key(&schema.model_name) {
38            return Err(TableStoreError::Metadata(format!(
39                "table schema registry already contains model `{}`",
40                schema.model_name
41            )));
42        }
43
44        self.tables_by_model
45            .insert(schema.model_name.clone(), schema.table_name.clone());
46        self.schemas_by_table
47            .insert(schema.table_name.clone(), schema);
48        Ok(self)
49    }
50
51    pub fn len(&self) -> usize {
52        self.schemas_by_table.len()
53    }
54
55    pub fn is_empty(&self) -> bool {
56        self.schemas_by_table.is_empty()
57    }
58
59    pub fn schemas(&self) -> impl Iterator<Item = &TableSchema> {
60        self.schemas_by_table.values()
61    }
62
63    pub fn table_names(&self) -> impl Iterator<Item = &str> {
64        self.schemas_by_table.keys().map(String::as_str)
65    }
66
67    pub fn schema_for_table(&self, table_name: &str) -> Option<&TableSchema> {
68        self.schemas_by_table.get(table_name)
69    }
70
71    pub fn schema_for_model(&self, model_name: &str) -> Option<&TableSchema> {
72        self.tables_by_model
73            .get(model_name)
74            .and_then(|table_name| self.schema_for_table(table_name))
75    }
76
77    pub fn validate(&self) -> Result<(), TableStoreError> {
78        let table_names = self
79            .schemas_by_table
80            .keys()
81            .cloned()
82            .collect::<BTreeSet<_>>();
83        for schema in self.schemas() {
84            schema.validate()?;
85            self.validate_column_foreign_keys(schema, &table_names)?;
86            self.validate_schema_foreign_keys(schema, &table_names)?;
87            self.validate_relationships(schema, &table_names)?;
88        }
89
90        Ok(())
91    }
92
93    fn validate_column_foreign_keys(
94        &self,
95        schema: &TableSchema,
96        table_names: &BTreeSet<String>,
97    ) -> Result<(), TableStoreError> {
98        for column in &schema.columns {
99            let Some(foreign_key) = &column.foreign_key else {
100                continue;
101            };
102            self.validate_foreign_key_target(
103                &schema.model_name,
104                &schema.table_name,
105                &column.column_name,
106                &foreign_key.table,
107                &foreign_key.column,
108                table_names,
109            )?;
110        }
111        Ok(())
112    }
113
114    fn validate_schema_foreign_keys(
115        &self,
116        schema: &TableSchema,
117        table_names: &BTreeSet<String>,
118    ) -> Result<(), TableStoreError> {
119        for foreign_key in &schema.foreign_keys {
120            self.validate_foreign_key_target(
121                &schema.model_name,
122                &schema.table_name,
123                "",
124                &foreign_key.table,
125                &foreign_key.column,
126                table_names,
127            )?;
128        }
129        Ok(())
130    }
131
132    fn validate_relationships(
133        &self,
134        schema: &TableSchema,
135        table_names: &BTreeSet<String>,
136    ) -> Result<(), TableStoreError> {
137        for relationship in &schema.relationships {
138            let target_schema = self
139                .schema_for_model(&relationship.target_model)
140                .ok_or_else(|| {
141                    TableStoreError::Metadata(format!(
142                        "model `{}` relationship `{}` targets unregistered model `{}`",
143                        schema.model_name, relationship.field_name, relationship.target_model
144                    ))
145                })?;
146
147            if let Some(through) = relationship.through.as_deref() {
148                if !table_names.contains(through) {
149                    return Err(TableStoreError::Metadata(format!(
150                        "model `{}` relationship `{}` references unregistered join table `{}`",
151                        schema.model_name, relationship.field_name, through
152                    )));
153                }
154            }
155
156            match relationship.kind {
157                RelationshipKind::HasMany | RelationshipKind::BelongsTo => {
158                    let _ = resolve_direct_join_keys(schema, relationship, target_schema)?;
159                }
160                RelationshipKind::ManyToMany => {
161                    let through = relationship.through.as_deref().ok_or_else(|| {
162                        TableStoreError::Metadata(format!(
163                            "model `{}` relationship `{}` many-to-many must declare `through`",
164                            schema.model_name, relationship.field_name
165                        ))
166                    })?;
167                    let through_schema = self.schema_for_table(through).ok_or_else(|| {
168                        TableStoreError::Metadata(format!(
169                            "model `{}` relationship `{}` references unavailable join table `{}`",
170                            schema.model_name, relationship.field_name, through
171                        ))
172                    })?;
173                    let _ =
174                        resolve_m2m_join_keys(schema, relationship, through_schema, target_schema)?;
175                }
176            }
177        }
178        Ok(())
179    }
180
181    fn validate_foreign_key_target(
182        &self,
183        model_name: &str,
184        table_name: &str,
185        column_name: &str,
186        target_table: &str,
187        target_column: &str,
188        table_names: &BTreeSet<String>,
189    ) -> Result<(), TableStoreError> {
190        if !table_names.contains(target_table) {
191            return Err(TableStoreError::Metadata(format!(
192                "model `{model_name}` table `{table_name}` references unregistered foreign-key table `{target_table}`"
193            )));
194        }
195
196        let target_schema = self.schemas_by_table.get(target_table).ok_or_else(|| {
197            TableStoreError::Metadata(format!(
198                "model `{model_name}` references unavailable foreign-key table `{target_table}`"
199            ))
200        })?;
201        if !target_schema
202            .columns
203            .iter()
204            .any(|column| column.column_name == target_column)
205        {
206            let local_column = if column_name.is_empty() {
207                "schema".to_string()
208            } else {
209                format!("column `{column_name}`")
210            };
211            return Err(TableStoreError::Metadata(format!(
212                "model `{model_name}` {local_column} references missing foreign-key column `{target_table}.{target_column}`"
213            )));
214        }
215
216        Ok(())
217    }
218}
219
220/// One through-table column paired with one end-table primary-key column.
221#[derive(Clone, Debug, PartialEq, Eq)]
222pub struct JoinColumnPair {
223    pub through_column: String,
224    pub end_column: String,
225}
226
227impl JoinColumnPair {
228    pub fn new(through_column: impl Into<String>, end_column: impl Into<String>) -> Self {
229        Self {
230            through_column: through_column.into(),
231            end_column: end_column.into(),
232        }
233    }
234}
235
236/// Through-table join keys for both ends of a many-to-many relationship.
237///
238/// `parent` pairs through columns with the source model's primary key.
239/// `target` pairs through columns with the target model's primary key.
240#[derive(Clone, Debug, PartialEq, Eq)]
241pub struct M2mJoinKeys {
242    pub parent: Vec<JoinColumnPair>,
243    pub target: Vec<JoinColumnPair>,
244}
245
246/// One foreign-key column paired with one primary-key column for a direct join.
247#[derive(Clone, Debug, PartialEq, Eq)]
248pub struct DirectJoinPair {
249    pub foreign_key_column: String,
250    pub primary_key_column: String,
251}
252
253impl DirectJoinPair {
254    pub fn new(
255        foreign_key_column: impl Into<String>,
256        primary_key_column: impl Into<String>,
257    ) -> Self {
258        Self {
259            foreign_key_column: foreign_key_column.into(),
260            primary_key_column: primary_key_column.into(),
261        }
262    }
263}
264
265/// Resolve `has_many` / `belongs_to` join equalities.
266///
267/// `foreign_key` lists the FK-holding table's columns in the other end's PK
268/// order, same arity as that PK (comma-separated when more than one).
269///
270/// - **HasMany**: FK columns live on the target; PK is the source.
271/// - **BelongsTo**: FK columns live on the source; PK is the target.
272pub fn resolve_direct_join_keys(
273    source: &TableSchema,
274    relationship: &RelationshipDef,
275    target: &TableSchema,
276) -> Result<Vec<DirectJoinPair>, TableStoreError> {
277    let (fk_schema, pk_schema) = match relationship.kind {
278        RelationshipKind::HasMany => (target, source),
279        RelationshipKind::BelongsTo => (source, target),
280        RelationshipKind::ManyToMany => {
281            return Err(TableStoreError::Metadata(format!(
282                "model `{}` relationship `{}` must be has_many or belongs_to to resolve a direct join",
283                source.model_name, relationship.field_name
284            )));
285        }
286    };
287    let pk_columns = &pk_schema.primary_key.columns;
288    if pk_columns.is_empty() {
289        return Err(TableStoreError::Metadata(format!(
290            "model `{}` relationship `{}` cannot join because `{}` has an empty primary key",
291            source.model_name, relationship.field_name, pk_schema.model_name
292        )));
293    }
294    let Some(fk_names) = parse_explicit_through_columns(
295        source,
296        relationship,
297        "foreign_key",
298        relationship.foreign_key.as_deref(),
299    )?
300    else {
301        return Err(TableStoreError::Metadata(format!(
302            "model `{}` relationship `{}` must declare a foreign key",
303            source.model_name, relationship.field_name
304        )));
305    };
306    if fk_names.len() != pk_columns.len() {
307        return Err(TableStoreError::Metadata(format!(
308            "model `{}` relationship `{}` foreign_key lists {} column(s) but `{}` primary key has {}",
309            source.model_name,
310            relationship.field_name,
311            fk_names.len(),
312            pk_schema.model_name,
313            pk_columns.len()
314        )));
315    }
316    let mut pairs = Vec::with_capacity(pk_columns.len());
317    for (fk_name, pk_column) in fk_names.iter().zip(pk_columns) {
318        let foreign_key_column = column_name_on(fk_schema, fk_name).ok_or_else(|| {
319            let side = match relationship.kind {
320                RelationshipKind::HasMany => "target",
321                _ => "source",
322            };
323            TableStoreError::Metadata(format!(
324                "model `{}` relationship `{}` foreign key `{fk_name}` is not a column on {side} model `{}`",
325                source.model_name, relationship.field_name, fk_schema.model_name
326            ))
327        })?;
328        pairs.push(DirectJoinPair::new(foreign_key_column, pk_column.clone()));
329    }
330    Ok(pairs)
331}
332
333/// Resolve every through-column ↔ end-PK pair for a many-to-many relationship.
334///
335/// Through columns are either:
336/// - the end's primary-key column names (present on the join table), or
337/// - `foreign_key` / `target_foreign_key` listing through columns in PK order,
338///   same arity as that end's primary key (comma-separated when more than one).
339///
340/// When those are absent, a unique through-column foreign-key reference to each
341/// PK column is accepted. Compile SQL must not invent a second pairing.
342pub fn resolve_m2m_join_keys(
343    source: &TableSchema,
344    relationship: &RelationshipDef,
345    through_schema: &TableSchema,
346    target_schema: &TableSchema,
347) -> Result<M2mJoinKeys, TableStoreError> {
348    if !matches!(relationship.kind, RelationshipKind::ManyToMany) {
349        return Err(TableStoreError::Metadata(format!(
350            "model `{}` relationship `{}` must be many-to-many to resolve join keys",
351            source.model_name, relationship.field_name
352        )));
353    }
354    Ok(M2mJoinKeys {
355        parent: m2m_key_pairs(
356            source,
357            relationship,
358            through_schema,
359            source,
360            "foreign_key",
361            relationship.foreign_key.as_deref(),
362        )?,
363        target: m2m_key_pairs(
364            source,
365            relationship,
366            through_schema,
367            target_schema,
368            "target_foreign_key",
369            relationship.target_foreign_key.as_deref(),
370        )?,
371    })
372}
373
374fn m2m_key_pairs(
375    source: &TableSchema,
376    relationship: &RelationshipDef,
377    through: &TableSchema,
378    end: &TableSchema,
379    field: &str,
380    explicit: Option<&str>,
381) -> Result<Vec<JoinColumnPair>, TableStoreError> {
382    let pk_columns = &end.primary_key.columns;
383    if pk_columns.is_empty() {
384        return Err(TableStoreError::Metadata(format!(
385            "model `{}` relationship `{}` cannot join through `{}` because `{}` has an empty primary key",
386            source.model_name, relationship.field_name, through.table_name, end.model_name
387        )));
388    }
389    if let Some(through_names) =
390        parse_explicit_through_columns(source, relationship, field, explicit)?
391    {
392        if through_names.len() != pk_columns.len() {
393            return Err(TableStoreError::Metadata(format!(
394                "model `{}` relationship `{}` {field} lists {} through column(s) but `{}` primary key has {}",
395                source.model_name,
396                relationship.field_name,
397                through_names.len(),
398                end.model_name,
399                pk_columns.len()
400            )));
401        }
402        let mut pairs = Vec::with_capacity(pk_columns.len());
403        for (through_name, end_column) in through_names.iter().zip(pk_columns) {
404            let through_column = column_name_on(through, through_name).ok_or_else(|| {
405                TableStoreError::Metadata(format!(
406                    "model `{}` relationship `{}` {field} `{through_name}` is not a column on join table `{}`",
407                    source.model_name, relationship.field_name, through.table_name
408                ))
409            })?;
410            pairs.push(JoinColumnPair::new(through_column, end_column.clone()));
411        }
412        return Ok(pairs);
413    }
414
415    let mut pairs = Vec::with_capacity(pk_columns.len());
416    let mut missing = Vec::new();
417    for end_column in pk_columns {
418        match column_name_on(through, end_column) {
419            Some(through_column) => {
420                pairs.push(JoinColumnPair::new(through_column, end_column.clone()));
421            }
422            None => missing.push(end_column.as_str()),
423        }
424    }
425    if missing.is_empty() {
426        return Ok(pairs);
427    }
428    if let Some(inferred) = infer_m2m_pairs_from_foreign_keys(through, end) {
429        return Ok(inferred);
430    }
431    Err(TableStoreError::Metadata(format!(
432        "model `{}` relationship `{}` cannot resolve {field} on join table `{}` for `{}` primary key [{}] \
433         (missing same-named through columns: {}); declare `{field}` as a PK-order through-column list",
434        source.model_name,
435        relationship.field_name,
436        through.table_name,
437        end.model_name,
438        pk_columns.join(", "),
439        missing.join(", ")
440    )))
441}
442
443fn parse_explicit_through_columns(
444    source: &TableSchema,
445    relationship: &RelationshipDef,
446    field: &str,
447    value: Option<&str>,
448) -> Result<Option<Vec<String>>, TableStoreError> {
449    let Some(raw) = value else {
450        return Ok(None);
451    };
452    if raw.trim().is_empty() {
453        return Err(TableStoreError::Metadata(format!(
454            "model `{}` relationship `{}` {field} must not be empty",
455            source.model_name, relationship.field_name
456        )));
457    }
458    let mut columns = Vec::new();
459    for part in raw.split(',') {
460        let name = part.trim();
461        if name.is_empty() {
462            return Err(TableStoreError::Metadata(format!(
463                "model `{}` relationship `{}` {field} lists an empty through column",
464                source.model_name, relationship.field_name
465            )));
466        }
467        columns.push(name.to_string());
468    }
469    Ok(Some(columns))
470}
471
472fn infer_m2m_pairs_from_foreign_keys(
473    through: &TableSchema,
474    end: &TableSchema,
475) -> Option<Vec<JoinColumnPair>> {
476    let mut pairs = Vec::with_capacity(end.primary_key.columns.len());
477    for end_column in &end.primary_key.columns {
478        let matches: Vec<&str> = through
479            .columns
480            .iter()
481            .filter(|column| {
482                column
483                    .foreign_key
484                    .as_ref()
485                    .is_some_and(|fk| fk.table == end.table_name && fk.column == *end_column)
486            })
487            .map(|column| column.column_name.as_str())
488            .collect();
489        let [only] = matches.as_slice() else {
490            return None;
491        };
492        pairs.push(JoinColumnPair::new(*only, end_column.clone()));
493    }
494    Some(pairs)
495}
496
497fn column_name_on<'a>(schema: &'a TableSchema, name: &str) -> Option<&'a str> {
498    schema.columns.iter().find_map(|column| {
499        if column.column_name == name || column.field_name == name {
500            Some(column.column_name.as_str())
501        } else {
502            None
503        }
504    })
505}
506
507/// Schema lifecycle operations an adapter can support.
508#[derive(Clone, Debug, Default, PartialEq, Eq)]
509pub struct TableSchemaAdapterCapabilities {
510    pub migration_artifacts: bool,
511    pub schema_verification: bool,
512    pub dev_bootstrap: bool,
513}
514
515impl TableSchemaAdapterCapabilities {
516    pub fn all() -> Self {
517        Self {
518            migration_artifacts: true,
519            schema_verification: true,
520            dev_bootstrap: true,
521        }
522    }
523}
524
525/// Generated or user-consumable migration artifact for registered schemas.
526#[derive(Clone, Debug, PartialEq, Eq)]
527pub struct TableMigrationArtifact {
528    pub name: String,
529    pub statements: Vec<String>,
530}
531
532impl TableMigrationArtifact {
533    pub fn new(name: impl Into<String>, statements: impl IntoIterator<Item = String>) -> Self {
534        Self {
535            name: name.into(),
536            statements: statements.into_iter().collect(),
537        }
538    }
539}
540
541/// Result of verifying registered metadata against an adapter-owned schema.
542#[derive(Clone, Debug, Default, PartialEq, Eq)]
543pub struct TableSchemaVerification {
544    pub issues: Vec<TableSchemaIssue>,
545}
546
547impl TableSchemaVerification {
548    pub fn verified() -> Self {
549        Self::default()
550    }
551
552    pub fn is_verified(&self) -> bool {
553        self.issues.is_empty()
554    }
555}
556
557/// Adapter-facing schema verification issue.
558#[derive(Clone, Debug, PartialEq, Eq)]
559pub struct TableSchemaIssue {
560    pub table_name: String,
561    pub column_name: Option<String>,
562    pub kind: TableSchemaIssueKind,
563    pub message: String,
564}
565
566impl TableSchemaIssue {
567    pub fn new(
568        table_name: impl Into<String>,
569        column_name: Option<impl Into<String>>,
570        kind: TableSchemaIssueKind,
571        message: impl Into<String>,
572    ) -> Self {
573        Self {
574            table_name: table_name.into(),
575            column_name: column_name.map(Into::into),
576            kind,
577            message: message.into(),
578        }
579    }
580}
581
582#[derive(Clone, Debug, PartialEq, Eq)]
583pub enum TableSchemaIssueKind {
584    MissingTable,
585    MissingColumn,
586    TypeMismatch,
587    PrimaryKeyMismatch,
588    ForeignKeyMismatch,
589    IndexMismatch,
590    NullabilityMismatch,
591    DefaultMismatch,
592    VersionColumnMismatch,
593    Unsupported(String),
594}
595
596/// Result of an explicit dev/test schema bootstrap operation.
597#[derive(Clone, Debug, Default, PartialEq, Eq)]
598pub struct TableSchemaBootstrap {
599    pub bootstrapped_tables: Vec<String>,
600}
601
602impl TableSchemaBootstrap {
603    pub fn new(bootstrapped_tables: impl IntoIterator<Item = String>) -> Self {
604        Self {
605            bootstrapped_tables: bootstrapped_tables.into_iter().collect(),
606        }
607    }
608}
609
610/// Adapter contract for schema generation, verification, and dev/test bootstrap.
611pub trait TableSchemaAdapter {
612    fn schema_capabilities(&self) -> TableSchemaAdapterCapabilities;
613
614    fn generate_migration_artifacts(
615        &self,
616        _registry: &TableSchemaRegistry,
617    ) -> Result<Vec<TableMigrationArtifact>, TableStoreError> {
618        Err(TableStoreError::Metadata(
619            "read-model schema adapter does not support migration artifact generation".into(),
620        ))
621    }
622
623    fn verify_schema(
624        &self,
625        _registry: &TableSchemaRegistry,
626    ) -> Result<TableSchemaVerification, TableStoreError> {
627        Err(TableStoreError::Metadata(
628            "read-model schema adapter does not support startup schema verification".into(),
629        ))
630    }
631
632    fn bootstrap_schema_for_dev(
633        &self,
634        _registry: &TableSchemaRegistry,
635    ) -> Result<TableSchemaBootstrap, TableStoreError> {
636        Err(TableStoreError::Metadata(
637            "read-model schema adapter does not support explicit dev/test bootstrap".into(),
638        ))
639    }
640}
641
642#[cfg(test)]
643mod m2m_join_key_tests {
644    use super::*;
645    use crate::table::{
646        ColumnType, ForeignKey, PrimaryKey, RelationshipDef, RelationshipKind, TableColumn,
647        TableKind, TableSchema,
648    };
649
650    fn pk_column(name: &str) -> TableColumn {
651        TableColumn {
652            primary_key: true,
653            ..TableColumn::new(name, name, ColumnType::Text)
654        }
655    }
656
657    fn column(name: &str) -> TableColumn {
658        TableColumn::new(name, name, ColumnType::Text)
659    }
660
661    fn schema(
662        model: &str,
663        table: &str,
664        columns: Vec<TableColumn>,
665        pk: &[&str],
666        relationships: Vec<RelationshipDef>,
667    ) -> TableSchema {
668        TableSchema {
669            model_name: model.into(),
670            table_name: table.into(),
671            columns,
672            primary_key: PrimaryKey::new(pk.iter().copied()),
673            version_column: None,
674            foreign_keys: Vec::new(),
675            indexes: Vec::new(),
676            relationships,
677            kind: TableKind::ReadModel,
678        }
679    }
680
681    fn labels() -> TableSchema {
682        schema(
683            "LabelView",
684            "labels",
685            vec![pk_column("label_id"), column("name")],
686            &["label_id"],
687            Vec::new(),
688        )
689    }
690
691    fn projects(rel: RelationshipDef) -> TableSchema {
692        schema(
693            "ProjectView",
694            "projects",
695            vec![pk_column("workspace_id"), pk_column("path"), column("kind")],
696            &["workspace_id", "path"],
697            vec![rel],
698        )
699    }
700
701    fn project_labels() -> TableSchema {
702        schema(
703            "ProjectLabel",
704            "project_labels",
705            vec![
706                pk_column("workspace_id"),
707                pk_column("path"),
708                pk_column("label_id"),
709            ],
710            &["workspace_id", "path", "label_id"],
711            Vec::new(),
712        )
713    }
714
715    fn labels_rel(foreign_key: Option<&str>, target_foreign_key: Option<&str>) -> RelationshipDef {
716        RelationshipDef {
717            field_name: "labels".into(),
718            kind: RelationshipKind::ManyToMany,
719            target_model: "LabelView".into(),
720            foreign_key: foreign_key.map(str::to_string),
721            through: Some("project_labels".into()),
722            target_foreign_key: target_foreign_key.map(str::to_string),
723        }
724    }
725
726    #[test]
727    fn same_named_through_columns_pair_composite_and_single_keys() {
728        let source = projects(labels_rel(None, None));
729        let keys = resolve_m2m_join_keys(
730            &source,
731            &source.relationships[0],
732            &project_labels(),
733            &labels(),
734        )
735        .unwrap();
736        assert_eq!(
737            keys.parent,
738            vec![
739                JoinColumnPair::new("workspace_id", "workspace_id"),
740                JoinColumnPair::new("path", "path"),
741            ]
742        );
743        assert_eq!(
744            keys.target,
745            vec![JoinColumnPair::new("label_id", "label_id")]
746        );
747    }
748
749    #[test]
750    fn explicit_pk_order_list_renames_composite_through_columns() {
751        let mut through = project_labels();
752        through.columns = vec![
753            pk_column("project_workspace_id"),
754            pk_column("project_path"),
755            pk_column("tag_id"),
756        ];
757        through.primary_key = PrimaryKey::new(["project_workspace_id", "project_path", "tag_id"]);
758        let source = projects(labels_rel(
759            Some("project_workspace_id,project_path"),
760            Some("tag_id"),
761        ));
762        let keys =
763            resolve_m2m_join_keys(&source, &source.relationships[0], &through, &labels()).unwrap();
764        assert_eq!(
765            keys.parent,
766            vec![
767                JoinColumnPair::new("project_workspace_id", "workspace_id"),
768                JoinColumnPair::new("project_path", "path"),
769            ]
770        );
771        assert_eq!(keys.target, vec![JoinColumnPair::new("tag_id", "label_id")]);
772    }
773
774    #[test]
775    fn partial_explicit_foreign_key_does_not_silently_ignore_composite_pk() {
776        let source = projects(labels_rel(Some("workspace_id"), Some("label_id")));
777        let error = resolve_m2m_join_keys(
778            &source,
779            &source.relationships[0],
780            &project_labels(),
781            &labels(),
782        )
783        .unwrap_err()
784        .to_string();
785        assert!(
786            error.contains("lists 1 through column") && error.contains("primary key has 2"),
787            "{error}"
788        );
789    }
790
791    #[test]
792    fn empty_string_foreign_key_is_not_a_missing_sentinel() {
793        let source = projects(labels_rel(Some(""), None));
794        let error = resolve_m2m_join_keys(
795            &source,
796            &source.relationships[0],
797            &project_labels(),
798            &labels(),
799        )
800        .unwrap_err()
801        .to_string();
802        assert!(error.contains("foreign_key must not be empty"), "{error}");
803    }
804
805    #[test]
806    fn infers_renamed_through_columns_from_column_foreign_keys() {
807        let mut through = project_labels();
808        through.columns = vec![
809            TableColumn {
810                primary_key: true,
811                foreign_key: Some(ForeignKey::new("projects", "workspace_id")),
812                ..TableColumn::new(
813                    "project_workspace_id",
814                    "project_workspace_id",
815                    ColumnType::Text,
816                )
817            },
818            TableColumn {
819                primary_key: true,
820                foreign_key: Some(ForeignKey::new("projects", "path")),
821                ..TableColumn::new("project_path", "project_path", ColumnType::Text)
822            },
823            TableColumn {
824                primary_key: true,
825                foreign_key: Some(ForeignKey::new("labels", "label_id")),
826                ..TableColumn::new("tag_id", "tag_id", ColumnType::Text)
827            },
828        ];
829        through.primary_key = PrimaryKey::new(["project_workspace_id", "project_path", "tag_id"]);
830        let source = projects(labels_rel(None, None));
831        let keys =
832            resolve_m2m_join_keys(&source, &source.relationships[0], &through, &labels()).unwrap();
833        assert_eq!(
834            keys.parent,
835            vec![
836                JoinColumnPair::new("project_workspace_id", "workspace_id"),
837                JoinColumnPair::new("project_path", "path"),
838            ]
839        );
840        assert_eq!(keys.target, vec![JoinColumnPair::new("tag_id", "label_id")]);
841    }
842
843    fn files_rel(foreign_key: &str) -> RelationshipDef {
844        RelationshipDef {
845            field_name: "files".into(),
846            kind: RelationshipKind::HasMany,
847            target_model: "ProjectFileView".into(),
848            foreign_key: Some(foreign_key.into()),
849            through: None,
850            target_foreign_key: None,
851        }
852    }
853
854    fn project_files() -> TableSchema {
855        schema(
856            "ProjectFileView",
857            "project_files",
858            vec![
859                pk_column("workspace_id"),
860                pk_column("path"),
861                pk_column("file_id"),
862            ],
863            &["workspace_id", "path", "file_id"],
864            Vec::new(),
865        )
866    }
867
868    #[test]
869    fn has_many_pairs_composite_parent_key_in_pk_order() {
870        let source = projects(files_rel("workspace_id,path"));
871        let pairs =
872            resolve_direct_join_keys(&source, &source.relationships[0], &project_files()).unwrap();
873        assert_eq!(
874            pairs,
875            vec![
876                DirectJoinPair::new("workspace_id", "workspace_id"),
877                DirectJoinPair::new("path", "path"),
878            ]
879        );
880    }
881
882    #[test]
883    fn belongs_to_pairs_composite_target_key_in_pk_order() {
884        let target = projects(files_rel("workspace_id,path"));
885        let source = schema(
886            "ProjectFileView",
887            "project_files",
888            vec![
889                pk_column("workspace_id"),
890                pk_column("path"),
891                pk_column("file_id"),
892            ],
893            &["workspace_id", "path", "file_id"],
894            vec![RelationshipDef {
895                field_name: "project".into(),
896                kind: RelationshipKind::BelongsTo,
897                target_model: "ProjectView".into(),
898                foreign_key: Some("workspace_id,path".into()),
899                through: None,
900                target_foreign_key: None,
901            }],
902        );
903        let pairs = resolve_direct_join_keys(&source, &source.relationships[0], &target).unwrap();
904        assert_eq!(
905            pairs,
906            vec![
907                DirectJoinPair::new("workspace_id", "workspace_id"),
908                DirectJoinPair::new("path", "path"),
909            ]
910        );
911    }
912
913    #[test]
914    fn partial_direct_foreign_key_does_not_silently_take_the_first_pk_column() {
915        let source = projects(files_rel("workspace_id"));
916        let error = resolve_direct_join_keys(&source, &source.relationships[0], &project_files())
917            .unwrap_err()
918            .to_string();
919        assert!(
920            error.contains("lists 1 column") && error.contains("primary key has 2"),
921            "{error}"
922        );
923    }
924
925    #[test]
926    fn renamed_direct_foreign_key_columns_pair_in_pk_order() {
927        let mut files = project_files();
928        files.columns = vec![
929            pk_column("project_workspace_id"),
930            pk_column("project_path"),
931            pk_column("file_id"),
932        ];
933        files.primary_key = PrimaryKey::new(["project_workspace_id", "project_path", "file_id"]);
934        let source = projects(files_rel("project_workspace_id,project_path"));
935        let pairs = resolve_direct_join_keys(&source, &source.relationships[0], &files).unwrap();
936        assert_eq!(
937            pairs,
938            vec![
939                DirectJoinPair::new("project_workspace_id", "workspace_id"),
940                DirectJoinPair::new("project_path", "path"),
941            ]
942        );
943    }
944}
distributed/table/registry.rs

distributed/table/
registry.rs
