fix

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

e2e_test/source/basic/kafka.slt

@@ -498,7 +498,7 @@ FORMAT DEBEZIUM ENCODE JSON (
   ignore_key = 'true'
 )
 
-statement error Payload column is only allowed when row encode is JSON, but got Bytes
+statement error INCLUDE payload is only allowed when using ENCODE JSON, but got ENCODE Bytes
 CREATE TABLE test_include_payload (a bytea)
 INCLUDE payload
 WITH (

src/connector/src/parser/mod.rs

@@ -488,16 +488,7 @@ 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)
-                }
+                (_, &Some(AdditionalColumnType::Payload(_))) => parse_field(desc),
                 (_, _) => {
                     // For normal columns, call the user provided closure.
                     parse_field(desc)

src/connector/src/parser/unified/kv_event.rs

@@ -79,6 +79,9 @@ where
     pub fn access_field(&self, desc: &SourceColumnDesc) -> AccessResult<DatumCow<'_>> {
         match desc.additional_column.column_type {
             Some(AdditionalColumnType::Key(_)) => self.access_key(&[&desc.name], &desc.data_type),
+            // hack here: Get the whole payload as a single column
+            // use a special mark "." as column name to represent the whole payload
+            Some(AdditionalColumnType::Payload(_)) => self.access_value(&[&"."], &desc.data_type),
             None => self.access_value(&[&desc.name], &desc.data_type),
             _ => unreachable!(),
         }