1use alloc::format;
18use alloc::string::{String, ToString};
19use alloc::vec::Vec;
20
21#[derive(Debug, PartialEq, Eq)]
23pub struct CopyFromSpec {
24 pub table: String,
27 pub columns: Option<Vec<String>>,
31}
32
33#[must_use]
41pub fn parse_copy_from_stdin_head(sql: &str) -> Option<CopyFromSpec> {
42 let trimmed = sql.trim();
43 let lower = trimmed.to_ascii_lowercase();
44 let rest = lower.strip_prefix("copy")?;
45 if !rest.starts_with(char::is_whitespace) {
46 return None;
47 }
48 let rest_orig = &trimmed[trimmed.len() - rest.len()..];
49 let bytes = rest.as_bytes();
50 let mut i = 0;
51 while i < bytes.len() && bytes[i].is_ascii_whitespace() {
52 i += 1;
53 }
54 let t0 = i;
56 while i < bytes.len() && !bytes[i].is_ascii_whitespace() && bytes[i] != b'(' {
57 i += 1;
58 }
59 if i == t0 {
60 return None;
61 }
62 let raw_table = &rest_orig[t0..i];
63 let table = match raw_table.rsplit_once('.') {
64 Some((_, bare)) => bare,
65 None => raw_table,
66 }
67 .trim_matches('"')
68 .to_string();
69 while i < bytes.len() && bytes[i].is_ascii_whitespace() {
70 i += 1;
71 }
72 let mut columns = None;
74 if bytes.get(i) == Some(&b'(') {
75 let cols_start = i + 1;
76 let mut depth = 1usize;
77 i += 1;
78 while i < bytes.len() && depth > 0 {
79 match bytes[i] {
80 b'(' => depth += 1,
81 b')' => depth -= 1,
82 _ => {}
83 }
84 i += 1;
85 }
86 let cols_str = &rest_orig[cols_start..i.saturating_sub(1)];
87 columns = Some(
88 cols_str
89 .split(',')
90 .map(|c| c.trim().trim_matches('"').to_string())
91 .filter(|c| !c.is_empty())
92 .collect::<Vec<_>>(),
93 );
94 while i < bytes.len() && bytes[i].is_ascii_whitespace() {
95 i += 1;
96 }
97 }
98 let tail = &rest[i..];
100 let tail = tail.trim_start();
101 let tail = tail.strip_prefix("from")?;
102 if !tail.starts_with(char::is_whitespace) {
103 return None;
104 }
105 let tail = tail.trim_start();
106 if !(tail == "stdin" || tail.starts_with("stdin")) {
107 return None;
108 }
109 let after = tail["stdin".len()..].trim();
110 if after.contains("format") && !after.contains("text") {
112 return None;
113 }
114 Some(CopyFromSpec { table, columns })
115}
116
117#[derive(Debug)]
121pub struct CopyToFileSpec {
122 pub table: String,
123 pub columns: Option<Vec<String>>,
124 pub query: Option<alloc::boxed::Box<spg_sql::ast::Statement>>,
125 pub path: String,
126 pub options: spg_sql::ast::CopyOptions,
127}
128
129#[must_use]
132pub fn parse_copy_to_file(sql: &str) -> Option<CopyToFileSpec> {
133 match spg_sql::parser::parse_statement(sql) {
134 Ok(spg_sql::ast::Statement::CopyToFile {
135 table,
136 columns,
137 query,
138 path,
139 options,
140 }) => Some(CopyToFileSpec {
141 table,
142 columns,
143 query,
144 path,
145 options,
146 }),
147 _ => None,
148 }
149}
150
151pub fn validate_copy_option_direction(
166 options: &spg_sql::ast::CopyOptions,
167 to_direction: bool,
168) -> Result<(), crate::EngineError> {
169 let is_csv = options.format == spg_sql::ast::CopyFormat::Csv;
170 let csv_only = |name: &str| {
171 crate::EngineError::Unsupported(alloc::format!("COPY {name} requires CSV mode"))
172 };
173 let wrong_way = |name: &str| {
174 crate::EngineError::Unsupported(alloc::format!(
175 "COPY {name} cannot be used with COPY {}",
176 if to_direction { "TO" } else { "FROM" }
177 ))
178 };
179 for (present, name, to_only) in [
180 (options.force_quote.is_some(), "FORCE_QUOTE", true),
181 (options.force_not_null.is_some(), "FORCE_NOT_NULL", false),
182 (options.force_null.is_some(), "FORCE_NULL", false),
183 ] {
184 if !present {
185 continue;
186 }
187 if !is_csv {
188 return Err(csv_only(name));
189 }
190 if to_only != to_direction {
191 return Err(wrong_way(name));
192 }
193 }
194 Ok(())
195}
196
197#[derive(Debug)]
201pub struct CopyFromFileSpec {
202 pub table: String,
203 pub columns: Option<Vec<String>>,
204 pub path: String,
205 pub options: spg_sql::ast::CopyOptions,
206}
207
208#[must_use]
213pub fn parse_copy_from_file(sql: &str) -> Option<CopyFromFileSpec> {
214 match spg_sql::parser::parse_statement(sql) {
215 Ok(spg_sql::ast::Statement::CopyFromFile {
216 table,
217 columns,
218 path,
219 options,
220 }) => Some(CopyFromFileSpec {
221 table,
222 columns,
223 path,
224 options,
225 }),
226 _ => None,
227 }
228}
229
230pub fn copy_buffer_inserts(
240 table: &str,
241 columns: Option<&[String]>,
242 target_cols: &[String],
243 options: &spg_sql::ast::CopyOptions,
244 data: &str,
245) -> Result<Vec<String>, crate::EngineError> {
246 let check_row = |values: &Vec<Option<String>>| -> Result<(), crate::EngineError> {
251 if values.len() > target_cols.len() {
252 return Err(crate::EngineError::Unsupported(String::from(
253 "extra data after last expected column",
254 )));
255 }
256 if values.len() < target_cols.len() {
257 return Err(crate::EngineError::Unsupported(format!(
258 "missing data for column \"{}\"",
259 target_cols[values.len()]
260 )));
261 }
262 Ok(())
263 };
264 use spg_sql::ast::CopyFormat;
265 let is_csv = options.format == CopyFormat::Csv;
266 let delimiter = options.delimiter.unwrap_or(if is_csv { ',' } else { '\t' });
267 let quote = options.quote.unwrap_or('"');
268 let null_str = options
269 .null_str
270 .clone()
271 .unwrap_or_else(|| String::from(if is_csv { "" } else { "\\N" }));
272 validate_copy_option_direction(options, false)?;
275 let in_list = |list: &Option<alloc::vec::Vec<String>>, idx: usize| -> bool {
276 match list {
277 None => false,
278 Some(cols) if cols.is_empty() => true,
279 Some(cols) => target_cols
280 .get(idx)
281 .is_some_and(|c| cols.iter().any(|w| w.eq_ignore_ascii_case(c))),
282 }
283 };
284 let mut inserts = Vec::new();
285 let mut first = true;
286 if is_csv {
287 let mut buf: Vec<u8> = data.as_bytes().to_vec();
288 if !buf.is_empty() && !buf.ends_with(b"\n") {
289 buf.push(b'\n');
290 }
291 let d8 = u8::try_from(delimiter as u32).unwrap_or(b',');
292 let q8 = u8::try_from(quote as u32).unwrap_or(b'"');
293 let mut start = 0;
294 while let Some(len) = csv_record_end(&buf[start..], d8, q8) {
295 let mut rec = &buf[start..start + len - 1];
296 start += len;
297 if rec.last() == Some(&b'\r') {
298 rec = &rec[..rec.len() - 1];
299 }
300 if rec.is_empty() {
301 continue;
302 }
303 if first && options.header {
304 first = false;
305 continue;
306 }
307 first = false;
308 let rec_str = core::str::from_utf8(rec).map_err(|_| {
309 crate::EngineError::Unsupported("COPY FROM: non-UTF-8 input".into())
310 })?;
311 let mut values = decode_copy_csv_record(rec_str, delimiter, quote, &null_str);
312 if options.force_not_null.is_some() || options.force_null.is_some() {
319 for (idx, cell) in values.iter_mut().enumerate() {
320 if in_list(&options.force_not_null, idx) && cell.is_none() {
321 *cell = Some(String::new());
322 }
323 if in_list(&options.force_null, idx)
324 && cell.as_deref() == Some(null_str.as_str())
325 {
326 *cell = None;
327 }
328 }
329 }
330 check_row(&values)?;
331 inserts.push(build_copy_insert(table, columns, &values));
332 }
333 } else {
334 for line in data.lines() {
335 let line = line.strip_suffix('\r').unwrap_or(line);
336 if line.is_empty() {
337 continue;
338 }
339 if first && options.header {
340 first = false;
341 continue;
342 }
343 first = false;
344 if line == "\\." {
345 break;
346 }
347 let values = decode_copy_text_row(line);
348 check_row(&values)?;
349 inserts.push(build_copy_insert(table, columns, &values));
350 }
351 }
352 Ok(inserts)
353}
354#[must_use]
357pub fn decode_copy_text_row(line: &str) -> Vec<Option<String>> {
358 line.split('\t')
359 .map(|cell| {
360 if cell == "\\N" {
361 None
362 } else {
363 let mut out = String::with_capacity(cell.len());
364 let mut chars = cell.chars();
365 while let Some(c) = chars.next() {
366 if c == '\\'
367 && let Some(n) = chars.next()
368 {
369 out.push(match n {
370 'b' => '\u{08}',
371 'f' => '\u{0c}',
372 'n' => '\n',
373 'r' => '\r',
374 't' => '\t',
375 'v' => '\u{0b}',
376 '\\' => '\\',
377 other => other,
378 });
379 } else {
380 out.push(c);
381 }
382 }
383 Some(out)
384 }
385 })
386 .collect()
387}
388
389#[must_use]
399pub fn decode_copy_csv_record(
400 record: &str,
401 delimiter: char,
402 quote: char,
403 null_str: &str,
404) -> Vec<Option<String>> {
405 let chars: Vec<char> = record.chars().collect();
406 let n = chars.len();
407 let mut fields: Vec<Option<String>> = Vec::new();
408 let mut i = 0;
409 loop {
410 if i < n && chars[i] == quote {
411 i += 1;
413 let mut content = String::new();
414 while i < n {
415 let c = chars[i];
416 if c == quote {
417 if i + 1 < n && chars[i + 1] == quote {
418 content.push(quote);
419 i += 2;
420 } else {
421 i += 1; break;
423 }
424 } else {
425 content.push(c);
426 i += 1;
427 }
428 }
429 fields.push(Some(content));
430 while i < n && chars[i] != delimiter {
433 i += 1;
434 }
435 } else {
436 let start = i;
438 while i < n && chars[i] != delimiter {
439 i += 1;
440 }
441 let content: String = chars[start..i].iter().collect();
442 fields.push(if content == null_str {
443 None
444 } else {
445 Some(content)
446 });
447 }
448 if i < n && chars[i] == delimiter {
449 i += 1; } else {
451 break;
452 }
453 }
454 fields
455}
456
457#[must_use]
467pub fn csv_record_end(buf: &[u8], delimiter: u8, quote: u8) -> Option<usize> {
468 let mut in_quote = false;
469 let mut at_field_start = true;
470 let mut i = 0;
471 while i < buf.len() {
472 let b = buf[i];
473 if in_quote {
474 if b == quote {
475 if buf.get(i + 1) == Some("e) {
476 i += 2; continue;
478 }
479 in_quote = false; }
481 } else if b == quote && at_field_start {
483 in_quote = true;
484 at_field_start = false;
485 } else if b == b'\n' {
486 return Some(i + 1);
487 } else {
488 at_field_start = b == delimiter;
489 }
490 i += 1;
491 }
492 None
493}
494
495#[must_use]
500pub fn build_copy_insert(
501 table: &str,
502 columns: Option<&[String]>,
503 values: &[Option<String>],
504) -> String {
505 let mut sql = format!("INSERT INTO {table} ");
506 if let Some(cols) = columns {
507 sql.push('(');
508 for (i, c) in cols.iter().enumerate() {
509 if i > 0 {
510 sql.push_str(", ");
511 }
512 sql.push_str(c);
513 }
514 sql.push_str(") ");
515 }
516 sql.push_str("VALUES (");
517 for (i, v) in values.iter().enumerate() {
518 if i > 0 {
519 sql.push_str(", ");
520 }
521 match v {
522 None => sql.push_str("NULL"),
523 Some(s) => {
524 if copy_cell_looks_numeric(s)
525 || matches!(s.as_str(), "true" | "false" | "TRUE" | "FALSE")
526 {
527 sql.push_str(s);
528 } else {
529 sql.push('\'');
530 for ch in s.chars() {
531 if ch == '\'' {
532 sql.push('\'');
533 }
534 sql.push(ch);
535 }
536 sql.push('\'');
537 }
538 }
539 }
540 }
541 sql.push(')');
542 sql
543}
544
545fn copy_cell_looks_numeric(s: &str) -> bool {
549 if s.is_empty() {
550 return false;
551 }
552 let b = s.as_bytes();
553 let mut i = 0;
554 if b[0] == b'-' || b[0] == b'+' {
555 if b.len() == 1 {
556 return false;
557 }
558 i = 1;
559 }
560 let mut seen_dot = false;
561 let mut seen_digit = false;
562 while i < b.len() {
563 match b[i] {
564 b'0'..=b'9' => seen_digit = true,
565 b'.' if !seen_dot => seen_dot = true,
566 _ => return false,
567 }
568 i += 1;
569 }
570 if !seen_dot && s.trim_start_matches(['-', '+']).len() > 1 {
573 let digits = s.trim_start_matches(['-', '+']);
574 if digits.starts_with('0') {
575 return false;
576 }
577 }
578 seen_digit
579}
580
581#[must_use]
586pub fn encode_copy_text_cells(cells: &[Option<String>]) -> String {
587 encode_copy_text_cells_opts(cells, '\t', "\\N")
588}
589
590#[must_use]
596pub fn encode_copy_text_cells_opts(
597 cells: &[Option<String>],
598 delimiter: char,
599 null_str: &str,
600) -> String {
601 let mut out = String::new();
602 for (i, cell) in cells.iter().enumerate() {
603 if i > 0 {
604 out.push(delimiter);
605 }
606 match cell {
607 None => out.push_str(null_str),
608 Some(s) => {
609 for c in s.chars() {
610 match c {
611 '\\' => out.push_str("\\\\"),
612 '\t' => out.push_str("\\t"),
613 '\n' => out.push_str("\\n"),
614 '\r' => out.push_str("\\r"),
615 '\u{08}' => out.push_str("\\b"),
616 '\u{0c}' => out.push_str("\\f"),
617 '\u{0b}' => out.push_str("\\v"),
618 other if other == delimiter => {
619 out.push('\\');
620 out.push(other);
621 }
622 other => out.push(other),
623 }
624 }
625 }
626 }
627 }
628 out
629}
630
631#[must_use]
639pub fn encode_copy_csv_cells(
640 cells: &[Option<String>],
641 delimiter: char,
642 quote: char,
643 null_str: &str,
644) -> String {
645 encode_copy_csv_cells_opts(cells, delimiter, quote, quote, None, null_str)
646}
647
648pub fn encode_copy_csv_cells_opts(
654 cells: &[Option<String>],
655 delimiter: char,
656 quote: char,
657 escape: char,
658 force_quote: Option<&[bool]>,
659 null_str: &str,
660) -> String {
661 let mut out = String::new();
662 for (i, cell) in cells.iter().enumerate() {
663 if i > 0 {
664 out.push(delimiter);
665 }
666 match cell {
667 None => out.push_str(null_str),
668 Some(s) => {
669 let forced = force_quote.and_then(|f| f.get(i)).copied().unwrap_or(false);
670 let needs_quote = forced
671 || s.as_str() == null_str
672 || s.chars().any(|c| {
673 c == delimiter || c == quote || c == escape || c == '\n' || c == '\r'
674 });
675 if needs_quote {
676 out.push(quote);
677 for c in s.chars() {
678 if c == quote || c == escape {
679 out.push(escape);
680 }
681 out.push(c);
682 }
683 out.push(quote);
684 } else {
685 out.push_str(s);
686 }
687 }
688 }
689 }
690 out
691}
692
693#[cfg(test)]
694mod tests {
695 use super::*;
696 use alloc::string::ToString;
697 use alloc::vec;
698
699 #[test]
700 fn parses_pg_dump_copy_head() {
701 let spec =
702 parse_copy_from_stdin_head("COPY public.messages (id, subject, body) FROM stdin")
703 .unwrap();
704 assert_eq!(spec.table, "messages");
705 assert_eq!(
706 spec.columns.as_deref(),
707 Some(&["id".to_string(), "subject".to_string(), "body".to_string()][..])
708 );
709 let bare = parse_copy_from_stdin_head("copy t from stdin").unwrap();
711 assert_eq!(bare.table, "t");
712 assert_eq!(bare.columns, None);
713 assert!(parse_copy_from_stdin_head("COPY t TO stdout").is_none());
715 assert!(parse_copy_from_stdin_head("COPY t FROM '/tmp/f.csv'").is_none());
716 assert!(parse_copy_from_stdin_head("COPY t FROM stdin WITH (FORMAT csv)").is_none());
717 }
718
719 #[test]
720 fn decodes_text_rows() {
721 assert_eq!(
722 decode_copy_text_row("1\thello\t\\N\ta\\tb"),
723 vec![
724 Some("1".to_string()),
725 Some("hello".to_string()),
726 None,
727 Some("a\tb".to_string())
728 ]
729 );
730 }
731
732 #[test]
733 fn builds_inserts_with_column_list() {
734 let cols = vec!["id".to_string(), "note".to_string()];
735 let row = vec![Some("7".to_string()), Some("it's".to_string())];
736 assert_eq!(
737 build_copy_insert("t", Some(&cols), &row),
738 "INSERT INTO t (id, note) VALUES (7, 'it''s')"
739 );
740 assert_eq!(
741 build_copy_insert("t", None, &[None, Some("0042".to_string())]),
742 "INSERT INTO t VALUES (NULL, '0042')"
743 );
744 }
745
746 fn csv(record: &str) -> Vec<Option<String>> {
747 decode_copy_csv_record(record, ',', '"', "")
748 }
749
750 #[test]
751 fn decodes_csv_quoting_and_null() {
752 assert_eq!(
754 csv("p,\"x,y\",\"a\"\"b\""),
755 vec![
756 Some("p".to_string()),
757 Some("x,y".to_string()),
758 Some("a\"b".to_string()),
759 ]
760 );
761 assert_eq!(
763 csv("q, spaced ,"),
764 vec![Some("q".to_string()), Some(" spaced ".to_string()), None]
765 );
766 assert_eq!(csv(",\"\""), vec![None, Some(String::new())]);
768 assert_eq!(
770 csv("\"line\nbreak\",r"),
771 vec![Some("line\nbreak".to_string()), Some("r".to_string())]
772 );
773 }
774
775 #[test]
776 fn decodes_csv_custom_delimiter_quote_and_null() {
777 assert_eq!(
778 decode_copy_csv_record("1;#a;b#;NULO", ';', '#', "NULO"),
779 vec![Some("1".to_string()), Some("a;b".to_string()), None]
780 );
781 }
782
783 #[test]
784 fn csv_record_end_is_quote_aware() {
785 assert_eq!(csv_record_end(b"a,b\nrest", b',', b'"'), Some(4));
787 assert_eq!(csv_record_end(b"a,\"x\ny\"\nnext", b',', b'"'), Some(8));
790 assert_eq!(csv_record_end(b"1,\"p\nq\"\n", b',', b'"'), Some(8));
793 assert_eq!(csv_record_end(b"\"a\"\"b\"\nx", b',', b'"'), Some(7));
795 assert_eq!(csv_record_end(b"\"unterminated\n", b',', b'"'), None);
797 assert_eq!(csv_record_end(b"a,b", b',', b'"'), None);
799 }
800}