fix(postgres): keep expression index DDL text and key order in sync

agents/common/src/main/java/com/dbx/agent/PostgresLikeAgent.java

@@ -542,10 +542,13 @@ public abstract class PostgresLikeAgent extends AbstractJdbcAgent {
             // dropped; their text comes from pg_get_indexdef instead of a.attname, and
             // is_expression records which case applied so Rust never has to guess from
             // characters in the text (#6312 review).
+            // `a.attname::text` keeps the cast: a bare COALESCE(name, text) resolves to
+            // `name`, which truncates an expression key part to 63 bytes (NAMEDATALEN - 1)
+            // and makes the rebuilt CREATE INDEX invalid (#9988).
             Map<String, IndexBuilder> byName = new LinkedHashMap<>();
             String sql = "SELECT i.relname AS index_name, am.amname AS index_type, " +
                 "(ix.indisunique AND ix.indisvalid) AS is_unique, ix.indisprimary AS is_primary, " +
-                "COALESCE(a.attname, " + profile.catalogPrefixedFunction("get_indexdef") + "(ix.indexrelid, k.n, true)) AS column_text, " +
+                "COALESCE(a.attname::text, " + profile.catalogPrefixedFunction("get_indexdef") + "(ix.indexrelid, k.n, true)) AS column_text, " +
                 "(a.attname IS NULL) AS is_expression, " +
                 "array_length(ix.indoption, 1) AS nkeyatts, k.n AS key_position, " +
                 "ix.indoption[(k.n - 1)::int] AS key_option " +

agents/common/src/test/java/com/dbx/agent/PostgresLikeAgentTest.java

@@ -201,6 +201,9 @@ class PostgresLikeAgentTest {
         String sql = String.join("\n", MetadataSqlFake.statements);
         assertTrue(sql.contains("LEFT JOIN pg_catalog.pg_attribute a ON a.attrelid = t.oid AND a.attnum = k.attnum AND k.attnum > 0"), sql);
         assertTrue(sql.contains("pg_catalog.pg_get_indexdef(ix.indexrelid, k.n, true)"), sql);
+        // #9988: a bare COALESCE(name, text) resolves to `name`, so PostgreSQL truncated a
+        // long expression key part to 63 bytes and the rebuilt CREATE INDEX was invalid.
+        assertTrue(sql.contains("COALESCE(a.attname::text, pg_catalog.pg_get_indexdef(ix.indexrelid, k.n, true)) AS column_text"), sql);
         assertTrue(sql.contains("ix.indisunique AND ix.indisvalid"), sql);
         assertTrue(sql.contains("ix.indoption[(k.n - 1)::int]"), sql);
         assertTrue(sql.contains("array_length(ix.indoption, 1) AS nkeyatts"), sql);

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

@@ -4782,9 +4782,13 @@ fn postgres_indexes_for_relations_query_tiers() -> [&'static str; 2] {
 // (schema, table) pair. Not merged into a shared fragment for the same
 // reason as the columns queries above — an alias would need renaming to
 // line up, and result columns are read positionally.
+// `COALESCE(a.attname::text, pg_get_indexdef(...))` keeps the cast: a bare
+// `COALESCE(name, text)` resolves to `name`, so PostgreSQL silently truncates an
+// expression key part to 63 bytes (NAMEDATALEN - 1) and the rebuilt CREATE INDEX
+// becomes invalid SQL (#9988).
 fn postgres_indexes_for_relations_sql() -> &'static str {
     "SELECT t.oid::bigint AS relid, i.relname AS index_name, \
-             array_agg(COALESCE(a.attname, pg_get_indexdef(ix.indexrelid, k.n::int, false)) ORDER BY k.n) AS columns, \
+             array_agg(COALESCE(a.attname::text, pg_get_indexdef(ix.indexrelid, k.n::int, false)) ORDER BY k.n) AS columns, \
              array_agg(CASE WHEN oc.opcdefault THEN NULL ELSE quote_ident(opcns.nspname) || '.' || quote_ident(oc.opcname) END ORDER BY k.n) AS column_opclasses, \
              (ix.indisunique AND ix.indisvalid) AS is_unique, \
              ix.indisprimary AS is_primary, \
@@ -4814,7 +4818,7 @@ fn postgres_indexes_for_relations_sql() -> &'static str {
 fn postgres_indexes_for_relations_compat_sql() -> &'static str {
     "SELECT t.oid::bigint AS relid, i.relname AS index_name, \
              ARRAY( \
-               SELECT COALESCE(a.attname, pg_get_indexdef(ix.indexrelid, pos.n, false)) \
+               SELECT COALESCE(a.attname::text, pg_get_indexdef(ix.indexrelid, pos.n, false)) \
                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 \
@@ -8844,8 +8848,12 @@ async fn execute_query_with_max_rows_inner(
 
 // Sibling of `postgres_indexes_for_relations_sql` (~line 3288), for a single
 // (schema, table) instead of a batch of oids — see the note there.
+// `COALESCE(a.attname::text, pg_get_indexdef(...))` keeps the cast: a bare
+// `COALESCE(name, text)` resolves to `name`, so PostgreSQL silently truncates an
+// expression key part to 63 bytes (NAMEDATALEN - 1) and the rebuilt CREATE INDEX
+// becomes invalid SQL (#9988).
 const POSTGRES_INDEXES_SQL: &str = "SELECT i.relname AS index_name, \
-             array_agg(COALESCE(a.attname, pg_get_indexdef(ix.indexrelid, k.n::int, false)) ORDER BY k.n) AS columns, \
+             array_agg(COALESCE(a.attname::text, pg_get_indexdef(ix.indexrelid, k.n::int, false)) ORDER BY k.n) AS columns, \
              array_agg(CASE WHEN oc.opcdefault THEN NULL ELSE quote_ident(opcns.nspname) || '.' || quote_ident(oc.opcname) END ORDER BY k.n) AS column_opclasses, \
              (ix.indisunique AND ix.indisvalid) AS is_unique, \
              ix.indisprimary AS is_primary, \
@@ -8874,7 +8882,7 @@ const POSTGRES_INDEXES_SQL: &str = "SELECT i.relname AS index_name, \
 // 3312) — see the note on `POSTGRES_INDEXES_SQL` above.
 const POSTGRES_INDEXES_COMPAT_SQL: &str = "SELECT i.relname AS index_name, \
              ARRAY( \
-               SELECT COALESCE(a.attname, pg_get_indexdef(ix.indexrelid, pos.n, false)) \
+               SELECT COALESCE(a.attname::text, pg_get_indexdef(ix.indexrelid, pos.n, false)) \
                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 \
@@ -13207,6 +13215,23 @@ mod tests {
         }
     }
 
+    #[test]
+    fn postgres_index_queries_cast_attname_to_text() {
+        // Regression for #9988: `COALESCE(a.attname, pg_get_indexdef(...))` resolves to the
+        // `name` type, so PostgreSQL silently truncated an expression key part to 63 bytes
+        // (NAMEDATALEN - 1). The rebuilt CREATE INDEX then failed with a syntax error and the
+        // whole publish transaction rolled back.
+        for sql in [
+            POSTGRES_INDEXES_SQL,
+            POSTGRES_INDEXES_COMPAT_SQL,
+            postgres_indexes_for_relations_sql(),
+            postgres_indexes_for_relations_compat_sql(),
+        ] {
+            assert!(sql.contains("COALESCE(a.attname::text, pg_get_indexdef("), "{sql}");
+            assert!(!sql.contains("COALESCE(a.attname, pg_get_indexdef("), "{sql}");
+        }
+    }
+
     #[test]
     fn postgres_column_metadata_marks_only_owned_integer_sequence_defaults_as_serial() {
         let modern_sql = [POSTGRES_COLUMNS_SQL, postgres_columns_for_relations_sql()];

crates/dbx-sql/src/schema_diff.rs

@@ -3995,16 +3995,38 @@ fn is_postgres_family_ddl(db_type: DatabaseType) -> bool {
     )
 }
 
-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).
+/// One PostgreSQL index key part for DDL: the key text plus its operator class
+/// and, for B-tree keys, the explicit ordering.
+///
+/// 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).
+///
+/// `pg_index.indoption` carries bit 0 = DESC and bit 1 = NULLS FIRST per key.
+/// Dropping it silently rebuilt `col DESC NULLS LAST` as a plain ASC key (#8559
+/// fixed exactly this for data transfer; the sync DDL kept losing it, #9988).
+/// Only B-tree keys carry meaningful flags, so other access methods keep the
+/// bare key text - same rule as the transfer path.
+fn postgres_index_column_sql(
+    column: &str,
+    is_expression: Option<bool>,
+    opclass: Option<&str>,
+    key_options: Option<i16>,
+    db_type: DatabaseType,
+) -> String {
     let trimmed = column.trim();
-    let is_expression = is_expression.unwrap_or(false);
-    if is_expression {
-        trimmed.to_string()
-    } else {
-        quote_id(column, db_type)
+    let base = if is_expression.unwrap_or(false) { trimmed.to_string() } else { quote_id(column, db_type) };
+    let with_opclass = match opclass.filter(|opclass| !opclass.is_empty()) {
+        Some(opclass) => format!("{base} {opclass}"),
+        None => base,
+    };
+    match key_options {
+        Some(options) => format!(
+            "{with_opclass} {} NULLS {}",
+            if options & 1 != 0 { "DESC" } else { "ASC" },
+            if options & 2 != 0 { "FIRST" } else { "LAST" }
+        ),
+        None => with_opclass,
     }
 }
 
@@ -4030,7 +4052,19 @@ pub fn create_index_sql(table_name: &str, index: &IndexInfo, db_type: DatabaseTy
             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)
+                let opclass = index.column_opclasses.get(i).and_then(|opclass| opclass.as_deref());
+                let key_options = index
+                    .index_type
+                    .as_deref()
+                    .filter(|index_type| index_type.trim().eq_ignore_ascii_case("btree"))
+                    .and_then(|_| index.key_options.get(i).copied());
+                postgres_index_column_sql(
+                    column,
+                    index.key_is_expression.get(i).copied(),
+                    opclass,
+                    key_options,
+                    db_type,
+                )
             } else {
                 quote_id(column, db_type)
             }
@@ -8499,6 +8533,139 @@ mod tests {
         assert!(!sql.contains(&format!("\"{expression_key_part}\"")));
     }
 
+    #[test]
+    fn postgres_sync_index_ddl_keeps_key_order_opclass_and_full_expression() {
+        // Regression for #9988: `pg_index.indoption` (bit 0 = DESC, bit 1 = NULLS FIRST)
+        // and `pg_index.indclass` were dropped by the sync DDL, so a published index came
+        // back as plain ASC keys with default null ordering. The expression key part is
+        // also asserted verbatim - the driver used to return it truncated to 63 bytes
+        // because `COALESCE(name, text)` resolves to `name` (see postgres.rs).
+        let expression_key_part =
+            "(((category)::text = ANY ((ARRAY['industrial'::character varying, 'macro'::character varying])::text[])))";
+        let new_index = index(IndexInfo {
+            name: "index_repro_rank".to_string(),
+            columns: vec![
+                "((has_current_analysis AND (NOT dirty) AND researchable))".to_string(),
+                "researchable".to_string(),
+                expression_key_part.to_string(),
+                "((new_events_24h > 0))".to_string(),
+                "evidence_count".to_string(),
+                "material_at".to_string(),
+                "topic_id".to_string(),
+            ],
+            is_unique: false,
+            is_primary: false,
+            filter: Some("(NOT suppressed)".to_string()),
+            index_type: Some("btree".to_string()),
+            included_columns: None,
+            comment: None,
+            key_is_expression: vec![true, false, true, true, false, false, false],
+            column_opclasses: vec![None, None, None, None, None, None, Some("pg_catalog.text_pattern_ops".to_string())],
+            key_options: vec![3, 3, 3, 3, 3, 3, 3],
+            constraint_backed: false,
+        });
+
+        let sql = generate_schema_sync_sql(
+            &[TableDiff {
+                diff_type: "modified".to_string(),
+                object_type: Some("table".to_string()),
+                name: "index_repro".to_string(),
+                target_name: None,
+                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::Postgres,
+            Some("public"),
+            false,
+            None,
+            &[],
+        );
+
+        assert!(
+            sql.contains(
+                "CREATE INDEX \"index_repro_rank\" ON \"public\".\"index_repro\" USING btree (((has_current_analysis AND (NOT dirty) AND researchable)) DESC NULLS FIRST, \"researchable\" DESC NULLS FIRST"
+            ),
+            "{sql}"
+        );
+        assert!(sql.contains(&format!("{expression_key_part} DESC NULLS FIRST")), "{sql}");
+        assert!(sql.contains("\"topic_id\" pg_catalog.text_pattern_ops DESC NULLS FIRST"), "{sql}");
+        assert!(sql.contains("WHERE (NOT suppressed);"), "{sql}");
+    }
+
+    #[test]
+    fn postgres_sync_index_ddl_keeps_non_btree_keys_bare() {
+        // `indoption` is only meaningful for B-tree keys, so a GIN index must not gain
+        // `ASC NULLS LAST` suffixes even when the introspection reports the flags.
+        let new_index = index(IndexInfo {
+            name: "idx_payload_gin".to_string(),
+            columns: vec!["payload".to_string()],
+            is_unique: false,
+            is_primary: false,
+            filter: None,
+            index_type: Some("gin".to_string()),
+            included_columns: None,
+            comment: None,
+            key_is_expression: vec![false],
+            column_opclasses: vec![Some("public.gin_trgm_ops".to_string())],
+            key_options: vec![0],
+            constraint_backed: false,
+        });
+
+        let sql = generate_schema_sync_sql(
+            &[TableDiff {
+                diff_type: "modified".to_string(),
+                object_type: Some("table".to_string()),
+                name: "events".to_string(),
+                target_name: None,
+                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::Postgres,
+            Some("public"),
+            false,
+            None,
+            &[],
+        );
+
+        assert!(sql.contains("USING gin (\"payload\" public.gin_trgm_ops)"), "{sql}");
+        assert!(!sql.contains("NULLS LAST"), "{sql}");
+        assert!(!sql.contains("NULLS FIRST"), "{sql}");
+    }
+
     #[test]
     fn quotes_real_columns_whose_names_contain_expression_like_characters() {
         // PR #6312 review: a quoted column identifier can legitimately contain whitespace,