test: s3 source with additional column payload

e2e_test/s3/fs_source_v2.py

@@ -69,7 +69,9 @@ def _encode():
         name TEXT,
         sex int,
         mark int,
-    ) WITH (
+    )
+    INCLUDE payload as rw_payload
+    WITH (
         connector = 's3',
         match_pattern = '{prefix}*.{fmt}',
         s3.region_name = '{config['S3_REGION']}',
@@ -105,6 +107,18 @@ def _assert_eq(field, got, expect):
     _assert_eq('sum(sex)', result[2], total_rows / 2)
     _assert_eq('sum(mark)', result[3], 0)
 
+    # check rw_payload
+    print('Check rw_payload')
+    stmt = f"select id, name, sex, mark, rw_payload from {_table()} limit 1;"
+    cur.execute(stmt)
+    result = cur.fetchone()
+    print("Got one line with rw_payload: ", result)
+    payload = json.loads(result[4])
+    _assert_eq('id', payload['id'], result[0])
+    _assert_eq('name', payload['name'], result[1])
+    _assert_eq('sex', payload['sex'], result[2])
+    _assert_eq('mark', payload['mark'], result[3])
+
     print('Test pass')
 
     if need_drop_table: