_bool (postgres): update pandas_columns, transport, postgres

Implement Vec<bool> and Option<Vec<bool>> for Postgres{CSV,Simple}SourceParser.
Since the values of true/false are stored like "t" and "f", they can't
be produced by using `v.parse()` like the other types; instead, we write
out the implementations with explicit match statements.

Author: Amar1729

connectorx-python/src/pandas/pandas_columns/array.rs

@@ -94,6 +94,33 @@ where
     }
 }
 
+impl PandasColumn<Vec<bool>> for ArrayColumn<bool> {
+    #[throws(ConnectorXPythonError)]
+    fn write(&mut self, val: Vec<bool>, row: usize) {
+        self.lengths.push(val.len());
+        self.buffer.extend_from_slice(&val[..]);
+        self.row_idx.push(row);
+        self.try_flush()?;
+    }
+}
+
+impl PandasColumn<Option<Vec<bool>>> for ArrayColumn<bool> {
+    #[throws(ConnectorXPythonError)]
+    fn write(&mut self, val: Option<Vec<bool>>, row: usize) {
+        match val {
+            Some(v) => {
+                self.lengths.push(v.len());
+                self.buffer.extend_from_slice(&v[..]);
+                self.row_idx.push(row);
+                self.try_flush()?;
+            }
+            None => {
+                self.lengths.push(usize::MAX);
+                self.row_idx.push(row);
+            }
+        }
+    }
+}
 impl PandasColumn<Vec<f64>> for ArrayColumn<f64> {
     #[throws(ConnectorXPythonError)]
     fn write(&mut self, val: Vec<f64>, row: usize) {
@@ -150,6 +177,14 @@ impl PandasColumn<Option<Vec<i64>>> for ArrayColumn<i64> {
     }
 }
 
+impl HasPandasColumn for Vec<bool> {
+    type PandasColumn<'a> = ArrayColumn<bool>;
+}
+
+impl HasPandasColumn for Option<Vec<bool>> {
+    type PandasColumn<'a> = ArrayColumn<bool>;
+}
+
 impl HasPandasColumn for Vec<f64> {
     type PandasColumn<'a> = ArrayColumn<f64>;
 }

connectorx-python/src/pandas/transports/postgres.rs

@@ -33,6 +33,7 @@ macro_rules! impl_postgres_transport {
                 { Int2[i16]                                     => I64[i64]                 | conversion auto }
                 { Int4[i32]                                     => I64[i64]                 | conversion auto }
                 { Int8[i64]                                     => I64[i64]                 | conversion auto }
+                { BoolArray[Vec<bool>]                          => BoolArray[Vec<bool>]     | conversion auto_vec }
                 { Int2Array[Vec<i16>]                           => I64Array[Vec<i64>]       | conversion auto_vec }
                 { Int4Array[Vec<i32>]                           => I64Array[Vec<i64>]       | conversion auto_vec }
                 { Int8Array[Vec<i64>]                           => I64Array[Vec<i64>]       | conversion auto }

connectorx/src/sources/postgres/mod.rs

@@ -472,6 +472,7 @@ impl_produce!(
     Vec<f64>,
     Vec<Decimal>,
     bool,
+    Vec<bool>,
     &'r str,
     Vec<u8>,
     NaiveTime,
@@ -706,6 +707,57 @@ impl<'r, 'a> Produce<'r, Option<bool>> for PostgresCSVSourceParser<'a> {
     }
 }
 
+impl<'r, 'a> Produce<'r, Vec<bool>> for PostgresCSVSourceParser<'a> {
+    type Error = PostgresSourceError;
+
+    #[throws(PostgresSourceError)]
+    fn produce(&mut self) -> Vec<bool> {
+        let (ridx, cidx) = self.next_loc()?;
+        let s = &self.rowbuf[ridx][cidx][..];
+        match s {
+            "{}" => vec![],
+            _ if s.len() < 3 => throw!(ConnectorXError::cannot_produce::<bool>(Some(s.into()))),
+            s => s[1..s.len() - 1]
+                .split(",")
+                .map(|v|
+                    match v {
+                        "t" => Ok(true),
+                        "f" => Ok(false),
+                        _ => throw!(ConnectorXError::cannot_produce::<bool>(Some(s.into()))),
+                    }
+                )
+                .collect::<Result<Vec<bool>, ConnectorXError>>()?,
+        }
+    }
+}
+
+impl<'r, 'a> Produce<'r, Option<Vec<bool>>> for PostgresCSVSourceParser<'a> {
+    type Error = PostgresSourceError;
+
+    #[throws(PostgresSourceError)]
+    fn produce(&mut self) -> Option<Vec<bool>> {
+        let (ridx, cidx) = self.next_loc()?;
+        let s = &self.rowbuf[ridx][cidx][..];
+        match s {
+            "" => None,
+            "{}" => Some(vec![]),
+            _ if s.len() < 3 => throw!(ConnectorXError::cannot_produce::<bool>(Some(s.into()))),
+            s => Some(
+                s[1..s.len() - 1]
+                    .split(",")
+                    .map(|v|
+                        match v {
+                            "t" => Ok(true),
+                            "f" => Ok(false),
+                            _ => throw!(ConnectorXError::cannot_produce::<bool>(Some(s.into()))),
+                        }
+                    )
+                    .collect::<Result<Vec<bool>, ConnectorXError>>()?,
+            ),
+        }
+    }
+}
+
 impl<'r, 'a> Produce<'r, DateTime<Utc>> for PostgresCSVSourceParser<'a> {
     type Error = PostgresSourceError;
 
@@ -1007,6 +1059,7 @@ impl_produce!(
     Vec<f64>,
     Vec<Decimal>,
     bool,
+    Vec<bool>,
     &'r str,
     Vec<u8>,
     NaiveTime,
@@ -1348,6 +1401,74 @@ macro_rules! impl_simple_vec_produce {
 }
 impl_simple_vec_produce!(i16, i32, i64, f32, f64, Decimal, String,);
 
+impl<'r> Produce<'r, Vec<bool>> for PostgresSimpleSourceParser {
+    type Error = PostgresSourceError;
+
+    #[throws(PostgresSourceError)]
+    fn produce(&'r mut self) -> Vec<bool> {
+        let (ridx, cidx) = self.next_loc()?;
+        let val = match &self.rows[ridx] {
+            SimpleQueryMessage::Row(row) => match row.try_get(cidx)? {
+                Some(s) => match s{
+                    "" => throw!(anyhow!("Cannot parse NULL in non-NULL column.")),
+                    "{}" => vec![],
+                    _ => rem_first_and_last(s)
+                            .split(",")
+                            .map(|token| match token {
+                                "t" => Ok(true),
+                                "f" => Ok(false),
+                                _ => throw!(ConnectorXError::cannot_produce::<Vec<bool>>(Some(s.into())))
+                            })
+                            .collect::<Result<Vec<bool>, ConnectorXError>>()?,
+                },
+                None => throw!(anyhow!("Cannot parse NULL in non-NULL column.")),
+            },
+            SimpleQueryMessage::CommandComplete(c) => {
+                panic!("get command: {}", c);
+            }
+            _ => {
+                panic!("what?");
+            }
+        };
+        val
+    }
+}
+
+impl<'r> Produce<'r, Option<Vec<bool>>> for PostgresSimpleSourceParser {
+    type Error = PostgresSourceError;
+
+    #[throws(PostgresSourceError)]
+    fn produce(&'r mut self) -> Option<Vec<bool>> {
+        let (ridx, cidx) = self.next_loc()?;
+        let val = match &self.rows[ridx] {
+            SimpleQueryMessage::Row(row) => match row.try_get(cidx)? {
+                Some(s) => match s{
+                    "" => None,
+                    "{}" => Some(vec![]),
+                    _ => Some(
+                        rem_first_and_last(s)
+                            .split(",")
+                            .map(|token| match token {
+                                "t" => Ok(true),
+                                "f" => Ok(false),
+                                _ => throw!(ConnectorXError::cannot_produce::<Vec<bool>>(Some(s.into())))
+                            })
+                            .collect::<Result<Vec<bool>, ConnectorXError>>()?
+                    ),
+                },
+                None => None,
+            },
+            SimpleQueryMessage::CommandComplete(c) => {
+                panic!("get command: {}", c);
+            }
+            _ => {
+                panic!("what?");
+            }
+        };
+        val
+    }
+}
+
 impl<'r> Produce<'r, NaiveDate> for PostgresSimpleSourceParser {
     type Error = PostgresSourceError;