sql: add sql::plan::DataSourceDesc to express more source types

We previously only had Option<Ingestion> to describe sources,
but with the coming progress subsources, we need more expressiveness.

Author: Sean Loiselle

src/adapter/src/catalog.rs

@@ -5545,20 +5545,25 @@ impl Catalog {
                 ..
             }) => CatalogItem::Source(Source {
                 create_sql: source.create_sql,
-                data_source: match source.ingestion {
-                    Some(ingestion) => DataSourceDesc::Ingestion(Ingestion {
-                        desc: ingestion.desc,
-                        source_imports: ingestion.source_imports,
-                        subsource_exports: ingestion.subsource_exports,
-                        cluster_id: match cluster_config {
-                            plan::SourceSinkClusterConfig::Existing { id } => id,
-                            plan::SourceSinkClusterConfig::Linked { .. }
-                            | plan::SourceSinkClusterConfig::Undefined => {
-                                self.state.clusters_by_linked_object_id[&id]
-                            }
-                        },
-                    }),
-                    None => DataSourceDesc::Source,
+                data_source: match source.data_source {
+                    mz_sql::plan::DataSourceDesc::Ingestion(ingestion) => {
+                        DataSourceDesc::Ingestion(Ingestion {
+                            desc: ingestion.desc,
+                            source_imports: ingestion.source_imports,
+                            subsource_exports: ingestion.subsource_exports,
+                            cluster_id: match cluster_config {
+                                plan::SourceSinkClusterConfig::Existing { id } => id,
+                                plan::SourceSinkClusterConfig::Linked { .. }
+                                | plan::SourceSinkClusterConfig::Undefined => {
+                                    self.state.clusters_by_linked_object_id[&id]
+                                }
+                            },
+                        })
+                    }
+                    mz_sql::plan::DataSourceDesc::Progress => {
+                        unreachable!("progress subsources error in purification")
+                    }
+                    mz_sql::plan::DataSourceDesc::Source => DataSourceDesc::Source,
                 },
                 desc: source.desc,
                 timeline,

src/adapter/src/coord/sequencer.rs

@@ -494,8 +494,8 @@ impl Coordinator {
             let source_oid = self.catalog.allocate_oid()?;
             let source = catalog::Source {
                 create_sql: plan.source.create_sql,
-                data_source: match plan.source.ingestion {
-                    Some(ingestion) => {
+                data_source: match plan.source.data_source {
+                    mz_sql::plan::DataSourceDesc::Ingestion(ingestion) => {
                         let cluster_id = self
                             .create_linked_cluster_ops(
                                 source_id,
@@ -511,7 +511,10 @@ impl Coordinator {
                             cluster_id,
                         })
                     }
-                    None => {
+                    mz_sql::plan::DataSourceDesc::Progress => {
+                        unreachable!("PROGRESS subsources error in purification");
+                    }
+                    mz_sql::plan::DataSourceDesc::Source => {
                         assert!(
                             matches!(
                                 plan.cluster_config,

src/sql/src/plan.rs

@@ -636,10 +636,20 @@ pub struct Table {
 #[derive(Clone, Debug)]
 pub struct Source {
     pub create_sql: String,
-    pub ingestion: Option<Ingestion>,
+    pub data_source: DataSourceDesc,
     pub desc: RelationDesc,
 }
 
+#[derive(Debug, Clone)]
+pub enum DataSourceDesc {
+    /// Receives data from an external system
+    Ingestion(Ingestion),
+    /// Receives data from some other source
+    Source,
+    /// Receives data from the source's reclocking/remapping operations.
+    Progress,
+}
+
 #[derive(Clone, Debug)]
 pub struct Ingestion {
     pub desc: SourceDesc,

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

@@ -109,10 +109,11 @@ use crate::plan::{
     ComputeReplicaIntrospectionConfig, CreateClusterPlan, CreateClusterReplicaPlan,
     CreateConnectionPlan, CreateDatabasePlan, CreateIndexPlan, CreateMaterializedViewPlan,
     CreateRolePlan, CreateSchemaPlan, CreateSecretPlan, CreateSinkPlan, CreateSourcePlan,
-    CreateTablePlan, CreateTypePlan, CreateViewPlan, DropClusterReplicasPlan, DropClustersPlan,
-    DropDatabasePlan, DropItemsPlan, DropRolesPlan, DropSchemaPlan, FullObjectName, HirScalarExpr,
-    Index, Ingestion, MaterializedView, Params, Plan, QueryContext, ReplicaConfig, RotateKeysPlan,
-    Secret, Sink, Source, SourceSinkClusterConfig, Table, Type, View,
+    CreateTablePlan, CreateTypePlan, CreateViewPlan, DataSourceDesc, DropClusterReplicasPlan,
+    DropClustersPlan, DropDatabasePlan, DropItemsPlan, DropRolesPlan, DropSchemaPlan,
+    FullObjectName, HirScalarExpr, Index, Ingestion, MaterializedView, Params, Plan, QueryContext,
+    ReplicaConfig, RotateKeysPlan, Secret, Sink, Source, SourceSinkClusterConfig, Table, Type,
+    View,
 };
 
 pub fn describe_create_database(
@@ -1069,7 +1070,7 @@ pub fn plan_create_source(
 
     let source = Source {
         create_sql,
-        ingestion: Some(Ingestion {
+        data_source: DataSourceDesc::Ingestion(Ingestion {
             desc: source_desc,
             // Currently no source reads from another source
             source_imports: BTreeSet::new(),
@@ -1179,7 +1180,7 @@ pub fn plan_create_subsource(
 
     let source = Source {
         create_sql,
-        ingestion: None,
+        data_source: DataSourceDesc::Source,
         desc,
     };