feat: replicate stored generated columns on Postgres 18

crates/etl-destinations/src/clickhouse/client.rs

@@ -762,6 +762,7 @@ mod tests {
             ordinal_position: 1,
             primary_key_ordinal_position: Some(1),
             nullable: false,
+            generated: false,
             default_expression: None,
         }
     }

crates/etl-destinations/src/clickhouse/core.rs

@@ -168,13 +168,18 @@ fn ensure_clickhouse_renames_are_supported(table_name: &str, plan: &SchemaPlan)
 /// source default. ClickHouse cannot represent a top-level nullable array, so a
 /// replication-mask expansion containing an array fails instead of silently
 /// exposing empty arrays for historical rows.
+///
+/// A generated column follows the same rule regardless of why it appeared. It
+/// carries source values that predate its arrival in the destination, so
+/// treating it as a plain table-schema addition would backfill historical rows
+/// with an empty array rather than the values the source already holds.
 fn clickhouse_add_column_definition(
     table_name: &str,
     after_column_schema: &ColumnSchema,
     reason: ColumnPresenceChangeReason,
 ) -> EtlResult<(ColumnSchema, bool)> {
     let mut destination_column_schema = after_column_schema.clone();
-    if reason == ColumnPresenceChangeReason::ReplicationMask {
+    if reason == ColumnPresenceChangeReason::ReplicationMask || after_column_schema.generated {
         if is_array_type(&after_column_schema.typ) {
             return Err(etl_error!(
                 ErrorKind::SourceSchemaError,

crates/etl-destinations/src/clickhouse/schema.rs

@@ -375,6 +375,7 @@ mod tests {
             ordinal_position: 1,
             primary_key_ordinal_position: Some(1),
             nullable: false,
+            generated: false,
             default_expression: None,
         }];
         // Pre-encoded table name with embedded quotes to verify the SQL
@@ -446,6 +447,7 @@ mod tests {
                 ordinal_position: 1,
                 primary_key_ordinal_position: Some(1),
                 nullable: false,
+                generated: false,
                 default_expression: None,
             },
             ColumnSchema {
@@ -455,6 +457,7 @@ mod tests {
                 ordinal_position: 2,
                 primary_key_ordinal_position: None,
                 nullable: true,
+                generated: false,
                 default_expression: None,
             },
         ];
@@ -543,6 +546,7 @@ mod tests {
             ordinal_position: 1,
             primary_key_ordinal_position: Some(1),
             nullable: false,
+            generated: false,
             default_expression: None,
         }];
         let sql = create_merge_tree_sql("public_t", &schemas);
@@ -561,6 +565,7 @@ mod tests {
             ordinal_position: 1,
             primary_key_ordinal_position: None,
             nullable: false,
+            generated: false,
             default_expression: None,
         }];
         let sql = create_merge_tree_sql("public_t", &schemas);
@@ -581,6 +586,7 @@ mod tests {
                 ordinal_position: 1,
                 primary_key_ordinal_position: Some(1),
                 nullable: false,
+                generated: false,
                 default_expression: None,
             },
             ColumnSchema {
@@ -590,6 +596,7 @@ mod tests {
                 ordinal_position: 2,
                 primary_key_ordinal_position: None,
                 nullable: true,
+                generated: false,
                 default_expression: None,
             },
         ];
@@ -615,6 +622,7 @@ mod tests {
                 ordinal_position: 1,
                 primary_key_ordinal_position: Some(2),
                 nullable: false,
+                generated: false,
                 default_expression: None,
             },
             ColumnSchema {
@@ -624,6 +632,7 @@ mod tests {
                 ordinal_position: 2,
                 primary_key_ordinal_position: None,
                 nullable: true,
+                generated: false,
                 default_expression: None,
             },
             ColumnSchema {
@@ -633,6 +642,7 @@ mod tests {
                 ordinal_position: 3,
                 primary_key_ordinal_position: Some(1),
                 nullable: false,
+                generated: false,
                 default_expression: None,
             },
         ];
@@ -655,6 +665,7 @@ mod tests {
             ordinal_position: 1,
             primary_key_ordinal_position: None,
             nullable: true,
+            generated: false,
             default_expression: None,
         }];
         // --- WHEN: build the ReplacingMergeTree DDL ---
@@ -673,6 +684,7 @@ mod tests {
             ordinal_position: 1,
             primary_key_ordinal_position: Some(1),
             nullable: false,
+            generated: false,
             default_expression: None,
         }];
         // --- WHEN/THEN: dispatcher selects the matching engine branch ---
@@ -695,6 +707,7 @@ mod tests {
                 ordinal_position: 1,
                 primary_key_ordinal_position: Some(1),
                 nullable: false,
+                generated: false,
                 default_expression: None,
             },
             ColumnSchema {
@@ -704,6 +717,7 @@ mod tests {
                 ordinal_position: 2,
                 primary_key_ordinal_position: None,
                 nullable: true,
+                generated: false,
                 default_expression: None,
             },
         ];

crates/etl-postgres/src/schema.rs

@@ -308,7 +308,7 @@ pub fn numeric_modifiers(modifier: TypeModifier) -> Option<NumericModifiers> {
 ///
 /// This type contains all metadata about a column including its name, data
 /// type, type modifier, ordinal position, primary key information, nullability,
-/// and default expression.
+/// generation kind, and default expression.
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct ColumnSchema {
     /// The name of the column.
@@ -324,7 +324,18 @@ pub struct ColumnSchema {
     pub primary_key_ordinal_position: Option<i32>,
     /// Whether the column can contain NULL values.
     pub nullable: bool,
+    /// Whether the column is a Postgres stored generated column.
+    ///
+    /// A stored generated column only reaches the logical stream on Postgres 18
+    /// with `publish_generated_columns = stored`; otherwise it stays in the
+    /// table schema but out of the replication mask. Virtual generated columns
+    /// are never represented because they are absent from the WAL.
+    pub generated: bool,
     /// The source default expression for this column, if one is defined.
+    ///
+    /// Always [`None`] for a generated column: Postgres stores the generation
+    /// expression in `pg_attrdef`, and that expression is not a destination
+    /// default.
     pub default_expression: Option<String>,
 }
 
@@ -344,6 +355,7 @@ impl ColumnSchema {
             ordinal_position,
             primary_key_ordinal_position: None,
             nullable,
+            generated: false,
             default_expression: None,
         }
     }
@@ -365,6 +377,12 @@ impl ColumnSchema {
         self
     }
 
+    /// Sets whether this column is a stored generated column.
+    pub fn with_generated(mut self, generated: bool) -> Self {
+        self.generated = generated;
+        self
+    }
+
     /// Sets the source default expression for this column.
     pub fn with_default_expression(mut self, default_expression: String) -> Self {
         self.default_expression = Some(default_expression);
@@ -392,6 +410,7 @@ pub struct ColumnSchemaBuilder {
     ordinal_position: i32,
     primary_key_ordinal_position: Option<i32>,
     nullable: bool,
+    generated: bool,
     default_expression: Option<String>,
 }
 
@@ -405,6 +424,7 @@ impl ColumnSchemaBuilder {
             ordinal_position,
             primary_key_ordinal_position: None,
             nullable: true,
+            generated: false,
             default_expression: None,
         }
     }
@@ -439,6 +459,12 @@ impl ColumnSchemaBuilder {
         self
     }
 
+    /// Marks the column as a stored generated column.
+    pub fn generated(mut self, generated: bool) -> Self {
+        self.generated = generated;
+        self
+    }
+
     /// Sets the source default expression for this column.
     pub fn default_expression(mut self, default_expression: String) -> Self {
         self.default_expression = Some(default_expression);
@@ -460,6 +486,7 @@ impl ColumnSchemaBuilder {
             ordinal_position: self.ordinal_position,
             primary_key_ordinal_position: self.primary_key_ordinal_position,
             nullable: self.nullable,
+            generated: self.generated,
             default_expression: self.default_expression,
         }
     }
@@ -614,9 +641,30 @@ mod tests {
         assert_eq!(schema.ordinal_position, 3);
         assert_eq!(schema.primary_key_ordinal_position, Some(1));
         assert!(!schema.nullable);
+        assert!(!schema.generated);
         assert_eq!(schema.default_expression.as_deref(), Some("'new'::text"));
     }
 
+    #[test]
+    fn column_schema_defaults_to_not_generated() {
+        let schema = ColumnSchema::new("height_cm".to_owned(), Type::NUMERIC, -1, 2, true);
+
+        assert!(!schema.generated);
+        assert!(!ColumnSchema::builder("height_cm".to_owned(), Type::NUMERIC, 2).build().generated);
+    }
+
+    #[test]
+    fn column_schema_marks_generated_columns() {
+        let schema =
+            ColumnSchema::new("height_in".to_owned(), Type::NUMERIC, -1, 3, true).with_generated(true);
+        let built = ColumnSchema::builder("height_in".to_owned(), Type::NUMERIC, 3)
+            .generated(true)
+            .build();
+
+        assert!(schema.generated);
+        assert!(built.generated);
+    }
+
     #[test]
     fn numeric_modifiers_unconstrained() {
         assert_eq!(numeric_modifiers(-1), None);

crates/etl-postgres/src/store/schema.rs

@@ -282,8 +282,8 @@ pub async fn store_table_schema(
             r#"
             insert into etl.table_columns
             (table_schema_id, column_name, column_type, type_modifier, nullable,
-             ordinal_position, primary_key_ordinal_position, default_expression)
-            values ($1, $2, $3, $4, $5, $6, $7, $8)
+             ordinal_position, primary_key_ordinal_position, generated, default_expression)
+            values ($1, $2, $3, $4, $5, $6, $7, $8, $9)
             "#,
         )
         .bind(table_schema_id)
@@ -293,6 +293,7 @@ pub async fn store_table_schema(
         .bind(column_schema.nullable)
         .bind(column_schema.ordinal_position)
         .bind(column_schema.primary_key_ordinal_position)
+        .bind(column_schema.generated)
         .bind(&column_schema.default_expression)
         .execute(&mut *tx)
         .await?;
@@ -345,6 +346,7 @@ pub async fn load_table_schema_at_snapshot(
             tc.nullable,
             tc.ordinal_position,
             tc.primary_key_ordinal_position,
+            tc.generated,
             tc.default_expression
         from etl.table_schemas ts
         inner join etl.table_columns tc on ts.id = tc.table_schema_id
@@ -446,6 +448,7 @@ pub async fn load_table_schemas_at_snapshot(
             tc.nullable,
             tc.ordinal_position,
             tc.primary_key_ordinal_position,
+            tc.generated,
             tc.default_expression
         from latest_schemas ls
         inner join etl.table_columns tc on ls.id = tc.table_schema_id
@@ -625,6 +628,7 @@ fn parse_column_schema(row: &PgRow) -> ColumnSchema {
     let ordinal_position: i32 = row.get("ordinal_position");
     let primary_key_ordinal_position: Option<i32> = row.get("primary_key_ordinal_position");
     let nullable: bool = row.get("nullable");
+    let generated: bool = row.get("generated");
     let default_expression: Option<String> = row.get("default_expression");
 
     ColumnSchema::new(
@@ -635,6 +639,7 @@ fn parse_column_schema(row: &PgRow) -> ColumnSchema {
         nullable,
     )
     .with_primary_key_ordinal_position(primary_key_ordinal_position)
+    .with_generated(generated)
     .with_default_expression_option(default_expression)
 }
 

crates/etl/migrations/postgres_store/20260908120000_table_column_generated.down.sql

@@ -0,0 +1,10 @@
+-- Drops the stored generated-column flag.
+--
+-- Stored schemas become indistinguishable from non-generated ones, so the
+-- replication-mask fallbacks would treat generated columns as replicated and
+-- fail row decoding. Before running this down, revert the matching source
+-- migration so `etl.describe_table_schema` stops returning generated columns,
+-- and reset every publication to `publish_generated_columns = none`.
+
+alter table etl.table_columns
+    drop column if exists generated;

crates/etl/migrations/postgres_store/20260908120000_table_column_generated.up.sql

@@ -0,0 +1,10 @@
+-- Records whether a stored table column is a PostgreSQL stored generated column.
+--
+-- Existing rows are correctly false: they were snapshotted by a
+-- `etl.describe_table_schema` version that excluded generated columns entirely,
+-- so no stored row can describe one. The constant default avoids a table
+-- rewrite and keeps the column readable by the previous release, which simply
+-- ignores it.
+
+alter table etl.table_columns
+    add column if not exists generated boolean not null default false;