add date support

pg_replicate/src/clients/bigquery.rs

@@ -1,7 +1,7 @@
 use std::{collections::HashSet, fs};
 
 use bytes::{Buf, BufMut};
-use chrono::{DateTime, NaiveDateTime, Utc};
+use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc};
 use futures::StreamExt;
 use gcp_bigquery_client::yup_oauth2::parse_service_account_key;
 use gcp_bigquery_client::{
@@ -75,8 +75,9 @@ impl BigQueryClient {
             &Type::NUMERIC => "bignumeric",
             &Type::BOOL => "bool",
             &Type::BYTEA => "bytes",
-            &Type::VARCHAR | &Type::BPCHAR | &Type::TEXT => "string",
+            &Type::CHAR | &Type::BPCHAR | &Type::VARCHAR | &Type::NAME | &Type::TEXT => "string",
             &Type::TIMESTAMP | &Type::TIMESTAMPTZ => "timestamp",
+            &Type::DATE => "date",
             _ => "bytes",
         }
     }
@@ -351,6 +352,7 @@ impl BigQueryClient {
             Cell::F32(i) => s.push_str(&format!("{i}")),
             Cell::F64(i) => s.push_str(&format!("{i}")),
             Cell::Numeric(n) => s.push_str(&format!("{n}")),
+            Cell::Date(t) => s.push_str(&format!("'{t}'")),
             Cell::TimeStamp(t) => s.push_str(&format!("'{t}'")),
             Cell::TimeStampTz(t) => s.push_str(&format!("'{t}'")),
             Cell::Bytes(b) => {
@@ -543,6 +545,10 @@ impl Message for TableRow {
                     let s = n.to_string();
                     ::prost::encoding::string::encode(tag, &s, buf);
                 }
+                Cell::Date(t) => {
+                    let s = t.format("%Y-%m-%d").to_string();
+                    ::prost::encoding::string::encode(tag, &s, buf);
+                }
                 Cell::TimeStamp(t) => {
                     let s = t.format("%Y-%m-%d %H:%M:%S%.f").to_string();
                     ::prost::encoding::string::encode(tag, &s, buf);
@@ -592,6 +598,10 @@ impl Message for TableRow {
                     let s = n.to_string();
                     ::prost::encoding::string::encoded_len(tag, &s)
                 }
+                Cell::Date(t) => {
+                    let s = t.format("%Y-%m-%d").to_string();
+                    ::prost::encoding::string::encoded_len(tag, &s)
+                }
                 Cell::TimeStamp(t) => {
                     let s = t.format("%Y-%m-%d %H:%M:%S%.f").to_string();
                     ::prost::encoding::string::encoded_len(tag, &s)
@@ -619,6 +629,7 @@ impl Message for TableRow {
                 Cell::F32(i) => *i = 0.,
                 Cell::F64(i) => *i = 0.,
                 Cell::Numeric(n) => *n = PgNumeric::default(),
+                Cell::Date(t) => *t = NaiveDate::default(),
                 Cell::TimeStamp(t) => *t = NaiveDateTime::default(),
                 Cell::TimeStampTz(t) => *t = DateTime::<Utc>::default(),
                 Cell::Bytes(b) => b.clear(),

pg_replicate/src/clients/duckdb.rs

@@ -101,10 +101,15 @@ impl DuckDbClient {
     fn postgres_typ_to_duckdb_typ(typ: &Type) -> &'static str {
         match typ {
             &Type::INT2 | &Type::INT4 | &Type::INT8 => "integer",
+            &Type::NUMERIC => "numeric",
+            &Type::FLOAT4 => "float",
+            &Type::FLOAT8 => "double",
             &Type::BOOL => "bool",
             &Type::BYTEA => "bytea",
-            &Type::VARCHAR | &Type::BPCHAR => "text",
+            &Type::CHAR | &Type::BPCHAR | &Type::VARCHAR | &Type::NAME | &Type::TEXT => "text",
             &Type::TIMESTAMP => "timestamp",
+            &Type::TIMESTAMPTZ => "timestamptz",
+            &Type::DATE => "date",
             typ => panic!("duckdb doesn't yet support type {typ}"),
         }
     }
@@ -365,6 +370,10 @@ impl ToSql for Cell {
                 let s = n.to_string();
                 Ok(ToSqlOutput::Owned(Value::Text(s)))
             }
+            Cell::Date(t) => {
+                let s = t.format("%Y-%m-%d").to_string();
+                Ok(ToSqlOutput::Owned(Value::Text(s)))
+            }
             Cell::TimeStamp(t) => {
                 let s = t.format("%Y-%m-%d %H:%M:%S%.f").to_string();
                 Ok(ToSqlOutput::Owned(Value::Text(s)))

pg_replicate/src/conversions/cdc_event.rs

@@ -5,7 +5,7 @@ use std::{
 };
 
 use bigdecimal::{BigDecimal, ParseBigDecimalError};
-use chrono::{DateTime, FixedOffset, NaiveDateTime};
+use chrono::{DateTime, FixedOffset, NaiveDate, NaiveDateTime};
 use postgres_protocol::message::backend::{
     BeginBody, CommitBody, DeleteBody, InsertBody, LogicalReplicationMessage, RelationBody,
     ReplicationMessage, TupleData, UpdateBody,
@@ -140,6 +140,11 @@ impl CdcEventConverter {
                 let val = from_bytea_hex(val)?;
                 Ok(Cell::Bytes(val))
             }
+            Type::DATE => {
+                let val = from_utf8(bytes)?;
+                let val = NaiveDate::parse_from_str(val, "%Y-%m-%d")?;
+                Ok(Cell::Date(val))
+            }
             Type::TIMESTAMP => {
                 let val = from_utf8(bytes)?;
                 let val = NaiveDateTime::parse_from_str(val, "%Y-%m-%d %H:%M:%S%.f")?;

pg_replicate/src/conversions/mod.rs

@@ -1,4 +1,4 @@
-use chrono::{DateTime, NaiveDateTime, Utc};
+use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc};
 use numeric::PgNumeric;
 
 pub mod cdc_event;
@@ -17,6 +17,7 @@ pub enum Cell {
     F32(f32),
     F64(f64),
     Numeric(PgNumeric),
+    Date(NaiveDate),
     TimeStamp(NaiveDateTime),
     TimeStampTz(DateTime<Utc>),
     Bytes(Vec<u8>),

pg_replicate/src/conversions/table_row.rs

@@ -1,4 +1,4 @@
-use chrono::{DateTime, FixedOffset, NaiveDateTime};
+use chrono::{DateTime, FixedOffset, NaiveDate, NaiveDateTime};
 #[cfg(feature = "unknown_types_to_bytes")]
 use postgres_protocol::types;
 use thiserror::Error;
@@ -120,6 +120,9 @@ impl TableRowConverter {
             Type::BYTEA => Self::get_from_row(row, i, column_schema.nullable, |val: Vec<u8>| {
                 Cell::Bytes(val)
             }),
+            Type::DATE => Self::get_from_row(row, i, column_schema.nullable, |val: NaiveDate| {
+                Cell::Date(val)
+            }),
             Type::TIMESTAMPTZ => Self::get_from_row(
                 row,
                 i,