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#[derive(Debug, Eq, PartialEq, Clone)]
21pub struct Tag {
22 command: String,
23 oid: Option<Oid>,
24 rows: Option<usize>,
25}
26
27impl Tag {
28 pub fn new(command: &str) -> Tag {
30 Tag {
31 command: command.to_owned(),
32 oid: None,
33 rows: None,
34 }
35 }
36
37 pub fn with_rows(mut self, rows: usize) -> Tag {
39 self.rows = Some(rows);
40 self
41 }
42
43 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#[derive(Debug, Eq, PartialEq, Clone, Copy)]
65pub enum FieldFormat {
66 Text,
67 Binary,
68}
69
70impl FieldFormat {
71 pub fn value(&self) -> i16 {
73 match self {
74 Self::Text => FORMAT_CODE_TEXT,
75 Self::Binary => FORMAT_CODE_BINARY,
76 }
77 }
78
79 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#[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#[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
130thread_local! {
143 static DEFAULT_FORMAT_OPTIONS: LazyLock<Arc<FormatOptions>> = LazyLock::new(Default::default);
144}
145
146#[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 #[new(value = "0")]
160 type_size: i16,
161 #[new(value = "-1")]
170 type_modifier: i32,
171}
172
173impl FieldInfo {
174 pub fn name(&self) -> &str {
176 &self.name
177 }
178
179 pub fn table_id(&self) -> Option<i32> {
181 self.table_id
182 }
183
184 pub fn column_id(&self) -> Option<i16> {
186 self.column_id
187 }
188
189 pub fn datatype(&self) -> &Type {
191 &self.datatype
192 }
193
194 pub fn format(&self) -> FieldFormat {
196 self.format
197 }
198
199 pub fn format_options(&self) -> &Arc<FormatOptions> {
201 &self.format_options
202 }
203
204 pub fn with_format_options(mut self, format_options: Arc<FormatOptions>) -> Self {
206 self.format_options = format_options;
207 self
208 }
209
210 pub fn type_size(&self) -> i16 {
212 self.type_size
213 }
214
215 pub fn with_type_size(mut self, type_size: i16) -> Self {
217 self.type_size = type_size;
218 self
219 }
220
221 pub fn type_modifier(&self) -> i32 {
223 self.type_modifier
224 }
225
226 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(), fi.table_id.unwrap_or(0), fi.column_id.unwrap_or(0), fi.datatype.oid(), fi.type_size, fi.type_modifier, 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
265pub type SendableRowStream = Pin<Box<dyn Stream<Item = PgWireResult<DataRow>> + Send>>;
267
268pub type SendableCopyDataStream = Pin<Box<dyn Stream<Item = PgWireResult<CopyData>> + Send>>;
270
271#[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 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 pub fn command_tag(&self) -> &str {
304 &self.command_tag
305 }
306
307 pub fn set_command_tag(&mut self, command_tag: &str) {
309 command_tag.clone_into(&mut self.command_tag);
310 }
311
312 pub fn row_schema(&self) -> Arc<Vec<FieldInfo>> {
314 self.row_schema.clone()
315 }
316
317 pub fn data_rows(&mut self) -> &mut SendableRowStream {
319 &mut self.data_rows
320 }
321}
322
323pub struct DataRowEncoder {
325 schema: Arc<Vec<FieldInfo>>,
326 row_buffer: BytesMut,
327 col_index: usize,
328}
329
330impl DataRowEncoder {
331 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 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 let prev_index = self.row_buffer.len();
356 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 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 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#[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
428pub 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 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 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 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 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 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 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 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 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 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, IsNull::No => {
616 let data = temp_buffer.as_ref();
617 data.contains(&delimiter_byte)
618 || data.contains("e_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 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 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 pub fn take_copy(&mut self) -> CopyData {
665 match &self.format {
666 CopyFormat::Binary => {
667 if !self.header_written {
668 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 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 self.buffer.put_u8(b'\n');
684 }
685 }
686
687 self.col_index = 0;
688 CopyData::new(self.buffer.split().freeze())
689 }
690
691 pub fn finish_copy_binary() -> CopyData {
697 CopyData::new(Bytes::from_static(&[0xFF, 0xFF]))
698 }
699
700 fn write_pgcop_header(&mut self) {
702 self.buffer.put_slice(b"PGCOPY\n\xFF\r\n\x00");
703 self.buffer.put_i32(0); self.buffer.put_i32(0); }
706}
707
708pub trait DescribeResponse {
710 fn parameters(&self) -> Option<&[Type]>;
712
713 fn fields(&self) -> &[FieldInfo];
715
716 fn no_data() -> Self;
719
720 fn is_no_data(&self) -> bool;
722}
723
724#[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 fn no_data() -> Self {
744 DescribeStatementResponse {
745 parameters: vec![],
746 fields: vec![],
747 }
748 }
749
750 fn is_no_data(&self) -> bool {
752 self.parameters.is_empty() && self.fields.is_empty()
753 }
754}
755
756#[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 fn no_data() -> Self {
775 DescribePortalResponse { fields: vec![] }
776 }
777
778 fn is_no_data(&self) -> bool {
780 self.fields.is_empty()
781 }
782}
783
784#[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 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 pub fn data_stream(&mut self) -> &mut SendableCopyDataStream {
827 &mut self.data_stream
828 }
829
830 pub fn column_formats(&self) -> Vec<i16> {
832 (0..self.columns).map(|_| self.format as i16).collect()
833 }
834}
835
836#[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 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 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 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 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 assert_eq!(&data[11..15], &[0x00, 0x00, 0x00, 0x00]);
977
978 assert_eq!(&data[15..19], &[0x00, 0x00, 0x00, 0x00]);
980
981 assert_eq!(&data[19..21], &[0x00, 0x01]); assert_eq!(&data[21..25], &[0x00, 0x00, 0x00, 0x04]); 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 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 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 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 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 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 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 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 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 encoder.encode_field(&1i32).unwrap();
1144 encoder.encode_field(&"Alice".to_string()).unwrap();
1145 let copy_data1 = encoder.take_copy();
1146
1147 encoder.encode_field(&2i32).unwrap();
1149 encoder.encode_field(&"Bob".to_string()).unwrap();
1150 let copy_data2 = encoder.take_copy();
1151
1152 let data1 = copy_data1.data.as_ref();
1154
1155 assert_eq!(&data1[19..21], &[0x00, 0x02]); let data2 = copy_data2.data.as_ref();
1160
1161 assert_eq!(&data2[0..2], &[0x00, 0x02]); }
1164}