fix(connector): warn on undefined nested field in struct when parsing JSON

src/common/src/types/mod.rs

@@ -1220,7 +1220,7 @@ mod tests {
         );
         assert_eq!(
             format!("{}", d),
-            "struct<i integer,j character varying>".to_string()
+            "struct<i integer, j character varying>".to_string()
         );
     }
 

src/common/src/types/struct_type.rs

@@ -140,7 +140,7 @@ impl Display for StructType {
                 (self.0.field_types.iter())
                     .zip_eq_fast(self.0.field_names.iter())
                     .map(|(d, s)| format!("{} {}", s, d))
-                    .join(",")
+                    .join(", ")
             )
         }
     }

src/connector/src/parser/json_parser.rs

@@ -515,6 +515,56 @@ mod tests {
         assert_eq!(row, expected.into());
     }
 
+    #[cfg(not(madsim))] // Traced test does not work with madsim
+    #[tokio::test]
+    #[tracing_test::traced_test]
+    async fn test_json_parse_struct_missing_field_warning() {
+        let descs = vec![ColumnDesc::new_struct(
+            "struct",
+            0,
+            "",
+            vec![
+                ColumnDesc::new_atomic(DataType::Varchar, "varchar", 1),
+                ColumnDesc::new_atomic(DataType::Boolean, "boolean", 2),
+            ],
+        )]
+        .iter()
+        .map(SourceColumnDesc::from)
+        .collect_vec();
+
+        let parser = JsonParser::new(
+            SpecificParserConfig::DEFAULT_PLAIN_JSON,
+            descs.clone(),
+            Default::default(),
+        )
+        .unwrap();
+        let payload = br#"
+        {
+            "struct": {
+                "varchar": "varchar"
+            }
+        }
+        "#
+        .to_vec();
+        let mut builder = SourceStreamChunkBuilder::with_capacity(descs, 1);
+        {
+            let writer = builder.row_writer();
+            parser.parse_inner(payload, writer).await.unwrap();
+        }
+        let chunk = builder.finish();
+        let (op, row) = chunk.rows().next().unwrap();
+        assert_eq!(op, Op::Insert);
+        let row = row.into_owned_row().into_inner();
+
+        let expected = vec![Some(ScalarImpl::Struct(StructValue::new(vec![
+            Some(ScalarImpl::Utf8("varchar".into())),
+            None,
+        ])))];
+        assert_eq!(row, expected.into());
+
+        assert!(logs_contain("undefined nested field, padding with `NULL`"));
+    }
+
     #[tokio::test]
     async fn test_json_upsert_parser() {
         let items = [

src/connector/src/parser/unified/json.rs

@@ -457,11 +457,17 @@ impl JsonParseOptions {
                     .names()
                     .zip_eq_fast(struct_type_info.types())
                     .map(|(field_name, field_type)| {
-                        self.parse(
-                            json_object_get_case_insensitive(value, field_name)
-                                .unwrap_or(&BorrowedValue::Static(simd_json::StaticNode::Null)),
-                            Some(field_type),
-                        )
+                        let field_value = json_object_get_case_insensitive(value, field_name)
+                            .unwrap_or_else(|| {
+                                let error = AccessError::Undefined {
+                                    name: field_name.to_owned(),
+                                    path: struct_type_info.to_string(), // TODO: this is not good, we should maintain a path stack
+                                };
+                                // TODO: is it possible to unify the logging with the one in `do_action`?
+                                tracing::warn!(%error, "undefined nested field, padding with `NULL`");
+                                &BorrowedValue::Static(simd_json::StaticNode::Null)
+                            });
+                        self.parse(field_value, Some(field_type))
                     })
                     .collect::<Result<_, _>>()?,
             )

src/frontend/planner_test/tests/testdata/output/struct_query.yaml

@@ -300,7 +300,7 @@
     Failed to bind expression: avg(country)
 
     Caused by:
-      Invalid input syntax: Invalid aggregation: avg(struct<address character varying,zipcode character varying>)
+      Invalid input syntax: Invalid aggregation: avg(struct<address character varying, zipcode character varying>)
   create_source:
     format: plain
     encode: protobuf
@@ -422,7 +422,7 @@
 - sql: |
     CREATE TABLE a (c STRUCT<i STRUCT<a INTEGER>, j INTEGER>);
     INSERT INTO a VALUES (1);
-  binder_error: 'Bind error: cannot cast type "integer" to "struct<i struct<a integer>,j integer>" in Assign context'
+  binder_error: 'Bind error: cannot cast type "integer" to "struct<i struct<a integer>, j integer>" in Assign context'
 - name: test struct type alignment in CASE expression
   sql: |
     select CASE WHEN false THEN ROW(0, INTERVAL '1') WHEN true THEN ROW(1.1, INTERVAL '1') ELSE ROW(1, INTERVAL '1') END;