From 2fc9257d999a8c6d65ab3c09620ad4ffbb316ea9 Mon Sep 17 00:00:00 2001 From: Dennis Felsing Date: Thu, 13 Aug 2026 13:33:31 +0200 Subject: [PATCH 1/8] dbt-materialize: quote the schema when copying default privileges (#38191) `internal_copy_schema_default_privs` interpolated the target schema name raw into `ALTER DEFAULT PRIVILEGES ... IN SCHEMA`, so the parser folded it to lowercase. #38066 started creating the deployment schema with a quoted name, which made the two disagree for any schema name that is not a bare lowercase identifier: `deploy_init` created `"Prod_dbt_deploy"` and then failed trying to alter default privileges in `prod_dbt_deploy`, after the schema already existed. The retry then took the schema-already-exists branch, which skips both privilege-copying macros, so the operation reported success while silently dropping the grants and default privileges the production schema had. Test run: https://buildkite.com/materialize/nightly/builds/18074 Co-authored-by: Claude Opus 5 (1M context) --- misc/dbt-materialize/CHANGELOG.md | 3 ++- .../materialize/macros/deploy/deploy_init.sql | 2 +- misc/dbt-materialize/tests/adapter/test_deploy.py | 14 +++++++++----- 3 files changed, 12 insertions(+), 7 deletions(-) diff --git a/misc/dbt-materialize/CHANGELOG.md b/misc/dbt-materialize/CHANGELOG.md index 01fdc3e3b205b..73c3c5a340572 100644 --- a/misc/dbt-materialize/CHANGELOG.md +++ b/misc/dbt-materialize/CHANGELOG.md @@ -10,7 +10,8 @@ * Fix `deploy_init` when the `CI_TAG` environment variable is set. The deployment schema was tagged without passing the schema name, so the operation failed with `COMMENT ON SCHEMA ""`. The deployment schema is - now also created with a quoted name, matching how it is dropped. + now also created with a quoted name, matching how it is dropped and how its + grants and default privileges are copied. ## 1.9.11 - 2026-07-26 diff --git a/misc/dbt-materialize/dbt/include/materialize/macros/deploy/deploy_init.sql b/misc/dbt-materialize/dbt/include/materialize/macros/deploy/deploy_init.sql index 8e2d450635ac3..acea343ff5ac4 100644 --- a/misc/dbt-materialize/dbt/include/materialize/macros/deploy/deploy_init.sql +++ b/misc/dbt-materialize/dbt/include/materialize/macros/deploy/deploy_init.sql @@ -170,7 +170,7 @@ SELECT WHEN object_owner = 'PUBLIC' THEN 'FOR ALL ROLES ' ELSE 'FOR ROLE ' || quote_ident(object_owner) || ' ' END || - 'IN SCHEMA {{ to }} ' || + 'IN SCHEMA ' || quote_ident({{ dbt.string_literal(to) }}) || ' ' || 'GRANT ' || privilege_type || ' ' || 'ON ' || object_type || 's ' || CASE diff --git a/misc/dbt-materialize/tests/adapter/test_deploy.py b/misc/dbt-materialize/tests/adapter/test_deploy.py index c5805cd66bd4e..34b745dbdfafb 100644 --- a/misc/dbt-materialize/tests/adapter/test_deploy.py +++ b/misc/dbt-materialize/tests/adapter/test_deploy.py @@ -64,14 +64,18 @@ def cleanup(self, project): project.run_sql("DROP ROLE IF EXISTS my_role") project.run_sql("DROP SCHEMA IF EXISTS blue_schema CASCADE") project.run_sql("DROP SCHEMA IF EXISTS green_schema CASCADE") + project.run_sql('DROP SCHEMA IF EXISTS "Green_Schema" CASCADE') project.run_sql("DROP CLUSTER IF EXISTS blue_cluster CASCADE") project.run_sql("DROP CLUSTER IF EXISTS green_cluster CASCADE") - def test_apply_schema_default_privileges(self, project): + # `deploy_init` creates the deployment schema with a quoted name, so a + # non-lowercase name has to survive into the generated statements too. + @pytest.mark.parametrize("to_schema", ["green_schema", "Green_Schema"]) + def test_apply_schema_default_privileges(self, project, to_schema): project.run_sql("CREATE ROLE my_role") project.run_sql("GRANT my_role TO materialize") project.run_sql("CREATE SCHEMA blue_schema") - project.run_sql("CREATE SCHEMA green_schema") + project.run_sql(f'CREATE SCHEMA "{to_schema}"') project.run_sql( "ALTER DEFAULT PRIVILEGES FOR ROLE my_role IN SCHEMA blue_schema GRANT SELECT ON TABLES TO my_role" ) @@ -81,15 +85,15 @@ def test_apply_schema_default_privileges(self, project): "run-operation", "internal_copy_schema_default_privs", "--args", - "{from: blue_schema, to: green_schema}", + f"{{from: blue_schema, to: {to_schema}}}", ] ) result = project.run_sql( - """SELECT count(*) = 1 + f"""SELECT count(*) = 1 FROM mz_internal.mz_show_default_privileges WHERE database = current_database() - AND schema = 'green_schema' + AND schema = '{to_schema}' AND grantee = 'my_role' AND object_type = 'table' AND privilege_type = 'SELECT'""", From cf4bca31d54e91b4ca1b10d851acaa2a2186acf0 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Thu, 13 Aug 2026 13:38:58 +0200 Subject: [PATCH 2/8] AGENTS.md: discourage semicolons instead of forbidding them (#38189) ### Motivation The style bullet in `AGENTS.md` told agents to "avoid em-dashes and semicolons", which read as an absolute ban. That produced a bad loop in practice: write a sentence with a semicolon, strip the semicolon to satisfy the rule, notice the remainder no longer parses, then rewrite the sentence from scratch. The rule was stronger than intended for semicolons. ### Description Split the single bullet in two, so the two marks get different strengths: * Em-dashes: unchanged, still avoid them. * Semicolons: discouraged rather than banned. Prefer a full stop or a comma, but a semicolon is acceptable where splitting genuinely hurts the sentence, and lists needing one for their own punctuation always are. The new bullet also names the failure mode directly, so an agent recognizes the loop instead of grinding through it: aim for prose that never needed the semicolon rather than prose with the semicolon taken back out, because a sentence left mangled by a deleted semicolon is worse than the semicolon. ### Verification Documentation-only change to `AGENTS.md` (`CLAUDE.md` is a symlink to it). No code or tests are affected. --- _Generated by [Claude Code](https://claude.ai/code/session_0127qk44MkFXAyUPEUWhtHP9)_ --------- Co-authored-by: Claude --- AGENTS.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 8560c347b78b9..bffc572450387 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -99,10 +99,10 @@ Two files control license policy, **keep in sync**: `deny.toml` (`[licenses].all we need to record that knowledge. In general comments need to stand on their own and make sense from just looking at them and the code around it, not previous changes. -* Avoid em-dashes and semicolons for structuring sentences, everywhere: code - comments, specs, design docs, all of it. Restructure with full stops and - commas instead. In most cases a sentence that wants an em-dash or semicolon - can just be split into two. +* Avoid em-dashes for structuring sentences in any prose: code comments, specs, + design docs, all of it. Restructure with full stops and commas instead. The + same goes for semicolons, though one is fine where splitting would mangle the + sentence. * Our guidance applies both when writing new code or designs, or when we notice deviations in code or architecture that we are working on. At the same time, we want to keep our changes minimal so it's good to call out deviations and From fbe634fd35eea8e6640239b002b153c04944aaf7 Mon Sep 17 00:00:00 2001 From: Dennis Felsing Date: Thu, 13 Aug 2026 14:10:27 +0200 Subject: [PATCH 3/8] ci: Nightly fixes (2026-08-13) (#38190) Based on https://buildkite.com/materialize/nightly/builds/18069 Test run: https://buildkite.com/materialize/nightly/builds/18073 & https://buildkite.com/materialize/nightly/builds/18077 --- ci/mkpipeline.py | 3 ++- ci/plugins/mzcompose/hooks/command | 2 +- src/dyncfg-file/Cargo.toml | 1 + .../subsource-resolution-empty.td | 11 ++--------- 4 files changed, 6 insertions(+), 11 deletions(-) diff --git a/ci/mkpipeline.py b/ci/mkpipeline.py index 586b3afdfb103..90e9363387b5a 100644 --- a/ci/mkpipeline.py +++ b/ci/mkpipeline.py @@ -761,7 +761,8 @@ def set_retry_on_agent_lost(pipeline: Any) -> None: retry.setdefault("automatic", []).extend( [ { - "exit_status": -1, # Agent lost or job timed out during checkout/setup + "exit_status": -1, + "signal_reason": "none", "limit": 2, }, { diff --git a/ci/plugins/mzcompose/hooks/command b/ci/plugins/mzcompose/hooks/command index 82c4de73868fa..c6956bfe7dfb5 100644 --- a/ci/plugins/mzcompose/hooks/command +++ b/ci/plugins/mzcompose/hooks/command @@ -404,7 +404,7 @@ cleanup() { && [ "$BUILDKITE_LABEL" != "QA Canary Environment Base Load" ] \ && [ "$BUILDKITE_LABEL" != "Parallel Benchmark against QA Canary Environment" ] \ && [ "$BUILDKITE_LABEL" != "Parallel Benchmark against QA Benchmarking Staging Environment" ] \ - && [ "$BUILDKITE_LABEL" != "Miri test (full)" ] \ + && [ "$BUILDKITE_LABEL" != ":rust: Miri test (full)" ] \ && [[ ! "$BUILDKITE_LABEL" =~ Terraform\ .* ]] \ && [[ ! "$BUILDKITE_LABEL" =~ Orchestratord\ .* ]] \ && [[ ! "$BUILDKITE_LABEL" =~ Cluster\ spec\ sheet.* ]]; then diff --git a/src/dyncfg-file/Cargo.toml b/src/dyncfg-file/Cargo.toml index afe4756d96398..a3d3e25f646de 100644 --- a/src/dyncfg-file/Cargo.toml +++ b/src/dyncfg-file/Cargo.toml @@ -20,6 +20,7 @@ tokio.workspace = true tracing.workspace = true [dev-dependencies] +mz-ore = { path = "../ore", features = ["test"] } tempfile.workspace = true [features] diff --git a/test/sql-server-cdc-old-syntax/subsource-resolution-empty.td b/test/sql-server-cdc-old-syntax/subsource-resolution-empty.td index a9230df920f23..d37831c484157 100644 --- a/test/sql-server-cdc-old-syntax/subsource-resolution-empty.td +++ b/test/sql-server-cdc-old-syntax/subsource-resolution-empty.td @@ -14,16 +14,9 @@ $ sql-server-connect name=sql-server server=tcp:sql-server,1433;IntegratedSecurity=true;TrustServerCertificate=true;User ID=${arg.default-sql-server-user};Password=${arg.default-sql-server-password} -$ sql-server-execute name=sql-server split-lines=false -IF EXISTS (SELECT name FROM sys.databases WHERE name = N'test') -BEGIN - ALTER DATABASE test SET SINGLE_USER WITH ROLLBACK IMMEDIATE; - DROP DATABASE test; -END; -CREATE DATABASE test; +$ sql-server-execute name=sql-server USE test; -EXEC sys.sp_cdc_enable_db; -ALTER DATABASE test SET ALLOW_SNAPSHOT_ISOLATION ON; +EXEC sys.sp_cdc_disable_table @source_schema = 'dbo', @source_name = 'dummy', @capture_instance = 'all'; > CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}' From 59c3bd161261c5042d75581f2026eb4d03d9d65a Mon Sep 17 00:00:00 2001 From: Jason Hernandez Date: Thu, 13 Aug 2026 05:13:35 -0700 Subject: [PATCH 4/8] (docs) simplify mz-debug download command for macOS (#38099) --- doc/user/content/integrations/mz-debug/_index.md | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/doc/user/content/integrations/mz-debug/_index.md b/doc/user/content/integrations/mz-debug/_index.md index 372a48f22d4d4..5a29555a5417c 100644 --- a/doc/user/content/integrations/mz-debug/_index.md +++ b/doc/user/content/integrations/mz-debug/_index.md @@ -19,9 +19,8 @@ disable_list: true {{< tab "macOS" >}} ```shell -ARCH=$(uname -m) sudo echo "Preparing to extract mz-debug..." -curl -L "https://binaries.materialize.com/mz-debug-latest-$ARCH-apple-darwin.tar.gz" \ +curl -L "https://binaries.materialize.com/mz-debug-latest-arm64-apple-darwin.tar.gz" \ | sudo tar -xzC /usr/local --strip-components=1 ``` {{}} From 07d08ad9c3504a24d7aa93759f8089e806df8ea8 Mon Sep 17 00:00:00 2001 From: Dennis Felsing Date: Thu, 13 Aug 2026 15:32:21 +0200 Subject: [PATCH 5/8] repr: bound regex compile memory by character class count (#38194) Follow-up to https://github.com/MaterializeInc/materialize/pull/37976, which still had gaps `Regex::new` projected a byte cost per AST node against a 1 GiB budget, charging a bracketed range like a literal. Case folding walks a range's whole span and keeps one Unicode range per fold mapping, so `[a-\x{2FFF}]` cost 22x its charge: a 1 MiB pattern of them was accepted and allocated 4.7 GiB before erroring, reachable from any `SELECT 'x' ~* `. An inline `(?i)` takes the same path, so case-sensitive operators were exposed too. Drop the cost model rather than adding a price for ranges. Per-kind byte prices need calibrating against the pinned `regex-syntax` and fail silently when one is set too low, which is how this got through. Two structural limits replace it: pattern bytes, which bound the node count and every kind whose cost is bounded, and `MAX_REGEX_CHARACTER_CLASSES` (2000), covering Unicode, Perl and POSIX classes and bracketed ranges. `PatternTooExpensive` becomes `TooManyCharacterClasses`. The counter's matches are exhaustive, so a `regex-syntax` bump adding a node kind fails to compile rather than defaulting it to "not a class". The two limits are independent, so the byte limit can now move on its own. It contributes the larger share of the bound, since 1 MiB of literals still translates to roughly 400 MiB. Counting rather than pricing means a pattern is rejected on its class count even where every class is cheap, and that is a deliberate break with released behavior. A machine-generated alternation carrying, say, 3000 `\d` compiles on every released version and does not compile here. Such a pattern inside an existing view const-folds to an error literal, so the object still loads but queries against it error, and plan-time compiles like `regexp_extract` fail to re-plan. The byte limit shipped with the same property, and the shapes that carry thousands of classes are contrived, so this is worth taking. `regex_two_limits_bound_what_a_compile_spends` replaces the per-kind charge table, holding measured translation heap against what the limits allow and pinning how each node kind is classified. `regex_wide_bracketed_range_counts_as_a_class` covers the wide-range vector, and `regex.slt` gains cases for it under both `~*` and an inline `(?i)`. Release notes: This release will reject regular expressions containing more than 2000 character classes, counting each Unicode, Perl, or POSIX class such as `\p{L}`, `\d`, or `[[:alpha:]]`, and each range such as `a-z`. Co-authored-by: Claude Opus 5 (1M context) --- doc/user/content/sql/functions/_index.md | 4 +- src/expr/src/scalar/like_pattern.rs | 2 +- src/repr/src/adt/regex.rs | 402 +++++++++++++---------- test/sqllogictest/regex.slt | 30 +- 4 files changed, 247 insertions(+), 191 deletions(-) diff --git a/doc/user/content/sql/functions/_index.md b/doc/user/content/sql/functions/_index.md index e7d84b4d3f45b..64613ef993b8f 100644 --- a/doc/user/content/sql/functions/_index.md +++ b/doc/user/content/sql/functions/_index.md @@ -113,7 +113,9 @@ Operator | Computes The regular expression syntax supported by Materialize is documented by the [Rust `regex` crate](https://docs.rs/regex/*/#syntax). The maximum length of a regular expression is 1 MiB in its raw form, and 10 MiB -after compiling it. +after compiling it. A regular expression may also contain at most 2000 character +classes, counting each Unicode, Perl, or POSIX class such as `\p{L}`, `\d`, or +`[[:alpha:]]`, and each range such as `a-z`. {{< warning >}} Materialize regular expressions are similar to, but not identical to, PostgreSQL diff --git a/src/expr/src/scalar/like_pattern.rs b/src/expr/src/scalar/like_pattern.rs index 75e9a33035836..06f4843a68037 100644 --- a/src/expr/src/scalar/like_pattern.rs +++ b/src/expr/src/scalar/like_pattern.rs @@ -369,7 +369,7 @@ fn build_regex(subpatterns: &[Subpattern], case_insensitive: bool) -> Result Ok(regex), Err(RegexCompilationError::PatternTooLarge { .. }) => Err(EvalError::LikePatternTooLong), - Err(RegexCompilationError::PatternTooExpensive { .. }) => { + Err(RegexCompilationError::TooManyCharacterClasses { .. }) => { Err(EvalError::LikePatternTooLong) } Err(RegexCompilationError::RegexError(regex::Error::CompiledTooBig(_))) => { diff --git a/src/repr/src/adt/regex.rs b/src/repr/src/adt/regex.rs index 304e3d4e10a58..8d536d0edb16d 100644 --- a/src/repr/src/adt/regex.rs +++ b/src/repr/src/adt/regex.rs @@ -35,96 +35,103 @@ const MAX_REGEX_SIZE_AFTER_COMPILATION: usize = 10 * 1024 * 1024; /// be the case. Since we compile regexes in envd, we need strict limits to prevent envd OOMs. /// See for an example. /// -/// This bounds the AST parse, which is what `estimate_compile_memory` needs before it can say -/// anything about the pattern. It does not bound the compile itself, see -/// `MAX_REGEX_COMPILE_MEMORY`. +/// This bounds the AST's node count, since every node needs at least one pattern byte, and with it +/// every node kind whose cost is bounded. Character classes are not, hence +/// `MAX_REGEX_CHARACTER_CLASSES`. /// /// Note: This number is mentioned in our user-facing docs at the "String operators" in the function /// reference. const MAX_REGEX_SIZE_BEFORE_COMPILATION: usize = 1 * 1024 * 1024; -/// The maximum heap a single compile may be projected to allocate. +/// The maximum number of character classes a pattern may contain. /// -/// Neither limit above bounds what a compile actually spends. `size_limit` bounds the compiled -/// NFA, and a pattern's byte length bounds only itself. The memory goes to `regex-syntax` -/// translating the pattern's AST into its HIR, a stage that runs before anything `size_limit` -/// measures. Cost there is linear in the AST's node count, but the cost *per node* spans two -/// orders of magnitude, from a few hundred bytes for a literal or a group up to tens of kilobytes -/// for a Unicode property under the `i` flag, so a byte budget cannot bound it. The per-kind costs -/// themselves are a fact about the `regex-syntax` version we pin, so they are measured rather than -/// written down here, in `regex_compile_memory_charges_cover_measured_cost`. +/// Byte length cannot bound what a compile spends, and neither can `size_limit`, which covers only +/// the compiled NFA. The memory goes to `regex-syntax` translating the AST into its HIR, where a +/// character class expands to hundreds of Unicode ranges out of a handful of pattern bytes. `\p{L}` +/// is five bytes, and under case folding `[a-\x{2FFF}]` is no cheaper, since the translator walks +/// the range codepoint by codepoint keeping one range per fold mapping. +/// See . /// -/// A 1 MiB pattern of `\p{L}` under the `i` flag, one byte under -/// `MAX_REGEX_SIZE_BEFORE_COMPILATION`, therefore allocates gigabytes before returning -/// `CompiledTooBig`. Since we compile regexes in envd, that made any `SELECT` an OOM vector, -/// which is what this limit exists to close. +/// Counting rather than pricing is deliberate. Per-kind byte prices need calibrating against the +/// pinned `regex-syntax` and fail silently when one is set too low, whereas a count only asks +/// whether a kind can expand without bound, and over-counting merely costs a few legitimate +/// patterns. /// -/// NOTE: a counted repetition needs no budget of its own. `\p{L}{200000}` stays two AST nodes, -/// and the expansion happens later in the NFA compiler, where `size_limit` already rejects it -/// incrementally. -const MAX_REGEX_COMPILE_MEMORY: usize = 1024 * 1024 * 1024; - -/// Charged per AST node against `MAX_REGEX_COMPILE_MEMORY`. Covers every node kind that expands -/// to at most a handful of Unicode ranges: literals, `.`, `[a-c]`, groups, repetitions, -/// assertions. Sized a few times above the costliest of them so it absorbs allocator differences, -/// but capped well under `MAX_REGEX_COMPILE_MEMORY / MAX_REGEX_SIZE_BEFORE_COMPILATION`, since a -/// 1 MiB pattern of plain literals, which is what a machine-generated alternation looks like, has -/// to stay within the budget. See `regex_cheap_nodes_do_not_collide_with_the_byte_limit`. -const COMPILE_MEMORY_PER_NODE: usize = 768; - -/// Charged in place of `COMPILE_MEMORY_PER_NODE` for a Unicode-property or Perl class item, the -/// node kinds that expand to hundreds of ranges. +/// This and the byte limit are independent, and multiply out to a bound on one compile that +/// `regex_two_limits_bound_what_a_compile_spends` holds against measurement. /// -/// Sized well above the costliest such node measured, which is what -/// `regex_compile_memory_charges_cover_measured_cost` checks. The margin is deliberate. Which -/// property costs the most is a fact about the Unicode tables `regex-syntax` ships, so a version -/// bump can shift it, and no legitimate pattern carries the thousands of Unicode classes it takes -/// to reach the budget. -const COMPILE_MEMORY_PER_CLASS: usize = 96 * 1024; - -/// Sums the projected translation cost of every node in an AST. -struct CompileMemoryEstimator { - estimate: usize, +/// Note: This number is mentioned in our user-facing docs at the "String operators" in the function +/// reference. +/// +/// NOTE: `\p{L}{200000}` stays one class in the AST. `size_limit` rejects its expansion later, in +/// the NFA compiler. +const MAX_REGEX_CHARACTER_CLASSES: usize = 2000; + +/// Counts the character classes in an AST, the node kinds whose translated size is not bounded by +/// the pattern bytes that spell them. +/// +/// Flags do not enter into it. Folding is the expensive direction for every kind, and `(?i)` turns +/// it on from inside the pattern, out of reach of the flag [`Regex`] is built with, so a kind is +/// judged by what it costs folded. +/// +/// NOTE: the matches are exhaustive on purpose. A `regex-syntax` bump adding a node kind has to +/// fail to compile here rather than default to "not a class", which no test could catch. +struct CharacterClassCounter { + count: usize, } -impl Visitor for CompileMemoryEstimator { +impl Visitor for CharacterClassCounter { type Output = usize; type Err = Infallible; fn finish(self) -> Result { - Ok(self.estimate) + Ok(self.count) } fn visit_pre(&mut self, ast: &Ast) -> Result<(), Infallible> { - self.estimate = self.estimate.saturating_add(match ast { - Ast::ClassUnicode(_) | Ast::ClassPerl(_) => COMPILE_MEMORY_PER_CLASS, - _ => COMPILE_MEMORY_PER_NODE, - }); + self.count += match ast { + Ast::ClassUnicode(_) | Ast::ClassPerl(_) => 1, + // A bracketed class carries no ranges itself. Its items do, and are counted below. + Ast::Empty(_) + | Ast::Flags(_) + | Ast::Literal(_) + | Ast::Dot(_) + | Ast::Assertion(_) + | Ast::ClassBracketed(_) + | Ast::Repetition(_) + | Ast::Group(_) + | Ast::Alternation(_) + | Ast::Concat(_) => 0, + }; Ok(()) } fn visit_class_set_item_pre(&mut self, item: &ClassSetItem) -> Result<(), Infallible> { - // Items nested inside a bracketed class, e.g. the `\p{L}` in `[a\p{L}]`. Each contributes - // its own ranges to the one translated class, so each is charged. A union can only merge - // ranges, never add them, so the sum of the parts is an upper bound on the whole. - self.estimate = self.estimate.saturating_add(match item { - ClassSetItem::Unicode(_) | ClassSetItem::Perl(_) => COMPILE_MEMORY_PER_CLASS, - _ => COMPILE_MEMORY_PER_NODE, - }); + // Items of a bracketed class, e.g. the `\p{L}` in `[a\p{L}]`. Each contributes its own + // ranges, and a union only merges ranges, so counting the parts bounds the whole. + self.count += match item { + ClassSetItem::Unicode(_) + | ClassSetItem::Perl(_) + | ClassSetItem::Ascii(_) + | ClassSetItem::Range(_) => 1, + ClassSetItem::Empty(_) + | ClassSetItem::Literal(_) + | ClassSetItem::Bracketed(_) + | ClassSetItem::Union(_) => 0, + }; Ok(()) } } -/// Projects the heap that compiling `pattern` would allocate, or `None` if it does not parse. +/// Counts the character classes in `pattern`, or `None` if it does not parse. /// -/// A pattern we cannot parse is left to [`RegexBuilder`], which rejects it with the message users -/// already see. That is not an escape hatch: both parse through the same `regex-syntax` version -/// with the same default configuration (`nest_limit` 250, `octal` off), so a pattern that fails -/// here fails there too. -fn estimate_compile_memory(pattern: &str) -> Option { +/// Not an escape hatch: an unparseable pattern is left to [`RegexBuilder`], and both parse through +/// the same `regex-syntax` with the same configuration (`nest_limit` 250, `octal` off), so a pattern +/// that fails here fails there too. +fn count_character_classes(pattern: &str) -> Option { let ast = ast::parse::Parser::new().parse(pattern).ok()?; - match ast::visit(&ast, CompileMemoryEstimator { estimate: 0 }) { - Ok(estimate) => Some(estimate), + match ast::visit(&ast, CharacterClassCounter { count: 0 }) { + Ok(count) => Some(count), Err(infallible) => match infallible {}, } } @@ -181,9 +188,9 @@ impl Regex { pattern_size: pattern.len(), }); } - if let Some(estimate) = estimate_compile_memory(pattern) { - if estimate > MAX_REGEX_COMPILE_MEMORY { - return Err(RegexCompilationError::PatternTooExpensive { estimate }); + if let Some(classes) = count_character_classes(pattern) { + if classes > MAX_REGEX_CHARACTER_CLASSES { + return Err(RegexCompilationError::TooManyCharacterClasses { classes }); } } let mut regex_builder = RegexBuilder::new(pattern); @@ -212,9 +219,8 @@ pub enum RegexCompilationError { RegexError(Error), /// Regex pattern size exceeds MAX_REGEX_SIZE_BEFORE_COMPILATION. PatternTooLarge { pattern_size: usize }, - /// Compiling the pattern is projected to allocate more than - /// MAX_REGEX_COMPILE_MEMORY. - PatternTooExpensive { estimate: usize }, + /// Regex pattern contains more than MAX_REGEX_CHARACTER_CLASSES character classes. + TooManyCharacterClasses { classes: usize }, } impl fmt::Display for RegexCompilationError { @@ -228,11 +234,12 @@ impl fmt::Display for RegexCompilationError { "regex pattern too large ({} bytes, max {} bytes)", patter_size, MAX_REGEX_SIZE_BEFORE_COMPILATION ), - RegexCompilationError::PatternTooExpensive { estimate } => write!( + RegexCompilationError::TooManyCharacterClasses { classes } => write!( f, - "regex pattern too expensive to compile (needs about {} bytes, max {} bytes); \ - reduce the number of character classes", - estimate, MAX_REGEX_COMPILE_MEMORY + "regex pattern has too many character classes ({}, max {}). \ + A character class is a Unicode, Perl or POSIX class such as `\\p{{L}}`, `\\d` \ + or `[[:alpha:]]`, or a range such as `a-z`", + classes, MAX_REGEX_CHARACTER_CLASSES ), } } @@ -437,13 +444,11 @@ mod tests { use super::*; - /// Wraps the system allocator to record the peak heap the calling thread has asked for, so - /// [`regex_compile_memory_charges_cover_measured_cost`] can hold the charges in this module - /// against what `regex-syntax` actually allocates. + /// Wraps the system allocator to record the peak heap the calling thread has asked for. /// - /// Counters are per thread, so tests running in parallel do not perturb each other. What is - /// counted is bytes requested, not resident bytes, which makes a measurement reproducible - /// across allocators at the cost of running a little above true RSS. + /// Counters are per thread, so tests running in parallel do not perturb each other. Bytes + /// requested are counted, not resident bytes, which keeps a measurement reproducible across + /// allocators at the cost of running a little above true RSS. struct TrackingAllocator; thread_local! { @@ -515,13 +520,12 @@ mod tests { static ALLOC: TrackingAllocator = TrackingAllocator; /// Peak heap the calling thread requests while parsing `pattern` and translating it to HIR, - /// the stage [`estimate_compile_memory`] projects. + /// the stage the two limits bound. /// - /// Compiling the NFA afterwards is deliberately left out. `size_limit` bounds it, and its - /// cost is close to a constant (a few times `MAX_REGEX_SIZE_AFTER_COMPILATION`) that would - /// swamp the per-node signal at the pattern sizes this test can afford. The translator flags - /// mirror what [`Regex::new_dot_matches_new_line`] hands to [`RegexBuilder`], since - /// `case_insensitive` alone moves a class's cost by an order of magnitude. + /// Compiling the NFA afterwards is left out: `size_limit` bounds it, and its near-constant cost + /// would swamp the per-node signal at the pattern sizes a test can afford. The translator flags + /// mirror [`Regex::new_dot_matches_new_line`], since `case_insensitive` alone moves a class's + /// cost by an order of magnitude. fn peak_translate_bytes(pattern: &str, case_insensitive: bool) -> usize { let base = LIVE_BYTES.with(|live| live.get()); PEAK_BYTES.with(|peak| peak.set(base)); @@ -542,155 +546,193 @@ mod tests { PEAK_BYTES.with(|peak| peak.get()).saturating_sub(base) } - /// A class-heavy pattern one byte under `MAX_REGEX_SIZE_BEFORE_COMPILATION` used to allocate - /// ~7.8 GB in envd before erroring, reachable from an unprivileged `SELECT 'x' ~* `. - /// It must now be rejected without compiling. + /// A class-heavy pattern one byte under `MAX_REGEX_SIZE_BEFORE_COMPILATION` costs gigabytes to + /// translate, and is reachable from an unprivileged `SELECT 'x' ~* `, so it has to be + /// rejected without compiling. #[mz_ore::test] #[cfg_attr(miri, ignore)] // too slow fn regex_class_heavy_pattern_rejected_before_compiling() { let pattern = r"\p{L}".repeat(MAX_REGEX_SIZE_BEFORE_COMPILATION / r"\p{L}".len()); assert!(pattern.len() <= MAX_REGEX_SIZE_BEFORE_COMPILATION); - // Case-insensitive is the expensive direction, but the estimate does not depend on the - // flag, so both must be rejected. + // The count does not depend on the flag, so both directions must be rejected. for case_insensitive in [true, false] { let err = Regex::new(&pattern, case_insensitive).expect_err("must be rejected"); assert!( - matches!(err, RegexCompilationError::PatternTooExpensive { .. }), - "expected PatternTooExpensive, got {err:?}" + matches!(err, RegexCompilationError::TooManyCharacterClasses { .. }), + "expected TooManyCharacterClasses, got {err:?}" ); } } - /// The guard has to charge classes nested in a bracketed class too, else `[\p{L}]` repeated - /// evades it while costing the same as `\p{L}` repeated. + /// Case folding walks a range's whole span, so a wide one buys as many Unicode ranges per + /// pattern byte as `\p{...}` does. It has to count as a class, not as the two codepoints it + /// spells. #[mz_ore::test] #[cfg_attr(miri, ignore)] // too slow - fn regex_bracketed_class_is_charged() { + fn regex_wide_bracketed_range_counts_as_a_class() { + let unit = r"[a-\x{2FFF}]"; + let pattern = unit.repeat(MAX_REGEX_SIZE_BEFORE_COMPILATION / unit.len()); + // `(?i)` reaches the same folding path from inside the pattern, so the flag we build with + // must make no difference. + for case_insensitive in [true, false] { + let err = Regex::new(&pattern, case_insensitive).expect_err("must be rejected"); + assert!( + matches!(err, RegexCompilationError::TooManyCharacterClasses { .. }), + "expected TooManyCharacterClasses, got {err:?}" + ); + } + } + + /// Nested classes have to count too, else `[\p{L}]` evades the limit while costing what + /// `\p{L}` costs. + #[mz_ore::test] + #[cfg_attr(miri, ignore)] // too slow + fn regex_bracketed_class_is_counted() { let unit = r"[\p{L}]"; let pattern = unit.repeat(MAX_REGEX_SIZE_BEFORE_COMPILATION / unit.len()); let err = Regex::new(&pattern, true).expect_err("must be rejected"); assert!( - matches!(err, RegexCompilationError::PatternTooExpensive { .. }), - "expected PatternTooExpensive, got {err:?}" + matches!(err, RegexCompilationError::TooManyCharacterClasses { .. }), + "expected TooManyCharacterClasses, got {err:?}" ); } - /// The budget has to admit as well as reject. A pattern one class short of it must still reach - /// the compiler, and one class past it must not, so that the limit lands where the constants - /// say it does rather than somewhere earlier. + /// The limit has to admit as well as reject: a pattern at the limit reaches the compiler, one + /// class past it does not. #[mz_ore::test] - #[cfg_attr(miri, ignore)] // Compiling tens of thousands of classes is far too slow under miri. - fn regex_budget_boundary_is_where_the_constants_put_it() { + #[cfg_attr(miri, ignore)] // Compiling thousands of classes is far too slow under miri. + fn regex_class_limit_boundary_is_where_the_constant_puts_it() { let unit = r"\p{L}"; - // The concat node holding the classes is charged too, hence the extra node. - let classes = - (MAX_REGEX_COMPILE_MEMORY - COMPILE_MEMORY_PER_NODE) / COMPILE_MEMORY_PER_CLASS; - assert!(unit.len() * (classes + 1) <= MAX_REGEX_SIZE_BEFORE_COMPILATION); - - // Case-sensitive on purpose. The estimate does not depend on the flag, but the compile - // this drives all the way into the NFA compiler does, and folding costs many times the - // memory for no extra coverage here. - let err = Regex::new(&unit.repeat(classes), false).expect_err("must be rejected"); assert!( - matches!(err, RegexCompilationError::RegexError(_)), - "one class under the budget must reach the compiler, got {err:?}" + unit.len() * (MAX_REGEX_CHARACTER_CLASSES + 1) <= MAX_REGEX_SIZE_BEFORE_COMPILATION + ); + + // Case-sensitive on purpose: this drives a real compile, and folding costs many times the + // memory for no extra coverage. Whether that compile succeeds or hits `size_limit` is not + // this test's business, only that our own limit lets it through. + let at_limit = Regex::new(&unit.repeat(MAX_REGEX_CHARACTER_CLASSES), false); + assert!( + !matches!( + at_limit, + Err(RegexCompilationError::TooManyCharacterClasses { .. }) + ), + "a pattern at the limit must reach the compiler, got {at_limit:?}" ); - let err = Regex::new(&unit.repeat(classes + 1), false).expect_err("must be rejected"); + let err = Regex::new(&unit.repeat(MAX_REGEX_CHARACTER_CLASSES + 1), false) + .expect_err("must be rejected"); assert!( - matches!(err, RegexCompilationError::PatternTooExpensive { .. }), - "one class over the budget must be turned away, got {err:?}" + matches!(err, RegexCompilationError::TooManyCharacterClasses { .. }), + "one class over the limit must be turned away, got {err:?}" ); } - /// The budget must not cost legitimate long patterns. A plain literal is orders of magnitude - /// cheaper per node than a Unicode class, so a large pattern of them still has to compile. + /// Literals carry no character classes, so a large pattern of them is bounded by its bytes + /// alone and still has to compile. #[mz_ore::test] #[cfg_attr(miri, ignore)] // too slow fn regex_long_literal_pattern_still_compiles() { // A long alternation of literals, the shape a generated pattern takes. Kept at 12000 - // branches: past that the compiled NFA runs into MAX_REGEX_SIZE_AFTER_COMPILATION, which - // would make this pass or fail for an unrelated reason. + // branches, past which the NFA runs into MAX_REGEX_SIZE_AFTER_COMPILATION instead. let pattern = vec!["abcdefgh"; 12_000].join("|"); assert!(pattern.len() > 100 * 1024); + assert_eq!(count_character_classes(&pattern), Some(0)); assert!(Regex::new(&pattern, true).is_ok()); } - /// `COMPILE_MEMORY_PER_NODE` has to stay low enough that the largest pattern - /// `MAX_REGEX_SIZE_BEFORE_COMPILATION` admits still fits the budget when every byte is a cheap - /// node. Otherwise the two limits collide and the byte limit becomes unreachable, silently - /// tightening what users can submit. - #[mz_ore::test] - #[cfg_attr(miri, ignore)] // too slow - fn regex_cheap_nodes_do_not_collide_with_the_byte_limit() { - let pattern = "a".repeat(MAX_REGEX_SIZE_BEFORE_COMPILATION); - assert!( - estimate_compile_memory(&pattern).unwrap() <= MAX_REGEX_COMPILE_MEMORY, - "a pattern of single-byte nodes at the byte limit must fit the memory budget" - ); - } - - /// The charges are calibrated against the `regex-syntax` version we pin, so hold them against - /// what that version actually allocates, one case per cost tier. A bump that makes a node kind - /// pricier then fails here, rather than silently reopening the OOM vector the budget closes. + /// The limits bound a compile only if the per-item ceilings below hold, and those are facts + /// about the pinned `regex-syntax`, so measure them. A bump that makes a node kind pricier + /// fails here. /// - /// Costs are measured here rather than written down, so nothing to keep in sync: the failure - /// message reports what a kind now costs against what it is charged, which is what a bump - /// needs in order to decide whether the charge or the case list has to move. + /// Each case also pins how [`count_character_classes`] classifies its kind, the one judgement + /// the limits rest on. A kind left uncounted while it can expand without bound shows up as the + /// measured cost outgrowing the bytes that bought it. #[mz_ore::test] #[cfg_attr(miri, ignore)] // Unicode class translation is far too slow under miri. - fn regex_compile_memory_charges_cover_measured_cost() { - /// Enough class nodes for their own cost to dominate a compile's fixed cost, few enough - /// that a case stays within a few tens of megabytes. - const CLASS_NODES: usize = 150; - /// Cheap nodes cost orders of magnitude less each, so they need proportionally more + fn regex_two_limits_bound_what_a_compile_spends() { + /// Ceiling on the heap spent translating one character class. However wide a range is, it + /// keeps at most one range per entry of the simple case-fold table, so it stays under this. + const MAX_MEMORY_PER_CHARACTER_CLASS: usize = 96 * 1024; + /// Ceiling on the heap spent translating one node that is not a character class. A literal + /// under `i` is the costliest, since it folds to a small class rather than staying a byte. + const MAX_MEMORY_PER_AST_NODE: usize = 768; + + /// Enough repetitions for a class's own cost to dominate a compile's fixed cost, few + /// enough that a case stays within a few tens of megabytes. + const CLASS_UNITS: usize = 150; + /// A non-class node costs orders of magnitude less, so it needs proportionally more /// repetitions to clear that same bar. - const CHEAP_NODES: usize = 10_000; - - let cases: &[(&str, bool, usize)] = &[ - // Unicode-property and Perl classes, charged `COMPILE_MEMORY_PER_CLASS`. - // `\p{Grapheme_Base}` under `i` is the costliest of these we have found. - (r"\p{Grapheme_Base}", true, CLASS_NODES), - (r"\p{XID_Continue}", true, CLASS_NODES), - (r"\p{Alphabetic}", true, CLASS_NODES), - (r"\p{L}", true, CLASS_NODES), - (r"\p{L}", false, CLASS_NODES), - (r"\w", true, CLASS_NODES), - (r"\W", true, CLASS_NODES), - (r"\d", true, CLASS_NODES), - (r"\s", true, CLASS_NODES), - // The same classes nested in a bracketed class, where each item is charged separately. - (r"[\p{L}]", true, CLASS_NODES), - (r"[a\p{L}\d]", true, CLASS_NODES), - // Everything else, charged `COMPILE_MEMORY_PER_NODE`. A literal under `i` is the - // costliest of these, since it folds to a small class rather than staying a byte. - ("a", true, CHEAP_NODES), - ("a", false, CHEAP_NODES), - (".", true, CHEAP_NODES), - ("[a-c]", true, CHEAP_NODES), - ("(a)", false, CHEAP_NODES), - ("a|", false, CHEAP_NODES), - ("a*", false, CHEAP_NODES), - ("a{2}", false, CHEAP_NODES), - ("^", false, CHEAP_NODES), + const CHEAP_UNITS: usize = 10_000; + + // (unit, case_insensitive, repetitions, character classes per repetition) + let cases: &[(&str, bool, usize, usize)] = &[ + // Unicode-property and Perl classes. `\p{Grapheme_Base}` under `i` is the costliest + // single class found, so it is what sizes `MAX_MEMORY_PER_CHARACTER_CLASS`. + (r"\p{Grapheme_Base}", true, CLASS_UNITS, 1), + (r"\p{XID_Continue}", true, CLASS_UNITS, 1), + (r"\p{Alphabetic}", true, CLASS_UNITS, 1), + (r"\p{L}", true, CLASS_UNITS, 1), + (r"\p{L}", false, CLASS_UNITS, 1), + (r"\w", true, CLASS_UNITS, 1), + (r"\W", true, CLASS_UNITS, 1), + (r"\d", true, CLASS_UNITS, 1), + (r"\s", true, CLASS_UNITS, 1), + // The same nested in a bracketed class, where every item counts on its own. + (r"[\p{L}]", true, CLASS_UNITS, 1), + (r"[a\p{L}\d]", true, CLASS_UNITS, 2), + // Ranges, from one too narrow to reach the case-fold table up to one covering all of + // it, including the stretch of Latin where it is densest. Span alone does not predict + // the cost: `[\x{100}-\x{17F}]` costs 2.5x what the equally wide `[\x00-\x7F]` does, + // which is why ranges are counted rather than priced. + (r"[a-\x{10FFFF}]", true, CLASS_UNITS, 1), + (r"[^a-\x{10FFFF}]", true, CLASS_UNITS, 1), + (r"[a-\x{2FFF}]", true, CLASS_UNITS, 1), + (r"[\x{100}-\x{250}]", true, CLASS_UNITS, 1), + (r"[\x{100}-\x{17F}]", true, CLASS_UNITS, 1), + (r"[a-\x{FF}]", true, CLASS_UNITS, 1), + (r"[a-z]", true, CLASS_UNITS, 1), + (r"[[:alpha:]]", true, CLASS_UNITS, 1), + // Nodes carrying no class, bounded by the byte limit alone. `[abc]` is here on + // purpose: a bracketed class of literals is not counted, so its cost has to stay + // within what its bytes buy. + ("a", true, CHEAP_UNITS, 0), + ("a", false, CHEAP_UNITS, 0), + (".", true, CHEAP_UNITS, 0), + ("[abc]", true, CHEAP_UNITS, 0), + ("(a)", false, CHEAP_UNITS, 0), + ("a|", false, CHEAP_UNITS, 0), + ("a*", false, CHEAP_UNITS, 0), + ("a{2}", false, CHEAP_UNITS, 0), + ("^", false, CHEAP_UNITS, 0), ]; - for (unit, case_insensitive, repeats) in cases { + for (unit, case_insensitive, repeats, classes_per_unit) in cases { let pattern = unit.repeat(*repeats); - let charged = estimate_compile_memory(&pattern).expect("pattern parses"); + let expected_classes = repeats * classes_per_unit; + let classes = count_character_classes(&pattern).expect("pattern parses"); + assert_eq!( + classes, expected_classes, + "`{unit}` x{repeats} counts as {classes} character classes, expected \ + {expected_classes}" + ); + + // The same product the two limits bound, evaluated for this pattern. Pattern bytes + // stand in for the node count, which they bound. + let bound = + classes * MAX_MEMORY_PER_CHARACTER_CLASS + pattern.len() * MAX_MEMORY_PER_AST_NODE; let measured = peak_translate_bytes(&pattern, *case_insensitive); assert!( - measured <= charged, + measured <= bound, "`{unit}` x{repeats} (case_insensitive: {case_insensitive}) allocated {measured} \ - bytes, above the {charged} bytes charged for it, so the budget no longer bounds \ - what a compile spends" + bytes, above the {bound} bytes the limits allow it, so they no longer bound what \ + a compile spends" ); } } - /// A counted repetition needs no budget of its own: it stays small in the AST, and the NFA - /// compiler's incremental `size_limit` check rejects it. Pin that, since the budget - /// deliberately does not multiply by repetition bounds. + /// A counted repetition stays one class in the AST, and the NFA compiler's incremental + /// `size_limit` check rejects it. Pin that, since the count does not multiply by repetition + /// bounds. #[mz_ore::test] #[cfg_attr(miri, ignore)] // too slow fn regex_counted_repetition_rejected_by_size_limit() { diff --git a/test/sqllogictest/regex.slt b/test/sqllogictest/regex.slt index 30a9688ccdd40..c5dd7e1d2dbc4 100644 --- a/test/sqllogictest/regex.slt +++ b/test/sqllogictest/regex.slt @@ -855,21 +855,33 @@ HINT: No function matches the given name and argument types. You might need to statement error invalid regular expression: regex pattern too large \(62400000 bytes, max 1048576 bytes\) SELECT 'aaaaaaaaaaa' ~ repeat('vxx.0.0-rc.4 (5b079d80c)', '2600000'); -# A class-heavy pattern stays far under the byte limit above while costing gigabytes to compile, -# an order of magnitude more heap per pattern byte than a plain literal costs. It has to be -# rejected on the projected cost instead. This is the shape that made any `SELECT` an OOM vector. -# Where exactly the budget cuts off is left to the unit tests in `mz_repr::adt::regex`, which can -# derive it from the constants instead of restating one here. -statement error invalid regular expression: regex pattern too expensive to compile \(needs about \d+ bytes, max \d+ bytes\); reduce the number of character classes +# A class-heavy pattern stays far under the byte limit while costing gigabytes to compile, so the +# number of character classes is limited separately. Where the limit cuts off is left to the unit +# tests, which derive it from the constant instead of restating one here. +statement error invalid regular expression: regex pattern has too many character classes \(\d+, max \d+\) SELECT 'x' ~* repeat(chr(92) || 'p{L}', 209715); -# The cost is per character class, not per byte, so a pattern of the same length built from plain -# literals must keep working. Kept below 12000 branches, past which the compiled NFA runs into the -# separate 10 MiB size limit and stops being a test of this one. +# Case folding walks a range's whole span, so `[a-\x{2FFF}]` buys as many Unicode ranges as a +# `\p{...}` class does, and counts as one. +statement error invalid regular expression: regex pattern has too many character classes \(\d+, max \d+\) +SELECT 'x' ~* repeat('[a-' || chr(92) || 'x{2FFF}]', 87381); + +# An inline `(?i)` reaches the same folding path, so a case-sensitive operator is guarded too. +statement error invalid regular expression: regex pattern has too many character classes \(\d+, max \d+\) +SELECT 'x' ~ ('(?i)' || repeat('[a-' || chr(92) || 'x{2FFF}]', 50000)); + +# Plain literals carry no classes, so a pattern of the same length must keep working. Kept below +# 12000 branches, past which the NFA runs into the separate 10 MiB size limit instead. query B SELECT 'abcdefgh' ~* (repeat('abcdefgh|', 12000) || 'x'); ---- true +# Ranges count as classes however narrow, so ordinary use has to stay well inside the limit. +query B +SELECT repeat('a', 5000) ~* repeat('[a-z]', 1500); +---- +true + statement ok DROP CLUSTER multiprocess; From 40a97b1b4071ec7d4b2b850c4729a8b3dd03a56c Mon Sep 17 00:00:00 2001 From: "claude[bot]" <209825114+claude[bot]@users.noreply.github.com> Date: Thu, 13 Aug 2026 16:34:45 +0200 Subject: [PATCH 6/8] compute: default arrangement dictionary compression gate on (#38195) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _Requested by **Moritz Hoffmann** · [Slack thread](https://materializeinc.slack.com/archives/C08AENAP691/p1786626884672969?thread_ts=1786626884.672969&cid=C08AENAP691)_ **Before:** a self-managed deployment had arrangement dictionary compression off even on a cluster with `EXPERIMENTAL ARRANGEMENT COMPRESSION` set. The environment-wide gate defaults off, and self-managed has no LaunchDarkly to flip it, so there was no way to turn the feature on at all. **After:** the environment-wide gate defaults on, so the per-cluster `EXPERIMENTAL ARRANGEMENT COMPRESSION` option is the only knob. This also aligns the out-of-the-box default with cloud, where LaunchDarkly already serves `true`. **How:** the compiled-in `default` of the `enable_arrangement_dictionary_compression_alpha` dyncfg (`ENABLE_ARRANGEMENT_DICTIONARY_COMPRESSION_ALPHA` in `src/compute-types/src/dyncfgs.rs`) changes from `false` to `true`. That is the whole functional change; the surrounding doc comment is updated to record why the default is on. The gate is consumed in `ComputeController::add_replica` (`src/compute-client/src/controller.rs`), where it is ANDed with the per-cluster `config.arrangement_compression` to produce `ReplicaConfig::arrangement_dictionary_compression`. The AND is untouched, so a cluster that does not request compression still gets none. **Caveat:** the environment-wide gate is read once at replica creation and is not part of `ReplicaShape`, so existing replicas do not pick up the change until they are recreated (restart, resize, or cluster reconfiguration). ### Motivation Self-managed customers cannot enable arrangement dictionary compression today. The feature is gated by two independent switches that are ANDed together at replica creation: the environment-wide dyncfg `enable_arrangement_dictionary_compression_alpha`, and the per-cluster SQL option `EXPERIMENTAL ARRANGEMENT COMPRESSION`. In cloud the dyncfg is set to `true` through LaunchDarkly, so only the cluster option matters in practice. Self-managed has no LaunchDarkly and therefore falls back to the compiled-in default, which is `false`, leaving the AND permanently false regardless of how the cluster is configured. Flipping the compiled-in default makes the cluster option the single knob everywhere. Requested in the Slack thread linked above. ### Description One-line functional change in `src/compute-types/src/dyncfgs.rs`: ```rust pub const ENABLE_ARRANGEMENT_DICTIONARY_COMPRESSION_ALPHA: Config = Config::new( "enable_arrangement_dictionary_compression_alpha", true, // was: false "Enable arrangement dictionary compression (alpha; not yet production-ready).", ); ``` The `_alpha` name and the description are deliberately left alone: this changes the default of the gate, not the maturity of the feature, and the flag remains a kill switch that can be set to `false` to disable compression on new or restarted replicas regardless of per-cluster configuration. The doc comment previously asserted "default off" as part of its disposition note, which would now be wrong, so it is corrected and gains a short note on why the compiled-in default is the one that matters for a deployment without LaunchDarkly. Notes for the reviewer: - **Existing replicas are unaffected.** `ComputeController::add_replica` captures the dyncfg once and stores it in `ReplicaConfig`, and the value is not part of `ReplicaShape`, so no reconciliation is triggered by the flip. Only replicas created after the upgrade see the new default. - **The per-cluster option default is unchanged.** `EXPERIMENTAL ARRANGEMENT COMPRESSION` still defaults to `false` (`src/sql/src/plan/statement/ddl.rs`, `experimental_arrangement_compression.unwrap_or(false)`), so no existing cluster starts compressing as a result of this PR. Compression is still strictly opt-in per cluster. - **CI behavior is unchanged.** `misc/python/materialize/mzcompose/__init__.py` pins `enable_arrangement_dictionary_compression_alpha` to `"false"` in the shared mzcompose settings, so the default flip does not silently turn compression on across the mzcompose test suites. That pin is left as-is in this PR; whether it should now be dropped, or flipped to `true` per the repo guidance that feature flags default on in test configuration, is a separate decision. ### Verification No test expectations depend on this default: - `test/testdrive/arrangement-dictionary-compression.td` sets the flag explicitly with `ALTER SYSTEM SET ... = true` before creating its cluster, so it is insensitive to the default. Its trailing `ALTER SYSTEM RESET` now resets to `true` rather than `false`, which is only cleanup at the end of the file. - The `.slt` expectations that mention `EXPERIMENTAL ARRANGEMENT COMPRESSION = false` (`test/sqllogictest/cluster.slt`, `managed_cluster.slt`, `materialized_views.slt`) render the per-cluster option, not the dyncfg, and that option's default is unchanged. - No Rust unit test asserts this dyncfg's default; the `clusterd-test-driver` cases in `src/clusterd-test-driver/src/text.rs` pass the value explicitly. - `misc/python/materialize/parallel_workload/action.py` already flips this flag across both values, so it keeps exercising both paths. `cargo check` could not be run in this environment (a git dependency, `MaterializeInc/async-compression`, is outside the sandbox's authorized repository set, so dependency resolution fails before compiling). `rustfmt --check` on the touched file is clean. The change is a `bool` literal in a `const` initializer plus a doc comment, so compilation risk is nil; CI will confirm. One thing worth a second pair of eyes: `test/launchdarkly-flag-consistency` fails when a flag's production LaunchDarkly default differs from the compiled-in default unless the flag is allowlisted, and `enable_arrangement_dictionary_compression_alpha` currently appears in neither `INTENTIONAL_LD_OVERRIDES` nor `KNOWN_CROSS_ENV_DIVERGENCES`. If production LaunchDarkly serves `true` as its flag default, this PR removes a discrepancy and the check stays green. If cloud instead has it on via a targeting rule while the served default is still `false`, this PR creates a new discrepancy there. Someone with LaunchDarkly access should confirm which it is. ### Note on `launchdarkly-flag-consistency` That check lives in the Nightly pipeline (`ci/nightly/pipeline.template.yml`), not the `test` pipeline, so it did not run on this PR. It compares a flag's compiled-in default against the production LaunchDarkly default and allowlists known divergences. This flag appears in neither `INTENTIONAL_LD_OVERRIDES` nor `KNOWN_CROSS_ENV_DIVERGENCES`. If LaunchDarkly's *served* default is already `true`, this change removes a divergence and Nightly stays green. If cloud is enabled via a targeting rule while the served default is still `false`, this change introduces one, and Nightly will need either the LaunchDarkly default updated or an allowlist entry. --- _Generated by [Claude Code](https://claude.ai/code/session_015y1vTXBhMNwFSinPccnZCZ)_ --------- Co-authored-by: Claude --- src/compute-types/src/dyncfgs.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs index 42cd46753beba..19ca6b4cd02fb 100644 --- a/src/compute-types/src/dyncfgs.rs +++ b/src/compute-types/src/dyncfgs.rs @@ -449,11 +449,11 @@ pub const COMPUTE_LOGICAL_BACKPRESSURE_INFLIGHT_SLACK: Config = Config /// production-ready, and the name is meant to make that unmissable at the /// `ALTER SYSTEM SET` call site rather than relying on out-of-band warnings. /// -/// Disposition: added 2026-06-09, default off; solicit feedback for one month -/// and remove in the absence of a positive response. +/// Disposition: added 2026-06-09; solicit feedback for one month and remove in +/// the absence of a positive response. pub const ENABLE_ARRANGEMENT_DICTIONARY_COMPRESSION_ALPHA: Config = Config::new( "enable_arrangement_dictionary_compression_alpha", - false, + true, "Enable arrangement dictionary compression (alpha; not yet production-ready).", ); From 77c7dfd06ff7f3d658d799b36ce8f39774c94d50 Mon Sep 17 00:00:00 2001 From: Dennis Felsing Date: Thu, 13 Aug 2026 17:09:46 +0200 Subject: [PATCH 7/8] console: Minor fixes to tutorial (#38192) Follow-up to #38067 --- console/src/platform/shell/Tutorial.tsx | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/console/src/platform/shell/Tutorial.tsx b/console/src/platform/shell/Tutorial.tsx index 5daec81dad754..3f41d1ad8f76b 100644 --- a/console/src/platform/shell/Tutorial.tsx +++ b/console/src/platform/shell/Tutorial.tsx @@ -53,7 +53,7 @@ import TutorialInsertionWidget from "./TutorialInsertionWidget"; import TutorialSchemaWidget from "./TutorialSchemaWidget"; /* Semantic version number used for instrumentation */ -const QUICKSTART_VERSION = "2.0.0"; +const QUICKSTART_VERSION = "2.1.0"; type RunnableProps = { runCommand: (value: string) => void; @@ -826,7 +826,7 @@ WHERE w2.amount > w1.amount runCommand={runCommand} value="CREATE INDEX wins_by_item_seller ON winning_bids (item, seller); CREATE INDEX wins_by_item_buyer ON winning_bids (item, buyer);" - title="View a sample row in flip_activities" + title="Create indexes on the winning_bids join keys" /> From 57107078dc6b54dcfd71a3b5f2619eecbfe2d096 Mon Sep 17 00:00:00 2001 From: mtabebe Date: Thu, 13 Aug 2026 12:37:51 -0400 Subject: [PATCH 8/8] catalog: convert the sink builtin tables to materialized views (SQL-493) (#38049) Problem: mz_sinks, mz_kafka_sinks and mz_iceberg_sinks are builtin tables that pack_sink_update fills on the leader environmentd, on every catalog change. Getting rid of those writes is the point of this whole series ([SQL-118](https://linear.app/materializeinc/issue/SQL-118) / multi-envd), and the sink family is the last item in M1. Everything the three tables hold is already sitting in the persisted create_sql, so none of it has to be packed by Rust in the first place. Note mz_continual_tasks is named in the issue but no longer exists. The CT feature was removed in #35967, and the v81_to_v82 migration asserts no CT items remain, so there is nothing there to convert. The Linear description wants correcting. Solution: Two parts. parse_catalog_create_sql gets a CreateSink arm, alongside the CreateSource one it mirrors. It emits sink_type, cluster_id, connection_id, envelope_type, format/key_format/value_format, the kafka topic, and the iceberg namespace/table. The deprecated format collapse stays in Rust rather than moving into the view SQL, because its edges are fiddly and here they are unit-testable. Only avro/avro and json/json collapse to a single name, and a key format only exists once the sink declares a KEY, which is easy to get wrong reading the statement alone. Then all three tables become BuiltinMaterializedViews over mz_catalog_raw and pack_sink_update goes away, since packing them was its only job. Each view keeps its columns, order, nullability, keys, comments, access and ontology as they were, and reuses its old OID so pg identity does not move. No privileges column, unlike what mz_sources grew, to keep this a straight parity conversion. - three MigrationStep::replacement entries at the current dev version. - the old 0.160.0 step naming mz_sinks as a Table had to go with them: it no longer resolves, and validate_migration_steps panics on that at catalog open. Since that only reproduces against a catalog already on disk, there is now a unit test asserting every step resolves. Testing: - New test/sqllogictest/mz_sinks.slt pins the column shape of all three views, that they are materialized views, that non-sink items are excluded, and the JSON keys the view SQL expects back from the helper. Sinks cannot be created in sqllogictest at all, since purification always talks to the broker, so derived values are not covered here. - mz-sinks.td picks up what nothing asserted before: key_format on a keyed bare format, the composite key-text-value-text form, and mz_kafka_sinks.topic. - test/iceberg/mode-append.td picks up envelope_type = append and mz_iceberg_sinks.namespace/table. The iceberg path had no catalog coverage. - Unit tests in jsonb.rs for the new arm: bare and key/value formats with and without a KEY, all three envelope values, the avro/json collapse against text/text and text/bytes, a missing TOPIC and a missing TABLE, and a non-sink statement as a regression guard. - The tables.td "cannot insert into system table" assertions move off mz_kafka_sinks and onto mz_storage_usage_by_shard, which the migration validator refuses to convert. --------- Co-authored-by: Claude Opus 5 --- .../checks/all_checks/builtin_version_pin.py | 12 +- .../src/catalog/builtin_table_updates.rs | 100 +----- .../catalog/open/builtin_schema_migration.rs | 29 +- .../open/builtin_schema_migration_tests.rs | 15 + src/catalog/src/builtin.rs | 6 +- src/catalog/src/builtin/mz_catalog.rs | 211 ++++++++---- src/catalog/src/memory/objects.rs | 11 + src/expr/src/scalar/func/impls/jsonb.rs | 307 +++++++++++++++++- src/pgrepr-consts/src/oid.rs | 6 +- src/storage-types/src/sinks.rs | 8 + test/iceberg/mode-append.td | 15 + test/sqllogictest/catalog_server_explain.slt | 49 ++- .../information_schema_tables.slt | 6 +- test/sqllogictest/mz_sinks.slt | 212 ++++++++++++ test/testdrive-old-kafka-src-syntax/tables.td | 7 +- test/testdrive/catalog.td | 8 +- test/testdrive/mz-sinks.td | 34 ++ test/testdrive/tables.td | 7 +- 18 files changed, 853 insertions(+), 190 deletions(-) create mode 100644 test/sqllogictest/mz_sinks.slt diff --git a/misc/python/materialize/checks/all_checks/builtin_version_pin.py b/misc/python/materialize/checks/all_checks/builtin_version_pin.py index a1cacca83b2e2..081bf28175519 100644 --- a/misc/python/materialize/checks/all_checks/builtin_version_pin.py +++ b/misc/python/materialize/checks/all_checks/builtin_version_pin.py @@ -36,7 +36,7 @@ _STABLE_BUILTINS: dict[str, MzVersion | None] = { "mz_array_types": None, "mz_audit_events": MzVersion.parse_mz("v26.33.0"), - "mz_aws_privatelink_connections": None, + "mz_aws_privatelink_connections": MzVersion.parse_mz("v26.37.0"), "mz_base_types": None, "mz_cluster_replica_sizes": None, "mz_cluster_replicas": MzVersion.parse_mz("v26.30.0"), @@ -45,17 +45,17 @@ "mz_default_privileges": MzVersion.parse_mz("v26.30.0"), "mz_egress_ips": None, "mz_functions": None, - "mz_iceberg_sinks": None, + "mz_iceberg_sinks": MzVersion.parse_mz("v26.38.0"), "mz_index_columns": None, - "mz_kafka_connections": None, - "mz_kafka_sinks": None, + "mz_kafka_connections": MzVersion.parse_mz("v26.37.0"), + "mz_kafka_sinks": MzVersion.parse_mz("v26.38.0"), "mz_kafka_sources": MzVersion.parse_mz("v26.34.0"), "mz_list_types": None, "mz_map_types": None, "mz_operators": None, "mz_pseudo_types": None, - "mz_sinks": None, - "mz_ssh_tunnel_connections": None, + "mz_sinks": MzVersion.parse_mz("v26.38.0"), + "mz_ssh_tunnel_connections": MzVersion.parse_mz("v26.37.0"), "mz_system_privileges": MzVersion.parse_mz("v26.30.0"), "mz_tables": None, "mz_types": None, diff --git a/src/adapter/src/catalog/builtin_table_updates.rs b/src/adapter/src/catalog/builtin_table_updates.rs index 24ea3082ccf43..709cc682ae7f5 100644 --- a/src/adapter/src/catalog/builtin_table_updates.rs +++ b/src/adapter/src/catalog/builtin_table_updates.rs @@ -17,16 +17,16 @@ use mz_catalog::SYSTEM_CONN_ID; use mz_catalog::builtin::{ BuiltinTable, MZ_AGGREGATES, MZ_ARRAY_TYPES, MZ_BASE_TYPES, MZ_CLUSTER_REPLICA_SIZE_INTERNAL, MZ_CLUSTER_REPLICA_SIZES, MZ_COLUMNS, MZ_EGRESS_IPS, MZ_FUNCTIONS, - MZ_HISTORY_RETENTION_STRATEGIES, MZ_ICEBERG_SINKS, MZ_INDEX_COLUMNS, MZ_KAFKA_SINKS, - MZ_LICENSE_KEYS, MZ_LIST_TYPES, MZ_MAP_TYPES, MZ_MATERIALIZED_VIEW_REFRESH_STRATEGIES, - MZ_OBJECT_DEPENDENCIES, MZ_OBJECT_GLOBAL_IDS, MZ_OPERATORS, MZ_PSEUDO_TYPES, MZ_REPLACEMENTS, - MZ_ROLE_AUTH, MZ_SESSIONS, MZ_SINKS, MZ_SOURCE_REFERENCES, MZ_STORAGE_USAGE_BY_SHARD, - MZ_SUBSCRIPTIONS, MZ_TABLES, MZ_TYPE_PG_METADATA, MZ_TYPES, MZ_VIEWS, MZ_WEBHOOKS_SOURCES, + MZ_HISTORY_RETENTION_STRATEGIES, MZ_INDEX_COLUMNS, MZ_LICENSE_KEYS, MZ_LIST_TYPES, + MZ_MAP_TYPES, MZ_MATERIALIZED_VIEW_REFRESH_STRATEGIES, MZ_OBJECT_DEPENDENCIES, + MZ_OBJECT_GLOBAL_IDS, MZ_OPERATORS, MZ_PSEUDO_TYPES, MZ_REPLACEMENTS, MZ_ROLE_AUTH, + MZ_SESSIONS, MZ_SOURCE_REFERENCES, MZ_STORAGE_USAGE_BY_SHARD, MZ_SUBSCRIPTIONS, MZ_TABLES, + MZ_TYPE_PG_METADATA, MZ_TYPES, MZ_VIEWS, MZ_WEBHOOKS_SOURCES, }; use mz_catalog::durable::SourceReferences; use mz_catalog::memory::error::Error; use mz_catalog::memory::objects::{ - CatalogEntry, CatalogItem, DataSourceDesc, Func, Index, MaterializedView, Sink, Table, + CatalogEntry, CatalogItem, DataSourceDesc, Func, Index, MaterializedView, Table, TableDataSource, Type, View, }; use mz_expr::MirScalarExpr; @@ -50,7 +50,6 @@ use mz_sql::func::FuncImplCatalogDetails; use mz_sql::names::SchemaSpecifier; use mz_sql_parser::ast::display::AstDisplay; use mz_storage_client::client::TableData; -use mz_storage_types::sinks::{IcebergSinkConnection, KafkaSinkConnection, StorageSinkConnection}; use smallvec::smallvec; // DO NOT add any more imports from `crate` outside of `crate::catalog`. @@ -192,9 +191,9 @@ impl CatalogState { CatalogItem::MaterializedView(mview) => { self.pack_materialized_view_update(id, mview, diff) } - CatalogItem::Sink(sink) => { - self.pack_sink_update(id, oid, schema_id, name, owner_id, sink, diff) - } + // mz_sinks, mz_kafka_sinks and mz_iceberg_sinks read create_sql + // out of mz_catalog_raw, so there is nothing to pack here. + CatalogItem::Sink(_) => vec![], CatalogItem::Type(ty) => { self.pack_type_update(id, oid, schema_id, name, owner_id, privileges, ty, diff) } @@ -514,87 +513,6 @@ impl CatalogState { updates } - fn pack_sink_update( - &self, - id: CatalogItemId, - oid: u32, - schema_id: &SchemaSpecifier, - name: &str, - owner_id: &RoleId, - sink: &Sink, - diff: Diff, - ) -> Vec> { - let mut updates = vec![]; - match &sink.connection { - StorageSinkConnection::Kafka(KafkaSinkConnection { - topic: topic_name, .. - }) => { - updates.push(BuiltinTableUpdate::row( - &*MZ_KAFKA_SINKS, - Row::pack_slice(&[ - Datum::String(&id.to_string()), - Datum::String(topic_name.as_str()), - ]), - diff, - )); - } - StorageSinkConnection::Iceberg(IcebergSinkConnection { - namespace, table, .. - }) => { - updates.push(BuiltinTableUpdate::row( - &*MZ_ICEBERG_SINKS, - Row::pack_slice(&[ - Datum::String(&id.to_string()), - Datum::String(namespace.as_str()), - Datum::String(table.as_str()), - ]), - diff, - )); - } - }; - - let create_stmt = mz_sql::parse::parse(&sink.create_sql) - .unwrap_or_else(|_| panic!("create_sql cannot be invalid: {}", sink.create_sql)) - .into_element() - .ast; - - let envelope = sink.envelope(); - - // The combined format string is used for the deprecated `format` column. - let combined_format = sink.combined_format(); - let (key_format, value_format) = match sink.formats() { - Some((key_format, value_format)) => (key_format, Some(value_format)), - None => (None, None), - }; - - updates.push(BuiltinTableUpdate::row( - &*MZ_SINKS, - Row::pack_slice(&[ - Datum::String(&id.to_string()), - Datum::UInt32(oid), - Datum::String(&schema_id.to_string()), - Datum::String(name), - Datum::String(sink.connection.name()), - Datum::from(sink.connection_id().map(|id| id.to_string()).as_deref()), - // size column now deprecated w/o linked clusters - Datum::Null, - Datum::from(envelope), - // FIXME: These key/value formats are kinda leaky! Should probably live in - // the kafka sink table. - Datum::from(combined_format.as_ref().map(|f| f.as_ref())), - Datum::from(key_format), - Datum::from(value_format), - Datum::String(&sink.cluster_id.to_string()), - Datum::String(&owner_id.to_string()), - Datum::String(&sink.create_sql), - Datum::String(&create_stmt.to_ast_string_redacted()), - ]), - diff, - )); - - updates - } - fn pack_index_update( &self, id: CatalogItemId, diff --git a/src/adapter/src/catalog/open/builtin_schema_migration.rs b/src/adapter/src/catalog/open/builtin_schema_migration.rs index 58b4ec995a88a..6d89904ba0724 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration.rs @@ -101,12 +101,6 @@ static MIGRATIONS: LazyLock> = LazyLock::new(|| { MZ_INTERNAL_SCHEMA, "mz_cluster_replica_metrics_history", ), - MigrationStep::replacement( - "0.160.0", - CatalogItemType::Table, - MZ_CATALOG_SCHEMA, - "mz_sinks", - ), MigrationStep::replacement( "26.18.0-dev.0", CatalogItemType::MaterializedView, @@ -355,6 +349,29 @@ static MIGRATIONS: LazyLock> = LazyLock::new(|| { MZ_CATALOG_SCHEMA, "mz_aws_privatelink_connections", ), + // The three sink tables became materialized views, which moves their + // fingerprints. The old `0.160.0` `mz_sinks` step had to go at the same + // time, because it names the `Table` description and that no longer + // resolves. Nothing is lost: anything that needed the old step upgrades + // from further back than this one, so this one covers it too. + MigrationStep::replacement( + "26.38.0-dev.0", + CatalogItemType::MaterializedView, + MZ_CATALOG_SCHEMA, + "mz_sinks", + ), + MigrationStep::replacement( + "26.38.0-dev.0", + CatalogItemType::MaterializedView, + MZ_CATALOG_SCHEMA, + "mz_kafka_sinks", + ), + MigrationStep::replacement( + "26.38.0-dev.0", + CatalogItemType::MaterializedView, + MZ_CATALOG_SCHEMA, + "mz_iceberg_sinks", + ), ] }); diff --git a/src/adapter/src/catalog/open/builtin_schema_migration_tests.rs b/src/adapter/src/catalog/open/builtin_schema_migration_tests.rs index eeacfea1c459f..787b887e8e6e1 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration_tests.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration_tests.rs @@ -402,3 +402,18 @@ fn init_persist(sim: &mut turmoil::Sim) -> PersistLocation { consensus_uri: "turmoil://consensus:7000".parse().unwrap(), } } + +/// A step naming a builtin that no longer exists, or that changed +/// `CatalogItemType`, panics `validate_migration_steps` at catalog open. That +/// needs a catalog already on disk to reproduce, so the default CI suite never +/// sees it and only the upgrade nightly does. Much cheaper to catch here. +#[mz_ore::test] +fn test_migration_steps_resolve_to_builtins() { + for step in MIGRATIONS.iter() { + assert!( + BUILTIN_LOOKUP.contains_key(&step.object), + "migration step for non-existent builtin: {:?}", + step.object + ); + } +} diff --git a/src/catalog/src/builtin.rs b/src/catalog/src/builtin.rs index 3dfb9e2ae8774..a29b875cf999f 100644 --- a/src/catalog/src/builtin.rs +++ b/src/catalog/src/builtin.rs @@ -1107,11 +1107,11 @@ pub static BUILTINS_STATIC: LazyLock>> = LazyLock::ne Builtin::Log(&MZ_COMPUTE_ERROR_COUNTS_RAW), Builtin::Log(&MZ_COMPUTE_HYDRATION_TIMES_PER_WORKER), Builtin::Log(&MZ_COMPUTE_OPERATOR_HYDRATION_STATUSES_PER_WORKER), - Builtin::Table(&MZ_KAFKA_SINKS), + Builtin::MaterializedView(&MZ_KAFKA_SINKS), Builtin::MaterializedView(&MZ_KAFKA_CONNECTIONS), Builtin::MaterializedView(&MZ_KAFKA_SOURCES), Builtin::Table(&MZ_OBJECT_DEPENDENCIES), - Builtin::Table(&MZ_ICEBERG_SINKS), + Builtin::MaterializedView(&MZ_ICEBERG_SINKS), Builtin::MaterializedView(&MZ_DATABASES), Builtin::MaterializedView(&MZ_SCHEMAS), Builtin::Table(&MZ_COLUMNS), @@ -1125,7 +1125,7 @@ pub static BUILTINS_STATIC: LazyLock>> = LazyLock::ne Builtin::MaterializedView(&MZ_MYSQL_SOURCE_TABLES), Builtin::MaterializedView(&MZ_SQL_SERVER_SOURCE_TABLES), Builtin::MaterializedView(&MZ_KAFKA_SOURCE_TABLES), - Builtin::Table(&MZ_SINKS), + Builtin::MaterializedView(&MZ_SINKS), Builtin::Table(&MZ_VIEWS), Builtin::Table(&MZ_TYPES), Builtin::Table(&MZ_TYPE_PG_METADATA), diff --git a/src/catalog/src/builtin/mz_catalog.rs b/src/catalog/src/builtin/mz_catalog.rs index c2fb6d2a75be2..719501805d533 100644 --- a/src/catalog/src/builtin/mz_catalog.rs +++ b/src/catalog/src/builtin/mz_catalog.rs @@ -198,69 +198,111 @@ pub const TYPE_MZ_ACL_ITEM_ARRAY: BuiltinType = BuiltinType { }, }; -pub static MZ_ICEBERG_SINKS: LazyLock = LazyLock::new(|| BuiltinTable { - name: "mz_iceberg_sinks", - schema: MZ_CATALOG_SCHEMA, - oid: oid::TABLE_MZ_ICEBERG_SINKS_OID, - desc: RelationDesc::builder() - .with_column("id", SqlScalarType::String.nullable(false)) - .with_column("namespace", SqlScalarType::String.nullable(false)) - .with_column("table", SqlScalarType::String.nullable(false)) - .finish(), - column_comments: BTreeMap::from_iter([ - ("id", "The ID of the sink."), - ( - "namespace", - "The namespace of the Iceberg table into which the sink is writing.", - ), - ("table", "The Iceberg table into which the sink is writing."), - ]), - is_retained_metrics_object: false, - access: vec![PUBLIC_SELECT], - ontology: Some(Ontology { - entity_name: "iceberg_sink", - description: "Iceberg-specific sink configuration (namespace, table)", - links: &const { - [OntologyLink { - name: "details_of", - target: "sink", - properties: LinkProperties::fk("id", "id", Cardinality::OneToOne), - }] - }, - column_semantic_types: &[("id", SemanticType::CatalogItemId)], - }), +pub static MZ_ICEBERG_SINKS: LazyLock = LazyLock::new(|| { + BuiltinMaterializedView { + name: "mz_iceberg_sinks", + schema: MZ_CATALOG_SCHEMA, + oid: oid::MV_MZ_ICEBERG_SINKS_OID, + desc: RelationDesc::builder() + .with_column("id", SqlScalarType::String.nullable(false)) + .with_column("namespace", SqlScalarType::String.nullable(false)) + .with_column("table", SqlScalarType::String.nullable(false)) + .finish(), + column_comments: BTreeMap::from_iter([ + ("id", "The ID of the sink."), + ( + "namespace", + "The namespace of the Iceberg table into which the sink is writing.", + ), + ("table", "The Iceberg table into which the sink is writing."), + ]), + // Planning requires `NAMESPACE` and `TABLE`, so both are always there. + sql: " +IN CLUSTER mz_catalog_server +WITH ( + ASSERT NOT NULL id, + ASSERT NOT NULL namespace, + ASSERT NOT NULL \"table\" +) AS +SELECT + mz_internal.parse_catalog_id(data->'key'->'gid') AS id, + parsed->>'namespace' AS namespace, + parsed->>'table' AS \"table\" +FROM + mz_internal.mz_catalog_raw + CROSS JOIN LATERAL ( + SELECT mz_internal.parse_catalog_create_sql(data->'value'->'definition'->'V1'->>'create_sql') + ) AS l(parsed) +WHERE + data->>'kind' = 'Item' AND + parsed->>'sink_type' = 'iceberg'", + is_retained_metrics_object: false, + access: vec![PUBLIC_SELECT], + ontology: Some(Ontology { + entity_name: "iceberg_sink", + description: "Iceberg-specific sink configuration (namespace, table)", + links: &const { + [OntologyLink { + name: "details_of", + target: "sink", + properties: LinkProperties::fk("id", "id", Cardinality::OneToOne), + }] + }, + column_semantic_types: &[("id", SemanticType::CatalogItemId)], + }), + } }); -pub static MZ_KAFKA_SINKS: LazyLock = LazyLock::new(|| BuiltinTable { - name: "mz_kafka_sinks", - schema: MZ_CATALOG_SCHEMA, - oid: oid::TABLE_MZ_KAFKA_SINKS_OID, - desc: RelationDesc::builder() - .with_column("id", SqlScalarType::String.nullable(false)) - .with_column("topic", SqlScalarType::String.nullable(false)) - .with_key(vec![0]) - .finish(), - column_comments: BTreeMap::from_iter([ - ("id", "The ID of the sink."), - ( - "topic", - "The name of the Kafka topic into which the sink is writing.", - ), - ]), - is_retained_metrics_object: false, - access: vec![PUBLIC_SELECT], - ontology: Some(Ontology { - entity_name: "kafka_sink", - description: "Kafka-specific sink configuration (topic)", - links: &const { - [OntologyLink { - name: "details_of", - target: "sink", - properties: LinkProperties::fk("id", "id", Cardinality::OneToOne), - }] - }, - column_semantic_types: &[("id", SemanticType::CatalogItemId)], - }), +pub static MZ_KAFKA_SINKS: LazyLock = LazyLock::new(|| { + BuiltinMaterializedView { + name: "mz_kafka_sinks", + schema: MZ_CATALOG_SCHEMA, + oid: oid::MV_MZ_KAFKA_SINKS_OID, + desc: RelationDesc::builder() + .with_column("id", SqlScalarType::String.nullable(false)) + .with_column("topic", SqlScalarType::String.nullable(false)) + .with_key(vec![0]) + .finish(), + column_comments: BTreeMap::from_iter([ + ("id", "The ID of the sink."), + ( + "topic", + "The name of the Kafka topic into which the sink is writing.", + ), + ]), + // Planning requires `TOPIC`, so it is always there. + sql: " +IN CLUSTER mz_catalog_server +WITH ( + ASSERT NOT NULL id, + ASSERT NOT NULL topic +) AS +SELECT + mz_internal.parse_catalog_id(data->'key'->'gid') AS id, + parsed->>'topic' AS topic +FROM + mz_internal.mz_catalog_raw + CROSS JOIN LATERAL ( + SELECT mz_internal.parse_catalog_create_sql(data->'value'->'definition'->'V1'->>'create_sql') + ) AS l(parsed) +WHERE + data->>'kind' = 'Item' AND + parsed->>'sink_type' = 'kafka'", + is_retained_metrics_object: false, + access: vec![PUBLIC_SELECT], + ontology: Some(Ontology { + entity_name: "kafka_sink", + description: "Kafka-specific sink configuration (topic)", + links: &const { + [OntologyLink { + name: "details_of", + target: "sink", + properties: LinkProperties::fk("id", "id", Cardinality::OneToOne), + }] + }, + column_semantic_types: &[("id", SemanticType::CatalogItemId)], + }), + } }); // Reads Item rows from `mz_catalog_raw`, pulls the broker addresses and any // explicit progress topic out of the persisted `create_sql` via @@ -1224,11 +1266,15 @@ WHERE }); // mz_sources is generated dynamically in BUILTINS_STATIC via builtin::make_mz_sources() // with builtin source/log entries inlined as VALUES. See builtin/builtin.rs. -pub static MZ_SINKS: LazyLock = LazyLock::new(|| { - BuiltinTable { +/// Sink metadata, all of it derived from the persisted `create_sql`. The +/// `CreateSink` arm of `parse_catalog_create_sql` does the parsing, including +/// the deprecated `format` column and its avro/json-only collapse. `size` is +/// deprecated and always NULL. +pub static MZ_SINKS: LazyLock = LazyLock::new(|| { + BuiltinMaterializedView { name: "mz_sinks", schema: MZ_CATALOG_SCHEMA, - oid: oid::TABLE_MZ_SINKS_OID, + oid: oid::MV_MZ_SINKS_OID, desc: RelationDesc::builder() .with_column("id", SqlScalarType::String.nullable(false)) .with_column("oid", SqlScalarType::Oid.nullable(false)) @@ -1294,6 +1340,43 @@ pub static MZ_SINKS: LazyLock = LazyLock::new(|| { "The redacted `CREATE` SQL statement for the sink.", ), ]), + sql: " +IN CLUSTER mz_catalog_server +WITH ( + ASSERT NOT NULL id, + ASSERT NOT NULL oid, + ASSERT NOT NULL schema_id, + ASSERT NOT NULL name, + ASSERT NOT NULL type, + ASSERT NOT NULL cluster_id, + ASSERT NOT NULL owner_id, + ASSERT NOT NULL create_sql, + ASSERT NOT NULL redacted_create_sql +) AS +SELECT + mz_internal.parse_catalog_id(data->'key'->'gid') AS id, + (data->'value'->>'oid')::oid AS oid, + mz_internal.parse_catalog_id(data->'value'->'schema_id') AS schema_id, + data->'value'->>'name' AS name, + parsed->>'sink_type' AS type, + parsed->>'connection_id' AS connection_id, + NULL::text AS size, + parsed->>'envelope_type' AS envelope_type, + parsed->>'format' AS format, + parsed->>'key_format' AS key_format, + parsed->>'value_format' AS value_format, + parsed->>'cluster_id' AS cluster_id, + mz_internal.parse_catalog_id(data->'value'->'owner_id') AS owner_id, + data->'value'->'definition'->'V1'->>'create_sql' AS create_sql, + mz_internal.redact_sql(data->'value'->'definition'->'V1'->>'create_sql') AS redacted_create_sql +FROM + mz_internal.mz_catalog_raw + CROSS JOIN LATERAL ( + SELECT mz_internal.parse_catalog_create_sql(data->'value'->'definition'->'V1'->>'create_sql') + ) AS l(parsed) +WHERE + data->>'kind' = 'Item' AND + parsed->>'type' = 'sink'", is_retained_metrics_object: true, access: vec![PUBLIC_SELECT], ontology: Some(Ontology { diff --git a/src/catalog/src/memory/objects.rs b/src/catalog/src/memory/objects.rs index c422cd79e0148..dd10215d8d8a1 100644 --- a/src/catalog/src/memory/objects.rs +++ b/src/catalog/src/memory/objects.rs @@ -1344,6 +1344,10 @@ impl Sink { } /// Envelope of the sink. + /// + /// NOTE: the `mz_sinks` view works this out from `create_sql` instead, in + /// the `CreateSink` arm of `parse_catalog_create_sql`. Kafka gets it from + /// `ENVELOPE` there, iceberg from `MODE`. Change both or they drift. pub fn envelope(&self) -> Option<&str> { match &self.envelope { SinkEnvelope::Debezium => Some("debezium"), @@ -1356,6 +1360,9 @@ impl Sink { /// if the key-format is none or the key & value formats are /// both the same (either avro or json), we return the value format name, /// otherwise we return a composite name. + /// + /// NOTE: `parse_catalog_create_sql` redoes this collapse for the `mz_sinks` + /// `format` column. Change both or they drift. pub fn combined_format(&self) -> Option> { match &self.connection { StorageSinkConnection::Kafka(connection) => Some(connection.format.get_format_name()), @@ -1364,6 +1371,10 @@ impl Sink { } /// Output distinct key_format and value_format of the sink. + /// + /// NOTE: also derived from `create_sql` for `mz_sinks`. Watch out that a key + /// format exists only when the sink has a `KEY`, so the SQL side has to work + /// that out from the statement for itself. pub fn formats(&self) -> Option<(Option<&str>, &str)> { match &self.connection { StorageSinkConnection::Kafka(connection) => { diff --git a/src/expr/src/scalar/func/impls/jsonb.rs b/src/expr/src/scalar/func/impls/jsonb.rs index 6929ea134192f..f7e29a38173f0 100644 --- a/src/expr/src/scalar/func/impls/jsonb.rs +++ b/src/expr/src/scalar/func/impls/jsonb.rs @@ -19,9 +19,10 @@ use mz_repr::{ArrayRustType, Datum, Row, RowPacker, SqlColumnType, SqlScalarType use mz_sql_parser::ast::display::AstDisplay; use mz_sql_parser::ast::{ AstInfo, AvroSchema, ConnectionOption, ConnectionOptionName, CreateConnectionType, - CreateSubsourceOptionName, Format, FormatSpecifier, KafkaSourceConfigOptionName, - PgConfigOptionName, ProtobufSchema, RawClusterName, RawItemName, SourceEnvelope, - SourceErrorPolicy, UnresolvedItemName, Value, WithOptionValue, + CreateSinkConnection, CreateSubsourceOptionName, Format, FormatSpecifier, + IcebergSinkConfigOptionName, IcebergSinkMode, KafkaSinkConfigOptionName, + KafkaSourceConfigOptionName, PgConfigOptionName, ProtobufSchema, RawClusterName, RawItemName, + SinkEnvelope, SourceEnvelope, SourceErrorPolicy, UnresolvedItemName, Value, WithOptionValue, }; use prost::Message as _; use serde::{Deserialize, Serialize}; @@ -576,7 +577,129 @@ fn parse_catalog_create_sql<'a>(a: &'a str) -> Result { "subsource" } - CreateSink(_) => "sink", + // 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!(get_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!(get_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!(get_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" + } CreateIndex(stmt) => { let Some(in_cluster) = stmt.in_cluster else { return Err("missing IN CLUSTER".into()); @@ -1697,4 +1820,180 @@ mod tests { let out = super::parse_catalog_create_sql(sql).expect("ok"); assert_eq!(as_serde(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::parse_catalog_create_sql(&sql).expect("ok"); + assert_eq!( + as_serde(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::parse_catalog_create_sql(&sql).expect("ok"); + assert_eq!( + as_serde(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::parse_catalog_create_sql(&sql).expect("ok"); + let out = as_serde(out); + 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 = as_serde(super::parse_catalog_create_sql(&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 = as_serde(super::parse_catalog_create_sql(&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 = as_serde(super::parse_catalog_create_sql(&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::parse_catalog_create_sql(sql).unwrap_err(); + assert!( + matches!(err, EvalError::InvalidCatalogJson(msg) if msg.contains("missing TOPIC")), + "wrong error variant/message" + ); + } + + #[mz_ore::test] + fn sink_iceberg_upsert_mode() { + let sql = iceberg_sink_sql("UPSERT"); + let out = super::parse_catalog_create_sql(&sql).expect("ok"); + assert_eq!( + as_serde(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 = as_serde(super::parse_catalog_create_sql(&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::parse_catalog_create_sql(sql).unwrap_err(); + assert!( + matches!(err, EvalError::InvalidCatalogJson(msg) if msg.contains("missing TABLE")), + "wrong error variant/message" + ); + } + + #[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::parse_catalog_create_sql(sql).expect("ok"); + assert_eq!(as_serde(out), json!({ "type": "view" })); + } } diff --git a/src/pgrepr-consts/src/oid.rs b/src/pgrepr-consts/src/oid.rs index 9d1c0672e5086..7a4716922990f 100644 --- a/src/pgrepr-consts/src/oid.rs +++ b/src/pgrepr-consts/src/oid.rs @@ -416,7 +416,7 @@ pub const LOG_MZ_MESSAGE_BATCH_COUNTS_SENT_RAW_OID: u32 = 16690; pub const LOG_MZ_MESSAGE_COUNTS_RECEIVED_RAW_OID: u32 = 16691; pub const LOG_MZ_MESSAGE_COUNTS_SENT_RAW_OID: u32 = 16692; pub const LOG_MZ_DATAFLOW_OPERATOR_REACHABILITY_RAW_OID: u32 = 16693; -pub const TABLE_MZ_KAFKA_SINKS_OID: u32 = 16694; +pub const MV_MZ_KAFKA_SINKS_OID: u32 = 16694; pub const MV_MZ_KAFKA_CONNECTIONS_OID: u32 = 16695; pub const MV_MZ_KAFKA_SOURCES_OID: u32 = 16696; pub const MV_MZ_POSTGRES_SOURCES_OID: u32 = 16697; @@ -433,7 +433,7 @@ pub const TABLE_MZ_TABLES_OID: u32 = 16707; pub const MV_MZ_CONNECTIONS_OID: u32 = 16708; pub const MV_MZ_SSH_TUNNEL_CONNECTIONS_OID: u32 = 16709; pub const MV_MZ_SOURCES_OID: u32 = 16710; -pub const TABLE_MZ_SINKS_OID: u32 = 16711; +pub const MV_MZ_SINKS_OID: u32 = 16711; pub const TABLE_MZ_VIEWS_OID: u32 = 16712; pub const MV_MZ_MATERIALIZED_VIEWS_OID: u32 = 16713; pub const TABLE_MZ_TYPES_OID: u32 = 16714; @@ -770,7 +770,7 @@ pub const VIEW_MZ_WALLCLOCK_GLOBAL_LAG_HISTOGRAM_OID: u32 = 17056; pub const MV_MZ_SQL_SERVER_SOURCE_TABLES_OID: u32 = 17057; pub const TABLE_MZ_LICENSE_KEYS_OID: u32 = 17058; pub const TABLE_MZ_ROLE_AUTH_OID: u32 = 17059; -pub const TABLE_MZ_ICEBERG_SINKS_OID: u32 = 17060; +pub const MV_MZ_ICEBERG_SINKS_OID: u32 = 17060; pub const VIEW_MZ_OBJECT_GLOBAL_IDS_OID: u32 = 17061; pub const TABLE_MZ_REPLACEMENTS_OID: u32 = 17062; pub const VIEW_PG_AUTHID_CORE_OID: u32 = 17063; diff --git a/src/storage-types/src/sinks.rs b/src/storage-types/src/sinks.rs index 99226b3483944..694fc42f8c0f0 100644 --- a/src/storage-types/src/sinks.rs +++ b/src/storage-types/src/sinks.rs @@ -186,6 +186,9 @@ impl IntoInlineConnection impl StorageSinkConnection { /// returns an option to not constrain ourselves in the future + /// + /// NOTE: `mz_sinks` digs this out of `create_sql` instead. Note the iceberg + /// case reports the catalog connection, not the optional AWS one. pub fn connection_id(&self) -> Option { use StorageSinkConnection::*; match self { @@ -198,6 +201,9 @@ impl StorageSinkConnection { } /// Returns the name of the sink connection. + /// + /// NOTE: `mz_sinks.type` comes from `create_sql`, not from here, so the two + /// sets of strings have to stay identical. pub fn name(&self) -> &'static str { use StorageSinkConnection::*; match self { @@ -485,6 +491,8 @@ impl KafkaSinkFormatType { } impl KafkaSinkFormat { + /// NOTE: the `mz_sinks` `format` column reimplements this in SQL, in + /// `parse_catalog_create_sql`. Change both or they drift. pub fn get_format_name<'a>(&'a self) -> Cow<'a, str> { // For legacy reasons, if the key-format is none or the key & value formats are // both the same (either avro or json), we return the value format name, diff --git a/test/iceberg/mode-append.td b/test/iceberg/mode-append.td index 3def611add9d1..ae18b97312ab5 100644 --- a/test/iceberg/mode-append.td +++ b/test/iceberg/mode-append.td @@ -40,6 +40,21 @@ MODE APPEND WITH (COMMIT INTERVAL '1s'); +# MODE APPEND is the only way to get envelope_type = append, and the catalog +# relations read it back out of create_sql. +> SELECT type, envelope_type, format, key_format, value_format, size, + connection_id = (SELECT id FROM mz_connections WHERE name = 'append_polaris') + FROM mz_sinks WHERE name = 'append_demo' +iceberg append true + +> SELECT namespace, "table" FROM mz_iceberg_sinks + WHERE id = (SELECT id FROM mz_sinks WHERE name = 'append_demo') +default_namespace append_demo_table + +> SELECT count(*) FROM mz_kafka_sinks + WHERE id = (SELECT id FROM mz_sinks WHERE name = 'append_demo') +0 + # Iceberg sinks commit data asynchronously; wait for at least one commit interval to elapse # before querying the table. $ sleep-is-probably-flaky-i-have-justified-my-need-with-a-comment duration=5s diff --git a/test/sqllogictest/catalog_server_explain.slt b/test/sqllogictest/catalog_server_explain.slt index 300281a804373..ce4e27e26a3c0 100644 --- a/test/sqllogictest/catalog_server_explain.slt +++ b/test/sqllogictest/catalog_server_explain.slt @@ -4756,6 +4756,21 @@ Target cluster: mz_catalog_server EOF +query T multiline +EXPLAIN MATERIALIZED VIEW "mz_catalog"."mz_iceberg_sinks"; +---- +mz_catalog.mz_iceberg_sinks: + →Read mz_internal.mz_catalog_raw + +Source mz_internal.mz_catalog_raw + project=(#2..=#4) + filter=(("Item" = (#0{data} ->> "kind")) AND ("iceberg" = (#1 ->> "sink_type"))) + map=(parse_catalog_create_sql(((((#0{data} -> "value") -> "definition") -> "V1") ->> "create_sql")), parse_catalog_id(((#0{data} -> "key") -> "gid")), (#1{parsed} ->> "namespace"), (#1{parsed} ->> "table")) + +Target cluster: mz_catalog_server + +EOF + query T multiline EXPLAIN MATERIALIZED VIEW "mz_catalog"."mz_indexes"; ---- @@ -4929,6 +4944,21 @@ Target cluster: mz_catalog_server EOF +query T multiline +EXPLAIN MATERIALIZED VIEW "mz_catalog"."mz_kafka_sinks"; +---- +mz_catalog.mz_kafka_sinks: + →Read mz_internal.mz_catalog_raw + +Source mz_internal.mz_catalog_raw + project=(#2, #3) + filter=(("Item" = (#0{data} ->> "kind")) AND ("kafka" = (#1 ->> "sink_type"))) + map=(parse_catalog_create_sql(((((#0{data} -> "value") -> "definition") -> "V1") ->> "create_sql")), parse_catalog_id(((#0{data} -> "key") -> "gid")), (#1{parsed} ->> "topic")) + +Target cluster: mz_catalog_server + +EOF + query T multiline EXPLAIN MATERIALIZED VIEW "mz_catalog"."mz_kafka_sources"; ---- @@ -4981,7 +5011,7 @@ mz_catalog.mz_materialized_views: Project: #4, #0, #5, #1, #6, #2, #9, #8, #3, #7 Map: {=r/s1, s1=r/s1}, "s1" →Arrange (#1{schema_name}, #2{name}) - →Constant (37 rows) + →Constant (40 rows) →Arrange (empty key) (#0{schema_name}) (#0{schema_name}, #1{name}) →Fused with Child Map/Filter/Project Project: #4, #3, #5 @@ -5149,6 +5179,21 @@ Target cluster: mz_catalog_server EOF +query T multiline +EXPLAIN MATERIALIZED VIEW "mz_catalog"."mz_sinks"; +---- +mz_catalog.mz_sinks: + →Read mz_internal.mz_catalog_raw + +Source mz_internal.mz_catalog_raw + project=(#4..=#9, #17, #10..=#15, #2, #16) + filter=(("Item" = (#0{data} ->> "kind")) AND ("sink" = (#3 ->> "type"))) + map=((#0{data} -> "value"), (((#1 -> "definition") -> "V1") ->> "create_sql"), parse_catalog_create_sql(#2), parse_catalog_id(((#0{data} -> "key") -> "gid")), text_to_oid((#1 ->> "oid")), parse_catalog_id((#1 -> "schema_id")), (#1 ->> "name"), (#3{parsed} ->> "sink_type"), (#3{parsed} ->> "connection_id"), (#3{parsed} ->> "envelope_type"), (#3{parsed} ->> "format"), (#3{parsed} ->> "key_format"), (#3{parsed} ->> "value_format"), (#3{parsed} ->> "cluster_id"), parse_catalog_id((#1 -> "owner_id")), redact_sql(#2), null) + +Target cluster: mz_catalog_server + +EOF + query T multiline EXPLAIN MATERIALIZED VIEW "mz_catalog"."mz_sources"; ---- @@ -7946,7 +7991,7 @@ query T multiline EXPLAIN SELECT * FROM "mz_internal"."mz_builtin_materialized_views"; ---- Explained Query (fast path): - →Constant (37 rows) + →Constant (40 rows) Target cluster: mz_catalog_server diff --git a/test/sqllogictest/information_schema_tables.slt b/test/sqllogictest/information_schema_tables.slt index 8ea3e4fdb71a7..3f1d1a0dca450 100644 --- a/test/sqllogictest/information_schema_tables.slt +++ b/test/sqllogictest/information_schema_tables.slt @@ -150,7 +150,7 @@ BASE TABLE materialize mz_catalog mz_iceberg_sinks -BASE TABLE +MATERIALIZED VIEW materialize mz_catalog mz_index_columns @@ -166,7 +166,7 @@ MATERIALIZED VIEW materialize mz_catalog mz_kafka_sinks -BASE TABLE +MATERIALIZED VIEW materialize mz_catalog mz_kafka_sources @@ -230,7 +230,7 @@ MATERIALIZED VIEW materialize mz_catalog mz_sinks -BASE TABLE +MATERIALIZED VIEW materialize mz_catalog mz_sources diff --git a/test/sqllogictest/mz_sinks.slt b/test/sqllogictest/mz_sinks.slt new file mode 100644 index 0000000000000..71ffe8123bcaa --- /dev/null +++ b/test/sqllogictest/mz_sinks.slt @@ -0,0 +1,212 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# 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. + +# Behavioural tests for mz_catalog.mz_sinks, mz_catalog.mz_kafka_sinks +# and mz_catalog.mz_iceberg_sinks. All three are materialized views over +# mz_internal.mz_catalog_raw and get every sink detail out of the +# persisted create_sql, in the CreateSink arm of parse_catalog_create_sql +# (src/expr/src/scalar/func/impls/jsonb.rs). +# +# Note we can't create a sink here at all: purification always talks to +# the broker or the iceberg catalog, and there is no VALIDATE = false to +# skip it. So the views stay empty and what we pin is the column shape, +# the item type, the non-sink exclusion, and the JSON keys the view SQL +# expects back from the helper. Real sinks are covered in +# test/testdrive/mz-sinks.td, the kafka-*-sinks.td files and +# test/iceberg/mode-append.td. + +mode cockroach + +reset-server + +# --auto-index-selects wraps the parse_catalog_create_sql calls below in +# indexed views, and that helper counts as an unstable dependency. +simple conn=mz_system,user=mz_system +ALTER SYSTEM SET unsafe_enable_unstable_dependencies = true +---- +COMPLETE 0 + +# --- Schema ------------------------------------------------------------------- + +# 15 columns, canonical order, types and nullability. `size` is deprecated +# and always NULL, but it still has to be here. +query TTT +SELECT c.name, c.type, c.nullable::text +FROM mz_catalog.mz_columns c +JOIN mz_catalog.mz_objects o ON c.id = o.id +JOIN mz_catalog.mz_schemas s ON o.schema_id = s.id +WHERE o.name = 'mz_sinks' AND s.name = 'mz_catalog' +ORDER BY c.position +---- +id text false +oid oid false +schema_id text false +name text false +type text false +connection_id text true +size text true +envelope_type text true +format text true +key_format text true +value_format text true +cluster_id text false +owner_id text false +create_sql text false +redacted_create_sql text false + +query TTT +SELECT c.name, c.type, c.nullable::text +FROM mz_catalog.mz_columns c +JOIN mz_catalog.mz_objects o ON c.id = o.id +JOIN mz_catalog.mz_schemas s ON o.schema_id = s.id +WHERE o.name = 'mz_kafka_sinks' AND s.name = 'mz_catalog' +ORDER BY c.position +---- +id text false +topic text false + +query TTT +SELECT c.name, c.type, c.nullable::text +FROM mz_catalog.mz_columns c +JOIN mz_catalog.mz_objects o ON c.id = o.id +JOIN mz_catalog.mz_schemas s ON o.schema_id = s.id +WHERE o.name = 'mz_iceberg_sinks' AND s.name = 'mz_catalog' +ORDER BY c.position +---- +id text false +namespace text false +table text false + +# All three are materialized views, not tables. +query TT +SELECT o.name, o.type +FROM mz_catalog.mz_objects o +JOIN mz_catalog.mz_schemas s ON o.schema_id = s.id +WHERE s.name = 'mz_catalog' + AND o.name IN ('mz_sinks', 'mz_kafka_sinks', 'mz_iceberg_sinks') +ORDER BY o.name +---- +mz_iceberg_sinks materialized-view +mz_kafka_sinks materialized-view +mz_sinks materialized-view + +# The retained-metrics index over mz_sinks survives the conversion. +query I +SELECT count(*) FROM mz_catalog.mz_indexes WHERE name = 'mz_sinks_ind' +---- +1 + +# --- Invariants --------------------------------------------------------------- + +# id and oid are both declared keys. +query I +SELECT count(*) - count(DISTINCT id) + count(*) - count(DISTINCT oid) +FROM mz_catalog.mz_sinks +---- +0 + +# `size` is deprecated and hardcoded NULL. +query I +SELECT count(*) FROM mz_catalog.mz_sinks WHERE size IS NOT NULL +---- +0 + +# The detail views partition mz_sinks by type, and every detail row joins back. +query I +SELECT + (SELECT count(*) FROM mz_catalog.mz_kafka_sinks) + - (SELECT count(*) FROM mz_catalog.mz_sinks WHERE type = 'kafka') ++ (SELECT count(*) FROM mz_catalog.mz_iceberg_sinks) + - (SELECT count(*) FROM mz_catalog.mz_sinks WHERE type = 'iceberg') +---- +0 + +query I +SELECT count(*) FROM mz_catalog.mz_kafka_sinks ks +LEFT JOIN mz_catalog.mz_sinks s ON ks.id = s.id +WHERE s.id IS NULL +---- +0 + +# --- Non-sink items are excluded ---------------------------------------------- + +statement ok +CREATE TABLE t (a int) + +statement ok +CREATE VIEW v AS SELECT * FROM t + +statement ok +CREATE MATERIALIZED VIEW mv AS SELECT * FROM t + +statement ok +CREATE SOURCE lg_src IN CLUSTER quickstart FROM LOAD GENERATOR COUNTER + +# No sinks are possible here, so all three stay empty even with other +# item types around. +query III +SELECT + (SELECT count(*) FROM mz_catalog.mz_sinks), + (SELECT count(*) FROM mz_catalog.mz_kafka_sinks), + (SELECT count(*) FROM mz_catalog.mz_iceberg_sinks) +---- +0 0 0 + +statement ok +DROP SOURCE lg_src + +# --- The JSON keys the view SQL reads ----------------------------------------- + +# The views read `parsed->>''`, so renaming a key on the helper side +# NULLs out a column instead of failing anything. These two calls pin the +# key names, against create_sql in the shape we actually persist. + +query TTTTTTTTT +SELECT + parsed->>'type', + parsed->>'sink_type', + parsed->>'cluster_id', + parsed->>'connection_id', + parsed->>'topic', + parsed->>'envelope_type', + parsed->>'format', + parsed->>'key_format', + parsed->>'value_format' +FROM ( + SELECT mz_internal.parse_catalog_create_sql( + 'CREATE SINK "materialize"."public"."snk" IN CLUSTER [u5] ' + || 'FROM [u1 AS "materialize"."public"."t"] ' + || 'INTO KAFKA CONNECTION [u2 AS "materialize"."public"."kc"] (TOPIC ''topic-x'') ' + || 'KEY (a) FORMAT JSON ENVELOPE UPSERT' + ) AS parsed +) +---- +sink kafka u5 u2 topic-x upsert json json json + +query TTTTTTTTT +SELECT + parsed->>'type', + parsed->>'sink_type', + parsed->>'cluster_id', + parsed->>'connection_id', + parsed->>'namespace', + parsed->>'table', + parsed->>'envelope_type', + parsed->>'format', + parsed->>'value_format' +FROM ( + SELECT mz_internal.parse_catalog_create_sql( + 'CREATE SINK "materialize"."public"."ice" IN CLUSTER [u5] ' + || 'FROM [u1 AS "materialize"."public"."t"] ' + || 'INTO ICEBERG CATALOG CONNECTION [u3 AS "materialize"."public"."cc"] ' + || '(NAMESPACE ''ns'', TABLE ''tbl'') MODE APPEND' + ) AS parsed +) +---- +sink iceberg u5 u3 ns tbl append NULL NULL diff --git a/test/testdrive-old-kafka-src-syntax/tables.td b/test/testdrive-old-kafka-src-syntax/tables.td index f90b6707dd0ae..cbbd8bb183b48 100644 --- a/test/testdrive-old-kafka-src-syntax/tables.td +++ b/test/testdrive-old-kafka-src-syntax/tables.td @@ -257,8 +257,11 @@ name type cluster comment ! INSERT INTO view VALUES (1) contains:cannot insert into view 'materialize.public.view' -! INSERT INTO mz_kafka_sinks VALUES ('bad', 'bad') -contains:cannot insert into system table 'mz_catalog.mz_kafka_sinks' +# Pick a target that will stay a system table: the builtin schema migration +# validator refuses to convert mz_storage_usage_by_shard, so this keeps +# testing the system-table path rather than some other error. +! INSERT INTO mz_internal.mz_storage_usage_by_shard VALUES (1, 'bad', 1, now()) +contains:cannot insert into system table 'mz_internal.mz_storage_usage_by_shard' > CREATE TABLE j (time TIMESTAMP NOT NULL); diff --git a/test/testdrive/catalog.td b/test/testdrive/catalog.td index eef5713d4fc51..b49f9ea805fee 100644 --- a/test/testdrive/catalog.td +++ b/test/testdrive/catalog.td @@ -516,15 +516,12 @@ mz_cluster_replica_sizes "" mz_columns "" mz_egress_ips "" mz_functions "" -mz_iceberg_sinks "" mz_index_columns "" -mz_kafka_sinks "" mz_list_types "" mz_map_types "" mz_operators "" mz_pseudo_types "" mz_role_auth "" -mz_sinks "" mz_tables "" mz_types "" mz_views "" @@ -549,14 +546,17 @@ mz_cluster_replicas mz_catalog_server "" mz_connections mz_catalog_server "" mz_databases mz_catalog_server "" mz_default_privileges mz_catalog_server "" +mz_iceberg_sinks mz_catalog_server "" mz_indexes mz_catalog_server "" mz_kafka_connections mz_catalog_server "" +mz_kafka_sinks mz_catalog_server "" mz_kafka_sources mz_catalog_server "" mz_role_members mz_catalog_server "" mz_role_parameters mz_catalog_server "" mz_roles mz_catalog_server "" mz_schemas mz_catalog_server "" mz_secrets mz_catalog_server "" +mz_sinks mz_catalog_server "" mz_sources mz_catalog_server "" mz_ssh_tunnel_connections mz_catalog_server "" mz_system_privileges mz_catalog_server "" @@ -830,7 +830,7 @@ test_table "" # `SHOW TABLES` and `mz_tables` should agree. > SELECT COUNT(*) FROM mz_tables WHERE id LIKE 's%' -34 +31 # There is one entry in mz_indexes for each field_number/expression of the index. > SELECT COUNT(id) FROM mz_indexes WHERE id LIKE 's%' diff --git a/test/testdrive/mz-sinks.td b/test/testdrive/mz-sinks.td index 84309b3a42851..db723ed449439 100644 --- a/test/testdrive/mz-sinks.td +++ b/test/testdrive/mz-sinks.td @@ -40,6 +40,19 @@ $ set-arg-default single-replica-cluster=quickstart FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_conn ENVELOPE UPSERT; +# Only avro/avro and json/json collapse to a shared name, so a keyed sink +# with a bare TEXT format has to report the composite form. +> CREATE MATERIALIZED VIEW mz_sinks_names AS + SELECT name FROM mz_sinks_table GROUP BY name; + +> CREATE SINK mz_sinks_text + IN CLUSTER ${arg.single-replica-cluster} + FROM mz_sinks_names + INTO KAFKA CONNECTION kafka_conn (TOPIC 'testdrive-mz-sinks-text-${testdrive.seed}') + KEY (name) + FORMAT TEXT + ENVELOPE UPSERT; + > SELECT envelope_type FROM mz_sinks WHERE name = 'mz_sinks_debezium' debezium @@ -48,3 +61,24 @@ upsert > SELECT format FROM mz_sinks WHERE name = 'mz_sinks_upsert' avro + +# Once the sink has a KEY, a bare FORMAT applies to the key as well. +> SELECT key_format, value_format FROM mz_sinks WHERE name = 'mz_sinks_upsert' +avro avro + +> SELECT format, key_format, value_format FROM mz_sinks WHERE name = 'mz_sinks_text' +key-text-value-text text text + +# The rest of the columns, plus the kafka detail relation. +> SELECT type, size, connection_id = (SELECT id FROM mz_connections WHERE name = 'kafka_conn'), + cluster_id = (SELECT id FROM mz_clusters WHERE name = '${arg.single-replica-cluster}') + FROM mz_sinks WHERE name = 'mz_sinks_upsert' +kafka true true + +> SELECT topic FROM mz_kafka_sinks + WHERE id = (SELECT id FROM mz_sinks WHERE name = 'mz_sinks_upsert') +testdrive-mz-upsert-debezium-${testdrive.seed} + +> SELECT count(*) FROM mz_iceberg_sinks + WHERE id IN (SELECT id FROM mz_sinks WHERE name LIKE 'mz_sinks_%') +0 diff --git a/test/testdrive/tables.td b/test/testdrive/tables.td index 82081855a7833..4cc53a8bee3d6 100644 --- a/test/testdrive/tables.td +++ b/test/testdrive/tables.td @@ -266,8 +266,11 @@ name type cluster comment ! INSERT INTO view VALUES (1) contains:cannot insert into view 'materialize.public.view' -! INSERT INTO mz_kafka_sinks VALUES ('bad', 'bad') -contains:cannot insert into system table 'mz_catalog.mz_kafka_sinks' +# Pick a target that will stay a system table: the builtin schema migration +# validator refuses to convert mz_storage_usage_by_shard, so this keeps +# testing the system-table path rather than some other error. +! INSERT INTO mz_internal.mz_storage_usage_by_shard VALUES (1, 'bad', 1, now()) +contains:cannot insert into system table 'mz_internal.mz_storage_usage_by_shard' > CREATE TABLE j (time TIMESTAMP NOT NULL);