adapter: migrate mz_source_references to a materialized view

src/adapter/src/catalog/apply.rs

@@ -1482,9 +1482,10 @@ impl CatalogState {
             }
             StateUpdateKind::Item(item) => self.pack_item_update(item.id, diff),
             StateUpdateKind::Comment(_) => Vec::new(),
-            StateUpdateKind::SourceReferences(source_references) => {
-                self.pack_source_references_update(&source_references, diff)
-            }
+            // mz_source_references is a MaterializedView backed by
+            // mz_internal.mz_catalog_raw, so source reference rows do not
+            // produce builtin table updates here.
+            StateUpdateKind::SourceReferences(_) => Vec::new(),
             // mz_audit_events is a MaterializedView backed by
             // mz_internal.mz_catalog_raw, so audit log rows do not produce
             // builtin table updates here.

src/adapter/src/catalog/builtin_table_updates.rs

@@ -19,11 +19,9 @@ use mz_catalog::builtin::{
     MZ_CLUSTER_REPLICA_SIZES, MZ_COLUMNS, MZ_EGRESS_IPS, MZ_FUNCTIONS,
     MZ_HISTORY_RETENTION_STRATEGIES, MZ_INDEX_COLUMNS, MZ_LICENSE_KEYS, MZ_LIST_TYPES,
     MZ_MAP_TYPES, MZ_MATERIALIZED_VIEW_REFRESH_STRATEGIES, MZ_OBJECT_GLOBAL_IDS, MZ_OPERATORS,
-    MZ_PSEUDO_TYPES, MZ_REPLACEMENTS, MZ_ROLE_AUTH, MZ_SESSIONS, MZ_SOURCE_REFERENCES,
-    MZ_STORAGE_USAGE_BY_SHARD, MZ_SUBSCRIPTIONS, MZ_TYPE_PG_METADATA, MZ_TYPES,
-    MZ_WEBHOOKS_SOURCES,
+    MZ_PSEUDO_TYPES, MZ_REPLACEMENTS, MZ_ROLE_AUTH, MZ_SESSIONS, MZ_STORAGE_USAGE_BY_SHARD,
+    MZ_SUBSCRIPTIONS, MZ_TYPE_PG_METADATA, MZ_TYPES, MZ_WEBHOOKS_SOURCES,
 };
-use mz_catalog::durable::SourceReferences;
 use mz_catalog::memory::error::Error;
 use mz_catalog::memory::objects::{
     CatalogEntry, CatalogItem, DataSourceDesc, Func, Index, MaterializedView, Table,
@@ -872,52 +870,4 @@ impl CatalogState {
             diff,
         )
     }
-
-    pub fn pack_source_references_update(
-        &self,
-        source_references: &SourceReferences,
-        diff: Diff,
-    ) -> Vec<BuiltinTableUpdate<&'static BuiltinTable>> {
-        let source_id = source_references.source_id.to_string();
-        let updated_at = &source_references.updated_at;
-        source_references
-            .references
-            .iter()
-            .map(|reference| {
-                let mut row = Row::default();
-                let mut packer = row.packer();
-                packer.extend([
-                    Datum::String(&source_id),
-                    reference
-                        .namespace
-                        .as_ref()
-                        .map(|s| Datum::String(s))
-                        .unwrap_or(Datum::Null),
-                    Datum::String(&reference.name),
-                    Datum::TimestampTz(
-                        mz_ore::now::to_datetime(*updated_at)
-                            .try_into()
-                            .expect("must fit"),
-                    ),
-                ]);
-                if reference.columns.len() > 0 {
-                    packer
-                        .try_push_array(
-                            &[ArrayDimension {
-                                lower_bound: 1,
-                                length: reference.columns.len(),
-                            }],
-                            reference.columns.iter().map(|col| Datum::String(col)),
-                        )
-                        .expect(
-                            "columns is 1 dimensional, and its length is used for the array length",
-                        );
-                } else {
-                    packer.push(Datum::Null);
-                }
-
-                BuiltinTableUpdate::row(&*MZ_SOURCE_REFERENCES, row, diff)
-            })
-            .collect()
-    }
 }

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

@@ -462,6 +462,17 @@ static MIGRATIONS: LazyLock<Vec<MigrationStep>> = LazyLock::new(|| {
             MZ_INTERNAL_SCHEMA,
             "mz_object_dependencies",
         ),
+        // Converting mz_source_references from a builtin table to a
+        // materialized view over mz_catalog_raw changes its catalog
+        // fingerprint, so it needs an explicit replacement step. See the
+        // NOTE above: this version must stay at the workspace's current dev
+        // version until the change ships.
+        MigrationStep::replacement(
+            "26.43.0-dev.0",
+            CatalogItemType::MaterializedView,
+            MZ_INTERNAL_SCHEMA,
+            "mz_source_references",
+        ),
     ]
 });
 

src/catalog/src/builtin.rs

@@ -1159,7 +1159,7 @@ pub static BUILTINS_STATIC: LazyLock<Vec<Builtin<NameReference>>> = LazyLock::ne
         Builtin::Table(&MZ_INDEX_COLUMNS),
         Builtin::MaterializedView(&MZ_TABLES),
         // mz_sources is generated dynamically below with inlined builtin VALUES.
-        Builtin::Table(&MZ_SOURCE_REFERENCES),
+        Builtin::MaterializedView(&MZ_SOURCE_REFERENCES),
         Builtin::MaterializedView(&MZ_POSTGRES_SOURCES),
         Builtin::MaterializedView(&MZ_POSTGRES_SOURCE_TABLES),
         Builtin::MaterializedView(&MZ_MYSQL_SOURCE_TABLES),
@@ -1536,7 +1536,7 @@ pub static BUILTINS_STATIC: LazyLock<Vec<Builtin<NameReference>>> = LazyLock::ne
         // all items that follow it in the list.
         let insert_pos = builtin_items
             .iter()
-            .position(|b| matches!(b, Builtin::Table(t) if t.name == "mz_source_references"))
+            .position(|b| b.name() == "mz_source_references")
             .expect("mz_source_references must be present in builtin_items");
         builtin_items.insert(insert_pos, Builtin::MaterializedView(mz_sources_ref));
     }

src/catalog/src/builtin/mz_internal.rs

@@ -3566,38 +3566,72 @@ FROM commented",
     }
 });
 
-pub static MZ_SOURCE_REFERENCES: LazyLock<BuiltinTable> = LazyLock::new(|| BuiltinTable {
-    name: "mz_source_references",
-    schema: MZ_INTERNAL_SCHEMA,
-    oid: oid::TABLE_MZ_SOURCE_REFERENCES_OID,
-    desc: RelationDesc::builder()
-        .with_column("source_id", SqlScalarType::String.nullable(false))
-        .with_column("namespace", SqlScalarType::String.nullable(true))
-        .with_column("name", SqlScalarType::String.nullable(false))
-        .with_column(
-            "updated_at",
-            SqlScalarType::TimestampTz { precision: None }.nullable(false),
-        )
-        .with_column(
-            "columns",
-            SqlScalarType::Array(Box::new(SqlScalarType::String)).nullable(true),
+pub static MZ_SOURCE_REFERENCES: LazyLock<BuiltinMaterializedView> = LazyLock::new(|| {
+    BuiltinMaterializedView {
+        name: "mz_source_references",
+        schema: MZ_INTERNAL_SCHEMA,
+        oid: oid::MV_MZ_SOURCE_REFERENCES_OID,
+        desc: RelationDesc::builder()
+            .with_column("source_id", SqlScalarType::String.nullable(false))
+            .with_column("namespace", SqlScalarType::String.nullable(true))
+            .with_column("name", SqlScalarType::String.nullable(false))
+            .with_column(
+                "updated_at",
+                SqlScalarType::TimestampTz { precision: None }.nullable(false),
+            )
+            .with_column(
+                "columns",
+                SqlScalarType::Array(Box::new(SqlScalarType::String)).nullable(true),
+            )
+            .finish(),
+        column_comments: BTreeMap::new(),
+        // Dropping a source removes its `SourceReferences` record (see
+        // `Op::DropObjects` in the catalog's `transact`), so no filter against
+        // surviving items is needed here.
+        //
+        // A reference that records no columns reads as NULL rather than an
+        // empty array. `WITH ORDINALITY` keeps the upstream column order.
+        //
+        // `updated_at` is an `EpochMillis`, serialized as `{"millis": <u64>}`.
+        sql: "
+IN CLUSTER mz_catalog_server
+WITH (
+    ASSERT NOT NULL source_id,
+    ASSERT NOT NULL name,
+    ASSERT NOT NULL updated_at
+) AS
+SELECT
+    mz_internal.parse_catalog_id(data->'key'->'source') AS source_id,
+    reference->>'namespace' AS namespace,
+    reference->>'name' AS name,
+    to_timestamp(((data->'value'->'updated_at'->>'millis')::float8) / 1000.0) AS updated_at,
+    CASE WHEN jsonb_array_length(reference->'columns') > 0 THEN
+        ARRAY(
+            SELECT c.value
+            FROM jsonb_array_elements_text(reference->'columns')
+                 WITH ORDINALITY AS c(value, ord)
+            ORDER BY c.ord
         )
-        .finish(),
-    column_comments: BTreeMap::new(),
-    is_retained_metrics_object: false,
-    access: vec![PUBLIC_SELECT],
-    ontology: Some(Ontology {
-        entity_name: "source_reference",
-        description: "External references tracked by sources",
-        links: &const {
-            [OntologyLink {
-                name: "references_source",
-                target: "source",
-                properties: LinkProperties::fk("source_id", "id", Cardinality::ManyToOne),
-            }]
-        },
-        column_semantic_types: &[("source_id", SemanticType::CatalogItemId)],
-    }),
+    END AS columns
+FROM
+    mz_internal.mz_catalog_raw,
+    jsonb_array_elements(data->'value'->'references') AS reference
+WHERE data->>'kind' = 'SourceReferences'",
+        is_retained_metrics_object: false,
+        access: vec![PUBLIC_SELECT],
+        ontology: Some(Ontology {
+            entity_name: "source_reference",
+            description: "External references tracked by sources",
+            links: &const {
+                [OntologyLink {
+                    name: "references_source",
+                    target: "source",
+                    properties: LinkProperties::fk("source_id", "id", Cardinality::ManyToOne),
+                }]
+            },
+            column_semantic_types: &[("source_id", SemanticType::CatalogItemId)],
+        }),
+    }
 });
 
 pub static MZ_WEBHOOKS_SOURCES: LazyLock<BuiltinTable> = LazyLock::new(|| BuiltinTable {

src/pgrepr-consts/src/oid.rs

@@ -732,7 +732,7 @@ pub const INDEX_MZ_OBJECTS_IND_OID: u32 = 17011;
 pub const INDEX_MZ_COLUMNS_IND_OID: u32 = 17012;
 pub const INDEX_MZ_SECRETS_IND_OID: u32 = 17013;
 pub const INDEX_MZ_VIEWS_IND_OID: u32 = 17014;
-pub const TABLE_MZ_SOURCE_REFERENCES_OID: u32 = 17015;
+pub const MV_MZ_SOURCE_REFERENCES_OID: u32 = 17015;
 pub const SOURCE_MZ_CLUSTER_REPLICA_METRICS_HISTORY_OID: u32 = 17016;
 pub const INDEX_MZ_CLUSTER_REPLICA_METRICS_HISTORY_IND_OID: u32 = 17017;
 pub const VIEW_MZ_CLUSTER_REPLICA_UTILIZATION_HISTORY_OID: u32 = 17018;