From 329a9f6e7de70464baaab494bc6ee92dcf43a696 Mon Sep 17 00:00:00 2001
From: Roshan Jobanputra <roshan@materialize.com>
Date: Thu, 17 Oct 2024 14:58:06 -0700
Subject: [PATCH] builtins: Populate envelope/key-format/value-format columns
 on mz_kafka_source_tables table

---
 .../content/sql/system-catalog/mz_internal.md |   3 +
 .../src/catalog/builtin_table_updates.rs      |  27 +++-
 src/catalog/src/builtin.rs                    |   3 +
 src/catalog/src/memory/objects.rs             | 144 +++++++++---------
 .../autogenerated/mz_internal.slt             |   3 +
 5 files changed, 105 insertions(+), 75 deletions(-)

diff --git a/doc/user/content/sql/system-catalog/mz_internal.md b/doc/user/content/sql/system-catalog/mz_internal.md
index ebf48da608f33..89ce961008f84 100644
--- a/doc/user/content/sql/system-catalog/mz_internal.md
+++ b/doc/user/content/sql/system-catalog/mz_internal.md
@@ -578,6 +578,9 @@ table and the corresponding upstream Kafka topic being ingested.
 | ------------------- | ---------------- | --------                                                                                                       |
 | `id`                | [`text`]         | The ID of the table. Corresponds to [`mz_catalog.mz_tables.id`](../mz_catalog#mz_tables).                   |
 | `topic`             | [`text`]         | The topic being ingested. |
+| `envelope_type`     | [`text`]         | The [envelope](/sql/create-source/#envelopes) type: `none`, `upsert`, or `debezium`. `NULL` for other source types. |
+| `key_format`        | [`text`]         | The [format](/sql/create-source/#formats) of the Kafka message key: `avro`, `protobuf`, `csv`, `regex`, `bytes`, `json`, `text`, or `NULL`. |
+| `value_format`      | [`text`]         | The [format](/sql/create-source/#formats) of the Kafka message value: `avro`, `protobuf`, `csv`, `regex`, `bytes`, `json`, `text`. `NULL` for other source types. |
 
 <!--
 ## `mz_prepared_statement_history`
diff --git a/src/adapter/src/catalog/builtin_table_updates.rs b/src/adapter/src/catalog/builtin_table_updates.rs
index 2d55a8f136a70..37bc00cd7a720 100644
--- a/src/adapter/src/catalog/builtin_table_updates.rs
+++ b/src/adapter/src/catalog/builtin_table_updates.rs
@@ -537,7 +537,17 @@ impl CatalogState {
                                 "kafka" => {
                                     mz_ore::soft_assert_eq_no_log!(external_reference.len(), 1);
                                     let topic = external_reference[0].to_ast_string();
-                                    self.pack_kafka_source_tables_update(id, &topic, diff)
+                                    let envelope = data_source.envelope();
+                                    let (key_format, value_format) = data_source.formats();
+
+                                    self.pack_kafka_source_tables_update(
+                                        id,
+                                        &topic,
+                                        envelope,
+                                        key_format,
+                                        value_format,
+                                        diff,
+                                    )
                                 }
                                 s => unreachable!("{s} sources do not have tables"),
                             }
@@ -551,7 +561,7 @@ impl CatalogState {
             CatalogItem::Source(source) => {
                 let source_type = source.source_type();
                 let connection_id = source.connection_id();
-                let envelope = source.envelope();
+                let envelope = source.data_source.envelope();
                 let cluster_entry = match source.data_source {
                     // Ingestion exports don't have their own cluster, but
                     // run on their ingestion's cluster.
@@ -563,7 +573,7 @@ impl CatalogState {
 
                 let cluster_id = cluster_entry.item().cluster_id().map(|id| id.to_string());
 
-                let (key_format, value_format) = source.formats();
+                let (key_format, value_format) = source.data_source.formats();
 
                 let mut updates = self.pack_source_update(
                     id,
@@ -975,11 +985,20 @@ impl CatalogState {
         &self,
         id: GlobalId,
         topic: &str,
+        envelope: Option<&str>,
+        key_format: Option<&str>,
+        value_format: Option<&str>,
         diff: Diff,
     ) -> Vec<BuiltinTableUpdate<&'static BuiltinTable>> {
         vec![BuiltinTableUpdate {
             id: &*MZ_KAFKA_SOURCE_TABLES,
-            row: Row::pack_slice(&[Datum::String(&id.to_string()), Datum::String(topic)]),
+            row: Row::pack_slice(&[
+                Datum::String(&id.to_string()),
+                Datum::String(topic),
+                Datum::from(envelope),
+                Datum::from(key_format),
+                Datum::from(value_format),
+            ]),
             diff,
         }]
     }
diff --git a/src/catalog/src/builtin.rs b/src/catalog/src/builtin.rs
index 2cb7f97baa29a..533e647313fec 100644
--- a/src/catalog/src/builtin.rs
+++ b/src/catalog/src/builtin.rs
@@ -2119,6 +2119,9 @@ pub static MZ_KAFKA_SOURCE_TABLES: LazyLock<BuiltinTable> = LazyLock::new(|| Bui
     desc: RelationDesc::builder()
         .with_column("id", ScalarType::String.nullable(false))
         .with_column("topic", ScalarType::String.nullable(false))
+        .with_column("envelope_type", ScalarType::String.nullable(true))
+        .with_column("key_format", ScalarType::String.nullable(true))
+        .with_column("value_format", ScalarType::String.nullable(true))
         .finish(),
     is_retained_metrics_object: true,
     access: vec![PUBLIC_SELECT],
diff --git a/src/catalog/src/memory/objects.rs b/src/catalog/src/memory/objects.rs
index 443ea7776885f..3dadeba0a418b 100644
--- a/src/catalog/src/memory/objects.rs
+++ b/src/catalog/src/memory/objects.rs
@@ -706,6 +706,79 @@ pub enum DataSourceDesc {
     },
 }
 
+impl DataSourceDesc {
+    /// The key and value formats of the data source.
+    pub fn formats(&self) -> (Option<&str>, Option<&str>) {
+        match &self {
+            DataSourceDesc::Ingestion { ingestion_desc, .. } => {
+                match &ingestion_desc.desc.primary_export.encoding.as_ref() {
+                    Some(encoding) => match &encoding.key {
+                        Some(key) => (Some(key.type_()), Some(encoding.value.type_())),
+                        None => (None, Some(encoding.value.type_())),
+                    },
+                    None => (None, None),
+                }
+            }
+            DataSourceDesc::IngestionExport { data_config, .. } => match &data_config.encoding {
+                Some(encoding) => match &encoding.key {
+                    Some(key) => (Some(key.type_()), Some(encoding.value.type_())),
+                    None => (None, Some(encoding.value.type_())),
+                },
+                None => (None, None),
+            },
+            DataSourceDesc::Introspection(_)
+            | DataSourceDesc::Webhook { .. }
+            | DataSourceDesc::Progress => (None, None),
+        }
+    }
+
+    /// Envelope of the data source.
+    pub fn envelope(&self) -> Option<&str> {
+        // Note how "none"/"append-only" is different from `None`. Source
+        // sources don't have an envelope (internal logs, for example), while
+        // other sources have an envelope that we call the "NONE"-envelope.
+
+        fn envelope_string(envelope: &SourceEnvelope) -> &str {
+            match envelope {
+                SourceEnvelope::None(_) => "none",
+                SourceEnvelope::Upsert(upsert_envelope) => match upsert_envelope.style {
+                    mz_storage_types::sources::envelope::UpsertStyle::Default(_) => "upsert",
+                    mz_storage_types::sources::envelope::UpsertStyle::Debezium { .. } => {
+                        // NOTE(aljoscha): Should we somehow mark that this is
+                        // using upsert internally? See note above about
+                        // DEBEZIUM.
+                        "debezium"
+                    }
+                    mz_storage_types::sources::envelope::UpsertStyle::ValueErrInline { .. } => {
+                        "upsert-value-err-inline"
+                    }
+                },
+                SourceEnvelope::CdcV2 => {
+                    // TODO(aljoscha): Should we even report this? It's
+                    // currently not exposed.
+                    "materialize"
+                }
+            }
+        }
+
+        match self {
+            // NOTE(aljoscha): We could move the block for ingestions into
+            // `SourceEnvelope` itself, but that one feels more like an internal
+            // thing and adapter should own how we represent envelopes as a
+            // string? It would not be hard to convince me otherwise, though.
+            DataSourceDesc::Ingestion { ingestion_desc, .. } => Some(envelope_string(
+                &ingestion_desc.desc.primary_export.envelope,
+            )),
+            DataSourceDesc::IngestionExport { data_config, .. } => {
+                Some(envelope_string(&data_config.envelope))
+            }
+            DataSourceDesc::Introspection(_)
+            | DataSourceDesc::Webhook { .. }
+            | DataSourceDesc::Progress => None,
+        }
+    }
+}
+
 #[derive(Debug, Clone, Serialize)]
 pub struct Source {
     pub create_sql: Option<String>,
@@ -809,77 +882,6 @@ impl Source {
         }
     }
 
-    /// The key and value formats of the source.
-    pub fn formats(&self) -> (Option<&str>, Option<&str>) {
-        match &self.data_source {
-            DataSourceDesc::Ingestion { ingestion_desc, .. } => {
-                match &ingestion_desc.desc.primary_export.encoding.as_ref() {
-                    Some(encoding) => match &encoding.key {
-                        Some(key) => (Some(key.type_()), Some(encoding.value.type_())),
-                        None => (None, Some(encoding.value.type_())),
-                    },
-                    None => (None, None),
-                }
-            }
-            DataSourceDesc::IngestionExport { data_config, .. } => match &data_config.encoding {
-                Some(encoding) => match &encoding.key {
-                    Some(key) => (Some(key.type_()), Some(encoding.value.type_())),
-                    None => (None, Some(encoding.value.type_())),
-                },
-                None => (None, None),
-            },
-            DataSourceDesc::Introspection(_)
-            | DataSourceDesc::Webhook { .. }
-            | DataSourceDesc::Progress => (None, None),
-        }
-    }
-
-    /// Envelope of the source.
-    pub fn envelope(&self) -> Option<&str> {
-        // Note how "none"/"append-only" is different from `None`. Source
-        // sources don't have an envelope (internal logs, for example), while
-        // other sources have an envelope that we call the "NONE"-envelope.
-
-        fn envelope_string(envelope: &SourceEnvelope) -> &str {
-            match envelope {
-                SourceEnvelope::None(_) => "none",
-                SourceEnvelope::Upsert(upsert_envelope) => match upsert_envelope.style {
-                    mz_storage_types::sources::envelope::UpsertStyle::Default(_) => "upsert",
-                    mz_storage_types::sources::envelope::UpsertStyle::Debezium { .. } => {
-                        // NOTE(aljoscha): Should we somehow mark that this is
-                        // using upsert internally? See note above about
-                        // DEBEZIUM.
-                        "debezium"
-                    }
-                    mz_storage_types::sources::envelope::UpsertStyle::ValueErrInline { .. } => {
-                        "upsert-value-err-inline"
-                    }
-                },
-                SourceEnvelope::CdcV2 => {
-                    // TODO(aljoscha): Should we even report this? It's
-                    // currently not exposed.
-                    "materialize"
-                }
-            }
-        }
-
-        match &self.data_source {
-            // NOTE(aljoscha): We could move the block for ingestions into
-            // `SourceEnvelope` itself, but that one feels more like an internal
-            // thing and adapter should own how we represent envelopes as a
-            // string? It would not be hard to convince me otherwise, though.
-            DataSourceDesc::Ingestion { ingestion_desc, .. } => Some(envelope_string(
-                &ingestion_desc.desc.primary_export.envelope,
-            )),
-            DataSourceDesc::IngestionExport { data_config, .. } => {
-                Some(envelope_string(&data_config.envelope))
-            }
-            DataSourceDesc::Introspection(_)
-            | DataSourceDesc::Webhook { .. }
-            | DataSourceDesc::Progress => None,
-        }
-    }
-
     /// Connection ID of the source, if one exists.
     pub fn connection_id(&self) -> Option<GlobalId> {
         match &self.data_source {
diff --git a/test/sqllogictest/autogenerated/mz_internal.slt b/test/sqllogictest/autogenerated/mz_internal.slt
index b56cff6162666..8f2f2d4207b3b 100644
--- a/test/sqllogictest/autogenerated/mz_internal.slt
+++ b/test/sqllogictest/autogenerated/mz_internal.slt
@@ -346,6 +346,9 @@ SELECT position, name, type FROM objects WHERE schema = 'mz_internal' AND object
 ----
 1  id  text
 2  topic  text
+3  envelope_type  text
+4  key_format  text
+5  value_format  text
 
 query ITT
 SELECT position, name, type FROM objects WHERE schema = 'mz_internal' AND object = 'mz_sessions' ORDER BY position