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.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/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/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, 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/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.", ); 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