refactor(postgres-cdc): refactor postgres_row_to_owned_row

e2e_test/source/cdc/cdc.check_new_rows.slt

@@ -128,6 +128,8 @@ select id, my_int from list_with_null_shared order by id;
 2 {NULL,3,4}
 3 {NULL,-3,-4}
 4 {-4,-5,-6}
+5 NULL
+6 NULL
 
 # my_num: varchar[]
 query II
@@ -137,6 +139,8 @@ select id, my_num from list_with_null_shared order by id;
 2 {2.2,0,NULL}
 3 NULL
 4 {NULL,-99999999999999999.9999}
+5 NULL
+6 NULL
 
 # my_num1: numeric[]
 query II
@@ -146,6 +150,8 @@ select id, my_num_1 from list_with_null_shared order by id;
 2 {2.2,0,NULL}
 3 NULL
 4 {NULL,-99999999999999999.9999}
+5 NULL
+6 NULL
 
 # my_num2: rw_int256[]
 query II
@@ -155,6 +161,8 @@ select id, my_num_2 from list_with_null_shared order by id;
 2 NULL
 3 NULL
 4 NULL
+5 NULL
+6 NULL
 
 # Due to the bug in Debezium, if a enum list contains `NULL`, the list will be converted to `NULL`
 query II
@@ -164,6 +172,8 @@ select id, my_mood from list_with_null_shared order by id;
 2 {happy,ok,sad}
 3 NULL
 4 NULL
+5 NULL
+6 NULL
 
 query II
 select id, my_uuid from list_with_null_shared order by id;
@@ -172,6 +182,8 @@ select id, my_uuid from list_with_null_shared order by id;
 2 {2de296df-eda7-4202-a81f-1036100ef4f6,2977afbc-0b12-459c-a36f-f623fc9e9840}
 3 {NULL,471acecf-a4b4-4ed3-a211-7fb2291f159f,9bc35adf-fb11-4130-944c-e7eadb96b829}
 4 {b2e4636d-fa03-4ad4-bf16-029a79dca3e2}
+5 NULL
+6 NULL
 
 query II
 select id, my_bytea from list_with_null_shared order by id;
@@ -180,3 +192,17 @@ select id, my_bytea from list_with_null_shared order by id;
 2 {"\\x00","\\x01","\\x02"}
 3 {NULL,"\\x99","\\xaa"}
 4 {"\\x88","\\x99","\\xaa"}
+5 NULL
+6 NULL
+
+query TTTTTTT
+SELECT * FROM postgres_all_types order by c_boolean,c_bigint,c_date;
+----
+f -32767 -2147483647 -9223372036854775807 -10.0 -10000 -10000 (empty) \x00 0001-01-01 00:00:00 2001-01-01 00:00:00 2001-01-01 08:00:00+00:00 00:00:00 {} bb488f9b-330d-4012-b849-12adeb49e57e happy {f} {-32767} {-2147483647} {-9223372036854775807} {-10.0} {-10000} {-10000} {""} {"\\x00"} {0001-01-01} {00:00:00} {"2001-01-01 00:00:00"} {"2001-01-01 08:00:00+00:00"} NULL {"{}"} {bb488f9b-330d-4012-b849-12adeb49e57e} NULL
+f 0 0 0 0 0 0 (empty) \x00 0001-01-01 00:00:00 2001-01-01 00:00:00 2001-01-01 08:00:00+00:00 00:00:00 {} NULL sad {} {} {} {} {} {} {} {} {} {} {} {} {} NULL {} {} {}
+f NULL NULL 1 NULL NULL NULL NULL NULL 0024-05-19 NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL
+f 1 123 1234567890 123.45 123.45 123.456 example \xdeadbeef 0024-01-01 12:34:56 2024-05-19 12:34:56 2024-05-19 12:34:56+00:00 1 day {"key": "value"} 123e4567-e89b-12d3-a456-426614174000 happy {NULL,t} {NULL,1} {NULL,123} {NULL,1234567890} {NULL,123.45} {NULL,123.45} {NULL,123.456} {NULL,example} {NULL,"\\xdeadbeef"} {NULL,2024-05-19} {NULL,12:34:56} {NULL,"2024-05-19 12:34:56"} {NULL,"2024-05-19 12:34:56+00:00"} NULL {NULL,"{\"key\": \"value\"}"} {NULL,123e4567-e89b-12d3-a456-426614174000} NULL
+t -32767 -2147483647 -9223372036854775807 -10.0 -10000 -10000 (empty) \x00 0001-01-01 00:00:00 2001-01-01 00:00:00 2001-01-01 08:00:00+00:00 00:00:00 {} bb488f9b-330d-4012-b849-12adeb49e57e happy {f} {-32767} {-2147483647} {-9223372036854775807} {-10.0} {-10000} {-10000} {""} {"\\x00"} {0001-01-01} {00:00:00} {"2001-01-01 00:00:00"} {"2001-01-01 08:00:00+00:00"} NULL {"{}"} {bb488f9b-330d-4012-b849-12adeb49e57e} NULL
+t 0 0 0 0 0 0 (empty) \x00 0001-01-01 00:00:00 2001-01-01 00:00:00 2001-01-01 08:00:00+00:00 00:00:00 {} NULL sad {} {} {} {} {} {} {} {} {} {} {} {} {} NULL {} {} {}
+t NULL NULL 1 NULL NULL NULL NULL NULL 0024-05-19 NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL NULL
+t 1 123 1234567890 123.45 123.45 123.456 example \xdeadbeef 0024-01-01 12:34:56 2024-05-19 12:34:56 2024-05-19 12:34:56+00:00 1 day {"key": "value"} 123e4567-e89b-12d3-a456-426614174000 happy {NULL,t} {NULL,1} {NULL,123} {NULL,1234567890} {NULL,123.45} {NULL,123.45} {NULL,123.456} {NULL,example} {NULL,"\\xdeadbeef"} {NULL,2024-05-19} {NULL,12:34:56} {NULL,"2024-05-19 12:34:56"} {NULL,"2024-05-19 12:34:56+00:00"} NULL {NULL,"{\"key\": \"value\"}"} {NULL,123e4567-e89b-12d3-a456-426614174000} NULL

e2e_test/source/cdc/cdc.share_stream.slt

@@ -193,6 +193,8 @@ CREATE TABLE IF NOT EXISTS postgres_all_types(
  c_timestamptz timestamptz,
  c_interval interval,
  c_jsonb jsonb,
+ c_uuid varchar,
+ c_enum varchar,
  c_boolean_array boolean[],
  c_smallint_array smallint[],
  c_integer_array integer[],
@@ -208,7 +210,8 @@ CREATE TABLE IF NOT EXISTS postgres_all_types(
  c_timestamptz_array timestamptz[],
  c_interval_array interval[],
  c_jsonb_array jsonb[],
- c_uuid varchar,
+ c_uuid_array varchar[],
+ c_enum_array varchar[],
  PRIMARY KEY (c_boolean,c_bigint,c_date)
 ) from pg_source table 'public.postgres_all_types';
 
@@ -240,7 +243,7 @@ sleep 3s
 query TTTTTTT
 SELECT c_boolean,c_date,c_time,c_timestamp,c_jsonb,c_smallint_array,c_timestamp_array,c_uuid FROM postgres_all_types where c_bigint=-9223372036854775807
 ----
-f 0001-01-01 00:00:00 0001-01-01 00:00:00 {} {-32767} {"0001-01-01 00:00:00"} bb488f9b-330d-4012-b849-12adeb49e57e
+f 0001-01-01 00:00:00 2001-01-01 00:00:00 {} {-32767} {"2001-01-01 00:00:00"} bb488f9b-330d-4012-b849-12adeb49e57e
 
 
 # postgres streaming test

e2e_test/source/cdc/postgres_cdc.sql

@@ -35,6 +35,7 @@ create table abs.t1 ("V1" int primary key, v2 double precision, v3 varchar, v4 n
 create publication my_publicaton for table abs.t1 ("V1", v3);
 insert into abs.t1 values (1, 1.1, 'aaa', '5431.1234');
 
+CREATE TYPE mood AS ENUM ('sad', 'ok', 'happy');
 
 CREATE TABLE IF NOT EXISTS postgres_all_types(
      c_boolean boolean,
@@ -52,6 +53,8 @@ CREATE TABLE IF NOT EXISTS postgres_all_types(
      c_timestamptz timestamptz,
      c_interval interval,
      c_jsonb jsonb,
+     c_uuid uuid,
+     c_enum mood,
      c_boolean_array boolean[],
      c_smallint_array smallint[],
      c_integer_array integer[],
@@ -67,11 +70,14 @@ CREATE TABLE IF NOT EXISTS postgres_all_types(
      c_timestamptz_array timestamptz[],
      c_interval_array interval[],
      c_jsonb_array jsonb[],
-     c_uuid uuid,
+     c_uuid_array uuid[],
+     c_enum_array mood[],
      PRIMARY KEY (c_boolean,c_bigint,c_date)
 );
-INSERT INTO postgres_all_types VALUES ( False, 0, 0, 0, 0, 0, 0, '', '\x00', '0001-01-01', '00:00:00', '0001-01-01 00:00:00'::timestamp, '0001-01-01 00:00:00'::timestamptz, interval '0 second', '{}', array[]::boolean[], array[]::smallint[], array[]::integer[], array[]::bigint[], array[]::decimal[], array[]::real[], array[]::double precision[], array[]::varchar[], array[]::bytea[], array[]::date[], array[]::time[], array[]::timestamp[], array[]::timestamptz[], array[]::interval[], array[]::jsonb[], null);
-INSERT INTO postgres_all_types VALUES ( False, -32767, -2147483647, -9223372036854775807, -10.0, -9999.999999, -10000.0, '', '\x00', '0001-01-01', '00:00:00', '0001-01-01 00:00:00'::timestamp, '0001-01-01 00:00:00'::timestamptz, interval '0 second', '{}', array[False::boolean]::boolean[], array[-32767::smallint]::smallint[], array[-2147483647::integer]::integer[], array[-9223372036854775807::bigint]::bigint[], array[-10.0::decimal]::decimal[], array[-9999.999999::real]::real[], array[-10000.0::double precision]::double precision[], array[''::varchar]::varchar[], array['\x00'::bytea]::bytea[], array['0001-01-01'::date]::date[], array['00:00:00'::time]::time[], array['0001-01-01 00:00:00'::timestamp::timestamp]::timestamp[], array['0001-01-01 00:00:00'::timestamptz::timestamptz]::timestamptz[], array[interval '0 second'::interval]::interval[], array['{}'::jsonb]::jsonb[], 'bb488f9b-330d-4012-b849-12adeb49e57e');
+INSERT INTO postgres_all_types VALUES ( False, 0, 0, 0, 0, 0, 0, '', '\x00', '0001-01-01', '00:00:00', '2001-01-01 00:00:00'::timestamp, '2001-01-01 00:00:00-8'::timestamptz, interval '0 second', '{}', null, 'sad', array[]::boolean[], array[]::smallint[], array[]::integer[], array[]::bigint[], array[]::decimal[], array[]::real[], array[]::double precision[], array[]::varchar[], array[]::bytea[], array[]::date[], array[]::time[], array[]::timestamp[], array[]::timestamptz[], array[]::interval[], array[]::jsonb[], array[]::uuid[], array[]::mood[]);
+INSERT INTO postgres_all_types VALUES ( False, -32767, -2147483647, -9223372036854775807, -10.0, -9999.999999, -10000.0, '', '\x00', '0001-01-01', '00:00:00', '2001-01-01 00:00:00'::timestamp, '2001-01-01 00:00:00-8'::timestamptz, interval '0 second', '{}', 'bb488f9b-330d-4012-b849-12adeb49e57e', 'happy', array[False::boolean]::boolean[], array[-32767::smallint]::smallint[], array[-2147483647::integer]::integer[], array[-9223372036854775807::bigint]::bigint[], array[-10.0::decimal]::decimal[], array[-9999.999999::real]::real[], array[-10000.0::double precision]::double precision[], array[''::varchar]::varchar[], array['\x00'::bytea]::bytea[], array['0001-01-01'::date]::date[], array['00:00:00'::time]::time[], array['2001-01-01 00:00:00'::timestamp::timestamp]::timestamp[], array['2001-01-01 00:00:00-8'::timestamptz::timestamptz]::timestamptz[], array[interval '0 second'::interval]::interval[], array['{}'::jsonb]::jsonb[], '{bb488f9b-330d-4012-b849-12adeb49e57e}', '{happy,ok,NULL}');
+INSERT INTO postgres_all_types VALUES ( False, 1, 123, 1234567890, 123.45, 123.45, 123.456, 'example', '\xDEADBEEF', '0024-01-01', '12:34:56', '2024-05-19 12:34:56', '2024-05-19 12:34:56+00', INTERVAL '1 day', '{"key": "value"}', '123e4567-e89b-12d3-a456-426614174000', 'happy', ARRAY[NULL, TRUE]::boolean[], ARRAY[NULL, 1::smallint], ARRAY[NULL, 123], ARRAY[NULL, 1234567890], ARRAY[NULL, 123.45::numeric], ARRAY[NULL, 123.45::real], ARRAY[NULL, 123.456], ARRAY[NULL, 'example'], ARRAY[NULL, '\xDEADBEEF'::bytea], ARRAY[NULL, '2024-05-19'::date], ARRAY[NULL, '12:34:56'::time], ARRAY[NULL, '2024-05-19 12:34:56'::timestamp], ARRAY[NULL, '2024-05-19 12:34:56+00'::timestamptz], ARRAY[NULL, INTERVAL '1 day'], ARRAY[NULL, '{"key": "value"}'::jsonb], ARRAY[NULL, '123e4567-e89b-12d3-a456-426614174000'::uuid], ARRAY[NULL, 'happy'::mood]);
+INSERT INTO postgres_all_types VALUES ( False, NULL, NULL, 1, NULL, NULL, NULL, NULL, NULL, '0024-05-19', NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL);
 
 create table numeric_table(id int PRIMARY KEY, num numeric);
 insert into numeric_table values(1, 3.14);
@@ -89,8 +95,6 @@ create table numeric_list(id int primary key, num numeric[]);
 insert into numeric_list values(1, '{3.14, 6, 57896044618658097711785492504343953926634992332820282019728792003956564819967, 57896044618658097711785492504343953926634992332820282019728792003956564819968, 115792089237316195423570985008687907853269984665640564039457584007913129639936.555555}');
 insert into numeric_list values(2, '{nan, infinity, -infinity}');
 
---- for https://github.com/risingwavelabs/risingwave/issues/16392
-CREATE TYPE mood AS ENUM ('sad', 'ok', 'happy');
 CREATE TABLE enum_table (
     id int PRIMARY KEY,
     current_mood mood
@@ -100,3 +104,4 @@ INSERT INTO enum_table VALUES (1, 'happy');
 CREATE TABLE list_with_null(id int primary key, my_int int[], my_num numeric[], my_num_1 numeric[], my_num_2 numeric[], my_mood mood[], my_uuid uuid[], my_bytea bytea[]);
 INSERT INTO list_with_null VALUES (1, '{1,2,NULL}', '{1.1,inf,NULL}', '{1.1,inf,NULL}', '{1.1,inf,NULL}', '{happy,ok,NULL}', '{bb488f9b-330d-4012-b849-12adeb49e57e,bb488f9b-330d-4012-b849-12adeb49e57f, NULL}', '{\\x00,\\x01,NULL}');
 INSERT INTO list_with_null VALUES (2, '{NULL,3,4}', '{2.2,0,NULL}' , '{2.2,0,NULL}', '{2.2,0,NULL}', '{happy,ok,sad}', '{2de296df-eda7-4202-a81f-1036100ef4f6,2977afbc-0b12-459c-a36f-f623fc9e9840}', '{\\x00,\\x01,\\x02}');
+INSERT INTO list_with_null VALUES (5, NULL, NULL, NULL, NULL, NULL, NULL, NULL);

e2e_test/source/cdc/postgres_cdc_insert.sql

@@ -12,6 +12,11 @@ SELECT pg_current_wal_lsn();
 select * from pg_publication_tables where pubname='rw_publication';
 select * from public.person order by id;
 
+INSERT INTO postgres_all_types VALUES ( True, 0, 0, 0, 0, 0, 0, '', '\x00', '0001-01-01', '00:00:00', '2001-01-01 00:00:00'::timestamp, '2001-01-01 00:00:00-8'::timestamptz, interval '0 second', '{}', null, 'sad', array[]::boolean[], array[]::smallint[], array[]::integer[], array[]::bigint[], array[]::decimal[], array[]::real[], array[]::double precision[], array[]::varchar[], array[]::bytea[], array[]::date[], array[]::time[], array[]::timestamp[], array[]::timestamptz[], array[]::interval[], array[]::jsonb[], array[]::uuid[], array[]::mood[]);
+INSERT INTO postgres_all_types VALUES ( True, -32767, -2147483647, -9223372036854775807, -10.0, -9999.999999, -10000.0, '', '\x00', '0001-01-01', '00:00:00', '2001-01-01 00:00:00'::timestamp, '2001-01-01 00:00:00-8'::timestamptz, interval '0 second', '{}', 'bb488f9b-330d-4012-b849-12adeb49e57e', 'happy', array[False::boolean]::boolean[], array[-32767::smallint]::smallint[], array[-2147483647::integer]::integer[], array[-9223372036854775807::bigint]::bigint[], array[-10.0::decimal]::decimal[], array[-9999.999999::real]::real[], array[-10000.0::double precision]::double precision[], array[''::varchar]::varchar[], array['\x00'::bytea]::bytea[], array['0001-01-01'::date]::date[], array['00:00:00'::time]::time[], array['2001-01-01 00:00:00'::timestamp::timestamp]::timestamp[], array['2001-01-01 00:00:00-8'::timestamptz::timestamptz]::timestamptz[], array[interval '0 second'::interval]::interval[], array['{}'::jsonb]::jsonb[], '{bb488f9b-330d-4012-b849-12adeb49e57e}', '{happy,ok,NULL}');
+INSERT INTO postgres_all_types VALUES ( True, 1, 123, 1234567890, 123.45, 123.45, 123.456, 'example', '\xDEADBEEF', '0024-01-01', '12:34:56', '2024-05-19 12:34:56', '2024-05-19 12:34:56+00', INTERVAL '1 day', '{"key": "value"}', '123e4567-e89b-12d3-a456-426614174000', 'happy', ARRAY[NULL, TRUE]::boolean[], ARRAY[NULL, 1::smallint], ARRAY[NULL, 123], ARRAY[NULL, 1234567890], ARRAY[NULL, 123.45::numeric], ARRAY[NULL, 123.45::real], ARRAY[NULL, 123.456], ARRAY[NULL, 'example'], ARRAY[NULL, '\xDEADBEEF'::bytea], ARRAY[NULL, '2024-05-19'::date], ARRAY[NULL, '12:34:56'::time], ARRAY[NULL, '2024-05-19 12:34:56'::timestamp], ARRAY[NULL, '2024-05-19 12:34:56+00'::timestamptz], ARRAY[NULL, INTERVAL '1 day'], ARRAY[NULL, '{"key": "value"}'::jsonb], ARRAY[NULL, '123e4567-e89b-12d3-a456-426614174000'::uuid], ARRAY[NULL, 'happy'::mood]);
+INSERT INTO postgres_all_types VALUES ( True, NULL, NULL, 1, NULL, NULL, NULL, NULL, NULL, '0024-05-19', NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL);
+
 insert into numeric_table values(102, 57896044618658097711785492504343953926634992332820282019728792003956564819967);
 --- 2^255
 insert into numeric_table values(103, 57896044618658097711785492504343953926634992332820282019728792003956564819968);
@@ -25,3 +30,5 @@ INSERT INTO enum_table VALUES (3, 'sad');
 --- to avoid escaping issues of psql -c "", we insert this row here and check the result in check_new_rows.slt
 INSERT INTO list_with_null VALUES (3, '{NULL,-3,-4}', '{NULL,nan,-inf}', '{NULL,nan,-inf}', '{NULL,nan,-inf}', '{NULL,sad,ok}', '{NULL,471acecf-a4b4-4ed3-a211-7fb2291f159f,9bc35adf-fb11-4130-944c-e7eadb96b829}', '{NULL,\\x99,\\xAA}');
 INSERT INTO list_with_null VALUES (4, '{-4,-5,-6}', '{NULL,-99999999999999999.9999}', '{NULL,-99999999999999999.9999}', '{NULL,-99999999999999999.9999}', '{NULL,sad,ok}', '{b2e4636d-fa03-4ad4-bf16-029a79dca3e2}', '{\\x88,\\x99,\\xAA}');
+INSERT INTO list_with_null VALUES (6, NULL, NULL, NULL, NULL, NULL, NULL, NULL);
+

java/connector-node/risingwave-connector-service/src/main/resources/debezium.properties

@@ -11,7 +11,7 @@ decimal.handling.mode=${debezium.decimal.handling.mode:-string}
 interval.handling.mode=string
 max.batch.size=${debezium.max.batch.size:-1024}
 max.queue.size=${debezium.max.queue.size:-8192}
-time.precision.mode=connect
+time.precision.mode=adaptive_time_microseconds
 # Quoted from the debezium document:
 # > Your application should always properly stop the engine to ensure graceful and complete
 # > shutdown and that each source record is sent to the application exactly one time.

src/common/src/types/datetime.rs

@@ -22,7 +22,7 @@ use std::str::FromStr;
 
 use bytes::{Bytes, BytesMut};
 use chrono::{Datelike, Days, Duration, NaiveDate, NaiveDateTime, NaiveTime, Timelike, Weekday};
-use postgres_types::{accepts, to_sql_checked, IsNull, ToSql, Type};
+use postgres_types::{accepts, to_sql_checked, FromSql, IsNull, ToSql, Type};
 use risingwave_common_estimate_size::ZeroHeapSize;
 use thiserror::Error;
 
@@ -88,6 +88,20 @@ impl ToSql for Date {
     }
 }
 
+impl<'a> FromSql<'a> for Date {
+    fn from_sql(
+        ty: &Type,
+        raw: &'a [u8],
+    ) -> std::result::Result<Self, Box<dyn std::error::Error + Sync + Send>> {
+        let instant = NaiveDate::from_sql(ty, raw)?;
+        Ok(Self::from(instant))
+    }
+
+    fn accepts(ty: &Type) -> bool {
+        matches!(*ty, Type::DATE)
+    }
+}
+
 impl ToSql for Time {
     accepts!(TIME);
 
@@ -105,6 +119,20 @@ impl ToSql for Time {
     }
 }
 
+impl<'a> FromSql<'a> for Time {
+    fn from_sql(
+        ty: &Type,
+        raw: &'a [u8],
+    ) -> std::result::Result<Self, Box<dyn std::error::Error + Sync + Send>> {
+        let instant = NaiveTime::from_sql(ty, raw)?;
+        Ok(Self::from(instant))
+    }
+
+    fn accepts(ty: &Type) -> bool {
+        matches!(*ty, Type::TIME)
+    }
+}
+
 impl ToSql for Timestamp {
     accepts!(TIMESTAMP);
 
@@ -122,6 +150,20 @@ impl ToSql for Timestamp {
     }
 }
 
+impl<'a> FromSql<'a> for Timestamp {
+    fn from_sql(
+        ty: &Type,
+        raw: &'a [u8],
+    ) -> std::result::Result<Self, Box<dyn std::error::Error + Sync + Send>> {
+        let instant = NaiveDateTime::from_sql(ty, raw)?;
+        Ok(Self::from(instant))
+    }
+
+    fn accepts(ty: &Type) -> bool {
+        matches!(*ty, Type::TIMESTAMP)
+    }
+}
+
 /// Parse a date from varchar.
 ///
 /// # Example