Skip to main content

pgwire/api/
results.rs

1use std::fmt::Debug;
2use std::pin::Pin;
3use std::sync::{Arc, LazyLock};
4
5use bytes::{BufMut, Bytes, BytesMut};
6use futures::{Stream, StreamExt, future, stream};
7use postgres_types::{IsNull, Oid, ToSql, Type};
8
9use crate::error::{ErrorInfo, PgWireError, PgWireResult};
10use crate::messages::copy::CopyData;
11use crate::messages::data::{
12    DataRow, FORMAT_CODE_BINARY, FORMAT_CODE_TEXT, FieldDescription, RowDescription,
13};
14use crate::messages::response::CommandComplete;
15use crate::types::ToSqlText;
16use crate::types::format::FormatOptions;
17use smol_str::SmolStr;
18
19/// Command completion tag for a query response.
20#[derive(Debug, Eq, PartialEq, Clone)]
21pub struct Tag {
22    command: String,
23    oid: Option<Oid>,
24    rows: Option<usize>,
25}
26
27impl Tag {
28    /// Create a new tag with the given command name.
29    pub fn new(command: &str) -> Tag {
30        Tag {
31            command: command.to_owned(),
32            oid: None,
33            rows: None,
34        }
35    }
36
37    /// Set the number of rows affected.
38    pub fn with_rows(mut self, rows: usize) -> Tag {
39        self.rows = Some(rows);
40        self
41    }
42
43    /// Set the OID of the inserted row.
44    pub fn with_oid(mut self, oid: Oid) -> Tag {
45        self.oid = Some(oid);
46        self
47    }
48}
49
50impl From<Tag> for CommandComplete {
51    fn from(tag: Tag) -> CommandComplete {
52        let tag_string = if let (Some(oid), Some(rows)) = (tag.oid, tag.rows) {
53            format!("{} {oid} {rows}", tag.command)
54        } else if let Some(rows) = tag.rows {
55            format!("{} {rows}", tag.command)
56        } else {
57            tag.command
58        };
59        CommandComplete::new(tag_string)
60    }
61}
62
63/// Describe encoding of a data field.
64#[derive(Debug, Eq, PartialEq, Clone, Copy)]
65pub enum FieldFormat {
66    Text,
67    Binary,
68}
69
70impl FieldFormat {
71    /// Get format code for the encoding.
72    pub fn value(&self) -> i16 {
73        match self {
74            Self::Text => FORMAT_CODE_TEXT,
75            Self::Binary => FORMAT_CODE_BINARY,
76        }
77    }
78
79    /// Parse FieldFormat from format code.
80    ///
81    /// 0 for text format, 1 for binary format. If the input is neither 0 nor 1,
82    /// here we return text as default value.
83    pub fn from(code: i16) -> Self {
84        if code == FORMAT_CODE_BINARY {
85            FieldFormat::Binary
86        } else {
87            FieldFormat::Text
88        }
89    }
90}
91
92/// Options for COPY text format.
93#[derive(Debug, Clone, Eq, PartialEq)]
94pub struct CopyTextOptions {
95    pub delimiter: SmolStr,
96    pub null_string: SmolStr,
97}
98
99impl Default for CopyTextOptions {
100    fn default() -> Self {
101        Self {
102            delimiter: "\t".into(),
103            null_string: "\\N".into(),
104        }
105    }
106}
107
108/// Options for COPY CSV format.
109#[derive(Debug, Clone, Eq, PartialEq)]
110pub struct CopyCsvOptions {
111    pub delimiter: SmolStr,
112    pub quote: SmolStr,
113    pub escape: SmolStr,
114    pub null_string: SmolStr,
115    pub force_quote: Vec<usize>,
116}
117
118impl Default for CopyCsvOptions {
119    fn default() -> Self {
120        Self {
121            delimiter: ",".into(),
122            quote: "\"".into(),
123            escape: "\"".into(),
124            null_string: "".into(),
125            force_quote: vec![],
126        }
127    }
128}
129
130// Default format options that are cloned in `FieldInfo::new` to avoid `Arc` allocation.
131//
132// Using thread-local storage avoids contention when multiple threads concurrently
133// clone the same `Arc<FormatOptions>` in `DataRowEncoder::encode_field`. Each thread
134// now clones its own thread-local instance rather than contending for a shared
135// global instance.
136//
137// This can be made a regular static if we remove format options cloning from
138// `DataRowEncoder::encode_field`.
139//
140// The issue with contention was observed in `examples/bench` benchmark:
141// https://github.com/sunng87/pgwire/pull/366#discussion_r2621917771
142thread_local! {
143    static DEFAULT_FORMAT_OPTIONS: LazyLock<Arc<FormatOptions>> = LazyLock::new(Default::default);
144}
145
146/// Metadata for a single field (column) in a query result.
147#[derive(Debug, new, Eq, PartialEq, Clone)]
148pub struct FieldInfo {
149    name: String,
150    table_id: Option<i32>,
151    column_id: Option<i16>,
152    datatype: Type,
153    format: FieldFormat,
154    #[new(value = "DEFAULT_FORMAT_OPTIONS.with(|opts| Arc::clone(&*opts))")]
155    format_options: Arc<FormatOptions>,
156    /// `pg_type.typlen`: the type's fixed byte width, or a negative value for a
157    /// variable-width type. Defaults to 0 ("unspecified"); set it with
158    /// [`FieldInfo::with_type_size`] to describe the value bytes precisely.
159    #[new(value = "0")]
160    type_size: i16,
161    /// `pg_attribute.atttypmod`: the type-specific modifier, such as the
162    /// precision and scale packed into a `numeric(p, s)` or the declared length
163    /// of a `varchar(n)`.
164    ///
165    /// Defaults to `-1`, PostgreSQL's encoding for "no modifier". This matters:
166    /// clients derive display scale from it, and a `0` here is not "unset" but a
167    /// valid-looking modifier that decodes to a nonsense scale — a `numeric`
168    /// then renders with a long tail of spurious zeros in JDBC-based clients.
169    #[new(value = "-1")]
170    type_modifier: i32,
171}
172
173impl FieldInfo {
174    /// Get the field name.
175    pub fn name(&self) -> &str {
176        &self.name
177    }
178
179    /// Get the source table OID, if any.
180    pub fn table_id(&self) -> Option<i32> {
181        self.table_id
182    }
183
184    /// Get the column number within the source table, if any.
185    pub fn column_id(&self) -> Option<i16> {
186        self.column_id
187    }
188
189    /// Get the PostgreSQL type of this field.
190    pub fn datatype(&self) -> &Type {
191        &self.datatype
192    }
193
194    /// Get the field encoding format (text or binary).
195    pub fn format(&self) -> FieldFormat {
196        self.format
197    }
198
199    /// Get the format options for text encoding.
200    pub fn format_options(&self) -> &Arc<FormatOptions> {
201        &self.format_options
202    }
203
204    /// Set custom format options for text encoding.
205    pub fn with_format_options(mut self, format_options: Arc<FormatOptions>) -> Self {
206        self.format_options = format_options;
207        self
208    }
209
210    /// Get the type size (`pg_type.typlen`).
211    pub fn type_size(&self) -> i16 {
212        self.type_size
213    }
214
215    /// Set the type size (`pg_type.typlen`).
216    pub fn with_type_size(mut self, type_size: i16) -> Self {
217        self.type_size = type_size;
218        self
219    }
220
221    /// Get the type modifier (`pg_attribute.atttypmod`).
222    pub fn type_modifier(&self) -> i32 {
223        self.type_modifier
224    }
225
226    /// Set the type modifier (`pg_attribute.atttypmod`); `-1` means none.
227    pub fn with_type_modifier(mut self, type_modifier: i32) -> Self {
228        self.type_modifier = type_modifier;
229        self
230    }
231}
232
233impl From<&FieldInfo> for FieldDescription {
234    fn from(fi: &FieldInfo) -> Self {
235        FieldDescription::new(
236            fi.name.clone(),           // name
237            fi.table_id.unwrap_or(0),  // table_id
238            fi.column_id.unwrap_or(0), // column_id
239            fi.datatype.oid(),         // type_id
240            fi.type_size,              // type_size
241            fi.type_modifier,          // type_modifier
242            fi.format.value(),
243        )
244    }
245}
246
247impl From<FieldDescription> for FieldInfo {
248    fn from(value: FieldDescription) -> Self {
249        FieldInfo::new(
250            value.name,
251            Some(value.table_id),
252            Some(value.column_id),
253            Type::from_oid(value.type_id).unwrap_or(Type::UNKNOWN),
254            FieldFormat::from(value.format_code),
255        )
256        .with_type_size(value.type_size)
257        .with_type_modifier(value.type_modifier)
258    }
259}
260
261pub(crate) fn into_row_description(fields: &[FieldInfo]) -> RowDescription {
262    RowDescription::new(fields.iter().map(Into::into).collect())
263}
264
265/// Type alias for a boxed, pinned, sendable stream of data rows.
266pub type SendableRowStream = Pin<Box<dyn Stream<Item = PgWireResult<DataRow>> + Send>>;
267
268/// Type alias for a boxed, pinned, sendable stream of copy data.
269pub type SendableCopyDataStream = Pin<Box<dyn Stream<Item = PgWireResult<CopyData>> + Send>>;
270
271/// Response containing row data for a SELECT-style query.
272#[non_exhaustive]
273pub struct QueryResponse {
274    pub command_tag: String,
275    pub row_schema: Arc<Vec<FieldInfo>>,
276    pub data_rows: SendableRowStream,
277}
278
279impl Debug for QueryResponse {
280    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
281        f.debug_struct("QueryResponse")
282            .field("command_tag", &self.command_tag)
283            .field("row_schema", &self.row_schema)
284            .finish()
285    }
286}
287
288impl QueryResponse {
289    /// Create `QueryResponse` from column schemas and stream of data row.
290    /// Sets "SELECT" as the command tag.
291    pub fn new<S>(field_defs: Arc<Vec<FieldInfo>>, row_stream: S) -> QueryResponse
292    where
293        S: Stream<Item = PgWireResult<DataRow>> + Send + 'static,
294    {
295        QueryResponse {
296            command_tag: "SELECT".to_owned(),
297            row_schema: field_defs,
298            data_rows: Box::pin(row_stream),
299        }
300    }
301
302    /// Get the command tag
303    pub fn command_tag(&self) -> &str {
304        &self.command_tag
305    }
306
307    /// Set the command tag
308    pub fn set_command_tag(&mut self, command_tag: &str) {
309        command_tag.clone_into(&mut self.command_tag);
310    }
311
312    /// Get schema of columns
313    pub fn row_schema(&self) -> Arc<Vec<FieldInfo>> {
314        self.row_schema.clone()
315    }
316
317    /// Get access to data rows stream
318    pub fn data_rows(&mut self) -> &mut SendableRowStream {
319        &mut self.data_rows
320    }
321}
322
323/// Encoder for building `DataRow` messages field by field.
324pub struct DataRowEncoder {
325    schema: Arc<Vec<FieldInfo>>,
326    row_buffer: BytesMut,
327    col_index: usize,
328}
329
330impl DataRowEncoder {
331    /// New DataRowEncoder from schema of column
332    pub fn new(fields: Arc<Vec<FieldInfo>>) -> DataRowEncoder {
333        Self {
334            schema: fields,
335            row_buffer: BytesMut::with_capacity(128),
336            col_index: 0,
337        }
338    }
339
340    /// Encode value with custom type and format
341    ///
342    /// This encode function ignores data type and format information from
343    /// schema of this encoder.
344    pub fn encode_field_with_type_and_format<T>(
345        &mut self,
346        value: &T,
347        data_type: &Type,
348        format: FieldFormat,
349        format_options: &FormatOptions,
350    ) -> PgWireResult<()>
351    where
352        T: ToSql + ToSqlText + Sized,
353    {
354        // remember the position of the 4-byte length field
355        let prev_index = self.row_buffer.len();
356        // write value length as -1 ahead of time
357        self.row_buffer.put_i32(-1);
358
359        let is_null = if format == FieldFormat::Text {
360            value.to_sql_text(data_type, &mut self.row_buffer, format_options)?
361        } else {
362            value.to_sql(data_type, &mut self.row_buffer)?
363        };
364
365        if let IsNull::No = is_null {
366            let value_length = self.row_buffer.len() - prev_index - 4;
367            let mut length_bytes = &mut self.row_buffer[prev_index..(prev_index + 4)];
368            length_bytes.put_i32(value_length as i32);
369        }
370
371        self.col_index += 1;
372
373        Ok(())
374    }
375
376    /// Encode value using type and format, defined by schema
377    ///
378    /// Panic when encoding more columns than provided as schema.
379    pub fn encode_field<T>(&mut self, value: &T) -> PgWireResult<()>
380    where
381        T: ToSql + ToSqlText + Sized,
382    {
383        let field = &self.schema[self.col_index];
384
385        let data_type = field.datatype().clone();
386        let format = field.format();
387        let format_options = field.format_options().clone();
388
389        self.encode_field_with_type_and_format(value, &data_type, format, format_options.as_ref())
390    }
391
392    #[deprecated(
393        since = "0.37.0",
394        note = "DataRowEncoder is reusable since 0.37, use `take_row() instead`"
395    )]
396    pub fn finish(self) -> PgWireResult<DataRow> {
397        Ok(DataRow::new(self.row_buffer, self.col_index as i16))
398    }
399
400    /// Takes the current row from the encoder, resetting the encoder for reuse.
401    ///
402    /// This method splits the inner buffer, taking the current row data and leaving the
403    /// encoder with an empty buffer (but retaining the capacity) enabling buffer reuse.
404    pub fn take_row(&mut self) -> DataRow {
405        let row = DataRow::new(self.row_buffer.split(), self.col_index as i16);
406        self.col_index = 0;
407        row
408    }
409}
410
411/// Internal COPY format representation.
412#[derive(Debug, Clone, Eq, PartialEq)]
413enum CopyFormat {
414    Binary,
415    Text {
416        delimiter: SmolStr,
417        null_string: SmolStr,
418    },
419    Csv {
420        delimiter: SmolStr,
421        quote: SmolStr,
422        escape: SmolStr,
423        null_string: SmolStr,
424        force_quote: Vec<usize>,
425    },
426}
427
428/// Encoder for COPY operations.
429///
430/// This encoder produces CopyData messages for PGCOPY binary, text, and CSV formats.
431pub struct CopyEncoder {
432    schema: Arc<Vec<FieldInfo>>,
433    buffer: BytesMut,
434    format: CopyFormat,
435    col_index: usize,
436    header_written: bool,
437}
438
439impl CopyEncoder {
440    /// Create a new binary format COPY encoder.
441    pub fn new_binary(schema: Arc<Vec<FieldInfo>>) -> Self {
442        Self {
443            schema,
444            buffer: BytesMut::with_capacity(128),
445            format: CopyFormat::Binary,
446            col_index: 0,
447            header_written: false,
448        }
449    }
450
451    /// Create a new text format COPY encoder.
452    pub fn new_text(schema: Arc<Vec<FieldInfo>>, options: CopyTextOptions) -> Self {
453        Self {
454            schema,
455            buffer: BytesMut::with_capacity(128),
456            format: CopyFormat::Text {
457                delimiter: options.delimiter,
458                null_string: options.null_string,
459            },
460            col_index: 0,
461            header_written: false,
462        }
463    }
464
465    /// Create a new CSV format COPY encoder.
466    pub fn new_csv(schema: Arc<Vec<FieldInfo>>, options: CopyCsvOptions) -> Self {
467        Self {
468            schema,
469            buffer: BytesMut::with_capacity(128),
470            format: CopyFormat::Csv {
471                delimiter: options.delimiter,
472                quote: options.quote,
473                escape: options.escape,
474                null_string: options.null_string,
475                force_quote: options.force_quote,
476            },
477            col_index: 0,
478            header_written: false,
479        }
480    }
481
482    /// Encode a field value.
483    ///
484    /// This method uses the type and format information from the schema.
485    pub fn encode_field<T>(&mut self, value: &T) -> PgWireResult<()>
486    where
487        T: ToSql + ToSqlText + Sized,
488    {
489        let datatype = self.schema[self.col_index].datatype().clone();
490        let col_index = self.col_index;
491        let num_fields = self.schema.len();
492
493        match &self.format {
494            CopyFormat::Binary => self.encode_field_binary(value, &datatype)?,
495            CopyFormat::Text { .. } => {
496                let is_last = col_index == num_fields - 1;
497                self.encode_field_text(value, &datatype, is_last)?;
498            }
499            CopyFormat::Csv { .. } => {
500                let is_last = col_index == num_fields - 1;
501                self.encode_field_csv(value, &datatype, is_last)?;
502            }
503        }
504
505        self.col_index += 1;
506        Ok(())
507    }
508
509    /// Encode a field in binary format (same as DataRow encoding).
510    fn encode_field_binary<T>(&mut self, value: &T, datatype: &Type) -> PgWireResult<()>
511    where
512        T: ToSql + ToSqlText,
513    {
514        let prev_index = self.buffer.len();
515        self.buffer.put_i32(-1);
516
517        let is_null = value.to_sql(datatype, &mut self.buffer)?;
518
519        if let IsNull::No = is_null {
520            let value_length = self.buffer.len() - prev_index - 4;
521            let mut length_bytes = &mut self.buffer[prev_index..(prev_index + 4)];
522            length_bytes.put_i32(value_length as i32);
523        }
524
525        Ok(())
526    }
527
528    /// Encode a field in text format.
529    fn encode_field_text<T>(
530        &mut self,
531        value: &T,
532        datatype: &Type,
533        is_last: bool,
534    ) -> PgWireResult<()>
535    where
536        T: ToSqlText,
537    {
538        if let CopyFormat::Text {
539            delimiter,
540            null_string,
541        } = &self.format
542        {
543            let mut temp_buffer = BytesMut::new();
544            let is_null =
545                value.to_sql_text(datatype, &mut temp_buffer, &FormatOptions::default())?;
546
547            if let IsNull::Yes = is_null {
548                self.buffer.put_slice(null_string.as_bytes());
549            } else {
550                // Backslash escape special characters
551                for &byte in temp_buffer.as_ref() {
552                    match byte {
553                        b'\n' => {
554                            self.buffer.put_slice(b"\\n");
555                        }
556                        b'\r' => {
557                            self.buffer.put_slice(b"\\r");
558                        }
559                        b'\t' => {
560                            self.buffer.put_slice(b"\\t");
561                        }
562                        b'\\' => {
563                            self.buffer.put_slice(b"\\\\");
564                        }
565                        _b if byte == delimiter.as_bytes()[0] => {
566                            self.buffer.put_u8(b'\\');
567                            self.buffer.put_u8(byte);
568                        }
569                        _ => {
570                            self.buffer.put_u8(byte);
571                        }
572                    }
573                }
574            }
575
576            // Add delimiter between fields
577            if !is_last {
578                self.buffer.put_slice(delimiter.as_bytes());
579            }
580
581            Ok(())
582        } else {
583            Err(PgWireError::IoError(std::io::Error::new(
584                std::io::ErrorKind::InvalidInput,
585                "Text format expected",
586            )))
587        }
588    }
589
590    /// Encode a field in CSV format.
591    fn encode_field_csv<T>(&mut self, value: &T, datatype: &Type, is_last: bool) -> PgWireResult<()>
592    where
593        T: ToSqlText,
594    {
595        if let CopyFormat::Csv {
596            delimiter,
597            quote,
598            null_string,
599            force_quote,
600            escape: _,
601        } = &self.format
602        {
603            let col_index = self.col_index;
604            let mut temp_buffer = BytesMut::new();
605            let is_null =
606                value.to_sql_text(datatype, &mut temp_buffer, &FormatOptions::default())?;
607
608            let delimiter_byte = delimiter.as_bytes()[0];
609            let quote_byte = quote.as_bytes()[0];
610            let null_string_bytes = null_string.as_bytes();
611
612            let should_quote = force_quote.contains(&col_index)
613                || match is_null {
614                    IsNull::Yes => false, // NULL values are never quoted in CSV (handled by null_string)
615                    IsNull::No => {
616                        let data = temp_buffer.as_ref();
617                        data.contains(&delimiter_byte)
618                            || data.contains(&quote_byte)
619                            || data.contains(&b'\n')
620                            || data.contains(&b'\r')
621                            || (!null_string_bytes.is_empty()
622                                && data
623                                    .windows(null_string_bytes.len())
624                                    .any(|w| w == null_string_bytes))
625                    }
626                };
627
628            if let IsNull::Yes = is_null {
629                self.buffer.put_slice(null_string_bytes);
630            } else if should_quote {
631                self.buffer.put_u8(quote_byte);
632
633                for &byte in temp_buffer.as_ref() {
634                    if byte == quote_byte {
635                        // Double the quote character
636                        self.buffer.put_u8(byte);
637                    }
638                    self.buffer.put_u8(byte);
639                }
640
641                self.buffer.put_u8(quote_byte);
642            } else {
643                self.buffer.put_slice(temp_buffer.as_ref());
644            }
645
646            // Add delimiter between fields
647            if !is_last {
648                self.buffer.put_slice(delimiter.as_bytes());
649            }
650
651            Ok(())
652        } else {
653            Err(PgWireError::IoError(std::io::Error::new(
654                std::io::ErrorKind::InvalidInput,
655                "CSV format expected",
656            )))
657        }
658    }
659
660    /// Take the current row as a CopyData message.
661    ///
662    /// For binary format: first call includes PGCOPY header.
663    /// For text/CSV format: each call returns one row with a trailing newline.
664    pub fn take_copy(&mut self) -> CopyData {
665        match &self.format {
666            CopyFormat::Binary => {
667                if !self.header_written {
668                    // Prepend header to field data
669                    let field_data = self.buffer.split();
670                    self.write_pgcop_header();
671                    self.buffer.put_i16(self.schema.len() as i16);
672                    self.buffer.extend_from_slice(&field_data);
673                    self.header_written = true;
674                } else {
675                    // Prepend field count before field data
676                    let field_data = self.buffer.split();
677                    self.buffer.put_i16(self.schema.len() as i16);
678                    self.buffer.extend_from_slice(&field_data);
679                }
680            }
681            CopyFormat::Text { .. } | CopyFormat::Csv { .. } => {
682                // Add newline at end of row
683                self.buffer.put_u8(b'\n');
684            }
685        }
686
687        self.col_index = 0;
688        CopyData::new(self.buffer.split().freeze())
689    }
690
691    /// Finish the COPY operation of binary format.
692    ///
693    /// For binary format: returns trailer (-1).
694    /// Note that this trailer is automatically appended to stream if you use
695    /// `CopyResponse` API.
696    pub fn finish_copy_binary() -> CopyData {
697        CopyData::new(Bytes::from_static(&[0xFF, 0xFF]))
698    }
699
700    /// Write PGCOPY binary header.
701    fn write_pgcop_header(&mut self) {
702        self.buffer.put_slice(b"PGCOPY\n\xFF\r\n\x00");
703        self.buffer.put_i32(0); // Flags (no OIDs)
704        self.buffer.put_i32(0); // Header extension length
705    }
706}
707
708/// Get response data for a `Describe` command
709pub trait DescribeResponse {
710    /// Get parameter types for the described statement.
711    fn parameters(&self) -> Option<&[Type]>;
712
713    /// Get result field descriptions.
714    fn fields(&self) -> &[FieldInfo];
715
716    /// Create an no_data instance of `DescribeResponse`. This is typically used
717    /// when client tries to describe an empty query.
718    fn no_data() -> Self;
719
720    /// Return true if the `DescribeResponse` is empty/nodata
721    fn is_no_data(&self) -> bool;
722}
723
724/// Response for frontend describe statement requests.
725#[non_exhaustive]
726#[derive(Debug, new)]
727pub struct DescribeStatementResponse {
728    pub parameters: Vec<Type>,
729    pub fields: Vec<FieldInfo>,
730}
731
732impl DescribeResponse for DescribeStatementResponse {
733    fn parameters(&self) -> Option<&[Type]> {
734        Some(self.parameters.as_ref())
735    }
736
737    fn fields(&self) -> &[FieldInfo] {
738        &self.fields
739    }
740
741    /// Create an no_data instance of `DescribeStatementResponse`. This is typically used
742    /// when client tries to describe an empty query.
743    fn no_data() -> Self {
744        DescribeStatementResponse {
745            parameters: vec![],
746            fields: vec![],
747        }
748    }
749
750    /// Return true if the `DescribeStatementResponse` is empty/nodata
751    fn is_no_data(&self) -> bool {
752        self.parameters.is_empty() && self.fields.is_empty()
753    }
754}
755
756/// Response for frontend describe portal requests.
757#[non_exhaustive]
758#[derive(Debug, new)]
759pub struct DescribePortalResponse {
760    pub fields: Vec<FieldInfo>,
761}
762
763impl DescribeResponse for DescribePortalResponse {
764    fn parameters(&self) -> Option<&[Type]> {
765        None
766    }
767
768    fn fields(&self) -> &[FieldInfo] {
769        &self.fields
770    }
771
772    /// Create an no_data instance of `DescribePortalResponse`. This is typically used
773    /// when client tries to describe an empty query.
774    fn no_data() -> Self {
775        DescribePortalResponse { fields: vec![] }
776    }
777
778    /// Return true if the `DescribePortalResponse` is empty/nodata
779    fn is_no_data(&self) -> bool {
780        self.fields.is_empty()
781    }
782}
783
784/// Response for copy operations
785#[non_exhaustive]
786pub struct CopyResponse {
787    pub format: i8,
788    pub columns: usize,
789    pub data_stream: SendableCopyDataStream,
790}
791
792impl std::fmt::Debug for CopyResponse {
793    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
794        f.debug_struct("CopyResponse")
795            .field("format", &self.format)
796            .field("columns", &self.columns)
797            .finish()
798    }
799}
800
801impl CopyResponse {
802    /// Create a new copy response. Binary format automatically appends a trailer.
803    pub fn new<S>(format: i8, columns: usize, data_stream: S) -> CopyResponse
804    where
805        S: Stream<Item = PgWireResult<CopyData>> + Send + 'static,
806    {
807        if format == 1 {
808            let data_stream = data_stream.chain(stream::once(future::ready(Ok(
809                CopyEncoder::finish_copy_binary(),
810            ))));
811            CopyResponse {
812                format,
813                columns,
814                data_stream: Box::pin(data_stream),
815            }
816        } else {
817            CopyResponse {
818                format,
819                columns,
820                data_stream: Box::pin(data_stream),
821            }
822        }
823    }
824
825    /// Get mutable access to the underlying copy data stream.
826    pub fn data_stream(&mut self) -> &mut SendableCopyDataStream {
827        &mut self.data_stream
828    }
829
830    /// Get the format code for each column.
831    pub fn column_formats(&self) -> Vec<i16> {
832        (0..self.columns).map(|_| self.format as i16).collect()
833    }
834}
835
836/// Query response types:
837///
838/// * Query: the response contains data rows
839/// * Execution: response for ddl/dml execution
840/// * Error: error response
841/// * EmptyQuery: when client sends an empty query
842/// * TransactionStart: indicate previous statement just started a transaction
843/// * TransactionEnd: indicate previous statement just ended a transaction
844/// * CopyIn: response for a copy-in request
845/// * CopyOut: response for a copy-out request
846/// * CopuBoth: response for a copy-both request
847#[derive(Debug)]
848pub enum Response {
849    EmptyQuery,
850    Query(QueryResponse),
851    Execution(Tag),
852    TransactionStart(Tag),
853    TransactionEnd(Tag),
854    Error(Box<ErrorInfo>),
855    CopyIn(CopyResponse),
856    CopyOut(CopyResponse),
857    CopyBoth(CopyResponse),
858}
859
860#[cfg(test)]
861mod test {
862
863    use super::*;
864
865    #[test]
866    fn field_info_defaults_to_no_type_modifier() {
867        // `0` is not "unset" on the wire — it decodes to a nonsense scale and
868        // makes JDBC clients render numerics with spurious trailing zeros.
869        let fi = FieldInfo::new("c".into(), None, None, Type::NUMERIC, FieldFormat::Text);
870        assert_eq!(fi.type_modifier(), -1);
871        assert_eq!(FieldDescription::from(&fi).type_modifier, -1);
872    }
873
874    #[test]
875    fn field_info_type_size_and_modifier_reach_the_field_description() {
876        // numeric(18, 4) packs as ((18 << 16) | 4) + 4.
877        let typmod = ((18 << 16) | 4) + 4;
878        let fi = FieldInfo::new(
879            "amount".into(),
880            None,
881            None,
882            Type::NUMERIC,
883            FieldFormat::Text,
884        )
885        .with_type_size(-1)
886        .with_type_modifier(typmod);
887        let fd = FieldDescription::from(&fi);
888        assert_eq!(fd.type_size, -1);
889        assert_eq!(fd.type_modifier, typmod);
890
891        // And they survive the round trip back into a `FieldInfo`.
892        let back = FieldInfo::from(fd);
893        assert_eq!(back.type_size(), -1);
894        assert_eq!(back.type_modifier(), typmod);
895    }
896
897    #[test]
898    fn test_command_complete() {
899        let tag = Tag::new("INSERT").with_rows(100);
900        let cc = CommandComplete::from(tag);
901
902        assert_eq!(cc.tag, "INSERT 100");
903
904        let tag = Tag::new("INSERT").with_oid(0).with_rows(100);
905        let cc = CommandComplete::from(tag);
906
907        assert_eq!(cc.tag, "INSERT 0 100");
908    }
909
910    #[test]
911    #[cfg(feature = "pg-type-chrono")]
912    fn test_data_row_encoder() {
913        use std::time::SystemTime;
914
915        let schema = Arc::new(vec![
916            FieldInfo::new("id".into(), None, None, Type::INT4, FieldFormat::Text),
917            FieldInfo::new("name".into(), None, None, Type::VARCHAR, FieldFormat::Text),
918            FieldInfo::new("ts".into(), None, None, Type::TIMESTAMP, FieldFormat::Text),
919        ]);
920        let now = SystemTime::now();
921        let mut encoder = DataRowEncoder::new(schema);
922        encoder.encode_field(&2001).unwrap();
923        encoder.encode_field(&"udev").unwrap();
924        encoder.encode_field(&now).unwrap();
925
926        let row = encoder.take_row();
927
928        assert_eq!(row.field_count, 3);
929
930        let mut expected = BytesMut::new();
931        expected.put_i32(4);
932        expected.put_slice("2001".as_bytes());
933        expected.put_i32(4);
934        expected.put_slice("udev".as_bytes());
935        expected.put_i32(26);
936        let _ = now.to_sql_text(&Type::TIMESTAMP, &mut expected, &FormatOptions::default());
937        assert_eq!(row.data, expected);
938    }
939
940    #[test]
941    fn test_copy_text_options_default() {
942        let opts = CopyTextOptions::default();
943        assert_eq!(opts.delimiter, "\t");
944        assert_eq!(opts.null_string, "\\N");
945    }
946
947    #[test]
948    fn test_copy_csv_options_default() {
949        let opts = CopyCsvOptions::default();
950        assert_eq!(opts.delimiter, ",");
951        assert_eq!(opts.quote, "\"");
952        assert_eq!(opts.escape, "\"");
953        assert_eq!(opts.null_string, "");
954        assert!(opts.force_quote.is_empty());
955    }
956
957    #[test]
958    fn test_copy_binary_header() {
959        let schema = Arc::new(vec![FieldInfo::new(
960            "id".into(),
961            None,
962            None,
963            Type::INT4,
964            FieldFormat::Binary,
965        )]);
966        let mut encoder = CopyEncoder::new_binary(schema.clone());
967
968        // First take_copy should include header
969        encoder.encode_field(&42).unwrap();
970        let copy_data = encoder.take_copy();
971
972        let data = copy_data.data.as_ref();
973        assert_eq!(&data[0..11], b"PGCOPY\n\xFF\r\n\0");
974
975        // Check flags (4 bytes, no OIDs = 0)
976        assert_eq!(&data[11..15], &[0x00, 0x00, 0x00, 0x00]);
977
978        // Check extension length (4 bytes, no extensions = 0)
979        assert_eq!(&data[15..19], &[0x00, 0x00, 0x00, 0x00]);
980
981        // Check field count (2 bytes)
982        assert_eq!(&data[19..21], &[0x00, 0x01]); // 1 field
983
984        // Check field length (4 bytes)
985        assert_eq!(&data[21..25], &[0x00, 0x00, 0x00, 0x04]); // 4 bytes
986
987        // Check field value (42 in network byte order)
988        assert_eq!(&data[25..29], &[0x00, 0x00, 0x00, 0x2A]);
989    }
990
991    #[test]
992    fn test_copy_binary_trailer() {
993        let copy_data = CopyEncoder::finish_copy_binary();
994        let data = copy_data.data.as_ref();
995
996        // Trailer is -1 as i16 (0xFFFF in network byte order)
997        assert_eq!(data, &[0xFF, 0xFF]);
998    }
999
1000    #[test]
1001    fn test_copy_text_default_delimiter() {
1002        let schema = Arc::new(vec![
1003            FieldInfo::new("id".into(), None, None, Type::INT4, FieldFormat::Text),
1004            FieldInfo::new("name".into(), None, None, Type::VARCHAR, FieldFormat::Text),
1005        ]);
1006        let mut encoder = CopyEncoder::new_text(schema, CopyTextOptions::default());
1007
1008        encoder.encode_field(&1).unwrap();
1009        encoder.encode_field(&"Alice").unwrap();
1010        let copy_data = encoder.take_copy();
1011
1012        // Expected: "1\tAlice\n"
1013        assert_eq!(copy_data.data.as_ref(), b"1\tAlice\n");
1014    }
1015
1016    #[test]
1017    fn test_copy_text_custom_delimiter() {
1018        let schema = Arc::new(vec![
1019            FieldInfo::new("id".into(), None, None, Type::INT4, FieldFormat::Text),
1020            FieldInfo::new("name".into(), None, None, Type::VARCHAR, FieldFormat::Text),
1021        ]);
1022        let mut encoder = CopyEncoder::new_text(
1023            schema,
1024            CopyTextOptions {
1025                delimiter: "|".into(),
1026                null_string: "\\N".into(),
1027            },
1028        );
1029
1030        encoder.encode_field(&1).unwrap();
1031        encoder.encode_field(&"Alice").unwrap();
1032        let copy_data = encoder.take_copy();
1033
1034        // Expected: "1|Alice\n"
1035        assert_eq!(copy_data.data.as_ref(), b"1|Alice\n");
1036    }
1037
1038    #[test]
1039    fn test_copy_text_null_handling() {
1040        let schema = Arc::new(vec![
1041            FieldInfo::new("id".into(), None, None, Type::INT4, FieldFormat::Text),
1042            FieldInfo::new("name".into(), None, None, Type::VARCHAR, FieldFormat::Text),
1043        ]);
1044        let mut encoder = CopyEncoder::new_text(schema, CopyTextOptions::default());
1045
1046        encoder.encode_field(&1).unwrap();
1047        encoder.encode_field(&None::<String>).unwrap();
1048        let copy_data = encoder.take_copy();
1049
1050        // Expected: "1\t\\N\n"
1051        assert_eq!(copy_data.data.as_ref(), b"1\t\\N\n");
1052    }
1053
1054    #[test]
1055    fn test_copy_text_backslash_escaping() {
1056        let schema = Arc::new(vec![FieldInfo::new(
1057            "value".into(),
1058            None,
1059            None,
1060            Type::VARCHAR,
1061            FieldFormat::Text,
1062        )]);
1063        let mut encoder = CopyEncoder::new_text(schema, CopyTextOptions::default());
1064
1065        encoder.encode_field(&"a\nb\tc\rd\\e").unwrap();
1066        let copy_data = encoder.take_copy();
1067
1068        // Expected: "a\\nb\\tc\\rd\\\\e\n"
1069        assert_eq!(copy_data.data.as_ref(), b"a\\nb\\tc\\rd\\\\e\n");
1070    }
1071
1072    #[test]
1073    fn test_copy_csv_default() {
1074        let schema = Arc::new(vec![
1075            FieldInfo::new("id".into(), None, None, Type::INT4, FieldFormat::Text),
1076            FieldInfo::new("name".into(), None, None, Type::VARCHAR, FieldFormat::Text),
1077        ]);
1078        let mut encoder = CopyEncoder::new_csv(schema, CopyCsvOptions::default());
1079
1080        encoder.encode_field(&1).unwrap();
1081        encoder.encode_field(&"Alice").unwrap();
1082        let copy_data = encoder.take_copy();
1083
1084        // Expected: "1,Alice\n"
1085        assert_eq!(copy_data.data.as_ref(), b"1,Alice\n");
1086    }
1087
1088    #[test]
1089    fn test_copy_csv_quoting() {
1090        let schema = Arc::new(vec![FieldInfo::new(
1091            "value".into(),
1092            None,
1093            None,
1094            Type::VARCHAR,
1095            FieldFormat::Text,
1096        )]);
1097        let mut encoder = CopyEncoder::new_csv(schema, CopyCsvOptions::default());
1098
1099        encoder.encode_field(&"a,b\"c\nd").unwrap();
1100        let copy_data = encoder.take_copy();
1101
1102        // Should be quoted because it contains comma and newline
1103        assert_eq!(copy_data.data.as_ref(), b"\"a,b\"\"c\nd\"\n");
1104    }
1105
1106    #[test]
1107    fn test_copy_csv_force_quote() {
1108        let schema = Arc::new(vec![
1109            FieldInfo::new("id".into(), None, None, Type::INT4, FieldFormat::Text),
1110            FieldInfo::new("name".into(), None, None, Type::VARCHAR, FieldFormat::Text),
1111        ]);
1112        let mut encoder = CopyEncoder::new_csv(
1113            schema,
1114            CopyCsvOptions {
1115                force_quote: vec![1],
1116                ..Default::default()
1117            },
1118        );
1119
1120        encoder.encode_field(&1).unwrap();
1121        encoder.encode_field(&"Alice").unwrap();
1122        let copy_data = encoder.take_copy();
1123
1124        // Expected: "1,\"Alice\"\n" - second column force quoted
1125        assert_eq!(copy_data.data.as_ref(), b"1,\"Alice\"\n");
1126    }
1127
1128    #[test]
1129    fn test_copy_binary_multiple_rows() {
1130        let schema = Arc::new(vec![
1131            FieldInfo::new("id".into(), None, None, Type::INT4, FieldFormat::Binary),
1132            FieldInfo::new(
1133                "name".into(),
1134                None,
1135                None,
1136                Type::VARCHAR,
1137                FieldFormat::Binary,
1138            ),
1139        ]);
1140        let mut encoder = CopyEncoder::new_binary(schema);
1141
1142        // First row
1143        encoder.encode_field(&1i32).unwrap();
1144        encoder.encode_field(&"Alice".to_string()).unwrap();
1145        let copy_data1 = encoder.take_copy();
1146
1147        // Second row
1148        encoder.encode_field(&2i32).unwrap();
1149        encoder.encode_field(&"Bob".to_string()).unwrap();
1150        let copy_data2 = encoder.take_copy();
1151
1152        // Verify first row format
1153        let data1 = copy_data1.data.as_ref();
1154
1155        // Header is 19 bytes, then field count (2 bytes)
1156        assert_eq!(&data1[19..21], &[0x00, 0x02]); // 2 fields
1157
1158        // Verify second row format
1159        let data2 = copy_data2.data.as_ref();
1160
1161        // Field count should be at the beginning (no header on second row)
1162        assert_eq!(&data2[0..2], &[0x00, 0x02]); // 2 fields
1163    }
1164}