sql: add CREATE SUBSOURCE options for data sources

Subsources need to persist whether or not they are progress
subsources, which will ultimately determine their initial read
policy in the storage controller. To make this unambiguous, also
include a type for the current reference-style subsources.

Author: Sean Loiselle

src/adapter/src/catalog/migrate.rs

@@ -11,10 +11,10 @@ use semver::Version;
 use std::collections::BTreeMap;
 
 use mz_ore::collections::CollectionExt;
-use mz_sql::ast::Raw;
 use mz_sql::ast::{
-    display::AstDisplay, CreateSourceStatement, CreateSourceSubsource, DeferredObjectName,
-    RawObjectName, ReferencedSubsources, Statement, UnresolvedObjectName,
+    display::AstDisplay, CreateSourceStatement, CreateSourceSubsource, CreateSubsourceOptionName,
+    CreateSubsourceStatement, DeferredObjectName, Raw, RawObjectName, ReferencedSubsources,
+    Statement, UnresolvedObjectName,
 };
 
 use crate::catalog::{Catalog, ConnCatalog, SerializedCatalogItem, SYSTEM_CONN_ID};
@@ -51,7 +51,10 @@ pub(crate) async fn migrate(catalog: &mut Catalog) -> Result<(), anyhow::Error>
     };
     let mut tx = storage.transaction().await?;
     // First, do basic AST -> AST transformations.
-    rewrite_items(&mut tx, |_stmt| Ok(()))?;
+    rewrite_items(&mut tx, |stmt| {
+        subsource_type_option_rewrite(stmt);
+        Ok(())
+    })?;
 
     // Then, load up a temporary catalog with the rewritten items, and perform
     // some transformations that require introspecting the catalog. These
@@ -89,6 +92,27 @@ pub(crate) async fn migrate(catalog: &mut Catalog) -> Result<(), anyhow::Error>
 // AST migrations -- Basic AST -> AST transformations
 // ****************************************************************************
 
+// Mark all current subsources as "references" subsources in anticipation of
+// adding "progress" subsources.
+// TODO: delete in version v0.45 (released in v0.43 + 1 additional release)
+fn subsource_type_option_rewrite(stmt: &mut mz_sql::ast::Statement<Raw>) {
+    if let Statement::CreateSubsource(CreateSubsourceStatement { with_options, .. }) = stmt {
+        if !with_options.iter().any(|option| {
+            matches!(
+                option.name,
+                CreateSubsourceOptionName::Progress | CreateSubsourceOptionName::References
+            )
+        }) {
+            with_options.push(mz_sql::ast::CreateSubsourceOption {
+                name: CreateSubsourceOptionName::References,
+                value: Some(mz_sql::ast::WithOptionValue::Value(
+                    mz_sql::ast::Value::Boolean(true),
+                )),
+            });
+        }
+    }
+}
+
 // ****************************************************************************
 // Semantic migrations -- Weird migrations that require access to the catalog
 // ****************************************************************************

src/sql-parser/src/ast/defs/statement.rs

@@ -659,13 +659,50 @@ impl<T: AstInfo> AstDisplay for ReferencedSubsources<T> {
 }
 impl_display_t!(ReferencedSubsources);
 
+/// An option in a `CREATE SUBSOURCE` statement.
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum CreateSubsourceOptionName {
+    Progress,
+    References,
+}
+
+impl AstDisplay for CreateSubsourceOptionName {
+    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
+        match self {
+            CreateSubsourceOptionName::Progress => {
+                f.write_str("PROGRESS");
+            }
+            CreateSubsourceOptionName::References => {
+                f.write_str("REFERENCES");
+            }
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct CreateSubsourceOption<T: AstInfo> {
+    pub name: CreateSubsourceOptionName,
+    pub value: Option<WithOptionValue<T>>,
+}
+
+impl<T: AstInfo> AstDisplay for CreateSubsourceOption<T> {
+    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
+        f.write_node(&self.name);
+        if let Some(v) = &self.value {
+            f.write_str(" = ");
+            f.write_node(v);
+        }
+    }
+}
+
 /// `CREATE SUBSOURCE`
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct CreateSubsourceStatement<T: AstInfo> {
     pub name: UnresolvedObjectName,
     pub columns: Vec<ColumnDef<T>>,
     pub constraints: Vec<TableConstraint<T>>,
     pub if_not_exists: bool,
+    pub with_options: Vec<CreateSubsourceOption<T>>,
 }
 
 impl<T: AstInfo> AstDisplay for CreateSubsourceStatement<T> {
@@ -682,6 +719,12 @@ impl<T: AstInfo> AstDisplay for CreateSubsourceStatement<T> {
             f.write_node(&display::comma_separated(&self.constraints));
         }
         f.write_str(")");
+
+        if !self.with_options.is_empty() {
+            f.write_str(" WITH (");
+            f.write_node(&display::comma_separated(&self.with_options));
+            f.write_str(")");
+        }
     }
 }
 impl_display_t!(CreateSubsourceStatement);

src/sql-parser/src/parser.rs

@@ -2392,14 +2392,44 @@ impl<'a> Parser<'a> {
 
         let (columns, constraints) = self.parse_columns(Mandatory)?;
 
+        let with_options = if self.parse_keyword(WITH) {
+            self.expect_token(&Token::LParen)?;
+            let options = self.parse_comma_separated(Parser::parse_create_subsource_option)?;
+            self.expect_token(&Token::RParen)?;
+            options
+        } else {
+            vec![]
+        };
+
         Ok(Statement::CreateSubsource(CreateSubsourceStatement {
             name,
             if_not_exists,
             columns,
             constraints,
+            with_options,
         }))
     }
 
+    /// Parse the name of a CREATE SINK optional parameter
+    fn parse_create_subsource_option_name(
+        &mut self,
+    ) -> Result<CreateSubsourceOptionName, ParserError> {
+        let name = match self.expect_one_of_keywords(&[PROGRESS, REFERENCES])? {
+            PROGRESS => CreateSubsourceOptionName::Progress,
+            REFERENCES => CreateSubsourceOptionName::References,
+            _ => unreachable!(),
+        };
+        Ok(name)
+    }
+
+    /// Parse a NAME = VALUE parameter for CREATE SINK
+    fn parse_create_subsource_option(&mut self) -> Result<CreateSubsourceOption<Raw>, ParserError> {
+        Ok(CreateSubsourceOption {
+            name: self.parse_create_subsource_option_name()?,
+            value: self.parse_optional_option_value()?,
+        })
+    }
+
     fn parse_create_source(&mut self) -> Result<Statement<Raw>, ParserError> {
         self.expect_keyword(SOURCE)?;
         let if_not_exists = self.parse_if_not_exists()?;