servers/postgres/
types.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15mod error;
16
17use std::collections::HashMap;
18use std::pin::Pin;
19use std::sync::{Arc, LazyLock};
20use std::task::{Context, Poll};
21
22use arrow::array::{Array, ArrayRef, AsArray};
23use arrow_pg::encoder::{Encoder, encode_value};
24use arrow_pg::list_encoder::encode_list;
25use arrow_schema::{DataType, TimeUnit};
26use bytes::BufMut;
27use chrono::{DateTime, FixedOffset, NaiveDate, NaiveDateTime};
28use common_recordbatch::error::Result as RecordBatchResult;
29use common_recordbatch::{RecordBatch, map_dictionary_to_values_data_type};
30use common_time::{IntervalDayTime, IntervalMonthDayNano, IntervalYearMonth};
31use datafusion_common::ScalarValue;
32use datafusion_expr::LogicalPlan;
33use datafusion_pg_catalog::pg_catalog::PgCatalogStaticTables;
34use datafusion_pg_catalog::pg_catalog::oid_field::{self, OID_ALIAS_KEY};
35use datatypes::arrow::datatypes::DataType as ArrowDataType;
36use datatypes::json::JsonSettings;
37use datatypes::prelude::{ConcreteDataType, DataType as _, Value};
38use datatypes::schema::{Schema, SchemaRef};
39use datatypes::types::{Decimal128Type, IntervalType, TimestampType, jsonb_to_string};
40use datatypes::value::try_value_from_array;
41use futures::Stream;
42use pg_interval::Interval as PgInterval;
43use pgwire::api::Type;
44use pgwire::api::portal::{Format, Portal};
45use pgwire::api::results::FieldInfo;
46use pgwire::error::{PgWireError, PgWireResult};
47use pgwire::types::ToSqlText;
48use pgwire::types::format::FormatOptions as PgFormatOptions;
49use postgres_types::{IsNull, ToSql};
50use query::planner::DfLogicalPlanner;
51use rust_decimal::Decimal;
52use rust_decimal::prelude::ToPrimitive;
53use session::context::QueryContextRef;
54use snafu::ResultExt;
55
56pub use self::error::{PgErrorCode, PgErrorSeverity};
57use crate::error::{self as server_error, InferParameterTypesSnafu, Result};
58use crate::postgres::handler::PgSqlPlan;
59use crate::postgres::utils::convert_err;
60
61pub(super) fn schema_to_pg(
62    origin: &Schema,
63    field_formats: &Format,
64    format_options: Option<Arc<PgFormatOptions>>,
65) -> Result<Vec<FieldInfo>> {
66    origin
67        .column_schemas()
68        .iter()
69        .enumerate()
70        .map(|(idx, col)| {
71            let pg_type = match pg_oid_alias_type(col) {
72                Some(pg_type) => pg_type,
73                None => type_gt_to_pg(&col.data_type)?,
74            };
75            let mut field_info = FieldInfo::new(
76                col.name.clone(),
77                None,
78                None,
79                pg_type,
80                field_formats.format_for(idx),
81            );
82            if let Some(format_options) = &format_options {
83                field_info = field_info.with_format_options(format_options.clone());
84            }
85            Ok(field_info)
86        })
87        .collect::<Result<Vec<FieldInfo>>>()
88}
89
90/// Maps `datafusion-pg-catalog` OID-alias metadata to PostgreSQL wire types.
91///
92/// The catalog exposes static catalog aliases as `Utf8` and dynamic aliases
93/// as `Int32`, so this must run before the ordinary `INT4`/`VARCHAR` fallbacks.
94/// The catalog crate only exposes named constants for the aliases it uses in
95/// dynamic catalog tables; keep the remaining aliases here for compatibility
96/// with its public `OID_ALIAS_TYPE_NAMES` contract. See
97/// https://github.com/datafusion-contrib/datafusion-postgres/issues/384.
98fn pg_oid_alias_type(column: &datatypes::schema::ColumnSchema) -> Option<Type> {
99    if !matches!(
100        &column.data_type,
101        ConcreteDataType::Int32(_) | ConcreteDataType::String(_)
102    ) {
103        return None;
104    }
105
106    pg_oid_alias_type_name(column.metadata().get(OID_ALIAS_KEY)?)
107}
108
109fn pg_oid_alias_type_name(alias: &str) -> Option<Type> {
110    match alias {
111        oid_field::kind::OID => Some(Type::OID),
112        oid_field::kind::REGPROC => Some(Type::REGPROC),
113        oid_field::kind::REGCLASS => Some(Type::REGCLASS),
114        oid_field::kind::REGTYPE => Some(Type::REGTYPE),
115        oid_field::kind::REGNAMESPACE => Some(Type::REGNAMESPACE),
116        "regprocedure" => Some(Type::REGPROCEDURE),
117        "regoper" => Some(Type::REGOPER),
118        "regoperator" => Some(Type::REGOPERATOR),
119        "regcollation" => Some(Type::REGCOLLATION),
120        "regconfig" => Some(Type::REGCONFIG),
121        "regdictionary" => Some(Type::REGDICTIONARY),
122        "regrole" => Some(Type::REGROLE),
123        _ => None,
124    }
125}
126
127/// OIDs keyed by their unambiguous `pg_proc.proname` in the static catalog.
128static REGPROC_OIDS: LazyLock<std::result::Result<HashMap<String, Option<u32>>, String>> =
129    LazyLock::new(|| {
130        let tables = PgCatalogStaticTables::try_new()
131            .map_err(|e| format!("load static PostgreSQL catalog tables: {e}"))?;
132        let mut oids = HashMap::new();
133
134        for batch in tables.pg_proc.data() {
135            let names = batch
136                .column_by_name("proname")
137                .ok_or("pg_proc is missing proname")?;
138            let procedure_oids = batch
139                .column_by_name("oid")
140                .ok_or("pg_proc is missing oid")?;
141            if names.data_type() != &DataType::Utf8
142                || procedure_oids.data_type() != &DataType::Int32
143            {
144                return Err("pg_proc has unexpected proname or oid types".to_string());
145            }
146
147            let names = names.as_string::<i32>();
148            let procedure_oids = procedure_oids.as_primitive::<arrow::datatypes::Int32Type>();
149            for (name, oid) in names.iter().zip(procedure_oids.iter()) {
150                let (Some(name), Some(oid)) = (name, oid) else {
151                    continue;
152                };
153                let oid = u32::try_from(oid)
154                    .map_err(|_| format!("pg_proc contains negative oid {oid}"))?;
155
156                if oids.insert(name.to_string(), Some(oid)).is_some() {
157                    oids.insert(name.to_string(), None);
158                }
159            }
160        }
161
162        Ok(oids)
163    });
164
165#[derive(Debug)]
166struct OidAliasError(String);
167
168impl std::fmt::Display for OidAliasError {
169    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
170        f.write_str(&self.0)
171    }
172}
173
174impl std::error::Error for OidAliasError {}
175
176fn resolve_oid_alias(value: &str, alias: &str) -> std::result::Result<u32, OidAliasError> {
177    if value == "-" {
178        return Ok(0);
179    }
180
181    if let Ok(oid) = value.parse() {
182        return Ok(oid);
183    }
184
185    if alias == oid_field::kind::REGPROC {
186        let oids = REGPROC_OIDS
187            .as_ref()
188            .map_err(|e| OidAliasError(e.clone()))?;
189        return match oids.get(value) {
190            Some(Some(oid)) => Ok(*oid),
191            Some(None) => Err(OidAliasError(format!("ambiguous regproc name: {value}"))),
192            None => Err(OidAliasError(format!("unknown regproc name: {value}"))),
193        };
194    }
195
196    Err(OidAliasError(format!(
197        "named oid aliases are only supported for regproc: {value}"
198    )))
199}
200
201#[derive(Debug)]
202struct OidAliasValue<'a> {
203    text: &'a str,
204    alias: &'a str,
205}
206
207impl ToSql for OidAliasValue<'_> {
208    fn to_sql(
209        &self,
210        _ty: &Type,
211        out: &mut bytes::BytesMut,
212    ) -> std::result::Result<IsNull, Box<dyn std::error::Error + Sync + Send>> {
213        let oid = resolve_oid_alias(self.text, self.alias)?;
214        out.put_u32(oid);
215        Ok(IsNull::No)
216    }
217
218    fn accepts(ty: &Type) -> bool {
219        pg_oid_alias_type_name(ty.name()).is_some()
220    }
221
222    postgres_types::to_sql_checked!();
223}
224
225impl ToSqlText for OidAliasValue<'_> {
226    fn to_sql_text(
227        &self,
228        _ty: &Type,
229        out: &mut bytes::BytesMut,
230        _format_options: &PgFormatOptions,
231    ) -> std::result::Result<IsNull, Box<dyn std::error::Error + Sync + Send>> {
232        out.put_slice(self.text.as_bytes());
233        Ok(IsNull::No)
234    }
235}
236
237/// this function will encode greptime's structured values (`StructValue`, or a
238/// list of them) into PostgreSQL jsonb type
239///
240/// Note that greptimedb has different types of StructValue for storing json data,
241/// based on policy defined in `JsonSettings`. But here the `StructValue`
242/// should be fully structured.
243///
244/// there are alternatives like records, arrays, etc. but there are also limitations:
245/// records: there is no support for include keys
246/// arrays: element in array must be the same type
247fn encode_struct<S: Encoder>(
248    _query_ctx: &QueryContextRef,
249    value: Value,
250    builder: &mut S,
251    pg_field: &FieldInfo,
252) -> PgWireResult<()> {
253    let encoding_setting = JsonSettings::default();
254    let json_value = encoding_setting
255        .decode(value)
256        .map_err(|e| PgWireError::ApiError(Box::new(e)))?;
257
258    builder.encode_field(&json_value, pg_field)
259}
260
261/// Encodes one row of a structured column (`Struct`, or `List` of `Struct`) as
262/// a PostgreSQL `json` value.
263///
264/// Returns an error for arrow field types greptimedb cannot represent
265/// (e.g. `Decimal256`) instead of panicking, so the client receives a proper
266/// error response.
267fn encode_structured_json<S: Encoder>(
268    query_ctx: &QueryContextRef,
269    column: &ArrayRef,
270    i: usize,
271    encoder: &mut S,
272    pg_field: &FieldInfo,
273) -> PgWireResult<()> {
274    let value = try_value_from_array(column.as_ref(), i).map_err(convert_err)?;
275    encode_struct(query_ctx, value, encoder, pg_field)
276}
277
278pub(crate) struct RecordBatchRowStream<S, B>
279where
280    S: Encoder,
281    B: Stream<Item = RecordBatchResult<RecordBatch>>,
282{
283    query_ctx: QueryContextRef,
284    pg_schema: Arc<Vec<FieldInfo>>,
285    schema: SchemaRef,
286    record_batches: Pin<Box<B>>,
287    encoder: S,
288}
289
290impl<S, B> Stream for RecordBatchRowStream<S, B>
291where
292    S: Encoder + Unpin,
293    B: Stream<Item = RecordBatchResult<RecordBatch>>,
294{
295    type Item = PgWireResult<Vec<S::Item>>;
296
297    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
298        match self.record_batches.as_mut().poll_next(cx) {
299            Poll::Ready(Some(Ok(batch))) => {
300                let record_batch = batch.into_df_record_batch();
301                let num_rows = record_batch.num_rows();
302
303                if num_rows == 0 {
304                    return Poll::Ready(Some(Ok(vec![])));
305                }
306
307                let arrow_schema = record_batch.schema();
308                let query_ctx = self.query_ctx.clone();
309                let pg_schema = self.pg_schema.clone();
310                let schema = self.schema.clone();
311                let mut results = Vec::with_capacity(num_rows);
312
313                for i in 0..num_rows {
314                    if let Err(e) = Self::encode_row(
315                        &query_ctx,
316                        &pg_schema,
317                        &schema,
318                        arrow_schema.as_ref(),
319                        &mut self.encoder,
320                        &record_batch,
321                        i,
322                    ) {
323                        return Poll::Ready(Some(Err(e)));
324                    }
325                    results.push(self.encoder.take_row());
326                }
327
328                Poll::Ready(Some(Ok(results)))
329            }
330            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(convert_err(e)))),
331            Poll::Ready(None) => Poll::Ready(None),
332            Poll::Pending => Poll::Pending,
333        }
334    }
335}
336
337impl<S, B> RecordBatchRowStream<S, B>
338where
339    S: Encoder,
340    B: Stream<Item = RecordBatchResult<RecordBatch>>,
341{
342    pub(crate) fn new(
343        query_ctx: QueryContextRef,
344        pg_schema: Arc<Vec<FieldInfo>>,
345        schema: SchemaRef,
346        record_batches: B,
347        encoder: S,
348    ) -> Self {
349        Self {
350            query_ctx,
351            pg_schema,
352            schema,
353            record_batches: Box::pin(record_batches),
354            encoder,
355        }
356    }
357
358    fn encode_row(
359        query_ctx: &QueryContextRef,
360        pg_schema: &Arc<Vec<FieldInfo>>,
361        schema: &SchemaRef,
362        arrow_schema: &arrow::datatypes::Schema,
363        encoder: &mut S,
364        record_batch: &arrow::record_batch::RecordBatch,
365        i: usize,
366    ) -> PgWireResult<()> {
367        for (j, column) in record_batch.columns().iter().enumerate() {
368            let pg_field = &pg_schema[j];
369
370            if column.is_null(i) {
371                encoder.encode_field(&None::<&i8>, pg_field)?;
372                continue;
373            }
374
375            match column.data_type() {
376                // these types are greptimedb specific or custom
377                DataType::Binary | DataType::LargeBinary | DataType::BinaryView => {
378                    // jsonb
379                    if let ConcreteDataType::Json(_) = &schema.column_schemas()[j].data_type {
380                        let v = datatypes::arrow_array::binary_array_value(column, i);
381                        let s = jsonb_to_string(v).map_err(convert_err)?;
382                        encoder.encode_field(&s, pg_field)?;
383                    } else {
384                        // bytea
385                        let arrow_field = arrow_schema.field(j);
386                        encode_value(encoder, column, i, arrow_field, pg_field)?;
387                    }
388                }
389
390                DataType::List(list_field)
391                    if matches!(list_field.data_type(), DataType::Struct(_)) =>
392                {
393                    encode_structured_json(query_ctx, column, i, encoder, pg_field)?;
394                }
395                DataType::List(_) => {
396                    let array = column.as_list::<i32>();
397                    let items = array.value(i);
398
399                    encode_list(encoder, items, pg_field)?;
400                }
401                DataType::Struct(_) => {
402                    encode_structured_json(query_ctx, column, i, encoder, pg_field)?;
403                }
404                DataType::Utf8 => {
405                    let arrow_field = arrow_schema.field(j);
406                    if let Some(alias) = arrow_field
407                        .metadata()
408                        .get(OID_ALIAS_KEY)
409                        .filter(|alias| pg_oid_alias_type_name(alias).is_some())
410                    {
411                        let value = OidAliasValue {
412                            text: column.as_string::<i32>().value(i),
413                            alias,
414                        };
415                        encoder.encode_field(&value, pg_field)?;
416                    } else {
417                        encode_value(encoder, column, i, arrow_field, pg_field)?;
418                    }
419                }
420                _ => {
421                    // Encode value using arrow-pg
422                    let arrow_field = arrow_schema.field(j);
423                    encode_value(encoder, column, i, arrow_field, pg_field)?;
424                }
425            }
426        }
427        Ok(())
428    }
429}
430
431pub(super) fn type_gt_to_pg(origin: &ConcreteDataType) -> Result<Type> {
432    let logical_type = map_dictionary_to_values_data_type(origin);
433    let origin = &logical_type;
434    match origin {
435        &ConcreteDataType::Null(_) => Ok(Type::UNKNOWN),
436        &ConcreteDataType::Boolean(_) => Ok(Type::BOOL),
437        &ConcreteDataType::Int8(_) => Ok(Type::INT2),
438        &ConcreteDataType::Int16(_) | &ConcreteDataType::UInt8(_) => Ok(Type::INT2),
439        &ConcreteDataType::Int32(_) | &ConcreteDataType::UInt16(_) => Ok(Type::INT4),
440        &ConcreteDataType::Int64(_) | &ConcreteDataType::UInt32(_) => Ok(Type::INT8),
441        &ConcreteDataType::UInt64(_) => Ok(Type::NUMERIC),
442        &ConcreteDataType::Float32(_) => Ok(Type::FLOAT4),
443        &ConcreteDataType::Float64(_) => Ok(Type::FLOAT8),
444        &ConcreteDataType::Binary(_) | &ConcreteDataType::Vector(_) => Ok(Type::BYTEA),
445        &ConcreteDataType::String(_) => Ok(Type::VARCHAR),
446        &ConcreteDataType::Date(_) => Ok(Type::DATE),
447        &ConcreteDataType::Timestamp(_) => Ok(Type::TIMESTAMP),
448        &ConcreteDataType::Time(_) => Ok(Type::TIME),
449        &ConcreteDataType::Interval(_) => Ok(Type::INTERVAL),
450        &ConcreteDataType::Decimal128(_) => Ok(Type::NUMERIC),
451        &ConcreteDataType::Json(_) => Ok(Type::JSON),
452        ConcreteDataType::List(list) => match list.item_type() {
453            &ConcreteDataType::Null(_) => Ok(Type::TEXT_ARRAY),
454            &ConcreteDataType::Boolean(_) => Ok(Type::BOOL_ARRAY),
455            &ConcreteDataType::Int8(_) => Ok(Type::INT2_ARRAY),
456            &ConcreteDataType::Int16(_) | &ConcreteDataType::UInt8(_) => Ok(Type::INT2_ARRAY),
457            &ConcreteDataType::Int32(_) | &ConcreteDataType::UInt16(_) => Ok(Type::INT4_ARRAY),
458            &ConcreteDataType::Int64(_) | &ConcreteDataType::UInt32(_) => Ok(Type::INT8_ARRAY),
459            &ConcreteDataType::UInt64(_) => Ok(Type::NUMERIC_ARRAY),
460            &ConcreteDataType::Float32(_) => Ok(Type::FLOAT4_ARRAY),
461            &ConcreteDataType::Float64(_) => Ok(Type::FLOAT8_ARRAY),
462            &ConcreteDataType::Binary(_) => Ok(Type::BYTEA_ARRAY),
463            &ConcreteDataType::String(_) => Ok(Type::VARCHAR_ARRAY),
464            &ConcreteDataType::Date(_) => Ok(Type::DATE_ARRAY),
465            &ConcreteDataType::Timestamp(_) => Ok(Type::TIMESTAMP_ARRAY),
466            &ConcreteDataType::Time(_) => Ok(Type::TIME_ARRAY),
467            &ConcreteDataType::Interval(_) => Ok(Type::INTERVAL_ARRAY),
468            &ConcreteDataType::Decimal128(_) => Ok(Type::NUMERIC_ARRAY),
469            &ConcreteDataType::Json(_) => Ok(Type::JSON_ARRAY),
470            &ConcreteDataType::Duration(_) => Ok(Type::INTERVAL_ARRAY),
471            // a list of structs is encoded as a single JSON array, matching the
472            // struct-to-json mapping above
473            &ConcreteDataType::Struct(_) => Ok(Type::JSON),
474            &ConcreteDataType::Dictionary(_)
475            | &ConcreteDataType::Vector(_)
476            | &ConcreteDataType::List(_) => server_error::UnsupportedDataTypeSnafu {
477                data_type: origin,
478                reason: "not implemented",
479            }
480            .fail(),
481        },
482        &ConcreteDataType::Dictionary(_) => server_error::UnsupportedDataTypeSnafu {
483            data_type: origin,
484            reason: "not implemented",
485        }
486        .fail(),
487        &ConcreteDataType::Duration(_) => Ok(Type::INTERVAL),
488        &ConcreteDataType::Struct(_) => Ok(Type::JSON),
489    }
490}
491
492pub(super) fn invalid_parameter_error(msg: &str, detail: Option<String>) -> PgWireError {
493    let mut error_info = PgErrorCode::Ec22023.to_err_info(msg.to_string());
494    error_info.detail = detail;
495    PgWireError::UserError(Box::new(error_info))
496}
497
498fn to_timestamp_scalar_value<T>(
499    data: Option<T>,
500    unit: &TimestampType,
501    ctype: &ConcreteDataType,
502) -> PgWireResult<ScalarValue>
503where
504    T: Into<i64>,
505{
506    if let Some(n) = data {
507        Value::Timestamp(unit.create_timestamp(n.into()))
508            .try_to_scalar_value(ctype)
509            .map_err(convert_err)
510    } else {
511        Ok(ScalarValue::Null)
512    }
513}
514
515fn to_decimal_scalar_value(data: Option<Decimal>, ctype: &Decimal128Type) -> ScalarValue {
516    if let Some(data) = data {
517        let mut value = data;
518        value.rescale(ctype.scale() as u32);
519
520        ScalarValue::Decimal128(Some(value.mantissa()), ctype.precision(), ctype.scale())
521    } else {
522        ScalarValue::Decimal128(None, ctype.precision(), ctype.scale())
523    }
524}
525
526fn numeric_out_of_range_error(value: impl std::fmt::Display) -> PgWireError {
527    invalid_parameter_error(
528        "numeric_value_out_of_range",
529        Some(format!("value {} is out of range for target type", value)),
530    )
531}
532
533fn string_parameter_to_scalar_value(
534    data: Option<String>,
535    data_type: &ConcreteDataType,
536) -> Option<ScalarValue> {
537    match data_type {
538        ConcreteDataType::String(string_type) => {
539            if string_type.is_large() {
540                Some(ScalarValue::LargeUtf8(data))
541            } else {
542                Some(ScalarValue::Utf8(data))
543            }
544        }
545        ConcreteDataType::Dictionary(dictionary) => Some(ScalarValue::Dictionary(
546            Box::new(dictionary.key_type().as_arrow_type()),
547            Box::new(string_parameter_to_scalar_value(
548                data,
549                dictionary.value_type(),
550            )?),
551        )),
552        _ => None,
553    }
554}
555
556pub(super) fn parameters_to_scalar_values(
557    plan: &LogicalPlan,
558    portal: &Portal<PgSqlPlan>,
559) -> PgWireResult<Vec<ScalarValue>> {
560    let param_count = portal.parameter_len();
561    let mut results = Vec::with_capacity(param_count);
562
563    let client_param_types = &portal.statement.parameter_types;
564    let server_param_types = DfLogicalPlanner::get_inferred_parameter_types(plan)
565        .context(InferParameterTypesSnafu)
566        .map_err(convert_err)?
567        .into_iter()
568        .map(|(k, v)| (k, v.map(|v| ConcreteDataType::from_arrow_type(&v))))
569        .collect::<HashMap<_, _>>();
570
571    for idx in 0..param_count {
572        let server_type = server_param_types
573            .get(&format!("${}", idx + 1))
574            .and_then(|t| t.as_ref());
575
576        let client_type = if let Some(Some(client_given_type)) = client_param_types.get(idx) {
577            client_given_type.clone()
578        } else if let Some(server_provided_type) = &server_type {
579            type_gt_to_pg(server_provided_type).map_err(convert_err)?
580        } else {
581            return Err(invalid_parameter_error(
582                "unknown_parameter_type",
583                Some(format!(
584                    "Cannot get type for parameter {}, try to provide a type using ${}::<type>",
585                    idx, idx
586                )),
587            ));
588        };
589
590        let value = match &client_type {
591            &Type::VARCHAR | &Type::TEXT | &Type::CHAR => {
592                let data = portal.parameter::<String>(idx, &client_type)?;
593                if let Some(server_type) = &server_type {
594                    string_parameter_to_scalar_value(data, server_type).ok_or_else(|| {
595                        invalid_parameter_error(
596                            "invalid_parameter_type",
597                            Some(format!("Expected: {}, found: {}", server_type, client_type)),
598                        )
599                    })?
600                } else {
601                    ScalarValue::Utf8(data)
602                }
603            }
604            &Type::BOOL => {
605                let data = portal.parameter::<bool>(idx, &client_type)?;
606                if let Some(server_type) = &server_type {
607                    match server_type {
608                        ConcreteDataType::Boolean(_) => ScalarValue::Boolean(data),
609                        _ => {
610                            return Err(invalid_parameter_error(
611                                "invalid_parameter_type",
612                                Some(format!("Expected: {}, found: {}", server_type, client_type)),
613                            ));
614                        }
615                    }
616                } else {
617                    ScalarValue::Boolean(data)
618                }
619            }
620            &Type::INT2 => {
621                let data = portal.parameter::<i16>(idx, &client_type)?;
622                if let Some(server_type) = &server_type {
623                    match server_type {
624                        ConcreteDataType::Int8(_) => ScalarValue::Int8(
625                            data.map(|n| n.to_i8().ok_or_else(|| numeric_out_of_range_error(n)))
626                                .transpose()?,
627                        ),
628                        ConcreteDataType::Int16(_) => ScalarValue::Int16(data),
629                        ConcreteDataType::Int32(_) => ScalarValue::Int32(data.map(|n| n as i32)),
630                        ConcreteDataType::Int64(_) => ScalarValue::Int64(data.map(|n| n as i64)),
631                        ConcreteDataType::UInt8(_) => ScalarValue::UInt8(
632                            data.map(|n| n.to_u8().ok_or_else(|| numeric_out_of_range_error(n)))
633                                .transpose()?,
634                        ),
635                        ConcreteDataType::UInt16(_) => ScalarValue::UInt16(
636                            data.map(|n| n.to_u16().ok_or_else(|| numeric_out_of_range_error(n)))
637                                .transpose()?,
638                        ),
639                        ConcreteDataType::UInt32(_) => ScalarValue::UInt32(
640                            data.map(|n| n.to_u32().ok_or_else(|| numeric_out_of_range_error(n)))
641                                .transpose()?,
642                        ),
643                        ConcreteDataType::UInt64(_) => ScalarValue::UInt64(
644                            data.map(|n| n.to_u64().ok_or_else(|| numeric_out_of_range_error(n)))
645                                .transpose()?,
646                        ),
647                        ConcreteDataType::Timestamp(unit) => {
648                            to_timestamp_scalar_value(data, unit, server_type)?
649                        }
650                        _ => {
651                            return Err(invalid_parameter_error(
652                                "invalid_parameter_type",
653                                Some(format!("Expected: {}, found: {}", server_type, client_type)),
654                            ));
655                        }
656                    }
657                } else {
658                    ScalarValue::Int16(data)
659                }
660            }
661            &Type::INT4 => {
662                let data = portal.parameter::<i32>(idx, &client_type)?;
663                if let Some(server_type) = &server_type {
664                    match server_type {
665                        ConcreteDataType::Int8(_) => ScalarValue::Int8(
666                            data.map(|n| n.to_i8().ok_or_else(|| numeric_out_of_range_error(n)))
667                                .transpose()?,
668                        ),
669                        ConcreteDataType::Int16(_) => ScalarValue::Int16(
670                            data.map(|n| n.to_i16().ok_or_else(|| numeric_out_of_range_error(n)))
671                                .transpose()?,
672                        ),
673                        ConcreteDataType::Int32(_) => ScalarValue::Int32(data),
674                        ConcreteDataType::Int64(_) => ScalarValue::Int64(data.map(|n| n as i64)),
675                        ConcreteDataType::UInt8(_) => ScalarValue::UInt8(
676                            data.map(|n| n.to_u8().ok_or_else(|| numeric_out_of_range_error(n)))
677                                .transpose()?,
678                        ),
679                        ConcreteDataType::UInt16(_) => ScalarValue::UInt16(
680                            data.map(|n| n.to_u16().ok_or_else(|| numeric_out_of_range_error(n)))
681                                .transpose()?,
682                        ),
683                        ConcreteDataType::UInt32(_) => ScalarValue::UInt32(
684                            data.map(|n| n.to_u32().ok_or_else(|| numeric_out_of_range_error(n)))
685                                .transpose()?,
686                        ),
687                        ConcreteDataType::UInt64(_) => ScalarValue::UInt64(
688                            data.map(|n| n.to_u64().ok_or_else(|| numeric_out_of_range_error(n)))
689                                .transpose()?,
690                        ),
691                        ConcreteDataType::Timestamp(unit) => {
692                            to_timestamp_scalar_value(data, unit, server_type)?
693                        }
694                        _ => {
695                            return Err(invalid_parameter_error(
696                                "invalid_parameter_type",
697                                Some(format!("Expected: {}, found: {}", server_type, client_type)),
698                            ));
699                        }
700                    }
701                } else {
702                    ScalarValue::Int32(data)
703                }
704            }
705            &Type::INT8 => {
706                let data = portal.parameter::<i64>(idx, &client_type)?;
707                if let Some(server_type) = &server_type {
708                    match server_type {
709                        ConcreteDataType::Int8(_) => ScalarValue::Int8(
710                            data.map(|n| n.to_i8().ok_or_else(|| numeric_out_of_range_error(n)))
711                                .transpose()?,
712                        ),
713                        ConcreteDataType::Int16(_) => ScalarValue::Int16(
714                            data.map(|n| n.to_i16().ok_or_else(|| numeric_out_of_range_error(n)))
715                                .transpose()?,
716                        ),
717                        ConcreteDataType::Int32(_) => ScalarValue::Int32(
718                            data.map(|n| n.to_i32().ok_or_else(|| numeric_out_of_range_error(n)))
719                                .transpose()?,
720                        ),
721                        ConcreteDataType::Int64(_) => ScalarValue::Int64(data),
722                        ConcreteDataType::UInt8(_) => ScalarValue::UInt8(
723                            data.map(|n| n.to_u8().ok_or_else(|| numeric_out_of_range_error(n)))
724                                .transpose()?,
725                        ),
726                        ConcreteDataType::UInt16(_) => ScalarValue::UInt16(
727                            data.map(|n| n.to_u16().ok_or_else(|| numeric_out_of_range_error(n)))
728                                .transpose()?,
729                        ),
730                        ConcreteDataType::UInt32(_) => ScalarValue::UInt32(
731                            data.map(|n| n.to_u32().ok_or_else(|| numeric_out_of_range_error(n)))
732                                .transpose()?,
733                        ),
734                        ConcreteDataType::UInt64(_) => ScalarValue::UInt64(
735                            data.map(|n| n.to_u64().ok_or_else(|| numeric_out_of_range_error(n)))
736                                .transpose()?,
737                        ),
738                        ConcreteDataType::Timestamp(unit) => {
739                            to_timestamp_scalar_value(data, unit, server_type)?
740                        }
741                        _ => {
742                            return Err(invalid_parameter_error(
743                                "invalid_parameter_type",
744                                Some(format!("Expected: {}, found: {}", server_type, client_type)),
745                            ));
746                        }
747                    }
748                } else {
749                    ScalarValue::Int64(data)
750                }
751            }
752            &Type::NUMERIC => {
753                let data = portal.parameter::<Decimal>(idx, &client_type)?;
754                match &server_type {
755                    Some(ConcreteDataType::Decimal128(dt)) => to_decimal_scalar_value(data, dt),
756                    Some(st @ ConcreteDataType::Timestamp(unit)) => {
757                        to_timestamp_scalar_value(data.and_then(|n| n.to_i64()), unit, st)?
758                    }
759                    Some(ConcreteDataType::UInt64(_)) | None => ScalarValue::UInt64(
760                        data.map(|n| n.to_u64().ok_or_else(|| numeric_out_of_range_error(n)))
761                            .transpose()?,
762                    ),
763                    Some(st) => {
764                        return Err(invalid_parameter_error(
765                            "invalid_parameter_type",
766                            Some(format!("Expected: {}, found: {}", st, client_type)),
767                        ));
768                    }
769                }
770            }
771            &Type::FLOAT4 => {
772                let data = portal.parameter::<f32>(idx, &client_type)?;
773                if let Some(server_type) = &server_type {
774                    match server_type {
775                        ConcreteDataType::Int8(_) => ScalarValue::Int8(
776                            data.map(|n| n.to_i8().ok_or_else(|| numeric_out_of_range_error(n)))
777                                .transpose()?,
778                        ),
779                        ConcreteDataType::Int16(_) => ScalarValue::Int16(
780                            data.map(|n| n.to_i16().ok_or_else(|| numeric_out_of_range_error(n)))
781                                .transpose()?,
782                        ),
783                        ConcreteDataType::Int32(_) => ScalarValue::Int32(
784                            data.map(|n| n.to_i32().ok_or_else(|| numeric_out_of_range_error(n)))
785                                .transpose()?,
786                        ),
787                        ConcreteDataType::Int64(_) => ScalarValue::Int64(
788                            data.map(|n| n.to_i64().ok_or_else(|| numeric_out_of_range_error(n)))
789                                .transpose()?,
790                        ),
791                        ConcreteDataType::UInt8(_) => ScalarValue::UInt8(
792                            data.map(|n| n.to_u8().ok_or_else(|| numeric_out_of_range_error(n)))
793                                .transpose()?,
794                        ),
795                        ConcreteDataType::UInt16(_) => ScalarValue::UInt16(
796                            data.map(|n| n.to_u16().ok_or_else(|| numeric_out_of_range_error(n)))
797                                .transpose()?,
798                        ),
799                        ConcreteDataType::UInt32(_) => ScalarValue::UInt32(
800                            data.map(|n| n.to_u32().ok_or_else(|| numeric_out_of_range_error(n)))
801                                .transpose()?,
802                        ),
803                        ConcreteDataType::UInt64(_) => ScalarValue::UInt64(
804                            data.map(|n| n.to_u64().ok_or_else(|| numeric_out_of_range_error(n)))
805                                .transpose()?,
806                        ),
807                        ConcreteDataType::Float32(_) => ScalarValue::Float32(data),
808                        ConcreteDataType::Float64(_) => {
809                            ScalarValue::Float64(data.map(|n| n as f64))
810                        }
811                        _ => {
812                            return Err(invalid_parameter_error(
813                                "invalid_parameter_type",
814                                Some(format!("Expected: {}, found: {}", server_type, client_type)),
815                            ));
816                        }
817                    }
818                } else {
819                    ScalarValue::Float32(data)
820                }
821            }
822            &Type::FLOAT8 => {
823                let data = portal.parameter::<f64>(idx, &client_type)?;
824                if let Some(server_type) = &server_type {
825                    match server_type {
826                        ConcreteDataType::Int8(_) => ScalarValue::Int8(
827                            data.map(|n| n.to_i8().ok_or_else(|| numeric_out_of_range_error(n)))
828                                .transpose()?,
829                        ),
830                        ConcreteDataType::Int16(_) => ScalarValue::Int16(
831                            data.map(|n| n.to_i16().ok_or_else(|| numeric_out_of_range_error(n)))
832                                .transpose()?,
833                        ),
834                        ConcreteDataType::Int32(_) => ScalarValue::Int32(
835                            data.map(|n| n.to_i32().ok_or_else(|| numeric_out_of_range_error(n)))
836                                .transpose()?,
837                        ),
838                        ConcreteDataType::Int64(_) => ScalarValue::Int64(
839                            data.map(|n| n.to_i64().ok_or_else(|| numeric_out_of_range_error(n)))
840                                .transpose()?,
841                        ),
842                        ConcreteDataType::UInt8(_) => ScalarValue::UInt8(
843                            data.map(|n| n.to_u8().ok_or_else(|| numeric_out_of_range_error(n)))
844                                .transpose()?,
845                        ),
846                        ConcreteDataType::UInt16(_) => ScalarValue::UInt16(
847                            data.map(|n| n.to_u16().ok_or_else(|| numeric_out_of_range_error(n)))
848                                .transpose()?,
849                        ),
850                        ConcreteDataType::UInt32(_) => ScalarValue::UInt32(
851                            data.map(|n| n.to_u32().ok_or_else(|| numeric_out_of_range_error(n)))
852                                .transpose()?,
853                        ),
854                        ConcreteDataType::UInt64(_) => ScalarValue::UInt64(
855                            data.map(|n| n.to_u64().ok_or_else(|| numeric_out_of_range_error(n)))
856                                .transpose()?,
857                        ),
858                        ConcreteDataType::Float32(_) => ScalarValue::Float32(
859                            data.map(|n| n.to_f32().ok_or_else(|| numeric_out_of_range_error(n)))
860                                .transpose()?,
861                        ),
862                        ConcreteDataType::Float64(_) => ScalarValue::Float64(data),
863                        _ => {
864                            return Err(invalid_parameter_error(
865                                "invalid_parameter_type",
866                                Some(format!("Expected: {}, found: {}", server_type, client_type)),
867                            ));
868                        }
869                    }
870                } else {
871                    ScalarValue::Float64(data)
872                }
873            }
874            &Type::TIMESTAMP => {
875                let data = portal.parameter::<NaiveDateTime>(idx, &client_type)?;
876                if let Some(server_type) = &server_type {
877                    match server_type {
878                        ConcreteDataType::Timestamp(unit) => match *unit {
879                            TimestampType::Second(_) => ScalarValue::TimestampSecond(
880                                data.map(|ts| ts.and_utc().timestamp()),
881                                None,
882                            ),
883                            TimestampType::Millisecond(_) => ScalarValue::TimestampMillisecond(
884                                data.map(|ts| ts.and_utc().timestamp_millis()),
885                                None,
886                            ),
887                            TimestampType::Microsecond(_) => ScalarValue::TimestampMicrosecond(
888                                data.map(|ts| ts.and_utc().timestamp_micros()),
889                                None,
890                            ),
891                            TimestampType::Nanosecond(_) => ScalarValue::TimestampNanosecond(
892                                data.and_then(|ts| ts.and_utc().timestamp_nanos_opt()),
893                                None,
894                            ),
895                        },
896                        _ => {
897                            return Err(invalid_parameter_error(
898                                "invalid_parameter_type",
899                                Some(format!("Expected: {}, found: {}", server_type, client_type)),
900                            ));
901                        }
902                    }
903                } else {
904                    ScalarValue::TimestampMillisecond(
905                        data.map(|ts| ts.and_utc().timestamp_millis()),
906                        None,
907                    )
908                }
909            }
910            &Type::TIMESTAMPTZ => {
911                let data = portal.parameter::<DateTime<FixedOffset>>(idx, &client_type)?;
912                if let Some(server_type) = &server_type {
913                    match server_type {
914                        ConcreteDataType::Timestamp(unit) => match *unit {
915                            TimestampType::Second(_) => {
916                                ScalarValue::TimestampSecond(data.map(|ts| ts.timestamp()), None)
917                            }
918                            TimestampType::Millisecond(_) => ScalarValue::TimestampMillisecond(
919                                data.map(|ts| ts.timestamp_millis()),
920                                None,
921                            ),
922                            TimestampType::Microsecond(_) => ScalarValue::TimestampMicrosecond(
923                                data.map(|ts| ts.timestamp_micros()),
924                                None,
925                            ),
926                            TimestampType::Nanosecond(_) => ScalarValue::TimestampNanosecond(
927                                data.and_then(|ts| ts.timestamp_nanos_opt()),
928                                None,
929                            ),
930                        },
931                        _ => {
932                            return Err(invalid_parameter_error(
933                                "invalid_parameter_type",
934                                Some(format!("Expected: {}, found: {}", server_type, client_type)),
935                            ));
936                        }
937                    }
938                } else {
939                    ScalarValue::TimestampMillisecond(data.map(|ts| ts.timestamp_millis()), None)
940                }
941            }
942            &Type::DATE => {
943                let data = portal.parameter::<NaiveDate>(idx, &client_type)?;
944                if let Some(server_type) = &server_type {
945                    match server_type {
946                        ConcreteDataType::Date(_) => ScalarValue::Date32(
947                            data.map(|d| (d - DateTime::UNIX_EPOCH.date_naive()).num_days() as i32),
948                        ),
949                        _ => {
950                            return Err(invalid_parameter_error(
951                                "invalid_parameter_type",
952                                Some(format!("Expected: {}, found: {}", server_type, client_type)),
953                            ));
954                        }
955                    }
956                } else {
957                    ScalarValue::Date32(
958                        data.map(|d| (d - DateTime::UNIX_EPOCH.date_naive()).num_days() as i32),
959                    )
960                }
961            }
962            &Type::INTERVAL => {
963                let data = portal.parameter::<PgInterval>(idx, &client_type)?;
964                if let Some(server_type) = &server_type {
965                    match server_type {
966                        ConcreteDataType::Interval(IntervalType::YearMonth(_)) => {
967                            ScalarValue::IntervalYearMonth(
968                                data.map(|i| {
969                                    if i.days != 0 || i.microseconds != 0 {
970                                        Err(invalid_parameter_error(
971                                            "invalid_parameter_type",
972                                            Some(format!(
973                                                "Expected: {}, found: {}",
974                                                server_type, client_type
975                                            )),
976                                        ))
977                                    } else {
978                                        Ok(IntervalYearMonth::new(i.months).to_i32())
979                                    }
980                                })
981                                .transpose()?,
982                            )
983                        }
984                        ConcreteDataType::Interval(IntervalType::DayTime(_)) => {
985                            ScalarValue::IntervalDayTime(
986                                data.map(|i| {
987                                    if i.months != 0 || i.microseconds % 1000 != 0 {
988                                        Err(invalid_parameter_error(
989                                            "invalid_parameter_type",
990                                            Some(format!(
991                                                "Expected: {}, found: {}",
992                                                server_type, client_type
993                                            )),
994                                        ))
995                                    } else {
996                                        Ok(IntervalDayTime::new(
997                                            i.days,
998                                            (i.microseconds / 1000) as i32,
999                                        )
1000                                        .into())
1001                                    }
1002                                })
1003                                .transpose()?,
1004                            )
1005                        }
1006                        ConcreteDataType::Interval(IntervalType::MonthDayNano(_)) => {
1007                            ScalarValue::IntervalMonthDayNano(data.map(|i| {
1008                                IntervalMonthDayNano::new(
1009                                    i.months,
1010                                    i.days,
1011                                    i.microseconds * 1_000i64,
1012                                )
1013                                .into()
1014                            }))
1015                        }
1016                        _ => {
1017                            return Err(invalid_parameter_error(
1018                                "invalid_parameter_type",
1019                                Some(format!("Expected: {}, found: {}", server_type, client_type)),
1020                            ));
1021                        }
1022                    }
1023                } else {
1024                    ScalarValue::IntervalMonthDayNano(data.map(|i| {
1025                        IntervalMonthDayNano::new(i.months, i.days, i.microseconds * 1_000i64)
1026                            .into()
1027                    }))
1028                }
1029            }
1030            &Type::BYTEA => {
1031                let data = portal.parameter::<Vec<u8>>(idx, &client_type)?;
1032                if let Some(server_type) = &server_type {
1033                    match server_type {
1034                        ConcreteDataType::String(t) => {
1035                            let s = data.map(|d| String::from_utf8_lossy(&d).to_string());
1036                            if t.is_large() {
1037                                ScalarValue::LargeUtf8(s)
1038                            } else {
1039                                ScalarValue::Utf8(s)
1040                            }
1041                        }
1042                        ConcreteDataType::Binary(_) => ScalarValue::Binary(data),
1043                        _ => {
1044                            return Err(invalid_parameter_error(
1045                                "invalid_parameter_type",
1046                                Some(format!("Expected: {}, found: {}", server_type, client_type)),
1047                            ));
1048                        }
1049                    }
1050                } else {
1051                    ScalarValue::Binary(data)
1052                }
1053            }
1054            &Type::JSONB => {
1055                let data = portal.parameter::<serde_json::Value>(idx, &client_type)?;
1056                if let Some(server_type) = &server_type {
1057                    match server_type {
1058                        ConcreteDataType::Binary(_) => {
1059                            ScalarValue::Binary(data.map(|d| d.to_string().into_bytes()))
1060                        }
1061                        _ => {
1062                            return Err(invalid_parameter_error(
1063                                "invalid_parameter_type",
1064                                Some(format!("Expected: {}, found: {}", server_type, client_type)),
1065                            ));
1066                        }
1067                    }
1068                } else {
1069                    ScalarValue::Binary(data.map(|d| d.to_string().into_bytes()))
1070                }
1071            }
1072            &Type::INT2_ARRAY => {
1073                let data = portal.parameter::<Vec<Option<i16>>>(idx, &client_type)?;
1074                if let Some(data) = data {
1075                    let values = data.into_iter().map(|i| i.into()).collect::<Vec<_>>();
1076                    ScalarValue::List(ScalarValue::new_list(&values, &ArrowDataType::Int16, true))
1077                } else {
1078                    ScalarValue::Null
1079                }
1080            }
1081            &Type::INT4_ARRAY => {
1082                let data = portal.parameter::<Vec<Option<i32>>>(idx, &client_type)?;
1083                if let Some(data) = data {
1084                    let values = data.into_iter().map(|i| i.into()).collect::<Vec<_>>();
1085                    ScalarValue::List(ScalarValue::new_list(&values, &ArrowDataType::Int32, true))
1086                } else {
1087                    ScalarValue::Null
1088                }
1089            }
1090            &Type::INT8_ARRAY => {
1091                let data = portal.parameter::<Vec<Option<i64>>>(idx, &client_type)?;
1092                if let Some(data) = data {
1093                    let values = data.into_iter().map(|i| i.into()).collect::<Vec<_>>();
1094                    ScalarValue::List(ScalarValue::new_list(&values, &ArrowDataType::Int64, true))
1095                } else {
1096                    ScalarValue::Null
1097                }
1098            }
1099            &Type::NUMERIC_ARRAY => {
1100                let data = portal.parameter::<Vec<Option<Decimal>>>(idx, &client_type)?;
1101                if let Some(data) = data {
1102                    let build_u64_list = |data: Vec<Option<Decimal>>| -> PgWireResult<ScalarValue> {
1103                        let values = data
1104                            .into_iter()
1105                            .map(|n| {
1106                                Ok(ScalarValue::UInt64(
1107                                    n.map(|n| {
1108                                        n.to_u64().ok_or_else(|| numeric_out_of_range_error(n))
1109                                    })
1110                                    .transpose()?,
1111                                ))
1112                            })
1113                            .collect::<PgWireResult<Vec<_>>>()?;
1114                        Ok(ScalarValue::List(ScalarValue::new_list(
1115                            &values,
1116                            &ArrowDataType::UInt64,
1117                            true,
1118                        )))
1119                    };
1120                    if let Some(server_type) = &server_type {
1121                        match server_type {
1122                            ConcreteDataType::List(list_type) => match list_type.item_type() {
1123                                ConcreteDataType::UInt64(_) => build_u64_list(data)?,
1124                                ConcreteDataType::Decimal128(dt) => {
1125                                    let values = data
1126                                        .into_iter()
1127                                        .map(|n| to_decimal_scalar_value(n, dt))
1128                                        .collect::<Vec<_>>();
1129                                    ScalarValue::List(ScalarValue::new_list(
1130                                        &values,
1131                                        &ArrowDataType::Decimal128(dt.precision(), dt.scale()),
1132                                        true,
1133                                    ))
1134                                }
1135                                _ => {
1136                                    // the server type is not a list of decimal or uint64
1137                                    return Err(invalid_parameter_error(
1138                                        "invalid_parameter_type",
1139                                        Some(format!(
1140                                            "Expected: {}, found: {}",
1141                                            list_type.item_type(),
1142                                            client_type
1143                                        )),
1144                                    ));
1145                                }
1146                            },
1147                            _ => {
1148                                // the server type is not a list
1149                                return Err(invalid_parameter_error(
1150                                    "invalid_parameter_type",
1151                                    Some(format!(
1152                                        "Expected: {}, found: {}",
1153                                        server_type, client_type
1154                                    )),
1155                                ));
1156                            }
1157                        }
1158                    } else {
1159                        // server type not provided
1160                        build_u64_list(data)?
1161                    }
1162                } else {
1163                    ScalarValue::Null
1164                }
1165            }
1166            &Type::VARCHAR_ARRAY | &Type::TEXT_ARRAY | &Type::CHAR_ARRAY => {
1167                let data = portal.parameter::<Vec<Option<String>>>(idx, &client_type)?;
1168                if let Some(data) = data {
1169                    let values = data.into_iter().map(|i| i.into()).collect::<Vec<_>>();
1170                    ScalarValue::List(ScalarValue::new_list(&values, &ArrowDataType::Utf8, true))
1171                } else {
1172                    ScalarValue::Null
1173                }
1174            }
1175            &Type::TIMESTAMP_ARRAY => {
1176                let data = portal.parameter::<Vec<Option<NaiveDateTime>>>(idx, &client_type)?;
1177                if let Some(data) = data {
1178                    if let Some(ConcreteDataType::List(list_type)) = &server_type {
1179                        match list_type.item_type() {
1180                            ConcreteDataType::Timestamp(unit) => match *unit {
1181                                TimestampType::Second(_) => {
1182                                    let values = data
1183                                        .into_iter()
1184                                        .map(|ts| {
1185                                            ScalarValue::TimestampSecond(
1186                                                ts.map(|ts| ts.and_utc().timestamp()),
1187                                                None,
1188                                            )
1189                                        })
1190                                        .collect::<Vec<_>>();
1191                                    ScalarValue::List(ScalarValue::new_list(
1192                                        &values,
1193                                        &ArrowDataType::Timestamp(TimeUnit::Second, None),
1194                                        true,
1195                                    ))
1196                                }
1197                                TimestampType::Millisecond(_) => {
1198                                    let values = data
1199                                        .into_iter()
1200                                        .map(|ts| {
1201                                            ScalarValue::TimestampMillisecond(
1202                                                ts.map(|ts| ts.and_utc().timestamp_millis()),
1203                                                None,
1204                                            )
1205                                        })
1206                                        .collect::<Vec<_>>();
1207                                    ScalarValue::List(ScalarValue::new_list(
1208                                        &values,
1209                                        &ArrowDataType::Timestamp(TimeUnit::Millisecond, None),
1210                                        true,
1211                                    ))
1212                                }
1213                                TimestampType::Microsecond(_) => {
1214                                    let values = data
1215                                        .into_iter()
1216                                        .map(|ts| {
1217                                            ScalarValue::TimestampMicrosecond(
1218                                                ts.map(|ts| ts.and_utc().timestamp_micros()),
1219                                                None,
1220                                            )
1221                                        })
1222                                        .collect::<Vec<_>>();
1223                                    ScalarValue::List(ScalarValue::new_list(
1224                                        &values,
1225                                        &ArrowDataType::Timestamp(TimeUnit::Microsecond, None),
1226                                        true,
1227                                    ))
1228                                }
1229                                TimestampType::Nanosecond(_) => {
1230                                    let values = data
1231                                        .into_iter()
1232                                        .map(|ts| match ts {
1233                                            None => {
1234                                                Ok(ScalarValue::TimestampNanosecond(None, None))
1235                                            }
1236                                            Some(ts) => ts
1237                                                .and_utc()
1238                                                .timestamp_nanos_opt()
1239                                                .map(|nanos| {
1240                                                    ScalarValue::TimestampNanosecond(
1241                                                        Some(nanos),
1242                                                        None,
1243                                                    )
1244                                                })
1245                                                .ok_or_else(|| numeric_out_of_range_error(ts)),
1246                                        })
1247                                        .collect::<PgWireResult<Vec<_>>>()?;
1248                                    ScalarValue::List(ScalarValue::new_list(
1249                                        &values,
1250                                        &ArrowDataType::Timestamp(TimeUnit::Nanosecond, None),
1251                                        true,
1252                                    ))
1253                                }
1254                            },
1255                            _ => {
1256                                return Err(invalid_parameter_error(
1257                                    "invalid_parameter_type",
1258                                    Some(format!(
1259                                        "Expected: {}, found: {}",
1260                                        list_type.item_type(),
1261                                        client_type
1262                                    )),
1263                                ));
1264                            }
1265                        }
1266                    } else {
1267                        let values = data
1268                            .into_iter()
1269                            .map(|ts| {
1270                                ScalarValue::TimestampMillisecond(
1271                                    ts.map(|ts| ts.and_utc().timestamp_millis()),
1272                                    None,
1273                                )
1274                            })
1275                            .collect::<Vec<_>>();
1276                        ScalarValue::List(ScalarValue::new_list(
1277                            &values,
1278                            &ArrowDataType::Timestamp(TimeUnit::Millisecond, None),
1279                            true,
1280                        ))
1281                    }
1282                } else {
1283                    ScalarValue::Null
1284                }
1285            }
1286            &Type::TIMESTAMPTZ_ARRAY => {
1287                let data =
1288                    portal.parameter::<Vec<Option<DateTime<FixedOffset>>>>(idx, &client_type)?;
1289                if let Some(data) = data {
1290                    if let Some(ConcreteDataType::List(list_type)) = &server_type {
1291                        match list_type.item_type() {
1292                            ConcreteDataType::Timestamp(unit) => match *unit {
1293                                TimestampType::Second(_) => {
1294                                    let values = data
1295                                        .into_iter()
1296                                        .map(|ts| {
1297                                            ScalarValue::TimestampSecond(
1298                                                ts.map(|ts| ts.timestamp()),
1299                                                None,
1300                                            )
1301                                        })
1302                                        .collect::<Vec<_>>();
1303                                    ScalarValue::List(ScalarValue::new_list(
1304                                        &values,
1305                                        &ArrowDataType::Timestamp(TimeUnit::Second, None),
1306                                        true,
1307                                    ))
1308                                }
1309                                TimestampType::Millisecond(_) => {
1310                                    let values = data
1311                                        .into_iter()
1312                                        .map(|ts| {
1313                                            ScalarValue::TimestampMillisecond(
1314                                                ts.map(|ts| ts.timestamp_millis()),
1315                                                None,
1316                                            )
1317                                        })
1318                                        .collect::<Vec<_>>();
1319                                    ScalarValue::List(ScalarValue::new_list(
1320                                        &values,
1321                                        &ArrowDataType::Timestamp(TimeUnit::Millisecond, None),
1322                                        true,
1323                                    ))
1324                                }
1325                                TimestampType::Microsecond(_) => {
1326                                    let values = data
1327                                        .into_iter()
1328                                        .map(|ts| {
1329                                            ScalarValue::TimestampMicrosecond(
1330                                                ts.map(|ts| ts.timestamp_micros()),
1331                                                None,
1332                                            )
1333                                        })
1334                                        .collect::<Vec<_>>();
1335                                    ScalarValue::List(ScalarValue::new_list(
1336                                        &values,
1337                                        &ArrowDataType::Timestamp(TimeUnit::Microsecond, None),
1338                                        true,
1339                                    ))
1340                                }
1341                                TimestampType::Nanosecond(_) => {
1342                                    let values = data
1343                                        .into_iter()
1344                                        .map(|ts| {
1345                                            ScalarValue::TimestampNanosecond(
1346                                                ts.and_then(|ts| ts.timestamp_nanos_opt()),
1347                                                None,
1348                                            )
1349                                        })
1350                                        .collect::<Vec<_>>();
1351                                    ScalarValue::List(ScalarValue::new_list(
1352                                        &values,
1353                                        &ArrowDataType::Timestamp(TimeUnit::Nanosecond, None),
1354                                        true,
1355                                    ))
1356                                }
1357                            },
1358                            _ => {
1359                                return Err(invalid_parameter_error(
1360                                    "invalid_parameter_type",
1361                                    Some(format!(
1362                                        "Expected: {}, found: {}",
1363                                        list_type.item_type(),
1364                                        client_type
1365                                    )),
1366                                ));
1367                            }
1368                        }
1369                    } else {
1370                        let values = data
1371                            .into_iter()
1372                            .map(|ts| {
1373                                ScalarValue::TimestampMillisecond(
1374                                    ts.map(|ts| ts.timestamp_millis()),
1375                                    None,
1376                                )
1377                            })
1378                            .collect::<Vec<_>>();
1379                        ScalarValue::List(ScalarValue::new_list(
1380                            &values,
1381                            &ArrowDataType::Timestamp(TimeUnit::Millisecond, None),
1382                            true,
1383                        ))
1384                    }
1385                } else {
1386                    ScalarValue::Null
1387                }
1388            }
1389            _ => Err(invalid_parameter_error(
1390                "unsupported_parameter_value",
1391                Some(format!("Found type: {}", client_type)),
1392            ))?,
1393        };
1394
1395        results.push(value);
1396    }
1397
1398    Ok(results)
1399}
1400
1401pub(super) fn param_types_to_pg_types(
1402    param_types: &HashMap<String, Option<ConcreteDataType>>,
1403) -> Result<Vec<Type>> {
1404    let param_count = param_types.len();
1405    let mut types = Vec::with_capacity(param_count);
1406    for i in 0..param_count {
1407        if let Some(Some(param_type)) = param_types.get(&format!("${}", i + 1)) {
1408            let pg_type = type_gt_to_pg(param_type)?;
1409            types.push(pg_type);
1410        } else {
1411            types.push(Type::UNKNOWN);
1412        }
1413    }
1414    Ok(types)
1415}
1416
1417pub fn format_options_from_query_ctx(query_ctx: &QueryContextRef) -> Arc<PgFormatOptions> {
1418    let config = query_ctx.configuration_parameter();
1419    let (date_style, date_order) = *config.pg_datetime_style();
1420
1421    let mut format_options = PgFormatOptions::default();
1422    format_options.date_style = format!("{}, {}", date_style, date_order);
1423    format_options.interval_style = config.pg_intervalstyle_format().to_string();
1424    format_options.bytea_output = config.postgres_bytea_output().to_string();
1425    format_options.time_zone = query_ctx.timezone().to_string();
1426
1427    Arc::new(format_options)
1428}
1429
1430#[cfg(test)]
1431mod test {
1432    use std::str::FromStr;
1433    use std::sync::Arc;
1434
1435    use arrow::array::{
1436        Float64Builder, Int64Builder, ListBuilder, StringBuilder, TimestampSecondBuilder,
1437    };
1438    use arrow_schema::{Field, IntervalUnit};
1439    use bytes::Bytes;
1440    use datafusion_expr::expr::Placeholder;
1441    use datafusion_expr::{Expr, LogicalPlanBuilder};
1442    use datatypes::schema::{ColumnSchema, Schema};
1443    use datatypes::vectors::{
1444        BinaryVector, BooleanVector, DateVector, Float32Vector, Float64Vector, Int8Vector,
1445        Int16Vector, Int32Vector, Int64Vector, IntervalDayTimeVector, IntervalMonthDayNanoVector,
1446        IntervalYearMonthVector, ListVector, NullVector, StringVector, TimeSecondVector,
1447        TimestampSecondVector, UInt8Vector, UInt16Vector, UInt32Vector, UInt64Vector, VectorRef,
1448    };
1449    use futures::{StreamExt as FuturesStreamExt, stream};
1450    use pgwire::api::Type;
1451    use pgwire::api::portal::{Format, Portal};
1452    use pgwire::api::results::{
1453        CopyEncoder, CopyTextOptions, DataRowEncoder, FieldFormat, FieldInfo,
1454    };
1455    use pgwire::api::stmt::StoredStatement;
1456    use pgwire::messages::extendedquery::Bind;
1457    use session::context::QueryContextBuilder;
1458
1459    use super::*;
1460    use crate::SqlPlan;
1461    use crate::postgres::handler::PgSqlPlan;
1462
1463    #[test]
1464    fn test_null_array_maps_to_text_array() {
1465        let array = ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::null_datatype()));
1466        assert_eq!(Type::TEXT_ARRAY, type_gt_to_pg(&array).unwrap());
1467    }
1468
1469    #[test]
1470    fn test_schema_convert() {
1471        let column_schemas = vec![
1472            ColumnSchema::new("nulls", ConcreteDataType::null_datatype(), true),
1473            ColumnSchema::new("bools", ConcreteDataType::boolean_datatype(), true),
1474            ColumnSchema::new("int8s", ConcreteDataType::int8_datatype(), true),
1475            ColumnSchema::new("int16s", ConcreteDataType::int16_datatype(), true),
1476            ColumnSchema::new("int32s", ConcreteDataType::int32_datatype(), true),
1477            ColumnSchema::new("int64s", ConcreteDataType::int64_datatype(), true),
1478            ColumnSchema::new("uint8s", ConcreteDataType::uint8_datatype(), true),
1479            ColumnSchema::new("uint16s", ConcreteDataType::uint16_datatype(), true),
1480            ColumnSchema::new("uint32s", ConcreteDataType::uint32_datatype(), true),
1481            ColumnSchema::new("uint64s", ConcreteDataType::uint64_datatype(), true),
1482            ColumnSchema::new("float32s", ConcreteDataType::float32_datatype(), true),
1483            ColumnSchema::new("float64s", ConcreteDataType::float64_datatype(), true),
1484            ColumnSchema::new("binaries", ConcreteDataType::binary_datatype(), true),
1485            ColumnSchema::new("strings", ConcreteDataType::string_datatype(), true),
1486            ColumnSchema::new(
1487                "timestamps",
1488                ConcreteDataType::timestamp_millisecond_datatype(),
1489                true,
1490            ),
1491            ColumnSchema::new("dates", ConcreteDataType::date_datatype(), true),
1492            ColumnSchema::new("times", ConcreteDataType::time_second_datatype(), true),
1493            ColumnSchema::new(
1494                "intervals",
1495                ConcreteDataType::interval_month_day_nano_datatype(),
1496                true,
1497            ),
1498        ];
1499        let pg_field_info = vec![
1500            FieldInfo::new("nulls".into(), None, None, Type::UNKNOWN, FieldFormat::Text),
1501            FieldInfo::new("bools".into(), None, None, Type::BOOL, FieldFormat::Text),
1502            FieldInfo::new("int8s".into(), None, None, Type::INT2, FieldFormat::Text),
1503            FieldInfo::new("int16s".into(), None, None, Type::INT2, FieldFormat::Text),
1504            FieldInfo::new("int32s".into(), None, None, Type::INT4, FieldFormat::Text),
1505            FieldInfo::new("int64s".into(), None, None, Type::INT8, FieldFormat::Text),
1506            FieldInfo::new("uint8s".into(), None, None, Type::INT2, FieldFormat::Text),
1507            FieldInfo::new("uint16s".into(), None, None, Type::INT4, FieldFormat::Text),
1508            FieldInfo::new("uint32s".into(), None, None, Type::INT8, FieldFormat::Text),
1509            FieldInfo::new(
1510                "uint64s".into(),
1511                None,
1512                None,
1513                Type::NUMERIC,
1514                FieldFormat::Text,
1515            ),
1516            FieldInfo::new(
1517                "float32s".into(),
1518                None,
1519                None,
1520                Type::FLOAT4,
1521                FieldFormat::Text,
1522            ),
1523            FieldInfo::new(
1524                "float64s".into(),
1525                None,
1526                None,
1527                Type::FLOAT8,
1528                FieldFormat::Text,
1529            ),
1530            FieldInfo::new(
1531                "binaries".into(),
1532                None,
1533                None,
1534                Type::BYTEA,
1535                FieldFormat::Text,
1536            ),
1537            FieldInfo::new(
1538                "strings".into(),
1539                None,
1540                None,
1541                Type::VARCHAR,
1542                FieldFormat::Text,
1543            ),
1544            FieldInfo::new(
1545                "timestamps".into(),
1546                None,
1547                None,
1548                Type::TIMESTAMP,
1549                FieldFormat::Text,
1550            ),
1551            FieldInfo::new("dates".into(), None, None, Type::DATE, FieldFormat::Text),
1552            FieldInfo::new("times".into(), None, None, Type::TIME, FieldFormat::Text),
1553            FieldInfo::new(
1554                "intervals".into(),
1555                None,
1556                None,
1557                Type::INTERVAL,
1558                FieldFormat::Text,
1559            ),
1560        ];
1561        let schema = Schema::new(column_schemas);
1562        let fs = schema_to_pg(&schema, &Format::UnifiedText, None).unwrap();
1563        assert_eq!(fs, pg_field_info);
1564    }
1565
1566    #[test]
1567    fn test_schema_convert_oid_alias_types() {
1568        let aliases = [
1569            (oid_field::kind::OID, Type::OID),
1570            (oid_field::kind::REGPROC, Type::REGPROC),
1571            ("regprocedure", Type::REGPROCEDURE),
1572            ("regoper", Type::REGOPER),
1573            ("regoperator", Type::REGOPERATOR),
1574            (oid_field::kind::REGCLASS, Type::REGCLASS),
1575            (oid_field::kind::REGTYPE, Type::REGTYPE),
1576            (oid_field::kind::REGNAMESPACE, Type::REGNAMESPACE),
1577            ("regrole", Type::REGROLE),
1578            ("regconfig", Type::REGCONFIG),
1579            ("regdictionary", Type::REGDICTIONARY),
1580            ("regcollation", Type::REGCOLLATION),
1581        ];
1582        let mut columns = Vec::new();
1583        let mut expected_oids = Vec::new();
1584        for (type_name, data_type, fallback_type) in [
1585            ("int32", ConcreteDataType::int32_datatype(), Type::INT4),
1586            ("utf8", ConcreteDataType::string_datatype(), Type::VARCHAR),
1587        ] {
1588            for (alias, pg_type) in &aliases {
1589                let mut column =
1590                    ColumnSchema::new(format!("{type_name}_{alias}"), data_type.clone(), true);
1591                column
1592                    .mut_metadata()
1593                    .insert(OID_ALIAS_KEY.to_string(), alias.to_string());
1594                columns.push(column);
1595                expected_oids.push(pg_type.oid());
1596            }
1597
1598            columns.push(ColumnSchema::new(
1599                format!("{type_name}_untagged"),
1600                data_type.clone(),
1601                true,
1602            ));
1603            expected_oids.push(fallback_type.oid());
1604
1605            let mut unknown = ColumnSchema::new(format!("{type_name}_unknown"), data_type, true);
1606            unknown
1607                .mut_metadata()
1608                .insert(OID_ALIAS_KEY.to_string(), "unknown".to_string());
1609            columns.push(unknown);
1610            expected_oids.push(fallback_type.oid());
1611        }
1612
1613        let fields = schema_to_pg(&Schema::new(columns), &Format::UnifiedText, None).unwrap();
1614        let actual_oids = fields
1615            .iter()
1616            .map(|field| field.datatype().oid())
1617            .collect::<Vec<_>>();
1618
1619        assert_eq!(actual_oids, expected_oids);
1620    }
1621
1622    #[test]
1623    fn test_encode_text_format_data() {
1624        let pg_schema = vec![
1625            FieldInfo::new("nulls".into(), None, None, Type::UNKNOWN, FieldFormat::Text),
1626            FieldInfo::new("bools".into(), None, None, Type::BOOL, FieldFormat::Text),
1627            FieldInfo::new("uint8s".into(), None, None, Type::INT2, FieldFormat::Text),
1628            FieldInfo::new("uint16s".into(), None, None, Type::INT4, FieldFormat::Text),
1629            FieldInfo::new("uint32s".into(), None, None, Type::INT8, FieldFormat::Text),
1630            FieldInfo::new(
1631                "uint64s".into(),
1632                None,
1633                None,
1634                Type::NUMERIC,
1635                FieldFormat::Text,
1636            ),
1637            FieldInfo::new("int8s".into(), None, None, Type::INT2, FieldFormat::Text),
1638            FieldInfo::new("int16s".into(), None, None, Type::INT2, FieldFormat::Text),
1639            FieldInfo::new("int32s".into(), None, None, Type::INT4, FieldFormat::Text),
1640            FieldInfo::new("int64s".into(), None, None, Type::INT8, FieldFormat::Text),
1641            FieldInfo::new(
1642                "float32s".into(),
1643                None,
1644                None,
1645                Type::FLOAT4,
1646                FieldFormat::Text,
1647            ),
1648            FieldInfo::new(
1649                "float64s".into(),
1650                None,
1651                None,
1652                Type::FLOAT8,
1653                FieldFormat::Text,
1654            ),
1655            FieldInfo::new(
1656                "strings".into(),
1657                None,
1658                None,
1659                Type::VARCHAR,
1660                FieldFormat::Text,
1661            ),
1662            FieldInfo::new(
1663                "binaries".into(),
1664                None,
1665                None,
1666                Type::BYTEA,
1667                FieldFormat::Text,
1668            ),
1669            FieldInfo::new("dates".into(), None, None, Type::DATE, FieldFormat::Text),
1670            FieldInfo::new("times".into(), None, None, Type::TIME, FieldFormat::Text),
1671            FieldInfo::new(
1672                "timestamps".into(),
1673                None,
1674                None,
1675                Type::TIMESTAMP,
1676                FieldFormat::Text,
1677            ),
1678            FieldInfo::new(
1679                "interval_year_month".into(),
1680                None,
1681                None,
1682                Type::INTERVAL,
1683                FieldFormat::Text,
1684            ),
1685            FieldInfo::new(
1686                "interval_day_time".into(),
1687                None,
1688                None,
1689                Type::INTERVAL,
1690                FieldFormat::Text,
1691            ),
1692            FieldInfo::new(
1693                "interval_month_day_nano".into(),
1694                None,
1695                None,
1696                Type::INTERVAL,
1697                FieldFormat::Text,
1698            ),
1699            FieldInfo::new(
1700                "int_list".into(),
1701                None,
1702                None,
1703                Type::INT8_ARRAY,
1704                FieldFormat::Text,
1705            ),
1706            FieldInfo::new(
1707                "float_list".into(),
1708                None,
1709                None,
1710                Type::FLOAT8_ARRAY,
1711                FieldFormat::Text,
1712            ),
1713            FieldInfo::new(
1714                "string_list".into(),
1715                None,
1716                None,
1717                Type::VARCHAR_ARRAY,
1718                FieldFormat::Text,
1719            ),
1720            FieldInfo::new(
1721                "timestamp_list".into(),
1722                None,
1723                None,
1724                Type::TIMESTAMP_ARRAY,
1725                FieldFormat::Text,
1726            ),
1727        ];
1728
1729        let arrow_schema = arrow_schema::Schema::new(vec![
1730            Field::new("x", DataType::Null, true),
1731            Field::new("x", DataType::Boolean, true),
1732            Field::new("x", DataType::UInt8, true),
1733            Field::new("x", DataType::UInt16, true),
1734            Field::new("x", DataType::UInt32, true),
1735            Field::new("x", DataType::UInt64, true),
1736            Field::new("x", DataType::Int8, true),
1737            Field::new("x", DataType::Int16, true),
1738            Field::new("x", DataType::Int32, true),
1739            Field::new("x", DataType::Int64, true),
1740            Field::new("x", DataType::Float32, true),
1741            Field::new("x", DataType::Float64, true),
1742            Field::new("x", DataType::Utf8, true),
1743            Field::new("x", DataType::Binary, true),
1744            Field::new("x", DataType::Date32, true),
1745            Field::new("x", DataType::Time32(TimeUnit::Second), true),
1746            Field::new("x", DataType::Timestamp(TimeUnit::Second, None), true),
1747            Field::new("x", DataType::Interval(IntervalUnit::YearMonth), true),
1748            Field::new("x", DataType::Interval(IntervalUnit::DayTime), true),
1749            Field::new("x", DataType::Interval(IntervalUnit::MonthDayNano), true),
1750            Field::new(
1751                "x",
1752                DataType::List(Arc::new(Field::new("item", DataType::Int64, true))),
1753                true,
1754            ),
1755            Field::new(
1756                "x",
1757                DataType::List(Arc::new(Field::new("item", DataType::Float64, true))),
1758                true,
1759            ),
1760            Field::new(
1761                "x",
1762                DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))),
1763                true,
1764            ),
1765            Field::new(
1766                "x",
1767                DataType::List(Arc::new(Field::new(
1768                    "item",
1769                    DataType::Timestamp(TimeUnit::Second, None),
1770                    true,
1771                ))),
1772                true,
1773            ),
1774        ]);
1775
1776        let mut builder = ListBuilder::new(Int64Builder::new());
1777        builder.append_value([Some(1i64), None, Some(2)]);
1778        builder.append_null();
1779        builder.append_value([Some(-1i64), None, Some(-2)]);
1780        let i64_list_array = builder.finish();
1781
1782        let mut builder = ListBuilder::new(Float64Builder::new());
1783        builder.append_value([Some(1.0f64), None, Some(2.0)]);
1784        builder.append_null();
1785        builder.append_value([Some(-1.0f64), None, Some(-2.0)]);
1786        let f64_list_array = builder.finish();
1787
1788        let mut builder = ListBuilder::new(StringBuilder::new());
1789        builder.append_value([Some("a"), None, Some("b")]);
1790        builder.append_null();
1791        builder.append_value([Some("c"), None, Some("d")]);
1792        let string_list_array = builder.finish();
1793
1794        let mut builder = ListBuilder::new(TimestampSecondBuilder::new());
1795        builder.append_value([Some(1i64), None, Some(2)]);
1796        builder.append_null();
1797        builder.append_value([Some(3i64), None, Some(4)]);
1798        let timestamp_list_array = builder.finish();
1799
1800        let values = vec![
1801            Arc::new(NullVector::new(3)) as VectorRef,
1802            Arc::new(BooleanVector::from(vec![Some(true), Some(false), None])),
1803            Arc::new(UInt8Vector::from(vec![Some(u8::MAX), Some(u8::MIN), None])),
1804            Arc::new(UInt16Vector::from(vec![
1805                Some(u16::MAX),
1806                Some(u16::MIN),
1807                None,
1808            ])),
1809            Arc::new(UInt32Vector::from(vec![
1810                Some(u32::MAX),
1811                Some(u32::MIN),
1812                None,
1813            ])),
1814            Arc::new(UInt64Vector::from(vec![
1815                Some(u64::MAX),
1816                Some(u64::MIN),
1817                None,
1818            ])),
1819            Arc::new(Int8Vector::from(vec![Some(i8::MAX), Some(i8::MIN), None])),
1820            Arc::new(Int16Vector::from(vec![
1821                Some(i16::MAX),
1822                Some(i16::MIN),
1823                None,
1824            ])),
1825            Arc::new(Int32Vector::from(vec![
1826                Some(i32::MAX),
1827                Some(i32::MIN),
1828                None,
1829            ])),
1830            Arc::new(Int64Vector::from(vec![
1831                Some(i64::MAX),
1832                Some(i64::MIN),
1833                None,
1834            ])),
1835            Arc::new(Float32Vector::from(vec![
1836                None,
1837                Some(f32::MAX),
1838                Some(f32::MIN),
1839            ])),
1840            Arc::new(Float64Vector::from(vec![
1841                None,
1842                Some(f64::MAX),
1843                Some(f64::MIN),
1844            ])),
1845            Arc::new(StringVector::from(vec![
1846                None,
1847                Some("hello"),
1848                Some("greptime"),
1849            ])),
1850            Arc::new(BinaryVector::from(vec![
1851                None,
1852                Some("hello".as_bytes().to_vec()),
1853                Some("world".as_bytes().to_vec()),
1854            ])),
1855            Arc::new(DateVector::from(vec![Some(1001), None, Some(1)])),
1856            Arc::new(TimeSecondVector::from(vec![Some(1001), None, Some(1)])),
1857            Arc::new(TimestampSecondVector::from(vec![
1858                Some(1000001),
1859                None,
1860                Some(1),
1861            ])),
1862            Arc::new(IntervalYearMonthVector::from(vec![Some(1), None, Some(2)])),
1863            Arc::new(IntervalDayTimeVector::from(vec![
1864                Some(arrow::datatypes::IntervalDayTime::new(1, 1)),
1865                None,
1866                Some(arrow::datatypes::IntervalDayTime::new(2, 2)),
1867            ])),
1868            Arc::new(IntervalMonthDayNanoVector::from(vec![
1869                Some(arrow::datatypes::IntervalMonthDayNano::new(1, 1, 10)),
1870                None,
1871                Some(arrow::datatypes::IntervalMonthDayNano::new(2, 2, 20)),
1872            ])),
1873            Arc::new(ListVector::from(i64_list_array)),
1874            Arc::new(ListVector::from(f64_list_array)),
1875            Arc::new(ListVector::from(string_list_array)),
1876            Arc::new(ListVector::from(timestamp_list_array)),
1877        ];
1878        let record_batch =
1879            RecordBatch::new(Arc::new(arrow_schema.try_into().unwrap()), values).unwrap();
1880
1881        let query_context = QueryContextBuilder::default()
1882            .configuration_parameter(Default::default())
1883            .build()
1884            .into();
1885        let schema = record_batch.schema.clone();
1886        let pg_schema_ref = Arc::new(pg_schema);
1887
1888        let encoder = DataRowEncoder::new(pg_schema_ref.clone());
1889
1890        let row_stream = RecordBatchRowStream::new(
1891            query_context,
1892            pg_schema_ref.clone(),
1893            schema,
1894            stream::once(async { Ok(record_batch) }),
1895            encoder,
1896        );
1897
1898        let rows: Vec<_> = futures::executor::block_on(
1899            row_stream
1900                .filter_map(|x: PgWireResult<_>| async move { x.ok() })
1901                .flat_map(stream::iter)
1902                .collect::<Vec<_>>(),
1903        );
1904        assert_eq!(rows.len(), 3);
1905        for row in rows {
1906            assert_eq!(row.field_count, pg_schema_ref.len() as i16);
1907        }
1908    }
1909
1910    #[test]
1911    fn test_encode_list_of_struct_as_json() {
1912        use datatypes::types::{StructField, StructType};
1913        use datatypes::value::{ListValue, StructValue};
1914        use datatypes::vectors::Helper;
1915        use pgwire::messages::data::DataRow;
1916
1917        // struct type: {a: int32, b: float64}
1918        let fields = vec![
1919            StructField::new("a".to_string(), ConcreteDataType::int32_datatype(), true),
1920            StructField::new("b".to_string(), ConcreteDataType::float64_datatype(), true),
1921        ];
1922        let struct_type =
1923            ConcreteDataType::struct_datatype(StructType::new(Arc::new(fields.clone())));
1924        let list_type = ConcreteDataType::list_datatype(Arc::new(struct_type.clone()));
1925
1926        let make_struct = |a: i32, b: f64| {
1927            Value::Struct(StructValue::new(
1928                vec![Value::Int32(a), Value::Float64(b.into())],
1929                StructType::new(Arc::new(fields.clone())),
1930            ))
1931        };
1932        let rows = vec![
1933            Value::List(ListValue::new(
1934                vec![make_struct(1, 1.5), Value::Null, make_struct(3, 3.5)],
1935                Arc::new(struct_type.clone()),
1936            )),
1937            Value::List(ListValue::new(vec![], Arc::new(struct_type.clone()))),
1938            Value::Null,
1939        ];
1940        let vector = Helper::try_from_row_into_vector(&rows, &list_type).unwrap();
1941
1942        let schema = Arc::new(Schema::new(vec![ColumnSchema::new(
1943            "l".to_string(),
1944            list_type.clone(),
1945            true,
1946        )]));
1947        let record_batch = RecordBatch::new(schema.clone(), vec![vector]).unwrap();
1948
1949        // The list-of-struct column is declared as a single json value.
1950        let pg_schema = Arc::new(vec![FieldInfo::new(
1951            "l".into(),
1952            None,
1953            None,
1954            type_gt_to_pg(&list_type).unwrap(),
1955            FieldFormat::Text,
1956        )]);
1957        assert_eq!(pg_schema[0].datatype(), &Type::JSON);
1958
1959        let query_context = QueryContextBuilder::default()
1960            .configuration_parameter(Default::default())
1961            .build()
1962            .into();
1963        let encoder = DataRowEncoder::new(pg_schema.clone());
1964        let row_stream = RecordBatchRowStream::new(
1965            query_context,
1966            pg_schema.clone(),
1967            schema,
1968            stream::once(async { Ok(record_batch) }),
1969            encoder,
1970        );
1971
1972        let data_rows: Vec<DataRow> = futures::executor::block_on(
1973            row_stream
1974                .filter_map(|x: PgWireResult<_>| async move { x.ok() })
1975                .flat_map(stream::iter)
1976                .collect(),
1977        );
1978        assert_eq!(data_rows.len(), 3);
1979
1980        // Decode text-format fields: [i32 BE length][bytes], -1 for NULL.
1981        let decoded: Vec<Option<String>> = data_rows
1982            .iter()
1983            .map(|row| {
1984                assert_eq!(row.field_count, 1);
1985                let len = i32::from_be_bytes(row.data[0..4].try_into().unwrap());
1986                if len < 0 {
1987                    return None;
1988                }
1989                Some(String::from_utf8(row.data[4..4 + len as usize].to_vec()).unwrap())
1990            })
1991            .collect();
1992
1993        // A NULL list row stays NULL.
1994        assert_eq!(decoded[2], None);
1995
1996        // A list of structs is a JSON array; null elements stay null.
1997        let json: serde_json::Value = serde_json::from_str(decoded[0].as_deref().unwrap()).unwrap();
1998        assert_eq!(
1999            json,
2000            serde_json::json!([
2001                {"a": 1, "b": 1.5},
2002                null,
2003                {"a": 3, "b": 3.5},
2004            ])
2005        );
2006
2007        // An empty list is an empty JSON array.
2008        let json: serde_json::Value = serde_json::from_str(decoded[1].as_deref().unwrap()).unwrap();
2009        assert_eq!(json, serde_json::json!([]));
2010    }
2011
2012    #[test]
2013    fn test_encode_struct_value_as_json() {
2014        use api::greptime_proto::io::prometheus::write::v2::histogram::{Count, ZeroCount};
2015        use api::greptime_proto::io::prometheus::write::v2::{BucketSpan, Histogram};
2016        use api::helper::pb_value_to_value_ref;
2017        use api::v1::Value as PbValue;
2018        use common_query::native_histogram::{
2019            encode_native_histogram, native_histogram_column_schema, native_histogram_value_type,
2020        };
2021        use datatypes::vectors::Helper;
2022        use pgwire::messages::data::DataRow;
2023
2024        fn histogram_value(histogram: &Histogram) -> Value {
2025            let value_data = encode_native_histogram(histogram).unwrap();
2026            let pb_value = PbValue {
2027                value_data: Some(value_data),
2028            };
2029            let column_schema = native_histogram_column_schema().unwrap();
2030            Value::from(pb_value_to_value_ref(
2031                &pb_value,
2032                column_schema.datatype_extension.as_ref(),
2033            ))
2034        }
2035
2036        // One integer-count histogram and one float-count histogram, mirroring
2037        // what prometheus remote-write v2 stores. The expected JSON matches the
2038        // HTTP query output for the same payloads.
2039        let int_histogram = Histogram {
2040            schema: 1,
2041            sum: 10.0,
2042            zero_threshold: 0.001,
2043            reset_hint: 2,
2044            start_timestamp: 1500,
2045            positive_spans: vec![BucketSpan {
2046                offset: 0,
2047                length: 3,
2048            }],
2049            positive_deltas: vec![1, 2, -1],
2050            negative_spans: vec![BucketSpan {
2051                offset: -2,
2052                length: 1,
2053            }],
2054            negative_deltas: vec![1],
2055            count: Some(Count::CountInt(8)),
2056            zero_count: Some(ZeroCount::ZeroCountInt(1)),
2057            ..Default::default()
2058        };
2059        let float_histogram = Histogram {
2060            schema: 2,
2061            sum: 20.0,
2062            zero_threshold: 0.002,
2063            reset_hint: 3,
2064            start_timestamp: 2500,
2065            positive_spans: vec![BucketSpan {
2066                offset: 3,
2067                length: 2,
2068            }],
2069            positive_counts: vec![2.0, 3.5],
2070            count: Some(Count::CountFloat(6.0)),
2071            zero_count: Some(ZeroCount::ZeroCountFloat(0.5)),
2072            ..Default::default()
2073        };
2074
2075        let rows = vec![
2076            histogram_value(&int_histogram),
2077            histogram_value(&float_histogram),
2078            Value::Null,
2079        ];
2080        let vector =
2081            Helper::try_from_row_into_vector(&rows, native_histogram_value_type()).unwrap();
2082
2083        let schema = Arc::new(Schema::new(vec![ColumnSchema::new(
2084            "greptime_native_histogram",
2085            native_histogram_value_type().clone(),
2086            true,
2087        )]));
2088        let record_batch = RecordBatch::new(schema.clone(), vec![vector]).unwrap();
2089
2090        let pg_schema = Arc::new(vec![FieldInfo::new(
2091            "greptime_native_histogram".into(),
2092            None,
2093            None,
2094            Type::JSON,
2095            FieldFormat::Text,
2096        )]);
2097
2098        let query_context = QueryContextBuilder::default()
2099            .configuration_parameter(Default::default())
2100            .build()
2101            .into();
2102        let encoder = DataRowEncoder::new(pg_schema.clone());
2103        let row_stream = RecordBatchRowStream::new(
2104            query_context,
2105            pg_schema.clone(),
2106            schema,
2107            stream::once(async { Ok(record_batch) }),
2108            encoder,
2109        );
2110
2111        let data_rows: Vec<DataRow> = futures::executor::block_on(
2112            row_stream
2113                .filter_map(|x: PgWireResult<_>| async move { x.ok() })
2114                .flat_map(stream::iter)
2115                .collect(),
2116        );
2117        assert_eq!(data_rows.len(), 3);
2118
2119        // Decode text-format fields: [i32 BE length][bytes], -1 for NULL.
2120        let decoded: Vec<Option<String>> = data_rows
2121            .iter()
2122            .map(|row| {
2123                assert_eq!(row.field_count, 1);
2124                let len = i32::from_be_bytes(row.data[0..4].try_into().unwrap());
2125                if len < 0 {
2126                    return None;
2127                }
2128                Some(String::from_utf8(row.data[4..4 + len as usize].to_vec()).unwrap())
2129            })
2130            .collect();
2131
2132        // A NULL struct row stays NULL.
2133        assert_eq!(decoded[2], None);
2134
2135        let json: serde_json::Value = serde_json::from_str(decoded[0].as_deref().unwrap()).unwrap();
2136        assert_eq!(
2137            json,
2138            serde_json::json!({
2139                "count_f64": null,
2140                "count_i64": 8,
2141                "custom_values": [],
2142                "negative_buckets_f64": [],
2143                "negative_buckets_i64": [1],
2144                "negative_span_lengths": [1],
2145                "negative_span_offsets": [-2],
2146                "positive_buckets_f64": [],
2147                "positive_buckets_i64": [1, 3, 2],
2148                "positive_span_lengths": [3],
2149                "positive_span_offsets": [0],
2150                "reset_hint": 2,
2151                "schema": 1,
2152                "start_timestamp": 1500,
2153                "sum": 10.0,
2154                "zero_count_f64": null,
2155                "zero_count_i64": 1,
2156                "zero_threshold": 0.001
2157            })
2158        );
2159
2160        let json: serde_json::Value = serde_json::from_str(decoded[1].as_deref().unwrap()).unwrap();
2161        assert_eq!(
2162            json,
2163            serde_json::json!({
2164                "count_f64": 6.0,
2165                "count_i64": null,
2166                "custom_values": [],
2167                "negative_buckets_f64": [],
2168                "negative_buckets_i64": [],
2169                "negative_span_lengths": [],
2170                "negative_span_offsets": [],
2171                "positive_buckets_f64": [2.0, 3.5],
2172                "positive_buckets_i64": [],
2173                "positive_span_lengths": [2],
2174                "positive_span_offsets": [3],
2175                "reset_hint": 3,
2176                "schema": 2,
2177                "start_timestamp": 2500,
2178                "sum": 20.0,
2179                "zero_count_f64": 0.5,
2180                "zero_count_i64": null,
2181                "zero_threshold": 0.002
2182            })
2183        );
2184    }
2185
2186    #[test]
2187    fn test_encode_utf8_oid_alias_data() {
2188        let aliases = [
2189            ("regproc_binary", FieldFormat::Binary, Some("boolrecv")),
2190            ("regproc_text", FieldFormat::Text, Some("boolrecv")),
2191            ("regproc_unknown_text", FieldFormat::Text, Some("unknown")),
2192            ("regproc_ambiguous_text", FieldFormat::Text, Some("int4")),
2193            ("regtype_text", FieldFormat::Text, Some("int4recv")),
2194            ("regproc_zero", FieldFormat::Binary, Some("-")),
2195            ("regproc_null", FieldFormat::Binary, None),
2196        ];
2197        let mut columns = Vec::new();
2198        let mut values = Vec::new();
2199        let mut formats = Vec::new();
2200
2201        for (name, format, value) in aliases {
2202            let mut column = ColumnSchema::new(name, ConcreteDataType::string_datatype(), true);
2203            column.mut_metadata().insert(
2204                OID_ALIAS_KEY.to_string(),
2205                if name == "regtype_text" {
2206                    oid_field::kind::REGTYPE.to_string()
2207                } else {
2208                    oid_field::kind::REGPROC.to_string()
2209                },
2210            );
2211            columns.push(column);
2212            values.push(Arc::new(StringVector::from(vec![value])) as VectorRef);
2213            formats.push(format.value());
2214        }
2215
2216        columns.push(ColumnSchema::new(
2217            "varchar",
2218            ConcreteDataType::string_datatype(),
2219            false,
2220        ));
2221        values.push(Arc::new(StringVector::from(vec![Some("varchar")])) as VectorRef);
2222        formats.push(FieldFormat::Binary.value());
2223
2224        let schema = Arc::new(Schema::new(columns));
2225        let pg_schema =
2226            Arc::new(schema_to_pg(&schema, &Format::Individual(formats), None).unwrap());
2227        let record_batch = RecordBatch::new(schema.clone(), values).unwrap();
2228        let query_context = QueryContextBuilder::default()
2229            .configuration_parameter(Default::default())
2230            .build()
2231            .into();
2232        let row_stream = RecordBatchRowStream::new(
2233            query_context,
2234            pg_schema.clone(),
2235            schema,
2236            stream::once(async { Ok(record_batch) }),
2237            DataRowEncoder::new(pg_schema),
2238        );
2239
2240        let row = futures::executor::block_on(row_stream.into_future())
2241            .0
2242            .unwrap()
2243            .unwrap()
2244            .pop()
2245            .unwrap();
2246        assert_eq!(row.field_count, 8);
2247        assert_eq!(
2248            &row.data[..],
2249            [
2250                0, 0, 0, 4, 0, 0, 9, 132, // boolrecv (OID 2436), binary
2251                0, 0, 0, 8, b'b', b'o', b'o', b'l', b'r', b'e', b'c', b'v', // text
2252                0, 0, 0, 7, b'u', b'n', b'k', b'n', b'o', b'w', b'n', // unknown text
2253                0, 0, 0, 4, b'i', b'n', b't', b'4', // ambiguous text
2254                0, 0, 0, 8, b'i', b'n', b't', b'4', b'r', b'e', b'c', b'v', // regtype text
2255                0, 0, 0, 4, 0, 0, 0, 0, // - is OID 0
2256                255, 255, 255, 255, // NULL
2257                0, 0, 0, 7, b'v', b'a', b'r', b'c', b'h', b'a', b'r',
2258            ]
2259        );
2260    }
2261
2262    #[test]
2263    fn test_encode_utf8_oid_alias_numeric_data() {
2264        let aliases = [
2265            oid_field::kind::OID,
2266            oid_field::kind::REGPROC,
2267            "regprocedure",
2268            "regoper",
2269            "regoperator",
2270            oid_field::kind::REGCLASS,
2271            oid_field::kind::REGTYPE,
2272            oid_field::kind::REGNAMESPACE,
2273            "regrole",
2274            "regconfig",
2275            "regdictionary",
2276            "regcollation",
2277        ];
2278        let mut columns = Vec::new();
2279        let mut values = Vec::new();
2280        for alias in aliases {
2281            let mut column = ColumnSchema::new(alias, ConcreteDataType::string_datatype(), false);
2282            column
2283                .mut_metadata()
2284                .insert(OID_ALIAS_KEY.to_string(), alias.to_string());
2285            columns.push(column);
2286            values.push(Arc::new(StringVector::from(vec![Some("4294967295")])) as VectorRef);
2287        }
2288
2289        let schema = Arc::new(Schema::new(columns));
2290        let pg_schema = Arc::new(schema_to_pg(&schema, &Format::UnifiedBinary, None).unwrap());
2291        let record_batch = RecordBatch::new(schema.clone(), values).unwrap();
2292        let query_context = QueryContextBuilder::default()
2293            .configuration_parameter(Default::default())
2294            .build()
2295            .into();
2296        let row_stream = RecordBatchRowStream::new(
2297            query_context,
2298            pg_schema.clone(),
2299            schema,
2300            stream::once(async { Ok(record_batch) }),
2301            DataRowEncoder::new(pg_schema),
2302        );
2303
2304        let row = futures::executor::block_on(row_stream.into_future())
2305            .0
2306            .unwrap()
2307            .unwrap()
2308            .pop()
2309            .unwrap();
2310        assert_eq!(row.field_count, aliases.len() as i16);
2311        assert_eq!(
2312            &row.data[..],
2313            &[[0, 0, 0, 4, 255, 255, 255, 255]; 12].concat()
2314        );
2315    }
2316
2317    #[test]
2318    fn test_encode_utf8_oid_alias_binary_errors() {
2319        for value in ["unknown", "int4"] {
2320            let mut column =
2321                ColumnSchema::new("regproc", ConcreteDataType::string_datatype(), false);
2322            column.mut_metadata().insert(
2323                OID_ALIAS_KEY.to_string(),
2324                oid_field::kind::REGPROC.to_string(),
2325            );
2326            let schema = Arc::new(Schema::new(vec![column]));
2327            let pg_schema = Arc::new(schema_to_pg(&schema, &Format::UnifiedBinary, None).unwrap());
2328            let record_batch = RecordBatch::new(
2329                schema.clone(),
2330                vec![Arc::new(StringVector::from(vec![Some(value)])) as VectorRef],
2331            )
2332            .unwrap();
2333            let query_context = QueryContextBuilder::default()
2334                .configuration_parameter(Default::default())
2335                .build()
2336                .into();
2337            let row_stream = RecordBatchRowStream::new(
2338                query_context,
2339                pg_schema.clone(),
2340                schema,
2341                stream::once(async { Ok(record_batch) }),
2342                DataRowEncoder::new(pg_schema),
2343            );
2344
2345            assert!(
2346                futures::executor::block_on(row_stream.into_future())
2347                    .0
2348                    .unwrap()
2349                    .is_err()
2350            );
2351        }
2352    }
2353
2354    #[test]
2355    fn test_copy_text_utf8_oid_alias_preserves_named_value() {
2356        let mut column = ColumnSchema::new("regproc", ConcreteDataType::string_datatype(), false);
2357        column.mut_metadata().insert(
2358            OID_ALIAS_KEY.to_string(),
2359            oid_field::kind::REGPROC.to_string(),
2360        );
2361        let schema = Arc::new(Schema::new(vec![column]));
2362        let pg_schema = Arc::new(schema_to_pg(&schema, &Format::UnifiedBinary, None).unwrap());
2363        let record_batch = RecordBatch::new(
2364            schema.clone(),
2365            vec![Arc::new(StringVector::from(vec![Some("unknown")])) as VectorRef],
2366        )
2367        .unwrap();
2368        let query_context = QueryContextBuilder::default()
2369            .configuration_parameter(Default::default())
2370            .build()
2371            .into();
2372        let row_stream = RecordBatchRowStream::new(
2373            query_context,
2374            pg_schema.clone(),
2375            schema,
2376            stream::once(async { Ok(record_batch) }),
2377            CopyEncoder::new_text(pg_schema, CopyTextOptions::default()),
2378        );
2379
2380        let row = futures::executor::block_on(row_stream.into_future())
2381            .0
2382            .unwrap()
2383            .unwrap()
2384            .pop()
2385            .unwrap();
2386        assert_eq!(row.data.as_ref(), b"unknown\n");
2387    }
2388
2389    #[test]
2390    fn test_invalid_parameter() {
2391        // test for refactor with PgErrorCode
2392        let msg = "invalid_parameter_count";
2393        let error = invalid_parameter_error(msg, None);
2394        if let PgWireError::UserError(value) = error {
2395            assert_eq!("ERROR", value.severity);
2396            assert_eq!("22023", value.code);
2397            assert_eq!(msg, value.message);
2398        } else {
2399            panic!("test_invalid_parameter failed");
2400        }
2401    }
2402
2403    #[test]
2404    fn test_to_decimal_scalar_value() {
2405        let dt = Decimal128Type::new(18, 4);
2406
2407        let d = Decimal::from_str("12345.6789").unwrap();
2408        assert_eq!(d.mantissa(), 123456789i128);
2409        let scalar = to_decimal_scalar_value(Some(d), &dt);
2410        assert_eq!(scalar, ScalarValue::Decimal128(Some(123456789), 18, 4));
2411
2412        let d = Decimal::from_str("100.5").unwrap();
2413        assert_eq!(d.mantissa(), 1005);
2414        let scalar = to_decimal_scalar_value(Some(d), &dt);
2415        assert_eq!(scalar, ScalarValue::Decimal128(Some(1005000), 18, 4));
2416
2417        let d = Decimal::from_str("-9876.5432").unwrap();
2418        let scalar = to_decimal_scalar_value(Some(d), &dt);
2419        assert_eq!(scalar, ScalarValue::Decimal128(Some(-98765432), 18, 4));
2420
2421        let scalar = to_decimal_scalar_value(None, &dt);
2422        assert_eq!(scalar, ScalarValue::Decimal128(None, 18, 4));
2423    }
2424
2425    fn s(v: &str) -> Option<String> {
2426        Some(v.to_string())
2427    }
2428
2429    fn typed_param(id: &str, dt: DataType) -> Expr {
2430        Expr::Placeholder(Placeholder::new_with_field(
2431            id.to_string(),
2432            Some(Arc::new(arrow_schema::Field::new(id, dt, true))),
2433        ))
2434    }
2435
2436    fn build_plan_with_params(params: Vec<(&str, DataType)>) -> LogicalPlan {
2437        let exprs: Vec<Expr> = params
2438            .into_iter()
2439            .map(|(id, dt)| typed_param(id, dt))
2440            .collect();
2441        LogicalPlanBuilder::empty(true)
2442            .project(exprs)
2443            .unwrap()
2444            .build()
2445            .unwrap()
2446    }
2447
2448    fn make_portal(
2449        client_param_types: Vec<Option<Type>>,
2450        param_data: Vec<Option<String>>,
2451    ) -> Portal<PgSqlPlan> {
2452        let bind = Bind::new(
2453            None,
2454            None,
2455            vec![],
2456            param_data
2457                .into_iter()
2458                .map(|opt| opt.map(Bytes::from))
2459                .collect(),
2460            vec![],
2461        );
2462        let statement = Arc::new(StoredStatement::new(
2463            String::new(),
2464            PgSqlPlan {
2465                plan: SqlPlan::Empty,
2466                copy_to_stdout_format: None,
2467            },
2468            client_param_types,
2469        ));
2470        Portal::try_new(&bind, statement).unwrap()
2471    }
2472
2473    #[test]
2474    fn test_dictionary_string_parameter() {
2475        let dictionary_type =
2476            DataType::Dictionary(Box::new(DataType::UInt32), Box::new(DataType::Utf8));
2477        let plan = build_plan_with_params(vec![("$1", dictionary_type)]);
2478        let portal = make_portal(vec![None], vec![s("host-a")]);
2479
2480        let values = parameters_to_scalar_values(&plan, &portal).unwrap();
2481        assert_eq!(
2482            vec![ScalarValue::Dictionary(
2483                Box::new(DataType::UInt32),
2484                Box::new(ScalarValue::Utf8(s("host-a"))),
2485            )],
2486            values
2487        );
2488
2489        let param_types = HashMap::from([(
2490            "$1".to_string(),
2491            Some(ConcreteDataType::dictionary_datatype(
2492                ConcreteDataType::uint32_datatype(),
2493                ConcreteDataType::string_datatype(),
2494            )),
2495        )]);
2496        assert_eq!(
2497            vec![Type::VARCHAR],
2498            param_types_to_pg_types(&param_types).unwrap()
2499        );
2500    }
2501
2502    #[test]
2503    fn test_int2_coerce_in_range() {
2504        let plan = build_plan_with_params(vec![
2505            ("$1", DataType::Int8),
2506            ("$2", DataType::Int16),
2507            ("$3", DataType::Int32),
2508            ("$4", DataType::Int64),
2509            ("$5", DataType::UInt8),
2510            ("$6", DataType::UInt16),
2511            ("$7", DataType::UInt32),
2512            ("$8", DataType::UInt64),
2513        ]);
2514        let portal = make_portal(
2515            vec![
2516                Some(Type::INT2),
2517                Some(Type::INT2),
2518                Some(Type::INT2),
2519                Some(Type::INT2),
2520                Some(Type::INT2),
2521                Some(Type::INT2),
2522                Some(Type::INT2),
2523                Some(Type::INT2),
2524            ],
2525            vec![
2526                s("100"),
2527                s("100"),
2528                s("100"),
2529                s("100"),
2530                s("100"),
2531                s("100"),
2532                s("100"),
2533                s("100"),
2534            ],
2535        );
2536
2537        let values = parameters_to_scalar_values(&plan, &portal).unwrap();
2538        assert_eq!(values[0], ScalarValue::Int8(Some(100)));
2539        assert_eq!(values[1], ScalarValue::Int16(Some(100)));
2540        assert_eq!(values[2], ScalarValue::Int32(Some(100)));
2541        assert_eq!(values[3], ScalarValue::Int64(Some(100)));
2542        assert_eq!(values[4], ScalarValue::UInt8(Some(100)));
2543        assert_eq!(values[5], ScalarValue::UInt16(Some(100)));
2544        assert_eq!(values[6], ScalarValue::UInt32(Some(100)));
2545        assert_eq!(values[7], ScalarValue::UInt64(Some(100)));
2546    }
2547
2548    #[test]
2549    fn test_int2_coerce_out_of_range() {
2550        let plan = build_plan_with_params(vec![("$1", DataType::Int8)]);
2551        let portal = make_portal(vec![Some(Type::INT2)], vec![s("200")]);
2552        let result = parameters_to_scalar_values(&plan, &portal);
2553        assert!(result.is_err());
2554    }
2555
2556    #[test]
2557    fn test_int2_coerce_negative_to_unsigned_out_of_range() {
2558        let plan = build_plan_with_params(vec![("$1", DataType::UInt64)]);
2559        let portal = make_portal(vec![Some(Type::INT2)], vec![s("-1")]);
2560        let result = parameters_to_scalar_values(&plan, &portal);
2561        assert!(result.is_err());
2562    }
2563
2564    #[test]
2565    fn test_int4_coerce_in_range() {
2566        let plan = build_plan_with_params(vec![
2567            ("$1", DataType::Int8),
2568            ("$2", DataType::Int16),
2569            ("$3", DataType::Int32),
2570            ("$4", DataType::Int64),
2571            ("$5", DataType::UInt8),
2572            ("$6", DataType::UInt16),
2573            ("$7", DataType::UInt32),
2574            ("$8", DataType::UInt64),
2575        ]);
2576        let portal = make_portal(
2577            vec![
2578                Some(Type::INT4),
2579                Some(Type::INT4),
2580                Some(Type::INT4),
2581                Some(Type::INT4),
2582                Some(Type::INT4),
2583                Some(Type::INT4),
2584                Some(Type::INT4),
2585                Some(Type::INT4),
2586            ],
2587            vec![
2588                s("100"),
2589                s("1000"),
2590                s("100000"),
2591                s("100000"),
2592                s("200"),
2593                s("1000"),
2594                s("100000"),
2595                s("100000"),
2596            ],
2597        );
2598
2599        let values = parameters_to_scalar_values(&plan, &portal).unwrap();
2600        assert_eq!(values[0], ScalarValue::Int8(Some(100)));
2601        assert_eq!(values[1], ScalarValue::Int16(Some(1000)));
2602        assert_eq!(values[2], ScalarValue::Int32(Some(100000)));
2603        assert_eq!(values[3], ScalarValue::Int64(Some(100000)));
2604        assert_eq!(values[4], ScalarValue::UInt8(Some(200)));
2605        assert_eq!(values[5], ScalarValue::UInt16(Some(1000)));
2606        assert_eq!(values[6], ScalarValue::UInt32(Some(100000)));
2607        assert_eq!(values[7], ScalarValue::UInt64(Some(100000)));
2608    }
2609
2610    #[test]
2611    fn test_int4_coerce_out_of_range() {
2612        let plan = build_plan_with_params(vec![("$1", DataType::Int8)]);
2613        let portal = make_portal(vec![Some(Type::INT4)], vec![s("200")]);
2614        let result = parameters_to_scalar_values(&plan, &portal);
2615        assert!(result.is_err());
2616    }
2617
2618    #[test]
2619    fn test_int4_coerce_i32_max_to_i16_out_of_range() {
2620        let plan = build_plan_with_params(vec![("$1", DataType::Int16)]);
2621        let portal = make_portal(vec![Some(Type::INT4)], vec![Some(i32::MAX.to_string())]);
2622        let result = parameters_to_scalar_values(&plan, &portal);
2623        assert!(result.is_err());
2624    }
2625
2626    #[test]
2627    fn test_int8_coerce_in_range() {
2628        let plan = build_plan_with_params(vec![
2629            ("$1", DataType::Int8),
2630            ("$2", DataType::Int16),
2631            ("$3", DataType::Int32),
2632            ("$4", DataType::Int64),
2633            ("$5", DataType::UInt8),
2634            ("$6", DataType::UInt16),
2635            ("$7", DataType::UInt32),
2636            ("$8", DataType::UInt64),
2637        ]);
2638        let portal = make_portal(
2639            vec![
2640                Some(Type::INT8),
2641                Some(Type::INT8),
2642                Some(Type::INT8),
2643                Some(Type::INT8),
2644                Some(Type::INT8),
2645                Some(Type::INT8),
2646                Some(Type::INT8),
2647                Some(Type::INT8),
2648            ],
2649            vec![
2650                s("100"),
2651                s("1000"),
2652                s("100000"),
2653                s("100000"),
2654                s("200"),
2655                s("1000"),
2656                s("3000000000"),
2657                s("3000000000"),
2658            ],
2659        );
2660
2661        let values = parameters_to_scalar_values(&plan, &portal).unwrap();
2662        assert_eq!(values[0], ScalarValue::Int8(Some(100)));
2663        assert_eq!(values[1], ScalarValue::Int16(Some(1000)));
2664        assert_eq!(values[2], ScalarValue::Int32(Some(100000)));
2665        assert_eq!(values[3], ScalarValue::Int64(Some(100000)));
2666        assert_eq!(values[4], ScalarValue::UInt8(Some(200)));
2667        assert_eq!(values[5], ScalarValue::UInt16(Some(1000)));
2668        assert_eq!(values[6], ScalarValue::UInt32(Some(3000000000)));
2669        assert_eq!(values[7], ScalarValue::UInt64(Some(3000000000)));
2670    }
2671
2672    #[test]
2673    fn test_int8_coerce_out_of_range() {
2674        let plan = build_plan_with_params(vec![("$1", DataType::Int32)]);
2675        let portal = make_portal(
2676            vec![Some(Type::INT8)],
2677            vec![Some((i32::MAX as i64 + 1).to_string())],
2678        );
2679        let result = parameters_to_scalar_values(&plan, &portal);
2680        assert!(result.is_err());
2681    }
2682
2683    #[test]
2684    fn test_int8_coerce_negative_to_unsigned_out_of_range() {
2685        let plan = build_plan_with_params(vec![("$1", DataType::UInt64)]);
2686        let portal = make_portal(vec![Some(Type::INT8)], vec![s("-1")]);
2687        let result = parameters_to_scalar_values(&plan, &portal);
2688        assert!(result.is_err());
2689    }
2690
2691    #[test]
2692    fn test_float4_coerce_in_range() {
2693        let plan =
2694            build_plan_with_params(vec![("$1", DataType::Float32), ("$2", DataType::Float64)]);
2695        let portal = make_portal(
2696            vec![Some(Type::FLOAT4), Some(Type::FLOAT4)],
2697            vec![s("1.5"), s("2.5")],
2698        );
2699
2700        let values = parameters_to_scalar_values(&plan, &portal).unwrap();
2701        assert_eq!(values[0], ScalarValue::Float32(Some(1.5)));
2702        assert_eq!(values[1], ScalarValue::Float64(Some(2.5)));
2703    }
2704
2705    #[test]
2706    fn test_float4_coerce_to_int_in_range() {
2707        let plan = build_plan_with_params(vec![
2708            ("$1", DataType::Int8),
2709            ("$2", DataType::Int32),
2710            ("$3", DataType::UInt64),
2711        ]);
2712        let portal = make_portal(
2713            vec![Some(Type::FLOAT4), Some(Type::FLOAT4), Some(Type::FLOAT4)],
2714            vec![s("100"), s("1000"), s("200")],
2715        );
2716
2717        let values = parameters_to_scalar_values(&plan, &portal).unwrap();
2718        assert_eq!(values[0], ScalarValue::Int8(Some(100)));
2719        assert_eq!(values[1], ScalarValue::Int32(Some(1000)));
2720        assert_eq!(values[2], ScalarValue::UInt64(Some(200)));
2721    }
2722
2723    #[test]
2724    fn test_float4_coerce_to_int_out_of_range() {
2725        let plan = build_plan_with_params(vec![("$1", DataType::Int8)]);
2726        let portal = make_portal(vec![Some(Type::FLOAT4)], vec![s("200")]);
2727        let result = parameters_to_scalar_values(&plan, &portal);
2728        assert!(result.is_err());
2729    }
2730
2731    #[test]
2732    fn test_float8_coerce_in_range() {
2733        let plan =
2734            build_plan_with_params(vec![("$1", DataType::Float32), ("$2", DataType::Float64)]);
2735        let portal = make_portal(
2736            vec![Some(Type::FLOAT8), Some(Type::FLOAT8)],
2737            vec![s("1.5"), s("2.5")],
2738        );
2739
2740        let values = parameters_to_scalar_values(&plan, &portal).unwrap();
2741        assert_eq!(values[0], ScalarValue::Float32(Some(1.5)));
2742        assert_eq!(values[1], ScalarValue::Float64(Some(2.5)));
2743    }
2744
2745    #[test]
2746    fn test_float8_coerce_to_int_in_range() {
2747        let plan = build_plan_with_params(vec![
2748            ("$1", DataType::Int8),
2749            ("$2", DataType::Int64),
2750            ("$3", DataType::UInt64),
2751        ]);
2752        let portal = make_portal(
2753            vec![Some(Type::FLOAT8), Some(Type::FLOAT8), Some(Type::FLOAT8)],
2754            vec![s("100"), s("1000000"), s("200")],
2755        );
2756
2757        let values = parameters_to_scalar_values(&plan, &portal).unwrap();
2758        assert_eq!(values[0], ScalarValue::Int8(Some(100)));
2759        assert_eq!(values[1], ScalarValue::Int64(Some(1000000)));
2760        assert_eq!(values[2], ScalarValue::UInt64(Some(200)));
2761    }
2762
2763    #[test]
2764    fn test_float8_coerce_to_int_out_of_range() {
2765        let plan = build_plan_with_params(vec![("$1", DataType::Int8)]);
2766        let portal = make_portal(vec![Some(Type::FLOAT8)], vec![s("200")]);
2767        let result = parameters_to_scalar_values(&plan, &portal);
2768        assert!(result.is_err());
2769    }
2770
2771    #[test]
2772    fn test_float8_coerce_negative_to_unsigned_out_of_range() {
2773        let plan = build_plan_with_params(vec![("$1", DataType::UInt64)]);
2774        let portal = make_portal(vec![Some(Type::FLOAT8)], vec![s("-1")]);
2775        let result = parameters_to_scalar_values(&plan, &portal);
2776        assert!(result.is_err());
2777    }
2778
2779    #[test]
2780    fn test_null_parameter() {
2781        let plan = build_plan_with_params(vec![("$1", DataType::Int8)]);
2782        let portal = make_portal(vec![Some(Type::INT2)], vec![None]);
2783
2784        let values = parameters_to_scalar_values(&plan, &portal).unwrap();
2785        assert_eq!(values[0], ScalarValue::Int8(None));
2786    }
2787
2788    fn numeric_uint64_array_plan() -> LogicalPlan {
2789        build_plan_with_params(vec![(
2790            "$1",
2791            DataType::List(Arc::new(Field::new("item", DataType::UInt64, true))),
2792        )])
2793    }
2794
2795    fn assert_numeric_out_of_range(result: PgWireResult<Vec<ScalarValue>>) {
2796        match result.unwrap_err() {
2797            PgWireError::UserError(error) => {
2798                assert_eq!("22023", error.code);
2799                assert_eq!("numeric_value_out_of_range", error.message);
2800            }
2801            error => panic!("expected numeric out-of-range error, got {error:?}"),
2802        }
2803    }
2804
2805    fn qbs_timestamp_nanosecond_array_plan() -> LogicalPlan {
2806        build_plan_with_params(vec![(
2807            "$1",
2808            DataType::List(Arc::new(Field::new(
2809                "item",
2810                DataType::Timestamp(TimeUnit::Nanosecond, None),
2811                true,
2812            ))),
2813        )])
2814    }
2815
2816    #[test]
2817    fn test_qbs_pg_timestamp_nanosecond_array_preserves_null_slot() {
2818        let portal = make_portal(
2819            vec![Some(Type::TIMESTAMP_ARRAY)],
2820            vec![s(
2821                r#"{"2024-01-01 00:00:00.000001",NULL,"2024-01-01 00:00:00.000003"}"#,
2822            )],
2823        );
2824
2825        let values =
2826            parameters_to_scalar_values(&qbs_timestamp_nanosecond_array_plan(), &portal).unwrap();
2827        let expected = ScalarValue::List(ScalarValue::new_list(
2828            &[
2829                ScalarValue::TimestampNanosecond(Some(1_704_067_200_000_001_000), None),
2830                ScalarValue::TimestampNanosecond(None, None),
2831                ScalarValue::TimestampNanosecond(Some(1_704_067_200_000_003_000), None),
2832            ],
2833            &ArrowDataType::Timestamp(TimeUnit::Nanosecond, None),
2834            true,
2835        ));
2836        assert_eq!(expected, values[0]);
2837    }
2838
2839    #[test]
2840    fn test_qbs_pg_timestamp_nanosecond_array_out_of_range_rejected() {
2841        let portal = make_portal(
2842            vec![Some(Type::TIMESTAMP_ARRAY)],
2843            vec![s(r#"{"3000-01-01 00:00:00"}"#)],
2844        );
2845
2846        assert_numeric_out_of_range(parameters_to_scalar_values(
2847            &qbs_timestamp_nanosecond_array_plan(),
2848            &portal,
2849        ));
2850    }
2851
2852    #[test]
2853    fn test_numeric_uint64_scalar_negative_rejected() {
2854        let plan = build_plan_with_params(vec![("$1", DataType::UInt64)]);
2855        let portal = make_portal(vec![Some(Type::NUMERIC)], vec![s("-1")]);
2856
2857        assert_numeric_out_of_range(parameters_to_scalar_values(&plan, &portal));
2858    }
2859
2860    #[test]
2861    fn test_numeric_uint64_scalar_above_u64_max_rejected() {
2862        let plan = build_plan_with_params(vec![("$1", DataType::UInt64)]);
2863        let portal = make_portal(vec![Some(Type::NUMERIC)], vec![s("18446744073709551616")]);
2864
2865        assert_numeric_out_of_range(parameters_to_scalar_values(&plan, &portal));
2866    }
2867
2868    #[test]
2869    fn test_numeric_uint64_scalar_null_preserved() {
2870        let plan = build_plan_with_params(vec![("$1", DataType::UInt64)]);
2871        let portal = make_portal(vec![Some(Type::NUMERIC)], vec![None]);
2872
2873        let values = parameters_to_scalar_values(&plan, &portal).unwrap();
2874        assert_eq!(ScalarValue::UInt64(None), values[0]);
2875    }
2876
2877    #[test]
2878    fn test_numeric_uint64_scalar_u64_max_preserved() {
2879        let plan = build_plan_with_params(vec![("$1", DataType::UInt64)]);
2880        let portal = make_portal(vec![Some(Type::NUMERIC)], vec![s(&u64::MAX.to_string())]);
2881
2882        let values = parameters_to_scalar_values(&plan, &portal).unwrap();
2883        assert_eq!(ScalarValue::UInt64(Some(u64::MAX)), values[0]);
2884    }
2885
2886    #[test]
2887    fn test_numeric_uint64_array_outer_null_preserved() {
2888        let portal = make_portal(vec![Some(Type::NUMERIC_ARRAY)], vec![None]);
2889
2890        let values = parameters_to_scalar_values(&numeric_uint64_array_plan(), &portal).unwrap();
2891        assert_eq!(ScalarValue::Null, values[0]);
2892    }
2893
2894    #[test]
2895    fn test_numeric_uint64_array_preserves_values_and_null_slots() {
2896        let portal = make_portal(
2897            vec![Some(Type::NUMERIC_ARRAY)],
2898            vec![s("{42,NULL,18446744073709551615}")],
2899        );
2900
2901        let values = parameters_to_scalar_values(&numeric_uint64_array_plan(), &portal).unwrap();
2902        let expected = ScalarValue::List(ScalarValue::new_list(
2903            &[
2904                ScalarValue::UInt64(Some(42)),
2905                ScalarValue::UInt64(None),
2906                ScalarValue::UInt64(Some(u64::MAX)),
2907            ],
2908            &ArrowDataType::UInt64,
2909            true,
2910        ));
2911        assert_eq!(expected, values[0]);
2912    }
2913
2914    #[test]
2915    fn test_numeric_uint64_array_invalid_after_valid_and_null_prefix_rejected() {
2916        let portal = make_portal(vec![Some(Type::NUMERIC_ARRAY)], vec![s("{42,NULL,-1}")]);
2917
2918        assert_numeric_out_of_range(parameters_to_scalar_values(
2919            &numeric_uint64_array_plan(),
2920            &portal,
2921        ));
2922    }
2923
2924    #[test]
2925    fn test_numeric_uint64_array_above_u64_max_rejected() {
2926        let portal = make_portal(
2927            vec![Some(Type::NUMERIC_ARRAY)],
2928            vec![s("{18446744073709551616}")],
2929        );
2930
2931        assert_numeric_out_of_range(parameters_to_scalar_values(
2932            &numeric_uint64_array_plan(),
2933            &portal,
2934        ));
2935    }
2936
2937    #[test]
2938    fn test_numeric_uint64_uninferred_array_invalid_value_rejected() {
2939        let plan = LogicalPlanBuilder::empty(true).build().unwrap();
2940        let portal = make_portal(vec![Some(Type::NUMERIC_ARRAY)], vec![s("{-1}")]);
2941
2942        assert_numeric_out_of_range(parameters_to_scalar_values(&plan, &portal));
2943    }
2944}
servers/postgres/types.rs

servers/postgres/
types.rs
