adapter,sql: move sink FROM check into planning

This commit moves the check that ensures that a sink's FROM collection
is a storage collection from sequencing into planning. Generally, it's
good for errors to be caught as early as possible, to avoid unnecessary
work.

This also fixes the bug that `ALTER SINK ... SET FROM` would skip
checking the FROM collection's type and then panic during sequencing.
That's because sink planning is shared between `CREATE SINK` and `ALTER
SINK`, whereas sequencing is not.

Author: teskje

src/adapter/src/coord/sequencer/inner.rs

@@ -1405,24 +1405,6 @@ impl Coordinator {
             owner_id: *ctx.session().current_role_id(),
         }];
 
-        let from = self.catalog().get_entry_by_global_id(&catalog_sink.from);
-        if let Err(e) = self
-            .controller
-            .storage
-            .check_exists(sink.from)
-            .map_err(|e| match e {
-                StorageError::IdentifierMissing(_) => AdapterError::Unstructured(anyhow!(
-                    "{} is a {}, which cannot be exported as a sink",
-                    from.name().item.clone(),
-                    from.item().typ(),
-                )),
-                e => AdapterError::Storage(e),
-            })
-        {
-            ctx.retire(Err(e));
-            return;
-        }
-
         let result = self.catalog_transact(Some(ctx.session()), ops).await;
 
         match result {

src/sql/src/plan/error.rs

@@ -143,6 +143,10 @@ pub enum PlanError {
         name: String,
         version: String,
     },
+    InvalidSinkFrom {
+        name: String,
+        item_type: CatalogItemType,
+    },
     MangedReplicaName(String),
     ParserStatement(ParserStatementError),
     Parser(ParserError),
@@ -637,6 +641,9 @@ impl fmt::Display for PlanError {
             Self::InvalidVersion { name, version } => {
                 write!(f, "invalid version {} for {}", version.quoted(), name.quoted())
             },
+            Self::InvalidSinkFrom { name, item_type } => {
+                write!(f, "{name} is a {item_type}, which cannot be exported as a sink")
+            },
             Self::DropViewOnMaterializedView(name)
             | Self::AlterViewOnMaterializedView(name)
             | Self::ShowCreateViewOnMaterializedView(name)

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

@@ -3376,6 +3376,21 @@ fn plan_sink(
 
     let from_name = &from;
     let from = scx.get_item_by_resolved_name(&from)?;
+
+    {
+        use CatalogItemType::*;
+        match from.item_type() {
+            Table | Source | MaterializedView | ContinualTask => {}
+            Sink | View | Index | Type | Func | Secret | Connection => {
+                let name = scx.catalog.minimal_qualification(from.name());
+                return Err(PlanError::InvalidSinkFrom {
+                    name: name.to_string(),
+                    item_type: from.item_type(),
+                });
+            }
+        }
+    }
+
     if from.id().is_system() {
         bail_unsupported!("creating a sink directly on a catalog object");
     }

test/testdrive/alter-sink.td

@@ -133,3 +133,7 @@ $ sleep-is-probably-flaky-i-have-justified-my-need-with-a-comment duration=10s
 
 $ kafka-verify-data format=json sink=materialize.public.sink key=false
 {"before": null, "after": {"created_post_name": "hundred", "created_post_value": 99}}
+
+> CREATE TABLE t (s text)
+> CREATE VIEW v AS SELECT s, 1 FROM t
+> ALTER SINK sink SET FROM v

test/testdrive/github-9923.td

@@ -0,0 +1,35 @@
+# 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.
+
+# Regression test for database-issues#9923.
+
+> CREATE CONNECTION kafka_conn
+  TO KAFKA (BROKER '${testdrive.kafka-addr}', SECURITY PROTOCOL PLAINTEXT)
+> CREATE CONNECTION csr_conn TO CONFLUENT SCHEMA REGISTRY (
+    URL '${testdrive.schema-registry-url}'
+  )
+
+> CREATE TABLE t (a int)
+> CREATE VIEW v AS SELECT a, 1 FROM t
+
+! CREATE SINK sink
+  FROM v
+  INTO KAFKA CONNECTION kafka_conn (TOPIC 'testdrive-alter-sink-${testdrive.seed}')
+  FORMAT JSON
+  ENVELOPE DEBEZIUM
+contains:v is a view, which cannot be exported as a sink
+
+> CREATE SINK sink
+  FROM t
+  INTO KAFKA CONNECTION kafka_conn (TOPIC 'testdrive-alter-sink-${testdrive.seed}')
+  FORMAT JSON
+  ENVELOPE DEBEZIUM
+
+! ALTER SINK sink SET FROM v
+contains:v is a view, which cannot be exported as a sink

test/testdrive/kafka-sink-errors.td

@@ -58,7 +58,7 @@ contains:LEGACY IDs option is not supported
   INTO KAFKA CONNECTION kafka_conn (TOPIC 'testdrive-kafka-sink-errors-${testdrive.seed}')
   FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_conn
   ENVELOPE DEBEZIUM
-contains:catalog item 'materialize.public.s1' is a sink and so cannot be depended upon
+contains:s1 is a sink, which cannot be exported as a sink
 
 ! CREATE VIEW v2 AS SELECT * FROM s1
 contains:catalog item 'materialize.public.s1' is a sink and so cannot be depended upon