fix(query): emit fill_with SQL expressions verbatim

`fill_with` is a raw SQL expression slot - whatever the user wrote is spliced
into the emitted UPDATE / INSERT ... SELECT via `Expr::cust`. But `add_column`
and `modify_column_nullable` ran it through `convert_default_for_backend`,
which is written for a column DEFAULT (a single literal or function call it is
free to canonicalise). Its PostgreSQL-cast branch split at the FIRST `::`,
lower-cased everything after it, and re-joined the halves.

A backfill such as

    (CASE WHEN plan_key::text = 'API' THEN 'MONTHLY_QUOTA' ELSE 'SEAT' END)::billing_metric

was emitted on PostgreSQL with `'api'` / `'monthly_quota'` / `'seat'`: the
comparison never matched, so the backfill silently did nothing, and the
lower-cased token is not a valid enum label so the cast failed. On MySQL and
SQLite the statement was truncated at the split point outright.

Changes:

- New `sql::fill_with::convert_fill_with_for_backend`. PostgreSQL - the dialect
  `fill_with` is authored in - always gets the value verbatim. Other backends
  only rewrite a value that is unambiguously a single simple literal, or a
  whole-string portable function spelling such as `NOW()`; anything with
  whitespace, parentheses or composite SQL keywords passes through untouched.
- `parse_pg_type_cast` now splits at the LAST *top-level* `::`, skipping
  operators inside single-quoted literals and inside parentheses, so
  `CASE WHEN tag = 'a::b' THEN 1 ELSE 2 END::integer` is no longer cut open
  inside its own string literal. Only the type name is lower-cased; the value
  is returned byte-for-byte.
- `convert_default_for_backend` recurses through cast chains, so `'x'::text::json`
  nests (`CAST(CAST('x' AS CHAR) AS JSON)` on MySQL) instead of collapsing.
- `modify_column_default.backfill` was already interpolated verbatim; locked
  with a regression test so the defect cannot be copied onto that path.

Author: devfive

.changepacks/changepack_log_Qp7rvN2xKdLm9aBcT4wZs.json

@@ -0,0 +1 @@
+{"changes":{"crates/vespertide-query/Cargo.toml":"Patch"},"note":"add_column.fill_with가 사용자 SQL 표현식을 훼손하던 버그 수정: fill_with는 raw SQL 표현식 슬롯이므로 DEFAULT 정규화(convert_default_for_backend)를 태우지 않고 그대로 방출한다. parse_pg_type_cast도 첫 번째 :: 대신 따옴표/괄호를 건너뛴 마지막 top-level :: 에서 분리하도록 수정","date":"2026-08-20T04:00:00.0000000Z"}

crates/vespertide-query/src/sql/add_column.rs

@@ -1,7 +1,8 @@
-use sea_query::{Alias, Expr, Query, Table, TableAlterStatement};
+use sea_query::{Alias, Expr, Query, Table, TableAlterStatement};
 
 use vespertide_core::{ColumnDef, TableDef};
 
+use super::fill_with::convert_fill_with_for_backend;
 use super::helpers::{
     build_create_enum_type_sql, build_sea_column_def_with_table, build_sqlite_temp_table_create,
     convert_default_for_backend, normalize_enum_default, normalize_fill_with,
@@ -77,7 +78,7 @@ pub fn build_add_column(
             columns_alias.push(alias);
         }
         let fill_expr = if let Some(fill) = normalize_fill_with(fill_with) {
-            let converted = convert_default_for_backend(fill, backend);
+            let converted = convert_fill_with_for_backend(fill, backend);
             Expr::cust(normalize_enum_default(&column.r#type, &converted))
         } else if let Some(def) = &column.default {
             let converted = convert_default_for_backend(&def.to_sql(), backend);
@@ -132,7 +133,7 @@ pub fn build_add_column(
 
         // Backfill with provided value
         if let Some(fill) = normalize_fill_with(fill_with) {
-            let fill = convert_default_for_backend(fill, backend);
+            let fill = convert_fill_with_for_backend(fill, backend);
             let update_stmt = Query::update()
                 .table(Alias::new(table))
                 .value(Alias::new(&column.name), Expr::cust(fill))
@@ -159,7 +160,7 @@ pub fn build_add_column(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::test_support::{joined_sql, joined_sql_semicolon};
+    use crate::test_support::{backend_tag, joined_sql, joined_sql_semicolon};
     use insta::{assert_snapshot, with_settings};
     use rstest::rstest;
     use vespertide_core::{ColumnType, SimpleColumnType, TableDef};
@@ -701,4 +702,142 @@ mod tests {
             assert_snapshot!(sql);
         });
     }
+
+    fn backfill_sql(backend: DatabaseBackend, column: &ColumnDef, fill: &str) -> String {
+        use crate::test_support::{col_n, table_def};
+
+        let current_schema = vec![table_def(
+            "subscription",
+            vec![
+                col_n("id", ColumnType::Simple(SimpleColumnType::Integer), false),
+                col_n(
+                    "plan_key",
+                    ColumnType::Simple(SimpleColumnType::Text),
+                    false,
+                ),
+                col_n(
+                    "plan_tag",
+                    ColumnType::Simple(SimpleColumnType::Text),
+                    false,
+                ),
+                col_n(
+                    "device_os",
+                    ColumnType::Simple(SimpleColumnType::Text),
+                    false,
+                ),
+                col_n(
+                    "device_family",
+                    ColumnType::Simple(SimpleColumnType::Text),
+                    false,
+                ),
+            ],
+            vec![],
+        )];
+        let queries = build_add_column(
+            backend,
+            "subscription",
+            column,
+            Some(fill),
+            &current_schema,
+            &[],
+        )
+        .expect("add_column with fill_with should build");
+        joined_sql_semicolon(backend, &queries)
+    }
+
+    fn not_null_column(name: &str, r#type: ColumnType) -> ColumnDef {
+        ColumnDef {
+            name: name.into(),
+            r#type,
+            nullable: false,
+            default: None,
+            comment: None,
+            primary_key: None,
+            unique: None,
+            index: None,
+            foreign_key: None,
+        }
+    }
+
+    /// Regression: a `fill_with` CASE expression comparing a text-cast column
+    /// to the uppercase literal `API` and returning `MONTHLY_QUOTA` / `SEAT`,
+    /// wrapped in parens and cast to an enum type.
+    ///
+    /// Splitting at the *first* `::` and lower-casing the remainder produced
+    /// `'api'` / `'monthly_quota'` / `'seat'`: the comparison never matched, so
+    /// the backfill silently did nothing, and the lower-cased token was not a
+    /// valid enum label so the cast failed.
+    #[rstest]
+    #[case::postgres(DatabaseBackend::Postgres)]
+    #[case::mysql(DatabaseBackend::MySql)]
+    #[case::sqlite(DatabaseBackend::Sqlite)]
+    fn fill_with_enum_cast_case_expression_is_verbatim(#[case] backend: DatabaseBackend) {
+        use vespertide_core::{ComplexColumnType, EnumValues};
+
+        const FILL: &str = "(CASE WHEN plan_key::text = 'API' THEN 'MONTHLY_QUOTA' ELSE 'SEAT' END)::billing_metric";
+
+        let column = not_null_column(
+            "metric",
+            ColumnType::Complex(ComplexColumnType::Enum {
+                name: "billing_metric".into(),
+                values: EnumValues::String(vec!["MONTHLY_QUOTA".into(), "SEAT".into()]),
+            }),
+        );
+        let sql = backfill_sql(backend, &column, FILL);
+
+        assert!(
+            sql.contains(FILL),
+            "fill_with must survive byte-for-byte, got: {sql}"
+        );
+
+        with_settings!({ snapshot_suffix => format!("fill_with_enum_cast_verbatim_{}", backend_tag(backend)) }, {
+            assert_snapshot!(sql);
+        });
+    }
+
+    /// Regression: uppercase `WINDOWS` sits *before* the first cast operator
+    /// and survived, while the `ELSE` / `END` keywords *after* it were
+    /// lower-cased — the observation that pinpointed the first-`::` split.
+    #[rstest]
+    #[case::postgres(DatabaseBackend::Postgres)]
+    #[case::mysql(DatabaseBackend::MySql)]
+    #[case::sqlite(DatabaseBackend::Sqlite)]
+    fn fill_with_json_array_case_expression_is_verbatim(#[case] backend: DatabaseBackend) {
+        const FILL: &str = "CASE WHEN device_os = 'win' THEN json_build_array('WINDOWS', device_family::text) ELSE '[]'::json END";
+
+        let column = not_null_column("os_tags", ColumnType::Simple(SimpleColumnType::Json));
+        let sql = backfill_sql(backend, &column, FILL);
+
+        assert!(
+            sql.contains(FILL),
+            "fill_with must survive byte-for-byte, got: {sql}"
+        );
+
+        with_settings!({ snapshot_suffix => format!("fill_with_json_array_verbatim_{}", backend_tag(backend)) }, {
+            assert_snapshot!(sql);
+        });
+    }
+
+    /// Regression: the comparison literal itself contains a cast operator
+    /// inside single quotes, followed by a trailing cast to integer. Splitting
+    /// on the first `::` cut the statement open inside the string literal.
+    #[rstest]
+    #[case::postgres(DatabaseBackend::Postgres)]
+    #[case::mysql(DatabaseBackend::MySql)]
+    #[case::sqlite(DatabaseBackend::Sqlite)]
+    fn fill_with_cast_operator_inside_quotes_is_verbatim(#[case] backend: DatabaseBackend) {
+        const FILL: &str = "CASE WHEN plan_tag = 'legacy::v1' THEN 1 ELSE 2 END::integer";
+
+        let column = not_null_column("tier", ColumnType::Simple(SimpleColumnType::Integer));
+        let sql = backfill_sql(backend, &column, FILL);
+
+        assert!(
+            sql.contains(FILL),
+            "fill_with must survive byte-for-byte, got: {sql}"
+        );
+
+        with_settings!({ snapshot_suffix => format!("fill_with_quoted_cast_verbatim_{}", backend_tag(backend)) }, {
+            assert_snapshot!(sql);
+        });
+    }
 }

crates/vespertide-query/src/sql/fill_with.rs

@@ -0,0 +1,213 @@
+//! Backend adaptation for `fill_with` / backfill values.
+//!
+//! `fill_with` is a **raw SQL expression slot**: whatever the user wrote is
+//! spliced into the emitted `UPDATE` / `INSERT ... SELECT` verbatim (via
+//! `Expr::cust`). That is a different contract from a column DEFAULT, which
+//! [`convert_default_for_backend`] was written for — a single literal or
+//! function call it is free to canonicalise.
+//!
+//! Running an expression through the DEFAULT path corrupted it. The
+//! PostgreSQL-cast branch split at the *first* `::`, lower-cased everything
+//! after it, and re-joined the halves, so
+//!
+//! ```sql
+//! (CASE WHEN plan_key::text = 'API' THEN 'MONTHLY_QUOTA' ELSE 'SEAT' END)::billing_metric
+//! ```
+//!
+//! was emitted with `'api'` / `'monthly_quota'` / `'seat'` — the comparison
+//! never matched (silent no-op backfill) and the lower-cased token was not a
+//! valid enum label, so the cast failed. On MySQL and SQLite the statement was
+//! truncated at the split point outright.
+//!
+//! The rule enforced here: **never mutate user SQL.**
+
+use super::helpers::{
+    TIMESTAMP_FUNCTION_SPELLINGS, UUID_FUNCTION_SPELLINGS, convert_default_for_backend,
+    find_last_top_level_cast, matches_any_spelling, quoted_literal_end,
+};
+use super::types::DatabaseBackend;
+
+/// Keywords that only occur in a *composite* SQL expression. Finding one
+/// outside a string literal proves the value is not a lone literal.
+const COMPOSITE_SQL_KEYWORDS: [&str; 16] = [
+    "case", "when", "then", "else", "end", "select", "from", "where", "and", "or", "not",
+    "between", "in", "like", "union", "join",
+];
+
+/// Adapt a `fill_with` / backfill expression for `backend`.
+///
+/// * PostgreSQL — the dialect `fill_with` is authored in — always receives the
+///   value **verbatim**.
+/// * Other backends are only allowed to rewrite a value that is unambiguously
+///   a single simple literal (or one of the portable function spellings).
+///   Anything composite passes through untouched.
+#[must_use]
+pub(crate) fn convert_fill_with_for_backend(fill: &str, backend: DatabaseBackend) -> String {
+    if backend == DatabaseBackend::Postgres || !is_simple_literal_fill(fill) {
+        return fill.to_string();
+    }
+    convert_default_for_backend(fill, backend)
+}
+
+/// Whether `fill` is safe to hand to [`convert_default_for_backend`], i.e. it
+/// is either a whole-string portable function spelling (`NOW()`,
+/// `gen_random_uuid()`, …) or a single simple literal / identifier optionally
+/// carrying one trailing `::type` cast.
+fn is_simple_literal_fill(fill: &str) -> bool {
+    let trimmed = fill.trim();
+    if matches_any_spelling(trimmed, &UUID_FUNCTION_SPELLINGS)
+        || matches_any_spelling(trimmed, &TIMESTAMP_FUNCTION_SPELLINGS)
+    {
+        return true;
+    }
+    if contains_composite_keyword(trimmed) {
+        return false;
+    }
+    let value = match find_last_top_level_cast(trimmed) {
+        Some(split) => trimmed[..split].trim(),
+        None => trimmed,
+    };
+    is_single_sql_atom(value)
+}
+
+/// Whether `value` is exactly one complete quoted string literal, or one bare
+/// token free of whitespace, parentheses, commas and quotes.
+fn is_single_sql_atom(value: &str) -> bool {
+    if value.is_empty() {
+        return false;
+    }
+    if value.starts_with('\'') {
+        return quoted_literal_end(value) == Some(value.len());
+    }
+    !value
+        .chars()
+        .any(|c| c.is_whitespace() || matches!(c, '(' | ')' | ',' | ';' | '\'' | '"'))
+}
+
+/// Whether `value` contains a [`COMPOSITE_SQL_KEYWORDS`] entry outside every
+/// single-quoted string literal.
+///
+/// Literal content is skipped because it is data, not syntax: an enum label
+/// such as `'not_started'` must not be mistaken for the `NOT` keyword and
+/// pushed onto the verbatim path, where MySQL would choke on its `::` cast.
+fn contains_composite_keyword(value: &str) -> bool {
+    let mut rest = value;
+    loop {
+        let (outside, next) = match rest.find('\'') {
+            Some(quote) => {
+                let after =
+                    quoted_literal_end(&rest[quote..]).map_or(rest.len(), |end| quote + end);
+                (&rest[..quote], &rest[after..])
+            }
+            None => (rest, ""),
+        };
+        if outside
+            .split(|c: char| !c.is_ascii_alphanumeric() && c != '_')
+            .any(|word| matches_any_spelling(word, &COMPOSITE_SQL_KEYWORDS))
+        {
+            return true;
+        }
+        if next.is_empty() {
+            return false;
+        }
+        rest = next;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rstest::rstest;
+
+    /// The three reported corruptions, at the unit level: every backend must
+    /// hand back the expression byte-for-byte.
+    #[rstest]
+    #[case::enum_cast(
+        "(CASE WHEN plan_key::text = 'API' THEN 'MONTHLY_QUOTA' ELSE 'SEAT' END)::billing_metric"
+    )]
+    #[case::json_array(
+        "CASE WHEN device_os = 'win' THEN json_build_array('WINDOWS', device_family::text) ELSE '[]'::json END"
+    )]
+    #[case::cast_inside_quotes("CASE WHEN plan_tag = 'legacy::v1' THEN 1 ELSE 2 END::integer")]
+    fn composite_expressions_survive_verbatim(#[case] fill: &str) {
+        for backend in [
+            DatabaseBackend::Postgres,
+            DatabaseBackend::MySql,
+            DatabaseBackend::Sqlite,
+        ] {
+            assert_eq!(
+                convert_fill_with_for_backend(fill, backend),
+                fill,
+                "{backend:?} must not rewrite a fill_with expression"
+            );
+        }
+    }
+
+    /// PostgreSQL is the authoring dialect, so even a value the DEFAULT path
+    /// would canonicalise is emitted exactly as written.
+    #[rstest]
+    #[case("NOW()")]
+    #[case("gen_random_uuid()")]
+    #[case("'[]'::json")]
+    #[case("0")]
+    fn postgres_never_rewrites(#[case] fill: &str) {
+        assert_eq!(
+            convert_fill_with_for_backend(fill, DatabaseBackend::Postgres),
+            fill
+        );
+    }
+
+    #[rstest]
+    #[case::now_mysql("NOW()", DatabaseBackend::MySql, "CURRENT_TIMESTAMP")]
+    #[case::now_sqlite("NOW()", DatabaseBackend::Sqlite, "CURRENT_TIMESTAMP")]
+    #[case::uuid_mysql("gen_random_uuid()", DatabaseBackend::MySql, "(UUID())")]
+    #[case::uuid_sqlite(
+        "gen_random_uuid()",
+        DatabaseBackend::Sqlite,
+        "lower(hex(randomblob(16)))"
+    )]
+    #[case::json_cast_mysql("'[]'::json", DatabaseBackend::MySql, "CAST('[]' AS JSON)")]
+    #[case::json_cast_sqlite("'[]'::json", DatabaseBackend::Sqlite, "'[]'")]
+    #[case::int_cast_mysql("0::integer", DatabaseBackend::MySql, "CAST(0 AS SIGNED)")]
+    #[case::identifier_cast_sqlite("legacy_id::text", DatabaseBackend::Sqlite, "legacy_id")]
+    #[case::empty_literal_mysql("''", DatabaseBackend::MySql, "''")]
+    #[case::plain_number_sqlite("0", DatabaseBackend::Sqlite, "0")]
+    fn simple_literals_still_convert_cross_backend(
+        #[case] fill: &str,
+        #[case] backend: DatabaseBackend,
+        #[case] expected: &str,
+    ) {
+        assert_eq!(convert_fill_with_for_backend(fill, backend), expected);
+    }
+
+    #[rstest]
+    #[case::plain_number("0", true)]
+    #[case::quoted_literal("'active'", true)]
+    #[case::quoted_literal_with_space("'in progress'", true)]
+    #[case::quoted_literal_with_keyword_inside("'not_started'::user_status", true)]
+    #[case::identifier_cast("legacy_id::text", true)]
+    #[case::portable_function("NOW()", true)]
+    #[case::nested_uuid_function("lower(hex(randomblob(16)))", true)]
+    #[case::empty("", false)]
+    #[case::whitespace_only("   ", false)]
+    #[case::function_call("json_build_array('a')", false)]
+    #[case::concatenation("'a' || 'b'", false)]
+    #[case::bare_keyword("END", false)]
+    #[case::case_expression("CASE WHEN a = 1 THEN 'x' ELSE 'y' END", false)]
+    #[case::parenthesised_cast("(a + b)::integer", false)]
+    #[case::unterminated_literal("'oops", false)]
+    fn simple_literal_classification(#[case] fill: &str, #[case] expected: bool) {
+        assert_eq!(is_simple_literal_fill(fill), expected, "input: {fill}");
+    }
+
+    /// A keyword inside a string literal is data. Without the quote-skipping
+    /// scan, `'not_started'::user_status` would take the verbatim path and
+    /// leave an unusable `::` cast in the MySQL statement.
+    #[test]
+    fn keyword_inside_string_literal_is_not_syntax() {
+        assert!(!contains_composite_keyword("'not_started'::user_status"));
+        assert!(contains_composite_keyword("a IN (1, 2)"));
+        assert!(!contains_composite_keyword("weekend_total::integer"));
+        assert!(!contains_composite_keyword("''"));
+    }
+}