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

mz_sql/plan/
with_options.rs
