Merge branch 'master' into master

Author: iPeluwa

Cargo.lock

@@ -35,12 +35,15 @@ pub fn into_pg_type(arrow_type: &DataType) -> PgWireResult<Type> {
         DataType::Time32(_) | DataType::Time64(_) => Type::TIME,
         DataType::Date32 | DataType::Date64 => Type::DATE,
         DataType::Interval(_) => Type::INTERVAL,
-        DataType::Binary | DataType::FixedSizeBinary(_) | DataType::LargeBinary => Type::BYTEA,
+        DataType::Binary
+        | DataType::FixedSizeBinary(_)
+        | DataType::LargeBinary
+        | DataType::BinaryView => Type::BYTEA,
         DataType::Float16 | DataType::Float32 => Type::FLOAT4,
         DataType::Float64 => Type::FLOAT8,
         DataType::Decimal128(_, _) => Type::NUMERIC,
         DataType::Utf8 => Type::VARCHAR,
-        DataType::LargeUtf8 => Type::TEXT,
+        DataType::LargeUtf8 | DataType::Utf8View => Type::TEXT,
         DataType::List(field) | DataType::FixedSizeList(field, _) | DataType::LargeList(field) => {
             match field.data_type() {
                 DataType::Boolean => Type::BOOL_ARRAY,
@@ -58,11 +61,14 @@ pub fn into_pg_type(arrow_type: &DataType) -> PgWireResult<Type> {
                 DataType::Time32(_) | DataType::Time64(_) => Type::TIME_ARRAY,
                 DataType::Date32 | DataType::Date64 => Type::DATE_ARRAY,
                 DataType::Interval(_) => Type::INTERVAL_ARRAY,
-                DataType::FixedSizeBinary(_) | DataType::Binary => Type::BYTEA_ARRAY,
+                DataType::FixedSizeBinary(_)
+                | DataType::Binary
+                | DataType::LargeBinary
+                | DataType::BinaryView => Type::BYTEA_ARRAY,
                 DataType::Float16 | DataType::Float32 => Type::FLOAT4_ARRAY,
                 DataType::Float64 => Type::FLOAT8_ARRAY,
                 DataType::Utf8 => Type::VARCHAR_ARRAY,
-                DataType::LargeUtf8 => Type::TEXT_ARRAY,
+                DataType::LargeUtf8 | DataType::Utf8View => Type::TEXT_ARRAY,
                 struct_type @ DataType::Struct(_) => Type::new(
                     Type::RECORD_ARRAY.name().into(),
                     Type::RECORD_ARRAY.oid(),
@@ -78,7 +84,6 @@ pub fn into_pg_type(arrow_type: &DataType) -> PgWireResult<Type> {
                 }
             }
         }
-        DataType::Utf8View => Type::TEXT,
         DataType::Dictionary(_, value_type) => into_pg_type(value_type)?,
         DataType::Struct(fields) => {
             let name: String = fields

arrow-pg/src/datatypes.rs

@@ -142,6 +142,15 @@ fn get_utf8_view_value(arr: &Arc<dyn Array>, idx: usize) -> Option<&str> {
     })
 }
 
+fn get_binary_view_value(arr: &Arc<dyn Array>, idx: usize) -> Option<&[u8]> {
+    (!arr.is_null(idx)).then(|| {
+        arr.as_any()
+            .downcast_ref::<BinaryViewArray>()
+            .unwrap()
+            .value(idx)
+    })
+}
+
 fn get_utf8_value(arr: &Arc<dyn Array>, idx: usize) -> Option<&str> {
     (!arr.is_null(idx)).then(|| {
         arr.as_any()
@@ -160,12 +169,15 @@ fn get_large_utf8_value(arr: &Arc<dyn Array>, idx: usize) -> Option<&str> {
     })
 }
 
-fn get_binary_value(arr: &Arc<dyn Array>, idx: usize) -> Option<&[u8]> {
+fn get_binary_value(arr: &Arc<dyn Array>, idx: usize) -> Option<String> {
     (!arr.is_null(idx)).then(|| {
-        arr.as_any()
-            .downcast_ref::<BinaryArray>()
-            .unwrap()
-            .value(idx)
+        String::from_utf8_lossy(
+            arr.as_any()
+                .downcast_ref::<BinaryArray>()
+                .unwrap()
+                .value(idx),
+        )
+        .to_string()
     })
 }
 
@@ -336,6 +348,11 @@ pub fn encode_value<T: Encoder>(
             type_,
             format,
         )?,
+        DataType::BinaryView => encoder.encode_field_with_type_and_format(
+            &get_binary_view_value(arr, idx),
+            type_,
+            format,
+        )?,
         DataType::LargeUtf8 => encoder.encode_field_with_type_and_format(
             &get_large_utf8_value(arr, idx),
             type_,
@@ -517,7 +534,8 @@ pub fn encode_value<T: Encoder>(
                 .or_else(|| get_dict_values!(UInt64Type))
                 .ok_or_else(|| {
                     ToSqlError::from(format!(
-                        "Unsupported dictionary key type for value type {value_type}"
+                      "Unsupported dictionary key type for value type {value_type}"
+
                     ))
                 })?;
 

arrow-pg/src/encoder.rs

@@ -9,6 +9,7 @@ use datafusion::prelude::SessionContext;
 
 pub mod auth;
 use getset::{Getters, Setters, WithSetters};
+use log::{info, warn};
 use pgwire::tokio::process_socket;
 use rustls_pemfile::{certs, pkcs8_private_keys};
 use rustls_pki_types::{CertificateDer, PrivateKeyDer};
@@ -101,7 +102,6 @@ pub async fn serve(
     // Bind to the specified host and port
     let server_addr = format!("{}:{}", opts.host, opts.port);
     let listener = TcpListener::bind(&server_addr).await?;
-
     if tls_acceptor.is_some() {
         println!("Listening on {server_addr} with TLS encryption");
     } else {
@@ -119,6 +119,7 @@ pub async fn serve(
                 tokio::spawn(async move {
                     if let Err(e) = process_socket(socket, tls_acceptor_ref, factory_ref).await {
                         eprintln!("Error processing socket: {e}");
+
                     }
                 });
             }

datafusion-postgres/src/lib.rs

@@ -1830,7 +1830,7 @@ pub fn setup_pg_catalog(
     session_context
         .catalog(catalog_name)
         .ok_or_else(|| {
-            DataFusionError::Configuration(format!(
+            DataFusionError::Configuration(format!(              
                 "Catalog not found when registering pg_catalog: {catalog_name}"
             ))
         })?