Examples: *"why is my materialized view stale?"*, *"what can I optimize to save costs?"*, *"is my source healthy?"* |
| `mz-debug-freshness` | Use for diagnosing why an object is behind wall-clock time, from the lag ranking down to the operator and the SQL responsible; pairs with the `materialize-developer` server. Covers reading `mz_wallclock_global_lag` as a distribution, the status sweep that catches a stalled source whose frontier still looks current, lag history, hop-by-hop attribution through `mz_materialization_lag`, hydration and replica selection before `EXPLAIN ANALYZE`, per-operator cost and worker skew, and mapping an operator back to the clause that produced it. Diagnosis only, not remediation.
Examples: *"which object is causing my freshness alert?"*, *"is this delay ingestion or compute?"*, *"why is this dataflow falling behind its input?"* |
| `mz-optimize-memory` | Use for reducing the memory footprint and cost of a compute cluster; works over SQL or the `materialize-developer` MCP server. Covers finding which objects hold the memory, choosing and sizing the fix (indexes to add or drop, `GROUP SIZE` hints, slimmer views, subquery rewrites), verifying each change by measurement on an experiment cluster, and sizing the replica afterwards.
Examples: *"why is this cluster using so much memory?"*, *"can I downsize this replica?"*, *"which of these indexes should I drop?"* |
-| `materialize-docs` | Use for authoring view definitions, learning concepts, and looking up patterns; useful with either MCP server. Covers comprehensive Materialize documentation, including SQL syntax, idiomatic patterns, data ingestion, concepts, and best practices (400+ reference files).
Examples: *"show me how to deduplicate a stream"*, *"what's the idiomatic top-K pattern?"*, *"how do I create a Kafka source?"* |
+| `mz-docs` | Use for authoring view definitions, learning concepts, and looking up patterns; useful with either MCP server. Covers comprehensive Materialize documentation, including SQL syntax, idiomatic patterns, data ingestion, concepts, and best practices (400+ reference files).
Examples: *"show me how to deduplicate a stream"*, *"what's the idiomatic top-K pattern?"*, *"how do I create a Kafka source?"* |
| `mz-dbt` | Use for managing Materialize pipelines with dbt. Covers dbt-materialize adapter usage: materializations, profile configuration, index creation, blue/green deployments, and testing.
Examples: *"write a dbt model for a materialized view"*, *"how do I do a blue/green deployment with dbt?"* |
| `mz-terraform-provider` | Use for managing Materialize resources declaratively with Terraform. Covers provider configuration for Cloud and self-managed, navigation into the provider's auto-generated resource reference, cross-resource patterns, import workflows, and gotchas.
Examples: *"create a Kafka source with Terraform"*, *"import my existing clusters into Terraform state"*, *"set up RBAC grants in Terraform"* |
| `mz-terraform-self-managed` | Use for deploying or operating self-managed Materialize infrastructure with Terraform. Covers module layout and variables for deploying on AWS, Azure, and GCP: networking, Kubernetes, backend URL formats, instance sizing, upgrades, and gotchas.
Examples: *"deploy Materialize on EKS"*, *"what instance types should Materialize nodes use?"*, *"upgrade my self-managed Materialize"* |
diff --git a/doc/user/layouts/index.skill.md b/doc/user/layouts/index.skill.md
index 097e604180c05..1b14ee05b16e8 100644
--- a/doc/user/layouts/index.skill.md
+++ b/doc/user/layouts/index.skill.md
@@ -1,6 +1,6 @@
{{- /* Home page template - generates SKILL.md for Claude */ -}}
---
-name: materialize-docs
+name: mz-docs
description: Materialize documentation for SQL syntax, data ingestion, concepts, and best practices. Use when users ask about Materialize queries, sources, sinks, views, or clusters.
---
diff --git a/misc/python/materialize/scratch.py b/misc/python/materialize/scratch.py
index 3e824165e1083..94b23caa5dbef 100644
--- a/misc/python/materialize/scratch.py
+++ b/misc/python/materialize/scratch.py
@@ -274,7 +274,7 @@ def msftp(
def setup_ai_tools(instance: Instance) -> None:
"""Transfer local Claude Code and Codex configs to a scratch instance
- and install the materialize-docs skill."""
+ and install the mz-docs skill."""
import io
import pathlib
import tarfile
@@ -323,7 +323,7 @@ def setup_ai_tools(instance: Instance) -> None:
instance,
"cd materialize && "
"NPM_CONFIG_UPDATE_NOTIFIER=false "
- "npx -q -y skills add MaterializeInc/agent-skills -g -a claude-code --copy -y --skill materialize-docs 2>/dev/null || true",
+ "npx -q -y skills add MaterializeInc/agent-skills -g -a claude-code --copy -y --skill mz-docs 2>/dev/null || true",
quiet=True,
)
@@ -331,7 +331,7 @@ def setup_ai_tools(instance: Instance) -> None:
instance,
"cd materialize && "
"NPM_CONFIG_UPDATE_NOTIFIER=false "
- "npx -q -y skills add MaterializeInc/agent-skills -g -a codex --copy -y --skill materialize-docs 2>/dev/null || true",
+ "npx -q -y skills add MaterializeInc/agent-skills -g -a codex --copy -y --skill mz-docs 2>/dev/null || true",
quiet=True,
)
diff --git a/misc/wasm/Cargo.lock b/misc/wasm/Cargo.lock
index 9d7018742b32f..d1d58bea8c7e9 100644
--- a/misc/wasm/Cargo.lock
+++ b/misc/wasm/Cargo.lock
@@ -134,6 +134,12 @@ version = "0.22.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
+[[package]]
+name = "base64"
+version = "0.23.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5"
+
[[package]]
name = "base64ct"
version = "1.8.3"
@@ -186,6 +192,17 @@ version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"
+[[package]]
+name = "chacha20"
+version = "0.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06"
+dependencies = [
+ "cfg-if",
+ "cpufeatures",
+ "rand_core 0.10.1",
+]
+
[[package]]
name = "core-foundation"
version = "0.10.1"
@@ -202,6 +219,15 @@ version = "0.8.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
+[[package]]
+name = "cpufeatures"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5ca28b0ae3115b884660db4118d803791fd6756b6e88f39c0f3f7859060d7566"
+dependencies = [
+ "libc",
+]
+
[[package]]
name = "crc32fast"
version = "1.5.2"
@@ -447,10 +473,22 @@ checksum = "26145e563e54f2cadc477553f1ec5ee650b00862f0a58bcd12cbdc5f0ea2d2f4"
dependencies = [
"cfg-if",
"libc",
- "r-efi",
+ "r-efi 5.2.0",
"wasi 0.14.2+wasi-0.2.4",
]
+[[package]]
+name = "getrandom"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "r-efi 6.0.0",
+ "rand_core 0.10.1",
+]
+
[[package]]
name = "gimli"
version = "0.31.1"
@@ -459,9 +497,9 @@ checksum = "07e28edb80900c19c28f1072f2e8aeca7fa06b23cd4169cefe1af5aa3260783f"
[[package]]
name = "h2"
-version = "0.4.10"
+version = "0.4.19"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a9421a676d1b147b16b82c9225157dc629087ef8ec4d5e2960f9437a90dac0a5"
+checksum = "ef8e5e5a340588f4452631496976cf8636d4a7ecf600239fdc27615d2530bc16"
dependencies = [
"atomic-waker",
"bytes",
@@ -546,9 +584,9 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
[[package]]
name = "hyper"
-version = "1.8.1"
+version = "1.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2ab2d4f250c3d7b1c9fcdff1cece94ea4e2dfbec68614f7b87cb205f24ca9d11"
+checksum = "27b501faa50e7a26c3d3560ca625132f4078a17771f4810baf70475ae48cbe43"
dependencies = [
"atomic-waker",
"bytes",
@@ -561,7 +599,6 @@ dependencies = [
"httpdate",
"itoa",
"pin-project-lite",
- "pin-utils",
"smallvec",
"tokio",
"want",
@@ -598,16 +635,17 @@ dependencies = [
[[package]]
name = "hyper-util"
-version = "0.1.20"
+version = "0.1.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0"
+checksum = "ddc03d96684f9226b8a787cdb71488417b53ab5ea8fdb1dac946cb9431cc8bff"
dependencies = [
- "base64",
+ "base64 0.23.1",
"bytes",
"futures-channel",
"futures-util",
"http",
"http-body",
+ "httparse",
"hyper",
"ipnet",
"libc",
@@ -946,7 +984,7 @@ dependencies = [
"sentry-tracing",
"serde",
"stacker",
- "thiserror 2.0.20",
+ "thiserror 2.0.21",
"tokio-native-tls",
"tonic",
"tracing",
@@ -1007,7 +1045,7 @@ dependencies = [
"mz-walkabout",
"serde",
"smallvec",
- "thiserror 2.0.20",
+ "thiserror 2.0.21",
"tracing",
]
@@ -1028,7 +1066,7 @@ dependencies = [
"itertools",
"mz-sql-parser",
"pretty",
- "thiserror 2.0.20",
+ "thiserror 2.0.21",
]
[[package]]
@@ -1225,23 +1263,23 @@ dependencies = [
[[package]]
name = "opentelemetry"
-version = "0.32.0"
+version = "0.33.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682"
+checksum = "6cdb0b1b267eb9db3331b434ed9ddab10d50e280a9adf9d13e5233e2002b61b5"
dependencies = [
"futures-core",
"futures-sink",
"js-sys",
"pin-project-lite",
- "thiserror 2.0.20",
+ "thiserror 2.0.21",
"tracing",
]
[[package]]
name = "opentelemetry-http"
-version = "0.32.0"
+version = "0.33.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5683015d09e2df236ef005b17f6f196f0d5f6313c4fa43a7b6a53b52776e4331"
+checksum = "ee2c3625b8aa04209f7e01e513bc8044da9687fa38a9eacf59628ca5f3b87300"
dependencies = [
"async-trait",
"bytes",
@@ -1252,18 +1290,19 @@ dependencies = [
[[package]]
name = "opentelemetry-otlp"
-version = "0.32.0"
+version = "0.33.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35"
+checksum = "699a67345e21962a231955b9059a157e428b219fa5df8efe11f08346fb23a344"
dependencies = [
"http",
+ "httpdate",
"opentelemetry",
"opentelemetry-http",
"opentelemetry-proto",
"opentelemetry_sdk",
"prost",
"reqwest",
- "thiserror 2.0.20",
+ "thiserror 2.0.21",
"tokio",
"tonic",
"tonic-types",
@@ -1271,9 +1310,9 @@ dependencies = [
[[package]]
name = "opentelemetry-proto"
-version = "0.32.0"
+version = "0.33.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "56d658ba1faf63f7b9c492cfbe6e0ec365440a16132d3270c1065f7b33f1b638"
+checksum = "25da1ac11a0aeccf38d7f77ee0348715adaf8340f65ad46c94a02c6b20e2f65d"
dependencies = [
"opentelemetry",
"opentelemetry_sdk",
@@ -1284,9 +1323,9 @@ dependencies = [
[[package]]
name = "opentelemetry_sdk"
-version = "0.32.1"
+version = "0.33.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9b59f80e1ac4d5ff7a2db8fb6c80badb7f0f3f858211fba08dd9aaec750894f9"
+checksum = "cb39533d9d1c912123efd7d41d7e0c29d16917b60ce15b4c8d87cb1af7f67520"
dependencies = [
"futures-channel",
"futures-executor",
@@ -1294,8 +1333,8 @@ dependencies = [
"opentelemetry",
"percent-encoding",
"portable-atomic",
- "rand",
- "thiserror 2.0.20",
+ "rand 0.9.5",
+ "thiserror 2.0.21",
"tokio",
"tokio-stream",
]
@@ -1436,12 +1475,6 @@ version = "0.2.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8afb450f006bf6385ca15ef45d71d2288452bc3683ce2e2cacc0d18e4be60b58"
-[[package]]
-name = "pin-utils"
-version = "0.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
-
[[package]]
name = "pkg-config"
version = "0.3.32"
@@ -1501,7 +1534,7 @@ dependencies = [
"memchr",
"parking_lot",
"protobuf",
- "thiserror 2.0.20",
+ "thiserror 2.0.21",
]
[[package]]
@@ -1581,6 +1614,12 @@ version = "5.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "74765f6d916ee2faa39bc8e68e4f3ed8949b48cccdac59983d287a7cb71ce9c5"
+[[package]]
+name = "r-efi"
+version = "6.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
+
[[package]]
name = "rand"
version = "0.9.5"
@@ -1588,7 +1627,18 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41"
dependencies = [
"rand_chacha",
- "rand_core",
+ "rand_core 0.9.3",
+]
+
+[[package]]
+name = "rand"
+version = "0.10.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "65c9fb96cbc91e3478eaae79a69fcd3f1ae4ad052e471fe6732fff548984b4af"
+dependencies = [
+ "chacha20",
+ "getrandom 0.4.3",
+ "rand_core 0.10.1",
]
[[package]]
@@ -1598,7 +1648,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb"
dependencies = [
"ppv-lite86",
- "rand_core",
+ "rand_core 0.9.3",
]
[[package]]
@@ -1607,9 +1657,15 @@ version = "0.9.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "99d9a13982dcf210057a8a78572b2217b667c3beacbf3a0d8b454f6f82837d38"
dependencies = [
- "getrandom",
+ "getrandom 0.3.3",
]
+[[package]]
+name = "rand_core"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69"
+
[[package]]
name = "redox_syscall"
version = "0.5.12"
@@ -1654,7 +1710,7 @@ version = "0.13.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "62e0021ea2c22aed41653bc7e1419abb2c97e038ff2c33d0e1309e49a97deec0"
dependencies = [
- "base64",
+ "base64 0.22.1",
"bytes",
"futures-channel",
"futures-core",
@@ -1780,9 +1836,9 @@ checksum = "56e6fa9c48d24d85fb3de5ad847117517440f6beceb7798af16b4a87d616b8d0"
[[package]]
name = "sentry"
-version = "0.49.2"
+version = "0.49.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "76a88b65feb368d9dde5d531a2b59b25016e36fd6e4978432371a3ee77746ca2"
+checksum = "9fa951f4e644464ebfd2347a7ab03b0828c512d2448a76262cad2607b343dff4"
dependencies = [
"cfg_aliases",
"httpdate",
@@ -1799,9 +1855,9 @@ dependencies = [
[[package]]
name = "sentry-backtrace"
-version = "0.49.2"
+version = "0.49.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2c51511d67ed670266a93afeaa26a08019b04ad1420cb9191da30f2338d22c48"
+checksum = "f4f27cf048aa63ccd5bc94c29989133eabb82e9c6bf7441b58e9d4a95837a105"
dependencies = [
"backtrace",
"regex",
@@ -1810,9 +1866,9 @@ dependencies = [
[[package]]
name = "sentry-contexts"
-version = "0.49.2"
+version = "0.49.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4b757ac1f335856e8d7f5062a1fd9bc07a05a7eb3cc48247db79bf2618a490bd"
+checksum = "c5ed680585dbdacdf9859c2b4dc53358ff02bd256766882e22e343836db36b22"
dependencies = [
"hostname",
"libc",
@@ -1824,11 +1880,11 @@ dependencies = [
[[package]]
name = "sentry-core"
-version = "0.49.2"
+version = "0.49.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d889520a375e5b93efb0a66def1630d21d168b0251eb55b286b03fa940ccfc84"
+checksum = "c3ed398494bb8fc412a11a182c1883a10ae147587b8f6314abd222d1d626f31b"
dependencies = [
- "rand",
+ "rand 0.10.3",
"sentry-types",
"serde",
"serde_json",
@@ -1837,9 +1893,9 @@ dependencies = [
[[package]]
name = "sentry-debug-images"
-version = "0.49.2"
+version = "0.49.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "656487fd5f7b7c0105b2e82171982aac64489e0cb679436038febf070f2f76d6"
+checksum = "55854bb40681e84ffd2a69578ad4ef9bfaf5bf2bf1380a745578b591fa24af22"
dependencies = [
"findshlibs",
"sentry-core",
@@ -1847,9 +1903,9 @@ dependencies = [
[[package]]
name = "sentry-tracing"
-version = "0.49.2"
+version = "0.49.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "77796d14eedbef22c2fe78e9c69fb09f14c7ad607e624d48dce92eedc17a136e"
+checksum = "e7e7db1aefdf623f4e48e21ec9e4af535024187aeafb2a4cc7e29d035def5a13"
dependencies = [
"bitflags",
"sentry-backtrace",
@@ -1860,16 +1916,16 @@ dependencies = [
[[package]]
name = "sentry-types"
-version = "0.49.2"
+version = "0.49.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fc71f5ca55942d9b2901af95d5df5c5d65c164134c22a83682cac3d0b6c7ef2d"
+checksum = "1c18bf7b159bdfc7c8f97816f476b4947a1937fb6bea9d38d3479e58631614ac"
dependencies = [
"debugid",
"hex",
- "rand",
+ "rand 0.10.3",
"serde",
"serde_json",
- "thiserror 2.0.20",
+ "thiserror 2.0.21",
"time",
"url",
"uuid",
@@ -1954,9 +2010,9 @@ dependencies = [
[[package]]
name = "smallvec"
-version = "1.16.1"
+version = "1.16.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ba467056f1b547ed52077911161fc86985becbc60e8e1857c8a144dab0def891"
+checksum = "f9395f0f0eee849a9b707b2f06bb92a6a422090e2123bb2ef8e87a0e61892a8e"
[[package]]
name = "socket2"
@@ -2082,11 +2138,11 @@ dependencies = [
[[package]]
name = "thiserror"
-version = "2.0.20"
+version = "2.0.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f"
+checksum = "09e52cb86a36cede5cb101bf8908837b3e4c6e5e59fe7fd85c23fb56200d189e"
dependencies = [
- "thiserror-impl 2.0.20",
+ "thiserror-impl 2.0.21",
]
[[package]]
@@ -2102,9 +2158,9 @@ dependencies = [
[[package]]
name = "thiserror-impl"
-version = "2.0.20"
+version = "2.0.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af"
+checksum = "fe5197923287db20a58125f0bc85c062f7f2c892de97b18c356f9efb14b28524"
dependencies = [
"proc-macro2",
"quote",
@@ -2230,7 +2286,7 @@ checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef"
dependencies = [
"async-trait",
"axum",
- "base64",
+ "base64 0.22.1",
"bytes",
"flate2",
"h2",
@@ -2368,9 +2424,9 @@ dependencies = [
[[package]]
name = "tracing-opentelemetry"
-version = "0.33.0"
+version = "0.34.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26"
+checksum = "0a904802a1b902f43638b677ff2a650847e3b4404101b6c586d648e8c1e3e8fe"
dependencies = [
"js-sys",
"opentelemetry",
@@ -2461,7 +2517,7 @@ version = "3.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fdc97a28575b85cfedf2a7e7d3cc64b3e11bd8ac766666318003abbacc7a21fc"
dependencies = [
- "base64",
+ "base64 0.22.1",
"der",
"log",
"native-tls",
@@ -2478,7 +2534,7 @@ version = "0.5.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d81f9efa9df032be5934a46a068815a10a042b494b6a58cb0a1a97bb5467ed6f"
dependencies = [
- "base64",
+ "base64 0.22.1",
"http",
"httparse",
"log",
diff --git a/src/adapter/src/catalog/apply.rs b/src/adapter/src/catalog/apply.rs
index a7c3721f3b7e8..ed0c6e5e06aea 100644
--- a/src/adapter/src/catalog/apply.rs
+++ b/src/adapter/src/catalog/apply.rs
@@ -1482,9 +1482,7 @@ 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)
- }
+ 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.
diff --git a/src/adapter/src/catalog/builtin_table_updates.rs b/src/adapter/src/catalog/builtin_table_updates.rs
index b6653b1305ad2..6b92124b586f9 100644
--- a/src/adapter/src/catalog/builtin_table_updates.rs
+++ b/src/adapter/src/catalog/builtin_table_updates.rs
@@ -19,10 +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_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_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::{
CatalogItem, DataSourceDesc, Func, Index, MaterializedView, Table, TableDataSource, Type,
@@ -853,52 +852,4 @@ impl CatalogState {
diff,
)
}
-
- pub fn pack_source_references_update(
- &self,
- source_references: &SourceReferences,
- diff: Diff,
- ) -> Vec> {
- 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()
- }
}
diff --git a/src/adapter/src/catalog/open/builtin_schema_migration.rs b/src/adapter/src/catalog/open/builtin_schema_migration.rs
index 48e295d21bf2a..707be62f46b59 100644
--- a/src/adapter/src/catalog/open/builtin_schema_migration.rs
+++ b/src/adapter/src/catalog/open/builtin_schema_migration.rs
@@ -487,6 +487,15 @@ static MIGRATIONS: LazyLock> = LazyLock::new(|| {
MZ_INTERNAL_SCHEMA,
"mz_object_global_ids",
),
+ // Converting mz_source_references from a builtin table to a
+ // materialized view over mz_catalog_raw changes its catalog
+ // fingerprint.
+ MigrationStep::replacement(
+ "26.45.0-dev.0",
+ CatalogItemType::MaterializedView,
+ MZ_INTERNAL_SCHEMA,
+ "mz_source_references",
+ ),
]
});
diff --git a/src/audit-log/src/lib.rs b/src/audit-log/src/lib.rs
index 0a0b202704a53..1d63b647128ea 100644
--- a/src/audit-log/src/lib.rs
+++ b/src/audit-log/src/lib.rs
@@ -25,7 +25,7 @@
//! `mz_catalog.mz_audit_events.details` exposes. The durable catalog stores
//! the proto twin from `mz_catalog_protos::objects::audit_log_event_v1`,
//! which `parse_catalog_audit_log_details` (in
-//! `src/expr/src/scalar/func/impls/jsonb.rs`) reshapes back into
+//! `src/catalog-decode/src/audit_log.rs`) reshapes back into
//! `as_json`'s output. Changes here that shift that output (new variants,
//! renamed fields, added `skip_serializing_if`, field-name diffs against
//! the proto) need matching updates there. The round-trip is covered by
diff --git a/src/catalog-decode/Cargo.toml b/src/catalog-decode/Cargo.toml
new file mode 100644
index 0000000000000..118c25f517b4e
--- /dev/null
+++ b/src/catalog-decode/Cargo.toml
@@ -0,0 +1,25 @@
+[package]
+name = "mz-catalog-decode"
+description = "Decoders for durable catalog data, backing the catalog-raw SQL functions."
+version = "0.0.0"
+edition.workspace = true
+rust-version.workspace = true
+publish = false
+
+[lints]
+workspace = true
+
+[dependencies]
+hex.workspace = true
+mz-ore = { path = "../ore", features = ["assert"] }
+mz-repr = { path = "../repr" }
+mz-sql-parser = { path = "../sql-parser" }
+prost.workspace = true
+serde_json.workspace = true
+tracing.workspace = true
+
+[dev-dependencies]
+mz-ore = { path = "../ore", features = ["test"] }
+
+[features]
+default = []
diff --git a/src/expr/src/scalar/func/impls/audit_log_details.rs b/src/catalog-decode/src/audit_log.rs
similarity index 95%
rename from src/expr/src/scalar/func/impls/audit_log_details.rs
rename to src/catalog-decode/src/audit_log.rs
index 1692460f730b0..54ae6c002c6e7 100644
--- a/src/expr/src/scalar/func/impls/audit_log_details.rs
+++ b/src/catalog-decode/src/audit_log.rs
@@ -26,12 +26,9 @@
//! variant and catches drift when either side gains a new variant, field,
//! or serde attribute.
-use mz_expr_derive::sqlfunc;
use mz_repr::adt::jsonb::{Jsonb, JsonbRef};
use mz_repr::{Datum, Row, RowPacker};
-use crate::EvalError;
-
/// Reshapes proto `audit_log_event_v1::Details` JSON from `mz_catalog_raw`
/// (e.g. `{"IdFullNameV1": {"id": "u1", "name": {...}}}`) into the shape
/// `mz_audit_log::EventDetails::as_json` produces: the format the prior
@@ -40,8 +37,7 @@ use crate::EvalError;
/// See the module docstring for the mechanism and the reciprocal side. Rule
/// changes must be paired with tests; the property test at
/// `src/catalog/tests/audit_log_details.rs` is the safety net.
-#[sqlfunc]
-fn parse_catalog_audit_log_details<'a>(a: JsonbRef<'a>) -> Result {
+pub fn details(a: JsonbRef<'_>) -> Result {
/// `(variant, path, field, sub_variant)`: `#[serde(flatten)]` sites.
/// `sub_variant` sets the context for rules on the hoisted content
/// when the flattened struct itself has a `#[serde(flatten)]`.
@@ -472,43 +468,39 @@ fn parse_catalog_audit_log_details<'a>(a: JsonbRef<'a>) -> Result Result {
- let Datum::Map(dict) = a.into_datum() else {
- return Err("expected object".into());
- };
- let mut iter = dict.iter();
- let (variant, inner) = iter
- .next()
- .ok_or_else(|| "empty details enum".to_string())?;
- if iter.next().is_some() {
- return Err("details enum had multiple keys".into());
- }
- let mut row = Row::default();
- // `ResetAllV1` is the only variant `as_json` maps to null (the
- // proto `Empty` payload serializes to `{}`).
- if variant == "ResetAllV1" {
- row.packer().push(Datum::JsonNull);
- return Ok(Jsonb::from_row(row));
- }
- let Datum::Map(_) = inner else {
- return Err(format!("expected inner object for variant {variant}"));
- };
- rewrite(inner, variant, "", &mut row.packer())?;
- Ok(Jsonb::from_row(row))
+ let Datum::Map(dict) = a.into_datum() else {
+ return Err("expected object".into());
};
-
- parse().map_err(|e| EvalError::InvalidCatalogJson(e.into()))
+ let mut iter = dict.iter();
+ let (variant, inner) = iter
+ .next()
+ .ok_or_else(|| "empty details enum".to_string())?;
+ if iter.next().is_some() {
+ return Err("details enum had multiple keys".into());
+ }
+ let mut row = Row::default();
+ // `ResetAllV1` is the only variant `as_json` maps to null (the
+ // proto `Empty` payload serializes to `{}`).
+ if variant == "ResetAllV1" {
+ row.packer().push(Datum::JsonNull);
+ return Ok(Jsonb::from_row(row));
+ }
+ let Datum::Map(_) = inner else {
+ return Err(format!("expected inner object for variant {variant}"));
+ };
+ rewrite(inner, variant, "", &mut row.packer())?;
+ Ok(Jsonb::from_row(row))
}
#[cfg(test)]
mod tests {
use super::*;
- /// Round-trip a JSON `input` through `parse_catalog_audit_log_details` and
+ /// Round-trip a JSON `input` through `details` and
/// assert the resulting JSON parses equal to `expected`.
fn check(input: &str, expected: &str) {
let input: Jsonb = input.parse().expect("valid input JSONB");
- let actual = parse_catalog_audit_log_details(input.as_ref())
+ let actual = details(input.as_ref())
.expect("helper succeeded")
.to_string();
let actual_value: serde_json::Value =
@@ -518,11 +510,11 @@ mod tests {
assert_eq!(actual_value, expected_value);
}
- /// Run `parse_catalog_audit_log_details` on `input` and assert it returns
- /// a `InvalidCatalogJson` error containing `expected_substr`.
+ /// Run `details` on `input` and assert it returns an error containing
+ /// `expected_substr`.
fn check_err(input: &str, expected_substr: &str) {
let input: Jsonb = input.parse().expect("valid input JSONB");
- let err = parse_catalog_audit_log_details(input.as_ref()).expect_err("helper should error");
+ let err = details(input.as_ref()).expect_err("helper should error");
let msg = format!("{err:?}");
assert!(
msg.contains(expected_substr),
diff --git a/src/catalog-decode/src/create_sql.rs b/src/catalog-decode/src/create_sql.rs
new file mode 100644
index 0000000000000..e79c46cc20a64
--- /dev/null
+++ b/src/catalog-decode/src/create_sql.rs
@@ -0,0 +1,1960 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+//! Decoders for catalog `create_sql` strings.
+//!
+//! Each function parses a persisted `create_sql` and returns a JSON value with
+//! the fields a builtin catalog view projects. Errors are human-readable
+//! strings.
+
+use std::collections::BTreeMap;
+
+use mz_sql_parser::ast::display::AstDisplay;
+use mz_sql_parser::ast::item_refs::collect_item_references;
+use mz_sql_parser::ast::{
+ AstInfo, AvroSchema, ConnectionOption, ConnectionOptionName, CreateConnectionType,
+ CreateSinkConnection, CreateSubsourceOptionName, Format, FormatSpecifier,
+ IcebergSinkConfigOptionName, IcebergSinkMode, KafkaSinkConfigOptionName,
+ KafkaSourceConfigOptionName, PgConfigOptionName, ProtobufSchema, Raw, RawClusterName,
+ RawItemName, SinkEnvelope, SourceEnvelope, SourceErrorPolicy, Statement, UnresolvedItemName,
+ Value, WithOptionValue,
+};
+use prost::Message as _;
+use serde_json::json;
+
+/// Parses `sql`, which must contain exactly one statement.
+fn parse_single_statement(sql: &str) -> Result, String> {
+ let mut stmts = mz_sql_parser::parser::parse_statements(sql)
+ .map_err(|e| format!("failed to parse create_sql: {e}"))?;
+ match stmts.len() {
+ 1 => Ok(stmts.remove(0).ast),
+ n => Err(format!("expected a single statement, found {n}")),
+ }
+}
+
+fn item_id(item: RawItemName) -> Result {
+ match item {
+ RawItemName::Id(id, _, _) => Ok(id),
+ RawItemName::Name(_) => Err("unresolved item name"),
+ }
+}
+
+fn cluster_id(in_cluster: RawClusterName) -> Result {
+ match in_cluster {
+ RawClusterName::Resolved(s) => Ok(s),
+ RawClusterName::Unresolved(_) => Err("unresolved cluster name"),
+ }
+}
+
+fn format_name(fmt: &Format) -> &'static str {
+ match fmt {
+ Format::Bytes => "bytes",
+ Format::Avro(_) => "avro",
+ Format::Protobuf(_) => "protobuf",
+ Format::Regex(_) => "regex",
+ Format::Csv { .. } => "csv",
+ Format::Json { .. } => "json",
+ Format::Text => "text",
+ }
+}
+
+/// Extracts the string form of a `WithOptionValue`, matching how the planner
+/// coerces option values. The blanket `TryFromValue>` impl in
+/// `src/sql/src/plan/with_options.rs` accepts a quoted string, a bare identifier,
+/// and a 1-part unresolved item name, all yielding a string. Any other variant
+/// yields None.
+///
+/// `AstDisplay` does not re-quote bare identifiers or 1-part names, so those
+/// forms persist unquoted in `create_sql`. Matching only `Value::String` here
+/// would miss them and silently fall back to a default, so the catalog-raw
+/// parsers below route their string options through this helper.
+fn option_string(value: &WithOptionValue) -> Option {
+ match value {
+ WithOptionValue::Value(Value::String(s)) => Some(s.clone()),
+ WithOptionValue::Ident(ident) => Some(ident.clone().into_string()),
+ WithOptionValue::UnresolvedItemName(UnresolvedItemName(parts)) if parts.len() == 1 => {
+ Some(parts[0].clone().into_string())
+ }
+ _ => None,
+ }
+}
+
+/// Parses a catalog `create_sql` string into a JSON object.
+///
+/// The returned JSON does not fully reflect the parsed SQL and instead contains only fields
+/// required by current callers.
+pub fn item_details(a: &str) -> Result {
+ let stmt = parse_single_statement(a)?;
+
+ let mut info = BTreeMap::<&str, serde_json::Value>::new();
+
+ use mz_sql_parser::ast::Statement::*;
+ let item_type = match stmt {
+ CreateSecret(_) => "secret",
+ CreateConnection(stmt) => {
+ let connection_type = stmt.connection_type.as_str();
+ info.insert("connection_type", json!(connection_type));
+
+ "connection"
+ }
+ CreateView(stmt) => {
+ let mut definition = stmt.definition.query.to_ast_string_stable();
+ // PostgreSQL appends a semicolon in `pg_views.definition`, we
+ // do the same for compatibility's sake.
+ definition.push(';');
+ info.insert("definition", json!(definition));
+
+ "view"
+ }
+ CreateMaterializedView(stmt) => {
+ let Some(in_cluster) = stmt.in_cluster else {
+ return Err("missing IN CLUSTER".into());
+ };
+ let cluster_id = match in_cluster {
+ RawClusterName::Unresolved(ident) => ident.into_string(),
+ RawClusterName::Resolved(s) => s,
+ };
+ info.insert("cluster_id", json!(cluster_id));
+
+ let mut definition = stmt.query.to_ast_string_stable();
+ definition.push(';');
+ info.insert("definition", json!(definition));
+
+ "materialized-view"
+ }
+ CreateTable(_) => "table",
+ CreateTableFromSource(stmt) => {
+ let source_id = item_id(stmt.source)?;
+ info.insert("source_id", json!(source_id));
+
+ "table"
+ }
+ CreateSource(stmt) => {
+ let Some(in_cluster) = stmt.in_cluster else {
+ return Err("missing IN CLUSTER".into());
+ };
+ let cluster_id = cluster_id(in_cluster)?;
+ info.insert("cluster_id", json!(cluster_id));
+
+ use mz_sql_parser::ast::CreateSourceConnection::*;
+ let (source_type, connection) = match stmt.connection {
+ Kafka { connection, .. } => ("kafka", Some(connection)),
+ Postgres { connection, .. } => ("postgres", Some(connection)),
+ MySql { connection, .. } => ("mysql", Some(connection)),
+ SqlServer { connection, .. } => ("sql-server", Some(connection)),
+ LoadGenerator { .. } => ("load-generator", None),
+ };
+ info.insert("source_type", json!(source_type));
+ if let Some(conn) = connection {
+ let conn_id = item_id(conn)?;
+ info.insert("connection_id", json!(conn_id));
+ }
+
+ let is_debezium = matches!(
+ stmt.envelope,
+ Some(mz_sql_parser::ast::SourceEnvelope::Debezium)
+ );
+
+ // An old-syntax kafka source ingests into its own relation, so an
+ // omitted ENVELOPE means the default ENVELOPE NONE and the pre-MV
+ // packer reported 'none'. A new-syntax source (no progress
+ // subsource, hence no EXPOSE PROGRESS AS in create_sql) ingests
+ // nothing itself. Its envelopes live on the per-table exports, so
+ // its own envelope_type stays absent (SQL NULL), matching released
+ // behavior. `progress_subsource.is_some()` is the planner's own
+ // old-vs-new discriminator (see `OldSyntaxIngestion` in
+ // plan_create_source). Non-kafka sources carry no envelope either.
+ // See the `mz_sources.envelope_type` column.
+ let envelope_type = match &stmt.envelope {
+ Some(envelope) => {
+ use mz_sql_parser::ast::SourceEnvelope::*;
+ Some(match envelope {
+ None => "none",
+ Debezium => "debezium",
+ Upsert { .. } => "upsert",
+ CdcV2 => "materialize",
+ })
+ }
+ None if source_type == "kafka" && stmt.progress_subsource.is_some() => Some("none"),
+ None => None,
+ };
+ if let Some(envelope_type) = envelope_type {
+ info.insert("envelope_type", json!(envelope_type));
+ }
+
+ if let Some(format_spec) = stmt.format {
+ match &format_spec {
+ FormatSpecifier::Bare(fmt) => {
+ // Debezium sources with a single format spec implicitly use
+ // the same format for both key and value.
+ if is_debezium {
+ info.insert("key_format", json!(format_name(fmt)));
+ }
+ info.insert("value_format", json!(format_name(fmt)));
+ }
+ FormatSpecifier::KeyValue { key, value } => {
+ info.insert("key_format", json!(format_name(key)));
+ info.insert("value_format", json!(format_name(value)));
+ }
+ }
+ }
+
+ "source"
+ }
+ CreateWebhookSource(stmt) => {
+ if stmt.is_table {
+ "table"
+ } else {
+ info.insert("source_type", json!("webhook"));
+ if let Some(in_cluster) = stmt.in_cluster {
+ let cluster_id = cluster_id(in_cluster)?;
+ info.insert("cluster_id", json!(cluster_id));
+ }
+ "source"
+ }
+ }
+ CreateSubsource(stmt) => {
+ use mz_sql_parser::ast::CreateSubsourceOptionName;
+ let is_progress = stmt
+ .with_options
+ .iter()
+ .any(|o| matches!(o.name, CreateSubsourceOptionName::Progress));
+ let source_type = if is_progress { "progress" } else { "subsource" };
+ info.insert("source_type", json!(source_type));
+
+ if let Some(of_source) = stmt.of_source {
+ let of_source_id = item_id(of_source)?;
+ info.insert("of_source_id", json!(of_source_id));
+ }
+
+ "subsource"
+ }
+ // Everything the mz_sinks, mz_kafka_sinks and mz_iceberg_sinks
+ // views read. The Rust side of each value lives in `Sink` and
+ // `StorageSinkConnection`, so those and this have to move together.
+ //
+ // NOTE: we bail below if a sink has no resolved `IN CLUSTER`, no
+ // `TOPIC` on kafka, or no `NAMESPACE`/`TABLE` on iceberg. Planning
+ // guarantees all four. But if one ever slipped through it would
+ // take down every view built on this function, not just the sink
+ // ones.
+ CreateSink(stmt) => {
+ let Some(in_cluster) = stmt.in_cluster else {
+ return Err("missing IN CLUSTER".into());
+ };
+ info.insert("cluster_id", json!(cluster_id(in_cluster)?));
+
+ match stmt.connection {
+ CreateSinkConnection::Kafka {
+ connection,
+ options,
+ key: sink_key,
+ ..
+ } => {
+ info.insert("sink_type", json!("kafka"));
+ info.insert("connection_id", json!(item_id(connection)?));
+
+ let topic = options
+ .into_iter()
+ .find(|o| o.name == KafkaSinkConfigOptionName::Topic)
+ .and_then(|o| o.value.as_ref().and_then(option_string))
+ .ok_or("kafka sink missing TOPIC")?;
+ info.insert("topic", json!(topic));
+
+ if let Some(envelope) = stmt.envelope {
+ let envelope_type = match envelope {
+ SinkEnvelope::Upsert => "upsert",
+ SinkEnvelope::Debezium => "debezium",
+ };
+ info.insert("envelope_type", json!(envelope_type));
+ }
+
+ if let Some(format_spec) = stmt.format {
+ // A key format only survives if the sink has a
+ // `KEY`. Without one `kafka_sink_builder` throws
+ // away the key half of a key/value spec, and does
+ // not copy a bare spec over to the key either.
+ let (key_format, value_format) = match (&format_spec, sink_key.is_some()) {
+ (FormatSpecifier::Bare(fmt), false) => (None, format_name(fmt)),
+ (FormatSpecifier::Bare(fmt), true) => {
+ (Some(format_name(fmt)), format_name(fmt))
+ }
+ (FormatSpecifier::KeyValue { value, .. }, false) => {
+ (None, format_name(value))
+ }
+ (FormatSpecifier::KeyValue { key, value }, true) => {
+ (Some(format_name(key)), format_name(value))
+ }
+ };
+ if let Some(key_format) = key_format {
+ info.insert("key_format", json!(key_format));
+ }
+ info.insert("value_format", json!(value_format));
+
+ // The deprecated combined `format`. Only avro/avro
+ // and json/json collapse to a single name.
+ // Everything else, text/text and bytes/bytes
+ // included, gets the composite form.
+ let combined = match key_format {
+ None => value_format.to_string(),
+ Some(key_format)
+ if key_format == value_format
+ && matches!(value_format, "avro" | "json") =>
+ {
+ value_format.to_string()
+ }
+ Some(key_format) => {
+ format!("key-{key_format}-value-{value_format}")
+ }
+ };
+ info.insert("format", json!(combined));
+ }
+ }
+ CreateSinkConnection::Iceberg {
+ catalog_connection,
+ options,
+ ..
+ } => {
+ info.insert("sink_type", json!("iceberg"));
+ // The catalog connection, not the optional AWS one.
+ info.insert("connection_id", json!(item_id(catalog_connection)?));
+
+ let mut namespace = None;
+ let mut table = None;
+ for option in options {
+ match option.name {
+ IcebergSinkConfigOptionName::Namespace => {
+ namespace = option.value.as_ref().and_then(option_string)
+ }
+ IcebergSinkConfigOptionName::Table => {
+ table = option.value.as_ref().and_then(option_string)
+ }
+ }
+ }
+ info.insert(
+ "namespace",
+ json!(namespace.ok_or("iceberg sink missing NAMESPACE")?),
+ );
+ info.insert("table", json!(table.ok_or("iceberg sink missing TABLE")?));
+
+ // Iceberg spells the envelope `MODE`, and has no format
+ // columns at all.
+ if let Some(mode) = stmt.mode {
+ let envelope_type = match mode {
+ IcebergSinkMode::Upsert => "upsert",
+ IcebergSinkMode::Append => "append",
+ };
+ info.insert("envelope_type", json!(envelope_type));
+ }
+ }
+ }
+
+ "sink"
+ }
+ CreateMetricSink(stmt) => {
+ let Some(in_cluster) = stmt.in_cluster else {
+ return Err("missing IN CLUSTER".into());
+ };
+ let cluster_id = cluster_id(in_cluster)?;
+ info.insert("cluster_id", json!(cluster_id));
+ let from_id = item_id(stmt.from)?;
+ info.insert("from_id", json!(from_id));
+ "metric-sink"
+ }
+ CreateIndex(stmt) => {
+ let Some(in_cluster) = stmt.in_cluster else {
+ return Err("missing IN CLUSTER".into());
+ };
+ let cluster_id = cluster_id(in_cluster)?;
+ info.insert("cluster_id", json!(cluster_id));
+ let on_id = item_id(stmt.on_name)?;
+ info.insert("on_id", json!(on_id));
+ "index"
+ }
+ CreateType(_) => "type",
+ // NOTE: every statement that creates a catalog item needs an arm above. These
+ // catalog views run this over every item row before their type filter drops the
+ // unwanted rows, so one unclassified `create_sql` takes out `mz_objects`,
+ // `mz_indexes`, and every sibling view at once. The match is exhaustive to make
+ // that a compile error here, not a runtime failure.
+ Select(_)
+ | Insert(_)
+ | Copy(_)
+ | Update(_)
+ | Delete(_)
+ | CreateDatabase(_)
+ | CreateSchema(_)
+ | CreateRole(_)
+ | CreateCluster(_)
+ | CreateClusterReplica(_)
+ | CreateNetworkPolicy(_)
+ | AlterCluster(_)
+ | AlterOwner(_)
+ | AlterObjectRename(_)
+ | AlterObjectSwap(_)
+ | AlterRetainHistory(_)
+ | AlterIndex(_)
+ | AlterSecret(_)
+ | AlterSetCluster(_)
+ | AlterSink(_)
+ | AlterSource(_)
+ | AlterSystemSet(_)
+ | AlterSystemReset(_)
+ | AlterSystemResetAll(_)
+ | AlterConnection(_)
+ | AlterNetworkPolicy(_)
+ | AlterRole(_)
+ | AlterTableAddColumn(_)
+ | AlterMaterializedViewApplyReplacement(_)
+ | Discard(_)
+ | DropObjects(_)
+ | DropOwned(_)
+ | SetVariable(_)
+ | ResetVariable(_)
+ | Show(_)
+ | StartTransaction(_)
+ | SetTransaction(_)
+ | Commit(_)
+ | Rollback(_)
+ | Subscribe(_)
+ | ExplainPlan(_)
+ | ExplainPushdown(_)
+ | ExplainTimestamp(_)
+ | ExplainSinkSchema(_)
+ | ExplainAnalyzeObject(_)
+ | ExplainAnalyzeCluster(_)
+ | Declare(_)
+ | Fetch(_)
+ | Close(_)
+ | Prepare(_)
+ | Execute(_)
+ | ExecuteUnitTest(_)
+ | Deallocate(_)
+ | Raise(_)
+ | GrantRole(_)
+ | RevokeRole(_)
+ | GrantPrivileges(_)
+ | RevokePrivileges(_)
+ | AlterDefaultPrivileges(_)
+ | ReassignOwned(_)
+ | ValidateConnection(_)
+ | Comment(_) => return Err("not a CREATE item statement".into()),
+ };
+ info.insert("type", json!(item_type));
+
+ let info = info.into_iter().map(|(k, v)| (k.to_string(), v)).collect();
+ Ok(info)
+}
+
+/// Extracts the catalog item references from a catalog `create_sql` string as a JSONB object.
+///
+/// - `ids`: array of catalog id strings from `[ AS ]` references.
+/// - `named_funcs`: array of function references such as "pg_catalog"."max".
+/// - `named_types`: array of type references such as "pg_catalog"."int4", exclusive from the ones
+/// in `ids`.
+/// - `named_relations`: array of relation references, exclusive from the ones in `ids`.
+pub fn item_references(a: &str) -> Result {
+ fn qualified(name: &UnresolvedItemName) -> serde_json::Value {
+ match &name.0[..] {
+ [.., schema, item] => json!({"schema": schema.as_str(), "name": item.as_str()}),
+ [item] => json!({"schema": serde_json::Value::Null, "name": item.as_str()}),
+ [] => json!({"schema": serde_json::Value::Null, "name": ""}),
+ }
+ }
+
+ let stmt = parse_single_statement(a)?;
+
+ let refs = collect_item_references(&stmt);
+ mz_ore::soft_assert_or_log!(
+ refs.named_array_elements.is_empty(),
+ "persisted create_sql should never carry an array type T[] \
+ and resolve its array type in `ids`: {:?}",
+ refs.named_array_elements
+ );
+ Ok(json!({
+ "ids": refs.ids.iter().collect::>(),
+ "named_funcs": refs.named_funcs.iter().map(qualified).collect::>(),
+ "named_types": refs.named_types.iter().map(qualified).collect::>(),
+ "named_relations": refs.named_relations.iter().map(qualified).collect::>(),
+ }))
+}
+
+/// Minimal decoder for `ProtoPostgresSourcePublicationDetails`. The
+/// canonical proto lives in `mz-storage-types`, which depends on this
+/// crate via `mz-expr`, so we redeclare the two tags we read here. Upstream tag
+/// renumbers slip past silently. The `mz_postgres_sources` lockdown
+/// SLTs catch them.
+#[derive(Clone, PartialEq, ::prost::Message)]
+struct PostgresPublicationDetailsSubset {
+ #[prost(string, tag = "2")]
+ slot: String,
+ #[prost(uint64, optional, tag = "3")]
+ timeline_id: Option,
+}
+
+/// Extracts postgres source publication details (slot, timeline_id) from a
+/// catalog `create_sql`. Returns:
+///
+/// - jsonb `{"slot": , "timeline_id": }` for `CREATE SOURCE ... FROM POSTGRES
+/// CONNECTION ... (DETAILS = ...)` statements.
+/// - jsonb `null` for any other statement.
+///
+/// Errors if the statement fails to parse, is a postgres source without
+/// a `DETAILS` option, or if the `DETAILS` value can't be hex- and
+/// proto-decoded.
+pub fn postgres_source_details(a: &str) -> Result {
+ let stmt = parse_single_statement(a)?;
+
+ use mz_sql_parser::ast::CreateSourceConnection;
+ use mz_sql_parser::ast::Statement::CreateSource;
+ let options = match stmt {
+ CreateSource(stmt) => match stmt.connection {
+ CreateSourceConnection::Postgres { options, .. } => options,
+ _ => return Ok(serde_json::Value::Null),
+ },
+ _ => return Ok(serde_json::Value::Null),
+ };
+
+ let details_hex = options
+ .into_iter()
+ .find(|opt| opt.name == PgConfigOptionName::Details)
+ .and_then(|opt| match opt.value {
+ Some(WithOptionValue::Value(Value::String(s))) => Some(s),
+ _ => None,
+ })
+ .ok_or("missing DETAILS option on postgres source")?;
+
+ let details_bytes =
+ hex::decode(&details_hex).map_err(|e| format!("DETAILS is not valid hex: {e}"))?;
+
+ let details = PostgresPublicationDetailsSubset::decode(&*details_bytes)
+ .map_err(|e| format!("DETAILS is not a valid publication-details proto: {e}"))?;
+
+ Ok(json!({
+ "slot": details.slot,
+ "timeline_id": details.timeline_id,
+ }))
+}
+
+/// Extracts kafka source configuration (topic, group id prefix, connection
+/// id) from a catalog `create_sql`. Returns:
+///
+/// - jsonb `{"topic": , "group_id_prefix": , "connection_id": }` for
+/// `CREATE SOURCE ... FROM KAFKA CONNECTION ... (TOPIC = ..., [GROUP ID PREFIX = ...])`
+/// statements.
+/// - jsonb `null` for any other statement.
+///
+/// Errors if the statement fails to parse, is a kafka source without a
+/// `TOPIC` option, or references an unresolved connection name (i.e. one
+/// that hasn't been through purification).
+pub fn kafka_source_details(a: &str) -> Result {
+ let stmt = parse_single_statement(a)?;
+
+ use mz_sql_parser::ast::CreateSourceConnection;
+ use mz_sql_parser::ast::Statement::CreateSource;
+ let (connection, options) = match stmt {
+ CreateSource(stmt) => match stmt.connection {
+ CreateSourceConnection::Kafka {
+ connection,
+ options,
+ } => (connection, options),
+ _ => return Ok(serde_json::Value::Null),
+ },
+ _ => return Ok(serde_json::Value::Null),
+ };
+
+ let connection_id = item_id(connection)?;
+
+ let mut topic: Option = None;
+ let mut group_id_prefix: Option = None;
+ for opt in options {
+ let string_value = opt.value.as_ref().and_then(option_string);
+ match opt.name {
+ KafkaSourceConfigOptionName::Topic => topic = string_value,
+ KafkaSourceConfigOptionName::GroupIdPrefix => group_id_prefix = string_value,
+ _ => {}
+ }
+ }
+
+ let topic = topic.ok_or("missing TOPIC option on kafka source")?;
+
+ Ok(json!({
+ "topic": topic,
+ "group_id_prefix": group_id_prefix,
+ "connection_id": connection_id,
+ }))
+}
+
+/// Extracts source-export (source table) metadata from a catalog `create_sql`.
+///
+/// Returns, for a `CREATE TABLE ... FROM SOURCE` or a non-progress
+/// `CREATE SUBSOURCE ... OF SOURCE ...` statement:
+///
+/// ```json
+/// {
+/// "source_id": "",
+/// "external_reference": ["part1", "part2", ...],
+/// "envelope_type": ,
+/// "key_format": ,
+/// "value_format":
+/// }
+/// ```
+///
+/// `envelope_type`, `key_format`, and `value_format` are always null for a
+/// `CREATE SUBSOURCE` (the postgres/mysql/sql-server exports that use the old
+/// subsource syntax carry neither format nor envelope). They may also be null
+/// for a `CREATE TABLE ... FROM SOURCE` that omits FORMAT/ENVELOPE.
+///
+/// Returns jsonb `null` for progress subsources and for any statement that is
+/// not a source export. The caller distinguishes the four source-table views
+/// by joining `source_id` against `mz_sources` and filtering on the parent's
+/// type, so this helper stays connection-type agnostic.
+///
+/// Errors if the statement fails to parse, references an unresolved item name,
+/// or is a non-progress subsource missing its OF SOURCE or EXTERNAL REFERENCE.
+///
+/// The `key_format`/`value_format` derivation mirrors the runtime
+/// `DataSourceDesc::formats()` that the removed `pack_kafka_source_tables_update`
+/// packer read. A bare FORMAT only carries a key when it resolves to an
+/// encoding that has one, which among bare formats is only Avro or Protobuf
+/// read from a Confluent Schema Registry whose purified seed carries a key
+/// schema. A KEY FORMAT ... VALUE FORMAT ... spec always carries both.
+pub fn source_export_details(a: &str) -> Result {
+ // A bare FORMAT resolves to an encoding with a key only for Avro or
+ // Protobuf read from a schema registry whose purified seed carries a key
+ // schema. Every other bare format is value-only.
+ fn bare_format_has_key(fmt: &Format) -> bool {
+ match fmt {
+ Format::Avro(AvroSchema::Csr { csr_connection }) => csr_connection
+ .seed
+ .as_ref()
+ .is_some_and(|seed| seed.key_schema.is_some()),
+ Format::Protobuf(ProtobufSchema::Csr { csr_connection }) => csr_connection
+ .seed
+ .as_ref()
+ .is_some_and(|seed| seed.key.is_some()),
+ _ => false,
+ }
+ }
+
+ fn key_value_formats(
+ spec: &FormatSpecifier,
+ ) -> (Option<&'static str>, Option<&'static str>) {
+ match spec {
+ FormatSpecifier::KeyValue { key, value } => {
+ (Some(format_name(key)), Some(format_name(value)))
+ }
+ FormatSpecifier::Bare(fmt) => {
+ let value = Some(format_name(fmt));
+ let key = bare_format_has_key(fmt).then(|| format_name(fmt));
+ (key, value)
+ }
+ }
+ }
+
+ fn envelope_name(envelope: &SourceEnvelope) -> &'static str {
+ match envelope {
+ SourceEnvelope::None => "none",
+ SourceEnvelope::Debezium => "debezium",
+ SourceEnvelope::Upsert {
+ value_decode_err_policy,
+ } => {
+ if value_decode_err_policy
+ .iter()
+ .any(|p| matches!(p, SourceErrorPolicy::Inline { .. }))
+ {
+ "upsert-value-err-inline"
+ } else {
+ "upsert"
+ }
+ }
+ SourceEnvelope::CdcV2 => "materialize",
+ }
+ }
+
+ let stmt = parse_single_statement(a)?;
+
+ use mz_sql_parser::ast::Statement::{CreateSubsource, CreateTableFromSource};
+ match stmt {
+ CreateTableFromSource(stmt) => {
+ let source_id = item_id(stmt.source)?;
+ let external_reference = stmt
+ .external_reference
+ .ok_or("missing external reference on CREATE TABLE FROM SOURCE")?
+ .0
+ .into_iter()
+ .map(|ident| ident.into_string())
+ .collect::>();
+
+ let envelope_type = stmt.envelope.as_ref().map(envelope_name);
+ let (key_format, value_format) = match &stmt.format {
+ Some(spec) => key_value_formats(spec),
+ None => (None, None),
+ };
+
+ Ok(json!({
+ "source_id": source_id,
+ "external_reference": external_reference,
+ "envelope_type": envelope_type,
+ "key_format": key_format,
+ "value_format": value_format,
+ }))
+ }
+ CreateSubsource(stmt) => {
+ // Progress subsources track ingestion progress and are not
+ // source tables. They have no external reference.
+ let is_progress = stmt
+ .with_options
+ .iter()
+ .any(|o| matches!(o.name, CreateSubsourceOptionName::Progress));
+ if is_progress {
+ return Ok(serde_json::Value::Null);
+ }
+
+ let source_id = stmt
+ .of_source
+ .ok_or("non-progress CREATE SUBSOURCE without OF SOURCE")
+ .and_then(item_id)?;
+
+ let external_reference = stmt
+ .with_options
+ .into_iter()
+ .find(|o| matches!(o.name, CreateSubsourceOptionName::ExternalReference))
+ .and_then(|o| match o.value {
+ Some(WithOptionValue::UnresolvedItemName(name)) => Some(name),
+ _ => None,
+ })
+ .ok_or("CREATE SUBSOURCE missing EXTERNAL REFERENCE option")?
+ .0
+ .into_iter()
+ .map(|ident| ident.into_string())
+ .collect::>();
+
+ Ok(json!({
+ "source_id": source_id,
+ "external_reference": external_reference,
+ "envelope_type": serde_json::Value::Null,
+ "key_format": serde_json::Value::Null,
+ "value_format": serde_json::Value::Null,
+ }))
+ }
+ _ => Ok(serde_json::Value::Null),
+ }
+}
+
+/// Extracts connection-detail metadata from a catalog `create_sql`.
+///
+/// Returns a per-connection-type object with the fields that the
+/// `mz_kafka_connections`, `mz_ssh_tunnel_connections`, and `mz_aws_connections`
+/// builtin views need. For everything else (other connection types, including
+/// aws-privatelink whose only detail is context-derived, and non-connection
+/// statements) it returns jsonb `null`, so callers filter on `IS NOT NULL` and
+/// gate on the connection type separately (via
+/// `parse_catalog_create_sql(...)->>'connection_type'`, the way `mz_connections`
+/// already does).
+///
+/// The shape per type:
+///
+/// ```json
+/// // kafka
+/// { "brokers": ["host:port", ...], "progress_topic": }
+/// // ssh-tunnel
+/// { "public_key_1": "", "public_key_2": "" }
+/// // aws
+/// {
+/// "auth_kind": "credentials" | "assume-role",
+/// "endpoint": , "region": ,
+/// "access_key_id": , "access_key_id_secret_id": ,
+/// "secret_access_key_secret_id": ,
+/// "session_token": , "session_token_secret_id": ,
+/// "assume_role_arn": , "assume_role_session_name":
+/// }
+/// ```
+///
+/// `progress_topic` is null when the connection does not set an explicit
+/// `PROGRESS TOPIC`. The default (`_materialize-progress--`) is
+/// reconstructed by the view, not here, because it needs the environment id and
+/// the connection's own id. Values derived only from environment context
+/// (AWS principal, external id, trust policy, privatelink principal) are also
+/// left to the view. This keeps the helper a pure function of the `create_sql`.
+///
+/// For aws, an option is either an inline value or a secret reference. Inline
+/// values land in `access_key_id`/`session_token`; a secret reference lands in
+/// the matching `*_secret_id` as the referenced secret's catalog item id (the
+/// persisted `create_sql` stores resolved references as `[uNNN AS name]`).
+/// `auth_kind` is `assume-role` when `ASSUME ROLE ARN` is present, else
+/// `credentials`, matching the `AwsAuth` variant the removed packer read.
+///
+/// Errors if the statement fails to parse.
+pub fn connection_details(a: &str) -> Result {
+ // The persisted `create_sql` stores an inline broker as a single `BROKER`
+ // option and a broker list as a `BROKERS (...)` sequence. Either way we
+ // only need the addresses, which are present regardless of the tunnel
+ // (direct, SSH, or PrivateLink).
+ fn broker_addresses(values: &[ConnectionOption]) -> Vec {
+ let mut brokers = Vec::new();
+ for opt in values {
+ match (&opt.name, &opt.value) {
+ (ConnectionOptionName::Broker, Some(WithOptionValue::ConnectionKafkaBroker(b))) => {
+ brokers.push(b.address.clone());
+ }
+ (ConnectionOptionName::Brokers, Some(WithOptionValue::Sequence(seq))) => {
+ for v in seq {
+ if let WithOptionValue::ConnectionKafkaBroker(b) = v {
+ brokers.push(b.address.clone());
+ }
+ }
+ }
+ _ => {}
+ }
+ }
+ brokers
+ }
+
+ fn string_option(
+ values: &[ConnectionOption],
+ name: ConnectionOptionName,
+ ) -> Option {
+ values
+ .iter()
+ .find(|o| o.name == name)
+ .and_then(|o| o.value.as_ref())
+ .and_then(option_string)
+ }
+
+ // The catalog id of the secret a `SECRET ...` option references. Resolved
+ // references persist as `RawItemName::Id`, so an unresolved name yields
+ // None (the same treatment [`item_id`] gives item names).
+ fn secret_id_option>(
+ values: &[ConnectionOption],
+ name: ConnectionOptionName,
+ ) -> Option {
+ values.iter().find_map(|o| match &o.value {
+ Some(WithOptionValue::Secret(RawItemName::Id(id, _, _))) if o.name == name => {
+ Some(id.clone())
+ }
+ _ => None,
+ })
+ }
+
+ let stmt = parse_single_statement(a)?;
+
+ use mz_sql_parser::ast::Statement::CreateConnection;
+ let stmt = match stmt {
+ CreateConnection(stmt) => stmt,
+ _ => return Ok(serde_json::Value::Null),
+ };
+
+ match stmt.connection_type {
+ CreateConnectionType::Kafka => Ok(json!({
+ "brokers": broker_addresses(&stmt.values),
+ "progress_topic": string_option(&stmt.values, ConnectionOptionName::ProgressTopic),
+ })),
+ CreateConnectionType::Ssh => Ok(json!({
+ "public_key_1": string_option(&stmt.values, ConnectionOptionName::PublicKey1),
+ "public_key_2": string_option(&stmt.values, ConnectionOptionName::PublicKey2),
+ })),
+ CreateConnectionType::Aws => {
+ let assume_role_arn = string_option(&stmt.values, ConnectionOptionName::AssumeRoleArn);
+ let auth_kind = if assume_role_arn.is_some() {
+ "assume-role"
+ } else {
+ "credentials"
+ };
+ Ok(json!({
+ "auth_kind": auth_kind,
+ // Planning coerces an empty ENDPOINT to None (see
+ // `src/sql/src/plan/statement/ddl/connection.rs`), so the
+ // removed packer wrote NULL for `ENDPOINT = ''`. Match that.
+ "endpoint": string_option(&stmt.values, ConnectionOptionName::Endpoint)
+ .filter(|s| !s.is_empty()),
+ "region": string_option(&stmt.values, ConnectionOptionName::Region),
+ "access_key_id": string_option(&stmt.values, ConnectionOptionName::AccessKeyId),
+ "access_key_id_secret_id":
+ secret_id_option(&stmt.values, ConnectionOptionName::AccessKeyId),
+ "secret_access_key_secret_id":
+ secret_id_option(&stmt.values, ConnectionOptionName::SecretAccessKey),
+ "session_token": string_option(&stmt.values, ConnectionOptionName::SessionToken),
+ "session_token_secret_id":
+ secret_id_option(&stmt.values, ConnectionOptionName::SessionToken),
+ "assume_role_arn": assume_role_arn,
+ "assume_role_session_name":
+ string_option(&stmt.values, ConnectionOptionName::AssumeRoleSessionName),
+ }))
+ }
+ _ => Ok(serde_json::Value::Null),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use prost::Message as _;
+ use serde_json::json;
+
+ /// Encode the two proto fields our decoder cares about, using the same
+ /// tag numbering as the canonical proto.
+ fn encode_pg_details(slot: &str, timeline_id: Option) -> String {
+ let details = super::PostgresPublicationDetailsSubset {
+ slot: slot.to_string(),
+ timeline_id,
+ };
+ hex::encode(details.encode_to_vec())
+ }
+
+ fn pg_source_sql(details_hex: &str) -> String {
+ format!(
+ "CREATE SOURCE \"materialize\".\"public\".\"pg_src\" \
+ IN CLUSTER [u42] \
+ FROM POSTGRES CONNECTION [u10 AS \"materialize\".\"public\".\"pg_conn\"] \
+ (DETAILS = '{details_hex}', PUBLICATION = 'mz_source') \
+ FOR ALL TABLES"
+ )
+ }
+
+ fn kafka_source_sql(with_prefix: bool) -> String {
+ let prefix_opt = if with_prefix {
+ ", GROUP ID PREFIX 'my-prefix-'"
+ } else {
+ ""
+ };
+ format!(
+ "CREATE SOURCE \"materialize\".\"public\".\"k_src\" \
+ IN CLUSTER [u42] \
+ FROM KAFKA CONNECTION [u11 AS \"materialize\".\"public\".\"k_conn\"] \
+ (TOPIC 'test'{prefix_opt}) FORMAT TEXT"
+ )
+ }
+
+ // --- parse_postgres_source_details ---------------------------------------
+
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
+ fn pg_happy_path_with_timeline() {
+ let hex = encode_pg_details("materialize_abc", Some(42));
+ let sql = pg_source_sql(&hex);
+ let out = super::postgres_source_details(&sql).expect("ok");
+ assert_eq!(out, json!({ "slot": "materialize_abc", "timeline_id": 42 }));
+ }
+
+ #[mz_ore::test]
+ fn pg_happy_path_null_timeline() {
+ // Pre-2024 sources have no timeline_id field. The decoder must
+ // surface that as JSON null, not error.
+ let hex = encode_pg_details("materialize_legacy", None);
+ let sql = pg_source_sql(&hex);
+ let out = super::postgres_source_details(&sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({ "slot": "materialize_legacy", "timeline_id": null }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn pg_non_postgres_source_returns_null_jsonb() {
+ let sql = "CREATE SOURCE \"materialize\".\"public\".\"lg\" \
+ IN CLUSTER [u42] FROM LOAD GENERATOR COUNTER";
+ let out = super::postgres_source_details(sql).expect("ok");
+ assert_eq!(out, serde_json::Value::Null);
+ }
+
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn pg_non_create_source_returns_null_jsonb() {
+ let sql = "CREATE VIEW v AS SELECT 1";
+ let out = super::postgres_source_details(sql).expect("ok");
+ assert_eq!(out, serde_json::Value::Null);
+ }
+
+ #[mz_ore::test]
+ fn pg_missing_details_option_errors() {
+ let sql = "CREATE SOURCE \"materialize\".\"public\".\"pg_src\" \
+ IN CLUSTER [u42] \
+ FROM POSTGRES CONNECTION [u10 AS \"materialize\".\"public\".\"pg_conn\"] \
+ (PUBLICATION = 'mz_source') FOR ALL TABLES";
+ let err = super::postgres_source_details(sql).unwrap_err();
+ assert!(
+ err.contains("missing DETAILS"),
+ "wrong error message: {err}"
+ );
+ }
+
+ #[mz_ore::test]
+ fn pg_malformed_hex_errors() {
+ let sql = pg_source_sql("not-hex!!");
+ let err = super::postgres_source_details(&sql).unwrap_err();
+ assert!(err.contains("valid hex"), "wrong error message: {err}");
+ }
+
+ #[mz_ore::test]
+ fn pg_malformed_proto_errors() {
+ // Valid hex, garbage bytes. Prost decoding fails on unexpected wire
+ // format.
+ let sql = pg_source_sql("ffff");
+ let err = super::postgres_source_details(&sql).unwrap_err();
+ assert!(
+ err.contains("publication-details proto"),
+ "wrong error message: {err}"
+ );
+ }
+
+ // --- parse_kafka_source_details ------------------------------------------
+
+ #[mz_ore::test]
+ fn kafka_happy_path_with_prefix() {
+ let sql = kafka_source_sql(true);
+ let out = super::kafka_source_details(&sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "topic": "test",
+ "group_id_prefix": "my-prefix-",
+ "connection_id": "u11",
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn kafka_happy_path_without_prefix() {
+ let sql = kafka_source_sql(false);
+ let out = super::kafka_source_details(&sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "topic": "test",
+ "group_id_prefix": null,
+ "connection_id": "u11",
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn kafka_non_kafka_source_returns_null_jsonb() {
+ let sql = "CREATE SOURCE \"materialize\".\"public\".\"lg\" \
+ IN CLUSTER [u42] FROM LOAD GENERATOR COUNTER";
+ let out = super::kafka_source_details(sql).expect("ok");
+ assert_eq!(out, serde_json::Value::Null);
+ }
+
+ #[mz_ore::test]
+ fn kafka_missing_topic_errors() {
+ let sql = "CREATE SOURCE \"materialize\".\"public\".\"k_src\" \
+ IN CLUSTER [u42] \
+ FROM KAFKA CONNECTION [u11 AS \"materialize\".\"public\".\"k_conn\"] \
+ FORMAT TEXT";
+ let err = super::kafka_source_details(sql).unwrap_err();
+ assert!(err.contains("missing TOPIC"), "wrong error message: {err}");
+ }
+
+ #[mz_ore::test]
+ fn kafka_unquoted_topic_and_prefix() {
+ // Planning accepts a bare identifier for TOPIC / GROUP ID PREFIX, and it
+ // persists unquoted in create_sql. Matching only quoted strings would
+ // drop TOPIC and error the whole mz_kafka_source_tables view.
+ let sql = "CREATE SOURCE \"materialize\".\"public\".\"k_src\" \
+ IN CLUSTER [u42] \
+ FROM KAFKA CONNECTION [u11 AS \"materialize\".\"public\".\"k_conn\"] \
+ (TOPIC = my_topic, GROUP ID PREFIX = my_prefix) FORMAT TEXT";
+ let out = super::kafka_source_details(sql).expect("ok");
+ assert_eq!(out["topic"], json!("my_topic"));
+ assert_eq!(out["group_id_prefix"], json!("my_prefix"));
+ }
+
+ #[mz_ore::test]
+ fn kafka_unresolved_connection_errors() {
+ // A bare-name connection reference never happens after purification,
+ // but the decoder must reject it explicitly rather than silently
+ // dropping the connection_id.
+ let sql = "CREATE SOURCE \"materialize\".\"public\".\"k_src\" \
+ IN CLUSTER [u42] \
+ FROM KAFKA CONNECTION k_conn (TOPIC 'test') FORMAT TEXT";
+ let err = super::kafka_source_details(sql).unwrap_err();
+ assert!(
+ err.contains("unresolved item name"),
+ "wrong error message: {err}"
+ );
+ }
+
+ // --- parse_source_export_details -----------------------------------------
+
+ fn table_from_source_sql(reference: &str, suffix: &str) -> String {
+ format!(
+ "CREATE TABLE \"materialize\".\"public\".\"tbl\" \
+ FROM SOURCE [u1 AS \"materialize\".\"public\".\"src\"] \
+ (REFERENCE = {reference}){suffix}"
+ )
+ }
+
+ #[mz_ore::test]
+ fn export_table_postgres_style_no_format() {
+ // Postgres/mysql/sql-server tables carry a multi-part external
+ // reference and no format or envelope.
+ let sql = table_from_source_sql("\"db\".\"public\".\"t\"", "");
+ let out = super::source_export_details(&sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "source_id": "u1",
+ "external_reference": ["db", "public", "t"],
+ "envelope_type": null,
+ "key_format": null,
+ "value_format": null,
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn export_table_kafka_bare_value_only() {
+ // A bare non-registry FORMAT is value-only: no key format.
+ let sql = table_from_source_sql("\"topic\"", " FORMAT TEXT ENVELOPE NONE");
+ let out = super::source_export_details(&sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "source_id": "u1",
+ "external_reference": ["topic"],
+ "envelope_type": "none",
+ "key_format": null,
+ "value_format": "text",
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn export_table_kafka_omitted_envelope_is_null() {
+ // Omitting ENVELOPE persists as absent in create_sql, so this
+ // source-type-agnostic helper reports null. The mz_kafka_source_tables
+ // view is responsible for defaulting kafka's null envelope to 'none'.
+ let sql = table_from_source_sql("\"topic\"", " FORMAT TEXT");
+ let out = super::source_export_details(&sql).expect("ok");
+ assert_eq!(out["envelope_type"], serde_json::Value::Null);
+ }
+
+ #[mz_ore::test]
+ fn export_table_kafka_key_value_format() {
+ let sql = table_from_source_sql(
+ "\"topic\"",
+ " KEY FORMAT TEXT VALUE FORMAT TEXT ENVELOPE NONE",
+ );
+ let out = super::source_export_details(&sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "source_id": "u1",
+ "external_reference": ["topic"],
+ "envelope_type": "none",
+ "key_format": "text",
+ "value_format": "text",
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn export_table_kafka_bare_avro_seed_with_key() {
+ // A bare Avro CSR format whose seed carries a key schema resolves to
+ // an encoding with a key, so key_format mirrors value_format. This is
+ // the upsert/debezium path.
+ let sql = table_from_source_sql(
+ "\"topic\"",
+ " FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY \
+ CONNECTION [u5 AS \"materialize\".\"public\".\"csr\"] \
+ SEED KEY SCHEMA 'k' VALUE SCHEMA 'v' ENVELOPE UPSERT",
+ );
+ let out = super::source_export_details(&sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "source_id": "u1",
+ "external_reference": ["topic"],
+ "envelope_type": "upsert",
+ "key_format": "avro",
+ "value_format": "avro",
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn export_table_kafka_bare_avro_seed_without_key() {
+ // A bare Avro CSR seed with only a value schema is value-only.
+ let sql = table_from_source_sql(
+ "\"topic\"",
+ " FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY \
+ CONNECTION [u5 AS \"materialize\".\"public\".\"csr\"] \
+ SEED VALUE SCHEMA 'v' ENVELOPE NONE",
+ );
+ let out = super::source_export_details(&sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "source_id": "u1",
+ "external_reference": ["topic"],
+ "envelope_type": "none",
+ "key_format": null,
+ "value_format": "avro",
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn export_subsource_non_progress() {
+ // Old-syntax subsource: external reference lives in a WITH option, and
+ // there is never a format or envelope.
+ let sql = "CREATE SUBSOURCE \"materialize\".\"public\".\"sub\" (id int4) \
+ OF SOURCE [u1 AS \"materialize\".\"public\".\"src\"] \
+ WITH (EXTERNAL REFERENCE = \"db\".\"public\".\"t\")";
+ let out = super::source_export_details(sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "source_id": "u1",
+ "external_reference": ["db", "public", "t"],
+ "envelope_type": null,
+ "key_format": null,
+ "value_format": null,
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn export_progress_subsource_returns_null_jsonb() {
+ let sql = "CREATE SUBSOURCE \"materialize\".\"public\".\"progress\" (id int4) \
+ WITH (PROGRESS)";
+ let out = super::source_export_details(sql).expect("ok");
+ assert_eq!(out, serde_json::Value::Null);
+ }
+
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn export_non_source_export_returns_null_jsonb() {
+ let sql = "CREATE VIEW v AS SELECT 1";
+ let out = super::source_export_details(sql).expect("ok");
+ assert_eq!(out, serde_json::Value::Null);
+ }
+
+ #[mz_ore::test]
+ fn export_unresolved_source_name_errors() {
+ let sql = "CREATE TABLE \"materialize\".\"public\".\"tbl\" \
+ FROM SOURCE src (REFERENCE = \"topic\")";
+ let err = super::source_export_details(sql).unwrap_err();
+ assert!(
+ err.contains("unresolved item name"),
+ "wrong error message: {err}"
+ );
+ }
+
+ // --- parse_connection_details --------------------------------------------
+
+ #[mz_ore::test]
+ fn connection_kafka_single_broker_default_progress() {
+ // No explicit PROGRESS TOPIC: the helper leaves it null and the view
+ // reconstructs the default.
+ let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO KAFKA \
+ (BROKER = 'localhost:9092', SECURITY PROTOCOL = plaintext)";
+ let out = super::connection_details(sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "brokers": ["localhost:9092"],
+ "progress_topic": null,
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn connection_kafka_explicit_progress_topic() {
+ let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO KAFKA \
+ (BROKER = 'localhost:9092', PROGRESS TOPIC = 'override', \
+ SECURITY PROTOCOL = plaintext)";
+ let out = super::connection_details(sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "brokers": ["localhost:9092"],
+ "progress_topic": "override",
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn connection_kafka_broker_list() {
+ let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO KAFKA \
+ (BROKERS ('b1:9092', 'b2:9092'), SECURITY PROTOCOL = plaintext)";
+ let out = super::connection_details(sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "brokers": ["b1:9092", "b2:9092"],
+ "progress_topic": null,
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn connection_ssh_public_keys() {
+ let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO SSH TUNNEL \
+ (HOST = 'ssh.example.com', PORT = 22, USER = 'mz', \
+ PUBLIC KEY 1 = 'ssh-ed25519 AAAA', PUBLIC KEY 2 = 'ssh-ed25519 BBBB')";
+ let out = super::connection_details(sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "public_key_1": "ssh-ed25519 AAAA",
+ "public_key_2": "ssh-ed25519 BBBB",
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn connection_aws_credentials_inline_key() {
+ // Inline ACCESS KEY ID, secret SECRET ACCESS KEY. Assume-role columns
+ // stay null and auth_kind is credentials.
+ let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO AWS \
+ (ACCESS KEY ID = 'AKIAEXAMPLE', \
+ SECRET ACCESS KEY = SECRET [u1 AS \"materialize\".\"public\".\"sk\"])";
+ let out = super::connection_details(sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "auth_kind": "credentials",
+ "endpoint": null,
+ "region": null,
+ "access_key_id": "AKIAEXAMPLE",
+ "access_key_id_secret_id": null,
+ "secret_access_key_secret_id": "u1",
+ "session_token": null,
+ "session_token_secret_id": null,
+ "assume_role_arn": null,
+ "assume_role_session_name": null,
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn connection_aws_credentials_secret_key_and_session_token() {
+ // Every credential provided as a secret reference lands in the matching
+ // *_secret_id column as the referenced secret's catalog id.
+ let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO AWS \
+ (ENDPOINT = 'http://localhost', REGION = 'us-east-1', \
+ ACCESS KEY ID = SECRET [u1 AS \"materialize\".\"public\".\"ak\"], \
+ SECRET ACCESS KEY = SECRET [u2 AS \"materialize\".\"public\".\"sk\"], \
+ SESSION TOKEN = SECRET [u3 AS \"materialize\".\"public\".\"st\"])";
+ let out = super::connection_details(sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "auth_kind": "credentials",
+ "endpoint": "http://localhost",
+ "region": "us-east-1",
+ "access_key_id": null,
+ "access_key_id_secret_id": "u1",
+ "secret_access_key_secret_id": "u2",
+ "session_token": null,
+ "session_token_secret_id": "u3",
+ "assume_role_arn": null,
+ "assume_role_session_name": null,
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn connection_aws_assume_role() {
+ // Assume-role sets auth_kind and the assume-role columns; credential
+ // columns stay null.
+ let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO AWS \
+ (ASSUME ROLE ARN 'arn:aws:iam::123:role/mz', \
+ ASSUME ROLE SESSION NAME 'sess')";
+ let out = super::connection_details(sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "auth_kind": "assume-role",
+ "endpoint": null,
+ "region": null,
+ "access_key_id": null,
+ "access_key_id_secret_id": null,
+ "secret_access_key_secret_id": null,
+ "session_token": null,
+ "session_token_secret_id": null,
+ "assume_role_arn": "arn:aws:iam::123:role/mz",
+ "assume_role_session_name": "sess",
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn connection_kafka_unquoted_progress_topic() {
+ // A bare identifier PROGRESS TOPIC persists unquoted in create_sql. The
+ // helper must surface it, else the view substitutes the default topic.
+ let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO KAFKA \
+ (BROKER = 'localhost:9092', PROGRESS TOPIC = my_topic, \
+ SECURITY PROTOCOL = plaintext)";
+ let out = super::connection_details(sql).expect("ok");
+ assert_eq!(out["progress_topic"], json!("my_topic"));
+ }
+
+ #[mz_ore::test]
+ fn connection_aws_unquoted_option_values() {
+ // Planning accepts bare identifiers for these options and persists them
+ // unquoted. The helper must surface them, not fall back to NULL.
+ let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO AWS \
+ (ENDPOINT = localhost, REGION = useast1, \
+ ASSUME ROLE ARN 'arn:aws:iam::123:role/mz', \
+ ASSUME ROLE SESSION NAME = mysession)";
+ let out = super::connection_details(sql).expect("ok");
+ assert_eq!(out["endpoint"], json!("localhost"));
+ assert_eq!(out["region"], json!("useast1"));
+ assert_eq!(out["assume_role_session_name"], json!("mysession"));
+ }
+
+ #[mz_ore::test]
+ fn connection_aws_empty_endpoint_is_null() {
+ // Planning coerces ENDPOINT = '' to None, so the packer wrote NULL. The
+ // view must match, not report an empty string.
+ let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO AWS \
+ (ENDPOINT = '', ASSUME ROLE ARN 'arn:aws:iam::123:role/mz')";
+ let out = super::connection_details(sql).expect("ok");
+ assert_eq!(out["endpoint"], serde_json::Value::Null);
+ }
+
+ #[mz_ore::test]
+ fn connection_other_type_returns_null_jsonb() {
+ // A connection type without a detail view (postgres) yields null.
+ let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO POSTGRES \
+ (HOST = 'db', DATABASE = 'postgres', USER = 'mz')";
+ let out = super::connection_details(sql).expect("ok");
+ assert_eq!(out, serde_json::Value::Null);
+ }
+
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn connection_non_connection_returns_null_jsonb() {
+ let sql = "CREATE VIEW v AS SELECT 1";
+ let out = super::connection_details(sql).expect("ok");
+ assert_eq!(out, serde_json::Value::Null);
+ }
+
+ // --- parse_catalog_create_sql envelope_type ------------------------------
+
+ #[mz_ore::test]
+ fn catalog_kafka_old_syntax_omitted_envelope_defaults_none() {
+ // An old-syntax kafka source (carries EXPOSE PROGRESS AS) ingests into
+ // its own relation, so an omitted ENVELOPE means the default NONE, which
+ // the pre-MV packer reported as 'none'.
+ let sql = "CREATE SOURCE \"materialize\".\"public\".\"k\" \
+ IN CLUSTER [u42] \
+ FROM KAFKA CONNECTION [u11 AS \"materialize\".\"public\".\"k_conn\"] \
+ (TOPIC 'test') FORMAT TEXT \
+ EXPOSE PROGRESS AS [u12 AS \"materialize\".\"public\".\"k_progress\"]";
+ let out = super::item_details(sql).expect("ok");
+ assert_eq!(out["envelope_type"], json!("none"));
+ }
+
+ #[mz_ore::test]
+ fn catalog_kafka_old_syntax_explicit_envelope() {
+ let sql = "CREATE SOURCE \"materialize\".\"public\".\"k\" \
+ IN CLUSTER [u42] \
+ FROM KAFKA CONNECTION [u11 AS \"materialize\".\"public\".\"k_conn\"] \
+ (TOPIC 'test') FORMAT BYTES ENVELOPE UPSERT \
+ EXPOSE PROGRESS AS [u12 AS \"materialize\".\"public\".\"k_progress\"]";
+ let out = super::item_details(sql).expect("ok");
+ assert_eq!(out["envelope_type"], json!("upsert"));
+ }
+
+ #[mz_ore::test]
+ fn catalog_kafka_new_syntax_source_omits_envelope_type() {
+ // A new-syntax kafka source has no progress subsource (no EXPOSE PROGRESS
+ // AS). It ingests nothing itself. Envelopes live on the per-table exports,
+ // so its own envelope_type stays absent (SQL NULL).
+ let sql = "CREATE SOURCE \"materialize\".\"public\".\"k\" \
+ IN CLUSTER [u42] \
+ FROM KAFKA CONNECTION [u11 AS \"materialize\".\"public\".\"k_conn\"]";
+ let out = super::item_details(sql).expect("ok");
+ assert_eq!(out.get("envelope_type"), None);
+ }
+
+ #[mz_ore::test]
+ fn catalog_non_kafka_source_omits_envelope_type() {
+ // Non-kafka sources carry no envelope, so envelope_type stays absent
+ // (SQL NULL), not 'none'.
+ let sql = "CREATE SOURCE \"materialize\".\"public\".\"lg\" \
+ IN CLUSTER [u42] FROM LOAD GENERATOR COUNTER";
+ let out = super::item_details(sql).expect("ok");
+ assert_eq!(out.get("envelope_type"), None);
+ }
+
+ // --- parse_catalog_create_sql, CreateSink arm ----------------------------
+
+ const AVRO_FORMAT: &str = "FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION \
+ [u12 AS \"materialize\".\"public\".\"csr_conn\"]";
+
+ /// A persisted kafka-sink `create_sql`: resolved names, and a `TOPIC` that
+ /// planning guarantees.
+ fn kafka_sink_sql(key: Option<&str>, format: &str, envelope: &str) -> String {
+ let key_clause = key.map(|k| format!(" KEY ({k})")).unwrap_or_default();
+ format!(
+ "CREATE SINK \"materialize\".\"public\".\"snk\" \
+ IN CLUSTER [u42] \
+ FROM [u1 AS \"materialize\".\"public\".\"t\"] \
+ INTO KAFKA CONNECTION [u10 AS \"materialize\".\"public\".\"k_conn\"] \
+ (TOPIC 'sink-topic'){key_clause} {format} ENVELOPE {envelope}"
+ )
+ }
+
+ fn iceberg_sink_sql(mode: &str) -> String {
+ format!(
+ "CREATE SINK \"materialize\".\"public\".\"ice\" \
+ IN CLUSTER [u42] \
+ FROM [u1 AS \"materialize\".\"public\".\"t\"] \
+ INTO ICEBERG CATALOG CONNECTION [u20 AS \"materialize\".\"public\".\"cat_conn\"] \
+ (NAMESPACE 'ns', TABLE 'tbl') \
+ USING AWS CONNECTION [u21 AS \"materialize\".\"public\".\"aws_conn\"] \
+ MODE {mode}"
+ )
+ }
+
+ #[mz_ore::test]
+ fn sink_kafka_bare_format_without_key() {
+ let sql = kafka_sink_sql(None, "FORMAT JSON", "DEBEZIUM");
+ let out = super::item_details(&sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "type": "sink",
+ "sink_type": "kafka",
+ "cluster_id": "u42",
+ "connection_id": "u10",
+ "topic": "sink-topic",
+ "envelope_type": "debezium",
+ "format": "json",
+ "value_format": "json",
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn sink_kafka_bare_format_with_key_derives_key_format() {
+ // A bare format applies to the key too once the sink has a KEY, which
+ // is what makes the deprecated `format` column collapse to `avro`.
+ let sql = kafka_sink_sql(Some("a"), AVRO_FORMAT, "UPSERT");
+ let out = super::item_details(&sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "type": "sink",
+ "sink_type": "kafka",
+ "cluster_id": "u42",
+ "connection_id": "u10",
+ "topic": "sink-topic",
+ "envelope_type": "upsert",
+ "format": "avro",
+ "key_format": "avro",
+ "value_format": "avro",
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn sink_kafka_bare_text_format_with_key_does_not_collapse() {
+ // Only avro/avro and json/json collapse, so a keyed text sink reports
+ // the composite form even though both halves are `text`.
+ let sql = kafka_sink_sql(Some("a"), "FORMAT TEXT", "UPSERT");
+ let out = super::item_details(&sql).expect("ok");
+ assert_eq!(out["format"], json!("key-text-value-text"));
+ assert_eq!(out["key_format"], json!("text"));
+ assert_eq!(out["value_format"], json!("text"));
+ }
+
+ #[mz_ore::test]
+ fn sink_kafka_key_value_json_collapses() {
+ let sql = kafka_sink_sql(Some("a"), "KEY FORMAT JSON VALUE FORMAT JSON", "UPSERT");
+ let out = super::item_details(&sql).expect("ok");
+ assert_eq!(out["format"], json!("json"));
+ assert_eq!(out["key_format"], json!("json"));
+ assert_eq!(out["value_format"], json!("json"));
+ }
+
+ #[mz_ore::test]
+ fn sink_kafka_key_value_mixed_is_composite() {
+ let sql = kafka_sink_sql(Some("a"), "KEY FORMAT TEXT VALUE FORMAT BYTES", "UPSERT");
+ let out = super::item_details(&sql).expect("ok");
+ assert_eq!(out["format"], json!("key-text-value-bytes"));
+ assert_eq!(out["key_format"], json!("text"));
+ assert_eq!(out["value_format"], json!("bytes"));
+ }
+
+ #[mz_ore::test]
+ fn sink_kafka_key_format_without_key_is_dropped() {
+ // `kafka_sink_builder` ignores the key half of the format spec when the
+ // sink has no KEY, so neither `key_format` nor the composite `format`
+ // may reflect it.
+ let sql = kafka_sink_sql(None, "KEY FORMAT JSON VALUE FORMAT TEXT", "DEBEZIUM");
+ let out = super::item_details(&sql).expect("ok");
+ assert_eq!(out["format"], json!("text"));
+ assert_eq!(out["key_format"], serde_json::Value::Null);
+ assert_eq!(out["value_format"], json!("text"));
+ }
+
+ #[mz_ore::test]
+ fn sink_kafka_missing_topic_errors() {
+ let sql = "CREATE SINK \"materialize\".\"public\".\"snk\" \
+ IN CLUSTER [u42] \
+ FROM [u1 AS \"materialize\".\"public\".\"t\"] \
+ INTO KAFKA CONNECTION [u10 AS \"materialize\".\"public\".\"k_conn\"] \
+ FORMAT JSON ENVELOPE DEBEZIUM";
+ let err = super::item_details(sql).unwrap_err();
+ assert!(err.contains("missing TOPIC"), "wrong error message: {err}");
+ }
+
+ #[mz_ore::test]
+ fn sink_iceberg_upsert_mode() {
+ let sql = iceberg_sink_sql("UPSERT");
+ let out = super::item_details(&sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "type": "sink",
+ "sink_type": "iceberg",
+ "cluster_id": "u42",
+ // The catalog connection, never the AWS connection (u21).
+ "connection_id": "u20",
+ "namespace": "ns",
+ "table": "tbl",
+ "envelope_type": "upsert",
+ }),
+ );
+ }
+
+ #[mz_ore::test]
+ fn sink_iceberg_append_mode() {
+ let sql = iceberg_sink_sql("APPEND");
+ let out = super::item_details(&sql).expect("ok");
+ // `append` is reachable only through an iceberg sink's MODE.
+ assert_eq!(out["envelope_type"], json!("append"));
+ }
+
+ #[mz_ore::test]
+ fn sink_iceberg_missing_table_errors() {
+ let sql = "CREATE SINK \"materialize\".\"public\".\"ice\" \
+ IN CLUSTER [u42] \
+ FROM [u1 AS \"materialize\".\"public\".\"t\"] \
+ INTO ICEBERG CATALOG CONNECTION [u20 AS \"materialize\".\"public\".\"cat_conn\"] \
+ (NAMESPACE 'ns') MODE UPSERT";
+ let err = super::item_details(sql).unwrap_err();
+ assert!(err.contains("missing TABLE"), "wrong error message: {err}");
+ }
+
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn sink_arm_leaves_other_item_types_alone() {
+ let sql = "CREATE VIEW \"materialize\".\"public\".\"v\" AS SELECT 1";
+ let out = super::item_details(sql).expect("ok");
+ assert_eq!(out, json!({ "type": "view", "definition": "SELECT 1;" }));
+ }
+
+ // --- parse_catalog_create_sql --------------------------------------------
+
+ /// `type` for a `create_sql`, or the error message if parsing failed.
+ fn item_type(sql: &str) -> Result {
+ match super::item_details(sql) {
+ Ok(out) => match out {
+ serde_json::Value::Object(mut m) => match m.remove("type") {
+ Some(serde_json::Value::String(s)) => Ok(s),
+ other => panic!("no string `type` key: {other:?}"),
+ },
+ other => panic!("not a JSON object: {other:?}"),
+ },
+ Err(msg) => Err(msg),
+ }
+ }
+
+ fn view_sql(query: &str) -> String {
+ format!("CREATE VIEW \"materialize\".\"public\".\"v\" AS {query}")
+ }
+
+ /// `definition` for a `CREATE VIEW` whose query is `query`.
+ fn view_definition(query: &str) -> String {
+ match super::item_details(&view_sql(query)).expect("ok") {
+ serde_json::Value::Object(mut m) => match m.remove("definition") {
+ Some(serde_json::Value::String(s)) => s,
+ other => panic!("no string `definition` key: {other:?}"),
+ },
+ other => panic!("not a JSON object: {other:?}"),
+ }
+ }
+
+ /// `mz_tables` and `mz_views` select rows by
+ /// `parse_catalog_create_sql(...)->>'type'`, and the function runs over
+ /// every `Item` row in the catalog, so a statement kind that changes its
+ /// reported type silently gains or loses rows in those relations. Pin the
+ /// type of every kind the catalog can hold.
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn catalog_item_type_per_statement_kind() {
+ let cases = [
+ (
+ "CREATE TABLE \"materialize\".\"public\".\"t\" (a int4)",
+ "table",
+ ),
+ // A table created from a source, and a webhook table, are both
+ // `table`, so both land in mz_tables.
+ (
+ "CREATE TABLE \"materialize\".\"public\".\"tbl\" \
+ FROM SOURCE [u1 AS \"materialize\".\"public\".\"src\"] \
+ (REFERENCE = \"topic\") FORMAT TEXT",
+ "table",
+ ),
+ (
+ "CREATE TABLE \"materialize\".\"public\".\"wht\" FROM WEBHOOK BODY FORMAT JSON",
+ "table",
+ ),
+ (
+ "CREATE VIEW \"materialize\".\"public\".\"v\" AS SELECT 1",
+ "view",
+ ),
+ (
+ "CREATE MATERIALIZED VIEW \"materialize\".\"public\".\"mv\" \
+ IN CLUSTER [u1] AS SELECT 1",
+ "materialized-view",
+ ),
+ (
+ "CREATE SOURCE \"materialize\".\"public\".\"lg\" \
+ IN CLUSTER [u1] FROM LOAD GENERATOR COUNTER",
+ "source",
+ ),
+ (
+ "CREATE SOURCE \"materialize\".\"public\".\"wh\" \
+ IN CLUSTER [u1] FROM WEBHOOK BODY FORMAT JSON",
+ "source",
+ ),
+ (
+ "CREATE SUBSOURCE \"materialize\".\"public\".\"sub\" (id int4) \
+ OF SOURCE [u1 AS \"materialize\".\"public\".\"src\"]",
+ "subsource",
+ ),
+ (
+ "CREATE SUBSOURCE \"materialize\".\"public\".\"progress\" (id int4) \
+ WITH (PROGRESS)",
+ "subsource",
+ ),
+ (
+ "CREATE SINK \"materialize\".\"public\".\"snk\" IN CLUSTER [u1] \
+ FROM [u1 AS \"materialize\".\"public\".\"t\"] \
+ INTO KAFKA CONNECTION [u2 AS \"materialize\".\"public\".\"c\"] \
+ (TOPIC 'tp') FORMAT JSON ENVELOPE DEBEZIUM",
+ "sink",
+ ),
+ (
+ "CREATE INDEX \"i\" IN CLUSTER [u1] \
+ ON [u1 AS \"materialize\".\"public\".\"t\"] (\"a\")",
+ "index",
+ ),
+ (
+ "CREATE TYPE \"materialize\".\"public\".\"ty\" AS LIST (ELEMENT TYPE = int4)",
+ "type",
+ ),
+ (
+ "CREATE SECRET \"materialize\".\"public\".\"s\" AS 'x'",
+ "secret",
+ ),
+ (
+ "CREATE CONNECTION \"materialize\".\"public\".\"c\" \
+ TO KAFKA (BROKER 'b', SECURITY PROTOCOL PLAINTEXT)",
+ "connection",
+ ),
+ ];
+ for (sql, expected) in cases {
+ assert_eq!(item_type(sql).as_deref(), Ok(expected), "for {sql}");
+ }
+ }
+
+ /// `mz_views.definition` is produced here. It used to be produced by
+ /// `pack_view_update` in the adapter, so the exact rendering is a
+ /// compatibility surface: `pg_views.definition` reads it.
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn catalog_view_definition() {
+ // Identifiers and function names come back fully quoted, literals
+ // untouched, and PostgreSQL's trailing semicolon is appended.
+ assert_eq!(view_definition("SELECT 1"), "SELECT 1;");
+ assert_eq!(
+ view_definition("WITH c AS (SELECT 1 AS a) SELECT a FROM c"),
+ "WITH \"c\" AS (SELECT 1 AS \"a\") SELECT \"a\" FROM \"c\";"
+ );
+ assert_eq!(
+ view_definition("SELECT 1 UNION ALL SELECT 2"),
+ "SELECT 1 UNION ALL SELECT 2;"
+ );
+ assert_eq!(
+ view_definition("SELECT (SELECT max(a) FROM [u1 AS \"materialize\".\"public\".\"t\"])"),
+ "SELECT (SELECT \"max\"(\"a\") FROM [u1 AS \"materialize\".\"public\".\"t\"]);"
+ );
+ // Identifiers needing quotes, an embedded double quote, non-ASCII, an
+ // embedded single quote in a literal, and ORDER BY all survive.
+ assert_eq!(
+ view_definition(
+ "SELECT \"a b\", \"héllo\", \"q\"\"x\" \
+ FROM [u1 AS \"materialize\".\"public\".\"t\"] \
+ WHERE s = 'lit''eral' AND n = 42 ORDER BY 1"
+ ),
+ "SELECT \"a b\", \"héllo\", \"q\"\"x\" \
+ FROM [u1 AS \"materialize\".\"public\".\"t\"] \
+ WHERE \"s\" = 'lit''eral' AND \"n\" = 42 ORDER BY 1;"
+ );
+ }
+
+ /// The rendering must be a fixed point: `pg_views` consumers re-issue
+ /// `definition` as the body of a new view, so a second pass through the
+ /// parser has to produce the identical string. The trailing `;` is part of
+ /// what gets re-parsed.
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn catalog_view_definition_is_idempotent() {
+ for query in [
+ "SELECT 1",
+ "WITH c AS (SELECT 1 AS a) SELECT a FROM c",
+ "SELECT 1 UNION ALL SELECT 2",
+ "SELECT \"a b\", \"q\"\"x\" FROM [u1 AS \"materialize\".\"public\".\"t\"] \
+ WHERE s = 'lit''eral' ORDER BY 1",
+ ] {
+ let once = view_definition(query);
+ assert_eq!(
+ view_definition(&once),
+ once,
+ "not a fixed point for {query}"
+ );
+ }
+ }
+
+ /// `mz_tables.source_id` comes from this key. A table with no source must
+ /// omit it entirely, so the MV's `->>'source_id'` yields SQL NULL.
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn catalog_table_source_id() {
+ let from_source = super::item_details(
+ "CREATE TABLE \"materialize\".\"public\".\"tbl\" \
+ FROM SOURCE [u1 AS \"materialize\".\"public\".\"src\"] \
+ (REFERENCE = \"topic\") FORMAT TEXT",
+ )
+ .expect("ok");
+ assert_eq!(from_source, json!({ "type": "table", "source_id": "u1" }));
+
+ for sql in [
+ "CREATE TABLE \"materialize\".\"public\".\"t\" (a int4)",
+ "CREATE TABLE \"materialize\".\"public\".\"wht\" FROM WEBHOOK BODY FORMAT JSON",
+ ] {
+ assert_eq!(
+ super::item_details(sql).expect("ok"),
+ json!({ "type": "table" }),
+ "for {sql}"
+ );
+ }
+ }
+
+ /// Every error here is fatal to the whole of `mz_tables`/`mz_views`, not to
+ /// one row: the MVs call this function inside their `WHERE` clause, so an
+ /// item the parser rejects makes the relation unreadable for everyone.
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn catalog_create_sql_errors() {
+ assert_eq!(
+ item_type("this is not sql"),
+ Err(
+ "failed to parse create_sql: Expected a keyword at the beginning of a statement, \
+ found identifier \"this\""
+ .to_string()
+ )
+ );
+ assert_eq!(
+ item_type("CREATE TABLE t (a int4); CREATE TABLE u (b int4)"),
+ Err("expected a single statement, found 2".to_string())
+ );
+ // A statement that is not a CREATE of a catalog item, e.g. if a future
+ // change persists something else in an Item record.
+ assert_eq!(
+ item_type("SELECT 1"),
+ Err("not a CREATE item statement".to_string())
+ );
+ // Catalog `create_sql` always names items by id. An unresolved name
+ // means the record was written wrong.
+ assert_eq!(
+ item_type(
+ "CREATE TABLE \"materialize\".\"public\".\"tbl\" \
+ FROM SOURCE src (REFERENCE = \"topic\")"
+ ),
+ Err("unresolved item name".to_string())
+ );
+ }
+
+ // --- parse_catalog_item_references ----------------------------------------
+
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn item_references_view() {
+ let sql = "CREATE VIEW \"materialize\".\"public\".\"v\" AS \
+ SELECT \"pg_catalog\".\"abs\"(\"a\"::[s20 AS \"pg_catalog\".\"int4\"]) \
+ FROM [u1 AS \"materialize\".\"public\".\"t\"]";
+ let out = super::item_references(sql).expect("ok");
+ assert_eq!(
+ out,
+ json!({
+ "ids": ["s20", "u1"],
+ "named_funcs": [{"schema": "pg_catalog", "name": "abs"}],
+ "named_types": [],
+ "named_relations": [],
+ })
+ );
+ }
+
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn item_references_connection() {
+ let sql = "CREATE CONNECTION \"materialize\".\"public\".\"kc\" TO KAFKA (\
+ BROKER 'kafka:9092', \
+ SSH TUNNEL [u5 AS \"materialize\".\"public\".\"ssh\"], \
+ SASL PASSWORD = SECRET [u7 AS \"materialize\".\"public\".\"pw\"])";
+ let out = super::item_references(sql).expect("ok");
+ assert_eq!(out["ids"], json!(["u5", "u7"]));
+ assert_eq!(out["named_funcs"], json!([]));
+ }
+
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn item_references_cte_shadowing() {
+ let sql = "CREATE VIEW \"materialize\".\"public\".\"v\" AS \
+ WITH \"c\" AS (SELECT * FROM [u1 AS \"materialize\".\"public\".\"t\"]) \
+ SELECT * FROM \"c\"";
+ let out = super::item_references(sql).expect("ok");
+ assert_eq!(out["ids"], json!(["u1"]));
+ assert_eq!(out["named_relations"], json!([]));
+ }
+
+ #[mz_ore::test]
+ #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
+ fn item_references_name_based_relation_surfaces() {
+ // Stored SQL always prints relations as id references. A name-based
+ // relation must surface in "named_relations" rather than vanish.
+ let sql = "CREATE VIEW \"materialize\".\"public\".\"v\" AS \
+ SELECT * FROM \"mz_catalog\".\"mz_tables\"";
+ let out = super::item_references(sql).expect("ok");
+ assert_eq!(
+ out["named_relations"],
+ json!([{"schema": "mz_catalog", "name": "mz_tables"}])
+ );
+ }
+
+ #[mz_ore::test]
+ fn item_references_parse_error() {
+ let err = super::item_references("NOT SQL").unwrap_err();
+ assert!(
+ err.contains("failed to parse"),
+ "wrong error message: {err}"
+ );
+ }
+}
diff --git a/src/catalog-decode/src/durable.rs b/src/catalog-decode/src/durable.rs
new file mode 100644
index 0000000000000..6fdd592bbc9a6
--- /dev/null
+++ b/src/catalog-decode/src/durable.rs
@@ -0,0 +1,146 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+//! Decoders for values in their durable catalog JSON encoding, the serde shape
+//! of the `mz-catalog-protos` types.
+
+use mz_repr::Datum;
+use mz_repr::adt::jsonb::JsonbRef;
+use mz_repr::adt::mz_acl_item::{AclMode, MzAclItem};
+use mz_repr::adt::numeric;
+use mz_repr::role_id::RoleId;
+
+/// Converts a JSONB `Datum` into a `u64`.
+fn jsonb_datum_to_u64<'a>(d: Datum<'a>) -> Result {
+ let Datum::Numeric(n) = d else {
+ return Err("expected numeric value".into());
+ };
+
+ let mut cx = numeric::cx_datum();
+ cx.try_into_u64(n.0)
+ .map_err(|_| format!("number out of u64 range: {n}"))
+}
+
+/// Decodes a JSONB object of shape `{"bitflags": }` into an `AclMode`.
+///
+/// Shared decoder for [`privileges`] (which embeds the object as the
+/// `acl_mode` field of each privilege) and [`acl_mode`] (which receives the
+/// object at the top level).
+fn jsonb_datum_to_acl_mode(d: Datum) -> Result {
+ let Datum::Map(dict) = d else {
+ return Err(format!("unexpected acl_mode: {d}"));
+ };
+ let mut bits = None;
+ for (key, val) in dict.iter() {
+ match key {
+ "bitflags" => bits = Some(jsonb_datum_to_u64(val)?),
+ other => return Err(format!("unexpected acl_mode field: {other}")),
+ }
+ }
+ let bits = bits.ok_or_else(|| "missing acl_mode bitflags".to_string())?;
+ AclMode::from_bits(bits).ok_or_else(|| format!("invalid acl_mode bitflags: {bits}"))
+}
+
+/// Converts a JSONB `Datum` into a `RoleId`.
+fn jsonb_datum_to_role_id(d: Datum) -> Result {
+ match d {
+ Datum::String("Public") => Ok(RoleId::Public),
+ Datum::String(other) => Err(format!("unexpected role ID variant: {other}")),
+ Datum::Map(dict) => {
+ let (key, val) = dict.iter().next().ok_or_else(|| "empty".to_string())?;
+ let n = jsonb_datum_to_u64(val)?;
+ match key {
+ "User" => Ok(RoleId::User(n)),
+ "System" => Ok(RoleId::System(n)),
+ "Predefined" => Ok(RoleId::Predefined(n)),
+ other => Err(format!("unexpected role ID variant: {other}")),
+ }
+ }
+ _ => Err("expected string or object".into()),
+ }
+}
+
+/// Converts a catalog JSON-serialized ID value into the appropriate string format.
+///
+/// Supports all of Materialize's various ID types of the form ``.
+pub fn catalog_id(a: JsonbRef<'_>) -> Result {
+ match a.into_datum() {
+ // Unit variant, e.g. "Public"
+ Datum::String(variant) => match variant {
+ "Explain" => Ok("e".to_string()),
+ "Public" => Ok("p".to_string()),
+ other => Err(format!("unexpected ID variant: {other}")),
+ },
+ // Newtype variant, e.g. {"User": 1}
+ Datum::Map(dict) => {
+ let (key, val) = dict.iter().next().ok_or_else(|| "empty".to_string())?;
+ let prefix = match key {
+ "IntrospectionSourceIndex" => "si",
+ "Predefined" => "g",
+ "System" => "s",
+ "Transient" => "t",
+ "User" => "u",
+ other => return Err(format!("unexpected ID variant: {other}")),
+ };
+ let n = jsonb_datum_to_u64(val)?;
+ Ok(format!("{prefix}{n}"))
+ }
+ _ => Err("expected string or object".into()),
+ }
+}
+
+/// Converts a catalog JSON-serialized privilege array into a list of `MzAclItem`s.
+pub fn privileges(a: JsonbRef<'_>) -> Result, String> {
+ let parse_one = |datum| match datum {
+ Datum::Map(dict) => {
+ let mut grantee = None;
+ let mut grantor = None;
+ let mut acl_mode = None;
+ for (key, val) in dict.iter() {
+ match key {
+ "grantee" => {
+ let id = jsonb_datum_to_role_id(val)?;
+ grantee = Some(id);
+ }
+ "grantor" => {
+ let id = jsonb_datum_to_role_id(val)?;
+ grantor = Some(id);
+ }
+ "acl_mode" => {
+ acl_mode = Some(jsonb_datum_to_acl_mode(val)?);
+ }
+ other => return Err(format!("unexpected privilege field: {other}")),
+ }
+ }
+ Ok(MzAclItem {
+ grantee: grantee.ok_or_else(|| format!("missing grantee: {dict:?}"))?,
+ grantor: grantor.ok_or_else(|| "missing grantor in privilege".to_string())?,
+ acl_mode: acl_mode.ok_or_else(|| "missing acl_mode in privilege".to_string())?,
+ })
+ }
+ other => Err(format!("expected object in array, found: {other}")),
+ };
+
+ match a.into_datum() {
+ Datum::List(list) => {
+ let mut result = Vec::new();
+ for item in list.iter() {
+ result.push(parse_one(item)?);
+ }
+ Ok(result)
+ }
+ _ => Err("expected array".to_string()),
+ }
+}
+
+/// Converts a catalog JSON-serialized `AclMode` bitflags object
+/// (e.g. `{"bitflags": 514}`) into an `AclMode`.
+pub fn acl_mode(a: JsonbRef<'_>) -> Result {
+ jsonb_datum_to_acl_mode(a.into_datum())
+}
diff --git a/src/catalog-decode/src/lib.rs b/src/catalog-decode/src/lib.rs
new file mode 100644
index 0000000000000..321a21c7610f9
--- /dev/null
+++ b/src/catalog-decode/src/lib.rs
@@ -0,0 +1,22 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+//! Decoders for durable catalog data.
+//!
+//! The builtin catalog views derive their columns from `mz_catalog_raw`, which
+//! exposes catalog objects in their durable encoding. The decoding logic lives
+//! here, and `mz-expr` exposes each decoder as an `mz_internal.parse_*` scalar
+//! function. It sits below `mz-expr` in the dependency graph, so it cannot use
+//! the crates that own the decoded formats (`mz-catalog-protos`, `mz-sql`,
+//! `mz-storage-types`, `mz-audit-log`) and mirrors their shapes instead.
+//! Tests in those crates and the catalog sqllogictests catch drift.
+
+pub mod audit_log;
+pub mod create_sql;
+pub mod durable;
diff --git a/src/catalog/src/builtin.rs b/src/catalog/src/builtin.rs
index 7d10a3a92531a..bb659d6d8a1fe 100644
--- a/src/catalog/src/builtin.rs
+++ b/src/catalog/src/builtin.rs
@@ -1159,7 +1159,7 @@ pub static BUILTINS_STATIC: LazyLock>> = 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>> = 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));
}
diff --git a/src/catalog/src/builtin/mz_internal.rs b/src/catalog/src/builtin/mz_internal.rs
index abd0bda3efd28..4b155ca94720a 100644
--- a/src/catalog/src/builtin/mz_internal.rs
+++ b/src/catalog/src/builtin/mz_internal.rs
@@ -3572,39 +3572,72 @@ FROM commented",
}
});
-pub static MZ_SOURCE_REFERENCES: LazyLock = 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),
- )
- .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)],
- }),
-});
+pub static MZ_SOURCE_REFERENCES: LazyLock =
+ 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(),
+ sql: "
+IN CLUSTER mz_catalog_server
+WITH (
+ ASSERT NOT NULL source_id,
+ ASSERT NOT NULL name,
+ ASSERT NOT NULL updated_at
+) AS
+-- `refs` projects away usage of the `data` column before the `columns` subquery.
+-- Without the CTE, each row would need a copy of `data` which contains all source
+-- references, leading to memory growing quadratically.
+WITH refs AS (
+ SELECT
+ mz_internal.parse_catalog_id(data->'key'->'source') AS source_id,
+ -- `updated_at` is an `EpochMillis`, serialized as `{\"millis\": }`
+ to_timestamp(((data->'value'->'updated_at'->>'millis')::double) / 1000.0) AS updated_at,
+ reference
+ FROM
+ mz_internal.mz_catalog_raw,
+ jsonb_array_elements(data->'value'->'references') AS reference
+ WHERE data->>'kind' = 'SourceReferences'
+)
+SELECT
+ source_id,
+ reference->>'namespace' AS namespace,
+ reference->>'name' AS name,
+ updated_at,
+ (
+ SELECT array_agg(c.value ORDER BY c.ord)
+ FROM jsonb_array_elements_text(reference->'columns')
+ WITH ORDINALITY AS c(value, ord)
+ ) AS columns
+FROM refs",
+ 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 = LazyLock::new(|| BuiltinTable {
name: "mz_webhook_sources",
diff --git a/src/compute-types/tests/snapshots/func_registry_source.json b/src/compute-types/tests/snapshots/func_registry_source.json
index 1a141a9baee86..f5aaf0b008968 100644
--- a/src/compute-types/tests/snapshots/func_registry_source.json
+++ b/src/compute-types/tests/snapshots/func_registry_source.json
@@ -527,16 +527,16 @@
"pad_char": {"display":"padchar","sqlfunc_decl":null,"body_fingerprint":null},
"pad_char[unbounded]": {"display":"padchar","sqlfunc_decl":null,"body_fingerprint":null},
"panic": {"display":"mz_panic","sqlfunc_decl":"#[sqlfunc(sqlname = \"mz_panic\")] fn panic<'a>(a: &'a str) -> String","body_fingerprint":"95469a722cc261a6"},
- "parse_catalog_acl_mode": {"display":"parse_catalog_acl_mode","sqlfunc_decl":"#[sqlfunc] fn parse_catalog_acl_mode<'a>(a: JsonbRef<'a>) -> Result","body_fingerprint":"9f9ba9300791097b"},
- "parse_catalog_audit_log_details": {"display":"parse_catalog_audit_log_details","sqlfunc_decl":"#[sqlfunc] fn parse_catalog_audit_log_details<'a>(a: JsonbRef<'a>) -> Result","body_fingerprint":"4d767508de48e9e0"},
- "parse_catalog_create_sql": {"display":"parse_catalog_create_sql","sqlfunc_decl":"#[sqlfunc] fn parse_catalog_create_sql<'a>(a: &'a str) -> Result","body_fingerprint":"525420f3abbc957f"},
- "parse_catalog_id": {"display":"parse_catalog_id","sqlfunc_decl":"#[sqlfunc] fn parse_catalog_id<'a>(a: JsonbRef<'a>) -> Result","body_fingerprint":"cf39474c1e9e1bb9"},
- "parse_catalog_item_references": {"display":"parse_catalog_item_references","sqlfunc_decl":"#[sqlfunc] fn parse_catalog_item_references<'a>(a: &'a str) -> Result","body_fingerprint":"3296877cab7a9c6b"},
- "parse_catalog_privileges": {"display":"parse_catalog_privileges","sqlfunc_decl":"#[sqlfunc] fn parse_catalog_privileges<'a>(a: JsonbRef<'a>) -> Result, EvalError>","body_fingerprint":"203b2915643066b2"},
- "parse_connection_details": {"display":"parse_connection_details","sqlfunc_decl":"#[sqlfunc] fn parse_connection_details<'a>(a: &'a str) -> Result","body_fingerprint":"4b1754595ee6cc23"},
- "parse_kafka_source_details": {"display":"parse_kafka_source_details","sqlfunc_decl":"#[sqlfunc] fn parse_kafka_source_details<'a>(a: &'a str) -> Result","body_fingerprint":"da33c684761dac6e"},
- "parse_postgres_source_details": {"display":"parse_postgres_source_details","sqlfunc_decl":"#[sqlfunc] fn parse_postgres_source_details<'a>(a: &'a str) -> Result","body_fingerprint":"c61ad0624dcc7138"},
- "parse_source_export_details": {"display":"parse_source_export_details","sqlfunc_decl":"#[sqlfunc] fn parse_source_export_details<'a>(a: &'a str) -> Result","body_fingerprint":"b73b420b90bf9bae"},
+ "parse_catalog_acl_mode": {"display":"parse_catalog_acl_mode","sqlfunc_decl":"#[sqlfunc] fn parse_catalog_acl_mode<'a>(a: JsonbRef<'a>) -> Result","body_fingerprint":"416a57d9f77c1f57"},
+ "parse_catalog_audit_log_details": {"display":"parse_catalog_audit_log_details","sqlfunc_decl":"#[sqlfunc] fn parse_catalog_audit_log_details<'a>(a: JsonbRef<'a>) -> Result","body_fingerprint":"d71918c119d2ecf5"},
+ "parse_catalog_create_sql": {"display":"parse_catalog_create_sql","sqlfunc_decl":"#[sqlfunc] fn parse_catalog_create_sql<'a>(a: &'a str) -> Result","body_fingerprint":"826a24b886478cf7"},
+ "parse_catalog_id": {"display":"parse_catalog_id","sqlfunc_decl":"#[sqlfunc] fn parse_catalog_id<'a>(a: JsonbRef<'a>) -> Result","body_fingerprint":"3df74fa5b95ad2a3"},
+ "parse_catalog_item_references": {"display":"parse_catalog_item_references","sqlfunc_decl":"#[sqlfunc] fn parse_catalog_item_references<'a>(a: &'a str) -> Result","body_fingerprint":"afac69531d801621"},
+ "parse_catalog_privileges": {"display":"parse_catalog_privileges","sqlfunc_decl":"#[sqlfunc] fn parse_catalog_privileges<'a>(a: JsonbRef<'a>) -> Result, EvalError>","body_fingerprint":"0189155db8d73f28"},
+ "parse_connection_details": {"display":"parse_connection_details","sqlfunc_decl":"#[sqlfunc] fn parse_connection_details<'a>(a: &'a str) -> Result","body_fingerprint":"479b90a7cbe943f6"},
+ "parse_kafka_source_details": {"display":"parse_kafka_source_details","sqlfunc_decl":"#[sqlfunc] fn parse_kafka_source_details<'a>(a: &'a str) -> Result","body_fingerprint":"8717ae9a730e89a0"},
+ "parse_postgres_source_details": {"display":"parse_postgres_source_details","sqlfunc_decl":"#[sqlfunc] fn parse_postgres_source_details<'a>(a: &'a str) -> Result","body_fingerprint":"00a79828148946a7"},
+ "parse_source_export_details": {"display":"parse_source_export_details","sqlfunc_decl":"#[sqlfunc] fn parse_source_export_details<'a>(a: &'a str) -> Result","body_fingerprint":"a8e38e11bf5b92ac"},
"pg_column_size": {"display":"pg_column_size","sqlfunc_decl":"#[sqlfunc] fn pg_column_size<'a>(a: Datum<'a>) -> Result