refactor: relationship model to column lists for compound FK (ADR-0043)

Move the FK column fields String->Vec<String> through all six
layers (AddRelationship/SqlForeignKey AST, RelationshipSchema,
metadata, project.yaml, ReadForeignKey, RelationshipEnd). Metadata
stores comma-joined lists in the existing TEXT cells; project.yaml
endpoints now columns: [a, b] (house style). Executor logic is
multi-column ready: resolve_fk_parent_columns (full-PK F-A +
auto-expand F-D), per-pair type-compat, schema_to_ddl multi-column
emission, pragma FK read grouped by id, auto-name + --create-fk
per-column, multi-column teaching echo. Single-column behaviour
preserved (one-element vecs); all 2181 tests green. The grammar to
parse multi-column input lands next.

docs/adr/0043-compound-pk-foreign-key-references.md

@@ -159,9 +159,14 @@ convention `project.yaml` already uses for `primary_key` and index
 - **Metadata** (`__rdbms_playground_relationships`): no
   `CREATE TABLE` change (the `TEXT` columns and
   `PRIMARY KEY (child_table, child_column)` are untouched).
-  `parent_column` / `child_column` store the list as a JSON array
+  `parent_column` / `child_column` store the list **comma-joined**
-  string — uniformly, including `["id"]` for a single column
+  in the same text cell (`a,b`; a single column is just its bare
-  (SQLite has no array type, so a text cell is where a list lives).
+  name). *As-built note:* the ADR first said "JSON array"; the
+  implementation uses a comma delimiter, which is safe because
+  column identifiers are `[A-Za-z0-9_]+` (no commas — `parser.rs`)
+  and simpler (no `serde_json` dependency). This is an internal
+  encoding detail below fork F-B — the user-visible `project.yaml`
+  is still the `columns: [a, b]` list.
   The actual enforced FK lives on the rebuilt child table's DDL
   (`FOREIGN KEY (a, b) REFERENCES P(x, y)`), emitted by
   `schema_to_ddl`, exactly as the single-column FK is today via the

src/app.rs

@@ -1968,12 +1968,14 @@ impl App {
             ),
             C::AddRelationship {
                 parent_table,
-                parent_column,
+                parent_columns,
                 ..
             } => (
                 Operation::AddRelationship,
                 Some(parent_table.as_str()),
-                Some(parent_column.as_str()),
+                // Single-column facts model (ADR-0019): the first PK
+                // column for a compound FK (ADR-0043).
+                parent_columns.first().map(String::as_str),
             ),
             C::DropRelationship { selector } => match selector {
                 RelationshipSelector::Endpoints {

src/dsl/command.rs

@@ -29,15 +29,20 @@ pub struct SqlForeignKey {
     /// FK or an unnamed table FK (auto-named at execution per
     /// ADR-0013).
     pub name: Option<String>,
-    /// The column in the table being created that holds the FK.
+    /// The column(s) in the table being created that hold the FK.
-    pub child_column: String,
+    /// One element for a single-column FK; ordered list for a
+    /// compound FK (ADR-0043). Positionally paired with
+    /// `parent_columns`.
+    pub child_columns: Vec<String>,
     /// The referenced (parent) table — may be the table being created
     /// (a self-referencing FK).
     pub parent_table: String,
-    /// The referenced parent column. `None` for the bare
+    /// The referenced parent column(s), positionally paired with
-    /// `REFERENCES <parent>` form, resolved at execution to the
+    /// `child_columns`. `None` for the bare `REFERENCES <parent>`
-    /// parent's single-column primary key (ADR-0035 §4b, user-confirmed).
+    /// form, resolved at execution to the parent's primary key —
-    pub parent_column: Option<String>,
+    /// the single-column PK, or (ADR-0043 F-D) the full compound PK
+    /// when the child arity matches.
+    pub parent_columns: Option<Vec<String>>,
     pub on_delete: ReferentialAction,
     pub on_update: ReferentialAction,
 }
@@ -253,9 +258,14 @@ pub enum Command {
     AddRelationship {
         name: Option<String>,
         parent_table: String,
-        parent_column: String,
+        /// Parent (referenced) PK column(s); one element for a
+        /// single-column FK, ordered list for a compound FK
+        /// (ADR-0043). Positionally paired with `child_columns`.
+        parent_columns: Vec<String>,
         child_table: String,
-        child_column: String,
+        /// Child (referencing) column(s), positionally paired with
+        /// `parent_columns`; equal, non-zero length.
+        child_columns: Vec<String>,
         on_delete: ReferentialAction,
         on_update: ReferentialAction,
         create_fk: bool,
@@ -1032,11 +1042,26 @@ impl Command {
         match self {
             Self::AddRelationship {
                 parent_table,
-                parent_column,
+                parent_columns,
                 child_table,
-                child_column,
+                child_columns,
                 ..
-            } => format!("from {parent_table}.{parent_column} to {child_table}.{child_column}"),
+            } => {
+                // `from P.col to C.col` (single) or `from P.(a, b) to
+                // C.(x, y)` (compound — ADR-0043), mirroring the DSL.
+                let fmt = |cols: &[String]| {
+                    if cols.len() == 1 {
+                        cols[0].clone()
+                    } else {
+                        format!("({})", cols.join(", "))
+                    }
+                };
+                format!(
+                    "from {parent_table}.{} to {child_table}.{}",
+                    fmt(parent_columns),
+                    fmt(child_columns),
+                )
+            }
             Self::DropRelationship { selector } => match selector {
                 RelationshipSelector::Named { name } => name.clone(),
                 RelationshipSelector::Endpoints {

src/dsl/grammar/ddl.rs

@@ -785,12 +785,24 @@ fn build_add_relationship(path: &MatchedPath, _source: &str) -> Result<Command,
         .iter()
         .any(|i| matches!(&i.kind, MatchedKind::Flag("create-fk")));
 
+    // Collect every matched `parent_column` / `child_column` ident, in
+    // order — one each for the single-column `from P.col to C.col`
+    // form, or the full lists for the parenthesized compound form
+    // `from P.(a, b) to C.(x, y)` (ADR-0043).
+    let parent_columns = collect_idents(path, "parent_column");
+    let child_columns = collect_idents(path, "child_column");
+    if parent_columns.is_empty() || child_columns.is_empty() {
+        return Err(ValidationError {
+            message_key: "parse.error_wrapper",
+            args: vec![("detail", "a relationship needs both endpoints".to_string())],
+        });
+    }
     Ok(Command::AddRelationship {
         name: ident(path, "relationship_name").map(str::to_string),
         parent_table: require_ident(path, "parent_table")?,
-        parent_column: require_ident(path, "parent_column")?,
+        parent_columns,
         child_table: require_ident(path, "child_table")?,
-        child_column: require_ident(path, "child_column")?,
+        child_columns,
         on_delete: on_delete.unwrap_or_else(ReferentialAction::default_action),
         on_update: on_update.unwrap_or_else(ReferentialAction::default_action),
         create_fk,
@@ -1680,9 +1692,12 @@ where
     }
     SqlForeignKey {
         name,
-        child_column,
+        // Single-column for now; the parenthesized multi-column parse
+        // (`FOREIGN KEY (a, b) REFERENCES P(x, y)`) lands with the
+        // grammar-node change (ADR-0043).
+        child_columns: vec![child_column],
         parent_table,
-        parent_column,
+        parent_columns: parent_column.map(|c| vec![c]),
         on_delete,
         on_update,
     }
@@ -3202,9 +3217,9 @@ mod sql_alter_table_tests {
                 assert_eq!(name, None);
                 match *constraint {
                     TableConstraint::ForeignKey(fk) => {
-                        assert_eq!(fk.child_column, "pid");
+                        assert_eq!(fk.child_columns, vec!["pid".to_string()]);
                         assert_eq!(fk.parent_table, "P");
-                        assert_eq!(fk.parent_column.as_deref(), Some("id"));
+                        assert_eq!(fk.parent_columns, Some(vec!["id".to_string()]));
                     }
                     other => panic!("expected ForeignKey, got {other:?}"),
                 }
@@ -3216,7 +3231,7 @@ mod sql_alter_table_tests {
                 assert_eq!(name.as_deref(), Some("fk_p"));
                 match *constraint {
                     TableConstraint::ForeignKey(fk) => {
-                        assert_eq!(fk.parent_column, None, "bare reference resolves at execution");
+                        assert_eq!(fk.parent_columns, None, "bare reference resolves at execution");
                     }
                     other => panic!("expected ForeignKey, got {other:?}"),
                 }

src/dsl/grammar/sql_create_table.rs

@@ -984,9 +984,9 @@ mod builder_tests {
         assert_eq!(fks.len(), 1);
         let fk = &fks[0];
         assert_eq!(fk.name, None, "inline FK is auto-named at execution");
-        assert_eq!(fk.child_column, "pid");
+        assert_eq!(fk.child_columns, vec!["pid".to_string()]);
         assert_eq!(fk.parent_table, "parent");
-        assert_eq!(fk.parent_column.as_deref(), Some("id"));
+        assert_eq!(fk.parent_columns, Some(vec!["id".to_string()]));
         assert_eq!(fk.on_delete, ReferentialAction::NoAction);
         assert_eq!(fk.on_update, ReferentialAction::NoAction);
     }
@@ -994,9 +994,9 @@ mod builder_tests {
     #[test]
     fn bare_inline_reference_has_no_parent_column() {
         let fks = parse_sct_fks("create table t (id int, pid int references parent)");
-        assert_eq!(fks[0].parent_column, None, "bare REFERENCES — resolved at execution");
+        assert_eq!(fks[0].parent_columns, None, "bare REFERENCES — resolved at execution");
         assert_eq!(fks[0].parent_table, "parent");
-        assert_eq!(fks[0].child_column, "pid");
+        assert_eq!(fks[0].child_columns, vec!["pid".to_string()]);
     }
 
     #[test]
@@ -1026,9 +1026,9 @@ mod builder_tests {
             parse_sct_fks("create table t (id int, pid int, foreign key (pid) references parent(id))");
         assert_eq!(fks.len(), 1);
         assert_eq!(fks[0].name, None);
-        assert_eq!(fks[0].child_column, "pid");
+        assert_eq!(fks[0].child_columns, vec!["pid".to_string()]);
         assert_eq!(fks[0].parent_table, "parent");
-        assert_eq!(fks[0].parent_column.as_deref(), Some("id"));
+        assert_eq!(fks[0].parent_columns, Some(vec!["id".to_string()]));
     }
 
     #[test]
@@ -1038,7 +1038,7 @@ mod builder_tests {
              constraint fk_parent foreign key (pid) references parent(id))",
         );
         assert_eq!(fks[0].name.as_deref(), Some("fk_parent"));
-        assert_eq!(fks[0].child_column, "pid");
+        assert_eq!(fks[0].child_columns, vec!["pid".to_string()]);
     }
 
     #[test]
@@ -1048,8 +1048,8 @@ mod builder_tests {
              foreign key (a) references p(id), foreign key (b) references q(id))",
         );
         assert_eq!(fks.len(), 2);
-        assert_eq!((fks[0].child_column.as_str(), fks[0].parent_table.as_str()), ("a", "p"));
+        assert_eq!((fks[0].child_columns[0].as_str(), fks[0].parent_table.as_str()), ("a", "p"));
-        assert_eq!((fks[1].child_column.as_str(), fks[1].parent_table.as_str()), ("b", "q"));
+        assert_eq!((fks[1].child_columns[0].as_str(), fks[1].parent_table.as_str()), ("b", "q"));
     }
 
     #[test]
@@ -1057,8 +1057,8 @@ mod builder_tests {
         let fks =
             parse_sct_fks("create table emp (id int primary key, mgr int references emp(id))");
         assert_eq!(fks[0].parent_table, "emp", "self-reference");
-        assert_eq!(fks[0].child_column, "mgr");
+        assert_eq!(fks[0].child_columns, vec!["mgr".to_string()]);
-        assert_eq!(fks[0].parent_column.as_deref(), Some("id"));
+        assert_eq!(fks[0].parent_columns, Some(vec!["id".to_string()]));
     }
 
     #[test]
@@ -1080,7 +1080,7 @@ mod builder_tests {
             } => {
                 assert_eq!(primary_key, vec!["id".to_string()]);
                 assert_eq!(foreign_keys.len(), 1);
-                assert_eq!(foreign_keys[0].child_column, "pid");
+                assert_eq!(foreign_keys[0].child_columns, vec!["pid".to_string()]);
                 // the column-level CHECK still attaches to `pid`
                 assert_eq!(
                     columns.iter().find(|c| c.name == "pid").unwrap().check_sql.as_deref(),

src/dsl/parser.rs

@@ -937,9 +937,9 @@ mod tests {
         Command::AddRelationship {
             name: name.map(String::from),
             parent_table: parent.0.to_string(),
-            parent_column: parent.1.to_string(),
+            parent_columns: vec![parent.1.to_string()],
             child_table: child.0.to_string(),
-            child_column: child.1.to_string(),
+            child_columns: vec![child.1.to_string()],
             on_delete,
             on_update,
             create_fk,

src/dsl/walker/mod.rs

@@ -3514,9 +3514,9 @@ mod tests {
             Command::AddRelationship {
                 name: None,
                 parent_table: "Customers".to_string(),
-                parent_column: "id".to_string(),
+                parent_columns: vec!["id".to_string()],
                 child_table: "Orders".to_string(),
-                child_column: "customer_id".to_string(),
+                child_columns: vec!["customer_id".to_string()],
                 on_delete: ReferentialAction::default_action(),
                 on_update: ReferentialAction::default_action(),
                 create_fk: false,
@@ -3535,9 +3535,9 @@ mod tests {
             Command::AddRelationship {
                 name: Some("cust_orders".to_string()),
                 parent_table: "Customers".to_string(),
-                parent_column: "id".to_string(),
+                parent_columns: vec!["id".to_string()],
                 child_table: "Orders".to_string(),
-                child_column: "customer_id".to_string(),
+                child_columns: vec!["customer_id".to_string()],
                 on_delete: ReferentialAction::Cascade,
                 on_update: ReferentialAction::SetNull,
                 create_fk: true,

src/echo.rs

@@ -264,12 +264,16 @@ pub(crate) fn render_drop_index(name: &str) -> String {
 pub(crate) fn render_add_relationship(
     name: &str,
     parent_table: &str,
-    parent_column: &str,
+    parent_columns: &[String],
     child_table: &str,
-    child_column: &str,
+    child_columns: &[String],
     on_delete: ReferentialAction,
     on_update: ReferentialAction,
 ) -> String {
+    // Multi-column FK (ADR-0043): comma-join each side; a
+    // single-column FK is the one-element case.
+    let child_column = child_columns.join(", ");
+    let parent_column = parent_columns.join(", ");
     let mut s = format!(
         "ALTER TABLE {child_table} ADD CONSTRAINT {name} FOREIGN KEY ({child_column}) REFERENCES {parent_table} ({parent_column})"
     );
@@ -325,28 +329,31 @@ pub(crate) fn render_drop_column_cascade(
 pub(crate) fn render_add_relationship_create_fk(
     name: &str,
     parent_table: &str,
-    parent_column: &str,
+    parent_columns: &[String],
     child_table: &str,
-    child_column: &str,
+    child_columns: &[String],
     on_delete: ReferentialAction,
     on_update: ReferentialAction,
-    new_child_column_type: crate::dsl::types::Type,
+    // The child columns `--create-fk` newly creates, with their types
+    // (ADR-0043: one per missing column, typed to the matching parent
+    // PK column's `fk_target_type`). Columns that already existed are
+    // omitted — no `ADD COLUMN` line for them.
+    new_columns: &[(String, crate::dsl::types::Type)],
 ) -> Vec<String> {
-    vec![
+    let mut lines: Vec<String> = new_columns
-        format!(
+        .iter()
-            "ALTER TABLE {child_table} ADD COLUMN {child_column} {}",
+        .map(|(col, ty)| format!("ALTER TABLE {child_table} ADD COLUMN {col} {}", ty.keyword()))
-            new_child_column_type.keyword()
+        .collect();
-        ),
+    lines.push(render_add_relationship(
-        render_add_relationship(
+        name,
-            name,
+        parent_table,
-            parent_table,
+        parent_columns,
-            parent_column,
+        child_table,
-            child_table,
+        child_columns,
-            child_column,
+        on_delete,
-            on_delete,
+        on_update,
-            on_update,
+    ));
-        ),
+    lines
-    ]
 }
 
 /// Append the `NOT NULL` / `UNIQUE` / `DEFAULT` / `CHECK` column-constraint
@@ -953,9 +960,9 @@ mod tests {
         let sql = render_add_relationship(
             "Orders_CustId_to_Customers_id",
             "Customers",
-            "id",
+            &["id".to_string()],
             "Orders",
-            "CustId",
+            &["CustId".to_string()],
             ReferentialAction::NoAction,
             ReferentialAction::NoAction,
         );
@@ -971,9 +978,9 @@ mod tests {
         let sql = render_add_relationship(
             "places",
             "Customers",
-            "id",
+            &["id".to_string()],
             "Orders",
-            "CustId",
+            &["CustId".to_string()],
             ReferentialAction::Cascade,
             ReferentialAction::SetNull,
         );
@@ -1029,14 +1036,14 @@ mod tests {
         let lines = render_add_relationship_create_fk(
             "Customers_id_to_Orders_CustId",
             "Customers",
-            "id",
+            &["id".to_string()],
             "Orders",
-            "CustId",
+            &["CustId".to_string()],
             ReferentialAction::Cascade,
             ReferentialAction::NoAction,
             // Parent PK is `serial` → child FK column is `int`
             // (`Type::fk_target_type` strips auto-gen semantics; ADR-0011).
-            crate::dsl::types::Type::Int,
+            &[("CustId".to_string(), crate::dsl::types::Type::Int)],
         );
         assert_eq!(
             lines.as_slice(),
@@ -1055,12 +1062,12 @@ mod tests {
         let lines = render_add_relationship_create_fk(
             "Items_code_to_Lines_code",
             "Items",
-            "code",
+            &["code".to_string()],
             "Lines",
-            "code",
+            &["code".to_string()],
             ReferentialAction::NoAction,
             ReferentialAction::NoAction,
-            crate::dsl::types::Type::Text,
+            &[("code".to_string(), crate::dsl::types::Type::Text)],
         );
         assert_eq!(lines[0], "ALTER TABLE Lines ADD COLUMN code text");
         // No referential clauses when both default.

src/output_render.rs

@@ -78,6 +78,16 @@ pub fn render_data_table(data: &DataResult) -> Vec<String> {
 /// — `References:` / `Referenced by:` blocks below as plain
 /// indented text (relationship visualization is its own
 /// future ADR per §5 OOS-1).
+/// Display a relationship-endpoint column list (ADR-0043): the bare
+/// column for a single-column FK, `(a, b)` for a compound one.
+fn cols_disp(cols: &[String]) -> String {
+    if cols.len() == 1 {
+        cols[0].clone()
+    } else {
+        format!("({})", cols.join(", "))
+    }
+}
+
 #[must_use]
 pub fn render_structure(desc: &TableDescription) -> Vec<String> {
     let mut out: Vec<String> = Vec::new();
@@ -112,9 +122,9 @@ pub fn render_structure(desc: &TableDescription) -> Vec<String> {
         for r in &desc.outbound_relationships {
             out.push(format!(
                 "  {} → {}.{}  ({}, on delete {}, on update {})",
-                r.local_column,
+                cols_disp(&r.local_columns),
                 r.other_table,
-                r.other_column,
+                cols_disp(&r.other_columns),
                 r.name,
                 r.on_delete,
                 r.on_update,
@@ -127,8 +137,8 @@ pub fn render_structure(desc: &TableDescription) -> Vec<String> {
             out.push(format!(
                 "  {}.{} → {}  ({}, on delete {}, on update {})",
                 r.other_table,
-                r.other_column,
+                cols_disp(&r.other_columns),
-                r.local_column,
+                cols_disp(&r.local_columns),
                 r.name,
                 r.on_delete,
                 r.on_update,
@@ -769,8 +779,8 @@ mod tests {
             inbound_relationships: vec![RelationshipEnd {
                 name: "cust_orders".to_string(),
                 other_table: "Orders".to_string(),
-                other_column: "cust_id".to_string(),
+                other_columns: vec!["cust_id".to_string()],
-                local_column: "id".to_string(),
+                local_columns: vec!["id".to_string()],
                 on_delete: ReferentialAction::Cascade,
                 on_update: ReferentialAction::NoAction,
             }],

src/persistence/mod.rs

@@ -245,9 +245,13 @@ pub struct IndexSchema {
 pub struct RelationshipSchema {
     pub name: String,
     pub parent_table: String,
-    pub parent_column: String,
+    /// Parent PK column(s); one element for single-column, ordered
+    /// list for a compound-PK FK (ADR-0043). Paired positionally
+    /// with `child_columns`.
+    pub parent_columns: Vec<String>,
     pub child_table: String,
-    pub child_column: String,
+    /// Child column(s), positionally paired with `parent_columns`.
+    pub child_columns: Vec<String>,
     pub on_delete: ReferentialAction,
     pub on_update: ReferentialAction,
 }

src/persistence/yaml.rs

@@ -188,20 +188,31 @@ fn write_relationship(out: &mut String, rel: &RelationshipSchema) {
     let _ = writeln!(out, "  - name: {}", quote_if_needed(&rel.name));
     let _ = writeln!(
         out,
-        "    parent: {{ table: {}, column: {} }}",
+        "    parent: {{ table: {}, columns: [{}] }}",
         quote_if_needed(&rel.parent_table),
-        quote_if_needed(&rel.parent_column),
+        write_col_list(&rel.parent_columns),
     );
     let _ = writeln!(
         out,
-        "    child:  {{ table: {}, column: {} }}",
+        "    child:  {{ table: {}, columns: [{}] }}",
         quote_if_needed(&rel.child_table),
-        quote_if_needed(&rel.child_column),
+        write_col_list(&rel.child_columns),
     );
     let _ = writeln!(out, "    on_delete: {}", action_keyword(rel.on_delete));
     let _ = writeln!(out, "    on_update: {}", action_keyword(rel.on_update));
 }
 
+/// Format a column list for an inline yaml flow sequence — `a, b`
+/// (the caller wraps in `[…]`), each element quoted if needed.
+/// Matches the `primary_key: [...]` / index `columns: [...]` house
+/// style (ADR-0043 D5). One element for a single-column endpoint.
+fn write_col_list(cols: &[String]) -> String {
+    cols.iter()
+        .map(|c| quote_if_needed(c))
+        .collect::<Vec<_>>()
+        .join(", ")
+}
+
 const fn action_keyword(action: ReferentialAction) -> &'static str {
     match action {
         ReferentialAction::NoAction => "no_action",
@@ -309,9 +320,9 @@ pub(crate) fn parse_schema(body: &str) -> Result<SchemaSnapshot, YamlError> {
         relationships.push(RelationshipSchema {
             name: r.name,
             parent_table: r.parent.table,
-            parent_column: r.parent.column,
+            parent_columns: r.parent.columns,
             child_table: r.child.table,
-            child_column: r.child.column,
+            child_columns: r.child.columns,
             on_delete,
             on_update,
         });
@@ -502,7 +513,10 @@ struct RawRelationship {
 #[derive(Deserialize)]
 struct RawEndpoint {
     table: String,
-    column: String,
+    /// FK endpoint column list (ADR-0043): `columns: [a, b]`, one
+    /// element for a single-column endpoint — matching the
+    /// `primary_key` / index `columns` house style.
+    columns: Vec<String>,
 }
 
 #[derive(Deserialize)]
@@ -551,9 +565,9 @@ mod tests {
             relationships: vec![RelationshipSchema {
                 name: "Customers_id_to_Orders_CustId".to_string(),
                 parent_table: "Customers".to_string(),
-                parent_column: "id".to_string(),
+                parent_columns: vec!["id".to_string()],
                 child_table: "Orders".to_string(),
-                child_column: "CustId".to_string(),
+                child_columns: vec!["CustId".to_string()],
                 on_delete: ReferentialAction::Cascade,
                 on_update: ReferentialAction::NoAction,
             }],
@@ -578,8 +592,8 @@ mod tests {
         assert!(body.contains("{ name: id, type: serial }"));
         assert!(body.contains("{ name: Name, type: text }"));
         assert!(body.contains("- name: Customers_id_to_Orders_CustId"));
-        assert!(body.contains("parent: { table: Customers, column: id }"));
+        assert!(body.contains("parent: { table: Customers, columns: [id] }"));
-        assert!(body.contains("child:  { table: Orders, column: CustId }"));
+        assert!(body.contains("child:  { table: Orders, columns: [CustId] }"));
         assert!(body.contains("on_delete: cascade"));
         assert!(body.contains("on_update: no_action"));
         assert!(body.contains("- name: Orders_CustId_idx"));
@@ -934,8 +948,8 @@ project:
 tables: []
 relationships:
   - name: R
-    parent: { table: A, column: id }
+    parent: { table: A, columns: [id] }
-    child:  { table: B, column: aid }
+    child:  { table: B, columns: [aid] }
     on_delete: blow_up
     on_update: no_action
 ";

src/runtime.rs

@@ -1591,16 +1591,16 @@ struct EchoLookups {
     /// teaching playground).
     drop_relationship: Option<(String, String)>,
     /// For `Command::AddRelationship { create_fk: true, .. }` — the
-    /// type of the child column the `--create-fk` flag will create, *if*
+    /// child columns the `--create-fk` flag will newly create, each with
-    /// the column did not already exist (`Some(ty)` → newly created →
+    /// its type (ADR-0043: one per child column that did **not** already
-    /// multi-line echo; `None` → already existed → single-line echo).
+    /// exist, typed to the matching parent PK column's `fk_target_type` —
-    /// The type is derived from the parent's PK column type via
+    /// ADR-0011: `serial → int`, `shortid → text`, others identity). An
-    /// `Type::fk_target_type` (ADR-0011: `serial → int`, `shortid →
+    /// **empty** vec means every child column already existed →
-    /// text`, others identity). The outer `Option` is `None` for
+    /// single-line echo; a non-empty vec → multi-line (one `ADD COLUMN`
-    /// not-applicable commands (not a `--create-fk` add, or simple mode,
+    /// per element). The outer `Option` is `None` for not-applicable
-    /// or a pre-execution lookup failed); the inner option encodes the
+    /// commands (not a `--create-fk` add, simple mode, or a
-    /// existed-vs-created distinction.
+    /// pre-execution lookup failed).
-    add_rel_create_fk_new_column_type: Option<Option<crate::dsl::types::Type>>,
+    add_rel_create_fk_new_columns: Option<Vec<(String, crate::dsl::types::Type)>>,
 }
 
 /// Resolve drop-target names and `--create-fk` pre-state **before**
@@ -1638,9 +1638,13 @@ async fn collect_echo_lookups(
         } => {
             if let Ok(desc) = database.describe_table(child_table.clone(), None).await
                 && let Some(rel) = desc.outbound_relationships.iter().find(|r| {
+                    // The Endpoints drop selector is single-column
+                    // (ADR-0043 keeps DROP by-endpoints single-column;
+                    // compound relationships drop by name) — match a
+                    // one-column relationship by its sole columns.
                     r.other_table == *parent_table
-                        && r.other_column == *parent_column
+                        && r.other_columns.as_slice() == std::slice::from_ref(parent_column)
-                        && r.local_column == *child_column
+                        && r.local_columns.as_slice() == std::slice::from_ref(child_column)
                 })
             {
                 out.drop_relationship = Some((rel.name.clone(), child_table.clone()));
@@ -1668,41 +1672,37 @@ async fn collect_echo_lookups(
         Command::AddRelationship {
             create_fk: true,
             parent_table,
-            parent_column,
+            parent_columns,
             child_table,
-            child_column,
+            child_columns,
             ..
         } => {
-            // Two pre-state facts feed the multi-line `--create-fk` echo
+            // Pre-state for the multi-line `--create-fk` echo (ADR-0038
-            // (ADR-0038 §7 Bucket B, category 2): whether the child
+            // §7 Bucket B, category 2 / ADR-0043): the subset of child
-            // column already exists (determines single- vs multi-line)
+            // columns that do NOT already exist, each typed to the
-            // and the parent PK column's user type (determines the
+            // matching parent PK column's `fk_target_type`. Needed
-            // newly-created child column's type via
+            // *before* execution to know which `ADD COLUMN` lines to
-            // `Type::fk_target_type`). Both are looked up post-exec from
+            // emit. The parent columns here are the explicit DSL list,
-            // the description for `add relationship` (no `--create-fk`),
+            // paired positionally with the child list.
-            // but the `--create-fk` multi-line case needs them *before*
+            let parent_desc = database.describe_table(parent_table.clone(), None).await;
-            // execution to know whether to emit an `ADD COLUMN` line.
+            let child_desc = database.describe_table(child_table.clone(), None).await;
-            let parent_pk_type = database
+            if let (Ok(parent_desc), Ok(child_desc)) = (parent_desc, child_desc) {
-                .describe_table(parent_table.clone(), None)
+                let mut new_columns: Vec<(String, crate::dsl::types::Type)> = Vec::new();
-                .await
+                for (child_col, parent_col) in child_columns.iter().zip(parent_columns) {
-                .ok()
+                    let already = child_desc.columns.iter().any(|c| c.name == *child_col);
-                .and_then(|d| {
+                    if already {
-                    d.columns
+                        continue;
+                    }
+                    if let Some(parent_ty) = parent_desc
+                        .columns
                         .iter()
-                        .find(|c| c.name == *parent_column)
+                        .find(|c| c.name == *parent_col)
                         .and_then(|c| c.user_type)
-                });
+                    {
-            let child_column_existed = database
+                        new_columns.push((child_col.clone(), parent_ty.fk_target_type()));
-                .describe_table(child_table.clone(), None)
+                    }
-                .await
+                }
-                .ok()
+                out.add_rel_create_fk_new_columns = Some(new_columns);
-                .map(|d| d.columns.iter().any(|c| c.name == *child_column));
-            if let (Some(parent_ty), Some(existed)) = (parent_pk_type, child_column_existed) {
-                out.add_rel_create_fk_new_column_type = Some(if existed {
-                    None
-                } else {
-                    Some(parent_ty.fk_target_type())
-                });
             }
         }
         _ => {}
@@ -1755,9 +1755,9 @@ fn build_schema_echo(
         Command::AddRelationship {
             name,
             parent_table,
-            parent_column,
+            parent_columns,
             child_table,
-            child_column,
+            child_columns,
             on_delete,
             on_update,
             create_fk,
@@ -1766,57 +1766,55 @@ fn build_schema_echo(
             // relationships (target_table for AddRelationship is the
             // parent — `database.add_relationship` returns the parent's
             // description per ADR-0013), falling back to the command's
-            // explicit `name` when the description is unavailable.
+            // explicit `name` when the description is unavailable. Match
+            // on the column lists (ADR-0043), the child as `other` and
+            // the parent as `local` from the parent's perspective.
             let resolved = description
                 .and_then(|d| {
                     d.inbound_relationships.iter().find(|r| {
                         r.other_table == *child_table
-                            && r.other_column == *child_column
+                            && r.other_columns == *child_columns
-                            && r.local_column == *parent_column
+                            && r.local_columns == *parent_columns
                     })
                 })
                 .map(|r| r.name.clone())
                 .or_else(|| name.clone())?;
             if *create_fk {
-                // Multi-line iff the child column was newly created
+                // The pre-execution lookup captured which child columns
-                // (`--create-fk`'s pre-state, captured pre-execution
+                // `--create-fk` newly creates (ADR-0043). An empty list
-                // into `add_rel_create_fk_new_column_type`). When the
+                // → every column existed → single-line FK echo; a
-                // column already existed the echo collapses to the
+                // non-empty list → one `ADD COLUMN` per new column then
-                // single-line FK form — the SQL `ADD COLUMN` would be
+                // the FK line.
-                // a no-op-with-error otherwise, and the catalogue is
+                let new_columns = lookups.add_rel_create_fk_new_columns.as_ref()?;
-                // explicit: "one line if the column already existed".
+                if new_columns.is_empty() {
-                Some(lookups.add_rel_create_fk_new_column_type?.map_or_else(
+                    Some(vec![crate::echo::render_add_relationship(
-                    || {
+                        &resolved,
-                        vec![crate::echo::render_add_relationship(
+                        parent_table,
-                            &resolved,
+                        parent_columns,
-                            parent_table,
+                        child_table,
-                            parent_column,
+                        child_columns,
-                            child_table,
+                        *on_delete,
-                            child_column,
+                        *on_update,
-                            *on_delete,
+                    )])
-                            *on_update,
+                } else {
-                        )]
+                    Some(crate::echo::render_add_relationship_create_fk(
-                    },
+                        &resolved,
-                    |new_ty| {
+                        parent_table,
-                        crate::echo::render_add_relationship_create_fk(
+                        parent_columns,
-                            &resolved,
+                        child_table,
-                            parent_table,
+                        child_columns,
-                            parent_column,
+                        *on_delete,
-                            child_table,
+                        *on_update,
-                            child_column,
+                        new_columns,
-                            *on_delete,
+                    ))
-                            *on_update,
+                }
-                            new_ty,
-                        )
-                    },
-                ))
             } else {
                 Some(vec![crate::echo::render_add_relationship(
                     &resolved,
                     parent_table,
-                    parent_column,
+                    parent_columns,
                     child_table,
-                    child_column,
+                    child_columns,
                     *on_delete,
                     *on_update,
                 )])
@@ -2013,17 +2011,19 @@ async fn enrich_fk_violation(
             };
             facts.table = Some(table.clone());
             for rel in outbound {
-                let value = user_value_for_column_with_schema(
+                // The friendly FK-error facts model is single-column
-                    database,
+                // (ADR-0019); for a compound FK (ADR-0043) we enrich
-                    command,
+                // from the first column pair — the error still surfaces,
-                    table,
+                // richer multi-column enrichment is a later refinement.
-                    &rel.local_column,
+                let Some(local_col) = rel.local_columns.first().cloned() else {
-                )
+                    continue;
-                .await;
+                };
+                let value =
+                    user_value_for_column_with_schema(database, command, table, &local_col).await;
                 if let Some(v) = value {
-                    facts.column = Some(rel.local_column);
+                    facts.column = Some(local_col);
                     facts.parent_table = Some(rel.other_table);
-                    facts.parent_column = Some(rel.other_column);
+                    facts.parent_column = rel.other_columns.into_iter().next();
                     facts.value = Some(v.to_string());
                     break;
                 }
@@ -2615,9 +2615,9 @@ async fn execute_command_typed(
         Command::AddRelationship {
             name,
             parent_table,
-            parent_column,
+            parent_columns,
             child_table,
-            child_column,
+            child_columns,
             on_delete,
             on_update,
             create_fk,
@@ -2625,9 +2625,9 @@ async fn execute_command_typed(
             .add_relationship(
                 name,
                 parent_table,
-                parent_column,
+                parent_columns,
                 child_table,
-                child_column,
+                child_columns,
                 on_delete,
                 on_update,
                 create_fk,
@@ -3193,9 +3193,9 @@ mod tests {
             .add_relationship(
                 None,
                 "Customers".to_string(),
-                "id".to_string(),
+                vec!["id".to_string()],
                 "Orders".to_string(),
-                "CustId".to_string(),
+                vec!["CustId".to_string()],
                 ReferentialAction::Cascade,
                 ReferentialAction::NoAction,
                 false,
@@ -3206,9 +3206,9 @@ mod tests {
         let add_rel_cmd = Command::AddRelationship {
             name: None,
             parent_table: "Customers".to_string(),
-            parent_column: "id".to_string(),
+            parent_columns: vec!["id".to_string()],
             child_table: "Orders".to_string(),
-            child_column: "CustId".to_string(),
+            child_columns: vec!["CustId".to_string()],
             on_delete: ReferentialAction::Cascade,
             on_update: ReferentialAction::NoAction,
             create_fk: false,
@@ -3366,9 +3366,9 @@ mod tests {
         let add_fk_cmd = Command::AddRelationship {
             name: None,
             parent_table: "Customers".to_string(),
-            parent_column: "id".to_string(),
+            parent_columns: vec!["id".to_string()],
             child_table: "Orders".to_string(),
-            child_column: "CustId".to_string(),
+            child_columns: vec!["CustId".to_string()],
             on_delete: ReferentialAction::Cascade,
             on_update: ReferentialAction::NoAction,
             create_fk: true,
@@ -3378,17 +3378,17 @@ mod tests {
         let pre_lookups =
             super::collect_echo_lookups(&db, &add_fk_cmd, EffectiveMode::AdvancedPersistent).await;
         assert_eq!(
-            pre_lookups.add_rel_create_fk_new_column_type,
+            pre_lookups.add_rel_create_fk_new_columns,
-            Some(Some(Type::Int)),
+            Some(vec![("CustId".to_string(), Type::Int)]),
             "pre-exec captures `serial → int` for the newly-created child column",
         );
         let parent_desc = db
             .add_relationship(
                 None,
                 "Customers".to_string(),
-                "id".to_string(),
+                vec!["id".to_string()],
                 "Orders".to_string(),
-                "CustId".to_string(),
+                vec!["CustId".to_string()],
                 ReferentialAction::Cascade,
                 ReferentialAction::NoAction,
                 true,
@@ -3435,17 +3435,17 @@ mod tests {
         let pre_lookups =
             super::collect_echo_lookups(&db, &add_fk_cmd, EffectiveMode::AdvancedPersistent).await;
         assert_eq!(
-            pre_lookups.add_rel_create_fk_new_column_type,
+            pre_lookups.add_rel_create_fk_new_columns,
-            Some(None),
+            Some(vec![]),
             "pre-exec records the child column already existed → single-line echo",
         );
         let parent_desc = db
             .add_relationship(
                 None,
                 "Customers".to_string(),
-                "id".to_string(),
+                vec!["id".to_string()],
                 "Orders".to_string(),
-                "CustId".to_string(),
+                vec!["CustId".to_string()],
                 ReferentialAction::Cascade,
                 ReferentialAction::NoAction,
                 true,

tests/it/column_op_guards.rs

@@ -132,9 +132,9 @@ fn add_relationship_refuses_internal_tables() {
         .block_on(db.add_relationship(
             None,
             internal.clone(),
-            "name".to_string(),
+            vec!["name".to_string()],
             "C".to_string(),
-            "x".to_string(),
+            vec!["x".to_string()],
             ReferentialAction::NoAction,
             ReferentialAction::NoAction,
             false,
@@ -161,9 +161,9 @@ fn add_relationship_refuses_internal_tables() {
         .block_on(db.add_relationship(
             None,
             "P".to_string(),
-            "id".to_string(),
+            vec!["id".to_string()],
             internal,
-            "x".to_string(),
+            vec!["x".to_string()],
             ReferentialAction::NoAction,
             ReferentialAction::NoAction,
             false,

tests/it/friendly_enrichment.rs

@@ -420,9 +420,9 @@ fn enrich_fk_insert_resolves_parent_table_column_and_value() {
         db.add_relationship(
             None,
             "Customers".to_string(),
-            "id".to_string(),
+            vec!["id".to_string()],
             "Orders".to_string(),
-            "CustId".to_string(),
+            vec!["CustId".to_string()],
             ReferentialAction::NoAction,
             ReferentialAction::NoAction,
             false,
@@ -496,9 +496,9 @@ fn enrich_fk_insert_natural_order_multi_value_resolves_via_schema() {
         db.add_relationship(
             None,
             "Customers".to_string(),
-            "id".to_string(),
+            vec!["id".to_string()],
             "Orders".to_string(),
-            "CustId".to_string(),
+            vec!["CustId".to_string()],
             ReferentialAction::NoAction,
             ReferentialAction::NoAction,
             false,
@@ -570,9 +570,9 @@ fn enrich_fk_delete_resolves_child_table() {
         db.add_relationship(
             None,
             "Customers".to_string(),
-            "id".to_string(),
+            vec!["id".to_string()],
             "Orders".to_string(),
-            "CustId".to_string(),
+            vec!["CustId".to_string()],
             ReferentialAction::NoAction,
             ReferentialAction::NoAction,
             false,

tests/it/iteration2_persistence.rs

@@ -192,9 +192,9 @@ fn delete_with_cascade_rewrites_both_csvs() {
         db.add_relationship(
             None,
             "Customers".to_string(),
-            "id".to_string(),
+            vec!["id".to_string()],
             "Orders".to_string(),
-            "CustId".to_string(),
+            vec!["CustId".to_string()],
             ReferentialAction::Cascade,
             ReferentialAction::NoAction,
             false,
@@ -424,9 +424,9 @@ fn project_yaml_carries_relationship_after_add() {
         db.add_relationship(
             None,
             "Customers".to_string(),
-            "id".to_string(),
+            vec!["id".to_string()],
             "Orders".to_string(),
-            "CustId".to_string(),
+            vec!["CustId".to_string()],
             ReferentialAction::Cascade,
             ReferentialAction::NoAction,
             false,

tests/it/iteration3_rebuild.rs

@@ -185,9 +185,9 @@ fn rebuild_restores_relationships_and_cascade_behaviour() {
             db.add_relationship(
                 None,
                 "Customers".to_string(),
-                "id".to_string(),
+                vec!["id".to_string()],
                 "Orders".to_string(),
-                "CustId".to_string(),
+                vec!["CustId".to_string()],
                 ReferentialAction::Cascade,
                 ReferentialAction::NoAction,
                 false,

tests/it/show_list.rs

@@ -144,9 +144,9 @@ async fn seed_schema(db: &Database) {
     db.add_relationship(
         Some("orders_customer".to_string()),
         "Customers".to_string(),
-        "id".to_string(),
+        vec!["id".to_string()],
         "Orders".to_string(),
-        "customer_id".to_string(),
+        vec!["customer_id".to_string()],
         ReferentialAction::Cascade,
         ReferentialAction::NoAction,
         false,

tests/it/sql_create_table.rs

@@ -834,9 +834,9 @@ fn dropping_a_column_a_table_check_references_fails_cleanly() {
 fn fk(child_column: &str, parent_table: &str, parent_column: Option<&str>) -> SqlForeignKey {
     SqlForeignKey {
         name: None,
-        child_column: child_column.to_string(),
+        child_columns: vec![child_column.to_string()],
         parent_table: parent_table.to_string(),
-        parent_column: parent_column.map(str::to_string),
+        parent_columns: parent_column.map(|c| vec![c.to_string()]),
         on_delete: ReferentialAction::NoAction,
         on_update: ReferentialAction::NoAction,
     }
@@ -929,7 +929,7 @@ fn foreign_key_creates_named_relationship_visible_in_describe() {
     let rel = &child.outbound_relationships[0];
     assert_eq!(rel.name, "parent_id_to_child_pid", "auto-named per ADR-0013");
     assert_eq!(rel.other_table, "parent");
-    assert_eq!(rel.local_column, "pid");
+    assert_eq!(rel.local_columns, vec!["pid".to_string()]);
 
     let parent = r.block_on(db.describe_table("parent".to_string(), None)).expect("describe parent");
     assert_eq!(parent.inbound_relationships.len(), 1, "parent is referenced by child");
@@ -974,7 +974,7 @@ fn bare_references_resolves_to_parent_single_column_pk() {
     ))
     .expect("create child with bare REFERENCES");
     let child = r.block_on(db.describe_table("child".to_string(), None)).expect("describe");
-    assert_eq!(child.outbound_relationships[0].other_column, "id", "resolved to parent PK");
+    assert_eq!(child.outbound_relationships[0].other_columns, vec!["id".to_string()], "resolved to parent PK");
 }
 
 #[test]
@@ -1341,7 +1341,7 @@ fn bare_self_reference_resolves_to_own_pk() {
     ))
     .expect("create self-referential emp with a bare reference");
     let emp = r.block_on(db.describe_table("emp".to_string(), None)).expect("describe");
-    assert_eq!(emp.outbound_relationships[0].other_column, "id", "bare self-ref resolved to own PK");
+    assert_eq!(emp.outbound_relationships[0].other_columns, vec!["id".to_string()], "bare self-ref resolved to own PK");
     // Enforced: a non-existent manager is rejected.
     r.block_on(db.insert(
         "emp".to_string(),

tests/it/sql_delete.rs

@@ -100,9 +100,9 @@ fn cascade_fixture(db: &Database, rt: &tokio::runtime::Runtime) {
     rt.block_on(db.add_relationship(
         Some("places".to_string()),
         "Customers".to_string(),
-        "id".to_string(),
+        vec!["id".to_string()],
         "Orders".to_string(),
-        "CustId".to_string(),
+        vec!["CustId".to_string()],
         ReferentialAction::Cascade,
         ReferentialAction::NoAction,
         false,
@@ -289,9 +289,9 @@ fn cascade_to_two_children_reports_both() {
         rt.block_on(db.add_relationship(
             Some(name.to_string()),
             "Customers".to_string(),
-            "id".to_string(),
+            vec!["id".to_string()],
             child.to_string(),
-            "CustId".to_string(),
+            vec!["CustId".to_string()],
             ReferentialAction::Cascade,
             ReferentialAction::NoAction,
             false,
@@ -358,9 +358,9 @@ fn delete_violating_fk_fails_and_persists_nothing() {
     rt.block_on(db.add_relationship(
         Some("places".to_string()),
         "Customers".to_string(),
-        "id".to_string(),
+        vec!["id".to_string()],
         "Orders".to_string(),
-        "CustId".to_string(),
+        vec!["CustId".to_string()],
         ReferentialAction::NoAction, // on delete: reject if referenced
         ReferentialAction::NoAction,
         false,
@@ -395,9 +395,9 @@ fn self_referential_cascade_counts_only_cascaded_rows() {
     rt.block_on(db.add_relationship(
         Some("parent_of".to_string()),
         "T".to_string(),
-        "id".to_string(),
+        vec!["id".to_string()],
         "T".to_string(),
-        "ParentId".to_string(),
+        vec!["ParentId".to_string()],
         ReferentialAction::Cascade,
         ReferentialAction::NoAction,
         false,

tests/it/sql_dml_e2e.rs

@@ -318,9 +318,9 @@ fn cascade_fixture(db: &Database, rt: &tokio::runtime::Runtime) {
     rt.block_on(db.add_relationship(
         Some("places".to_string()),
         "Customers".to_string(),
-        "id".to_string(),
+        vec!["id".to_string()],
         "Orders".to_string(),
-        "CustId".to_string(),
+        vec!["CustId".to_string()],
         ReferentialAction::Cascade,
         ReferentialAction::NoAction,
         false,

tests/it/sql_drop_table.rs

@@ -104,9 +104,9 @@ fn dropping_a_referenced_parent_is_refused() {
         vec![],
         vec![SqlForeignKey {
             name: None,
-            child_column: "pid".to_string(),
+            child_columns: vec!["pid".to_string()],
             parent_table: "parent".to_string(),
-            parent_column: Some("id".to_string()),
+            parent_columns: Some(vec!["id".to_string()]),
             on_delete: rdbms_playground::dsl::ReferentialAction::NoAction,
             on_update: rdbms_playground::dsl::ReferentialAction::NoAction,
         }],

tests/it/walking_skeleton.rs

@@ -420,9 +420,9 @@ fn add_relationship_flow_shows_parent_side_with_inbound_section() {
         &Command::AddRelationship {
             name: None,
             parent_table: "Customers".to_string(),
-            parent_column: "Id".to_string(),
+            parent_columns: vec!["Id".to_string()],
             child_table: "Orders".to_string(),
-            child_column: "CustId".to_string(),
+            child_columns: vec!["CustId".to_string()],
             on_delete: ReferentialAction::Cascade,
             on_update: ReferentialAction::NoAction,
             create_fk: false,
@@ -449,8 +449,8 @@ fn add_relationship_flow_shows_parent_side_with_inbound_section() {
         inbound_relationships: vec![RelationshipEnd {
             name: "Customers_Id_to_Orders_CustId".to_string(),
             other_table: "Orders".to_string(),
-            other_column: "CustId".to_string(),
+            other_columns: vec!["CustId".to_string()],
-            local_column: "Id".to_string(),
+            local_columns: vec!["Id".to_string()],
             on_delete: ReferentialAction::Cascade,
             on_update: ReferentialAction::NoAction,
         }],
@@ -462,9 +462,9 @@ fn add_relationship_flow_shows_parent_side_with_inbound_section() {
         command: Command::AddRelationship {
             name: None,
             parent_table: "Customers".to_string(),
-            parent_column: "Id".to_string(),
+            parent_columns: vec!["Id".to_string()],
             child_table: "Orders".to_string(),
-            child_column: "CustId".to_string(),
+            child_columns: vec!["CustId".to_string()],
             on_delete: ReferentialAction::Cascade,
             on_update: ReferentialAction::NoAction,
             create_fk: false,
@@ -504,8 +504,8 @@ fn add_relationship_flow_shows_inbound_section_on_parent() {
         inbound_relationships: vec![RelationshipEnd {
             name: "Customers_Id_to_Orders_CustId".to_string(),
             other_table: "Orders".to_string(),
-            other_column: "CustId".to_string(),
+            other_columns: vec!["CustId".to_string()],
-            local_column: "Id".to_string(),
+            local_columns: vec!["Id".to_string()],
             on_delete: ReferentialAction::Cascade,
             on_update: ReferentialAction::NoAction,
         }],