fix(schema-diff): require explicit expression index provenance

crates/dbx-core/src/db/clickhouse_driver.rs

@@ -540,6 +540,7 @@ fn clickhouse_index_from_skipping_row(row: &[serde_json::Value]) -> IndexInfo {
         }),
         included_columns: None,
         comment: None,
+        key_is_expression: Vec::new(),
     }
 }
 
@@ -709,6 +710,7 @@ pub async fn list_indexes(client: &ChClient, database: &str, table: &str) -> Res
             index_type: Some("primary".to_string()),
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         });
     }
 

crates/dbx-core/src/db/cloudflare_d1/mod.rs

@@ -199,6 +199,7 @@ pub async fn list_indexes(client: &CloudflareD1Client, _schema: &str, table: &st
             index_type: None,
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         });
     }
     Ok(indexes)

crates/dbx-core/src/db/manticoresearch.rs

@@ -209,6 +209,7 @@ fn index_info_from_row(row: &mysql_async::Row) -> IndexInfo {
         index_type: Some(get_str_by_name(row, "Type")),
         included_columns: None,
         comment: (!comment_parts.is_empty()).then(|| comment_parts.join(", ")),
+        key_is_expression: Vec::new(),
     }
 }
 

crates/dbx-core/src/db/mongo_driver.rs

@@ -821,6 +821,7 @@ fn index_info_from_model(model: IndexModel) -> IndexInfo {
         index_type,
         included_columns: None,
         comment: None,
+        key_is_expression: Vec::new(),
     }
 }
 
@@ -3525,6 +3526,7 @@ mod tests {
             index_type: Some("email: 1".to_string()),
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         });
 
         assert_eq!(spec.name, "email_1");
@@ -3549,6 +3551,7 @@ mod tests {
             index_type: None,
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         });
 
         assert_eq!(
@@ -3791,6 +3794,7 @@ mod tests {
                 index_type: Some("_id: 1".to_string()),
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             },
             IndexInfo {
                 name: "users_email_unique".to_string(),
@@ -3801,6 +3805,7 @@ mod tests {
                 index_type: Some("email: 1".to_string()),
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             },
             IndexInfo {
                 name: "users_status_idx".to_string(),
@@ -3811,6 +3816,7 @@ mod tests {
                 index_type: Some("status: 1".to_string()),
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             },
         ];
         let after = vec![before[0].clone(), before[2].clone()];

crates/dbx-core/src/db/mysql.rs

@@ -5170,6 +5170,7 @@ pub async fn list_indexes(pool: &MySqlPool, database: &str, table: &str) -> Resu
                     index_type: Some(get_str_by_name(&row, "INDEX_TYPE")),
                     included_columns: None,
                     comment: get_opt_str(&row, "INDEX_COMMENT").filter(|value| !value.is_empty()),
+                    key_is_expression: Vec::new(),
                 });
                 index_position
             };

crates/dbx-core/src/db/mysql_compatible.rs

@@ -323,6 +323,7 @@ fn table_key_index(name: &str, line: &str, is_unique: bool, is_primary: bool, in
         index_type: Some(index_type.to_string()),
         included_columns: None,
         comment: None,
+        key_is_expression: Vec::new(),
     })
 }
 
@@ -342,6 +343,7 @@ fn secondary_index(line: &str) -> Option<IndexInfo> {
         index_type: mysql_keyword_argument(after_name, "USING").or_else(|| Some("INDEX".to_string())),
         included_columns: None,
         comment: mysql_quoted_string_argument(after_name, "COMMENT"),
+        key_is_expression: Vec::new(),
     })
 }
 

crates/dbx-core/src/db/ob_oracle.rs

@@ -241,6 +241,7 @@ pub async fn list_indexes(pool: &mysql_async::Pool, schema: &str, table: &str) -
                 index_type: None,
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             }
         })
         .collect())

crates/dbx-core/src/db/postgres.rs

@@ -5281,7 +5281,8 @@ const POSTGRES_INDEXES_SQL: &str = "SELECT i.relname AS index_name, \
              am.amname AS index_type, \
              ix.indnkeyatts AS nkeyatts, \
              ix.indkey AS indkey, \
-             obj_description(i.oid, 'pg_class') AS index_comment \
+             obj_description(i.oid, 'pg_class') AS index_comment, \
+             array_agg(a.attname IS NULL ORDER BY k.n) AS key_is_expression \
              FROM pg_index ix \
              JOIN pg_class t ON t.oid = ix.indrelid \
              JOIN pg_class i ON i.oid = ix.indexrelid \
@@ -5309,7 +5310,16 @@ const POSTGRES_INDEXES_COMPAT_SQL: &str = "SELECT i.relname AS index_name, \
              am.amname AS index_type, \
              NULL::smallint AS nkeyatts, \
              ix.indkey AS indkey, \
-             obj_description(i.oid, 'pg_class') AS index_comment \
+             obj_description(i.oid, 'pg_class') AS index_comment, \
+             ARRAY( \
+               SELECT a.attname IS NULL \
+               FROM generate_series(1, array_length(string_to_array(ix.indkey::text, ' '), 1)) AS pos(n) \
+               LEFT JOIN pg_attribute a \
+                 ON a.attrelid = t.oid \
+                AND a.attnum = (string_to_array(ix.indkey::text, ' '))[pos.n]::int2 \
+                AND a.attnum > 0 \
+               ORDER BY pos.n \
+             ) AS key_is_expression \
              FROM pg_index ix \
              JOIN pg_class t ON t.oid = ix.indrelid \
              JOIN pg_class i ON i.oid = ix.indexrelid \
@@ -5385,6 +5395,11 @@ async fn list_indexes_with_sql(
             let split_at = nkeyatts.min(all_cols.len());
             let key_cols = all_cols[..split_at].to_vec();
             let included = if split_at < all_cols.len() { all_cols[split_at..].to_vec() } else { vec![] };
+            // `a.attname IS NULL` at a given key position means that key part came back from
+            // pg_get_indexdef (a functional/expression key part), not from a real column (#6295).
+            let all_is_expr: Vec<bool> = row.try_get::<_, Vec<bool>>(9).unwrap_or_default();
+            let key_is_expression =
+                if all_is_expr.len() == all_cols.len() { all_is_expr[..split_at].to_vec() } else { Vec::new() };
             IndexInfo {
                 name: pg_row_try_string(row, 0),
                 columns: key_cols,
@@ -5394,6 +5409,7 @@ async fn list_indexes_with_sql(
                 index_type: row.try_get::<_, Option<String>>(5).ok().flatten(),
                 included_columns: if included.is_empty() { None } else { Some(included) },
                 comment: row.try_get::<_, Option<String>>(8).ok().flatten(),
+                key_is_expression,
             }
         })
         .collect())
@@ -8432,6 +8448,15 @@ mod tests {
         assert!(POSTGRES_INDEXES_COMPAT_SQL.contains("string_to_array(ix.indkey::text, ' ')"));
     }
 
+    #[test]
+    fn postgres_index_metadata_tracks_expression_key_provenance() {
+        // #6295 fix: each index key part's `a.attname IS NULL` tags whether it came from a real
+        // column or from pg_get_indexdef, so DDL generation never has to guess from the text.
+        assert!(POSTGRES_INDEXES_SQL.contains("array_agg(a.attname IS NULL ORDER BY k.n) AS key_is_expression"));
+        assert!(POSTGRES_INDEXES_COMPAT_SQL.contains("SELECT a.attname IS NULL"));
+        assert!(POSTGRES_INDEXES_COMPAT_SQL.contains("AS key_is_expression"));
+    }
+
     #[test]
     fn postgres_owner_metadata_casts_relkind_to_text() {
         assert!(POSTGRES_OWNERS_SQL.contains("c.relkind::text AS relkind"));

crates/dbx-core/src/db/questdb.rs

@@ -176,6 +176,7 @@ pub async fn list_indexes(pool: &Pool, _schema: &str, table: &str) -> Result<Vec
                 index_type: None,
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             }
         })
         .collect())

crates/dbx-core/src/db/rqlite_driver.rs

@@ -179,6 +179,7 @@ pub async fn list_indexes(client: &RqliteClient, _schema: &str, table: &str) ->
             index_type: None,
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         });
     }
 

crates/dbx-core/src/db/sqlite.rs

@@ -2272,6 +2272,7 @@ pub async fn list_indexes(pool: &SqliteHandle, schema: &str, table: &str) -> Res
                     index_type: None,
                     included_columns: None,
                     comment: None,
+                    key_is_expression: Vec::new(),
                 });
             }
             Ok(indexes)

crates/dbx-core/src/db/sqlserver.rs

@@ -2656,6 +2656,7 @@ pub async fn list_indexes(client: &mut SqlServerClient, schema: &str, table: &st
                     Some(inc_str.split(',').map(|s| s.to_string()).collect())
                 },
                 comment: row.get::<&str, _>(7).filter(|s: &&str| !s.is_empty()).map(|s: &str| s.to_string()),
+                key_is_expression: Vec::new(),
             }
         })
         .collect())

crates/dbx-core/src/db/turso_driver.rs

@@ -221,6 +221,7 @@ pub async fn list_indexes(client: &TursoClient, _schema: &str, table: &str) -> R
             index_type: None,
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         });
     }
 

crates/dbx-core/src/docs/dbml.rs

@@ -599,6 +599,7 @@ mod tests {
             index_type: None,
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         };
         let table = doc_table("orders", vec![col("user_id", "integer")], vec![index]);
 
@@ -621,6 +622,7 @@ mod tests {
             index_type: None,
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         };
         let table = doc_table("orders", vec![col("reference", "text")], vec![index]);
         let mut warnings = Vec::new();
@@ -639,6 +641,7 @@ mod tests {
             index_type: None,
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         };
         let mut id = col("id", "integer");
         id.is_primary_key = true;
@@ -660,6 +663,7 @@ mod tests {
             index_type: None,
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         };
         let table = doc_table("orders", vec![col("status", "text")], vec![index]);
 
@@ -861,6 +865,7 @@ mod tests {
             index_type: None,
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         };
         let out =
             to_dbml(&snapshot(vec![doc_table("orders", vec![col("status", "text")], vec![index])], vec!["public"]));

crates/dbx-core/src/docs/relations.rs

@@ -193,6 +193,7 @@ mod tests {
             index_type: None,
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         }
     }
 

crates/dbx-core/src/mongo_ops.rs

@@ -1846,6 +1846,7 @@ for line in sys.stdin:
                     index_type: Some("_id: 1".to_string()),
                     included_columns: None,
                     comment: None,
+                    key_is_expression: Vec::new(),
                 },
                 IndexInfo {
                     name: "email_1".to_string(),
@@ -1856,6 +1857,7 @@ for line in sys.stdin:
                     index_type: Some("email: 1".to_string()),
                     included_columns: None,
                     comment: None,
+                    key_is_expression: Vec::new(),
                 },
             ],
             1,

crates/dbx-core/src/schema.rs

@@ -8572,6 +8572,7 @@ mod ddl_tests {
             index_type: Some("btree".to_string()),
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         }];
         let partition_info = db::postgres::PostgresTablePartitionInfo {
             is_partition: true,
@@ -8618,6 +8619,7 @@ mod ddl_tests {
             index_type: Some("btree".to_string()),
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         }];
         let partition_info = db::postgres::PostgresTablePartitionInfo {
             is_partition: true,

crates/dbx-core/src/schema_diff.rs

@@ -3162,23 +3162,56 @@ fn mysql_index_column_sql(column: &str) -> String {
     }
 }
 
+fn is_postgres_family_ddl(db_type: DatabaseType) -> bool {
+    matches!(
+        db_type,
+        DatabaseType::Postgres
+            | DatabaseType::Redshift
+            | DatabaseType::Gaussdb
+            | DatabaseType::Kingbase
+            | DatabaseType::Highgo
+            | DatabaseType::Vastbase
+            | DatabaseType::OpenGauss
+            | DatabaseType::Kwdb
+            | DatabaseType::Firebird
+            | DatabaseType::Vertica
+            | DatabaseType::Exasol
+            | DatabaseType::Uxdb
+    )
+}
+
+fn postgres_index_column_sql(column: &str, is_expression: Option<bool>, db_type: DatabaseType) -> String {
+    // Expression/functional index key parts (e.g. from pg_get_indexdef) arrive as raw
+    // expression text, not a plain column name; quoting the whole expression as an
+    // identifier turns it into a literal column reference that doesn't exist (#6295).
+    let trimmed = column.trim();
+    let is_expression = is_expression.unwrap_or(false);
+    if is_expression {
+        trimmed.to_string()
+    } else {
+        quote_id(column, db_type)
+    }
+}
+
 fn create_index_sql(table_name: &str, index: &IndexInfo, db_type: DatabaseType, schema: Option<&str>) -> String {
     use crate::sql_dialect::ddl_profile::IndexTypePlacement;
     let profile = profile_for(db_type);
     let table = qualified_name(table_name, db_type, schema);
-    let columns =
-        index
-            .columns
-            .iter()
-            .map(|column| {
-                if db_type == DatabaseType::Mysql {
-                    mysql_index_column_sql(column)
-                } else {
-                    quote_id(column, db_type)
-                }
-            })
-            .collect::<Vec<_>>()
-            .join(", ");
+    let columns = index
+        .columns
+        .iter()
+        .enumerate()
+        .map(|(i, column)| {
+            if db_type == DatabaseType::Mysql {
+                mysql_index_column_sql(column)
+            } else if is_postgres_family_ddl(db_type) {
+                postgres_index_column_sql(column, index.key_is_expression.get(i).copied(), db_type)
+            } else {
+                quote_id(column, db_type)
+            }
+        })
+        .collect::<Vec<_>>()
+        .join(", ");
     let unique = if index.is_unique { "UNIQUE " } else { "" };
     let index_type = index.index_type.as_deref().unwrap_or_default();
     let (type_prefix, using_before_on, using_suffix) = if index_type.is_empty() {
@@ -4352,6 +4385,7 @@ mod tests {
             index_type: overrides.index_type,
             included_columns: overrides.included_columns,
             comment: overrides.comment,
+            key_is_expression: overrides.key_is_expression,
         }
     }
 
@@ -5239,6 +5273,7 @@ mod tests {
                 index_type: None,
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             })],
             &[index(IndexInfo {
                 name: "idx_orders_status".to_string(),
@@ -5249,6 +5284,7 @@ mod tests {
                 index_type: None,
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             })],
         );
 
@@ -5269,6 +5305,7 @@ mod tests {
             index_type: None,
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         });
         let target_index = index(IndexInfo {
             name: "test_UNIQUE".to_string(),
@@ -5279,6 +5316,7 @@ mod tests {
             index_type: None,
             included_columns: None,
             comment: None,
+            key_is_expression: Vec::new(),
         });
 
         let diffs = diff_indexes(std::slice::from_ref(&source_index), &[target_index]);
@@ -5319,6 +5357,261 @@ mod tests {
         assert!(!sql.contains("`((case"));
     }
 
+    #[test]
+    fn preserves_bare_expression_in_postgres_family_unique_index_ddl() {
+        // Regression for #6295: highgo/postgres-family expression index key parts (e.g. from
+        // pg_get_indexdef) arrived as raw expression text in IndexInfo.columns; quoting the
+        // whole expression as an identifier turned it into a literal (and nonexistent) column.
+        let expression_key_part = "COALESCE(height, '-1'::integer::double precision)";
+        let new_index = index(IndexInfo {
+            name: "uq_tankong_sta_type_time".to_string(),
+            columns: vec![
+                "sta_id".to_string(),
+                "data_type".to_string(),
+                "data_time".to_string(),
+                expression_key_part.to_string(),
+            ],
+            is_unique: true,
+            is_primary: false,
+            filter: None,
+            index_type: None,
+            included_columns: None,
+            comment: None,
+            key_is_expression: vec![false, false, false, true],
+        });
+
+        let sql = generate_schema_sync_sql(
+            &[TableDiff {
+                diff_type: "modified".to_string(),
+                object_type: Some("table".to_string()),
+                name: "tankong_data".to_string(),
+                columns: None,
+                indexes: Some(vec![IndexDiff {
+                    diff_type: "added".to_string(),
+                    name: new_index.name.clone(),
+                    source: Some(new_index),
+                    target: None,
+                    changes: vec![],
+                }]),
+                foreign_keys: None,
+                triggers: None,
+                ddl: None,
+                target_ddl: None,
+                source_table_comment: None,
+                target_table_comment: None,
+                sync_sql: None,
+            }],
+            &[],
+            &[],
+            &[],
+            &[],
+            DatabaseType::Highgo,
+            Some("public"),
+            false,
+            None,
+            &[],
+        );
+
+        assert!(sql.contains(&format!(
+            "CREATE UNIQUE INDEX \"uq_tankong_sta_type_time\" ON \"public\".\"tankong_data\" (\"sta_id\", \"data_type\", \"data_time\", {expression_key_part})"
+        )));
+        assert!(!sql.contains(&format!("\"{expression_key_part}\"")));
+    }
+
+    #[test]
+    fn quotes_real_columns_whose_names_contain_expression_like_characters() {
+        // PR #6312 review: a quoted column identifier can legitimately contain whitespace,
+        // `(`, or `::` (e.g. PostgreSQL metadata returning the ordinary column name
+        // `order item` through a.attname). The old character-based heuristic mistook such
+        // columns for expressions and left them bare, generating an invalid
+        // `CREATE INDEX ... (order item)` instead of `CREATE INDEX ... ("order item")`.
+        // With real per-key provenance (`key_is_expression`), only genuine expression key
+        // parts from pg_get_indexdef are left unquoted.
+        let expression_key_part = "COALESCE(height, '-1'::integer::double precision)";
+        let new_index = index(IndexInfo {
+            name: "uq_weird_columns".to_string(),
+            columns: vec![
+                "order item".to_string(),
+                "a(b)".to_string(),
+                "a::b".to_string(),
+                expression_key_part.to_string(),
+            ],
+            key_is_expression: vec![false, false, false, true],
+            is_unique: true,
+            is_primary: false,
+            filter: None,
+            index_type: None,
+            included_columns: None,
+            comment: None,
+        });
+
+        let sql = generate_schema_sync_sql(
+            &[TableDiff {
+                diff_type: "modified".to_string(),
+                object_type: Some("table".to_string()),
+                name: "tankong_data".to_string(),
+                columns: None,
+                indexes: Some(vec![IndexDiff {
+                    diff_type: "added".to_string(),
+                    name: new_index.name.clone(),
+                    source: Some(new_index),
+                    target: None,
+                    changes: vec![],
+                }]),
+                foreign_keys: None,
+                triggers: None,
+                ddl: None,
+                target_ddl: None,
+                source_table_comment: None,
+                target_table_comment: None,
+                sync_sql: None,
+            }],
+            &[],
+            &[],
+            &[],
+            &[],
+            DatabaseType::Highgo,
+            Some("public"),
+            false,
+            None,
+            &[],
+        );
+
+        assert!(sql.contains(&format!(
+            "CREATE UNIQUE INDEX \"uq_weird_columns\" ON \"public\".\"tankong_data\" (\"order item\", \"a(b)\", \"a::b\", {expression_key_part})"
+        )));
+        assert!(!sql.contains(&format!("\"{expression_key_part}\"")));
+
+        // Real PostgreSQL-family validation: parse the generated DDL with the PostgreSQL
+        // dialect so this also proves the statement is syntactically valid, not just that the
+        // expected substring is present.
+        use sqlparser::dialect::PostgreSqlDialect;
+        use sqlparser::parser::Parser;
+        let create_index_sql = sql
+            .lines()
+            .find(|line| line.starts_with("CREATE UNIQUE INDEX \"uq_weird_columns\""))
+            .expect("generated DDL should include the CREATE INDEX statement");
+        let statements = Parser::parse_sql(&PostgreSqlDialect {}, create_index_sql)
+            .unwrap_or_else(|error| panic!("generated DDL must be valid PostgreSQL: {error}\nSQL: {create_index_sql}"));
+        assert_eq!(statements.len(), 1);
+    }
+
+    #[test]
+    fn quotes_expression_like_column_names_without_agent_provenance() {
+        for db_type in [DatabaseType::Kingbase, DatabaseType::Vastbase] {
+            let new_index = index(IndexInfo {
+                name: "idx_weird_columns".to_string(),
+                columns: vec!["order item".to_string(), "a(b)".to_string(), "a::b".to_string()],
+                key_is_expression: Vec::new(),
+                is_unique: false,
+                is_primary: false,
+                filter: None,
+                index_type: None,
+                included_columns: None,
+                comment: None,
+            });
+
+            let sql = generate_schema_sync_sql(
+                &[TableDiff {
+                    diff_type: "modified".to_string(),
+                    object_type: Some("table".to_string()),
+                    name: "tankong_data".to_string(),
+                    columns: None,
+                    indexes: Some(vec![IndexDiff {
+                        diff_type: "added".to_string(),
+                        name: new_index.name.clone(),
+                        source: Some(new_index),
+                        target: None,
+                        changes: vec![],
+                    }]),
+                    foreign_keys: None,
+                    triggers: None,
+                    ddl: None,
+                    target_ddl: None,
+                    source_table_comment: None,
+                    target_table_comment: None,
+                    sync_sql: None,
+                }],
+                &[],
+                &[],
+                &[],
+                &[],
+                db_type,
+                Some("public"),
+                false,
+                None,
+                &[],
+            );
+
+            assert!(sql.contains("(\"order item\", \"a(b)\", \"a::b\")"), "{db_type:?}: {sql}");
+        }
+    }
+
+    #[tokio::test]
+    #[ignore = "requires DBX_TEST_POSTGRES_URL pointing at a writable PostgreSQL-family database"]
+    async fn real_postgres_round_trip_quotes_columns_and_leaves_expressions_bare() {
+        // PR #6312 review: exercise the full path end-to-end against a real PostgreSQL server
+        // instead of only asserting on generated text. Introspects a table whose unique index
+        // mixes a real column with an expression-hostile name ("order item", i.e. exactly the
+        // a.attname case the reviewer called out) and a genuine pg_get_indexdef expression key
+        // part, generates DDL with the same `create_index_sql` schema-diff sync uses, and
+        // executes that DDL back against the database to prove it's actually valid — not just
+        // plausible-looking text.
+        let url = std::env::var("DBX_TEST_POSTGRES_URL").expect("DBX_TEST_POSTGRES_URL");
+        let pool =
+            crate::db::postgres::connect(&url, std::time::Duration::from_secs(5)).await.expect("connect postgres");
+        let schema = format!("dbx_key_expr_{}", uuid::Uuid::new_v4().simple());
+        crate::db::postgres::execute_query(&pool, &format!("CREATE SCHEMA {schema}")).await.expect("create schema");
+
+        let exercise = async {
+            crate::db::postgres::execute_query(
+                &pool,
+                &format!(
+                    "CREATE TABLE {schema}.tankong_data (\
+                     \"order item\" integer, data_type text, data_time timestamp, height double precision)"
+                ),
+            )
+            .await?;
+            crate::db::postgres::execute_query(
+                &pool,
+                &format!(
+                    "CREATE UNIQUE INDEX uq_tankong_sta_type_time ON {schema}.tankong_data \
+                     (\"order item\", data_type, data_time, \
+                     (COALESCE(height, '-1'::integer::double precision)))"
+                ),
+            )
+            .await?;
+
+            let indexes = crate::db::postgres::list_indexes(&pool, &schema, "tankong_data").await?;
+            let index = indexes
+                .into_iter()
+                .find(|index| index.name == "uq_tankong_sta_type_time")
+                .ok_or_else(|| "introspection should return the created index".to_string())?;
+
+            // Regenerate the index DDL through the exact same production function schema-diff
+            // sync calls, then execute it back against the real database to prove it's valid.
+            let ddl = create_index_sql("tankong_data", &index, DatabaseType::Highgo, Some(&schema));
+            crate::db::postgres::execute_query(&pool, &format!("DROP INDEX {schema}.uq_tankong_sta_type_time")).await?;
+            crate::db::postgres::execute_query(&pool, &ddl).await?;
+
+            Ok::<_, String>((index, ddl))
+        }
+        .await;
+
+        let cleanup = crate::db::postgres::execute_query(&pool, &format!("DROP SCHEMA {schema} CASCADE")).await;
+        cleanup.expect("drop schema");
+        let (index, recreate_ddl) = exercise.expect("exercise real postgres round trip");
+
+        assert_eq!(
+            index.columns,
+            vec!["order item", "data_type", "data_time", "COALESCE(height, '-1'::integer::double precision)"]
+        );
+        assert_eq!(index.key_is_expression, vec![false, false, false, true]);
+        assert!(recreate_ddl.contains("\"order item\""));
+        assert!(recreate_ddl.contains("COALESCE(height, '-1'::integer::double precision)"));
+        assert!(!recreate_ddl.contains("\"COALESCE"));
+    }
+
     #[test]
     fn detects_foreign_key_additions_removals_and_target_changes() {
         let diffs = diff_foreign_keys(
@@ -5394,6 +5687,7 @@ mod tests {
                     index_type: None,
                     included_columns: None,
                     comment: None,
+                    key_is_expression: Vec::new(),
                 })),
                 target: None,
                 changes: Vec::new(),
@@ -5698,6 +5992,7 @@ mod tests {
                     index_type: None,
                     included_columns: None,
                     comment: None,
+                    key_is_expression: Vec::new(),
                 })),
                 target: None,
                 changes: Vec::new(),
@@ -6301,6 +6596,7 @@ mod tests {
                     index_type: Some("btree".to_string()),
                     included_columns: Some(vec!["User ID".to_string()]),
                     comment: None,
+                    key_is_expression: Vec::new(),
                 })],
                 foreign_keys: vec![foreign_key(ForeignKeyInfo {
                     name: "Order User FK".to_string(),
@@ -6363,6 +6659,7 @@ mod tests {
                     index_type: Some("BTREE".to_string()),
                     included_columns: None,
                     comment: Some("status lookup".to_string()),
+                    key_is_expression: Vec::new(),
                 })],
                 foreign_keys: vec![foreign_key(ForeignKeyInfo {
                     name: "user-fk".to_string(),
@@ -6415,6 +6712,7 @@ mod tests {
                     index_type: None,
                     included_columns: None,
                     comment: None,
+                    key_is_expression: Vec::new(),
                 })],
                 foreign_keys: vec![foreign_key(ForeignKeyInfo {
                     name: "parent item fk".to_string(),
@@ -7179,6 +7477,7 @@ mod tests {
                     index_type: None,
                     included_columns: None,
                     comment: None,
+                    key_is_expression: Vec::new(),
                 })),
                 target: None,
                 changes: vec![],
@@ -7219,6 +7518,7 @@ mod tests {
                     index_type: None,
                     included_columns: None,
                     comment: None,
+                    key_is_expression: Vec::new(),
                 })),
                 changes: vec![],
             }]),
@@ -7616,6 +7916,7 @@ mod tests {
                 index_type: Some("BTREE".into()),
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             })],
             &[index(IndexInfo {
                 name: "idx_t".into(),
@@ -7626,6 +7927,7 @@ mod tests {
                 index_type: Some("HASH".into()),
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             })],
         );
         assert_eq!(diffs.len(), 1, "index type diff detected");
@@ -7644,6 +7946,7 @@ mod tests {
                 index_type: Some("FULLTEXT".into()),
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             })],
             &[index(IndexInfo {
                 name: "idx_t".into(),
@@ -7654,6 +7957,7 @@ mod tests {
                 index_type: None,
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             })],
         );
         assert_eq!(diffs[0].changes.iter().filter(|c| c.contains("FULLTEXT")).count(), 1, "fulltext change");
@@ -7672,6 +7976,7 @@ mod tests {
                 index_type: None,
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             })],
             &[index(IndexInfo {
                 name: "idx_t".into(),
@@ -7682,6 +7987,7 @@ mod tests {
                 index_type: None,
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             })],
         );
         assert_eq!(diffs.len(), 1, "order diff detected");
@@ -7701,6 +8007,7 @@ mod tests {
                 index_type: None,
                 included_columns: Some(vec!["b".into(), "c".into()]),
                 comment: None,
+                key_is_expression: Vec::new(),
             })],
             &[index(IndexInfo {
                 name: "idx_t".into(),
@@ -7711,6 +8018,7 @@ mod tests {
                 index_type: None,
                 included_columns: Some(vec!["b".into()]),
                 comment: None,
+                key_is_expression: Vec::new(),
             })],
         );
         assert_eq!(diffs.len(), 1, "included columns diff detected");
@@ -7729,6 +8037,7 @@ mod tests {
                 index_type: None,
                 included_columns: Some(vec!["b".into()]),
                 comment: None,
+                key_is_expression: Vec::new(),
             })],
             &[index(IndexInfo {
                 name: "idx_t".into(),
@@ -7739,6 +8048,7 @@ mod tests {
                 index_type: None,
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             })],
         );
         assert_eq!(diffs.len(), 1, "included added");
@@ -7757,6 +8067,7 @@ mod tests {
                 index_type: None,
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             })],
             &[index(IndexInfo {
                 name: "idx_t".into(),
@@ -7767,6 +8078,7 @@ mod tests {
                 index_type: None,
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             })],
         );
         assert_eq!(diffs.len(), 1, "filter diff");
@@ -7787,6 +8099,7 @@ mod tests {
                     index_type: None,
                     included_columns: None,
                     comment: None,
+                    key_is_expression: Vec::new(),
                 }),
                 index(IndexInfo {
                     name: "idx_modified".into(),
@@ -7797,6 +8110,7 @@ mod tests {
                     index_type: Some("BTREE".into()),
                     included_columns: None,
                     comment: None,
+                    key_is_expression: Vec::new(),
                 }),
             ],
             &[
@@ -7809,6 +8123,7 @@ mod tests {
                     index_type: None,
                     included_columns: None,
                     comment: None,
+                    key_is_expression: Vec::new(),
                 }),
                 index(IndexInfo {
                     name: "idx_modified".into(),
@@ -7819,6 +8134,7 @@ mod tests {
                     index_type: None,
                     included_columns: None,
                     comment: None,
+                    key_is_expression: Vec::new(),
                 }),
             ],
         );
@@ -8004,6 +8320,7 @@ mod tests {
                         index_type: Some("BTREE".into()),
                         included_columns: None,
                         comment: None,
+                        key_is_expression: Vec::new(),
                     })),
                     target: None,
                     changes: vec![],
@@ -8021,6 +8338,7 @@ mod tests {
                         index_type: None,
                         included_columns: None,
                         comment: None,
+                        key_is_expression: Vec::new(),
                     })),
                     changes: vec![],
                 },
@@ -8413,6 +8731,7 @@ mod tests {
                 filter: None,
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             }),
             target: None,
             changes: vec![],
@@ -8664,6 +8983,7 @@ mod tests {
                 index_type: None,
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             }),
             target: None,
             changes: vec![],

crates/dbx-core/src/table_structure_sql/indexes.rs

@@ -132,6 +132,55 @@ fn mysql_index_column_sql(column: &str) -> String {
     }
 }
 
+fn postgres_index_column_sql(column: &str, is_expression: bool) -> String {
+    // Expression/functional index key parts arrive as raw expression text, not a plain
+    // column name; quoting the whole expression as an identifier turns it into a literal
+    // column reference that doesn't exist (#6295).
+    if is_expression {
+        column.trim().to_string()
+    } else {
+        quote_ident(StructureDialect::Postgres, column)
+    }
+}
+
+/// Per-key expression provenance for `columns` (the edited/current key list), computed once for
+/// the whole index. The editor only lets users toggle real table columns onto an index (see
+/// TableStructureEditor.vue's column checkbox list) — there is no free-text expression input —
+/// so an edited key part is only ever an expression when it corresponds to one from the original
+/// introspected snapshot.
+///
+/// Matches each edited key part against the first *not yet claimed* original key part with the
+/// same text, rather than the first textual match overall: a real column whose name happens to
+/// equal another key's expression text (or two key parts with identical text) can otherwise steal
+/// the wrong original's provenance. Consuming each original slot at most once keeps provenance
+/// tied to its true ordinal position instead of being re-derived by scanning for a text match
+/// (#6312 review). A key without explicit provenance is treated as a real column: the editor can
+/// only add table columns, and guessing from the identifier text breaks valid quoted names.
+fn key_expression_flags(index: &EditableStructureIndex, columns: &[String]) -> Vec<bool> {
+    let original = match &index.original {
+        Some(original) if !original.key_is_expression.is_empty() => original,
+        _ => return vec![false; columns.len()],
+    };
+    let mut consumed = vec![false; original.columns.len()];
+    columns
+        .iter()
+        .map(|column| {
+            let claimed = original
+                .columns
+                .iter()
+                .enumerate()
+                .find(|(i, original_column)| !consumed[*i] && *original_column == column);
+            match claimed {
+                Some((i, _)) => {
+                    consumed[i] = true;
+                    original.key_is_expression.get(i).copied().unwrap_or(false)
+                }
+                None => false,
+            }
+        })
+        .collect()
+}
+
 pub(super) fn build_drop_index_sql(
     database_type: Option<DatabaseType>,
     dialect: StructureDialect,
@@ -179,11 +228,15 @@ pub(super) fn build_create_index_statements(
 
     let unique = if index.is_unique { "UNIQUE " } else { "" };
     let replace = if or_replace { "OR REPLACE " } else { "" };
+    let key_is_expression = key_expression_flags(index, &columns);
     let cols = columns
         .iter()
-        .map(|column| {
+        .enumerate()
+        .map(|(i, column)| {
             if dialect == StructureDialect::Mysql {
                 mysql_index_column_sql(column)
+            } else if dialect == StructureDialect::Postgres {
+                postgres_index_column_sql(column, key_is_expression[i])
             } else {
                 quote_ident(dialect, column)
             }

crates/dbx-core/src/table_structure_sql/sqlite_rebuild.rs

@@ -2018,6 +2018,7 @@ mod tests {
                 index_type: None,
                 included_columns: None,
                 comment: None,
+                key_is_expression: Vec::new(),
             }),
             marked_for_drop: true,
         });

crates/dbx-core/src/table_structure_sql/tests.rs

@@ -90,6 +90,7 @@ fn existing_index(name: &str, columns: &[&str], is_unique: bool) -> EditableStru
         index_type: None,
         included_columns: None,
         comment: None,
+        key_is_expression: Vec::new(),
     });
     index
 }
@@ -169,6 +170,7 @@ fn builds_mysql_column_and_index_changes() {
         index_type: None,
         included_columns: None,
         comment: None,
+        key_is_expression: Vec::new(),
     });
     let mut email_index = index("uniq_users_email", &["email"]);
     email_index.is_unique = true;
@@ -741,6 +743,7 @@ fn builds_informix_column_and_index_changes() {
         index_type: None,
         included_columns: None,
         comment: None,
+        key_is_expression: Vec::new(),
     });
     let mut email_index = index("uniq_users_email", &["email"]);
     email_index.is_unique = true;
@@ -939,6 +942,7 @@ fn iris_drop_index_includes_table_name() {
         index_type: None,
         included_columns: None,
         comment: None,
+        key_is_expression: Vec::new(),
     });
 
     let result = build_table_structure_change_sql(TableStructureSqlOptions {
@@ -1218,6 +1222,7 @@ fn gbase8a_uses_limited_mysql_ddl() {
         index_type: None,
         included_columns: None,
         comment: None,
+        key_is_expression: Vec::new(),
     });
 
     let result = build_table_structure_change_sql(TableStructureSqlOptions {
@@ -1632,6 +1637,74 @@ fn builds_postgres_create_table_with_comments_and_index() {
     );
 }
 
+#[test]
+fn quotes_expression_like_new_index_columns_without_provenance() {
+    let expression_like_column = "COALESCE(height, '-1'::integer::double precision)";
+    let idx = index("idx_expression_like_column", &[expression_like_column]);
+
+    let result = build_create_table_sql(TableStructureSqlOptions {
+        database_type: Some(DatabaseType::Kingbase),
+        schema: Some("public".to_string()),
+        table_name: "tankong_data".to_string(),
+        columns: vec![column(expression_like_column)],
+        indexes: vec![idx],
+        foreign_keys: Vec::new(),
+        triggers: Vec::new(),
+        table_comment: None,
+        original_table_comment: None,
+    });
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert!(result.statements.iter().any(|statement| statement.contains(&format!("(\"{expression_like_column}\")"))));
+}
+
+#[test]
+fn preserves_key_provenance_when_rebuilding_an_untouched_postgres_index() {
+    // PR #6312 review: a quoted column identifier can legitimately contain whitespace, `(`,
+    // or `::` (e.g. PostgreSQL metadata returning the ordinary column name `order item`
+    // through a.attname). Regenerating an *unedited* existing index (e.g. only its uniqueness
+    // changed) must trust the original snapshot's real per-key provenance rather than guessing
+    // from characters, so a weirdly-named real column stays quoted and only the genuine
+    // expression key part stays bare.
+    let expression_key_part = "COALESCE(height, '-1'::integer::double precision)";
+    let mut changed = existing_index("uq_weird_columns", &["order item", "a(b)", "a::b", expression_key_part], false);
+    changed.is_unique = true;
+    changed.original.as_mut().unwrap().key_is_expression = vec![false, false, false, true];
+
+    let result =
+        build_table_structure_change_sql(index_change_options(DatabaseType::Postgres, Some("public"), changed));
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert_eq!(result.statements.len(), 2);
+    assert!(result.statements[0].starts_with("DROP INDEX "));
+    assert_eq!(
+        result.statements[1],
+        format!(
+            "CREATE UNIQUE INDEX \"uq_weird_columns\" ON \"public\".\"USERS\" (\"order item\", \"a(b)\", \"a::b\", {expression_key_part});"
+        )
+    );
+}
+
+#[test]
+fn preserves_key_provenance_by_ordinal_position_not_first_text_match() {
+    // PR #6312 review (round 2): provenance must stay tied to each key part's original ordinal
+    // slot, not be re-derived by scanning for the first original key part with matching text. Two
+    // key parts sharing identical text with different provenance (a pathological but real case —
+    // e.g. a genuine expression key part and a real column whose name happens to equal that same
+    // text) must not let the first one's provenance leak onto the second.
+    let mut changed = existing_index("idx_dup", &["dup", "dup"], false);
+    changed.is_unique = true;
+    changed.original.as_mut().unwrap().key_is_expression = vec![true, false];
+
+    let result =
+        build_table_structure_change_sql(index_change_options(DatabaseType::Postgres, Some("public"), changed));
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert_eq!(result.statements.len(), 2);
+    assert!(result.statements[0].starts_with("DROP INDEX "));
+    assert_eq!(result.statements[1], "CREATE UNIQUE INDEX \"idx_dup\" ON \"public\".\"USERS\" (dup, \"dup\");");
+}
+
 #[test]
 fn create_table_trims_table_name_whitespace_for_all_statements() {
     let mut id = column("id");
@@ -1706,6 +1779,7 @@ fn qualifies_attached_sqlite_table_and_index_changes() {
         index_type: None,
         included_columns: None,
         comment: None,
+        key_is_expression: Vec::new(),
     });
     let email_index = index("idx_users_email", &["email"]);
 
@@ -4615,6 +4689,7 @@ fn oscar_drop_index_with_schema_qualifier() {
         index_type: None,
         included_columns: None,
         comment: None,
+        key_is_expression: Vec::new(),
     });
 
     let result = build_table_structure_change_sql(TableStructureSqlOptions {

crates/dbx-core/src/table_structure_sql/types.rs

@@ -119,6 +119,10 @@ pub struct IndexInfo {
     pub included_columns: Option<Vec<String>>,
     #[serde(default)]
     pub comment: Option<String>,
+    /// Parallel to `columns`: `true` at index `i` means `columns[i]` is a raw expression
+    /// (e.g. sourced from `pg_get_indexdef`), not a plain column name.
+    #[serde(default)]
+    pub key_is_expression: Vec<bool>,
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]

crates/dbx-core/src/transfer.rs

@@ -9977,6 +9977,7 @@ mod tests {
             index_type: Some("btree".to_string()),
             included_columns: Some(vec!["created_at".to_string()]),
             comment: Some("lookup index".to_string()),
+            key_is_expression: Vec::new(),
         }];
         let foreign_keys = vec![
             db::ForeignKeyInfo {

crates/dbx-core/src/types.rs

@@ -405,6 +405,11 @@ pub struct IndexInfo {
     pub index_type: Option<String>,
     pub included_columns: Option<Vec<String>>,
     pub comment: Option<String>,
+    /// Parallel to `columns`: `true` at index `i` means `columns[i]` is a raw expression
+    /// (e.g. sourced from `pg_get_indexdef`), not a plain column name. Empty when the
+    /// introspection source doesn't track this (provenance unknown for that dialect/path).
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub key_is_expression: Vec<bool>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]