catalog: add pg_type.typsend and type typreceive as regproc

ci/test/pipeline.template.yml

@@ -760,6 +760,17 @@ steps:
     agents:
       queue: hetzner-aarch64-4cpu-8gb
 
+  - id: adbc
+    label: Arrow ADBC driver
+    depends_on: build-aarch64
+    timeout_in_minutes: 20
+    inputs: [test/adbc]
+    plugins:
+      - ./ci/plugins/mzcompose:
+          composition: adbc
+    agents:
+      queue: hetzner-aarch64-4cpu-8gb
+
   - id: chbench-demo
     topics: [debezium, kafka, mysql]
     label: chbench smoke

console/types/materialize.d.ts

@@ -4004,6 +4004,7 @@ export interface MzTypePgMetadata {
   id: Generated<string>;
   typinput: Generated<number>;
   typreceive: Generated<number>;
+  typsend: Generated<number | null>;
 }
 
 export interface MzTypes {
@@ -4245,6 +4246,7 @@ export interface PgTypeAllDatabases {
   typowner: number;
   typreceive: number;
   typrelid: number;
+  typsend: string;
   typtype: string;
   typtypmod: number;
 }

src/adapter/src/catalog.rs

@@ -2928,6 +2928,7 @@ mod tests {
                 array: u32,
                 input: u32,
                 receive: u32,
+                send: u32,
             }
 
             struct PgOper {
@@ -2967,7 +2968,7 @@ mod tests {
             let pg_type: BTreeMap<_, _> = query(
                 &client,
                 sql!(
-                    "SELECT oid, typname, typtype::text, typelem, typarray, typinput::oid, typreceive::oid as typreceive FROM pg_type"
+                    "SELECT oid, typname, typtype::text, typelem, typarray, typinput::oid, typreceive::oid as typreceive, typsend::oid as typsend FROM pg_type"
                 ),
                 &[],
             )
@@ -2983,6 +2984,7 @@ mod tests {
                         array: row.get("typarray"),
                         input: row.get("typinput"),
                         receive: row.get("typreceive"),
+                        send: row.get("typsend"),
                     };
                     (oid, pg_type)
                 })
@@ -3074,10 +3076,15 @@ mod tests {
                             ty.oid, pg_ty.name, ty.name,
                         );
 
-                        let (typinput_oid, typreceive_oid) = match &ty.details.pg_metadata {
-                            None => (0, 0),
-                            Some(pgmeta) => (pgmeta.typinput_oid, pgmeta.typreceive_oid),
-                        };
+                        let (typinput_oid, typreceive_oid, typsend_oid) =
+                            match &ty.details.pg_metadata {
+                                None => (0, 0, 0),
+                                Some(pgmeta) => (
+                                    pgmeta.typinput_oid,
+                                    pgmeta.typreceive_oid,
+                                    pgmeta.typsend_oid,
+                                ),
+                            };
                         assert_eq!(
                             typinput_oid, pg_ty.input,
                             "type {} has typinput OID {:?} in mz but {:?} in pg",
@@ -3088,6 +3095,15 @@ mod tests {
                             "type {} has typreceive OID {:?} in mz but {:?} in pg",
                             ty.name, typreceive_oid, pg_ty.receive,
                         );
+                        // Unlike typinput and typreceive below, typsend is not also
+                        // checked against `func_oids`. Nothing resolves a typsend OID
+                        // to a name, so the corresponding `*send` functions are
+                        // deliberately not registered as builtins.
+                        assert_eq!(
+                            typsend_oid, pg_ty.send,
+                            "type {} has typsend OID {:?} in mz but {:?} in pg",
+                            ty.name, typsend_oid, pg_ty.send,
+                        );
                         if typinput_oid != 0 {
                             assert!(
                                 func_oids.contains(&typinput_oid),

src/adapter/src/catalog/builtin_table_updates.rs

@@ -1185,6 +1185,7 @@ impl CatalogState {
                     Datum::String(&id.to_string()),
                     Datum::UInt32(pg_metadata.typinput_oid),
                     Datum::UInt32(pg_metadata.typreceive_oid),
+                    Datum::UInt32(pg_metadata.typsend_oid),
                 ]),
                 diff,
             ));

src/adapter/src/catalog/open/builtin_schema_migration.rs

@@ -301,6 +301,27 @@ static MIGRATIONS: LazyLock<Vec<MigrationStep>> = LazyLock::new(|| {
             MZ_CATALOG_SCHEMA,
             "mz_kafka_sources",
         ),
+        // `mz_type_pg_metadata` gained a trailing `typsend` column. Appending a
+        // column would be backward compatible enough for `Evolution`, but
+        // `Replacement` is the right mechanism for a builtin table anyway.
+        // `Coordinator::bootstrap_tables` retracts every system table's persist
+        // snapshot and re-derives the rows from the catalog on each boot, so a
+        // fresh shard loses nothing. More importantly, only `Replacement`
+        // reports the item in `MigrationResult::replaced_items`, which is what
+        // makes the read-only coordinator back-fill the new shard so dependent
+        // dataflows such as `pg_type_all_databases_ind` can hydrate. Under
+        // `Evolution` a read-only replica would instead have to read the old
+        // parts through the schema-migration path, which no builtin table has
+        // ever exercised.
+        //
+        // See the NOTE above: this version must stay at the workspace's current
+        // dev version until the change ships.
+        MigrationStep::replacement(
+            "26.36.0-dev.0",
+            CatalogItemType::Table,
+            MZ_INTERNAL_SCHEMA,
+            "mz_type_pg_metadata",
+        ),
     ]
 });
 

src/catalog/src/builtin.rs

@@ -1659,6 +1659,64 @@ mod tests {
 
     use super::*;
 
+    /// `mz_pgrepr::regproc::NAMES` is a table of every builtin function OID and
+    /// the text a `regproc` renders it as. It has to be checked in as data
+    /// because `mz-pgrepr` sits below this crate in the dependency graph and so
+    /// cannot read the registry itself. This test is what keeps it honest: it
+    /// recomputes the whole table from the registry and, on drift, prints the
+    /// corrected table to paste into `src/pgrepr-consts/src/regproc.rs`.
+    #[mz_ore::test]
+    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+    fn test_regproc_names_match_builtin_functions() {
+        // `effective_search_path` unconditionally prepends these two, so a
+        // uniquely named function in either resolves from its bare name.
+        const IMPLICITLY_SEARCHED: &[&str] = &[MZ_CATALOG_SCHEMA, PG_CATALOG_SCHEMA];
+
+        // A bare name only identifies one OID when exactly one impl anywhere in
+        // the registry carries it, so count impls across schemas.
+        let mut impls_per_name: BTreeMap<&str, usize> = BTreeMap::new();
+        for func in BUILTINS::funcs() {
+            *impls_per_name.entry(func.name).or_default() += func.inner.func_impls().len();
+        }
+
+        let mut expected: BTreeMap<u32, String> = BTreeMap::new();
+        for func in BUILTINS::funcs() {
+            // The rule `regprocout` applies: print the bare name when it
+            // resolves back to this OID on its own, otherwise qualify it.
+            let rendered =
+                if impls_per_name[func.name] == 1 && IMPLICITLY_SEARCHED.contains(&func.schema) {
+                    func.name.to_string()
+                } else {
+                    format!("{}.{}", func.schema, func.name)
+                };
+            for details in func.inner.func_impls() {
+                let previous = expected.insert(details.oid, rendered.clone());
+                assert_eq!(
+                    previous, None,
+                    "two builtin functions share OID {}",
+                    details.oid
+                );
+            }
+        }
+
+        let actual: BTreeMap<u32, String> = mz_pgrepr::regproc::NAMES
+            .iter()
+            .map(|(oid, name)| (*oid, name.to_string()))
+            .collect();
+
+        if actual != expected {
+            let table: String = expected
+                .iter()
+                .map(|(oid, name)| format!("    ({}, \"{}\"),\n", oid, name))
+                .collect();
+            panic!(
+                "mz_pgrepr::regproc::NAMES has drifted from the builtin function registry. \
+                 Replace the entries of NAMES in src/pgrepr-consts/src/regproc.rs with:\n{}",
+                table
+            );
+        }
+    }
+
     #[mz_ore::test]
     #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
     fn test_builtin_type_schema() {

src/catalog/src/builtin/mz_catalog.rs

@@ -194,6 +194,7 @@ pub const TYPE_MZ_ACL_ITEM_ARRAY: BuiltinType<NameReference> = BuiltinType {
         pg_metadata: Some(CatalogTypePgMetadata {
             typinput_oid: 750,
             typreceive_oid: 2400,
+            typsend_oid: 2401,
         }),
     },
 };

src/catalog/src/builtin/mz_internal.rs

@@ -586,6 +586,14 @@ pub static MZ_TYPE_PG_METADATA: LazyLock<BuiltinTable> = LazyLock::new(|| Builti
         .with_column("id", SqlScalarType::String.nullable(false))
         .with_column("typinput", SqlScalarType::Oid.nullable(false))
         .with_column("typreceive", SqlScalarType::Oid.nullable(false))
+        // `pack_type_update` always writes an OID here, so this declaration is
+        // wider than the data requires. Downstream `COALESCE`s on this column
+        // are there for the `LEFT JOIN` against this table, not for this.
+        //
+        // TODO: Tighten to `nullable(false)` to match `typinput` and
+        // `typreceive`. That also changes the descs of the views projecting the
+        // column.
+        .with_column("typsend", SqlScalarType::Oid.nullable(true))
         .finish(),
     column_comments: BTreeMap::new(),
     is_retained_metrics_object: false,
@@ -4372,6 +4380,9 @@ pub static PG_TYPE_ALL_DATABASES: LazyLock<BuiltinView> = LazyLock::new(|| {
             .with_column("typcollation", SqlScalarType::Oid.nullable(false))
             .with_column("typdefault", SqlScalarType::String.nullable(true))
             .with_column("database_name", SqlScalarType::String.nullable(true))
+            // Appended rather than placed at PostgreSQL's ordinal position, to
+            // keep the existing columns' positions stable.
+            .with_column("typsend", SqlScalarType::RegProc.nullable(false))
             .finish(),
         column_comments: BTreeMap::new(),
         sql: "
@@ -4440,7 +4451,8 @@ SELECT
     -- MZ doesn't support COLLATE so typcollation is filled with 0
     0::pg_catalog.oid AS typcollation,
     NULL::pg_catalog.text AS typdefault,
-    d.name as database_name
+    d.name as database_name,
+    COALESCE(mz_internal.mz_type_pg_metadata.typsend, 0)::pg_catalog.regproc AS typsend
 FROM
     mz_catalog.mz_types
     LEFT JOIN mz_internal.mz_type_pg_metadata ON mz_catalog.mz_types.id = mz_internal.mz_type_pg_metadata.id