fix(translate): repair dialect detection — UPSERT routing, whitespace, MySQL quoting, dead code

CRITICAL BUG-1: UPSERT routing generated invalid ON CONFLICT for MySQL,
MSSQL, Snowflake, Oracle, DuckDB. Now uses explicit UPSERT_SUPPORTED_DIALECTS
({'postgresql', 'redshift'}) — unknown dialects get plain INSERT.

BUG-2: _extract_dialect did not strip whitespace, inconsistent with
get_dialect_from_database. Fixed guard + normalized value.

BUG-3: Whitespace-only input ('   ', '\n') returned itself instead of 'unknown'.

BUG-4: Dead code — removed 'greenplum' from POSTGRESQL_DIALECTS
(always normalized to 'postgresql').

BUG-5: MySQL identifier quoting used double quotes instead of native backticks.
Added BACKTICK_DIALECTS set.

MOCK-FIX: test_at_least_one_row_per_batch patched wrong path
(executor.estimate_token_budget -> _batch_sizer.estimate_token_budget).

Adds 131 orthogonal tests covering all 4 dialect detection code paths
across input formats, normalization, routing, quoting, encoding,
schema validation, and cross-component consistency.

backend/src/plugins/translate/__tests__/test_dialect_detection_orthogonal.py

@@ -0,0 +1,932 @@
+# #region DialectDetectionOrthogonalTests [C:4] [TYPE Module] [SEMANTICS test,dialect,orthogonal,postgres,clickhouse]
+# @BRIEF Orthogonal critical testing of database dialect detection algorithm across all 4 code paths.
+# @LAYER Test
+# @RELATION BINDS_TO -> [_extract_dialect]
+# @RELATION BINDS_TO -> [get_dialect_from_database]
+# @RELATION BINDS_TO -> [SQLGenerator]
+# @RELATION BINDS_TO -> [validate_target_table_schema]
+# @TEST_CONTRACT Dialect detection is consistent across all code paths for equivalent inputs.
+# @TEST_INVARIANT Normalized dialect from any path MUST be a member of SUPPORTED_DIALECTS or "unknown".
+# @TEST_EDGE Case sensitivity, whitespace, URI variants, driver suffixes, unknown dialects, empty inputs.
+
+import pytest
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+from src.plugins.translate.service_utils import _extract_dialect
+from src.plugins.translate.service_datasource import (
+    get_dialect_from_database,
+    SUPPORTED_DIALECTS,
+)
+from src.plugins.translate.sql_generator import (
+    SQLGenerator,
+    POSTGRESQL_DIALECTS,
+    CLICKHOUSE_DIALECTS,
+    UPSERT_SUPPORTED_DIALECTS,
+    BACKTICK_DIALECTS,
+    _quote_identifier,
+    _encode_sql_value,
+)
+
+
+# #region TestExtractDialect_Orthogonal [TYPE Class]
+# @BRIEF Orthogonal tests for _extract_dialect: input format x normalization x edge cases.
+class TestExtractDialect_Orthogonal:
+
+    # region test_plain_engine_names [TYPE Function]
+    # @PURPOSE: Plain engine names return as-is (lowercased).
+    @pytest.mark.parametrize("input_val,expected", [
+        ("postgresql", "postgresql"),
+        ("clickhouse", "clickhouse"),
+        ("mysql", "mysql"),
+        ("sqlite", "sqlite"),
+        ("mssql", "mssql"),
+        ("redshift", "redshift"),
+        ("snowflake", "snowflake"),
+        ("bigquery", "bigquery"),
+    ])
+    def test_plain_engine_names(self, input_val: str, expected: str) -> None:
+        assert _extract_dialect(input_val) == expected
+    # endregion test_plain_engine_names
+
+    # region test_uri_formats [TYPE Function]
+    # @PURPOSE: URI formats extract scheme correctly.
+    @pytest.mark.parametrize("input_val,expected", [
+        ("postgresql://localhost:5432/mydb", "postgresql"),
+        ("clickhousedb://localhost:8123/mydb", "clickhouse"),
+        ("clickhouse://localhost:8123/mydb", "clickhouse"),
+        ("mysql://localhost:3306/mydb", "mysql"),
+        ("sqlite:///mydb.sqlite", "sqlite"),
+        ("mssql://localhost:1433/mydb", "mssql"),
+        ("redshift://localhost:5439/mydb", "redshift"),
+    ])
+    def test_uri_formats(self, input_val: str, expected: str) -> None:
+        assert _extract_dialect(input_val) == expected
+    # endregion test_uri_formats
+
+    # region test_uri_with_driver_suffix [TYPE Function]
+    # @PURPOSE: URI with driver suffix (e.g. postgresql+asyncpg) strips driver correctly.
+    @pytest.mark.parametrize("input_val,expected", [
+        ("postgresql+asyncpg://localhost:5432/mydb", "postgresql"),
+        ("postgresql+psycopg2://localhost:5432/mydb", "postgresql"),
+        ("clickhousedb+native://localhost:8123/mydb", "clickhouse"),
+        ("mysql+pymysql://localhost:3306/mydb", "mysql"),
+        ("mysql+mysqldb://localhost:3306/mydb", "mysql"),
+    ])
+    def test_uri_with_driver_suffix(self, input_val: str, expected: str) -> None:
+        assert _extract_dialect(input_val) == expected
+    # endregion test_uri_with_driver_suffix
+
+    # region test_normalization_mappings [TYPE Function]
+    # @PURPOSE: Known Superset backend variants are normalized correctly.
+    @pytest.mark.parametrize("input_val,expected", [
+        ("clickhousedb", "clickhouse"),
+        ("clickhousedb://host:8123/db", "clickhouse"),
+        ("greenplum", "postgresql"),
+        ("greenplum://host:5432/db", "postgresql"),
+    ])
+    def test_normalization_mappings(self, input_val: str, expected: str) -> None:
+        assert _extract_dialect(input_val) == expected
+    # endregion test_normalization_mappings
+
+    # region test_case_sensitivity [TYPE Function]
+    # @PURPOSE: Mixed-case inputs are lowercased before normalization.
+    @pytest.mark.parametrize("input_val,expected", [
+        ("PostgreSQL", "postgresql"),
+        ("POSTGRESQL", "postgresql"),
+        ("ClickHouse", "clickhouse"),
+        ("CLICKHOUSE", "clickhouse"),
+        ("ClickHouseDB", "clickhouse"),
+        ("CLICKHOUSEDB", "clickhouse"),
+        ("GreenPlum", "postgresql"),
+        ("GREENPLUM", "postgresql"),
+        ("MySQL", "mysql"),
+        ("MYSQL", "mysql"),
+    ])
+    def test_case_sensitivity(self, input_val: str, expected: str) -> None:
+        assert _extract_dialect(input_val) == expected
+    # endregion test_case_sensitivity
+
+    # region test_whitespace_stripping [TYPE Function]
+    # @PURPOSE: Whitespace in input is stripped before normalization.
+    def test_leading_whitespace_stripped(self) -> None:
+        assert _extract_dialect(" postgresql") == "postgresql"
+
+    def test_trailing_whitespace_stripped(self) -> None:
+        assert _extract_dialect("postgresql ") == "postgresql"
+
+    def test_leading_and_trailing_whitespace(self) -> None:
+        assert _extract_dialect("  clickhouse  ") == "clickhouse"
+    # endregion test_whitespace_stripping
+
+    # region test_empty_and_none_equivalent [TYPE Function]
+    # @PURPOSE: Empty/None-equivalent inputs return "unknown".
+    @pytest.mark.parametrize("input_val", [
+        "",
+        None,
+    ])
+    def test_empty_and_none_equivalent(self, input_val) -> None:
+        # None is falsy, returns "unknown" immediately
+        # Empty string is falsy, returns "unknown" immediately
+        assert _extract_dialect(input_val) == "unknown"
+    # endregion test_empty_and_none_equivalent
+
+    # region test_whitespace_only_treated_as_empty [TYPE Function]
+    # @PURPOSE: Whitespace-only input returns "unknown".
+    def test_whitespace_only_returns_unknown(self) -> None:
+        assert _extract_dialect("   ") == "unknown"
+
+    def test_newline_only_returns_unknown(self) -> None:
+        assert _extract_dialect("\n") == "unknown"
+
+    def test_tab_only_returns_unknown(self) -> None:
+        assert _extract_dialect("\t") == "unknown"
+    # endregion test_whitespace_only_treated_as_empty
+
+    # region test_unknown_dialects [TYPE Function]
+    # @PURPOSE: Unknown dialects pass through as-is (lowercased).
+    @pytest.mark.parametrize("input_val,expected", [
+        ("duckdb", "duckdb"),
+        ("oracle", "oracle"),
+        ("presto", "presto"),
+        ("trino", "trino"),
+        ("druid", "druid"),
+        ("hive", "hive"),
+        ("spark", "spark"),
+        ("databricks", "databricks"),
+        ("unknown_backend", "unknown_backend"),
+    ])
+    def test_unknown_dialects(self, input_val: str, expected: str) -> None:
+        assert _extract_dialect(input_val) == expected
+    # endregion test_unknown_dialects
+
+    # region test_malformed_uris [TYPE Function]
+    # @PURPOSE: Malformed URIs should not crash.
+    @pytest.mark.parametrize("input_val", [
+        "://no-scheme",
+        "postgresql+://weird",
+        "postgresql+asyncpg+extra://host",
+        "://",
+    ])
+    def test_malformed_uris_no_crash(self, input_val: str) -> None:
+        # Should not raise — returns something (possibly wrong, but not crash)
+        result = _extract_dialect(input_val)
+        assert isinstance(result, str)
+    # endregion test_malformed_uris
+
+    # region test_redshift_not_normalized [TYPE Function]
+    # @PURPOSE: redshift is NOT normalized to postgresql by _extract_dialect.
+    # This is intentional — redshift is a distinct dialect in Superset.
+    def test_redshift_not_normalized(self) -> None:
+        assert _extract_dialect("redshift") == "redshift"
+        assert _extract_dialect("redshift://host:5439/db") == "redshift"
+    # endregion test_redshift_not_normalized
+# #endregion TestExtractDialect_Orthogonal
+
+
+# #region TestGetDialectFromDatabase_Orthogonal [TYPE Class]
+# @BRIEF Orthogonal tests for get_dialect_from_database: record format x normalization x validation.
+class TestGetDialectFromDatabase_Orthogonal:
+
+    # region test_backend_key [TYPE Function]
+    # @PURPOSE: Extracts from "backend" key.
+    @pytest.mark.parametrize("backend,expected", [
+        ("postgresql", "postgresql"),
+        ("clickhouse", "clickhouse"),
+        ("clickhousedb", "clickhouse"),
+        ("greenplum", "postgresql"),
+        ("redshift", "redshift"),
+        ("mysql", "mysql"),
+    ])
+    def test_backend_key(self, backend: str, expected: str) -> None:
+        record = {"backend": backend}
+        assert get_dialect_from_database(record) == expected
+    # endregion test_backend_key
+
+    # region test_engine_key_fallback [TYPE Function]
+    # @PURPOSE: Falls back to "engine" key when "backend" is missing.
+    def test_engine_key_fallback(self) -> None:
+        record = {"engine": "clickhousedb"}
+        assert get_dialect_from_database(record) == "clickhouse"
+
+    def test_backend_takes_precedence_over_engine(self) -> None:
+        record = {"backend": "postgresql", "engine": "mysql"}
+        assert get_dialect_from_database(record) == "postgresql"
+    # endregion test_engine_key_fallback
+
+    # region test_case_sensitivity [TYPE Function]
+    # @PURPOSE: Case-insensitive matching.
+    @pytest.mark.parametrize("backend,expected", [
+        ("PostgreSQL", "postgresql"),
+        ("CLICKHOUSEDB", "clickhouse"),
+        ("GreenPlum", "postgresql"),
+        ("MySQL", "mysql"),
+    ])
+    def test_case_sensitivity(self, backend: str, expected: str) -> None:
+        record = {"backend": backend}
+        assert get_dialect_from_database(record) == expected
+    # endregion test_case_sensitivity
+
+    # region test_whitespace_stripping [TYPE Function]
+    # @PURPOSE: Whitespace IS stripped in get_dialect_from_database (unlike _extract_dialect).
+    @pytest.mark.parametrize("backend,expected", [
+        (" postgresql ", "postgresql"),
+        ("clickhouse\n", "clickhouse"),
+        ("  greenplum  ", "postgresql"),
+    ])
+    def test_whitespace_stripping(self, backend: str, expected: str) -> None:
+        record = {"backend": backend}
+        assert get_dialect_from_database(record) == expected
+    # endregion test_whitespace_stripping
+
+    # region test_empty_backend_raises [TYPE Function]
+    # @PURPOSE: Empty backend raises ValueError.
+    @pytest.mark.parametrize("record", [
+        {"backend": ""},
+        {"backend": "   "},
+        {"backend": None},
+        {},
+    ])
+    def test_empty_backend_raises(self, record: dict) -> None:
+        with pytest.raises(ValueError, match="Could not determine database dialect"):
+            get_dialect_from_database(record)
+    # endregion test_empty_backend_raises
+
+    # region test_unsupported_dialect_raises [TYPE Function]
+    # @PURPOSE: Unsupported dialect raises ValueError with helpful message.
+    def test_unsupported_dialect_raises(self) -> None:
+        record = {"backend": "cobol_db"}
+        with pytest.raises(ValueError, match="Unsupported database dialect"):
+            get_dialect_from_database(record)
+
+    def test_unsupported_dialect_error_lists_supported(self) -> None:
+        record = {"backend": "cobol_db"}
+        with pytest.raises(ValueError) as exc_info:
+            get_dialect_from_database(record)
+        error_msg = str(exc_info.value)
+        # Error message should list supported dialects
+        assert "Supported dialects" in error_msg
+        assert "postgresql" in error_msg
+        assert "clickhouse" in error_msg
+    # endregion test_unsupported_dialect_raises
+
+    # region test_all_supported_dialects [TYPE Function]
+    # @PURPOSE: Every dialect in SUPPORTED_DIALECTS is accepted.
+    def test_all_supported_dialects_accepted(self) -> None:
+        for dialect in SUPPORTED_DIALECTS:
+            record = {"backend": dialect}
+            result = get_dialect_from_database(record)
+            assert result in SUPPORTED_DIALECTS, f"{dialect} -> {result} not in SUPPORTED_DIALECTS"
+    # endregion test_all_supported_dialects
+# #endregion TestGetDialectFromDatabase_Orthogonal
+
+
+# #region TestCrossComponentConsistency [TYPE Class]
+# @BRIEF Tests that all 4 dialect detection points produce consistent results.
+# @TEST_INVARIANT For any Superset backend value, all normalization paths MUST agree.
+class TestCrossComponentConsistency:
+
+    # region test_clickhousedb_consistency [TYPE Function]
+    # @PURPOSE: "clickhousedb" MUST normalize to "clickhouse" everywhere.
+    def test_clickhousedb_consistency(self) -> None:
+        # _extract_dialect
+        assert _extract_dialect("clickhousedb") == "clickhouse"
+        assert _extract_dialect("clickhousedb://host:8123/db") == "clickhouse"
+
+        # get_dialect_from_database
+        assert get_dialect_from_database({"backend": "clickhousedb"}) == "clickhouse"
+
+        # CLICKHOUSE_DIALECTS set
+        assert "clickhouse" in CLICKHOUSE_DIALECTS
+    # endregion test_clickhousedb_consistency
+
+    # region test_greenplum_consistency [TYPE Function]
+    # @PURPOSE: "greenplum" MUST normalize to "postgresql" everywhere.
+    def test_greenplum_consistency(self) -> None:
+        # _extract_dialect
+        assert _extract_dialect("greenplum") == "postgresql"
+        assert _extract_dialect("greenplum://host:5432/db") == "postgresql"
+
+        # get_dialect_from_database
+        assert get_dialect_from_database({"backend": "greenplum"}) == "postgresql"
+
+        # POSTGRESQL_DIALECTS set — greenplum entry is dead code since it's normalized away
+        assert "postgresql" in POSTGRESQL_DIALECTS
+    # endregion test_greenplum_consistency
+
+    # region test_redshift_inconsistency [TYPE Function]
+    # @PURPOSE: "redshift" is NOT normalized by _extract_dialect but IS in POSTGRESQL_DIALECTS.
+    # This is a DESIGN INCONSISTENCY — redshift passes through as "redshift" but
+    # SQLGenerator treats it as PostgreSQL-compatible.
+    def test_redshift_in_postgresql_dialects(self) -> None:
+        # _extract_dialect returns "redshift" as-is
+        assert _extract_dialect("redshift") == "redshift"
+
+        # get_dialect_from_database returns "redshift" as-is
+        assert get_dialect_from_database({"backend": "redshift"}) == "redshift"
+
+        # But SQLGenerator treats redshift as PostgreSQL-compatible
+        assert "redshift" in POSTGRESQL_DIALECTS
+
+        # This means: database_dialect="redshift" -> SQLGenerator uses ON CONFLICT
+        # Redshift DOES support ON CONFLICT (it's PostgreSQL-derived), so this is correct behavior.
+        # But the inconsistency is that redshift is NOT in the normalization maps.
+    # endregion test_redshift_inconsistency
+
+    # region test_whitespace_consistency_between_functions [TYPE Function]
+    # @PURPOSE: Both functions strip whitespace consistently now.
+    def test_whitespace_consistency(self) -> None:
+        # Both functions strip whitespace
+        assert get_dialect_from_database({"backend": " postgresql "}) == "postgresql"
+        assert _extract_dialect(" postgresql") == "postgresql"
+        assert _extract_dialect(" clickhouse ") == "clickhouse"
+    # endregion test_whitespace_consistency_between_functions
+
+    # region test_clickhousedb_in_clickhouse_dialects_defensive [TYPE Function]
+    # @PURPOSE: "clickhousedb" stays in CLICKHOUSE_DIALECTS for defensive
+    # quoting/encoding in case it's passed directly to _quote_identifier or _encode_sql_value.
+    def test_clickhousedb_in_set_is_defensive(self) -> None:
+        # "clickhousedb" is in quoting/encoding sets for defensive handling
+        assert "clickhousedb" in CLICKHOUSE_DIALECTS
+        assert "clickhousedb" in BACKTICK_DIALECTS
+
+        # Normalization still works correctly
+        assert _extract_dialect("clickhousedb") == "clickhouse"
+        assert get_dialect_from_database({"backend": "clickhousedb"}) == "clickhouse"
+
+        # But quoting/encoding functions handle it defensively
+        from src.plugins.translate.sql_generator import _quote_identifier
+        assert _quote_identifier("my_col", "clickhousedb") == "`my_col`"
+    # endregion test_clickhousedb_in_clickhouse_dialects_defensive
+
+    # region test_greenplum_normalized_away [TYPE Function]
+    # @PURPOSE: "greenplum" is normalized to "postgresql" and NOT in POSTGRESQL_DIALECTS.
+    # The UPSERT routing is via UPSERT_SUPPORTED_DIALECTS which doesn't need it either.
+    def test_greenplum_normalized_away(self) -> None:
+        # greenplum is NOT in POSTGRESQL_DIALECTS (cleaned up)
+        assert "greenplum" not in POSTGRESQL_DIALECTS
+        assert "greenplum" not in UPSERT_SUPPORTED_DIALECTS
+        # But it still normalizes correctly
+        assert _extract_dialect("greenplum") == "postgresql"
+        assert get_dialect_from_database({"backend": "greenplum"}) == "postgresql"
+    # endregion test_greenplum_normalized_away
+
+    # region test_non_postgresql_upsert_falls_back_to_insert [TYPE Function]
+    # @PURPOSE: Non-PostgreSQL dialects with MERGE strategy fall back to plain INSERT.
+    # This prevents generating invalid ON CONFLICT syntax for MySQL, MSSQL, Snowflake, etc.
+    def test_mysql_upsert_falls_back_to_insert(self) -> None:
+        rows = [{"id": 1, "name": "test"}]
+        sql, count = SQLGenerator.generate(
+            dialect="mysql",
+            target_schema=None,
+            target_table="t",
+            columns=["id", "name"],
+            rows=rows,
+            key_columns=["id"],
+            upsert_strategy="MERGE",
+        )
+        # MySQL does NOT support ON CONFLICT — falls back to plain INSERT
+        assert "ON CONFLICT" not in sql
+        assert "INSERT INTO" in sql
+        assert count == 1
+
+    def test_mssql_upsert_falls_back_to_insert(self) -> None:
+        rows = [{"id": 1, "name": "test"}]
+        sql, count = SQLGenerator.generate(
+            dialect="mssql",
+            target_schema=None,
+            target_table="t",
+            columns=["id", "name"],
+            rows=rows,
+            key_columns=["id"],
+            upsert_strategy="MERGE",
+        )
+        assert "ON CONFLICT" not in sql
+        assert "INSERT INTO" in sql
+
+    def test_snowflake_upsert_falls_back_to_insert(self) -> None:
+        rows = [{"id": 1, "name": "test"}]
+        sql, count = SQLGenerator.generate(
+            dialect="snowflake",
+            target_schema=None,
+            target_table="t",
+            columns=["id", "name"],
+            rows=rows,
+            key_columns=["id"],
+            upsert_strategy="MERGE",
+        )
+        assert "ON CONFLICT" not in sql
+        assert "INSERT INTO" in sql
+    # endregion test_non_postgresql_upsert_falls_back_to_insert
+
+    # region test_postgresql_and_clickhouse_sql_correctness [TYPE Function]
+    # @PURPOSE: Verify that the two primary dialects generate correct SQL.
+    def test_postgresql_upsert_correct(self) -> None:
+        rows = [{"id": 1, "name": "test"}]
+        sql, count = SQLGenerator.generate(
+            dialect="postgresql",
+            target_schema="public",
+            target_table="t",
+            columns=["id", "name"],
+            rows=rows,
+            key_columns=["id"],
+            upsert_strategy="MERGE",
+        )
+        assert "ON CONFLICT" in sql
+        assert "DO UPDATE SET" in sql
+        assert count == 1
+
+    def test_clickhouse_insert_correct(self) -> None:
+        rows = [{"id": 1, "name": "test"}]
+        sql, count = SQLGenerator.generate(
+            dialect="clickhouse",
+            target_schema="default",
+            target_table="t",
+            columns=["id", "name"],
+            rows=rows,
+            key_columns=["id"],
+            upsert_strategy="MERGE",
+        )
+        assert "ON CONFLICT" not in sql
+        assert "INSERT INTO" in sql
+        assert count == 1
+    # endregion test_postgresql_and_clickhouse_sql_correctness
+# #endregion TestCrossComponentConsistency
+
+
+# #region TestDialectQuoting_Orthogonal [TYPE Class]
+# @BRIEF Orthogonal tests for identifier quoting across dialects.
+class TestDialectQuoting_Orthogonal:
+
+    # region test_postgresql_quoting [TYPE Function]
+    # @PURPOSE: PostgreSQL and Redshift use double quotes. Greenplum (not in POSTGRESQL_DIALECTS) falls through to ANSI double-quote.
+    @pytest.mark.parametrize("dialect", ["postgresql", "redshift", "greenplum"])
+    def test_postgresql_family_quoting(self, dialect: str) -> None:
+        assert _quote_identifier("my_col", dialect) == '"my_col"'
+        assert _quote_identifier("MyCol", dialect) == '"MyCol"'
+        assert _quote_identifier("select", dialect) == '"select"'  # reserved word
+    # endregion test_postgresql_quoting
+
+    # region test_clickhouse_quoting [TYPE Function]
+    # @PURPOSE: ClickHouse uses backticks.
+    @pytest.mark.parametrize("dialect", ["clickhouse", "clickhousedb"])
+    def test_clickhouse_family_quoting(self, dialect: str) -> None:
+        assert _quote_identifier("my_col", dialect) == "`my_col`"
+        assert _quote_identifier("MyCol", dialect) == "`MyCol`"
+    # endregion test_clickhouse_quoting
+
+    # region test_unknown_dialect_defaults_to_double_quotes [TYPE Function]
+    # @PURPOSE: Unknown dialects (not in POSTGRESQL_DIALECTS or BACKTICK_DIALECTS) default to ANSI double-quote.
+    @pytest.mark.parametrize("dialect", ["mssql", "snowflake", "duckdb", "oracle", "unknown"])
+    def test_unknown_dialect_defaults_to_double_quotes(self, dialect: str) -> None:
+        # These are NOT in POSTGRESQL_DIALECTS or BACKTICK_DIALECTS
+        # They fall through to the else branch -> double quotes
+        assert _quote_identifier("my_col", dialect) == '"my_col"'
+    # endregion test_unknown_dialect_defaults_to_double_quotes
+
+    # region test_special_characters_in_identifiers [TYPE Function]
+    # @PURPOSE: Identifiers with special characters are handled.
+    def test_identifier_with_spaces(self) -> None:
+        result = _quote_identifier("my column", "postgresql")
+        assert result == '"my column"'
+
+    def test_identifier_with_hyphens(self) -> None:
+        result = _quote_identifier("my-column", "clickhouse")
+        assert result == "`my-column`"
+
+    def test_empty_identifier(self) -> None:
+        assert _quote_identifier("", "postgresql") == ""
+        assert _quote_identifier("", "clickhouse") == ""
+    # endregion test_special_characters_in_identifiers
+
+    # region test_mysql_uses_backticks [TYPE Function]
+    # @PURPOSE: MySQL uses backticks natively (now fixed).
+    def test_mysql_uses_backticks(self) -> None:
+        assert _quote_identifier("my_col", "mysql") == "`my_col`"
+        assert _quote_identifier("select", "mysql") == "`select`"
+    # endregion test_mysql_uses_backticks
+# #endregion TestDialectQuoting_Orthogonal
+
+
+# #region TestDialectEncoding_Orthogonal [TYPE Class]
+# @BRIEF Orthogonal tests for value encoding across dialects.
+class TestDialectEncoding_Orthogonal:
+
+    # region test_clickhouse_timestamp_normalization [TYPE Function]
+    # @PURPOSE: ClickHouse normalizes Unix timestamps to date strings.
+    @pytest.mark.parametrize("value,expected", [
+        (1726358400, "'2024-09-15'"),       # Unix timestamp (seconds)
+        (1726358400000, "'2024-09-15'"),    # Unix timestamp (milliseconds)
+        ("1726358400", "'2024-09-15'"),     # String timestamp
+        ("1726358400000.0", "'2024-09-15'"), # String timestamp with decimal
+    ])
+    def test_clickhouse_timestamp_normalization(self, value: Any, expected: str) -> None:
+        assert _encode_sql_value(value, dialect="clickhouse") == expected
+    # endregion test_clickhouse_timestamp_normalization
+
+    # region test_postgresql_no_timestamp_normalization [TYPE Function]
+    # @PURPOSE: PostgreSQL does NOT normalize timestamps.
+    def test_postgresql_no_timestamp_normalization(self) -> None:
+        # PostgreSQL keeps timestamps as-is (numeric)
+        assert _encode_sql_value(1726358400, dialect="postgresql") == "1726358400"
+        assert _encode_sql_value("1726358400", dialect="postgresql") == "'1726358400'"
+    # endregion test_postgresql_no_timestamp_normalization
+
+    # region test_none_dialect_no_normalization [TYPE Function]
+    # @PURPOSE: No dialect specified -> no timestamp normalization.
+    def test_none_dialect_no_normalization(self) -> None:
+        assert _encode_sql_value(1726358400, dialect=None) == "1726358400"
+    # endregion test_none_dialect_no_normalization
+
+    # region test_boolean_encoding [TYPE Function]
+    # @PURPOSE: Booleans encode as TRUE/FALSE (ANSI SQL).
+    def test_boolean_encoding(self) -> None:
+        assert _encode_sql_value(True, dialect="postgresql") == "TRUE"
+        assert _encode_sql_value(False, dialect="postgresql") == "FALSE"
+        assert _encode_sql_value(True, dialect="clickhouse") == "TRUE"
+        assert _encode_sql_value(False, dialect="clickhouse") == "FALSE"
+        # Note: ClickHouse uses 1/0 for booleans, but we encode as TRUE/FALSE
+        # This may cause issues with strict ClickHouse type checking
+    # endregion test_boolean_encoding
+
+    # region test_null_encoding [TYPE Function]
+    # @PURPOSE: None encodes as NULL for all dialects.
+    def test_null_encoding(self) -> None:
+        assert _encode_sql_value(None, dialect="postgresql") == "NULL"
+        assert _encode_sql_value(None, dialect="clickhouse") == "NULL"
+        assert _encode_sql_value(None, dialect="mysql") == "NULL"
+    # endregion test_null_encoding
+# #endregion TestDialectEncoding_Orthogonal
+
+
+# #region TestDialectRouting_Orthogonal [TYPE Class]
+# @BRIEF Tests for SQLGenerator dialect routing logic (line 286).
+# @TEST_INVARIANT The routing condition `dialect in POSTGRESQL_DIALECTS or dialect not in CLICKHOUSE_DIALECTS`
+#   MUST correctly classify all supported dialects.
+class TestDialectRouting_Orthogonal:
+
+    # region test_postgresql_dialects_route_to_upsert [TYPE Function]
+    # @PURPOSE: All UPSERT_SUPPORTED_DIALECTS members route to UPSERT path.
+    @pytest.mark.parametrize("dialect", UPSERT_SUPPORTED_DIALECTS)
+    def test_postgresql_dialects_route_to_upsert(self, dialect: str) -> None:
+        rows = [{"id": 1, "name": "test"}]
+        sql, _ = SQLGenerator.generate(
+            dialect=dialect,
+            target_schema=None,
+            target_table="t",
+            columns=["id", "name"],
+            rows=rows,
+            key_columns=["id"],
+            upsert_strategy="MERGE",
+        )
+        assert "ON CONFLICT" in sql, f"{dialect} should generate ON CONFLICT"
+    # endregion test_postgresql_dialects_route_to_upsert
+
+    # region test_clickhouse_dialects_route_to_insert [TYPE Function]
+    # @PURPOSE: All CLICKHOUSE_DIALECTS members route to plain INSERT path.
+    @pytest.mark.parametrize("dialect", CLICKHOUSE_DIALECTS)
+    def test_clickhouse_dialects_route_to_insert(self, dialect: str) -> None:
+        rows = [{"id": 1, "name": "test"}]
+        sql, _ = SQLGenerator.generate(
+            dialect=dialect,
+            target_schema=None,
+            target_table="t",
+            columns=["id", "name"],
+            rows=rows,
+            key_columns=["id"],
+            upsert_strategy="MERGE",
+        )
+        assert "ON CONFLICT" not in sql, f"{dialect} should NOT generate ON CONFLICT"
+        assert "INSERT INTO" in sql
+    # endregion test_clickhouse_dialects_route_to_insert
+
+    # region test_unknown_dialects_route_to_insert [TYPE Function]
+    # @PURPOSE: Unknown dialects route to plain INSERT (safe fallback).
+    # No longer generates invalid ON CONFLICT syntax for unsupported dialects.
+    @pytest.mark.parametrize("dialect", ["mysql", "mssql", "snowflake", "oracle", "duckdb"])
+    def test_unknown_dialects_route_to_insert(self, dialect: str) -> None:
+        rows = [{"id": 1, "name": "test"}]
+        sql, _ = SQLGenerator.generate(
+            dialect=dialect,
+            target_schema=None,
+            target_table="t",
+            columns=["id", "name"],
+            rows=rows,
+            key_columns=["id"],
+            upsert_strategy="MERGE",
+        )
+        # These dialects are not in UPSERT_SUPPORTED_DIALECTS → plain INSERT
+        assert "ON CONFLICT" not in sql, f"{dialect} should NOT generate ON CONFLICT"
+        assert "INSERT INTO" in sql
+    # endregion test_unknown_dialects_route_to_insert
+
+    # region test_insert_strategy_bypasses_upsert [TYPE Function]
+    # @PURPOSE: INSERT strategy bypasses UPSERT for all dialects.
+    @pytest.mark.parametrize("dialect", ["postgresql", "clickhouse", "mysql", "redshift"])
+    def test_insert_strategy_bypasses_upsert(self, dialect: str) -> None:
+        rows = [{"id": 1, "name": "test"}]
+        sql, _ = SQLGenerator.generate(
+            dialect=dialect,
+            target_schema=None,
+            target_table="t",
+            columns=["id", "name"],
+            rows=rows,
+            key_columns=["id"],
+            upsert_strategy="INSERT",
+        )
+        assert "ON CONFLICT" not in sql
+        assert "INSERT INTO" in sql
+    # endregion test_insert_strategy_bypasses_upsert
+# #endregion TestDialectRouting_Orthogonal
+
+
+# #region TestSchemaValidationDialectRouting [TYPE Class]
+# @BRIEF Tests for dialect-based SQL query selection in validate_target_table_schema.
+# @TEST_INVARIANT is_clickhouse MUST be True only for clickhouse backends.
+class TestSchemaValidationDialectRouting:
+
+    # region test_clickhouse_uses_system_columns [TYPE Function]
+    # @PURPOSE: ClickHouse backend generates system.columns query.
+    @patch("src.plugins.translate.service_target_schema.SupersetSqlLabExecutor")
+    def test_clickhouse_uses_system_columns(self, mock_executor) -> None:
+        from src.schemas.translate import TargetSchemaValidationRequest
+        from src.plugins.translate.service_target_schema import validate_target_table_schema
+
+        req = TargetSchemaValidationRequest(
+            environment_id="env-1",
+            target_database_id="db-1",
+            target_schema="default",
+            target_table="my_table",
+            target_key_cols=["id"],
+            target_column="text",
+        )
+        config_manager = MagicMock()
+
+        instance = mock_executor.return_value
+        instance.resolve_database_id.return_value = None
+        instance.get_database_backend.return_value = "clickhouse"
+        instance.get_database_name.return_value = "test_ch"
+        instance.execute_and_poll.return_value = {
+            "status": "success",
+            "raw_response": {
+                "data": [
+                    {"name": "id", "type": "UInt64"},
+                    {"name": "text", "type": "String"},
+                    {"name": "context", "type": "String"},
+                    {"name": "is_original", "type": "UInt8"},
+                ],
+            },
+        }
+
+        validate_target_table_schema(req, config_manager)
+
+        # Verify the SQL query used system.columns
+        call_args = instance.execute_and_poll.call_args
+        sql = call_args.kwargs.get("sql") or call_args.args[0]
+        assert "system.columns" in sql
+        assert "information_schema" not in sql
+    # endregion test_clickhouse_uses_system_columns
+
+    # region test_postgresql_uses_information_schema [TYPE Function]
+    # @PURPOSE: PostgreSQL backend generates information_schema.columns query.
+    @patch("src.plugins.translate.service_target_schema.SupersetSqlLabExecutor")
+    def test_postgresql_uses_information_schema(self, mock_executor) -> None:
+        from src.schemas.translate import TargetSchemaValidationRequest
+        from src.plugins.translate.service_target_schema import validate_target_table_schema
+
+        req = TargetSchemaValidationRequest(
+            environment_id="env-1",
+            target_database_id="db-1",
+            target_schema="public",
+            target_table="my_table",
+            target_key_cols=["id"],
+            target_column="text",
+        )
+        config_manager = MagicMock()
+
+        instance = mock_executor.return_value
+        instance.resolve_database_id.return_value = None
+        instance.get_database_backend.return_value = "postgresql"
+        instance.get_database_name.return_value = "test_pg"
+        instance.execute_and_poll.return_value = {
+            "status": "success",
+            "raw_response": {
+                "data": [
+                    {"column_name": "id", "data_type": "integer"},
+                    {"column_name": "text", "data_type": "text"},
+                    {"column_name": "context", "data_type": "text"},
+                    {"column_name": "is_original", "data_type": "boolean"},
+                ],
+            },
+        }
+
+        validate_target_table_schema(req, config_manager)
+
+        call_args = instance.execute_and_poll.call_args
+        sql = call_args.kwargs.get("sql") or call_args.args[0]
+        assert "information_schema.columns" in sql
+        assert "system.columns" not in sql
+    # endregion test_postgresql_uses_information_schema
+
+    # region test_clickhousedb_normalized_to_clickhouse [TYPE Function]
+    # @PURPOSE: "clickhousedb" backend is normalized to clickhouse -> uses system.columns.
+    @patch("src.plugins.translate.service_target_schema.SupersetSqlLabExecutor")
+    def test_clickhousedb_normalized_to_clickhouse(self, mock_executor) -> None:
+        from src.schemas.translate import TargetSchemaValidationRequest
+        from src.plugins.translate.service_target_schema import validate_target_table_schema
+
+        req = TargetSchemaValidationRequest(
+            environment_id="env-1",
+            target_database_id="db-1",
+            target_schema="default",
+            target_table="my_table",
+            target_key_cols=["id"],
+            target_column="text",
+        )
+        config_manager = MagicMock()
+
+        instance = mock_executor.return_value
+        instance.resolve_database_id.return_value = None
+        instance.get_database_backend.return_value = "clickhousedb"
+        instance.get_database_name.return_value = "test_ch"
+        instance.execute_and_poll.return_value = {
+            "status": "success",
+            "raw_response": {
+                "data": [
+                    {"name": "id", "type": "UInt64"},
+                    {"name": "text", "type": "String"},
+                    {"name": "context", "type": "String"},
+                    {"name": "is_original", "type": "UInt8"},
+                ],
+            },
+        }
+
+        validate_target_table_schema(req, config_manager)
+
+        call_args = instance.execute_and_poll.call_args
+        sql = call_args.kwargs.get("sql") or call_args.args[0]
+        assert "system.columns" in sql
+    # endregion test_clickhousedb_normalized_to_clickhouse
+
+    # region test_greenplum_normalized_to_postgresql [TYPE Function]
+    # @PURPOSE: "greenplum" backend is normalized to postgresql -> uses information_schema.
+    @patch("src.plugins.translate.service_target_schema.SupersetSqlLabExecutor")
+    def test_greenplum_normalized_to_postgresql(self, mock_executor) -> None:
+        from src.schemas.translate import TargetSchemaValidationRequest
+        from src.plugins.translate.service_target_schema import validate_target_table_schema
+
+        req = TargetSchemaValidationRequest(
+            environment_id="env-1",
+            target_database_id="db-1",
+            target_schema="public",
+            target_table="my_table",
+            target_key_cols=["id"],
+            target_column="text",
+        )
+        config_manager = MagicMock()
+
+        instance = mock_executor.return_value
+        instance.resolve_database_id.return_value = None
+        instance.get_database_backend.return_value = "greenplum"
+        instance.get_database_name.return_value = "test_gp"
+        instance.execute_and_poll.return_value = {
+            "status": "success",
+            "raw_response": {
+                "data": [
+                    {"column_name": "id", "data_type": "integer"},
+                    {"column_name": "text", "data_type": "text"},
+                    {"column_name": "context", "data_type": "text"},
+                    {"name": "is_original", "type": "boolean"},
+                ],
+            },
+        }
+
+        validate_target_table_schema(req, config_manager)
+
+        call_args = instance.execute_and_poll.call_args
+        sql = call_args.kwargs.get("sql") or call_args.args[0]
+        assert "information_schema.columns" in sql
+    # endregion test_greenplum_normalized_to_postgresql
+
+    # region test_redshift_uses_information_schema [TYPE Function]
+    # @PURPOSE: Redshift uses information_schema (it's PostgreSQL-derived).
+    @patch("src.plugins.translate.service_target_schema.SupersetSqlLabExecutor")
+    def test_redshift_uses_information_schema(self, mock_executor) -> None:
+        from src.schemas.translate import TargetSchemaValidationRequest
+        from src.plugins.translate.service_target_schema import validate_target_table_schema
+
+        req = TargetSchemaValidationRequest(
+            environment_id="env-1",
+            target_database_id="db-1",
+            target_schema="public",
+            target_table="my_table",
+            target_key_cols=["id"],
+            target_column="text",
+        )
+        config_manager = MagicMock()
+
+        instance = mock_executor.return_value
+        instance.resolve_database_id.return_value = None
+        instance.get_database_backend.return_value = "redshift"
+        instance.get_database_name.return_value = "test_rs"
+        instance.execute_and_poll.return_value = {
+            "status": "success",
+            "raw_response": {
+                "data": [
+                    {"column_name": "id", "data_type": "integer"},
+                    {"column_name": "text", "data_type": "text"},
+                    {"column_name": "context", "data_type": "text"},
+                    {"column_name": "is_original", "data_type": "boolean"},
+                ],
+            },
+        }
+
+        validate_target_table_schema(req, config_manager)
+
+        call_args = instance.execute_and_poll.call_args
+        sql = call_args.kwargs.get("sql") or call_args.args[0]
+        assert "information_schema.columns" in sql
+    # endregion test_redshift_uses_information_schema
+
+    # region test_case_insensitive_backend_normalization [TYPE Function]
+    # @PURPOSE: Backend normalization is case-insensitive.
+    @patch("src.plugins.translate.service_target_schema.SupersetSqlLabExecutor")
+    def test_case_insensitive_backend_normalization(self, mock_executor) -> None:
+        from src.schemas.translate import TargetSchemaValidationRequest
+        from src.plugins.translate.service_target_schema import validate_target_table_schema
+
+        req = TargetSchemaValidationRequest(
+            environment_id="env-1",
+            target_database_id="db-1",
+            target_schema="default",
+            target_table="my_table",
+            target_key_cols=["id"],
+            target_column="text",
+        )
+        config_manager = MagicMock()
+
+        instance = mock_executor.return_value
+        instance.resolve_database_id.return_value = None
+        instance.get_database_backend.return_value = "ClickHouseDB"
+        instance.get_database_name.return_value = "test_ch"
+        instance.execute_and_poll.return_value = {
+            "status": "success",
+            "raw_response": {
+                "data": [
+                    {"name": "id", "type": "UInt64"},
+                    {"name": "text", "type": "String"},
+                    {"name": "context", "type": "String"},
+                    {"name": "is_original", "type": "UInt8"},
+                ],
+            },
+        }
+
+        validate_target_table_schema(req, config_manager)
+
+        call_args = instance.execute_and_poll.call_args
+        sql = call_args.kwargs.get("sql") or call_args.args[0]
+        assert "system.columns" in sql
+    # endregion test_case_insensitive_backend_normalization
+# #endregion TestSchemaValidationDialectRouting
+
+
+# #region TestDialectDetectionSummary [TYPE Class]
+# @BRIEF Summary test that documents all identified bugs and inconsistencies.
+# @TEST_EDGE This test serves as a living document of known issues.
+class TestDialectDetectionSummary:
+    """
+    DIALECT DETECTION — FIXED BUGS
+    ===============================
+
+    FIX-1: Whitespace stripping in _extract_dialect
+      - Added .strip() to guard AND to the normalized value
+      - " postgresql" → "postgresql" (consistent with get_dialect_from_database)
+      - "   " → "unknown" (whitespace-only treated as empty)
+
+    FIX-2: UPSERT routing now uses UPSERT_SUPPORTED_DIALECTS
+      - Only {"postgresql", "redshift"} get ON CONFLICT syntax
+      - MySQL, MSSQL, Snowflake, Oracle, DuckDB → plain INSERT
+      - Previously any non-ClickHouse dialect got ON CONFLICT (invalid SQL)
+
+    FIX-3: Dead code cleaned up
+      - "greenplum" removed from POSTGRESQL_DIALECTS (normalized to "postgresql")
+      - "clickhousedb" kept in CLICKHOUSE_DIALECTS for defensive quoting/encoding
+      - UPSERT_SUPPORTED_DIALECTS introduced for explicit routing
+
+    FIX-4: MySQL backtick quoting
+      - Added BACKTICK_DIALECTS = {"clickhouse", "clickhousedb", "mysql"}
+      - MySQL now uses backticks instead of double quotes
+      - Works with default MySQL sql_mode (no ANSI_QUOTES required)
+
+    REMAINING OBSERVATIONS
+    ----------------------
+    - Redshift is not in normalization maps but works correctly via
+      UPSERT_SUPPORTED_DIALECTS and information_schema fallthrough
+    - Boolean encoding as TRUE/FALSE may cause issues with ClickHouse strict typing
+    """
+
+    def test_documentation_only(self) -> None:
+        """This test exists to document known issues. No assertions."""
+        pass
+# #endregion TestDialectDetectionSummary
+# #endregion DialectDetectionOrthogonalTests

backend/src/plugins/translate/__tests__/test_executor.py

@@ -663,7 +663,7 @@ class TestAutoSizeBatches:
     # endregion test_resolve_provider_model_exception
 
     # region test_at_least_one_row_per_batch [TYPE Function]
-    @patch("src.plugins.translate.executor.estimate_token_budget")
+    @patch("src.plugins.translate._batch_sizer.estimate_token_budget")
     def test_at_least_one_row_per_batch(
         self,
         mock_estimate: MagicMock,

backend/src/plugins/translate/service_utils.py

@@ -6,7 +6,7 @@ from ...models.translate import TranslationJob
 from ...schemas.translate import TranslateJobResponse
 
 
-# #region _extract_dialect [TYPE Function]
+# #region _extract_dialect [C:2] [TYPE Function]
 # @BRIEF Extract database dialect from Superset backend URI or engine name.
 def _extract_dialect(backend: str) -> str:
     """Extract dialect name from a Superset database backend URI or engine name.
@@ -15,13 +15,13 @@ def _extract_dialect(backend: str) -> str:
     and plain engine names (e.g. 'postgresql', 'clickhousedb') returned by Superset.
     Normalises known variants like 'clickhousedb' → 'clickhouse'.
     """
-    if not backend:
+    if not backend or not backend.strip():
         return "unknown"
     try:
         # Extract scheme from URI or use plain name
         scheme = backend.split("://")[0]
         dialect = scheme.split("+")[0]
-        raw = dialect.lower()
+        raw = dialect.strip().lower()
         # Normalise known Superset backend variants
         dialect_map = {
             "clickhousedb": "clickhouse",
@@ -33,7 +33,7 @@ def _extract_dialect(backend: str) -> str:
 # #endregion _extract_dialect
 
 
-# #region job_to_response [TYPE Function]
+# #region job_to_response [C:1] [TYPE Function]
 # @BRIEF Convert a TranslationJob ORM model to a TranslateJobResponse schema with dictionary_ids.
 def job_to_response(job: TranslationJob, dict_ids: list[str] | None = None) -> TranslateJobResponse:
     return TranslateJobResponse(

backend/src/plugins/translate/sql_generator.py

@@ -15,10 +15,14 @@ from typing import Any
 
 from ...core.logger import belief_scope, logger
 
-# PostgreSQL/Greenplum dialects that support ON CONFLICT
-POSTGRESQL_DIALECTS = {"postgresql", "redshift", "greenplum"}
-# Dialects that use backtick or no quoting
+# PostgreSQL-family dialects that support ON CONFLICT (UPSERT)
+POSTGRESQL_DIALECTS = {"postgresql", "redshift"}
+# Dialects that support UPSERT via ON CONFLICT (subset of all supported dialects)
+UPSERT_SUPPORTED_DIALECTS = {"postgresql", "redshift"}
+# Dialects that use backtick quoting
 CLICKHOUSE_DIALECTS = {"clickhouse", "clickhousedb"}
+# Dialects that use backtick quoting (MySQL native, ClickHouse)
+BACKTICK_DIALECTS = {"clickhouse", "clickhousedb", "mysql"}
 
 
 # #region _normalize_timestamp_value [C:2] [TYPE Function] [SEMANTICS translate,sql,timestamp]
@@ -62,15 +66,25 @@ def _normalize_timestamp_value(value: Any) -> str | None:
 # @BRIEF Quote an identifier per dialect rules. PostgreSQL uses double quotes; ClickHouse uses backticks.
 # @PRE  identifier is a non-empty string.
 # @POST  Returns safely quoted identifier.
+# @RATIONALE Native quoting per dialect prevents syntax errors.
+#           PostgreSQL/Redshift require double quotes for case-sensitive/reserved identifiers.
+#           ClickHouse/MySQL use backticks natively.
+#           ANSI double-quote fallback works with any modern SQL engine.
+# @REJECTED  Single-quote quoting — would break all dialects for identifiers.
+#            No-op (pass-through) quoting — allows SQL injection via identifier names.
 def _quote_identifier(identifier: str, dialect: str) -> str:
-    """Quote a SQL identifier per dialect rules."""
+    """Quote a SQL identifier per dialect rules.
+
+    PostgreSQL/Redshift uses double quotes; ClickHouse/MySQL uses backticks.
+    Unknown dialects default to ANSI double quotes as a safe fallback.
+    """
     if not identifier:
         return identifier
     # Remove any existing quotes to avoid double-quoting
     cleaned = identifier.strip().strip('"').strip('`').strip('[]')
     if dialect in POSTGRESQL_DIALECTS:
         return f'"{cleaned}"'
-    elif dialect in CLICKHOUSE_DIALECTS:
+    elif dialect in BACKTICK_DIALECTS:
         return f"`{cleaned}`"
     else:
         # Generic ANSI double-quote
@@ -163,7 +177,11 @@ def generate_insert_sql(
 # @BRIEF Generate PostgreSQL dialect UPSERT SQL with ON CONFLICT DO UPDATE.
 # @PRE dialect is postgresql-compatible. target_table, columns, key_columns are non-empty.
 # @POST Returns UPSERT SQL string or raises ValueError.
-# @COMPLEXITY 4
+# @RATIONALE ON CONFLICT DO UPDATE is the standard PostgreSQL upsert syntax.
+#           The DO NOTHING variant covers the pure-insert-without-overwrite case
+#           when all columns are conflict keys. EXCLUDED references the proposed row.
+# @REJECTED MERGE INTO (SQL standard) — PostgreSQL only supports it via ON CONFLICT.
+#           Custom UPDATE+INSERT in a transaction — ON CONFLICT is atomic and faster.
 
 def generate_upsert_sql(
     target_schema: str | None,
@@ -283,8 +301,8 @@ class SQLGenerator:
             # Generate SQL per dialect and strategy
             use_upsert = upsert_strategy.upper() == "MERGE" and key_columns
 
-            if dialect in POSTGRESQL_DIALECTS or dialect not in CLICKHOUSE_DIALECTS:
-                # PostgreSQL and other ANSI dialects: support UPSERT via ON CONFLICT
+            if dialect in UPSERT_SUPPORTED_DIALECTS:
+                # PostgreSQL/Redshift: support UPSERT via ON CONFLICT
                 if use_upsert:
                     sql = generate_upsert_sql(
                         target_schema=None,
@@ -315,7 +333,8 @@ class SQLGenerator:
                         "note": "ClickHouse does not support ON CONFLICT. Use ReplacingMergeTree for dedup.",
                     })
             else:
-                # Fallback: plain INSERT
+                # Other dialects (MySQL, MSSQL, Snowflake, etc.): plain INSERT
+                # Do NOT generate ON CONFLICT — most dialects don't support it
                 sql = generate_insert_sql(
                     target_schema=None,
                     target_table=table_ref,
@@ -323,6 +342,11 @@ class SQLGenerator:
                     rows=rows,
                     dialect=dialect,
                 )
+                if use_upsert:
+                    logger.reason(
+                        f"UPSERT not supported for dialect '{dialect}'; using plain INSERT", {
+                        "note": "Use INSERT strategy explicitly to silence this warning.",
+                    })
 
             logger.reflect("SQL generated", {
                 "dialect": dialect,