Report planner error when inserting into sources or select from sinks

Author: mwylde

crates/arroyo-planner/src/builder.rs

@@ -323,9 +323,8 @@ impl PlanToGraphVisitor<'_> {
             })
             .collect::<Result<Vec<_>>>()?;
 
-        let NodeWithIncomingEdges { node, edges } = extension
-            .plan_node(&self.planner, self.graph.node_count(), input_schemas)
-            .map_err(|e| e.context(format!("planning operator {extension:?}")))?;
+        let NodeWithIncomingEdges { node, edges } =
+            extension.plan_node(&self.planner, self.graph.node_count(), input_schemas)?;
 
         let node_index = self.graph.add_node(node);
         self.add_index_to_traversal(node_index);

crates/arroyo-planner/src/extension/sink.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 use arroyo_datastream::logical::{LogicalEdge, LogicalEdgeType, LogicalNode, OperatorName};
 use arroyo_rpc::{
     UPDATING_META_FIELD,
+    api_types::connections::ConnectionType,
     df::{ArroyoSchema, ArroyoSchemaRef},
 };
 use datafusion::common::{DFSchemaRef, Result, TableReference, plan_err};
@@ -46,6 +47,13 @@ impl SinkExtension {
             .has_column_with_unqualified_name(UPDATING_META_FIELD);
         match &table {
             Table::ConnectorTable(connector_table) => {
+                if connector_table.connection_type == ConnectionType::Source {
+                    return plan_err!(
+                        "attempted to insert into table '{}', but it is a source",
+                        connector_table.name
+                    );
+                }
+
                 match (input_is_updating, connector_table.is_updating()) {
                     (_, true) => {
                         let to_debezium_extension =

crates/arroyo-planner/src/rewriters.rs

@@ -15,8 +15,7 @@ use crate::{
 };
 
 use arrow_schema::DataType;
-use arroyo_rpc::TIMESTAMP_FIELD;
-use arroyo_rpc::UPDATING_META_FIELD;
+use arroyo_rpc::{TIMESTAMP_FIELD, UPDATING_META_FIELD, api_types::connections::ConnectionType};
 use datafusion::logical_expr::UserDefinedLogicalNode;
 
 use crate::extension::AsyncUDFExtension;
@@ -192,6 +191,13 @@ impl SourceRewriter<'_> {
         table_scan: &TableScan,
         table: &ConnectorTable,
     ) -> DFResult<Transformed<LogicalPlan>> {
+        if table.connection_type == ConnectionType::Sink {
+            return plan_err!(
+                "attempted to read from table '{}', but it is a sink",
+                table.name
+            );
+        }
+
         let input = self.projection(table_scan, table)?;
 
         let schema = input.schema().clone();

crates/arroyo-planner/src/test/queries/filesystem_invalid_partition.sql

@@ -37,5 +37,5 @@ create table events_sink (
     type = 'sink'
 );
 
-INSERT INTO events
+INSERT INTO events_sink
 SELECT * from events;

crates/arroyo-planner/src/test/queries/filesystem_partition.sql

@@ -36,5 +36,5 @@ create table events_sink (
     'rolling_policy.interval' = interval '6000 seconds'
 );
 
-INSERT INTO events
+INSERT INTO events_sink
 SELECT * from events;

crates/arroyo-planner/src/test/queries/insert_into_source.sql

@@ -0,0 +1,8 @@
+--fail=attempted to insert into table 'source', but it is a source
+CREATE TABLE source with (
+    connector = 'impulse',
+    event_rate = 10
+);
+
+INSERT INTO source
+select * from source;

crates/arroyo-planner/src/test/queries/select_from_sink.sql

@@ -0,0 +1,26 @@
+--fail=attempted to read from table 'cars_output', but it is a sink
+
+CREATE TABLE cars (
+	timestamp TIMESTAMP,
+	driver_id BIGINT,
+	event_type TEXT,
+	location TEXT
+) WITH (
+	connector = 'single_file',
+	path = 'cars.json',
+	format = 'json',
+	type = 'source'
+);
+
+CREATE TABLE cars_output (
+	timestamp TIMESTAMP,
+	driver_id BIGINT,
+	event_type TEXT,
+	location TEXT
+) WITH (
+	connector = 'single_file',
+	path = 'cars_output.json',
+	format = 'json',
+	type = 'sink'
+);
+INSERT INTO cars_output SELECT * from cars_output;