access layer

Signed-off-by: tabVersion <[email protected]>

src/connector/src/parser/mod.rs

@@ -488,6 +488,16 @@ impl SourceStreamChunkRowWriter<'_> {
                             .map(|ele| ScalarRefImpl::Utf8(ele.split_id)),
                     ));
                 }
+                (_, &Some(AdditionalColumnType::Payload(_))) => {
+                    // Get the whole payload as a single column
+                    // The frontend check guarantees that row encode must be `SourceEncode::Json`
+                    // here fake a column named "." to represent the whole payload
+                    // see the json accessor hack in `impl Access for JsonAccess<'_>`
+                    let mut desc_mock = desc.clone();
+                    desc_mock.name = ".".to_string();
+                    desc_mock.additional_column.column_type = None;
+                    parse_field(&desc_mock)
+                }
                 (_, _) => {
                     // For normal columns, call the user provided closure.
                     parse_field(desc)

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

@@ -646,6 +646,10 @@ impl<'a> JsonAccess<'a> {
 impl Access for JsonAccess<'_> {
     fn access<'a>(&'a self, path: &[&str], type_expected: &DataType) -> AccessResult<DatumCow<'a>> {
         let mut value = &self.value;
+        if path.len() == 1 && path[0] == "." {
+            return self.options.parse(value, type_expected);
+        }
+
         for (idx, &key) in path.iter().enumerate() {
             if let Some(sub_value) = if self.options.ignoring_keycase {
                 json_object_get_case_insensitive(value, key)