mz_sql/plan/
with_options.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Provides tooling to handle `WITH` options.
11
12use std::collections::BTreeMap;
13use std::time::Duration;
14
15use mz_repr::adt::interval::Interval;
16use mz_repr::bytes::ByteSize;
17use mz_repr::{CatalogItemId, RelationVersionSelector, strconv};
18use mz_sql_parser::ast::{
19    ClusterAlterOptionValue, ClusterAutoScalingStrategyOptionValue, ClusterScheduleOptionValue,
20    ConnectionDefaultAwsPrivatelink, Expr, Ident, KafkaBroker, KafkaMatchingBrokerRule,
21    NetworkPolicyRuleDefinition, RefreshOptionValue, ReplicaDefinition,
22};
23use mz_storage_types::connections::IcebergCatalogType;
24use mz_storage_types::connections::string_or_secret::StringOrSecret;
25use serde::{Deserialize, Serialize};
26
27use crate::ast::{AstInfo, UnresolvedItemName, Value, WithOptionValue};
28use crate::catalog::SessionCatalog;
29use crate::names::{ResolvedDataType, ResolvedItemName};
30use crate::plan::{Aug, PlanError, literal};
31
32pub trait TryFromValue<T>: Sized {
33    fn try_from_value(v: T) -> Result<Self, PlanError>;
34
35    fn try_into_value(self, catalog: &dyn SessionCatalog) -> Option<T>;
36
37    fn name() -> String;
38}
39
40pub trait ImpliedValue: Sized {
41    fn implied_value() -> Result<Self, PlanError>;
42}
43
44impl TryFromValue<WithOptionValue<Aug>> for IcebergCatalogType {
45    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
46        match String::try_from_value(v)? {
47            s if s.eq_ignore_ascii_case("rest") => Ok(IcebergCatalogType::Rest),
48            s if s.eq_ignore_ascii_case("s3tablesrest") => Ok(IcebergCatalogType::S3TablesRest),
49            _ => sql_bail!("invalid iceberg catalog type"),
50        }
51    }
52
53    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
54        Some(WithOptionValue::Value(Value::String(match self {
55            IcebergCatalogType::Rest => "rest".to_string(),
56            IcebergCatalogType::S3TablesRest => "s3tablesrest".to_string(),
57        })))
58    }
59
60    fn name() -> String {
61        "iceberg catalog type".to_string()
62    }
63}
64
65impl ImpliedValue for IcebergCatalogType {
66    fn implied_value() -> Result<Self, PlanError> {
67        sql_bail!("must provide an iceberg catalog type")
68    }
69}
70
71#[derive(Copy, Clone, Debug)]
72pub struct Secret(CatalogItemId);
73
74impl From<Secret> for CatalogItemId {
75    fn from(secret: Secret) -> Self {
76        secret.0
77    }
78}
79
80impl TryFromValue<WithOptionValue<Aug>> for Secret {
81    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
82        match StringOrSecret::try_from_value(v)? {
83            StringOrSecret::Secret(id) => Ok(Secret(id)),
84            StringOrSecret::String(_) => sql_bail!("must provide a secret value"),
85        }
86    }
87
88    fn try_into_value(self, catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
89        let secret = catalog.get_item(&self.0);
90        let name = ResolvedItemName::Item {
91            id: self.0,
92            qualifiers: secret.name().qualifiers.clone(),
93            full_name: catalog.resolve_full_name(secret.name()),
94            print_id: false,
95            version: RelationVersionSelector::Latest,
96        };
97        Some(WithOptionValue::Secret(name))
98    }
99
100    fn name() -> String {
101        "secret".to_string()
102    }
103}
104
105impl ImpliedValue for Secret {
106    fn implied_value() -> Result<Self, PlanError> {
107        sql_bail!("must provide a secret value")
108    }
109}
110
111#[derive(Copy, Clone, Debug)]
112pub struct Object(CatalogItemId);
113
114impl From<Object> for CatalogItemId {
115    fn from(obj: Object) -> Self {
116        obj.0
117    }
118}
119
120impl From<&Object> for CatalogItemId {
121    fn from(obj: &Object) -> Self {
122        obj.0
123    }
124}
125
126impl TryFromValue<WithOptionValue<Aug>> for Object {
127    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
128        Ok(match v {
129            WithOptionValue::Item(ResolvedItemName::Item { id, .. }) => Object(id),
130            _ => sql_bail!("must provide an object"),
131        })
132    }
133
134    fn try_into_value(self, catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
135        let item = catalog.get_item(&self.0);
136        let name = ResolvedItemName::Item {
137            id: self.0,
138            qualifiers: item.name().qualifiers.clone(),
139            full_name: catalog.resolve_full_name(item.name()),
140            print_id: false,
141            // TODO(alter_table): Evaluate if this is correct.
142            version: RelationVersionSelector::Latest,
143        };
144        Some(WithOptionValue::Item(name))
145    }
146
147    fn name() -> String {
148        "object reference".to_string()
149    }
150}
151
152impl ImpliedValue for Object {
153    fn implied_value() -> Result<Self, PlanError> {
154        sql_bail!("must provide an object")
155    }
156}
157
158impl TryFromValue<WithOptionValue<Aug>> for Ident {
159    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
160        Ok(match v {
161            WithOptionValue::UnresolvedItemName(UnresolvedItemName(mut inner))
162                if inner.len() == 1 =>
163            {
164                inner.remove(0)
165            }
166            WithOptionValue::Ident(inner) => inner,
167            _ => sql_bail!("must provide an unqualified identifier"),
168        })
169    }
170
171    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
172        Some(WithOptionValue::Ident(self))
173    }
174
175    fn name() -> String {
176        "identifier".to_string()
177    }
178}
179
180impl ImpliedValue for Ident {
181    fn implied_value() -> Result<Self, PlanError> {
182        sql_bail!("must provide an identifier")
183    }
184}
185
186impl TryFromValue<WithOptionValue<Aug>> for Expr<Aug> {
187    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
188        Ok(match v {
189            WithOptionValue::Expr(e) => e,
190            _ => sql_bail!("must provide an expr"),
191        })
192    }
193
194    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
195        Some(WithOptionValue::Expr(self))
196    }
197
198    fn name() -> String {
199        "expression".to_string()
200    }
201}
202
203impl ImpliedValue for Expr<Aug> {
204    fn implied_value() -> Result<Self, PlanError> {
205        sql_bail!("must provide an expression")
206    }
207}
208
209impl TryFromValue<WithOptionValue<Aug>> for UnresolvedItemName {
210    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
211        Ok(match v {
212            WithOptionValue::UnresolvedItemName(name) => name,
213            WithOptionValue::Ident(inner) => UnresolvedItemName(vec![inner]),
214            _ => sql_bail!("must provide an object name"),
215        })
216    }
217
218    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
219        Some(WithOptionValue::UnresolvedItemName(self))
220    }
221
222    fn name() -> String {
223        "object name".to_string()
224    }
225}
226
227impl ImpliedValue for UnresolvedItemName {
228    fn implied_value() -> Result<Self, PlanError> {
229        sql_bail!("must provide an object name")
230    }
231}
232
233impl TryFromValue<WithOptionValue<Aug>> for ResolvedDataType {
234    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
235        Ok(match v {
236            WithOptionValue::DataType(ty) => ty,
237            _ => sql_bail!("must provide a data type"),
238        })
239    }
240
241    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
242        Some(WithOptionValue::DataType(self))
243    }
244
245    fn name() -> String {
246        "data type".to_string()
247    }
248}
249
250impl ImpliedValue for ResolvedDataType {
251    fn implied_value() -> Result<Self, PlanError> {
252        sql_bail!("must provide a data type")
253    }
254}
255
256impl TryFromValue<WithOptionValue<Aug>> for StringOrSecret {
257    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
258        Ok(match v {
259            WithOptionValue::Secret(ResolvedItemName::Item { id, .. }) => {
260                StringOrSecret::Secret(id)
261            }
262            v => StringOrSecret::String(String::try_from_value(v)?),
263        })
264    }
265
266    fn try_into_value(self, catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
267        Some(match self {
268            StringOrSecret::Secret(secret) => Secret(secret).try_into_value(catalog)?,
269            StringOrSecret::String(s) => s.try_into_value(catalog)?,
270        })
271    }
272
273    fn name() -> String {
274        "string or secret".to_string()
275    }
276}
277
278impl ImpliedValue for StringOrSecret {
279    fn implied_value() -> Result<Self, PlanError> {
280        sql_bail!("must provide a string or secret value")
281    }
282}
283
284impl TryFromValue<Value> for Duration {
285    fn try_from_value(v: Value) -> Result<Self, PlanError> {
286        let interval = Interval::try_from_value(v)?;
287        let duration = interval.duration()?;
288        // `try_into_value` (used during unplanning) requires that the `Duration`
289        // is convertible back to an `Interval`, which has a smaller range than
290        // `Duration`. Enforce that here so we return a graceful error instead of
291        // panicking later. See database-issues SQL-361.
292        Interval::from_duration(&duration).map_err(|_| sql_err!("interval out of range"))?;
293        Ok(duration)
294    }
295
296    fn try_into_value(self, catalog: &dyn SessionCatalog) -> Option<Value> {
297        let interval = Interval::from_duration(&self)
298            .expect("planning ensured that this is convertible back to Interval");
299        interval.try_into_value(catalog)
300    }
301
302    fn name() -> String {
303        "interval".to_string()
304    }
305}
306
307impl ImpliedValue for Duration {
308    fn implied_value() -> Result<Self, PlanError> {
309        sql_bail!("must provide an interval value")
310    }
311}
312
313impl TryFromValue<Value> for ByteSize {
314    fn try_from_value(v: Value) -> Result<Self, PlanError> {
315        match v {
316            Value::Number(value) | Value::String(value) => Ok(value
317                .parse::<ByteSize>()
318                .map_err(|e| sql_err!("invalid bytes value: {e}"))?),
319            _ => sql_bail!("cannot use value as bytes"),
320        }
321    }
322
323    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<Value> {
324        Some(Value::String(self.to_string()))
325    }
326
327    fn name() -> String {
328        "bytes".to_string()
329    }
330}
331
332impl ImpliedValue for ByteSize {
333    fn implied_value() -> Result<Self, PlanError> {
334        sql_bail!("must provide a value for bytes")
335    }
336}
337
338impl TryFromValue<Value> for Interval {
339    fn try_from_value(v: Value) -> Result<Self, PlanError> {
340        match v {
341            Value::Interval(value) => literal::plan_interval(&value),
342            Value::Number(value) | Value::String(value) => Ok(strconv::parse_interval(&value)?),
343            _ => sql_bail!("cannot use value as interval"),
344        }
345    }
346
347    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<Value> {
348        let interval_value = literal::unplan_interval(&self);
349        Some(Value::Interval(interval_value))
350    }
351
352    fn name() -> String {
353        "interval".to_string()
354    }
355}
356
357impl ImpliedValue for Interval {
358    fn implied_value() -> Result<Self, PlanError> {
359        sql_bail!("must provide an interval value")
360    }
361}
362
363#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
364pub struct OptionalString(pub Option<String>);
365
366impl TryFromValue<Value> for OptionalString {
367    fn try_from_value(v: Value) -> Result<Self, PlanError> {
368        Ok(match v {
369            Value::Null => Self(None),
370            v => Self(Some(String::try_from_value(v)?)),
371        })
372    }
373
374    fn try_into_value(self, catalog: &dyn SessionCatalog) -> Option<Value> {
375        Some(match self.0 {
376            None => Value::Null,
377            Some(s) => s.try_into_value(catalog)?,
378        })
379    }
380
381    fn name() -> String {
382        "optional string".to_string()
383    }
384}
385
386impl ImpliedValue for OptionalString {
387    fn implied_value() -> Result<Self, PlanError> {
388        sql_bail!("must provide a string value")
389    }
390}
391
392#[derive(
393    Debug,
394    Clone,
395    Copy,
396    PartialEq,
397    Eq,
398    PartialOrd,
399    Ord,
400    Serialize,
401    Hash,
402    Deserialize
403)]
404pub struct OptionalDuration(pub Option<Duration>);
405
406impl From<Duration> for OptionalDuration {
407    fn from(i: Duration) -> OptionalDuration {
408        // An interval of 0 disables the setting.
409        let inner = if i == Duration::ZERO { None } else { Some(i) };
410        OptionalDuration(inner)
411    }
412}
413
414impl TryFromValue<Value> for OptionalDuration {
415    fn try_from_value(v: Value) -> Result<Self, PlanError> {
416        Ok(match v {
417            Value::Null => OptionalDuration(None),
418            v => Duration::try_from_value(v)?.into(),
419        })
420    }
421
422    fn try_into_value(self, catalog: &dyn SessionCatalog) -> Option<Value> {
423        Some(match self.0 {
424            None => Value::Null,
425            Some(duration) => duration.try_into_value(catalog)?,
426        })
427    }
428
429    fn name() -> String {
430        "optional interval".to_string()
431    }
432}
433
434impl ImpliedValue for OptionalDuration {
435    fn implied_value() -> Result<Self, PlanError> {
436        sql_bail!("must provide an interval value")
437    }
438}
439
440impl TryFromValue<Value> for String {
441    fn try_from_value(v: Value) -> Result<Self, PlanError> {
442        match v {
443            Value::String(v) => Ok(v),
444            _ => sql_bail!("cannot use value as string"),
445        }
446    }
447
448    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<Value> {
449        Some(Value::String(self))
450    }
451
452    fn name() -> String {
453        "text".to_string()
454    }
455}
456
457impl ImpliedValue for String {
458    fn implied_value() -> Result<Self, PlanError> {
459        sql_bail!("must provide a string value")
460    }
461}
462
463impl TryFromValue<Value> for bool {
464    fn try_from_value(v: Value) -> Result<Self, PlanError> {
465        match v {
466            Value::Boolean(v) => Ok(v),
467            _ => sql_bail!("cannot use value as boolean"),
468        }
469    }
470
471    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<Value> {
472        Some(Value::Boolean(self))
473    }
474
475    fn name() -> String {
476        "bool".to_string()
477    }
478}
479
480impl ImpliedValue for bool {
481    fn implied_value() -> Result<Self, PlanError> {
482        Ok(true)
483    }
484}
485
486impl TryFromValue<Value> for f64 {
487    fn try_from_value(v: Value) -> Result<Self, PlanError> {
488        match v {
489            Value::Number(v) => v
490                .parse::<f64>()
491                .map_err(|e| sql_err!("invalid numeric value: {e}")),
492            _ => sql_bail!("cannot use value as number"),
493        }
494    }
495
496    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<Value> {
497        Some(Value::Number(self.to_string()))
498    }
499
500    fn name() -> String {
501        "float8".to_string()
502    }
503}
504
505impl ImpliedValue for f64 {
506    fn implied_value() -> Result<Self, PlanError> {
507        sql_bail!("must provide a float value")
508    }
509}
510
511impl TryFromValue<Value> for i32 {
512    fn try_from_value(v: Value) -> Result<Self, PlanError> {
513        match v {
514            Value::Number(v) => v
515                .parse::<i32>()
516                .map_err(|e| sql_err!("invalid numeric value: {e}")),
517            _ => sql_bail!("cannot use value as number"),
518        }
519    }
520
521    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<Value> {
522        Some(Value::Number(self.to_string()))
523    }
524
525    fn name() -> String {
526        "int".to_string()
527    }
528}
529
530impl ImpliedValue for i32 {
531    fn implied_value() -> Result<Self, PlanError> {
532        sql_bail!("must provide an integer value")
533    }
534}
535
536impl TryFromValue<Value> for i64 {
537    fn try_from_value(v: Value) -> Result<Self, PlanError> {
538        match v {
539            Value::Number(v) => v
540                .parse::<i64>()
541                .map_err(|e| sql_err!("invalid numeric value: {e}")),
542            _ => sql_bail!("cannot use value as number"),
543        }
544    }
545    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<Value> {
546        Some(Value::Number(self.to_string()))
547    }
548    fn name() -> String {
549        "int8".to_string()
550    }
551}
552
553impl ImpliedValue for i64 {
554    fn implied_value() -> Result<Self, PlanError> {
555        sql_bail!("must provide an integer value")
556    }
557}
558
559impl TryFromValue<Value> for u16 {
560    fn try_from_value(v: Value) -> Result<Self, PlanError> {
561        match v {
562            Value::Number(v) => v
563                .parse::<u16>()
564                .map_err(|e| sql_err!("invalid numeric value: {e}")),
565            _ => sql_bail!("cannot use value as number"),
566        }
567    }
568    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<Value> {
569        Some(Value::Number(self.to_string()))
570    }
571    fn name() -> String {
572        "uint2".to_string()
573    }
574}
575
576impl ImpliedValue for u16 {
577    fn implied_value() -> Result<Self, PlanError> {
578        sql_bail!("must provide an integer value")
579    }
580}
581
582impl TryFromValue<Value> for u32 {
583    fn try_from_value(v: Value) -> Result<Self, PlanError> {
584        match v {
585            Value::Number(v) => v
586                .parse::<u32>()
587                .map_err(|e| sql_err!("invalid numeric value: {e}")),
588            _ => sql_bail!("cannot use value as number"),
589        }
590    }
591    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<Value> {
592        Some(Value::Number(self.to_string()))
593    }
594    fn name() -> String {
595        "uint4".to_string()
596    }
597}
598
599impl ImpliedValue for u32 {
600    fn implied_value() -> Result<Self, PlanError> {
601        sql_bail!("must provide an integer value")
602    }
603}
604
605impl TryFromValue<Value> for u64 {
606    fn try_from_value(v: Value) -> Result<Self, PlanError> {
607        match v {
608            Value::Number(v) => v
609                .parse::<u64>()
610                .map_err(|e| sql_err!("invalid unsigned numeric value: {e}")),
611            _ => sql_bail!("cannot use value as number"),
612        }
613    }
614    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<Value> {
615        Some(Value::Number(self.to_string()))
616    }
617    fn name() -> String {
618        "uint8".to_string()
619    }
620}
621
622impl ImpliedValue for u64 {
623    fn implied_value() -> Result<Self, PlanError> {
624        sql_bail!("must provide an unsigned integer value")
625    }
626}
627
628impl<V: TryFromValue<WithOptionValue<Aug>>> TryFromValue<WithOptionValue<Aug>> for Vec<V> {
629    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
630        match v {
631            WithOptionValue::Sequence(a) => {
632                let mut out = Vec::with_capacity(a.len());
633                for i in a {
634                    out.push(
635                        V::try_from_value(i)
636                            .map_err(|_| anyhow::anyhow!("cannot use value in array"))?,
637                    )
638                }
639                Ok(out)
640            }
641            _ => sql_bail!("cannot use value as array"),
642        }
643    }
644
645    fn try_into_value(self, catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
646        Some(WithOptionValue::Sequence(
647            self.into_iter()
648                .map(|v| v.try_into_value(catalog))
649                .collect::<Option<_>>()?,
650        ))
651    }
652
653    fn name() -> String {
654        format!("array of {}", V::name())
655    }
656}
657
658impl<V: ImpliedValue> ImpliedValue for Vec<V> {
659    fn implied_value() -> Result<Self, PlanError> {
660        sql_bail!("must provide an array value")
661    }
662}
663
664impl<T: AstInfo, V: TryFromValue<WithOptionValue<T>>> TryFromValue<WithOptionValue<T>>
665    for Option<V>
666{
667    fn try_from_value(v: WithOptionValue<T>) -> Result<Self, PlanError> {
668        Ok(Some(V::try_from_value(v)?))
669    }
670
671    fn try_into_value(self, catalog: &dyn SessionCatalog) -> Option<WithOptionValue<T>> {
672        match self {
673            Some(v) => v.try_into_value(catalog),
674            None => None,
675        }
676    }
677
678    fn name() -> String {
679        format!("optional {}", V::name())
680    }
681}
682
683impl<V: ImpliedValue> ImpliedValue for Option<V> {
684    fn implied_value() -> Result<Self, PlanError> {
685        Ok(Some(V::implied_value()?))
686    }
687}
688
689impl<V: TryFromValue<Value>, T: AstInfo + std::fmt::Debug> TryFromValue<WithOptionValue<T>> for V {
690    fn try_from_value(v: WithOptionValue<T>) -> Result<Self, PlanError> {
691        match v {
692            WithOptionValue::Value(v) => V::try_from_value(v),
693            WithOptionValue::UnresolvedItemName(UnresolvedItemName(mut inner))
694                if inner.len() == 1 =>
695            {
696                V::try_from_value(Value::String(inner.remove(0).into_string()))
697            }
698            WithOptionValue::Ident(v) => V::try_from_value(Value::String(v.into_string())),
699            WithOptionValue::RetainHistoryFor(v) => V::try_from_value(v),
700            WithOptionValue::Sequence(_)
701            | WithOptionValue::Map(_)
702            | WithOptionValue::Item(_)
703            | WithOptionValue::UnresolvedItemName(_)
704            | WithOptionValue::Secret(_)
705            | WithOptionValue::DataType(_)
706            | WithOptionValue::Expr(_)
707            | WithOptionValue::ClusterReplicas(_)
708            | WithOptionValue::ConnectionKafkaBroker(_)
709            | WithOptionValue::ConnectionAwsPrivatelink(_)
710            | WithOptionValue::KafkaMatchingBrokerRule(_)
711            | WithOptionValue::ClusterAlterStrategy(_)
712            | WithOptionValue::Refresh(_)
713            | WithOptionValue::ClusterScheduleOptionValue(_)
714            | WithOptionValue::ClusterAutoScalingStrategyOptionValue(_)
715            | WithOptionValue::NetworkPolicyRules(_) => sql_bail!(
716                "incompatible value types: cannot convert {} to {}",
717                match v {
718                    // The first few are unreachable because they are handled at the top of the outer match.
719                    WithOptionValue::Value(_) => unreachable!(),
720                    WithOptionValue::RetainHistoryFor(_) => unreachable!(),
721                    WithOptionValue::ClusterAlterStrategy(_) => "cluster alter strategy",
722                    WithOptionValue::Sequence(_) => "sequences",
723                    WithOptionValue::Map(_) => "maps",
724                    WithOptionValue::Item(_) => "object references",
725                    WithOptionValue::UnresolvedItemName(_) => "object names",
726                    WithOptionValue::Ident(_) => "identifiers",
727                    WithOptionValue::Secret(_) => "secrets",
728                    WithOptionValue::DataType(_) => "data types",
729                    WithOptionValue::Expr(_) => "exprs",
730                    WithOptionValue::ClusterReplicas(_) => "cluster replicas",
731                    WithOptionValue::ConnectionKafkaBroker(_) => "connection kafka brokers",
732                    WithOptionValue::ConnectionAwsPrivatelink(_) => "connection privatelink",
733                    WithOptionValue::KafkaMatchingBrokerRule(_) => "matching broker rule",
734                    WithOptionValue::Refresh(_) => "refresh option values",
735                    WithOptionValue::ClusterScheduleOptionValue(_) => "cluster schedule",
736                    WithOptionValue::ClusterAutoScalingStrategyOptionValue(_) =>
737                        "cluster auto scaling strategy",
738                    WithOptionValue::NetworkPolicyRules(_) => "network policy rules",
739                },
740                V::name()
741            ),
742        }
743    }
744
745    fn try_into_value(self, catalog: &dyn SessionCatalog) -> Option<WithOptionValue<T>> {
746        Some(WithOptionValue::Value(self.try_into_value(catalog)?))
747    }
748
749    fn name() -> String {
750        V::name()
751    }
752}
753
754impl<T, V: TryFromValue<T> + ImpliedValue> TryFromValue<Option<T>> for V {
755    fn try_from_value(v: Option<T>) -> Result<Self, PlanError> {
756        match v {
757            Some(v) => V::try_from_value(v),
758            None => V::implied_value(),
759        }
760    }
761
762    fn try_into_value(self, catalog: &dyn SessionCatalog) -> Option<Option<T>> {
763        Some(Some(self.try_into_value(catalog)?))
764    }
765
766    fn name() -> String {
767        V::name()
768    }
769}
770
771impl TryFromValue<WithOptionValue<Aug>> for Vec<ReplicaDefinition<Aug>> {
772    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
773        match v {
774            WithOptionValue::ClusterReplicas(replicas) => Ok(replicas),
775            _ => sql_bail!("cannot use value as cluster replicas"),
776        }
777    }
778
779    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
780        Some(WithOptionValue::ClusterReplicas(self))
781    }
782
783    fn name() -> String {
784        "cluster replicas".to_string()
785    }
786}
787
788impl ImpliedValue for Vec<ReplicaDefinition<Aug>> {
789    fn implied_value() -> Result<Self, PlanError> {
790        sql_bail!("must provide a set of cluster replicas")
791    }
792}
793
794impl TryFromValue<WithOptionValue<Aug>> for Vec<KafkaBroker<Aug>> {
795    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
796        let mut out = vec![];
797        match v {
798            WithOptionValue::ConnectionKafkaBroker(broker) => {
799                out.push(broker);
800            }
801            WithOptionValue::Sequence(values) => {
802                for value in values {
803                    out.extend(Self::try_from_value(value)?);
804                }
805            }
806            _ => sql_bail!("cannot use value as a kafka broker"),
807        }
808        Ok(out)
809    }
810
811    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
812        Some(WithOptionValue::Sequence(
813            self.into_iter()
814                .map(WithOptionValue::ConnectionKafkaBroker)
815                .collect(),
816        ))
817    }
818
819    fn name() -> String {
820        "kafka broker".to_string()
821    }
822}
823
824impl ImpliedValue for Vec<KafkaBroker<Aug>> {
825    fn implied_value() -> Result<Self, PlanError> {
826        sql_bail!("must provide a kafka broker")
827    }
828}
829
830impl TryFromValue<WithOptionValue<Aug>> for RefreshOptionValue<Aug> {
831    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
832        if let WithOptionValue::Refresh(r) = v {
833            Ok(r)
834        } else {
835            sql_bail!("cannot use value `{}` for a refresh option", v)
836        }
837    }
838
839    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
840        Some(WithOptionValue::Refresh(self))
841    }
842
843    fn name() -> String {
844        "refresh option value".to_string()
845    }
846}
847
848impl ImpliedValue for RefreshOptionValue<Aug> {
849    fn implied_value() -> Result<Self, PlanError> {
850        sql_bail!("must provide a refresh option value")
851    }
852}
853
854impl TryFromValue<WithOptionValue<Aug>> for ConnectionDefaultAwsPrivatelink<Aug> {
855    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
856        if let WithOptionValue::ConnectionAwsPrivatelink(r) = v {
857            Ok(r)
858        } else {
859            sql_bail!("cannot use value `{}` for a privatelink", v)
860        }
861    }
862
863    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
864        Some(WithOptionValue::ConnectionAwsPrivatelink(self))
865    }
866
867    fn name() -> String {
868        "privatelink option value".to_string()
869    }
870}
871
872impl TryFromValue<WithOptionValue<Aug>> for KafkaMatchingBrokerRule<Aug> {
873    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
874        if let WithOptionValue::KafkaMatchingBrokerRule(r) = v {
875            Ok(r)
876        } else {
877            sql_bail!("cannot use value `{}` for a matching broker rule", v)
878        }
879    }
880
881    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
882        Some(WithOptionValue::KafkaMatchingBrokerRule(self))
883    }
884
885    fn name() -> String {
886        "matching broker rule".to_string()
887    }
888}
889
890impl ImpliedValue for ConnectionDefaultAwsPrivatelink<Aug> {
891    fn implied_value() -> Result<Self, PlanError> {
892        sql_bail!("must provide a value")
893    }
894}
895
896impl ImpliedValue for KafkaMatchingBrokerRule<Aug> {
897    fn implied_value() -> Result<Self, PlanError> {
898        sql_bail!("must provide a value")
899    }
900}
901
902/// A list of broker entries that can contain both static `KafkaBroker` entries
903/// and `KafkaMatchingBrokerRule` entries (from `MATCHING` clauses in `BROKERS`).
904#[derive(Debug)]
905pub struct BrokersList {
906    pub static_entries: Vec<KafkaBroker<Aug>>,
907    pub matching_rules: Vec<KafkaMatchingBrokerRule<Aug>>,
908}
909
910impl TryFromValue<WithOptionValue<Aug>> for BrokersList {
911    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
912        match v {
913            WithOptionValue::Sequence(entries) => {
914                let mut static_entries = vec![];
915                let mut matching_rules = vec![];
916                for entry in entries {
917                    match entry {
918                        WithOptionValue::ConnectionKafkaBroker(b) => static_entries.push(b),
919                        WithOptionValue::KafkaMatchingBrokerRule(m) => matching_rules.push(m),
920                        other => sql_bail!("unexpected value in BROKERS: {}", other),
921                    }
922                }
923                Ok(BrokersList {
924                    static_entries,
925                    matching_rules,
926                })
927            }
928            WithOptionValue::ConnectionKafkaBroker(b) => Ok(BrokersList {
929                static_entries: vec![b],
930                matching_rules: vec![],
931            }),
932            WithOptionValue::KafkaMatchingBrokerRule(m) => Ok(BrokersList {
933                static_entries: vec![],
934                matching_rules: vec![m],
935            }),
936            other => sql_bail!("cannot use {} as brokers list", other),
937        }
938    }
939
940    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
941        let mut entries: Vec<WithOptionValue<Aug>> = vec![];
942        for b in self.static_entries {
943            entries.push(WithOptionValue::ConnectionKafkaBroker(b));
944        }
945        for m in self.matching_rules {
946            entries.push(WithOptionValue::KafkaMatchingBrokerRule(m));
947        }
948        Some(WithOptionValue::Sequence(entries))
949    }
950
951    fn name() -> String {
952        "brokers list".to_string()
953    }
954}
955
956impl ImpliedValue for BrokersList {
957    fn implied_value() -> Result<Self, PlanError> {
958        sql_bail!("must provide a value for BROKERS")
959    }
960}
961
962impl ImpliedValue for ClusterScheduleOptionValue {
963    fn implied_value() -> Result<Self, PlanError> {
964        sql_bail!("must provide a cluster schedule option value")
965    }
966}
967
968impl ImpliedValue for ClusterAlterOptionValue<Aug> {
969    fn implied_value() -> Result<Self, PlanError> {
970        sql_bail!("must provide a value")
971    }
972}
973
974impl TryFromValue<WithOptionValue<Aug>> for ClusterScheduleOptionValue {
975    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
976        if let WithOptionValue::ClusterScheduleOptionValue(r) = v {
977            Ok(r)
978        } else {
979            sql_bail!("cannot use value `{}` for a cluster schedule", v)
980        }
981    }
982
983    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
984        Some(WithOptionValue::ClusterScheduleOptionValue(self))
985    }
986
987    fn name() -> String {
988        "cluster schedule option value".to_string()
989    }
990}
991
992impl ImpliedValue for ClusterAutoScalingStrategyOptionValue {
993    fn implied_value() -> Result<Self, PlanError> {
994        sql_bail!("must provide an auto scaling strategy option value")
995    }
996}
997
998impl TryFromValue<WithOptionValue<Aug>> for ClusterAutoScalingStrategyOptionValue {
999    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
1000        if let WithOptionValue::ClusterAutoScalingStrategyOptionValue(r) = v {
1001            Ok(r)
1002        } else {
1003            sql_bail!("cannot use value `{}` for an auto scaling strategy", v)
1004        }
1005    }
1006
1007    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
1008        Some(WithOptionValue::ClusterAutoScalingStrategyOptionValue(self))
1009    }
1010
1011    fn name() -> String {
1012        "auto scaling strategy option value".to_string()
1013    }
1014}
1015
1016impl<V: ImpliedValue> ImpliedValue for BTreeMap<String, V> {
1017    fn implied_value() -> Result<Self, PlanError> {
1018        sql_bail!("must provide a map of key-value pairs")
1019    }
1020}
1021
1022impl<V: TryFromValue<WithOptionValue<Aug>>> TryFromValue<WithOptionValue<Aug>>
1023    for BTreeMap<String, V>
1024{
1025    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
1026        match v {
1027            WithOptionValue::Map(a) => a
1028                .into_iter()
1029                .map(|(k, v)| Ok((k, V::try_from_value(v)?)))
1030                .collect(),
1031            _ => sql_bail!("cannot use value as map"),
1032        }
1033    }
1034
1035    fn try_into_value(self, catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
1036        Some(WithOptionValue::Map(
1037            self.into_iter()
1038                .map(|(k, v)| {
1039                    let v = v.try_into_value(catalog);
1040                    v.map(|v| (k, v))
1041                })
1042                .collect::<Option<_>>()?,
1043        ))
1044    }
1045
1046    fn name() -> String {
1047        format!("map of string to {}", V::name())
1048    }
1049}
1050
1051impl TryFromValue<WithOptionValue<Aug>> for ClusterAlterOptionValue<Aug> {
1052    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
1053        if let WithOptionValue::ClusterAlterStrategy(r) = v {
1054            Ok(r)
1055        } else {
1056            sql_bail!("cannot use value `{}` for a cluster alter strategy", v)
1057        }
1058    }
1059
1060    fn name() -> String {
1061        "cluster alter strategyoption value".to_string()
1062    }
1063
1064    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
1065        Some(WithOptionValue::ClusterAlterStrategy(self))
1066    }
1067}
1068
1069impl TryFromValue<WithOptionValue<Aug>> for Vec<NetworkPolicyRuleDefinition<Aug>> {
1070    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
1071        match v {
1072            WithOptionValue::NetworkPolicyRules(rules) => Ok(rules),
1073            _ => sql_bail!("cannot use value as cluster replicas"),
1074        }
1075    }
1076
1077    fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option<WithOptionValue<Aug>> {
1078        Some(WithOptionValue::NetworkPolicyRules(self))
1079    }
1080
1081    fn name() -> String {
1082        "network policy rules".to_string()
1083    }
1084}
1085
1086impl ImpliedValue for Vec<NetworkPolicyRuleDefinition<Aug>> {
1087    fn implied_value() -> Result<Self, PlanError> {
1088        sql_bail!("must provide a set of network policy rules")
1089    }
1090}
mz_sql/plan/with_options.rs

mz_sql/plan/
with_options.rs
