fix(builder): validate dialect clauses and translate MySQL upserts

docs/changelog.rst

@@ -153,8 +153,20 @@ v0.63.0 - Transactions, table fixtures, SQL fragments, storage, and kwargs param
 
 **Changed:**
 
+* Query builder validates dialect capabilities in ``build()`` and ``to_statement()`` instead of silently dropping unsupported clauses.
+  ``.for_update()`` and ``.for_share()`` raise ``SQLBuilderError`` on dialects without these locking clauses
+  (T-SQL, SQLite, DuckDB, BigQuery), and ``skip_locked=True`` validates against ``supports_skip_locked``.
+  ``.on_conflict()`` automatically transpiles to ``ON DUPLICATE KEY UPDATE`` for MySQL and MariaDB (with ``do_nothing()``
+  rewriting to self-assignment), while raising ``SQLBuilderError`` suggesting ``sql.merge()`` on dialects lacking native
+  upsert support (Oracle, T-SQL, BigQuery). Spanner supports native upserts and plain ``FOR UPDATE``;
+  PostgreSQL-mode upserts validate assignment restrictions. Oracle rejects shared locks, while MariaDB renders
+  ``LOCK IN SHARE MODE``. Query builder ``build()`` also normalizes dialect aliases
+  (``mssql`` to ``tsql``, ``mariadb`` to ``mysql``, and ``cockroachdb`` to ``postgres``).
+  (`#778 <https://github.com/litestar-org/sqlspec/pull/778>`_)
+
 * Decoupled the ``ParadeDB`` dialect so it inherits directly from ``Postgres`` rather than ``PGVector``,
   allowing clean independent combinations of vector search and BM25 extensions.
+
 * Standardized PostgreSQL extension detection across all adapters on a single first-connection probe
   via :func:`~sqlspec.core.config_runtime.build_postgres_extension_probe_names`, removing ad-hoc ADK
   probe branches.

docs/usage/query_builder.rst

@@ -44,6 +44,16 @@ Upserts (ON CONFLICT)
 Use ``.on_conflict()`` to handle insert conflicts. Chain ``.do_nothing()`` to skip
 conflicting rows, or ``.do_update(**columns)`` to update them.
 
+Dialects natively supporting ``ON CONFLICT`` (PostgreSQL, CockroachDB, SQLite, DuckDB, and Spanner)
+render standard ``ON CONFLICT`` syntax. For MySQL and MariaDB, the builder automatically
+transpiles ``.on_conflict().do_update()`` to ``ON DUPLICATE KEY UPDATE``, and ``.do_nothing()``
+to a no-op self-assignment (e.g., ``col = col``). This requires a conflict column or
+explicit insert columns. MySQL handles conflicts on any unique key, regardless of the
+requested conflict target; the no-op update can still fire update triggers. References
+to ``excluded.column`` in update expressions become ``VALUES(column)``. Dialects without native upsert clauses
+(Oracle, T-SQL / SQL Server, and BigQuery) raise :class:`~sqlspec.exceptions.SQLBuilderError`
+in both ``build()`` and ``to_statement()`` advising the use of :func:`sql.merge`.
+
 .. literalinclude:: /examples/builder/upsert.py
    :language: python
    :caption: ``upsert with on_conflict``
@@ -81,6 +91,20 @@ Joins
 Query Modifiers
 ---------------
 
+Row-level locking clauses such as ``.for_update()`` and ``.for_share()`` are validated against
+dialect capabilities at build time. On dialects without these locking clauses (T-SQL, SQLite,
+DuckDB, and BigQuery), building a locked query raises :class:`~sqlspec.exceptions.SQLBuilderError`.
+Oracle also rejects ``.for_share()``; MariaDB renders it as ``LOCK IN SHARE MODE``
+and rejects ``of=`` targets for all locking clauses. PostgreSQL key lock variants
+are rejected on other dialect families.
+Spanner supports plain ``FOR UPDATE`` in both SQL modes, but rejects shared locks,
+``SKIP LOCKED``, ``NOWAIT``, and ``OF`` modifiers. Its PostgreSQL mode requires conflict
+updates to assign every inserted column from the matching ``excluded`` column and
+does not accept conflict predicates or named constraints.
+Similarly, ``skip_locked=True`` is validated against the dialect's ``supports_skip_locked`` capability.
+The builder also normalizes common dialect aliases during build (e.g., ``mssql`` to ``tsql``,
+``mariadb`` to ``mysql``, and ``cockroachdb`` to ``postgres``).
+
 .. literalinclude:: /examples/builder/query_modifiers.py
    :language: python
    :caption: ``where helpers + pagination``

sqlspec/builder/_base.py

@@ -21,7 +21,7 @@
 from typing_extensions import Self
 
 from sqlspec.builder._locking import register_lock_generator
-from sqlspec.builder._parsing_utils import _resolve_dialect
+from sqlspec.builder._parsing_utils import _normalize_dialect, _resolve_dialect
 from sqlspec.builder._vector_distance import has_vector_distance_ancestor
 from sqlspec.core import (
     SQL,
@@ -35,6 +35,7 @@
 )
 from sqlspec.core.filters import StatementFilter
 from sqlspec.core.hashing import _expression_cache_fingerprint
+from sqlspec.data_dictionary import get_dialect_config
 from sqlspec.exceptions import SQLBuilderError
 from sqlspec.utils.logging import get_logger
 from sqlspec.utils.type_guards import has_expression_and_parameters, has_name, has_with_method, is_expression
@@ -611,7 +612,8 @@ def build(self, dialect: DialectType = None) -> "BuiltQuery":
         if self.enable_optimization and isinstance(final_expression, exp.Expr):
             final_expression = self._optimize_expression(final_expression)
 
-        target_dialect = str(dialect) if dialect else self.dialect_name
+        target_dialect = self._build_dialect(dialect)
+        final_expression = self._prepare_dialect_expression(final_expression, target_dialect, dialect)
 
         try:
             if isinstance(final_expression, exp.Expr):
@@ -631,9 +633,98 @@ def build(self, dialect: DialectType = None) -> "BuiltQuery":
             err_msg = f"Error generating SQL from expression: {e!s}"
             self._raise_builder_error(err_msg, e)
 
-        return BuiltQuery(
-            sql=sql_string, parameters=self._parameters.copy(), dialect=_resolve_dialect(dialect, self.dialect)
-        )
+        return BuiltQuery(sql=sql_string, parameters=self._parameters.copy(), dialect=target_dialect)
+
+    def _build_dialect(self, dialect: DialectType = None) -> str | None:
+        return _normalize_dialect(dialect or self.dialect)
+
+    def _prepare_dialect_expression(
+        self, expression: exp.Expr, dialect: str | None, source_dialect: DialectType = None
+    ) -> exp.Expr:
+        """Validate and translate a copy so builds never mutate the builder AST."""
+        if dialect is None:
+            return expression
+        try:
+            config = get_dialect_config("spanner" if dialect == "spangres" else dialect)
+        except ValueError:
+            return expression
+        if str(source_dialect or self.dialect).lower() == "mariadb" and expression.find(exp.Lock):
+            expression = expression.copy()
+            for lock in expression.find_all(exp.Lock):
+                if lock.expressions:
+                    self._raise_builder_error("MariaDB locking clauses do not support OF targets.")
+                if not lock.args.get("update"):
+                    lock.set("sqlspec_share_mode", True)
+        for lock in expression.find_all(exp.Lock):
+            if dialect in {"spanner", "spangres"} and (
+                not lock.args.get("update")
+                or lock.args.get("wait") is not None
+                or lock.expressions
+                or lock.args.get("key")
+            ):
+                self._raise_builder_error(f"Dialect '{dialect}' supports only plain FOR UPDATE without lock modifiers.")
+            if lock.args.get("key") and dialect != "postgres":
+                self._raise_builder_error(f"Dialect '{dialect}' does not support PostgreSQL key lock modes.")
+            if dialect == "oracle" and not lock.args.get("update"):
+                self._raise_builder_error("Dialect 'oracle' does not support FOR SHARE.")
+            if dialect not in {"spanner", "spangres"} and config.get_feature_flag("supports_for_update") is False:
+                self._raise_builder_error(f"Dialect '{dialect}' does not support FOR UPDATE / row locking.")
+            if lock.args.get("wait") is False and config.get_feature_flag("supports_skip_locked") is False:
+                self._raise_builder_error(f"Dialect '{dialect}' does not support SKIP LOCKED.")
+        if dialect == "spangres":
+            self._validate_spangres_conflicts(expression)
+        if config.get_feature_flag("supports_on_conflict") is not False or not expression.find(exp.OnConflict):
+            return expression
+        if dialect != "mysql":
+            self._raise_builder_error(f"Dialect '{dialect}' does not support ON CONFLICT; use sql.merge() instead.")
+        expression = expression.copy()
+        for conflict in expression.find_all(exp.OnConflict):
+            if conflict.args.get("duplicate"):
+                continue
+            if any(conflict.args.get(key) for key in ("where", "index_predicate", "constraint")):
+                self._raise_builder_error("MySQL cannot preserve ON CONFLICT predicates or named constraints.")
+            assignments = conflict.args.get("expressions")
+            if str(conflict.args.get("action", "")).upper() == "DO NOTHING":
+                keys = conflict.args.get("conflict_keys")
+                insert = conflict.find_ancestor(exp.Insert)
+                schema = insert.this if insert is not None else None
+                columns = keys or (schema.expressions if isinstance(schema, exp.Schema) else None)
+                if not columns:
+                    self._raise_builder_error("MySQL DO NOTHING requires a conflict column or explicit insert columns.")
+                column = exp.column(columns[0].name)
+                assignments = [exp.EQ(this=column, expression=column.copy())]
+            elif not assignments:
+                self._raise_builder_error("ON CONFLICT DO UPDATE requires at least one assignment.")
+            for assignment in assignments:
+                for column in list(assignment.find_all(exp.Column)):
+                    if column.table.lower() == "excluded":
+                        column.replace(exp.Anonymous(this="VALUES", expressions=[exp.column(column.name)]))
+            conflict.replace(exp.OnConflict(duplicate=True, action=exp.var("UPDATE"), expressions=assignments))
+        return expression
+
+    def _validate_spangres_conflicts(self, expression: exp.Expr) -> None:
+        """Check PostgreSQL-mode Spanner upsert restrictions visible in the AST."""
+        for conflict in expression.find_all(exp.OnConflict):
+            if any(conflict.args.get(key) for key in ("where", "index_predicate", "constraint", "duplicate")):
+                self._raise_builder_error("Spanner PostgreSQL does not support these ON CONFLICT modifiers.")
+            assignments = conflict.args.get("expressions") or []
+            if assignments:
+                insert = conflict.find_ancestor(exp.Insert)
+                schema = insert.this if insert is not None else None
+                assigned_columns = set()
+                for assignment in assignments:
+                    value = assignment.expression
+                    if (
+                        not isinstance(value, exp.Column)
+                        or value.table.lower() != "excluded"
+                        or value.name != assignment.this.name
+                    ):
+                        self._raise_builder_error("Spanner PostgreSQL conflict updates require excluded column values.")
+                    assigned_columns.add(assignment.this.name)
+                if isinstance(schema, exp.Schema) and assigned_columns != {
+                    column.name for column in schema.expressions
+                }:
+                    self._raise_builder_error("Spanner PostgreSQL conflict updates must assign every inserted column.")
 
     def to_sql(self, show_parameters: bool = False, dialect: DialectType = None) -> str:
         """Return SQL string with optional parameter substitution.
@@ -825,12 +916,16 @@ def _to_statement(self, config: "StatementConfig | None" = None) -> "SQL":
 
     def _create_builder_cache_entry(self, config: "StatementConfig | None") -> "_BuilderCacheEntry":
         dialect_override = config.dialect if config is not None else None
-        resolved_dialect = _resolve_dialect(dialect_override, self.dialect)
+        resolved_dialect = self._build_dialect(dialect_override)
         statement_expression = self._build_final_expression(copy=True)
 
         if self.enable_optimization and isinstance(statement_expression, exp.Expr):
             statement_expression = self._optimize_expression(statement_expression)
 
+        statement_expression = self._prepare_dialect_expression(
+            statement_expression, resolved_dialect, dialect_override
+        )
+
         if statement_expression.find(exp.Lock):
             register_lock_generator(resolved_dialect)
         if self._is_oracle_dialect(resolved_dialect):
@@ -842,6 +937,8 @@ def _statement_from_cache_entry(self, cache_entry: "_BuilderCacheEntry", config:
         kwargs, parameters = self._statement_parameters(self._parameters.copy())
 
         statement_config = config
+        if statement_config is not None and statement_config.dialect != cache_entry.dialect:
+            statement_config = statement_config.replace(dialect=cache_entry.dialect)
         if statement_config is None:
             statement_config = StatementConfig(
                 parameter_config=ParameterStyleConfig(

sqlspec/builder/_locking.py

@@ -38,14 +38,17 @@ def _render_lock_targets(generator: "Generator", expressions: "Iterable[exp.Expr
 
 
 def _lock_sql(generator: "Generator", expression: exp.Lock) -> str:
-    if not generator.LOCKING_READS_SUPPORTED:
+    if not generator.LOCKING_READS_SUPPORTED and type(generator.dialect).__name__ != "Spanner":
         generator.unsupported("Locking reads using 'FOR UPDATE/SHARE' are not supported")
         return ""
 
     update = expression.args["update"]
     key = expression.args.get("key")
     lock_type = ("FOR NO KEY UPDATE" if key else "FOR UPDATE") if update else "FOR KEY SHARE" if key else "FOR SHARE"
 
+    if expression.args.get("sqlspec_share_mode"):
+        lock_type = "LOCK IN SHARE MODE"
+
     targets = _render_lock_targets(generator, expression.expressions)
     target_sql = f" OF {targets}" if targets else ""
     wait = expression.args.get("wait")

sqlspec/builder/_select.py

@@ -1511,7 +1511,7 @@ def for_no_key_update(self) -> "Self":
         assert self._expression is not None
         select_expr = cast("exp.Select", self._expression)
 
-        lock = exp.Lock(update=True, key=False)
+        lock = exp.Lock(update=True, key=True)
 
         current_locks = select_expr.args.get("locks", [])
         current_locks.append(lock)

sqlspec/core/hashing.py

@@ -257,7 +257,7 @@ def _expression_cache_fingerprint(
     settings: Any = None,
 ) -> str:
     components = (
-        hash(expr),
+        hash_expression(expr),
         parameter_signature,
         str(dialect) if dialect is not None else "default",
         _freeze_cache_value(schema),

sqlspec/data_dictionary/_types.py

@@ -1422,6 +1422,7 @@ class FeatureFlags(TypedDict, total=False):
     supports_interleaved_tables: bool
     supports_json: bool
     supports_maps: bool
+    supports_on_conflict: bool
     supports_partitioning: bool
     supports_prepared_statements: bool
     supports_resource_groups: bool

sqlspec/data_dictionary/dialects/bigquery/config.py

@@ -26,6 +26,7 @@
     "supports_uuid": False,
     "supports_for_update": False,
     "supports_skip_locked": False,
+    "supports_on_conflict": False,
 }
 
 BIGQUERY_TYPE_MAPPINGS: dict[str, str] = {

sqlspec/data_dictionary/dialects/cockroachdb/config.py

@@ -24,6 +24,7 @@
     "supports_for_update": True,
     "supports_skip_locked": True,
     "supports_crdb_internal_metadata": False,
+    "supports_on_conflict": True,
 }
 
 COCKROACHDB_TYPE_MAPPINGS: dict[str, str] = {

sqlspec/data_dictionary/dialects/duckdb/config.py

@@ -22,6 +22,7 @@
     "supports_uuid": True,
     "supports_for_update": False,
     "supports_skip_locked": False,
+    "supports_on_conflict": True,
 }
 
 DUCKDB_TYPE_MAPPINGS: dict[str, str] = {

sqlspec/data_dictionary/dialects/mssql/config.py

@@ -123,6 +123,7 @@
     "supports_in_memory": True,
     "supports_for_update": False,
     "supports_skip_locked": False,
+    "supports_on_conflict": False,
 }
 
 MSSQL_TYPE_MAPPINGS: dict[str, str] = {

sqlspec/data_dictionary/dialects/mysql/config.py

@@ -61,6 +61,7 @@
     "supports_for_update": True,
     "supports_sequences": False,
     "supports_system_versioned_tables": False,
+    "supports_on_conflict": False,
 }
 
 MYSQL_TYPE_MAPPINGS: dict[str, str] = {
@@ -113,6 +114,7 @@
     "supports_invisible_columns": False,
     "supports_invisible_indexes": False,
     "supports_resource_groups": False,
+    "supports_on_conflict": False,
 }
 
 MARIADB_CONFIG = DialectConfig(

sqlspec/data_dictionary/dialects/oracle/config.py

@@ -53,6 +53,7 @@
     "supports_in_memory": True,
     "supports_for_update": True,
     "supports_skip_locked": True,
+    "supports_on_conflict": False,
 }
 
 ORACLE_TYPE_MAPPINGS: dict[str, str] = {

sqlspec/data_dictionary/dialects/postgres/config.py

@@ -25,6 +25,7 @@
     "supports_prepared_statements": True,
     "supports_schemas": True,
     "supports_for_update": True,
+    "supports_on_conflict": True,
 }
 
 POSTGRES_TYPE_MAPPINGS: dict[str, str] = {

sqlspec/data_dictionary/dialects/spanner/config.py

@@ -13,6 +13,7 @@
     "supports_interleaved_tables": True,
     "supports_for_update": False,
     "supports_skip_locked": False,
+    "supports_on_conflict": True,
 }
 
 SPANNER_TYPE_MAPPINGS: dict[str, str] = {

sqlspec/data_dictionary/dialects/sqlite/config.py

@@ -23,6 +23,7 @@
     "supports_uuid": False,
     "supports_for_update": False,
     "supports_skip_locked": False,
+    "supports_on_conflict": True,
 }
 
 SQLITE_TYPE_MAPPINGS: dict[str, str] = {