diff --git a/src/impls/ast_builder.rs b/src/impls/ast_builder.rs index 72b5b7bed..344c235bf 100644 --- a/src/impls/ast_builder.rs +++ b/src/impls/ast_builder.rs @@ -60,7 +60,11 @@ pub(crate) fn trigger( } pub(crate) fn insert(table: ObjectName, columns: Vec, source: Query) -> Statement { - Statement::Insert(Insert { + Statement::Insert(insert_body(table, columns, source)) +} + +pub(crate) fn insert_body(table: ObjectName, columns: Vec, source: Query) -> Insert { + Insert { insert_token: AttachedToken::empty(), optimizer_hints: Vec::new(), or: None, @@ -89,7 +93,7 @@ pub(crate) fn insert(table: ObjectName, columns: Vec, source: Query) multi_table_into_clauses: Vec::new(), multi_table_when_clauses: Vec::new(), multi_table_else_clause: None, - }) + } } pub(crate) fn values(rows: Vec>) -> Query { diff --git a/src/impls/translator_impls.rs b/src/impls/translator_impls.rs index d3f868ff0..0574edb39 100644 --- a/src/impls/translator_impls.rs +++ b/src/impls/translator_impls.rs @@ -17,6 +17,7 @@ pub(crate) mod column; pub(crate) mod column_option; mod condition_injection; mod constraint_characteristic; +mod copy; pub(crate) mod create_index; mod create_table; mod create_trigger; diff --git a/src/impls/translator_impls/copy.rs b/src/impls/translator_impls/copy.rs new file mode 100644 index 000000000..6feb6dde1 --- /dev/null +++ b/src/impls/translator_impls/copy.rs @@ -0,0 +1,558 @@ +//! `COPY ... FROM STDIN`, translated into the `INSERT` that writes its inline +//! rows. +//! +//! The payload is decoded with PostgreSQL's own text and CSV rules, and each +//! field becomes a string literal, which PostgreSQL coerces to the column type +//! through the same input function `COPY` calls. The `INSERT` then takes the +//! ordinary translation, so every column conversion applies to it. + +#[cfg(not(feature = "std"))] +use alloc::{ + format, + string::{String, ToString}, + vec, + vec::Vec, +}; +use core::{iter::Peekable, str::Chars}; + +use sql_traits::{ + structs::ParserDB, + traits::{ColumnLike, TableLike}, +}; +use sqlparser::ast::{ + ColumnOption, CopyLegacyCsvOption, CopyLegacyOption, CopyOption, CopyPayload, CopySource, + CopyTarget, Expr, Ident, ObjectName, ObjectNamePart, Statement, TableObject, Value, +}; + +use super::{insert::backing_table_name, rls}; +use crate::{ + errors::Error, + impls::{ast_builder, object_name::resolve_translation_table}, + options::TranslationContext, + traits::translator::TranslatorWithContext, + warnings::TranslationWarning, +}; + +/// The parts of a `COPY` statement. +pub(super) struct Copy<'a> { + pub(super) source: &'a CopySource, + pub(super) to: bool, + pub(super) target: &'a CopyTarget, + pub(super) options: &'a [CopyOption], + pub(super) legacy_options: &'a [CopyLegacyOption], + pub(super) payload: Option<&'a CopyPayload>, +} + +/// Translates `COPY`, which becomes an `INSERT` for an inline payload and is +/// refused in every other form. +pub(super) fn translate_copy( + copy: &Copy<'_>, + schema: &ParserDB, + options: &TranslationContext<'_>, + emit: &mut dyn FnMut(TranslationWarning), +) -> Result, Error> { + let &Copy { source, to, target, options: copy_options, legacy_options, payload } = copy; + let CopySource::Table { table_name, columns } = source else { + return Err(reject_copy("COPY (query)", to)); + }; + if to || !matches!(target, CopyTarget::Stdin) { + return Err(reject_copy(&format!("COPY {table_name}"), to)); + } + let Some(payload) = payload.filter(|payload| payload.terminated) else { + return Err(Error::forward_refusal(format!( + "COPY {table_name} FROM STDIN has no payload ended by a `\\.` line, and PostgreSQL \ + refuses a payload that runs to the end of the input. End the rows with `\\.` on a \ + line of its own." + ))); + }; + let format = Format::new(copy_options, legacy_options)?; + let columns = + if columns.is_empty() { table_columns(table_name, schema)? } else { columns.clone() }; + let rows = match format.kind { + Kind::Text => text_rows(&payload.data, &format)?, + Kind::Csv => csv_rows(&payload.data, &format, &columns)?, + }; + if let Some((row, fields)) = + rows.iter().enumerate().find(|(_, fields)| fields.len() != columns.len()) + { + return Err(Error::forward_refusal(format!( + "row {} of COPY {table_name} has {} fields, and the statement writes {} columns, so \ + PostgreSQL refuses the whole COPY.", + row + 1, + fields.len(), + columns.len() + ))); + } + if rows.is_empty() { + return Ok(Vec::new()); + } + let values = rows + .into_iter() + .map(|fields| { + fields + .into_iter() + .map(|field| { + Expr::Value( + field.map_or(Value::Null, Value::SingleQuotedString).with_empty_span(), + ) + }) + .collect() + }) + .collect(); + let column_names = columns.into_iter().map(|column| ObjectName::from(vec![column])).collect(); + let insert = + ast_builder::insert_body(table_name.clone(), column_names, ast_builder::values(values)); + let mut translated = insert.translate_with_warnings(schema, options, emit)?; + // A `pg_dump` restore runs as the table owner, which PostgreSQL lets past + // the policies, so the rows go to the backing table. + if let Some(last) = table_name.0.last().and_then(ObjectNamePart::as_ident) + && rls::table_has_rls(&last.value, schema)? + && let TableObject::TableName(name) = &translated.table + { + translated.table = TableObject::TableName(backing_table_name(name, options)); + } + Ok(vec![Statement::Insert(translated)]) +} + +/// Refuses an export or a `COPY` that reads a file or a program. +fn reject_copy(subject: &str, to: bool) -> Error { + let advice = if to { + "SQLite has no COPY statement, and the translator cannot write the destination. Export the \ + rows with your client instead." + } else { + "SQLite has no COPY statement, and the translator cannot read the source. Load the rows \ + with INSERT statements instead." + }; + Error::forward_refusal(format!("{subject} cannot be translated. {advice}")) +} + +/// The columns a `COPY` without a column list writes, every column but the +/// generated ones, in declaration order. +fn table_columns(table_name: &ObjectName, schema: &ParserDB) -> Result, Error> { + // PostgreSQL looks an unqualified name up in `pg_temp` first. + let resolved = match resolve_translation_table(schema, table_name)? { + Some(table) => Some(table), + None => schema.resolve_table_object_name_on_search_path(table_name)?, + }; + let Some(table) = resolved else { + return Err(Error::forward_refusal(format!( + "COPY {table_name} names no table the translation declares, so the columns its rows \ + fill are unknown." + ))); + }; + Ok(table + .columns(schema)? + .filter(|column| { + !column.attribute().options.iter().any(|option| { + matches!(option.option, ColumnOption::Generated { generation_expr: Some(_), .. }) + }) + }) + .map(|column| { + let name = column.column_name().to_string(); + if column.column_name_is_quoted() { + Ident::with_quote('"', name) + } else { + Ident::new(name) + } + }) + .collect()) +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum Kind { + Text, + Csv, +} + +/// How a payload is decoded, after PostgreSQL's defaults and checks. +struct Format { + kind: Kind, + delimiter: char, + null: String, + header: bool, + quote: char, + escape: char, + force_null: Vec, + force_not_null: Vec, +} + +/// The options as written, each at most once. +#[derive(Default)] +struct Written { + kind: Option, + delimiter: Option, + null: Option, + header: Option, + quote: Option, + escape: Option, + force_null: Option>, + force_not_null: Option>, +} + +impl Format { + fn new(options: &[CopyOption], legacy: &[CopyLegacyOption]) -> Result { + let mut written = Written::default(); + for option in options { + match option { + CopyOption::Format(name) => { + let kind = match name.value.to_ascii_lowercase().as_str() { + "text" => Kind::Text, + "csv" => Kind::Csv, + _ => return Err(unsupported_option(&option.to_string())), + }; + once(&mut written.kind, kind, "FORMAT")?; + } + CopyOption::Delimiter(delimiter) => { + once(&mut written.delimiter, *delimiter, "DELIMITER")?; + } + CopyOption::Null(null) => once(&mut written.null, null.clone(), "NULL")?, + CopyOption::Header(header) => once(&mut written.header, *header, "HEADER")?, + CopyOption::Quote(quote) => once(&mut written.quote, *quote, "QUOTE")?, + CopyOption::Escape(escape) => once(&mut written.escape, *escape, "ESCAPE")?, + CopyOption::ForceNull(columns) => { + once(&mut written.force_null, columns.clone(), "FORCE_NULL")?; + } + CopyOption::ForceNotNull(columns) => { + once(&mut written.force_not_null, columns.clone(), "FORCE_NOT_NULL")?; + } + CopyOption::Freeze(_) | CopyOption::ForceQuote(_) | CopyOption::Encoding(_) => { + return Err(unsupported_option(&option.to_string())); + } + } + } + for option in legacy { + match option { + CopyLegacyOption::Delimiter(delimiter) => { + once(&mut written.delimiter, *delimiter, "DELIMITER")?; + } + CopyLegacyOption::Null(null) => once(&mut written.null, null.clone(), "NULL")?, + CopyLegacyOption::Csv(csv) => { + once(&mut written.kind, Kind::Csv, "CSV")?; + for csv_option in csv { + match csv_option { + CopyLegacyCsvOption::Header => { + once(&mut written.header, true, "HEADER")?; + } + CopyLegacyCsvOption::Quote(quote) => { + once(&mut written.quote, *quote, "QUOTE")?; + } + CopyLegacyCsvOption::Escape(escape) => { + once(&mut written.escape, *escape, "ESCAPE")?; + } + CopyLegacyCsvOption::ForceNotNull(columns) => { + once( + &mut written.force_not_null, + columns.clone(), + "FORCE NOT NULL", + )?; + } + CopyLegacyCsvOption::ForceQuote(_) => { + return Err(unsupported_option(&csv_option.to_string())); + } + } + } + } + other => return Err(unsupported_option(&other.to_string())), + } + } + Self::checked(written) + } + + /// Applies PostgreSQL's defaults and refuses what its option check refuses. + fn checked(written: Written) -> Result { + let kind = written.kind.unwrap_or(Kind::Text); + let csv = kind == Kind::Csv; + if !csv { + for (name, given) in [ + ("QUOTE", written.quote.is_some()), + ("ESCAPE", written.escape.is_some()), + ("FORCE_NULL", written.force_null.is_some()), + ("FORCE_NOT_NULL", written.force_not_null.is_some()), + ] { + if given { + return Err(invalid_option(&format!("COPY {name} requires CSV mode"))); + } + } + } + let delimiter = written.delimiter.unwrap_or(if csv { ',' } else { '\t' }); + let quote = written.quote.unwrap_or('"'); + let escape = written.escape.unwrap_or(quote); + let null = written.null.unwrap_or_else(|| if csv { String::new() } else { "\\N".into() }); + if matches!(delimiter, '\n' | '\r') { + return Err(invalid_option("COPY delimiter cannot be newline or carriage return")); + } + if null.contains(['\n', '\r']) { + return Err(invalid_option( + "COPY null representation cannot use newline or carriage return", + )); + } + if !csv && "\\.abcdefghijklmnopqrstuvwxyz0123456789".contains(delimiter) { + return Err(invalid_option(&format!("COPY delimiter cannot be \"{delimiter}\""))); + } + if csv && delimiter == quote { + return Err(invalid_option("COPY delimiter and quote must be different")); + } + if null.contains(delimiter) { + return Err(invalid_option( + "COPY delimiter character must not appear in the NULL specification", + )); + } + if csv && null.contains(quote) { + return Err(invalid_option( + "CSV quote character must not appear in the NULL specification", + )); + } + Ok(Self { + kind, + delimiter, + null, + header: written.header.unwrap_or(false), + quote, + escape, + force_null: written.force_null.unwrap_or_default(), + force_not_null: written.force_not_null.unwrap_or_default(), + }) + } +} + +fn once(slot: &mut Option, value: T, name: &str) -> Result<(), Error> { + if slot.replace(value).is_some() { + return Err(invalid_option(&format!("COPY {name} is given more than once"))); + } + Ok(()) +} + +fn unsupported_option(option: &str) -> Error { + Error::forward_refusal(format!( + "COPY option {option} cannot be translated. Only the text and CSV formats with DELIMITER, \ + NULL, HEADER, QUOTE, ESCAPE, FORCE_NULL and FORCE_NOT_NULL become INSERT statements." + )) +} + +fn invalid_option(reason: &str) -> Error { + Error::forward_refusal(format!("{reason}, so PostgreSQL refuses the COPY.")) +} + +fn malformed(record: usize, reason: &str) -> Error { + Error::forward_refusal(format!( + "row {record} of the COPY payload {reason}, so PostgreSQL refuses the whole COPY." + )) +} + +/// The line ending a payload uses, which PostgreSQL requires to be the same +/// on every line. +#[derive(Clone, Copy, PartialEq, Eq)] +enum LineEnd { + Lf, + CrLf, + Cr, +} + +/// Consumes the line ending that starts with `first`, and refuses one that +/// differs from the payload's first. +fn line_end( + first: char, + chars: &mut Peekable>, + seen: &mut Option, + record: usize, +) -> Result<(), Error> { + let end = match first { + '\r' if chars.peek() == Some(&'\n') => { + chars.next(); + LineEnd::CrLf + } + '\r' => LineEnd::Cr, + _ => LineEnd::Lf, + }; + if *seen.get_or_insert(end) == end { + Ok(()) + } else { + Err(malformed(record, "ends with a different line ending than the first row")) + } +} + +/// Decodes the text format, matching the null string before any backslash +/// escape is resolved. +fn text_rows(data: &str, format: &Format) -> Result>>, Error> { + let mut rows = Vec::new(); + let mut chars = data.chars().peekable(); + let mut seen = None; + let mut record = 0; + while chars.peek().is_some() { + record += 1; + let mut fields = Vec::new(); + let mut raw = String::new(); + let mut value = Vec::new(); + loop { + match chars.next() { + None => break, + Some(end @ ('\n' | '\r')) => { + line_end(end, &mut chars, &mut seen, record)?; + break; + } + Some(character) if character == format.delimiter => { + let field = text_field(&raw, value, format, record)?; + fields.push(field); + raw.clear(); + value = Vec::new(); + } + Some('\\') => { + raw.push('\\'); + let Some(escaped) = chars.next() else { break }; + raw.push(escaped); + unescape(escaped, &mut chars, &mut raw, &mut value); + } + Some(character) => { + raw.push(character); + push_char(&mut value, character); + } + } + } + fields.push(text_field(&raw, value, format, record)?); + if !(format.header && record == 1) { + rows.push(fields); + } + } + Ok(rows) +} + +/// Resolves the backslash escape whose character is `escaped`. +fn unescape(escaped: char, chars: &mut Peekable>, raw: &mut String, value: &mut Vec) { + let byte = match escaped { + 'b' => 0x08, + 'f' => 0x0c, + 'n' => b'\n', + 'r' => b'\r', + 't' => b'\t', + 'v' => 0x0b, + '0'..='7' => { + let mut code = escaped.to_digit(8).unwrap_or_default(); + for _ in 0..2 { + let Some(digit) = chars.peek().and_then(|next| next.to_digit(8)) else { break }; + raw.extend(chars.next()); + code = code * 8 + digit; + } + // PostgreSQL keeps the low byte of a three-digit octal escape. + u8::try_from(code & 0o377).unwrap_or_default() + } + 'x' if chars.peek().is_some_and(char::is_ascii_hexdigit) => { + let mut code = 0; + for _ in 0..2 { + let Some(digit) = chars.peek().and_then(|next| next.to_digit(16)) else { break }; + raw.extend(chars.next()); + code = code * 16 + digit; + } + u8::try_from(code).unwrap_or_default() + } + other => { + push_char(value, other); + return; + } + }; + value.push(byte); +} + +fn push_char(value: &mut Vec, character: char) { + value.extend_from_slice(character.encode_utf8(&mut [0; 4]).as_bytes()); +} + +fn text_field( + raw: &str, + value: Vec, + format: &Format, + record: usize, +) -> Result, Error> { + if raw == format.null { + return Ok(None); + } + String::from_utf8(value) + .map(Some) + .map_err(|_| malformed(record, "decodes to bytes that are not valid UTF-8")) +} + +/// Decodes the CSV format, where only an unquoted field matches the null +/// string unless `FORCE_NULL` or `FORCE_NOT_NULL` says otherwise. +fn csv_rows( + data: &str, + format: &Format, + columns: &[Ident], +) -> Result>>, Error> { + let force_null = positions(&format.force_null, columns, "FORCE_NULL")?; + let force_not_null = positions(&format.force_not_null, columns, "FORCE_NOT_NULL")?; + let field = |value: String, quoted: bool, position: usize| { + let null = value == format.null + && if quoted { + force_null.contains(&position) + } else { + !force_not_null.contains(&position) + }; + (!null).then_some(value) + }; + let mut rows = Vec::new(); + let mut chars = data.chars().peekable(); + let mut seen = None; + let mut record = 0; + while chars.peek().is_some() { + record += 1; + let mut fields = Vec::new(); + let mut value = String::new(); + let (mut quoted, mut in_quotes) = (false, false); + loop { + let next = chars.next(); + if in_quotes { + match next { + None => return Err(malformed(record, "ends inside a quoted field")), + Some(character) + if character == format.escape + && chars.peek().is_some_and(|&after| { + after == format.quote || after == format.escape + }) => + { + value.extend(chars.next()); + } + Some(character) if character == format.quote => in_quotes = false, + Some(character) => value.push(character), + } + continue; + } + match next { + None => break, + Some(end @ ('\n' | '\r')) => { + line_end(end, &mut chars, &mut seen, record)?; + break; + } + Some(character) if character == format.delimiter => { + fields.push(field(core::mem::take(&mut value), quoted, fields.len())); + quoted = false; + } + Some(character) if character == format.quote => (quoted, in_quotes) = (true, true), + Some(character) => value.push(character), + } + } + fields.push(field(value, quoted, fields.len())); + if !(format.header && record == 1) { + rows.push(fields); + } + } + Ok(rows) +} + +/// The positions of `named` among the columns the `COPY` writes. +fn positions(named: &[Ident], columns: &[Ident], option: &str) -> Result, Error> { + named + .iter() + .map(|name| { + columns.iter().position(|column| same_column(column, name)).ok_or_else(|| { + invalid_option(&format!("{option} column \"{name}\" is not referenced by COPY")) + }) + }) + .collect() +} + +fn same_column(left: &Ident, right: &Ident) -> bool { + if left.quote_style.is_some() && right.quote_style.is_some() { + left.value == right.value + } else { + left.value.eq_ignore_ascii_case(&right.value) + } +} diff --git a/src/impls/translator_impls/insert.rs b/src/impls/translator_impls/insert.rs index 131b4939c..d9d525062 100644 --- a/src/impls/translator_impls/insert.rs +++ b/src/impls/translator_impls/insert.rs @@ -939,17 +939,25 @@ fn rewrite_rls_view_insert<'schema>( // Strict mode: redirect to the backing table. The BEFORE INSERT guard on // the backing table keeps WITH CHECK enforcement on this path. - let suffix = options.get_rls_table_suffix(); - let backing_name = format!("{}{suffix}", last.value); - let mut new_parts = table_name.0.clone(); - if let Some(last_part) = new_parts.last_mut() { - *last_part = - sqlparser::ast::ObjectNamePart::Identifier(sqlparser::ast::Ident::new(backing_name)); - } - insert.table = TableObject::TableName(sqlparser::ast::ObjectName(new_parts)); + insert.table = TableObject::TableName(backing_table_name(table_name, options)); Ok(target) } +/// `table_name` with its last part renamed to the RLS backing table. +pub(crate) fn backing_table_name( + table_name: &sqlparser::ast::ObjectName, + options: &crate::options::TranslationContext<'_>, +) -> sqlparser::ast::ObjectName { + let mut parts = table_name.0.clone(); + if let Some(last) = parts.last_mut() + && let Some(ident) = last.as_ident() + { + let backing = format!("{}{}", ident.value, options.get_rls_table_suffix()); + *last = sqlparser::ast::ObjectNamePart::Identifier(sqlparser::ast::Ident::new(backing)); + } + sqlparser::ast::ObjectName(parts) +} + /// The first returned column whose value the database supplies rather than the /// caller: a declared default, a generated column, or the integer primary key /// SQLite assigns. Those are exactly the columns a view cannot answer for. diff --git a/src/impls/translator_impls/statement.rs b/src/impls/translator_impls/statement.rs index d445df3c8..7a43c0773 100644 --- a/src/impls/translator_impls/statement.rs +++ b/src/impls/translator_impls/statement.rs @@ -44,13 +44,12 @@ use sql_traits::{ }; use sqlparser::ast::{ AlterTable, AlterTableOperation, AlterTableType, BeginTransactionKind, BinaryOperator, - CascadeOption, ColumnDef, ColumnOption, CopySource, CopyTarget, CreateFunction, - CreateTableOptions, CreateView, Delete, DescribeAlias, DiscardObject, DropBehavior, - ExceptionWhen, Expr, FromTable, Ident, Merge, ObjectName, ObjectNamePart, ObjectType, Query, - RenameTable, RenameTableNameKind, Set, SqlOption, Statement, TableFactor, TableWithJoins, - TransactionAccessMode, TransactionMode, TransactionModifier, TriggerEvent, TriggerPeriod, - Truncate, TruncateIdentityOption, UnaryOperator, VacuumStatement, ViewColumnDef, - helpers::attached_token::AttachedToken, + CascadeOption, ColumnDef, ColumnOption, CreateFunction, CreateTableOptions, CreateView, Delete, + DescribeAlias, DiscardObject, DropBehavior, ExceptionWhen, Expr, FromTable, Ident, Merge, + ObjectName, ObjectNamePart, ObjectType, Query, RenameTable, RenameTableNameKind, Set, + SqlOption, Statement, TableFactor, TableWithJoins, TransactionAccessMode, TransactionMode, + TransactionModifier, TriggerEvent, TriggerPeriod, Truncate, TruncateIdentityOption, + UnaryOperator, VacuumStatement, ViewColumnDef, helpers::attached_token::AttachedToken, }; use crate::{ @@ -1081,34 +1080,6 @@ fn reject_untranslatable_truncate_options(truncate: &Truncate) -> Result<(), Err } } -/// Rejects `COPY`, which SQLite has no statement for in any of its forms. -/// -/// Refusing rather than dropping matters most for `COPY ... FROM stdin`, whose -/// rows are carried inline in the migration file, so a silent drop loses data. -/// -/// Translating that form into multi-row `INSERT`s would be the better answer -/// for a migration translator, and it is not implemented, so it is refused with -/// the rest. -fn reject_copy(source: &CopySource, to: bool, target: &CopyTarget) -> Error { - let subject = match source { - CopySource::Table { table_name, .. } => format!("COPY {table_name}"), - CopySource::Query(_) => "COPY (query)".to_owned(), - }; - - let advice = if to { - "SQLite has no COPY statement, and the translator cannot write the destination. Export the \ - rows with your client instead." - } else if matches!(target, CopyTarget::Stdin) { - "SQLite has no COPY statement. The rows are carried inline in this statement, so they \ - would be lost outright. Rewrite the payload as INSERT statements." - } else { - "SQLite has no COPY statement, and the translator cannot read the source. Load the rows \ - with INSERT statements instead." - }; - - Error::forward_refusal(format!("{subject} cannot be translated. {advice}")) -} - /// Rejects `MERGE`, which has no SQLite form. /// /// `INSERT ... ON CONFLICT DO UPDATE` resembles a translation without being @@ -2086,8 +2057,23 @@ impl crate::traits::translator::TranslatorWithContext for Statement { } Statement::RenameTable(renames) => return Err(reject_rename_table(renames)), Statement::Truncate(truncate) => translate_truncate(truncate, schema, options, emit)?, - Statement::Copy { source, to, target, .. } => { - return Err(reject_copy(source, *to, target)); + Statement::Copy { + source, + to, + target, + options: copy_options, + legacy_options, + payload, + } => { + let copy = super::copy::Copy { + source, + to: *to, + target, + options: copy_options, + legacy_options, + payload: payload.as_ref(), + }; + super::copy::translate_copy(©, schema, options, emit)? } Statement::Merge(merge) => return Err(reject_merge(merge)), Statement::Prepare { name, .. } => { diff --git a/tests/test_copy.rs b/tests/test_copy.rs index f1a87fbfc..99e457062 100644 --- a/tests/test_copy.rs +++ b/tests/test_copy.rs @@ -1,63 +1,322 @@ -//! `COPY` must be refused, not discarded. +//! `COPY ... FROM STDIN` becomes the `INSERT` of its inline rows, and every +//! other `COPY` is refused rather than discarded. //! -//! `COPY ... FROM stdin` carries its rows inline in the migration file, so -//! dropping the statement loses data with no diagnostic. It used to sit in the -//! `unsupported_statement_patterns!()` catch-all alongside the other silent -//! drops. -//! -//! Translating it to multi-row `INSERT`s would be the better answer for a -//! migration translator, and it is not implemented, so every form is refused. +//! The emitted statements run as text because that text is the artifact under +//! test, and everything read back goes through the typed diesel DSL. + +use diesel::{connection::SimpleConnection, prelude::*}; +use pg2sqlite::prelude::{Error, Pg2Sqlite, Pg2SqliteOptions}; + +diesel::define_sql_function! { + /// Reports whether generated write guards are exempt, true for a restore. + fn write_is_exempt() -> diesel::sql_types::Bool; +} -use pg2sqlite::prelude::{Pg2Sqlite, Pg2SqliteOptions}; +mod schema { + diesel::table! { + t (rowid) { + rowid -> BigInt, + a -> Nullable, + b -> Nullable, + } + } + + diesel::table! { + g (id) { + id -> Integer, + flag -> Bool, + twice -> Integer, + } + } + + diesel::table! { + notes_rls (id) { + id -> Integer, + owner -> Text, + } + } + + diesel::table! { + q (id) { + id -> Integer, + #[sql_name = "Odd Name"] + odd -> Nullable, + } + } +} const BASE: &str = "CREATE TABLE t (a INTEGER, b TEXT);"; -/// Translates `pg` and returns the error message, failing if translation -/// succeeds. -fn rejection(pg: &str) -> String { - let result = Pg2Sqlite::default() +fn translate(pg: &str, options: &Pg2SqliteOptions) -> Result, Error> { + Pg2Sqlite::default() .sql(pg) .expect("the fixture must parse") - .translate(&Pg2SqliteOptions::default()) - .map(pg2sqlite::script::Script::into_vec); - - match result { - Ok(statements) => { - panic!( - "COPY must be refused rather than dropped, but translation produced {statements:?}" - ) - } - Err(error) => error.to_string(), + .translate_to_sql(options) + .map(pg2sqlite::script::Script::into_vec) +} + +/// Runs the translation of `pg` on a new database. +fn apply(pg: &str) -> SqliteConnection { + let mut connection = SqliteConnection::establish(":memory:").expect("connect"); + for statement in translate(pg, &Pg2SqliteOptions::default()).expect("translate") { + connection.batch_execute(&statement).unwrap_or_else(|error| panic!("{error}\n{statement}")); + } + connection +} + +fn rows(connection: &mut SqliteConnection) -> Vec<(Option, Option)> { + schema::t::table + .select((schema::t::a, schema::t::b)) + .order(schema::t::rowid) + .load(connection) + .expect("read t") +} + +fn refused(pg: &str) { + let translated = translate(pg, &Pg2SqliteOptions::default()); + assert!(matches!(translated, Err(Error::TranslationRefusal(_))), "{translated:?}"); +} + +#[test] +fn a_text_payload_becomes_its_rows() { + let mut connection = apply(&format!("{BASE} COPY t (a, b) FROM stdin;\n1\t\\N\n2\ty\n\\.\n")); + assert_eq!(rows(&mut connection), [(Some(1), None), (Some(2), Some("y".to_owned()))]); +} + +#[test] +fn text_escapes_decode_and_the_null_string_matches_before_them() { + let mut connection = apply(&format!( + "{BASE} COPY t (a, b) FROM stdin;\n\ + 1\ttab\\there\n\ + 2\t\\101\\x42\\\\\\N\n\ + 3\tsplit\\\tfield\n\ + 4\t\\\\N\n\ + \\." + )); + assert_eq!( + rows(&mut connection), + [ + (Some(1), Some("tab\there".to_owned())), + (Some(2), Some("AB\\N".to_owned())), + (Some(3), Some("split\tfield".to_owned())), + (Some(4), Some("\\N".to_owned())), + ] + ); +} + +#[test] +fn a_custom_delimiter_and_null_string_apply() { + let mut connection = apply(&format!( + "{BASE} COPY t (a, b) FROM stdin WITH (DELIMITER '|', NULL 'nil');\n1|nil\n2|x\n\\.\n" + )); + assert_eq!(rows(&mut connection), [(Some(1), None), (Some(2), Some("x".to_owned()))]); +} + +#[test] +fn csv_quotes_carry_delimiters_newlines_and_doubled_quotes() { + let mut connection = apply(&format!( + "{BASE} COPY t (a, b) FROM stdin WITH (FORMAT csv, HEADER);\n\ + a,b\n\ + 1,\"one, two\"\n\ + 2,\"line\nbreak\"\n\ + 3,\"say \"\"hi\"\"\"\n\ + 4,\n\ + 5,\"\"\n\ + \\.\n" + )); + assert_eq!( + rows(&mut connection), + [ + (Some(1), Some("one, two".to_owned())), + (Some(2), Some("line\nbreak".to_owned())), + (Some(3), Some("say \"hi\"".to_owned())), + (Some(4), None), + (Some(5), Some(String::new())), + ] + ); +} + +#[test] +fn force_null_and_force_not_null_flip_the_null_match() { + let mut connection = apply(&format!( + "{BASE} COPY t (a, b) FROM stdin WITH (FORMAT csv, FORCE_NOT_NULL (b));\n1,\n\\.\n\ + COPY t (a, b) FROM stdin WITH (FORMAT csv, FORCE_NULL (b));\n2,\"\"\n\\.\n" + )); + assert_eq!(rows(&mut connection), [(Some(1), Some(String::new())), (Some(2), None)]); +} + +#[test] +fn a_copy_without_columns_fills_every_column_but_the_generated_ones() { + let mut connection = apply( + "CREATE TABLE g (id INT PRIMARY KEY, flag BOOLEAN NOT NULL, \ + twice INT GENERATED ALWAYS AS (id * 2) STORED);\n\ + COPY g FROM stdin;\n3\tt\n4\tf\n\\.\n", + ); + let loaded: Vec<(i32, bool, i32)> = schema::g::table + .select((schema::g::id, schema::g::flag, schema::g::twice)) + .order(schema::g::id) + .load(&mut connection) + .expect("read g"); + assert_eq!(loaded, [(3, true, 6), (4, false, 8)]); +} + +#[test] +fn an_empty_payload_writes_nothing() { + let mut connection = apply(&format!("{BASE} COPY t (a, b) FROM stdin;\n\\.\n")); + assert_eq!(rows(&mut connection), []); +} + +#[test] +fn a_secured_table_receives_its_rows_in_the_backing_table() { + let pg = "CREATE TABLE notes (id INT PRIMARY KEY, owner TEXT NOT NULL); + ALTER TABLE notes ENABLE ROW LEVEL SECURITY; + CREATE POLICY notes_read ON notes FOR SELECT USING (false); + COPY notes (id, owner) FROM stdin;\n1\talice\n\\.\n"; + let mut connection = SqliteConnection::establish(":memory:").expect("connect"); + write_is_exempt_utils::register_impl(&mut connection, || true) + .expect("register the write exemption"); + let options = Pg2SqliteOptions::default() + .with_rls_audit_table_name("_rls_audit") + .with_write_exemption_function("write_is_exempt"); + for statement in translate(pg, &options).expect("translate") { + connection.batch_execute(&statement).unwrap_or_else(|error| panic!("{error}\n{statement}")); } + let stored: Vec<(i32, String)> = schema::notes_rls::table + .select((schema::notes_rls::id, schema::notes_rls::owner)) + .load(&mut connection) + .expect("read the backing table"); + assert_eq!(stored, [(1, "alice".to_owned())]); +} + +#[test] +fn a_row_with_the_wrong_field_count_is_refused() { + refused(&format!("{BASE} COPY t (a, b) FROM stdin;\n1\tx\textra\n\\.\n")); + refused(&format!("{BASE} COPY t (a, b) FROM stdin;\n1\n\\.\n")); +} + +#[test] +fn a_payload_without_its_terminator_is_refused() { + refused(&format!("{BASE} COPY t (a, b) FROM stdin;\n1\tx\n")); +} + +#[test] +fn an_unterminated_csv_quote_is_refused() { + refused(&format!("{BASE} COPY t (a, b) FROM stdin WITH (FORMAT csv);\n1,\"open\n\\.\n")); +} + +#[test] +fn options_postgresql_refuses_or_the_translator_cannot_decode_are_refused() { + for options in [ + "(FORMAT binary)", + "(FORMAT json)", + "(FREEZE)", + "(ENCODING 'LATIN1')", + "(FORCE_QUOTE (a))", + "(QUOTE '''')", + "(ESCAPE '|')", + "(FORCE_NOT_NULL (b))", + "(FORMAT csv, DELIMITER '\"')", + "(FORMAT csv, FORCE_NULL (missing))", + "(FORMAT csv, NULL 'a\"b')", + "(DELIMITER ',', DELIMITER ';')", + "(DELIMITER 'a')", + "(DELIMITER '\\')", + "(DELIMITER '|', NULL 'a|b')", + ] { + refused(&format!("{BASE} COPY t (a, b) FROM stdin WITH {options};\n\\.\n")); + } + for legacy in ["BINARY", "CSV FORCE QUOTE a"] { + refused(&format!("{BASE} COPY t (a, b) FROM stdin {legacy};\n\\.\n")); + } +} + +#[test] +fn the_legacy_option_spellings_apply() { + let mut connection = apply(&format!( + "{BASE} COPY t (a, b) FROM stdin WITH DELIMITER '|' NULL 'nil';\n1|nil\n\\.\n\ + COPY t (a, b) FROM stdin WITH CSV HEADER QUOTE '|' ESCAPE '!' FORCE NOT NULL b;\n\ + a,b\n2,\n3,|x!|y!!|\n\\.\n" + )); + assert_eq!( + rows(&mut connection), + [(Some(1), None), (Some(2), Some(String::new())), (Some(3), Some("x|y!".to_owned()))] + ); +} + +#[test] +fn carriage_return_line_endings_are_read_when_every_row_uses_them() { + let mut connection = + apply(&format!("{BASE} COPY t (a, b) FROM stdin;\r\n1\tx\r\n2\ty\r\n\\.\r\n")); + assert_eq!( + rows(&mut connection), + [(Some(1), Some("x".to_owned())), (Some(2), Some("y".to_owned()))] + ); + refused(&format!("{BASE} COPY t (a, b) FROM stdin;\n1\tx\r\n2\ty\n\\.\n")); } -/// The data-loss case. The rows live in the statement, so a silent drop -/// discards them. #[test] -fn copy_from_stdin_is_rejected() { - let error = rejection(&format!("{BASE} COPY t (a, b) FROM stdin;\n1\tx\n2\ty\n\\.\n")); - assert!(error.contains("COPY"), "the error must name the statement, got: {error}"); +fn short_octal_and_hex_escapes_decode_and_invalid_utf8_is_refused() { + let mut connection = + apply(&format!("{BASE} COPY t (a, b) FROM stdin;\n5\t\\x4g\\7z\\xq\n\\.\n")); + assert_eq!(rows(&mut connection), [(Some(5), Some("\u{4}g\u{7}zxq".to_owned()))]); + refused(&format!("{BASE} COPY t (a, b) FROM stdin;\n1\t\\377\n\\.\n")); +} + +#[test] +fn a_copy_without_columns_into_an_undeclared_table_is_refused() { + refused("COPY missing FROM stdin;\n\\.\n"); } /// A file source cannot be read by the translator and SQLite has no equivalent /// statement. #[test] fn copy_from_a_file_is_rejected() { - let error = rejection(&format!("{BASE} COPY t (a, b) FROM '/tmp/rows.csv';")); - assert!(error.contains("COPY"), "the error must name the statement, got: {error}"); + refused(&format!("{BASE} COPY t (a, b) FROM '/tmp/rows.csv';")); } /// `COPY ... TO` exports rows. Dropping it silently produces no export at all. #[test] fn copy_to_stdout_is_rejected() { - let error = rejection(&format!("{BASE} COPY t TO stdout;")); - assert!(error.contains("COPY"), "the error must name the statement, got: {error}"); + refused(&format!("{BASE} COPY t TO stdout;")); } /// `COPY (SELECT ...) TO` is the query form of an export, and is refused for /// the same reason. #[test] fn copy_from_a_query_is_rejected() { - let error = rejection(&format!("{BASE} COPY (SELECT a FROM t) TO stdout;")); - assert!(error.contains("COPY"), "the error must name the statement, got: {error}"); + refused(&format!("{BASE} COPY (SELECT a FROM t) TO stdout;")); +} + +#[test] +fn the_control_character_escapes_decode() { + let mut connection = + apply(&format!("{BASE} COPY t (a, b) FROM stdin;\n6\t\\b\\f\\n\\r\\v\n\\.\n")); + assert_eq!(rows(&mut connection), [(Some(6), Some("\u{8}\u{c}\n\r\u{b}".to_owned()))]); +} + +#[test] +fn an_explicit_text_format_reads_carriage_return_rows() { + let mut connection = + apply(&format!("{BASE} COPY t (a, b) FROM stdin WITH (FORMAT text);\r1\tx\r\\.\r")); + assert_eq!(rows(&mut connection), [(Some(1), Some("x".to_owned()))]); +} + +#[test] +fn a_line_break_in_the_delimiter_or_null_string_is_refused() { + for options in ["(DELIMITER '\n')", "(NULL 'a\rb')"] { + refused(&format!("{BASE} COPY t (a, b) FROM stdin WITH {options};\n\\.\n")); + } +} + +#[test] +fn a_quoted_column_is_written_and_matched_by_its_exact_name() { + let mut connection = apply( + "CREATE TABLE q (id INT PRIMARY KEY, \"Odd Name\" TEXT);\n\ + COPY q FROM stdin WITH (FORMAT csv, FORCE_NULL (\"Odd Name\"));\n1,\"\"\n\\.\n", + ); + let loaded: Vec<(i32, Option)> = schema::q::table + .select((schema::q::id, schema::q::odd)) + .load(&mut connection) + .expect("read q"); + assert_eq!(loaded, [(1, None)]); }