fix NaN

risingwavelabs · Apr 17, 2024 · 037c7e8 · 037c7e8
1 parent 0404896
commit 037c7e8
Show file tree

Hide file tree

Showing 6 changed files with 53 additions and 19 deletions.
diff --git a/e2e_test/source/cdc/cdc.check.slt b/e2e_test/source/cdc/cdc.check.slt
@@ -55,6 +55,8 @@ select * from numeric_to_varchar order by id;
 3 57896044618658097711785492504343953926634992332820282019728792003956564819968
 4 115792089237316195423570985008687907853269984665640564039457584007913129639936
 5 115792089237316195423570985008687907853269984665640564039457584007913129639936.555555
+6 NaN
+7 NULL
 
 # The invalid data for rw_int256 is converted to NULL
 query II
@@ -65,3 +67,5 @@ select * from numeric_to_rw_int256 order by id;
 3 NULL
 4 NULL
 5 NULL
+6 NULL
+7 NULL
diff --git a/e2e_test/source/cdc/cdc.check_new_rows.slt b/e2e_test/source/cdc/cdc.check_new_rows.slt
@@ -87,10 +87,14 @@ select * from numeric_to_varchar order by id;
 3   57896044618658097711785492504343953926634992332820282019728792003956564819968
 4   115792089237316195423570985008687907853269984665640564039457584007913129639936
 5   115792089237316195423570985008687907853269984665640564039457584007913129639936.555555
+6   NaN
+7   NULL
 102 57896044618658097711785492504343953926634992332820282019728792003956564819967
 103 57896044618658097711785492504343953926634992332820282019728792003956564819968
 104 115792089237316195423570985008687907853269984665640564039457584007913129639936
 105 115792089237316195423570985008687907853269984665640564039457584007913129639936.555555
+106 NAN
+107 POSITIVE_INFINITY
 
 # The invalid data for rw_int256 is converted to NULL
 query II
@@ -101,7 +105,11 @@ select * from numeric_to_rw_int256 order by id;
 3   NULL
 4   NULL
 5   NULL
+6   NULL
+7   NULL
 102 57896044618658097711785492504343953926634992332820282019728792003956564819967
 103 NULL
 104 NULL
-105 NULL
+105 NULL
+106 NULL
+107 NULL
diff --git a/e2e_test/source/cdc/cdc.share_stream.slt b/e2e_test/source/cdc/cdc.share_stream.slt
@@ -306,6 +306,8 @@ insert into numeric_table values(103, 578960446186580977117854925043439539266349
 --- 2^256
 insert into numeric_table values(104, 115792089237316195423570985008687907853269984665640564039457584007913129639936);
 insert into numeric_table values(105, 115792089237316195423570985008687907853269984665640564039457584007913129639936.555555);
+insert into numeric_table values(106, 'NaN'::numeric);
+insert into numeric_table values(107, 'Infinity'::numeric);
 "
 
 sleep 3s
@@ -318,10 +320,14 @@ select * from numeric_to_varchar_shared order by id;
 3   57896044618658097711785492504343953926634992332820282019728792003956564819968
 4   115792089237316195423570985008687907853269984665640564039457584007913129639936
 5   115792089237316195423570985008687907853269984665640564039457584007913129639936.555555
+6   NaN
+7   NULL
 102 57896044618658097711785492504343953926634992332820282019728792003956564819967
 103 57896044618658097711785492504343953926634992332820282019728792003956564819968
 104 115792089237316195423570985008687907853269984665640564039457584007913129639936
 105 115792089237316195423570985008687907853269984665640564039457584007913129639936.555555
+106 NAN
+107 POSITIVE_INFINITY
 
 # The invalid data for rw_int256 is converted to NULL
 query II 
@@ -332,24 +338,28 @@ select * from numeric_to_rw_int256_shared order by id;
 3   NULL
 4   NULL
 5   NULL
+6   NULL
+7   NULL
 102 57896044618658097711785492504343953926634992332820282019728792003956564819967
 103 NULL
 104 NULL
 105 NULL
+106 NULL
+107 NULL
 
 system ok
 psql -c "
-DELETE FROM numeric_table WHERE id IN (102, 103, 104, 105);
+DELETE FROM numeric_table WHERE id IN (102, 103, 104, 105, 106, 107);
 "
 
-query II 
+query II
 select * from numeric_list_to_varchar_list_shared order by id;
 ----
-1   3.14
-2   57896044618658097711785492504343953926634992332820282019728792003956564819967
+1 {3.14,6,57896044618658097711785492504343953926634992332820282019728792003956564819967,57896044618658097711785492504343953926634992332820282019728792003956564819968,115792089237316195423570985008687907853269984665640564039457584007913129639936.555555}
+2 {}
 
-query II 
+query II
 select * from numeric_list_to_rw_int256_list_shared order by id;
 ----
-1   3.14
-2   57896044618658097711785492504343953926634992332820282019728792003956564819967
+1 {NULL,6,57896044618658097711785492504343953926634992332820282019728792003956564819967,NULL,NULL}
+2 {}
diff --git a/e2e_test/source/cdc/postgres_cdc.sql b/e2e_test/source/cdc/postgres_cdc.sql
@@ -82,6 +82,8 @@ insert into numeric_table values(3, 57896044618658097711785492504343953926634992
 --- 2^256
 insert into numeric_table values(4, 115792089237316195423570985008687907853269984665640564039457584007913129639936);
 insert into numeric_table values(5, 115792089237316195423570985008687907853269984665640564039457584007913129639936.555555);
+insert into numeric_table values(6, 'NaN'::numeric);
+insert into numeric_table values(7, 'Infinity'::numeric);
 
 create table numeric_list(id int primary key, num numeric[]);
 insert into numeric_list values(1, '{3.14, 6, 57896044618658097711785492504343953926634992332820282019728792003956564819967, 57896044618658097711785492504343953926634992332820282019728792003956564819968, 115792089237316195423570985008687907853269984665640564039457584007913129639936.555555}');

diff --git a/e2e_test/source/cdc/postgres_cdc_insert.sql b/e2e_test/source/cdc/postgres_cdc_insert.sql
@@ -18,3 +18,5 @@ insert into numeric_table values(103, 578960446186580977117854925043439539266349
 --- 2^256
 insert into numeric_table values(104, 115792089237316195423570985008687907853269984665640564039457584007913129639936);
 insert into numeric_table values(105, 115792089237316195423570985008687907853269984665640564039457584007913129639936.555555);
+insert into numeric_table values(106, 'NaN'::numeric);
+insert into numeric_table values(107, 'Infinity'::numeric);
diff --git a/src/connector/src/parser/postgres.rs b/src/connector/src/parser/postgres.rs
@@ -429,8 +429,16 @@ pub fn postgres_row_to_owned_row(row: tokio_postgres::Row, schema: &Schema) -> O
 }
 
 fn pg_numeric_to_rw_int256(val: Option<PgNumeric>) -> Option<ScalarImpl> {
-    let string = pg_numeric_to_string(val)?;
-    match Int256::from_str(string.as_str()) {
+    let val = if let Some(PgNumeric {
+        n: Some(big_decimal),
+    }) = val
+    {
+        big_decimal.to_string()
+    } else {
+        // either NULL or NaN
+        return None;
+    };
+    match Int256::from_str(val.as_str()) {
         Ok(num) => Some(ScalarImpl::from(num)),
         Err(err) => {
             if let Ok(suppressed_count) = LOG_SUPPERSSER.check() {
@@ -446,16 +454,16 @@ fn pg_numeric_to_rw_int256(val: Option<PgNumeric>) -> Option<ScalarImpl> {
 }
 
 fn pg_numeric_to_varchar(val: Option<PgNumeric>) -> Option<ScalarImpl> {
-    pg_numeric_to_string(val).map(ScalarImpl::from)
-}
-
-fn pg_numeric_to_string(val: Option<PgNumeric>) -> Option<String> {
-    if let Some(PgNumeric {
-        n: Some(big_decimal),
-    }) = val
-    {
-        Some(big_decimal.to_string())
+    // TODO(kexiang): support Infinity and -Infinity
+    if let Some(v) = val {
+        if let Some(big_decimal) = v.n {
+            Some(ScalarImpl::from(big_decimal.to_string()))
+        } else {
+            // NaN
+            Some(ScalarImpl::from(String::from("NaN")))
+        }
     } else {
+        // NULL
         None
     }
 }