catalog: add pg_type.typsend and type typreceive as regproc

The Apache Arrow ADBC PostgreSQL driver runs a type-resolution query at
connect time that references pg_catalog.pg_type.typsend. Materialize did
not expose that column, so every adbc_driver_postgresql connection failed
with `column "typsend" does not exist` before any user query ran.

Carry the real PostgreSQL typsend OID per builtin type, verified against a
live PostgreSQL 16 server for all 76 types that have a PG counterpart. Nine
types get 0, matching PostgreSQL, which gives them no binary send function.
_mz_aclitem is Materialize-invented and takes array_send by analogy with
every other array type.

typsend is appended to mz_internal.mz_type_pg_metadata, then projected
through mz_internal.pg_type_all_databases and pg_catalog.pg_type. The new
columns are appended rather than placed at PostgreSQL's ordinal positions,
so existing column positions stay stable.

pg_catalog.pg_type.typreceive also becomes regproc, which is what
PostgreSQL declares and what lets a client's typreceive::TEXT comparison
resolve a function name instead of digits. The underlying all-databases
view keeps oid, because it backs a builtin index and resolving a regproc to
a name reads current_database(), which is not materializable.

Adds a test/adbc mzcompose suite covering the driver end to end. Connecting
and streaming rows are hard assertions. Full Arrow type fidelity is not yet
reachable, because Materialize encodes regproc over pgwire as a bare OID
rather than as the function name, so the driver's name-keyed type map
misses and every column resolves to Arrow binary. Those tests are marked
expectedFailure so they report an unexpected success once that encoding is
fixed.

Author: claude

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,16 @@ 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 is backward compatible, so the shard's schema can be evolved
+        // in place. See the NOTE above: this version must stay at the
+        // workspace's current dev version until the change ships.
+        MigrationStep::evolution(
+            "26.36.0-dev.0",
+            CatalogItemType::Table,
+            MZ_INTERNAL_SCHEMA,
+            "mz_type_pg_metadata",
+        ),
     ]
 });
 

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,9 @@ 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))
+        // Always populated, but declared nullable because persist schema
+        // evolution only accepts an appended field if it is nullable.
+        .with_column("typsend", SqlScalarType::Oid.nullable(true))
         .finish(),
     column_comments: BTreeMap::new(),
     is_retained_metrics_object: false,
@@ -4372,6 +4375,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 +4446,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