support casting that doesn't change types, for virtual fields. (#436)

arroyo-controller/src/states/running.rs

@@ -111,6 +111,11 @@ impl State for Running {
                         },
                         Err(err) => {
                             error!(message = "error while running", error = format!("{:?}", err), job_id = ctx.config.id);
+                            log_event("running_error", json!({
+                                "service": "controller",
+                                "job_id": ctx.config.id,
+                                "error": format!("{:?}", err),
+                            }));
                             if ctx.status.restarts >= RESTARTS_ALLOWED as i32 {
                                 return Err(fatal(
                                     "Job has restarted too many times",

arroyo-sql-testing/src/full_query_tests.rs

@@ -557,3 +557,18 @@ INSERT INTO avro_writes
 select cast(bid.datetime as TIMESTAMP), bid.auction as u64, bid.auction as i64, bid.extra
 from nexmark;"
 }
+
+full_pipeline_codegen! {
+  "raw_string_cast",
+  "create table logs (
+    value TEXT NOT NULL,
+    parsed TEXT GENERATED ALWAYS AS (value)
+) with (
+    connector = 'sse',
+    endpoint = 'http://host.docker.internal:9563/sse',
+    format = 'raw_string'
+);
+
+SELECT * from logs;
+"
+}

arroyo-sql/src/expressions.rs

@@ -1882,6 +1882,9 @@ impl CastExpression {
         matches!(data_type, DataType::Utf8 | DataType::LargeUtf8)
     }
     fn cast_expr(input_type: &DataType, output_type: &DataType, sub_expr: syn::Expr) -> syn::Expr {
+        if input_type == output_type {
+            return sub_expr;
+        }
         if Self::is_numeric(input_type) && Self::is_numeric(output_type) {
             let cast_type: syn::Type =
                 parse_str(&StructField::data_type_name(output_type)).unwrap();