feat: add some types from pg_catalog (#216)

* feat: add some types from pg_catalog

* fix: pin arrow-odbc version

Author: xhwhis

Cargo.toml

@@ -12,7 +12,7 @@ arrow-flight = { version = "53", optional = true, features = [
   "flight-sql-experimental",
   "tls",
 ] }
-arrow-odbc = { version = "14.0", optional = true }
+arrow-odbc = { version = "=14.0.1", optional = true }
 async-stream = { version = "0.3.6", optional = true }
 async-trait = "0.1"
 bb8 = { version = "0.8", optional = true }

src/sql/arrow_sql_gen/arrow.rs

@@ -86,6 +86,7 @@ pub fn map_data_type_to_array_builder(data_type: &DataType) -> Box<dyn ArrayBuil
                 DataType::Int16 => Box::new(ListBuilder::new(Int16Builder::new())),
                 DataType::Int32 => Box::new(ListBuilder::new(Int32Builder::new())),
                 DataType::Int64 => Box::new(ListBuilder::new(Int64Builder::new())),
+                DataType::UInt32 => Box::new(ListBuilder::new(UInt32Builder::new())),
                 DataType::Float32 => Box::new(ListBuilder::new(Float32Builder::new())),
                 DataType::Float64 => Box::new(ListBuilder::new(Float64Builder::new())),
                 DataType::Utf8 => Box::new(ListBuilder::new(StringBuilder::new())),
@@ -111,6 +112,10 @@ pub fn map_data_type_to_array_builder(data_type: &DataType) -> Box<dyn ArrayBuil
                 Int64Builder::new(),
                 size.to_owned(),
             )),
+            DataType::UInt32 => Box::new(FixedSizeListBuilder::new(
+                UInt32Builder::new(),
+                size.to_owned(),
+            )),
             DataType::Float32 => Box::new(FixedSizeListBuilder::new(
                 Float32Builder::new(),
                 size.to_owned(),

src/sql/arrow_sql_gen/postgres.rs

@@ -258,12 +258,36 @@ pub fn rows_to_arrow(rows: &[Row], projected_schema: &Option<SchemaRef>) -> Resu
                 Type::OID => {
                     handle_primitive_type!(builder, Type::OID, UInt32Builder, u32, row, i);
                 }
+                Type::XID => {
+                    let Some(builder) = builder else {
+                        return NoBuilderForIndexSnafu { index: i }.fail();
+                    };
+                    let Some(builder) = builder.as_any_mut().downcast_mut::<UInt32Builder>() else {
+                        return FailedToDowncastBuilderSnafu {
+                            postgres_type: format!("{postgres_type}"),
+                        }
+                        .fail();
+                    };
+                    let v = row
+                        .try_get::<usize, Option<XidFromSql>>(i)
+                        .with_context(|_| FailedToGetRowValueSnafu { pg_type: Type::XID })?;
+
+                    match v {
+                        Some(v) => {
+                            builder.append_value(v.xid);
+                        }
+                        None => builder.append_null(),
+                    }
+                }
                 Type::FLOAT4 => {
                     handle_primitive_type!(builder, Type::FLOAT4, Float32Builder, f32, row, i);
                 }
                 Type::FLOAT8 => {
                     handle_primitive_type!(builder, Type::FLOAT8, Float64Builder, f64, row, i);
                 }
+                Type::CHAR => {
+                    handle_primitive_type!(builder, Type::CHAR, Int8Builder, i8, row, i);
+                }
                 Type::TEXT => {
                     handle_primitive_type!(builder, Type::TEXT, StringBuilder, &str, row, i);
                 }
@@ -645,6 +669,14 @@ pub fn rows_to_arrow(rows: &[Row], projected_schema: &Option<SchemaRef>) -> Resu
                     ListBuilder<Int64Builder>,
                     i64
                 ),
+                Type::OID_ARRAY => handle_primitive_array_type!(
+                    Type::OID_ARRAY,
+                    builder,
+                    row,
+                    i,
+                    ListBuilder<UInt32Builder>,
+                    u32
+                ),
                 Type::FLOAT4_ARRAY => handle_primitive_array_type!(
                     Type::FLOAT4_ARRAY,
                     builder,
@@ -841,9 +873,10 @@ fn map_column_type_to_data_type(column_type: &Type) -> Option<DataType> {
         Type::INT2 => Some(DataType::Int16),
         Type::INT4 => Some(DataType::Int32),
         Type::INT8 | Type::MONEY => Some(DataType::Int64),
-        Type::OID => Some(DataType::UInt32),
+        Type::OID | Type::XID => Some(DataType::UInt32),
         Type::FLOAT4 => Some(DataType::Float32),
         Type::FLOAT8 => Some(DataType::Float64),
+        Type::CHAR => Some(DataType::Int8),
         Type::TEXT | Type::VARCHAR | Type::BPCHAR | Type::UUID | Type::NAME => Some(DataType::Utf8),
         Type::BYTEA => Some(DataType::Binary),
         Type::BOOL => Some(DataType::Boolean),
@@ -859,6 +892,7 @@ fn map_column_type_to_data_type(column_type: &Type) -> Option<DataType> {
             Arc::new(Field::new("item", DataType::Float64, true)),
             2,
         )),
+        Type::PG_NODE_TREE => Some(DataType::Utf8),
         Type::INT2_ARRAY => Some(DataType::List(Arc::new(Field::new(
             "item",
             DataType::Int16,
@@ -874,6 +908,11 @@ fn map_column_type_to_data_type(column_type: &Type) -> Option<DataType> {
             DataType::Int64,
             true,
         )))),
+        Type::OID_ARRAY => Some(DataType::List(Arc::new(Field::new(
+            "item",
+            DataType::UInt32,
+            true,
+        )))),
         Type::FLOAT4_ARRAY => Some(DataType::List(Arc::new(Field::new(
             "item",
             DataType::Float32,
@@ -1126,6 +1165,25 @@ impl<'a> FromSql<'a> for GeometryFromSql<'a> {
     }
 }
 
+struct XidFromSql {
+    xid: u32,
+}
+
+impl<'a> FromSql<'a> for XidFromSql {
+    fn from_sql(
+        _ty: &Type,
+        raw: &'a [u8],
+    ) -> Result<Self, Box<dyn std::error::Error + Sync + Send>> {
+        let mut cursor = std::io::Cursor::new(raw);
+        let xid = cursor.read_u32::<BigEndian>()?;
+        Ok(XidFromSql { xid })
+    }
+
+    fn accepts(ty: &Type) -> bool {
+        matches!(*ty, Type::XID)
+    }
+}
+
 fn get_decimal_column_precision_and_scale(
     column_name: &str,
     projected_schema: &SchemaRef,

src/sql/arrow_sql_gen/postgres/schema.rs

@@ -14,13 +14,14 @@ pub(crate) fn pg_data_type_to_arrow_type(
         "smallint" => Ok(DataType::Int16),
         "integer" | "int" | "int4" => Ok(DataType::Int32),
         "bigint" | "int8" | "money" => Ok(DataType::Int64),
-        "oid" => Ok(DataType::UInt32),
+        "oid" | "xid" | "regproc" => Ok(DataType::UInt32),
         "numeric" | "decimal" => {
             let (precision, scale) = parse_numeric_type(pg_type)?;
             Ok(DataType::Decimal128(precision, scale))
         }
         "real" | "float4" => Ok(DataType::Float32),
         "double precision" | "float8" => Ok(DataType::Float64),
+        "\"char\"" => Ok(DataType::Int8),
         "character" | "char" | "character varying" | "varchar" | "text" | "bpchar" | "uuid"
         | "name" => Ok(DataType::Utf8),
         "bytea" => Ok(DataType::Binary),
@@ -48,7 +49,13 @@ pub(crate) fn pg_data_type_to_arrow_type(
         "bit" | "bit varying" => Ok(DataType::Binary),
         "tsvector" | "tsquery" => Ok(DataType::LargeUtf8),
         "xml" | "json" | "jsonb" => Ok(DataType::LargeUtf8),
+        "aclitem" | "pg_node_tree" => Ok(DataType::Utf8),
         "array" => parse_array_type(type_details),
+        "anyarray" => Ok(DataType::List(Arc::new(Field::new(
+            "item",
+            DataType::Binary,
+            true,
+        )))),
         "int4range" => Ok(DataType::Struct(Fields::from(vec![
             Field::new("lower", DataType::Int32, true),
             Field::new("upper", DataType::Int32, true),
@@ -213,6 +220,11 @@ mod tests {
             pg_data_type_to_arrow_type("boolean", None).expect("Failed to convert boolean"),
             DataType::Boolean
         );
+        assert_eq!(
+            pg_data_type_to_arrow_type("\"char\"", None)
+                .expect("Failed to convert single character"),
+            DataType::Int8
+        );
 
         // Test string types
         assert_eq!(