From 447da2b53e26d93a6c4b7512234c0e7d67766b80 Mon Sep 17 00:00:00 2001 From: Aljoscha Krettek Date: Thu, 6 Aug 2026 10:28:24 +0200 Subject: [PATCH 1/3] sql: support IF NOT EXISTS for CREATE CLUSTER and CREATE CLUSTER REPLICA (#38061) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `CREATE CLUSTER` and `CREATE CLUSTER REPLICA` now accept an `IF NOT EXISTS` clause, matching the 13 other `CREATE` statements that already support it. With it, creating a cluster or replica whose name is already taken succeeds and emits a `cluster "c" already exists, skipping` notice (SQLSTATE 42710) instead of failing, so idempotent provisioning scripts no longer need a pre-flight existence check. Skipping leaves the existing object completely untouched. The configuration in the statement is not applied, so this is not an upsert. Previously these two statements produced a misleading parse error, because `IF` was consumed as the object name and the failure surfaced one token later: ``` CREATE CLUSTER IF NOT EXISTS c (SIZE '25cc') ERROR: Expected one of AUTO or AVAILABILITY or DISK or ..., found NOT ``` ## Implementation The machinery already exists end to end, so this is mostly threading a flag: `if_not_exists` on the two AST nodes and the two plans, a `parse_if_not_exists()` call in each parse function, and the established "turn the catalog's already-exists error into a notice" arm in the sequencer. Two things worth calling out: * The skip happens at the catalog transaction, where every sibling path puts it, rather than short-circuiting early on the in-memory catalog. That means the statement is validated in full first, so `CREATE CLUSTER IF NOT EXISTS c (SIZE 'typo')` still reports the bad size rather than silently succeeding because `c` happens to exist. The cost is that a skipped statement still allocates (and discards) a cluster id, which is a durable catalog write. That is the same cost every other `CREATE ... IF NOT EXISTS` already pays, and silently accepting a typo'd config in a provisioning script is the worse failure mode. * A bare `if` identifier now renders quoted, joining the existing list of keywords quoted for exactly this reason (`PREPARE`, `INTO`, `WHEN`, …). The `IF [NOT] EXISTS` clause sits exactly where the object name goes, so `CREATE CLUSTER if (SIZE …)` rendered in Simple mode no longer reparses. That affects the parser's own display roundtrip test, the user-facing `pretty_sql()` function, and `mz-deploy`, which renders a project file's statement with `to_string()` and executes it. `SHOW CREATE CLUSTER` and the plan/unplan roundtrip assertion in `plan_create_cluster` both use Stable mode, which already quoted everything, so neither was ever affected. The flip side is a grammar break: `CREATE CLUSTER if …` and `CREATE CLUSTER REPLICA if.r …` used to parse, since the parser accepts keywords as identifiers, and now do not. Quoting still works, and `DROP CLUSTER if` was already broken the same way. I smoke tested the upgrade: created a cluster named `if` (unquoted) on `main` along with a table, index and materialized view in it, then restarted on this branch against the same catalog. Bootstrap is clean, the cluster and its objects survive and serve queries, and `ALTER`/`DROP` by quoted name work. The only observable change is that renderings which used to emit a bare `if` now quote it, for example `SHOW CREATE INDEX` goes from `IN CLUSTER if` to `IN CLUSTER "if"`. `SHOW CREATE CLUSTER` does not render `IF NOT EXISTS`. It is a property of the statement, not of the cluster. ## Tests * `src/sql-parser/tests/testdata/ddl`: syntax for both statements, the `IF EXISTS` typo, and a cluster literally named `if`. * `test/sqllogictest/cluster.slt`: idempotence, that the existing object is left untouched, that a bad `SIZE` is still reported, the managed and unmanaged cluster variants, `FEATURES`, reserved names, case sensitivity, full `SHOW CREATE CLUSTER` output, that the replica form still requires the cluster to exist, and that it does not soften the managed-cluster rejection. * `test/pgtest-mz/notice.pt`: the wire-level notice for both statements. Fixes [SQL-601](https://linear.app/materializeinc/issue/SQL-601) --- doc/user/data/examples/create_cluster.yml | 8 +- .../data/examples/create_cluster_replica.yml | 8 +- .../src/coord/sequencer/inner/cluster.rs | 42 ++++- src/catalog/src/memory/objects.rs | 11 ++ src/sql-parser/src/ast/defs/name.rs | 10 +- src/sql-parser/src/ast/defs/statement.rs | 8 + src/sql-parser/src/parser.rs | 4 + src/sql-parser/tests/testdata/ddl | 117 +++++++++----- src/sql-pretty/src/doc.rs | 16 +- src/sql/src/plan.rs | 2 + src/sql/src/plan/statement/ddl.rs | 12 +- src/sql/src/rbac.rs | 2 + test/pgtest-mz/notice.pt | 33 ++++ test/sqllogictest/cluster.slt | 150 ++++++++++++++++++ 14 files changed, 367 insertions(+), 56 deletions(-) diff --git a/doc/user/data/examples/create_cluster.yml b/doc/user/data/examples/create_cluster.yml index 79b8dec01d124..7eb636d9ffcec 100644 --- a/doc/user/data/examples/create_cluster.yml +++ b/doc/user/data/examples/create_cluster.yml @@ -1,6 +1,6 @@ - name: "syntax" code: | - CREATE CLUSTER ( + CREATE CLUSTER [IF NOT EXISTS] ( SIZE = [, REPLICATION FACTOR = ] [, MANAGED = ] @@ -12,6 +12,12 @@ )] ); syntax_elements: + - name: "**IF NOT EXISTS**" + description: | + *Optional.* If specified, do not throw an error if a cluster with the + same name already exists. Instead, issue a notice and skip the cluster + creation. Note that the existing cluster is left untouched, its + configuration is not updated to match the statement. - name: "``" description: | A name for the cluster. diff --git a/doc/user/data/examples/create_cluster_replica.yml b/doc/user/data/examples/create_cluster_replica.yml index 029b7ced78e49..44658b2b0d9d3 100644 --- a/doc/user/data/examples/create_cluster_replica.yml +++ b/doc/user/data/examples/create_cluster_replica.yml @@ -1,9 +1,15 @@ - name: "syntax" code: | - CREATE CLUSTER REPLICA . ( + CREATE CLUSTER REPLICA [IF NOT EXISTS] . ( SIZE = ); syntax_elements: + - name: "**IF NOT EXISTS**" + description: | + *Optional.* If specified, do not throw an error if a replica with the + same name already exists on the cluster. Instead, issue a notice and + skip the replica creation. Note that the existing replica is left + untouched, its configuration is not updated to match the statement. - name: "``" description: | The cluster you want to attach a replica to. diff --git a/src/adapter/src/coord/sequencer/inner/cluster.rs b/src/adapter/src/coord/sequencer/inner/cluster.rs index 06232d8765f00..4fd7b174d467a 100644 --- a/src/adapter/src/coord/sequencer/inner/cluster.rs +++ b/src/adapter/src/coord/sequencer/inner/cluster.rs @@ -15,6 +15,7 @@ use maplit::btreeset; use mz_adapter_types::cluster_state::ReconfigurationAudit; use mz_catalog::builtin::BUILTINS; use mz_catalog::durable::managed_cluster_replica_name; +use mz_catalog::memory::error::ErrorKind; use mz_catalog::memory::objects::{ ClusterConfig, ClusterReplica, ClusterVariant, ClusterVariantManaged, ManagedReplicaConfigShape, ReconfigurationState, ReconfigurationStatus, ReconfigurationTarget, @@ -31,7 +32,7 @@ use mz_repr::Timestamp; use mz_repr::adt::numeric::Numeric; use mz_repr::role_id::RoleId; use mz_sql::ast::{Ident, QualifiedReplica}; -use mz_sql::catalog::{CatalogCluster, ObjectType}; +use mz_sql::catalog::{CatalogCluster, CatalogError, ObjectType}; use mz_sql::plan::{ self, AlterClusterPlanStrategy, AlterClusterRenamePlan, AlterClusterReplicaRenamePlan, AlterClusterSwapPlan, AlterOptionParameter, AlterSetClusterPlan, @@ -61,7 +62,7 @@ use crate::coord::{ AlterClusterWaitForHydrated, ClusterReplicaStatuses, ClusterStage, Coordinator, Message, PlanValidity, StageResult, Staged, }; -use crate::{AdapterError, ExecuteContext, ExecuteResponse, session::Session}; +use crate::{AdapterError, AdapterNotice, ExecuteContext, ExecuteResponse, session::Session}; const PENDING_REPLICA_SUFFIX: &str = "-pending"; @@ -1210,6 +1211,7 @@ impl Coordinator { name, variant, workload_class, + if_not_exists, }: CreateClusterPlan, ) -> Result { tracing::debug!("sequence_create_cluster"); @@ -1260,14 +1262,26 @@ impl Coordinator { match variant { CreateClusterVariant::Managed(plan) => { - self.sequence_create_managed_cluster(session, plan, id, name, ops) + self.sequence_create_managed_cluster(session, plan, id, name.clone(), ops) .await } CreateClusterVariant::Unmanaged(plan) => { - self.sequence_create_unmanaged_cluster(session, plan, id, name, ops) + self.sequence_create_unmanaged_cluster(session, plan, id, name.clone(), ops) .await } } + .or_else(|err| match err { + AdapterError::Catalog(mz_catalog::memory::error::Error { + kind: ErrorKind::Sql(CatalogError::ClusterAlreadyExists(_)), + }) if if_not_exists => { + session.add_notice(AdapterNotice::ObjectAlreadyExists { + name, + ty: "cluster", + }); + Ok(ExecuteResponse::CreatedCluster) + } + err => Err(err), + }) } #[mz_ore::instrument(level = "debug")] @@ -1668,6 +1682,7 @@ impl Coordinator { name, cluster_id, config, + if_not_exists, }: CreateClusterReplicaPlan, ) -> Result { // Choose default AZ if necessary @@ -1765,6 +1780,9 @@ impl Coordinator { let cluster_name = cluster.name.clone(); let is_builtin = cluster_id.is_system(); + // A replica name is only unique within its cluster, so the notice on the + // `IF NOT EXISTS` path below has to name both. + let qualified_name = format!("{cluster_name}.{name}"); // Pre-allocate the replica id out-of-band via the durable allocator, // picking the id type from the target cluster, which may be a system @@ -1822,9 +1840,19 @@ impl Coordinator { } } - self.catalog_transact(Some(session), ops).await?; - - Ok(ExecuteResponse::CreatedClusterReplica) + match self.catalog_transact(Some(session), ops).await { + Ok(()) => Ok(ExecuteResponse::CreatedClusterReplica), + Err(AdapterError::Catalog(mz_catalog::memory::error::Error { + kind: ErrorKind::Sql(CatalogError::DuplicateReplica(_, _)), + })) if if_not_exists => { + session.add_notice(AdapterNotice::ObjectAlreadyExists { + name: qualified_name, + ty: "cluster replica", + }); + Ok(ExecuteResponse::CreatedClusterReplica) + } + Err(err) => Err(err), + } } /// When this is called by the automated cluster scheduling, `scheduling_decision_reason` should diff --git a/src/catalog/src/memory/objects.rs b/src/catalog/src/memory/objects.rs index beae6a2c0aeb9..c422cd79e0148 100644 --- a/src/catalog/src/memory/objects.rs +++ b/src/catalog/src/memory/objects.rs @@ -404,6 +404,13 @@ impl Cluster { } } + /// Builds the plan that a `CREATE CLUSTER` statement for this cluster would + /// have produced. + /// + /// Clusters are stored as a structured config, not as SQL text, so `SHOW + /// CREATE CLUSTER` cannot just print a stored statement. It rebuilds one, + /// and this is the first half of that: config to plan. `unplan_create_cluster` + /// then turns the plan back into a statement. pub fn try_to_plan(&self) -> Result { let name = self.name.clone(); let variant = match &self.config.variant { @@ -462,6 +469,10 @@ impl Cluster { name, variant, workload_class, + // Nothing about `IF NOT EXISTS` is stored. It only ever affected how + // the original statement handled a name collision, so the rebuilt + // statement never shows it. + if_not_exists: false, }) } } diff --git a/src/sql-parser/src/ast/defs/name.rs b/src/sql-parser/src/ast/defs/name.rs index c6559ebdfcc4d..eb445ac2925a8 100644 --- a/src/sql-parser/src/ast/defs/name.rs +++ b/src/sql-parser/src/ast/defs/name.rs @@ -19,7 +19,9 @@ // limitations under the License. use mz_ore::str::StrExt; -use mz_sql_lexer::keywords::{ALL, ANY, AS, DISTINCT, INTO, Keyword, LIST, PREPARE, SOME, WHEN}; +use mz_sql_lexer::keywords::{ + ALL, ANY, AS, DISTINCT, IF, INTO, Keyword, LIST, PREPARE, SOME, WHEN, +}; use mz_sql_lexer::lexer::{IdentString, MAX_IDENTIFIER_LENGTH}; use serde::{Deserialize, Serialize}; use std::fmt; @@ -359,6 +361,12 @@ impl Ident { // (`COPY into FROM x` -> `COPY INTO …`, which // then fails expecting the FROM/TO direction). || kw == INTO + // The `IF [NOT] EXISTS` clauses of CREATE/DROP/ALTER sit + // exactly where the object name goes, so a bare `if` name + // is consumed as the start of such a clause on reparse + // (`CREATE CLUSTER if (SIZE …)` -> "expected NOT, found + // left parenthesis"). + || kw == IF }) .unwrap_or(false) } diff --git a/src/sql-parser/src/ast/defs/statement.rs b/src/sql-parser/src/ast/defs/statement.rs index a131594360f3e..055f45a5780bb 100644 --- a/src/sql-parser/src/ast/defs/statement.rs +++ b/src/sql-parser/src/ast/defs/statement.rs @@ -2467,11 +2467,15 @@ pub struct CreateClusterStatement { pub options: Vec>, /// The comma-separated features enabled on the cluster. pub features: Vec>, + pub if_not_exists: bool, } impl AstDisplay for CreateClusterStatement { fn fmt(&self, f: &mut AstFormatter) { f.write_str("CREATE CLUSTER "); + if self.if_not_exists { + f.write_str("IF NOT EXISTS "); + } f.write_node(&self.name); if !self.options.is_empty() { f.write_str(" ("); @@ -2566,11 +2570,15 @@ pub struct CreateClusterReplicaStatement { pub of_cluster: Ident, /// The replica's definition. pub definition: ReplicaDefinition, + pub if_not_exists: bool, } impl AstDisplay for CreateClusterReplicaStatement { fn fmt(&self, f: &mut AstFormatter) { f.write_str("CREATE CLUSTER REPLICA "); + if self.if_not_exists { + f.write_str("IF NOT EXISTS "); + } f.write_node(&self.of_cluster); f.write_str("."); f.write_node(&self.definition.name); diff --git a/src/sql-parser/src/parser.rs b/src/sql-parser/src/parser.rs index 34763fdb55465..be20b55a1de32 100644 --- a/src/sql-parser/src/parser.rs +++ b/src/sql-parser/src/parser.rs @@ -4564,6 +4564,7 @@ impl<'a> Parser<'a> { } fn parse_create_cluster(&mut self) -> Result, ParserError> { + let if_not_exists = self.parse_if_not_exists()?; let name = self.parse_identifier()?; // For historical reasons, the parentheses around the options can be // omitted. @@ -4586,6 +4587,7 @@ impl<'a> Parser<'a> { name, options, features, + if_not_exists, })) } @@ -4871,6 +4873,7 @@ impl<'a> Parser<'a> { fn parse_create_cluster_replica(&mut self) -> Result, ParserError> { self.next_token(); + let if_not_exists = self.parse_if_not_exists()?; let of_cluster = self.parse_identifier()?; self.expect_token(&Token::Dot)?; let name = self.parse_identifier()?; @@ -4885,6 +4888,7 @@ impl<'a> Parser<'a> { CreateClusterReplicaStatement { of_cluster, definition: ReplicaDefinition { name, options }, + if_not_exists, }, )) } diff --git a/src/sql-parser/tests/testdata/ddl b/src/sql-parser/tests/testdata/ddl index 6ee7e4a46002d..2806964651181 100644 --- a/src/sql-parser/tests/testdata/ddl +++ b/src/sql-parser/tests/testdata/ddl @@ -1973,7 +1973,7 @@ CREATE CLUSTER cluster REPLICAS () ---- CREATE CLUSTER cluster (REPLICAS = ()) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Replicas, value: Some(ClusterReplicas([])) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Replicas, value: Some(ClusterReplicas([])) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster WITH REPLICAS () @@ -1994,91 +1994,112 @@ CREATE CLUSTER cluster REPLICAS (a (STORAGECTL ADDRESSES ['host1'])) ---- CREATE CLUSTER cluster (REPLICAS = (a (STORAGECTL ADDRESSES = ('host1')))) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Replicas, value: Some(ClusterReplicas([ReplicaDefinition { name: Ident("a"), options: [ReplicaOption { name: StoragectlAddresses, value: Some(Sequence([Value(String("host1"))])) }] }])) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Replicas, value: Some(ClusterReplicas([ReplicaDefinition { name: Ident("a"), options: [ReplicaOption { name: StoragectlAddresses, value: Some(Sequence([Value(String("host1"))])) }] }])) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster REPLICAS (a (COMPUTECTL ADDRESSES ['host1']), b (SIZE 'scale=1,workers=1')) ---- CREATE CLUSTER cluster (REPLICAS = (a (COMPUTECTL ADDRESSES = ('host1')), b (SIZE = 'scale=1,workers=1'))) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Replicas, value: Some(ClusterReplicas([ReplicaDefinition { name: Ident("a"), options: [ReplicaOption { name: ComputectlAddresses, value: Some(Sequence([Value(String("host1"))])) }] }, ReplicaDefinition { name: Ident("b"), options: [ReplicaOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }] }])) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Replicas, value: Some(ClusterReplicas([ReplicaDefinition { name: Ident("a"), options: [ReplicaOption { name: ComputectlAddresses, value: Some(Sequence([Value(String("host1"))])) }] }, ReplicaDefinition { name: Ident("b"), options: [ReplicaOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }] }])) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster REPLICAS (a (COMPUTE ADDRESSES ['host1'], INTROSPECTION INTERVAL '1s', INTROSPECTION DEBUGGING true), b (SIZE 'scale=1,workers=1', INTROSPECTION INTERVAL 0)) ---- CREATE CLUSTER cluster (REPLICAS = (a (COMPUTE ADDRESSES = ('host1'), INTROSPECTION INTERVAL = '1s', INTROSPECTION DEBUGGING = true), b (SIZE = 'scale=1,workers=1', INTROSPECTION INTERVAL = 0))) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Replicas, value: Some(ClusterReplicas([ReplicaDefinition { name: Ident("a"), options: [ReplicaOption { name: ComputeAddresses, value: Some(Sequence([Value(String("host1"))])) }, ReplicaOption { name: IntrospectionInterval, value: Some(Value(String("1s"))) }, ReplicaOption { name: IntrospectionDebugging, value: Some(Value(Boolean(true))) }] }, ReplicaDefinition { name: Ident("b"), options: [ReplicaOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ReplicaOption { name: IntrospectionInterval, value: Some(Value(Number("0"))) }] }])) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Replicas, value: Some(ClusterReplicas([ReplicaDefinition { name: Ident("a"), options: [ReplicaOption { name: ComputeAddresses, value: Some(Sequence([Value(String("host1"))])) }, ReplicaOption { name: IntrospectionInterval, value: Some(Value(String("1s"))) }, ReplicaOption { name: IntrospectionDebugging, value: Some(Value(Boolean(true))) }] }, ReplicaDefinition { name: Ident("b"), options: [ReplicaOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ReplicaOption { name: IntrospectionInterval, value: Some(Value(Number("0"))) }] }])) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster REPLICAS (a (SIZE 'scale=1,workers=1')) ---- CREATE CLUSTER cluster (REPLICAS = (a (SIZE = 'scale=1,workers=1'))) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Replicas, value: Some(ClusterReplicas([ReplicaDefinition { name: Ident("a"), options: [ReplicaOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }] }])) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Replicas, value: Some(ClusterReplicas([ReplicaDefinition { name: Ident("a"), options: [ReplicaOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }] }])) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster REPLICAS (a (STORAGECTL ADDRESSES ['123'], STORAGE ADDRESSES ['124'], COMPUTECTL ADDRESSES ['host1:2400', 'host2:2400'], COMPUTE ADDRESSES ['host1:2401', 'host2:2401'], WORKERS '1')) ---- CREATE CLUSTER cluster (REPLICAS = (a (STORAGECTL ADDRESSES = ('123'), STORAGE ADDRESSES = ('124'), COMPUTECTL ADDRESSES = ('host1:2400', 'host2:2400'), COMPUTE ADDRESSES = ('host1:2401', 'host2:2401'), WORKERS = '1'))) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Replicas, value: Some(ClusterReplicas([ReplicaDefinition { name: Ident("a"), options: [ReplicaOption { name: StoragectlAddresses, value: Some(Sequence([Value(String("123"))])) }, ReplicaOption { name: StorageAddresses, value: Some(Sequence([Value(String("124"))])) }, ReplicaOption { name: ComputectlAddresses, value: Some(Sequence([Value(String("host1:2400")), Value(String("host2:2400"))])) }, ReplicaOption { name: ComputeAddresses, value: Some(Sequence([Value(String("host1:2401")), Value(String("host2:2401"))])) }, ReplicaOption { name: Workers, value: Some(Value(String("1"))) }] }])) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Replicas, value: Some(ClusterReplicas([ReplicaDefinition { name: Ident("a"), options: [ReplicaOption { name: StoragectlAddresses, value: Some(Sequence([Value(String("123"))])) }, ReplicaOption { name: StorageAddresses, value: Some(Sequence([Value(String("124"))])) }, ReplicaOption { name: ComputectlAddresses, value: Some(Sequence([Value(String("host1:2400")), Value(String("host2:2400"))])) }, ReplicaOption { name: ComputeAddresses, value: Some(Sequence([Value(String("host1:2401")), Value(String("host2:2401"))])) }, ReplicaOption { name: Workers, value: Some(Value(String("1"))) }] }])) }], features: [], if_not_exists: false }) + +parse-statement +CREATE CLUSTER IF NOT EXISTS cluster SIZE 'scale=1,workers=1' +---- +CREATE CLUSTER IF NOT EXISTS cluster (SIZE = 'scale=1,workers=1') +=> +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }], features: [], if_not_exists: true }) + +parse-statement +CREATE CLUSTER IF EXISTS cluster SIZE 'scale=1,workers=1' +---- +error: Expected NOT, found EXISTS +CREATE CLUSTER IF EXISTS cluster SIZE 'scale=1,workers=1' + ^ + +parse-statement +CREATE CLUSTER "if" SIZE 'scale=1,workers=1' +---- +CREATE CLUSTER "if" (SIZE = 'scale=1,workers=1') +=> +CreateCluster(CreateClusterStatement { name: Ident("if"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster SIZE 'scale=1,workers=1' ---- CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1') => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster (SIZE 'scale=1,workers=1') FEATURES (enable eager delta joins = true) ---- CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1') FEATURES (ENABLE EAGER DELTA JOINS = true) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }], features: [ClusterFeature { name: EnableEagerDeltaJoins, value: Some(Value(Boolean(true))) }] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }], features: [ClusterFeature { name: EnableEagerDeltaJoins, value: Some(Value(Boolean(true))) }], if_not_exists: false }) parse-statement CREATE CLUSTER cluster REPLICATION FACTOR 1 ---- CREATE CLUSTER cluster (REPLICATION FACTOR = 1) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: ReplicationFactor, value: Some(Value(Number("1"))) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: ReplicationFactor, value: Some(Value(Number("1"))) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster AVAILABILITY ZONES ('1') ---- CREATE CLUSTER cluster (AVAILABILITY ZONES = ('1')) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: AvailabilityZones, value: Some(Sequence([Value(String("1"))])) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: AvailabilityZones, value: Some(Sequence([Value(String("1"))])) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster MANAGED true ---- CREATE CLUSTER cluster (MANAGED = true) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Managed, value: Some(Value(Boolean(true))) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Managed, value: Some(Value(Boolean(true))) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster MANAGED ---- CREATE CLUSTER cluster (MANAGED) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Managed, value: None }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Managed, value: None }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster MANAGED, DISK = true ---- CREATE CLUSTER cluster (MANAGED, DISK = true) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Managed, value: None }, ClusterOption { name: Disk, value: Some(Value(Boolean(true))) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Managed, value: None }, ClusterOption { name: Disk, value: Some(Value(Boolean(true))) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster (MANAGED, DISK = true) ---- CREATE CLUSTER cluster (MANAGED, DISK = true) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Managed, value: None }, ClusterOption { name: Disk, value: Some(Value(Boolean(true))) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Managed, value: None }, ClusterOption { name: Disk, value: Some(Value(Boolean(true))) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster (MANAGED, DISK = true @@ -2099,56 +2120,56 @@ CREATE CLUSTER cluster INTROSPECTION INTERVAL '1' ---- CREATE CLUSTER cluster (INTROSPECTION INTERVAL = '1') => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: IntrospectionInterval, value: Some(Value(String("1"))) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: IntrospectionInterval, value: Some(Value(String("1"))) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster INTROSPECTION DEBUGGING true ---- CREATE CLUSTER cluster (INTROSPECTION DEBUGGING = true) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: IntrospectionDebugging, value: Some(Value(Boolean(true))) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: IntrospectionDebugging, value: Some(Value(Boolean(true))) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster (SIZE = '1', EXPERIMENTAL ARRANGEMENT COMPRESSION = true) ---- CREATE CLUSTER cluster (SIZE = '1', EXPERIMENTAL ARRANGEMENT COMPRESSION = true) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("1"))) }, ClusterOption { name: ExperimentalArrangementCompression, value: Some(Value(Boolean(true))) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("1"))) }, ClusterOption { name: ExperimentalArrangementCompression, value: Some(Value(Boolean(true))) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER REPLICA cluster.r1 (SIZE = '1', EXPERIMENTAL ARRANGEMENT COMPRESSION = true) ---- CREATE CLUSTER REPLICA cluster.r1 (SIZE = '1', EXPERIMENTAL ARRANGEMENT COMPRESSION = true) => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("cluster"), definition: ReplicaDefinition { name: Ident("r1"), options: [ReplicaOption { name: Size, value: Some(Value(String("1"))) }, ReplicaOption { name: ExperimentalArrangementCompression, value: Some(Value(Boolean(true))) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("cluster"), definition: ReplicaDefinition { name: Ident("r1"), options: [ReplicaOption { name: Size, value: Some(Value(String("1"))) }, ReplicaOption { name: ExperimentalArrangementCompression, value: Some(Value(Boolean(true))) }] }, if_not_exists: false }) parse-statement CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', SCHEDULE = MANUAL) ---- CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', SCHEDULE = MANUAL) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: Schedule, value: Some(ClusterScheduleOptionValue(Manual)) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: Schedule, value: Some(ClusterScheduleOptionValue(Manual)) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', SCHEDULE = ON REFRESH) ---- CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', SCHEDULE = ON REFRESH) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: Schedule, value: Some(ClusterScheduleOptionValue(Refresh { hydration_time_estimate: None })) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: Schedule, value: Some(ClusterScheduleOptionValue(Refresh { hydration_time_estimate: None })) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', SCHEDULE = ON REFRESH (HYDRATION TIME ESTIMATE = '1 hour')) ---- CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', SCHEDULE = ON REFRESH (HYDRATION TIME ESTIMATE = '1 hour')) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: Schedule, value: Some(ClusterScheduleOptionValue(Refresh { hydration_time_estimate: Some(IntervalValue { value: "1 hour", precision_high: Year, precision_low: Second, fsec_max_precision: None }) })) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: Schedule, value: Some(ClusterScheduleOptionValue(Refresh { hydration_time_estimate: Some(IntervalValue { value: "1 hour", precision_high: Year, precision_low: Second, fsec_max_precision: None }) })) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', SCHEDULE = ON REFRESH (HYDRATION TIME ESTIMATE '1 hour')) ---- CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', SCHEDULE = ON REFRESH (HYDRATION TIME ESTIMATE = '1 hour')) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: Schedule, value: Some(ClusterScheduleOptionValue(Refresh { hydration_time_estimate: Some(IntervalValue { value: "1 hour", precision_high: Year, precision_low: Second, fsec_max_precision: None }) })) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: Schedule, value: Some(ClusterScheduleOptionValue(Refresh { hydration_time_estimate: Some(IntervalValue { value: "1 hour", precision_high: Year, precision_low: Second, fsec_max_precision: None }) })) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', SCHEDULE = ON REFRESH (HYDRATION TIME ESTIMATE)) @@ -2183,21 +2204,21 @@ CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', AUTO SCALING STRATEGY = (ON ---- CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', AUTO SCALING STRATEGY = (ON HYDRATION (HYDRATION SIZE = '400cc', LINGER DURATION = '600s'))) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: AutoScalingStrategy, value: Some(ClusterAutoScalingStrategyOptionValue(ClusterAutoScalingStrategyOptionValue { on_hydration: Some(OnHydrationOptionValue { hydration_size: String("400cc"), linger_duration: Some(String("600s")) }) })) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: AutoScalingStrategy, value: Some(ClusterAutoScalingStrategyOptionValue(ClusterAutoScalingStrategyOptionValue { on_hydration: Some(OnHydrationOptionValue { hydration_size: String("400cc"), linger_duration: Some(String("600s")) }) })) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', AUTO SCALING STRATEGY = (ON HYDRATION (HYDRATION SIZE = '400cc'))) ---- CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', AUTO SCALING STRATEGY = (ON HYDRATION (HYDRATION SIZE = '400cc'))) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: AutoScalingStrategy, value: Some(ClusterAutoScalingStrategyOptionValue(ClusterAutoScalingStrategyOptionValue { on_hydration: Some(OnHydrationOptionValue { hydration_size: String("400cc"), linger_duration: None }) })) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: AutoScalingStrategy, value: Some(ClusterAutoScalingStrategyOptionValue(ClusterAutoScalingStrategyOptionValue { on_hydration: Some(OnHydrationOptionValue { hydration_size: String("400cc"), linger_duration: None }) })) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', AUTO SCALING STRATEGY = ()) ---- CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', AUTO SCALING STRATEGY = ()) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: AutoScalingStrategy, value: Some(ClusterAutoScalingStrategyOptionValue(ClusterAutoScalingStrategyOptionValue { on_hydration: None })) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: AutoScalingStrategy, value: Some(ClusterAutoScalingStrategyOptionValue(ClusterAutoScalingStrategyOptionValue { on_hydration: None })) }], features: [], if_not_exists: false }) parse-statement ALTER CLUSTER cluster SET (AUTO SCALING STRATEGY = (ON HYDRATION (HYDRATION SIZE = '400cc'))) @@ -2232,14 +2253,14 @@ CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', WORKLOAD CLASS NULL) ---- CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', WORKLOAD CLASS = NULL) => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: WorkloadClass, value: Some(Value(Null)) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: WorkloadClass, value: Some(Value(Null)) }], features: [], if_not_exists: false }) parse-statement CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', WORKLOAD CLASS 'production') ---- CREATE CLUSTER cluster (SIZE = 'scale=1,workers=1', WORKLOAD CLASS = 'production') => -CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: WorkloadClass, value: Some(Value(String("production"))) }], features: [] }) +CreateCluster(CreateClusterStatement { name: Ident("cluster"), options: [ClusterOption { name: Size, value: Some(Value(String("scale=1,workers=1"))) }, ClusterOption { name: WorkloadClass, value: Some(Value(String("production"))) }], features: [], if_not_exists: false }) parse-statement ALTER CLUSTER cluster SET (SIZE 'scale=1,workers=1') @@ -2396,103 +2417,117 @@ error: Expected dot, found EOF CREATE CLUSTER REPLICA replica ^ +parse-statement +CREATE CLUSTER REPLICA IF NOT EXISTS default.replica SIZE 'small' +---- +CREATE CLUSTER REPLICA IF NOT EXISTS default.replica (SIZE = 'small') +=> +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }] }, if_not_exists: true }) + +parse-statement +CREATE CLUSTER REPLICA IF EXISTS default.replica SIZE 'small' +---- +error: Expected NOT, found EXISTS +CREATE CLUSTER REPLICA IF EXISTS default.replica SIZE 'small' + ^ + parse-statement CREATE CLUSTER REPLICA default.replica SIZE 'small' ---- CREATE CLUSTER REPLICA default.replica (SIZE = 'small') => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }] }, if_not_exists: false }) parse-statement CREATE CLUSTER REPLICA default.replica SIZE 'small', INTERNAL, BILLED AS 'free' ---- CREATE CLUSTER REPLICA default.replica (SIZE = 'small', INTERNAL, BILLED AS = 'free') => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: Internal, value: None }, ReplicaOption { name: BilledAs, value: Some(Value(String("free"))) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: Internal, value: None }, ReplicaOption { name: BilledAs, value: Some(Value(String("free"))) }] }, if_not_exists: false }) parse-statement CREATE CLUSTER REPLICA default.replica (SIZE 'small', INTERNAL, BILLED AS 'free') ---- CREATE CLUSTER REPLICA default.replica (SIZE = 'small', INTERNAL, BILLED AS = 'free') => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: Internal, value: None }, ReplicaOption { name: BilledAs, value: Some(Value(String("free"))) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: Internal, value: None }, ReplicaOption { name: BilledAs, value: Some(Value(String("free"))) }] }, if_not_exists: false }) parse-statement CREATE CLUSTER REPLICA default.replica SIZE 'small', INTERNAL = true, BILLED AS 'free' ---- CREATE CLUSTER REPLICA default.replica (SIZE = 'small', INTERNAL = true, BILLED AS = 'free') => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: Internal, value: Some(Value(Boolean(true))) }, ReplicaOption { name: BilledAs, value: Some(Value(String("free"))) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: Internal, value: Some(Value(Boolean(true))) }, ReplicaOption { name: BilledAs, value: Some(Value(String("free"))) }] }, if_not_exists: false }) parse-statement CREATE CLUSTER REPLICA default.replica SIZE 'small', INTERNAL = false, BILLED AS 'free' ---- CREATE CLUSTER REPLICA default.replica (SIZE = 'small', INTERNAL = false, BILLED AS = 'free') => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: Internal, value: Some(Value(Boolean(false))) }, ReplicaOption { name: BilledAs, value: Some(Value(String("free"))) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: Internal, value: Some(Value(Boolean(false))) }, ReplicaOption { name: BilledAs, value: Some(Value(String("free"))) }] }, if_not_exists: false }) parse-statement CREATE CLUSTER REPLICA default.replica SIZE 'small', BILLED AS 'free' ---- CREATE CLUSTER REPLICA default.replica (SIZE = 'small', BILLED AS = 'free') => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: BilledAs, value: Some(Value(String("free"))) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: BilledAs, value: Some(Value(String("free"))) }] }, if_not_exists: false }) parse-statement CREATE CLUSTER REPLICA default.replica SIZE 'small', AVAILABILITY ZONE 'a' ---- CREATE CLUSTER REPLICA default.replica (SIZE = 'small', AVAILABILITY ZONE = 'a') => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: AvailabilityZone, value: Some(Value(String("a"))) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: AvailabilityZone, value: Some(Value(String("a"))) }] }, if_not_exists: false }) parse-statement CREATE CLUSTER REPLICA default.replica SIZE 'small', AVAILABILITY ZONE 'a', DISK = false ---- CREATE CLUSTER REPLICA default.replica (SIZE = 'small', AVAILABILITY ZONE = 'a', DISK = false) => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: AvailabilityZone, value: Some(Value(String("a"))) }, ReplicaOption { name: Disk, value: Some(Value(Boolean(false))) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: AvailabilityZone, value: Some(Value(String("a"))) }, ReplicaOption { name: Disk, value: Some(Value(Boolean(false))) }] }, if_not_exists: false }) parse-statement CREATE CLUSTER REPLICA default.replica AVAILABILITY ZONE 'a', AVAILABILITY ZONE 'b' ---- CREATE CLUSTER REPLICA default.replica (AVAILABILITY ZONE = 'a', AVAILABILITY ZONE = 'b') => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: AvailabilityZone, value: Some(Value(String("a"))) }, ReplicaOption { name: AvailabilityZone, value: Some(Value(String("b"))) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: AvailabilityZone, value: Some(Value(String("a"))) }, ReplicaOption { name: AvailabilityZone, value: Some(Value(String("b"))) }] }, if_not_exists: false }) parse-statement CREATE CLUSTER REPLICA default.replica SIZE 'small', AVAILABILITY ZONE 'a' ---- CREATE CLUSTER REPLICA default.replica (SIZE = 'small', AVAILABILITY ZONE = 'a') => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: AvailabilityZone, value: Some(Value(String("a"))) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: AvailabilityZone, value: Some(Value(String("a"))) }] }, if_not_exists: false }) parse-statement CREATE CLUSTER REPLICA default.replica SIZE 'small', INTROSPECTION INTERVAL '1s', INTROSPECTION DEBUGGING = false ---- CREATE CLUSTER REPLICA default.replica (SIZE = 'small', INTROSPECTION INTERVAL = '1s', INTROSPECTION DEBUGGING = false) => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: IntrospectionInterval, value: Some(Value(String("1s"))) }, ReplicaOption { name: IntrospectionDebugging, value: Some(Value(Boolean(false))) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: Size, value: Some(Value(String("small"))) }, ReplicaOption { name: IntrospectionInterval, value: Some(Value(String("1s"))) }, ReplicaOption { name: IntrospectionDebugging, value: Some(Value(Boolean(false))) }] }, if_not_exists: false }) parse-statement CREATE CLUSTER REPLICA default.replica INTROSPECTION INTERVAL = 0, SIZE 'small' ---- CREATE CLUSTER REPLICA default.replica (INTROSPECTION INTERVAL = 0, SIZE = 'small') => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: IntrospectionInterval, value: Some(Value(Number("0"))) }, ReplicaOption { name: Size, value: Some(Value(String("small"))) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: IntrospectionInterval, value: Some(Value(Number("0"))) }, ReplicaOption { name: Size, value: Some(Value(String("small"))) }] }, if_not_exists: false }) parse-statement CREATE CLUSTER REPLICA default.replica INTROSPECTION INTERVAL NULL ---- CREATE CLUSTER REPLICA default.replica (INTROSPECTION INTERVAL = NULL) => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: IntrospectionInterval, value: Some(Value(Null)) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: IntrospectionInterval, value: Some(Value(Null)) }] }, if_not_exists: false }) parse-statement CREATE CLUSTER REPLICA default.replica STORAGECTL ADDRESSES ('1', '2'), COMPUTECTL ADDRESSES ('1', '2'), COMPUTE ADDRESSES ('3', '4'), WORKERS 2, INTROSPECTION INTERVAL = NULL ---- CREATE CLUSTER REPLICA default.replica (STORAGECTL ADDRESSES = ('1', '2'), COMPUTECTL ADDRESSES = ('1', '2'), COMPUTE ADDRESSES = ('3', '4'), WORKERS = 2, INTROSPECTION INTERVAL = NULL) => -CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: StoragectlAddresses, value: Some(Sequence([Value(String("1")), Value(String("2"))])) }, ReplicaOption { name: ComputectlAddresses, value: Some(Sequence([Value(String("1")), Value(String("2"))])) }, ReplicaOption { name: ComputeAddresses, value: Some(Sequence([Value(String("3")), Value(String("4"))])) }, ReplicaOption { name: Workers, value: Some(Value(Number("2"))) }, ReplicaOption { name: IntrospectionInterval, value: Some(Value(Null)) }] } }) +CreateClusterReplica(CreateClusterReplicaStatement { of_cluster: Ident("default"), definition: ReplicaDefinition { name: Ident("replica"), options: [ReplicaOption { name: StoragectlAddresses, value: Some(Sequence([Value(String("1")), Value(String("2"))])) }, ReplicaOption { name: ComputectlAddresses, value: Some(Sequence([Value(String("1")), Value(String("2"))])) }, ReplicaOption { name: ComputeAddresses, value: Some(Sequence([Value(String("3")), Value(String("4"))])) }, ReplicaOption { name: Workers, value: Some(Value(Number("2"))) }, ReplicaOption { name: IntrospectionInterval, value: Some(Value(Null)) }] }, if_not_exists: false }) parse-statement diff --git a/src/sql-pretty/src/doc.rs b/src/sql-pretty/src/doc.rs index 02c69af4ef4d9..a8741930b7a65 100644 --- a/src/sql-pretty/src/doc.rs +++ b/src/sql-pretty/src/doc.rs @@ -438,8 +438,12 @@ impl Pretty { ) -> RcDoc<'a> { let mut docs = Vec::new(); - // CREATE CLUSTER name - docs.push(nest_title("CREATE CLUSTER", self.doc_display_pass(&v.name))); + // CREATE CLUSTER [IF NOT EXISTS] name + let mut title = "CREATE CLUSTER".to_string(); + if v.if_not_exists { + title.push_str(" IF NOT EXISTS"); + } + docs.push(nest_title(&title, self.doc_display_pass(&v.name))); // OPTIONS (...) if !v.options.is_empty() { @@ -468,13 +472,17 @@ impl Pretty { ) -> RcDoc<'a> { let mut docs = Vec::new(); - // CREATE CLUSTER REPLICA cluster.replica + // CREATE CLUSTER REPLICA [IF NOT EXISTS] cluster.replica + let mut title = "CREATE CLUSTER REPLICA".to_string(); + if v.if_not_exists { + title.push_str(" IF NOT EXISTS"); + } let replica_name = RcDoc::concat([ self.doc_display_pass(&v.of_cluster), RcDoc::text("."), self.doc_display_pass(&v.definition.name), ]); - docs.push(nest_title("CREATE CLUSTER REPLICA", replica_name)); + docs.push(nest_title(&title, replica_name)); // OPTIONS (...) docs.push(bracket( diff --git a/src/sql/src/plan.rs b/src/sql/src/plan.rs index 69b62c5b552f1..311209e13164e 100644 --- a/src/sql/src/plan.rs +++ b/src/sql/src/plan.rs @@ -566,6 +566,7 @@ pub struct CreateClusterPlan { pub name: String, pub variant: CreateClusterVariant, pub workload_class: Option, + pub if_not_exists: bool, } #[derive(Debug, PartialEq, Eq, Clone)] @@ -597,6 +598,7 @@ pub struct CreateClusterReplicaPlan { pub cluster_id: ClusterId, pub name: String, pub config: ReplicaConfig, + pub if_not_exists: bool, } /// Configuration of introspection for a cluster replica. diff --git a/src/sql/src/plan/statement/ddl.rs b/src/sql/src/plan/statement/ddl.rs index 7ef6e2b2fcf8e..19f7073711fdc 100644 --- a/src/sql/src/plan/statement/ddl.rs +++ b/src/sql/src/plan/statement/ddl.rs @@ -4936,6 +4936,7 @@ pub fn plan_create_cluster_inner( name, options, features, + if_not_exists, }: CreateClusterStatement, ) -> Result { let ClusterOptionExtracted { @@ -5072,6 +5073,7 @@ pub fn plan_create_cluster_inner( auto_scaling_strategy, }), workload_class, + if_not_exists, }) } else { let Some(replica_defs) = replicas else { @@ -5119,19 +5121,24 @@ pub fn plan_create_cluster_inner( name: normalize::ident(name), variant: CreateClusterVariant::Unmanaged(CreateClusterUnmanagedPlan { replicas }), workload_class, + if_not_exists, }) } } /// Convert a [`CreateClusterPlan`] into a [`CreateClusterStatement`]. /// -/// The reverse of [`plan_create_cluster`]. +/// The reverse of [`plan_create_cluster`], so this renders `IF NOT EXISTS` +/// when the plan carries it. A caller that wants the cluster's canonical +/// definition rather than a faithful reverse must pass `if_not_exists: false`, +/// which is what `Cluster::try_to_plan` produces for `SHOW CREATE CLUSTER`. pub fn unplan_create_cluster( scx: &StatementContext, CreateClusterPlan { name, variant, workload_class, + if_not_exists, }: CreateClusterPlan, ) -> Result, PlanError> { match variant { @@ -5232,6 +5239,7 @@ pub fn unplan_create_cluster( name, options, features, + if_not_exists, }) } CreateClusterVariant::Unmanaged(_) => { @@ -5561,6 +5569,7 @@ pub fn plan_create_cluster_replica( CreateClusterReplicaStatement { definition: ReplicaDefinition { name, options }, of_cluster, + if_not_exists, }: CreateClusterReplicaStatement, ) -> Result { let cluster = scx @@ -5581,6 +5590,7 @@ pub fn plan_create_cluster_replica( name: normalize::ident(name), cluster_id: cluster.id(), config, + if_not_exists, })) } diff --git a/src/sql/src/rbac.rs b/src/sql/src/rbac.rs index 8b9c8cd826c91..47dc2e27ecf91 100644 --- a/src/sql/src/rbac.rs +++ b/src/sql/src/rbac.rs @@ -532,6 +532,7 @@ fn generate_rbac_requirements( name: _, variant: _, workload_class: _, + if_not_exists: _, }) => RbacRequirements { privileges: vec![(SystemObjectId::System, AclMode::CREATE_CLUSTER, role_id)], item_usage: &CREATE_ITEM_USAGE, @@ -541,6 +542,7 @@ fn generate_rbac_requirements( cluster_id, name: _, config: _, + if_not_exists: _, }) => RbacRequirements { ownership: vec![ObjectId::Cluster(*cluster_id)], item_usage: &CREATE_ITEM_USAGE, diff --git a/test/pgtest-mz/notice.pt b/test/pgtest-mz/notice.pt index 06302fa6ab27d..5aec742896123 100644 --- a/test/pgtest-mz/notice.pt +++ b/test/pgtest-mz/notice.pt @@ -505,3 +505,36 @@ ReadyForQuery {"status":"I"} NoticeResponse {"fields":[{"typ":"S","value":"NOTICE"},{"typ":"C","value":"00000"},{"typ":"M","value":"drop cascades to 2 other objects"}]} CommandComplete {"tag":"DROP TABLE"} ReadyForQuery {"status":"I"} + +# Cluster case: `CREATE CLUSTER IF NOT EXISTS` for a name that is already taken +# skips creation and reports success with an "already exists, skipping" notice. +# `CREATE CLUSTER REPLICA IF NOT EXISTS` behaves the same for a replica name +# already used on the target cluster. +send +Query {"query": "CREATE CLUSTER nc REPLICAS (r1 (SIZE = 'scale=1,workers=1'))"} +Query {"query": "CREATE CLUSTER IF NOT EXISTS nc REPLICAS ()"} +Query {"query": "CREATE CLUSTER REPLICA IF NOT EXISTS nc.r1 SIZE = 'scale=1,workers=1'"} +Query {"query": "CREATE CLUSTER REPLICA IF NOT EXISTS nc.r2 SIZE = 'scale=1,workers=1'"} +Query {"query": "DROP CLUSTER nc"} +---- + +until err_field_typs=SCM +ReadyForQuery +ReadyForQuery +ReadyForQuery +ReadyForQuery +ReadyForQuery +---- +CommandComplete {"tag":"CREATE CLUSTER"} +ReadyForQuery {"status":"I"} +NoticeResponse {"fields":[{"typ":"S","value":"NOTICE"},{"typ":"C","value":"42710"},{"typ":"M","value":"cluster \"nc\" already exists, skipping"}]} +CommandComplete {"tag":"CREATE CLUSTER"} +ReadyForQuery {"status":"I"} +NoticeResponse {"fields":[{"typ":"S","value":"NOTICE"},{"typ":"C","value":"42710"},{"typ":"M","value":"cluster replica \"nc.r1\" already exists, skipping"}]} +CommandComplete {"tag":"CREATE CLUSTER REPLICA"} +ReadyForQuery {"status":"I"} +CommandComplete {"tag":"CREATE CLUSTER REPLICA"} +ReadyForQuery {"status":"I"} +NoticeResponse {"fields":[{"typ":"S","value":"NOTICE"},{"typ":"C","value":"00000"},{"typ":"M","value":"drop cascades to 2 other objects"}]} +CommandComplete {"tag":"DROP CLUSTER"} +ReadyForQuery {"status":"I"} diff --git a/test/sqllogictest/cluster.slt b/test/sqllogictest/cluster.slt index 6f062878b8577..5766418e810a0 100644 --- a/test/sqllogictest/cluster.slt +++ b/test/sqllogictest/cluster.slt @@ -1041,3 +1041,153 @@ u3 NULL statement ok DROP CLUSTER c1 + +# `IF NOT EXISTS` + +statement ok +CREATE CLUSTER ine1 (SIZE 'scale=1,workers=1') + +# Skipping leaves the existing cluster untouched: the configuration in the +# statement is not applied. +statement ok +CREATE CLUSTER IF NOT EXISTS ine1 (SIZE 'scale=1,workers=2', REPLICATION FACTOR 2) + +query TIT +SELECT name, replication_factor, size FROM mz_clusters WHERE name = 'ine1' +---- +ine1 1 scale=1,workers=1 + +# The statement is validated in full before the collision is reported, exactly +# as it is without `IF NOT EXISTS`. A typo must not hide behind the clause. +statement error unknown cluster replica size +CREATE CLUSTER IF NOT EXISTS ine1 (SIZE 'lol') + +statement error cluster 'ine1' already exists +CREATE CLUSTER ine1 (SIZE 'scale=1,workers=2') + +# The unmanaged variant takes the clause too. +statement ok +CREATE CLUSTER ineu REPLICAS (r1 (SIZE 'scale=1,workers=1')) + +statement ok +CREATE CLUSTER IF NOT EXISTS ineu REPLICAS (r2 (SIZE 'scale=1,workers=1')) + +query T rowsort +SELECT r.name +FROM mz_cluster_replicas r JOIN mz_clusters c ON c.id = r.cluster_id +WHERE c.name = 'ineu' +---- +r1 + +statement ok +DROP CLUSTER ineu + +# `IF NOT EXISTS` still creates the cluster when the name is free. +statement ok +CREATE CLUSTER IF NOT EXISTS ine3 (SIZE 'scale=1,workers=1') + +query T +SELECT name FROM mz_clusters WHERE name = 'ine3' +---- +ine3 + +# Cluster names are case sensitive, so an unquoted `INE3` is `ine3` but a +# quoted `INE3` is a distinct, free name. +statement ok +CREATE CLUSTER IF NOT EXISTS INE3 (SIZE 'scale=1,workers=1') + +statement ok +CREATE CLUSTER IF NOT EXISTS "INE3" (SIZE 'scale=1,workers=1') + +query T rowsort +SELECT name FROM mz_clusters WHERE name ILIKE 'ine3' +---- +INE3 +ine3 + +# `FEATURES` is system-user only, and it is the shape with the most plan/unplan +# roundtrip surface, so cover it with the clause too. +simple conn=mz_system,user=mz_system +CREATE CLUSTER IF NOT EXISTS inef (SIZE 'scale=1,workers=1') FEATURES (ENABLE EAGER DELTA JOINS = true) +---- +COMPLETE 0 + +simple conn=mz_system,user=mz_system +CREATE CLUSTER IF NOT EXISTS inef (SIZE 'scale=1,workers=2') FEATURES (ENABLE EAGER DELTA JOINS = false) +---- +COMPLETE 0 + +query TT +SHOW CREATE CLUSTER inef +---- +inef CREATE␠CLUSTER␠"inef"␠(EXPERIMENTAL␠ARRANGEMENT␠COMPRESSION␠=␠false,␠INTROSPECTION␠DEBUGGING␠=␠false,␠INTROSPECTION␠INTERVAL␠=␠INTERVAL␠'00:00:01',␠MANAGED␠=␠true,␠REPLICATION␠FACTOR␠=␠1,␠SIZE␠=␠'scale=1,workers=1',␠SCHEDULE␠=␠MANUAL)␠FEATURES␠(ENABLE␠EAGER␠DELTA␠JOINS␠=␠true) + +simple conn=mz_system,user=mz_system +DROP CLUSTER inef +---- +COMPLETE 0 + +# `IF NOT EXISTS` is not part of the cluster's definition, so it does not show +# up here. +query TT +SHOW CREATE CLUSTER ine3 +---- +ine3 CREATE␠CLUSTER␠"ine3"␠(EXPERIMENTAL␠ARRANGEMENT␠COMPRESSION␠=␠false,␠INTROSPECTION␠DEBUGGING␠=␠false,␠INTROSPECTION␠INTERVAL␠=␠INTERVAL␠'00:00:01',␠MANAGED␠=␠true,␠REPLICATION␠FACTOR␠=␠1,␠SIZE␠=␠'scale=1,workers=1',␠SCHEDULE␠=␠MANUAL) + +# A reserved name is rejected before the name is ever checked for a collision, +# so `IF NOT EXISTS` must report the reserved name rather than a skip. +statement error cluster name "mz_system" is reserved +CREATE CLUSTER IF NOT EXISTS mz_system (SIZE 'scale=1,workers=1') + +statement error cluster name "mz_nope" is reserved +CREATE CLUSTER IF NOT EXISTS mz_nope (SIZE 'scale=1,workers=1') + +statement ok +DROP CLUSTER ine1 + +statement ok +DROP CLUSTER ine3 + +statement ok +DROP CLUSTER "INE3" + +# `IF NOT EXISTS` for cluster replicas. + +statement ok +CREATE CLUSTER iner REPLICAS (r1 (SIZE 'scale=1,workers=1')) + +statement error cannot create multiple replicas named 'r1' on cluster 'iner' +CREATE CLUSTER REPLICA iner.r1 SIZE 'scale=1,workers=2' + +statement ok +CREATE CLUSTER REPLICA IF NOT EXISTS iner.r1 SIZE 'scale=1,workers=2' + +statement ok +CREATE CLUSTER REPLICA IF NOT EXISTS iner.r2 SIZE 'scale=1,workers=1' + +query TT rowsort +SELECT r.name, r.size +FROM mz_cluster_replicas r JOIN mz_clusters c ON c.id = r.cluster_id +WHERE c.name = 'iner' +---- +r1 scale=1,workers=1 +r2 scale=1,workers=1 + +# `IF NOT EXISTS` covers the replica name only, the cluster must exist. +statement error unknown cluster 'nope' +CREATE CLUSTER REPLICA IF NOT EXISTS nope.r1 SIZE 'scale=1,workers=1' + +# Replicas cannot be added to a managed cluster at all, and `IF NOT EXISTS` +# does not soften that: the rejection happens at plan time, before the replica +# name is ever considered. +statement ok +CREATE CLUSTER inem (SIZE 'scale=1,workers=1') + +statement error cannot modify managed cluster inem +CREATE CLUSTER REPLICA IF NOT EXISTS inem.r1 SIZE 'scale=1,workers=1' + +statement ok +DROP CLUSTER inem + +statement ok +DROP CLUSTER iner From b5624d20bb818e0f6bd67c0633eaea75d22fe608 Mon Sep 17 00:00:00 2001 From: Aljoscha Krettek Date: Thu, 6 Aug 2026 10:50:41 +0200 Subject: [PATCH 2/3] adapter: let the group committer write at a caller-chosen timestamp (#37921) ### Motivation Part 4 of 7 in a stack that moves `DELETE`, `UPDATE` and `INSERT ... SELECT` off the coordinator onto the session task, using optimistic concurrency control. Design doc: [`20260210_incremental_occ_read_then_write.md`](https://github.com/MaterializeInc/materialize/blob/aljoscha/occ-06-occ-path/doc/developer/design/20260210_incremental_occ_read_then_write.md) (lands in part 6). The group committer picks a write timestamp from the oracle, which is right for a write whose diffs are valid whenever they land. A write computed against a snapshot is not like that. Its diffs are only correct at the frontier they were read at, so the timestamp has to come from the caller and the committer has to be able to refuse it. Closes [SQL-590](https://linear.app/materializeinc/issue/SQL-590) ### Description `TableWriteCmd::TimestampedWrite` carries a caller-chosen timestamp and gets exactly one attempt. An `InvalidUppers` means another writer holds the upper at or past that timestamp, so the snapshot the diffs were computed from is stale, and the answer is `TimestampPassed`. Never a retry at a fresh timestamp: retrying stale diffs later is precisely the lost update this exists to prevent. Deciding what to do about a conflict is the caller's job, and it needs a new snapshot to do it. A concurrent `ALTER TABLE` is caught by a separate guard on the blind path, not here. `UserWriteResponder::Internal` names the generation its diffs were computed against in `WriteTarget`, and `stage_group_commit` answers `TargetChanged` if the table's latest `GlobalId` has moved on. That check runs on the coordinator loop rather than in the committer. `TimestampedWriteRequest` carries pre-resolved `GlobalId`s and has no such check. It relies on its submitter resolving them against the current catalog on the coordinator loop and sending with no await in between, so channel order keeps the append ahead of any `Forget` for the same table. Part 6's `handle_attempt_write` is that submitter. The struct's doc now states this as a contract, because a non-empty append for a table whose write handle is gone trips an assert in the storage controller. Both txns-shard write paths now share `attempt_write_to_txns`, so the catalog upper advance, the metric, the `InvalidUppers` classification and the oracle apply exist once. The two differ only in what they do with a conflict, which is the difference worth seeing. The result travels back through a responder whose `Drop` reports `Indeterminate`. Without that, a dropped sender on shutdown would panic the waiting session task instead of failing its statement. **Temporary scaffolding.** Nothing calls this until part 6, so a `dead_code` allowance on the unreached items keeps the crate warning-free. Part 6 removes it in the same commit that adds the callers. ### Verification No caller yet, so this part is carried by the tests in parts 6 and 7, and by existing coverage showing the shared `attempt_write_to_txns` did not change ordinary group commit behavior. --- src/adapter/src/coord.rs | 2 + src/adapter/src/coord/appends.rs | 482 +++++++++++++++++++---- src/adapter/src/coord/message_handler.rs | 6 + src/adapter/src/session.rs | 5 + 4 files changed, 418 insertions(+), 77 deletions(-) diff --git a/src/adapter/src/coord.rs b/src/adapter/src/coord.rs index 6356d9d3ed0b9..66650756c64ef 100644 --- a/src/adapter/src/coord.rs +++ b/src/adapter/src/coord.rs @@ -362,6 +362,8 @@ pub enum Message { responses: Vec, /// Statement executions associated with this commit. statement_logging_ids: Vec, + /// Frontend-sequenced writes to complete after local timestamp bookkeeping. + internal_results: Vec, /// The applied write timestamp. write_ts: Timestamp, }, diff --git a/src/adapter/src/coord/appends.rs b/src/adapter/src/coord/appends.rs index a0c482c179938..732e50fd1ff3d 100644 --- a/src/adapter/src/coord/appends.rs +++ b/src/adapter/src/coord/appends.rs @@ -164,12 +164,98 @@ pub(crate) enum BuiltinTableUpdateSource { Background(oneshot::Sender<()>), } +/// Result of a write submitted by frontend sequencing. +// The read-then-write path that submits these writes lands later in this +// stack, this attribute goes away with it. +#[allow(dead_code)] +#[derive(Debug, Clone)] +pub enum WriteResult { + /// The write committed at this timestamp. + Success { timestamp: Timestamp }, + /// The requested timestamp was no longer eligible. + TimestampPassed { + target_timestamp: Timestamp, + next_eligible_timestamp: Timestamp, + }, + /// The write was canceled before it entered the committer. + Canceled, + /// The coordinator cannot accept writes. + ReadOnly, + /// The target table was dropped or changed after planning. + TargetChanged, + /// The committer shut down with the write's outcome unknown. + Indeterminate, +} + +/// Delivers an internal write result, including on task shutdown. +/// +/// The `Drop` impl is load-bearing. The session task waiting on the other end +/// `expect`s a reply, so a `oneshot::Sender` that is dropped silently would +/// panic that session on coordinator shutdown or on a dead group-committer +/// task. Reporting [`WriteResult::Indeterminate`] instead lets the session +/// report an error. +#[derive(Debug)] +pub struct InternalWriteResponder { + tx: Option>, +} + +impl InternalWriteResponder { + // The read-then-write path that uses this lands later in this stack, this + // attribute goes away with it. + #[allow(dead_code)] + pub(crate) fn new(tx: oneshot::Sender) -> Self { + Self { tx: Some(tx) } + } + + pub(crate) fn send(mut self, result: WriteResult) { + if let Some(tx) = self.tx.take() { + let _ = tx.send(result); + } + } +} + +impl Drop for InternalWriteResponder { + fn drop(&mut self) { + if let Some(tx) = self.tx.take() { + let _ = tx.send(WriteResult::Indeterminate); + } + } +} + /// Where to deliver the result of a [`PendingWriteTxn::User`] write. #[derive(Debug)] pub(crate) enum UserWriteResponder { /// Session-bound write. The coordinator retires the session's /// `ExecuteContext` once the write commits. Session(PendingTxn), + /// Frontend-sequenced blind write. + // The read-then-write path that uses this lands later in this stack, this + // attribute goes away with it. + #[allow(dead_code)] + Internal { + conn_id: ConnectionId, + /// The table the diffs were computed against, item id and the generation + /// current at that time. Group commit refuses the write if the table's + /// latest generation has moved on. + target: WriteTarget, + result: InternalWriteResponder, + }, +} + +/// A write's target table, pinned to one generation of it. +#[derive(Debug, Clone, Copy)] +pub(crate) struct WriteTarget { + pub(crate) item_id: CatalogItemId, + pub(crate) global_id: GlobalId, +} + +impl UserWriteResponder { + pub(crate) fn conn_id(&self) -> &ConnectionId { + match self { + UserWriteResponder::Session(pending) => pending.ctx.session().conn_id(), + UserWriteResponder::Internal { conn_id, .. } => conn_id, + } + } } /// A pending write transaction that will be committing during the next group commit. @@ -209,6 +295,10 @@ impl PendingWriteTxn { pub(crate) enum TableWriteCmd { GroupCommit(GroupCommitRequest), + // The read-then-write path that uses this lands later in this stack, this + // attribute goes away with it. + #[allow(dead_code)] + TimestampedWrite(TimestampedWriteRequest), Register { tables: Vec, result: oneshot::Sender, @@ -219,6 +309,21 @@ pub(crate) enum TableWriteCmd { }, } +/// An OCC write whose diffs are valid only at `target_timestamp`. +/// +/// The `GlobalId`s reach the table write worker unvalidated, so a submitter +/// must resolve them against the current catalog on the coordinator loop and +/// send with no await in between. Channel order then keeps the append ahead of +/// any `Forget` for the same table. A non-empty append for a table whose write +/// handle is already gone trips an assert in the storage controller and takes +/// the process down. +pub(crate) struct TimestampedWriteRequest { + pub(crate) appends: Vec<(GlobalId, Vec)>, + pub(crate) target_timestamp: Timestamp, + pub(crate) result: InternalWriteResponder, + pub(crate) span: Span, +} + /// A group commit staged on the coordinator loop for the [`GroupCommitter`]. pub(crate) struct GroupCommitRequest { /// Appends resolved to their latest [`GlobalId`]. Empty for a keepalive. @@ -226,6 +331,7 @@ pub(crate) struct GroupCommitRequest { responses: Vec, statement_logging_ids: Vec, notifies: Vec>, + internal_results: Vec, write_locks: GroupCommitWriteLocks, /// In-progress permits held until the commit is applied. permits: Vec, @@ -240,6 +346,7 @@ impl GroupCommitRequest { responses, statement_logging_ids, notifies, + internal_results, write_locks, permits, contains_internal_system_write, @@ -249,6 +356,7 @@ impl GroupCommitRequest { self.responses.extend(responses); self.statement_logging_ids.extend(statement_logging_ids); self.notifies.extend(notifies); + self.internal_results.extend(internal_results); self.write_locks.extend(write_locks); self.permits.extend(permits); self.contains_internal_system_write |= contains_internal_system_write; @@ -270,6 +378,17 @@ pub(crate) struct GroupCommitter { max_attempts: ConfigValHandle, } +/// Outcome of one txns-shard write attempt. +enum TxnsWriteAttempt { + /// The write landed and the oracle has applied its timestamp. + Applied, + /// Another writer holds the upper at or past the attempted timestamp. The + /// write did not land. + UpperConflict, + /// The table write worker is gone, so the outcome is unknown. + WorkerGone, +} + impl GroupCommitter { async fn run(mut self) { while let Some(cmd) = self.rx.recv().await { @@ -285,6 +404,17 @@ impl GroupCommitter { ControlFlow::Break(()) => return, } } + TableWriteCmd::TimestampedWrite(request) => { + let span = request.span.clone(); + if self + .commit_timestamped(request) + .instrument(span) + .await + .is_break() + { + return; + } + } TableWriteCmd::Register { tables, result } => { let Some(write_ts) = self .write_to_txns(None, |ts, _advance_to| { @@ -312,7 +442,85 @@ impl GroupCommitter { } } - /// Writes at a fresh oracle timestamp and applies a successful write to the oracle. + /// Attempts an OCC write exactly once at its requested timestamp. + /// + /// A conflict is reported as `TimestampPassed` and is the caller's to + /// resolve with a new snapshot. Retrying the same diffs at a fresh + /// timestamp would apply a mutation to state it was not computed from. + /// + /// What [`Self::commit`] does that this skips, and why that is safe: + /// + /// * The wall-clock throttle. `target_timestamp` is the caller's to choose, + /// and it must not run the write timeline ahead of the clock. + /// * A [`GroupCommitPermit`]. The caller bounds how many of these are in + /// flight, and that is the backpressure for this path. + /// * Merging queued commits. There is nothing to merge into: these diffs + /// are valid at this one timestamp, so they cannot share a timestamp with + /// another write. + /// * Write locks. The point of OCC is to detect a conflicting write after + /// the fact, through the timestamp, rather than to exclude it. + /// + /// `Break` means the table worker shut down. + async fn commit_timestamped(&self, request: TimestampedWriteRequest) -> ControlFlow<(), ()> { + let TimestampedWriteRequest { + appends, + target_timestamp, + result, + span: _, + } = request; + + let oracle_write_ts = self.oracle.peek_write_ts().await; + if target_timestamp <= oracle_write_ts { + result.send(WriteResult::TimestampPassed { + target_timestamp, + next_eligible_timestamp: oracle_write_ts.step_forward(), + }); + return ControlFlow::Continue(()); + } + + let write_ts = WriteTimestamp { + timestamp: target_timestamp, + advance_to: target_timestamp.step_forward(), + }; + match self + .attempt_write_to_txns( + &write_ts, + Some(&self.metrics.append_table_duration_seconds), + |ts, advance_to| self.table_write_handle.append(ts, advance_to, appends), + ) + .await + { + TxnsWriteAttempt::Applied => {} + TxnsWriteAttempt::UpperConflict => { + result.send(WriteResult::TimestampPassed { + target_timestamp, + next_eligible_timestamp: write_ts.advance_to, + }); + return ControlFlow::Continue(()); + } + TxnsWriteAttempt::WorkerGone => { + warn!("table write worker gone with a timestamped write outstanding"); + return ControlFlow::Break(()); + } + } + + if self + .internal_cmd_tx + .send(Message::GroupCommitApplied { + responses: Vec::new(), + statement_logging_ids: Vec::new(), + internal_results: vec![result], + write_ts: target_timestamp, + }) + .is_err() + { + warn!("coordinator shut down before a timestamped write could be finalized"); + } + ControlFlow::Continue(()) + } + + /// Writes at a fresh oracle timestamp, retrying an upper conflict at a new + /// timestamp, and applies a successful write to the oracle. /// /// Returns `None` when the table write worker shuts down. async fn write_to_txns( @@ -322,7 +530,7 @@ impl GroupCommitter { ) -> Option { // Persistent conflicts indicate an unexpected writer. Halt instead of spinning forever. let mut attempt = 0; - let write_ts = loop { + loop { let max_attempts = self.max_attempts.get().max(1); if attempt >= max_attempts { halt!( @@ -332,26 +540,16 @@ impl GroupCommitter { attempt += 1; let write_ts = self.oracle.write_ts().await; - // A post-fence retry has an advance frontier above this handle's stale upper, so this - // reaches Persist and observes the fence. - let catalog_upper_start = Instant::now(); - self.catalog_upper - .advance_upper(write_ts.advance_to) + // A post-fence retry has an advance frontier above this handle's stale upper, so the + // advance inside reaches Persist and observes the fence. + match self + .attempt_write_to_txns(&write_ts, op_duration_metric, |ts, advance_to| { + op(ts, advance_to) + }) .await - .unwrap_or_terminate("unable to advance catalog upper"); - self.metrics - .group_commit_catalog_upper_seconds - .observe(catalog_upper_start.elapsed().as_secs_f64()); - - let op_start = Instant::now(); - let op_res = op(write_ts.timestamp, write_ts.advance_to).await; - if let Some(metric) = op_duration_metric { - metric.observe(op_start.elapsed().as_secs_f64()); - } - - match op_res { - Ok(Ok(())) => break write_ts, - Ok(Err(StorageError::InvalidUppers(_))) => { + { + TxnsWriteAttempt::Applied => return Some(write_ts), + TxnsWriteAttempt::UpperConflict => { warn!( write_ts = %write_ts.timestamp, attempt, @@ -359,24 +557,57 @@ impl GroupCommitter { ); continue; } - Ok(Err(other)) => { - Err::<(), _>(other).unwrap_or_terminate("cannot fail to write to txns shard"); - unreachable!("unwrap_or_terminate does not return on Err"); - } - Err(_recv) => { + TxnsWriteAttempt::WorkerGone => { // The outcome is indeterminate. Stop before processing more writes. warn!("table write worker gone (process shutting down), winding down"); return None; } } - }; + } + } + + /// Runs `op` against the txns shard once, at `write_ts`. + /// + /// Advancing the catalog upper first keeps the catalog readable at the + /// oracle read timestamp. A write that lands is applied to the oracle + /// before this returns. + async fn attempt_write_to_txns( + &self, + write_ts: &WriteTimestamp, + op_duration_metric: Option<&prometheus::Histogram>, + op: impl FnOnce(Timestamp, Timestamp) -> oneshot::Receiver>, + ) -> TxnsWriteAttempt { + let catalog_upper_start = Instant::now(); + self.catalog_upper + .advance_upper(write_ts.advance_to) + .await + .unwrap_or_terminate("unable to advance catalog upper"); + self.metrics + .group_commit_catalog_upper_seconds + .observe(catalog_upper_start.elapsed().as_secs_f64()); + + let op_start = Instant::now(); + let op_res = op(write_ts.timestamp, write_ts.advance_to).await; + if let Some(metric) = op_duration_metric { + metric.observe(op_start.elapsed().as_secs_f64()); + } + + match op_res { + Ok(Ok(())) => {} + Ok(Err(StorageError::InvalidUppers(_))) => return TxnsWriteAttempt::UpperConflict, + Ok(Err(other)) => { + Err::<(), _>(other).unwrap_or_terminate("cannot fail to write to txns shard"); + unreachable!("unwrap_or_terminate does not return on Err"); + } + Err(_recv) => return TxnsWriteAttempt::WorkerGone, + } let now: Timestamp = (self.now)().into(); crate::coord::timeline::check_runaway_write_ts(&now, write_ts.timestamp); self.oracle.apply_write(write_ts.timestamp).await; - Some(write_ts) + TxnsWriteAttempt::Applied } /// Applies a staged group commit. @@ -439,6 +670,7 @@ impl GroupCommitter { responses, statement_logging_ids, notifies, + internal_results, write_locks, permits, contains_internal_system_write: _, @@ -487,6 +719,7 @@ impl GroupCommitter { .send(Message::GroupCommitApplied { responses, statement_logging_ids, + internal_results, write_ts: timestamp, }) .is_err() @@ -640,90 +873,151 @@ impl Coordinator { // Validate, merge, and possibly acquire write locks for as many pending writes as possible. for pending_write in pending_writes { match pending_write { - // We always allow system writes to proceed. PendingWriteTxn::System { .. } => validated_writes.push(pending_write), - // We have a set of locks! Validate they're correct (expected). PendingWriteTxn::User { span, write_locks: Some(write_locks), writes, - responder: UserWriteResponder::Session(pending_txn), + responder, } => match write_locks.validate(writes.keys().copied()) { Ok(validated_locks) => { - // Merge all of our write locks together since we can allow concurrent - // writes at the same timestamp. + // Locks from different sessions can be merged into one + // group because every write in the group commits at the + // same timestamp. group_write_locks.merge(validated_locks); - - let validated_write = PendingWriteTxn::User { + validated_writes.push(PendingWriteTxn::User { span, writes, write_locks: None, - responder: UserWriteResponder::Session(pending_txn), - }; - validated_writes.push(validated_write); + responder, + }); } - // This is very unexpected since callers of this method should be validating. - // - // We cannot allow these write to occur since if the correct set of locks was - // not taken we could violate serializability. + // Callers validate before they get here, so a partial set is + // a bug. We must not let the write proceed: without the + // right locks it can violate serializability. Err(missing) => { let writes: Vec<_> = writes.keys().collect(); panic!( - "got to group commit with partial set of locks!\nmissing: {:?}, writes: {:?}, txn: {:?}", - missing, writes, pending_txn, + "got to group commit with partial set of locks!\nmissing: {:?}, writes: {:?}, conn_id: {}", + missing, + writes, + responder.conn_id(), ); } }, - // If we don't have any locks, try to acquire them, otherwise defer the write. + // Without handed-off locks, acquire just in time. On a miss a + // session write defers, an internal write re-queues. PendingWriteTxn::User { span, writes, write_locks: None, - responder: UserWriteResponder::Session(pending_txn), + responder, } => { let missing = group_write_locks.missing_locks(writes.keys().copied()); - if missing.is_empty() { - // We have all the locks! Queue the pending write. - let validated_write = PendingWriteTxn::User { + validated_writes.push(PendingWriteTxn::User { span, writes, write_locks: None, - responder: UserWriteResponder::Session(pending_txn), - }; - validated_writes.push(validated_write); - } else { - // Try to acquire the locks we're missing. - let mut just_in_time_locks = WriteLocks::builder(missing.clone()); - for collection in missing { - if let Some(lock) = self.try_grant_object_write_lock(collection) { - just_in_time_locks.insert_lock(collection, lock); + responder, + }); + continue; + } + + match responder { + UserWriteResponder::Session(pending_txn) => { + let mut just_in_time_locks = WriteLocks::builder(missing.clone()); + for collection in missing { + if let Some(lock) = self.try_grant_object_write_lock(collection) { + just_in_time_locks.insert_lock(collection, lock); + } + } + match just_in_time_locks + .all_or_nothing(pending_txn.ctx.session().conn_id()) + { + Ok(locks) => { + group_write_locks.merge(locks); + validated_writes.push(PendingWriteTxn::User { + span, + writes, + write_locks: None, + responder: UserWriteResponder::Session(pending_txn), + }); + } + Err(missing) => { + let acquire_future = + self.grant_object_write_lock(missing).map(Option::Some); + deferred_writes.push(( + acquire_future, + DeferredWrite { + span, + writes, + pending_txn, + }, + )); + } } } - - match just_in_time_locks.all_or_nothing(pending_txn.ctx.session().conn_id()) - { - // We acquired all of the locks! Proceed with the write. - Ok(locks) => { - group_write_locks.merge(locks); - let validated_write = PendingWriteTxn::User { + UserWriteResponder::Internal { + conn_id, + target, + result, + } => { + // All-or-nothing, like `WriteLocks::all_or_nothing` + // for session writes: `collect` into an `Option` + // drops every lock it did acquire as soon as one is + // unavailable. Holding a partial set across the + // re-queue below could deadlock against another + // writer holding the complement. + let acquired = missing + .into_iter() + .map(|id| { + self.try_grant_object_write_lock(id).map(|lock| (id, lock)) + }) + .collect::>>(); + if let Some(acquired) = acquired { + for (id, lock) in acquired { + group_write_locks.insert_lock(id, lock); + } + validated_writes.push(PendingWriteTxn::User { span, writes, write_locks: None, - responder: UserWriteResponder::Session(pending_txn), - }; - validated_writes.push(validated_write); - } - // Darn. We couldn't acquire the locks, defer the write. - Err(missing) => { - let acquire_future = - self.grant_object_write_lock(missing).map(Option::Some); - let write = DeferredWrite { + responder: UserWriteResponder::Internal { + conn_id, + target, + result, + }, + }); + } else { + // Retry by riding the next group commit + // initiate, at the latest the periodic + // timeline advancement tick. Internal writes + // have no `ExecuteContext`, so they can't use + // `defer_op` like session writes. + // + // Deliberately without `trigger_group_commit`. + // The lock is held by a writer that is not + // waiting on us, so an immediate retry would + // find it held, re-queue, and trigger again, + // spinning for as long as the holder keeps it. + // Waiting for a trigger someone else raises + // costs at most one tick and no CPU. + // + // Lock hold times are short while frontend OCC + // sequencing is enabled because the + // coordinator's lock-based read-then-write + // path is disabled. + self.pending_writes.push(PendingWriteTxn::User { span, writes, - pending_txn, - }; - deferred_writes.push((acquire_future, write)); + write_locks: None, + responder: UserWriteResponder::Internal { + conn_id, + target, + result, + }, + }); } } } @@ -744,6 +1038,7 @@ impl Coordinator { let mut responses = Vec::with_capacity(validated_writes.len()); let mut statement_logging_ids = Vec::new(); let mut notifies = Vec::new(); + let mut internal_results = Vec::new(); for validated_write_txn in validated_writes { match validated_write_txn { @@ -775,6 +1070,38 @@ impl Coordinator { responses.push(CompletedClientTransmitter::new(ctx, response, action)); } + PendingWriteTxn::User { + span: _, + writes, + write_locks, + responder: UserWriteResponder::Internal { target, result, .. }, + } => { + assert_none!(write_locks, "should have merged together all locks above"); + let current_global_id = self + .catalog() + .try_get_entry(&target.item_id) + .map(|entry| entry.latest_global_id()); + if current_global_id != Some(target.global_id) { + result.send(WriteResult::TargetChanged); + continue; + } + // A frontend write's data all belongs to `target`, which + // `handle_attempt_write` enforces by building `writes` with + // that single key. Folding it under `target.item_id` + // regardless would append to the wrong table, so check + // rather than trust the submitter. + assert!( + writes.keys().all(|id| *id == target.item_id), + "frontend write for {:?} carries other tables: {:?}", + target.item_id, + writes.keys().collect::>(), + ); + appends + .entry(target.item_id) + .or_default() + .extend(writes.into_values().flatten()); + internal_results.push(result); + } PendingWriteTxn::System { updates, source } => { for update in updates { appends.entry(update.id).or_default().push(update.data); @@ -821,6 +1148,7 @@ impl Coordinator { responses, statement_logging_ids, notifies, + internal_results, write_locks: group_write_locks, permits: permit.into_iter().collect(), contains_internal_system_write, diff --git a/src/adapter/src/coord/message_handler.rs b/src/adapter/src/coord/message_handler.rs index c9eb4e0e17c95..988000c1abfb4 100644 --- a/src/adapter/src/coord/message_handler.rs +++ b/src/adapter/src/coord/message_handler.rs @@ -114,6 +114,7 @@ impl Coordinator { Message::GroupCommitApplied { responses, statement_logging_ids, + internal_results, write_ts, } => { // Record statement timestamps before retiring, since retiring ends the statement @@ -130,6 +131,11 @@ impl Coordinator { // that and we can downgrade the local read holds without an oracle round trip. self.downgrade_local_read_holds(write_ts); self.advance_custom_timelines().boxed_local().await; + for result in internal_results { + result.send(crate::coord::appends::WriteResult::Success { + timestamp: write_ts, + }); + } } Message::AdvanceTimelines => { // Only sent by the periodic tick in read-only mode, where group commits (which diff --git a/src/adapter/src/session.rs b/src/adapter/src/session.rs index 09cf80b7ce5fc..428e82a1086be 100644 --- a/src/adapter/src/session.rs +++ b/src/adapter/src/session.rs @@ -1863,6 +1863,11 @@ impl GroupCommitWriteLocks { self.locks.extend(std::mem::take(&mut other.locks)); } + /// Inserts a single lock, keyed by the collection it guards. + pub fn insert_lock(&mut self, id: CatalogItemId, lock: tokio::sync::OwnedMutexGuard<()>) { + self.locks.insert(id, lock); + } + /// Returns the collections we're missing locks for, if any. pub fn missing_locks( &self, From bd299ae61e5a47e05283620db8a5aadfd68e0409 Mon Sep 17 00:00:00 2001 From: Petros Angelatos Date: Thu, 6 Aug 2026 12:50:09 +0300 Subject: [PATCH 3/3] compute: disable by default the alternative persist sink implemenetation (#38074) Follow up to incident 1175 where it was observed that the new sink implementation schedules long running blocking work on tokio worker threads causing them to stop scheduling all the other tasks, leading to lost reader leases. This is only to guard self-managed deployments from the problem. In cloud we have already disabled it for everyone via LD. We should consider re-flipping the default after having rolled it out gradually via LaunchDarkly for some time. Signed-off-by: Petros Angelatos --- src/compute-types/src/dyncfgs.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs index 210dd347c927f..42cd46753beba 100644 --- a/src/compute-types/src/dyncfgs.rs +++ b/src/compute-types/src/dyncfgs.rs @@ -187,7 +187,7 @@ pub const ENABLE_MZ_JOIN_CORE: Config = Config::new( /// Use sync Timely operators with Tokio tasks for the MV sink. pub const ENABLE_SYNC_MV_SINK: Config = Config::new( "enable_compute_sync_mv_sink", - true, + false, "Use sync Timely operators with Tokio tasks for the MV sink.", );