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>;
 }
 
 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,15 @@ static MIGRATIONS: LazyLock<Vec<MigrationStep>> = LazyLock::new(|| {
             MZ_CATALOG_SCHEMA,
             "mz_kafka_sources",
         ),
+        // `mz_type_pg_metadata` gained a trailing `typsend` column. 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,62 @@ mod tests {
 
     use super::*;
 
+    /// Recomputes `mz_pgrepr::regproc::NAMES` from the builtin function registry
+    /// and fails with the corrected table when the checked-in copy has drifted.
+    /// It is checked in as data because `mz-pgrepr` sits below this crate in the
+    /// dependency graph and so cannot read the registry itself.
+    #[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() {
+            // Mirrors PostgreSQL's `regprocout`, which qualifies a name that
+            // would not resolve back to this OID on its own.
+            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,10 @@ 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))
+        // NOTE: `pg_type_all_databases` still needs `COALESCE` on this column,
+        // because its `LEFT JOIN` against this table yields NULLs for types with
+        // no PostgreSQL metadata.
+        .with_column("typsend", SqlScalarType::Oid.nullable(false))
         .finish(),
     column_comments: BTreeMap::new(),
     is_retained_metrics_object: false,
@@ -4372,6 +4376,7 @@ 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))
+            .with_column("typsend", SqlScalarType::RegProc.nullable(false))
             .finish(),
         column_comments: BTreeMap::new(),
         sql: "
@@ -4440,7 +4445,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