Repository: ArroyoSystems/arroyo (language: rust)

Issue:
Invalid unstructured json schema in source is not detected at plan time
This query DDL is invalid because it specifies an unstructured JSON schema, but it does not have a single column of type TEXT/JSON:

```sql
CREATE TABLE source (
  x INT
) WITH (
    connector = 'websocket',
    endpoint = 'wss://ws-feed.exchange.coinbase.com',
    subscription_message = '{
      "type": "subscribe",
      "product_ids": [
        "BTC-USD"
      ],
      "channels": ["ticker"]
    }',
    format = 'json',
    'json.unstructured' = 'true'
);
```

This should produce a plan-time failure, but it does not. Instead it produces a runtime failure:

```
2024-08-30T18:08:30.232087Z ERROR arroyo_server_common: panicked at crates/arroyo-formats/src/de.rs:319:14:
no 'value' column for RawString format panic.file="crates/arroyo-formats/src/de.rs" panic.line=319 panic.column=14
```


Discussion / hints:

Fix patch:
diff --git a/crates/arroyo-rpc/src/api_types/connections.rs b/crates/arroyo-rpc/src/api_types/connections.rs
index ac6ec49cd..fccde33f9 100644
--- a/crates/arroyo-rpc/src/api_types/connections.rs
+++ b/crates/arroyo-rpc/src/api_types/connections.rs
@@ -283,6 +283,16 @@ impl ConnectionSchema {
                     bail!("raw_string format requires a schema with a single field called `value` of type TEXT");
                 }
             }
+            Some(Format::Json(json_format)) => {
+                if json_format.unstructured
+                    && (self.fields.len() != 1
+                        || self.fields.first().unwrap().field_type.r#type
+                            != FieldType::Primitive(PrimitiveType::Json)
+                        || self.fields.first().unwrap().field_name != "value")
+                {
+                    bail!("json format with unstructured flag enabled requires a schema with a single field called `value` of type JSON");
+                }
+            }
             _ => {
                 // Right now only RawString has checks, but we may add checks for other formats in the future
             }


Test patch: