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

servers/postgres/
types.rs
