sql: add useless EXPOSE PROGRESS AS clause to CREATE SOURCE

we will let users specify a name for progress collections, so
add parsing but error if specified

Author: Sean Loiselle

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

@@ -560,6 +560,7 @@ pub struct CreateSourceStatement<T: AstInfo> {
     pub key_constraint: Option<KeyConstraint>,
     pub with_options: Vec<CreateSourceOption<T>>,
     pub referenced_subsources: Option<ReferencedSubsources<T>>,
+    pub progress_subsource: Option<DeferredObjectName<T>>,
 }
 
 impl<T: AstInfo> AstDisplay for CreateSourceStatement<T> {
@@ -604,6 +605,11 @@ impl<T: AstInfo> AstDisplay for CreateSourceStatement<T> {
             f.write_node(subsources);
         }
 
+        if let Some(progress) = &self.progress_subsource {
+            f.write_str(" EXPOSE PROGRESS AS ");
+            f.write_node(progress);
+        }
+
         if !self.with_options.is_empty() {
             f.write_str(" WITH (");
             f.write_node(&display::comma_separated(&self.with_options));

src/sql-parser/src/keywords.txt

@@ -126,6 +126,7 @@ Execute
 Exists
 Expected
 Explain
+Expose
 Extract
 Factor
 False

src/sql-parser/src/parser.rs

@@ -2439,6 +2439,12 @@ impl<'a> Parser<'a> {
             None
         };
 
+        let progress_subsource = if self.parse_keywords(&[EXPOSE, PROGRESS, AS]) {
+            Some(self.parse_deferred_object_name()?)
+        } else {
+            None
+        };
+
         // New WITH block
         let with_options = if self.parse_keyword(WITH) {
             self.expect_token(&Token::LParen)?;
@@ -2459,8 +2465,9 @@ impl<'a> Parser<'a> {
             envelope,
             if_not_exists,
             key_constraint,
-            with_options,
             referenced_subsources,
+            progress_subsource,
+            with_options,
         }))
     }
 

src/sql-parser/tests/testdata/ddl

@@ -1631,3 +1631,10 @@ CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source') WI
 error: Expected end of statement, found FOR
 CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source') WITH (SIZE = 'small') FOR ALL TABLES;
                                                                                                     ^
+
+parse-statement
+CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION 'mz_source') FOR ALL TABLES EXPOSE PROGRESS AS foo.bar WITH (SIZE = 'small');
+----
+CREATE SOURCE mz_source FROM POSTGRES CONNECTION pg (PUBLICATION = 'mz_source') FOR ALL TABLES EXPOSE PROGRESS AS foo.bar WITH (SIZE = 'small')
+=>
+CreateSource(CreateSourceStatement { name: UnresolvedObjectName([Ident("mz_source")]), in_cluster: None, col_names: [], connection: Postgres { connection: Name(UnresolvedObjectName([Ident("pg")])), options: [PgConfigOption { name: Publication, value: Some(Value(String("mz_source"))) }] }, include_metadata: [], format: None, envelope: None, if_not_exists: false, key_constraint: None, with_options: [CreateSourceOption { name: Size, value: Some(Value(String("small"))) }], referenced_subsources: Some(All), progress_subsource: Some(Deferred(UnresolvedObjectName([Ident("foo"), Ident("bar")]))) })

src/sql/src/normalize.rs

@@ -307,6 +307,7 @@ pub fn create_statement(
             key_constraint: _,
             with_options: _,
             referenced_subsources: _,
+            progress_subsource: _,
         }) => {
             *name = allocate_name(name)?;
             *if_not_exists = false;

src/sql/src/plan/statement/ddl.rs

@@ -347,8 +347,14 @@ pub fn plan_create_source(
         include_metadata,
         with_options,
         referenced_subsources,
+        progress_subsource,
     } = &stmt;
 
+    // TODO: enable progress subsources.
+    if progress_subsource.is_some() {
+        sql_bail!("[internal error] should have errored in purification");
+    }
+
     let envelope = envelope.clone().unwrap_or(Envelope::None);
 
     const SAFE_WITH_OPTIONS: &[CreateSourceOptionName] = &[CreateSourceOptionName::Size];

src/sql/src/pure.rs

@@ -111,9 +111,14 @@ pub async fn purify_create_source(
         envelope,
         include_metadata: _,
         referenced_subsources: requested_subsources,
+        progress_subsource,
         ..
     } = &mut stmt;
 
+    if progress_subsource.is_some() {
+        bail_unsupported!("PROGRESS subsources not yet supported");
+    }
+
     // Disallow manually targetting subsources, this syntax is reserved for purification only
     if let Some(ReferencedSubsources::Subset(subsources)) = requested_subsources {
         for CreateSourceSubsource {

test/testdrive/kafka-progress.td

@@ -0,0 +1,30 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Create sources and verify they can ingest data while `environmentd` is online.
+
+$ kafka-create-topic topic=data
+$ kafka-ingest format=bytes topic=data
+one
+
+> CREATE CONNECTION kafka_conn
+  TO KAFKA (BROKER '${testdrive.kafka-addr}');
+
+> CREATE SOURCE data
+  FROM KAFKA CONNECTION kafka_conn (TOPIC 'testdrive-data-${testdrive.seed}')
+  FORMAT TEXT;
+
+> SELECT * from data
+one
+
+! CREATE SOURCE d
+  FROM KAFKA CONNECTION kafka_conn (TOPIC 'testdrive-data-${testdrive.seed}')
+  FORMAT TEXT
+  EXPOSE PROGRESS AS exposed_progress_data;
+contains:PROGRESS subsources not yet supported