feat(duckdb): use native MERGE for incremental by unique key models

docs/concepts/models/model_kinds.md

@@ -618,11 +618,14 @@ MODEL (
 
 * BigQuery
 * Databricks
+* DuckDB (version 1.4 or later, see below)
 * Postgres
 * Redshift
 * Snowflake
 * Spark
 
+In DuckDB's case, the native `MERGE` statement is used on version 1.4 or later when the target table is in a DuckDB, DuckLake or MotherDuck catalog. Older DuckDB versions, and target tables in other attached catalogs (such as PostgreSQL, SQLite or MySQL), use `DELETE` + `INSERT` instead, and `when_matched` and `merge_filter` raise an error there. Source rows are de-duplicated by `unique_key` before the merge so there is one row per key. Which of several duplicate rows is kept is unspecified.
+
 In Redshift's case, to enable the use of the native `MERGE` statement, you need to pass the `enable_merge` flag in the connection and set it to `true`. It is disabled by default.
 
 ```yaml linenums="1"
@@ -668,7 +671,7 @@ Depending on the target engine, models of the `INCREMENTAL_BY_UNIQUE_KEY` kind a
 | BigQuery   | MERGE ON unique key                 |
 | Redshift   | MERGE ON unique key                 |
 | Postgres   | MERGE ON unique key                 |
-| DuckDB     | DELETE ON matched + INSERT new rows |
+| DuckDB     | MERGE ON unique key (version 1.4 or later, DuckDB, DuckLake and MotherDuck catalogs); otherwise DELETE ON matched + INSERT new rows |
 
 ## FULL
 Models of the `FULL` kind cause the dataset associated with a model to be fully refreshed (rewritten) upon each model evaluation.

docs/integrations/engines/duckdb.md

@@ -196,6 +196,10 @@ Some connections, like SQLite, do not support schema names and therefore objects
 
 Example: mounting a SQLite database with the name `sqlite` that has a table `example_table` will be accessible as `sqlite.main.example_table`.
 
+##### Incremental by unique key models
+
+On DuckDB version 1.4 or later, [`INCREMENTAL_BY_UNIQUE_KEY`](../../concepts/models/model_kinds.md#incremental_by_unique_key) models targeting a DuckDB, DuckLake or MotherDuck catalog use a single, atomic `MERGE` statement and support `when_matched` and `merge_filter`. Older DuckDB versions and tables in other attached catalogs (such as PostgreSQL or SQLite) use `DELETE` + `INSERT`, which runs as two statements because DuckDB adapter transactions are disabled, and do not support `when_matched` or `merge_filter`.
+
 ##### Sensitive fields in paths
 
 If a connector, like Postgres, requires sensitive information in the path, it might support defining environment variables instead.

sqlmesh/core/engine_adapter/base.py

@@ -1771,6 +1771,10 @@ def update_table(
     ) -> None:
         self.execute(exp.update(table_name, properties, where=where))
 
+    def _prepare_merge_source_query(self, query: Query, unique_key: t.Sequence[exp.Expr]) -> Query:
+        """Hook that lets an engine adjust the MERGE source query. By default it is used as is."""
+        return query
+
     def _merge(
         self,
         target_table: TableName,
@@ -2316,7 +2320,7 @@ def merge(
             with source_query as query:
                 self._merge(
                     target_table=target_table,
-                    query=query,
+                    query=self._prepare_merge_source_query(query, unique_key),
                     on=on,
                     whens=exp.Whens(expressions=match_expressions),
                 )

sqlmesh/core/engine_adapter/duckdb.py

@@ -1,13 +1,16 @@
 from __future__ import annotations
 
 import typing as t
+from functools import cached_property, partial
 from sqlglot import exp
 from pathlib import Path
+from packaging import version
 
+from sqlmesh.core.engine_adapter.base import MERGE_TARGET_ALIAS
 from sqlmesh.core.engine_adapter.mixins import (
     GetCurrentCatalogFromFunctionMixin,
-    LogicalMergeMixin,
     RowDiffMixin,
+    logical_merge,
 )
 from sqlmesh.core.engine_adapter.shared import (
     CatalogSupport,
@@ -21,11 +24,11 @@
 
 if t.TYPE_CHECKING:
     from sqlmesh.core._typing import SchemaName, TableName
-    from sqlmesh.core.engine_adapter._typing import DF
+    from sqlmesh.core.engine_adapter._typing import DF, Query, QueryOrDF
 
 
 @set_catalog(override_mapping={"_get_data_objects": CatalogSupport.REQUIRES_SET_CATALOG})
-class DuckDBEngineAdapter(LogicalMergeMixin, GetCurrentCatalogFromFunctionMixin, RowDiffMixin):
+class DuckDBEngineAdapter(GetCurrentCatalogFromFunctionMixin, RowDiffMixin):
     DIALECT = "duckdb"
     SUPPORTS_TRANSACTIONS = False
     SCHEMA_DIFFER_KWARGS = {
@@ -37,6 +40,11 @@ class DuckDBEngineAdapter(LogicalMergeMixin, GetCurrentCatalogFromFunctionMixin,
     COMMENT_CREATION_VIEW = CommentCreationView.COMMENT_COMMAND_ONLY
     SUPPORTS_CREATE_DROP_CATALOG = True
     SUPPORTED_DROP_CASCADE_OBJECT_KINDS = ["SCHEMA", "TABLE", "VIEW"]
+    # MERGE INTO was added in DuckDB 1.4.0
+    MIN_NATIVE_MERGE_VERSION = version.parse("1.4.0")
+    # Catalog types whose storage extension supports MERGE INTO. Other attached catalogs
+    # (postgres, sqlite, mysql, ...) fall back to DELETE + INSERT.
+    NATIVE_MERGE_CATALOG_TYPES = {"duckdb", "ducklake"}
 
     @property
     def catalog_support(self) -> CatalogSupport:
@@ -72,6 +80,76 @@ def _drop_catalog(self, catalog_name: exp.Identifier) -> None:
                 )
             )
 
+    @cached_property
+    def _duckdb_supports_merge(self) -> bool:
+        import duckdb
+
+        return version.parse(duckdb.__version__) >= self.MIN_NATIVE_MERGE_VERSION
+
+    def _supports_native_merge(self, target_table: TableName) -> bool:
+        if not self._duckdb_supports_merge:
+            return False
+        if self._is_motherduck:
+            return True
+        catalog = exp.to_table(target_table).catalog or self.get_current_catalog()
+        # Errors from the catalog lookup propagate on purpose; only a None/unknown type falls back.
+        return self._get_catalog_type(catalog) in self.NATIVE_MERGE_CATALOG_TYPES
+
+    def merge(
+        self,
+        target_table: TableName,
+        source_table: QueryOrDF,
+        target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]],
+        unique_key: t.Sequence[exp.Expr],
+        when_matched: t.Optional[exp.Whens] = None,
+        merge_filter: t.Optional[exp.Expr] = None,
+        source_columns: t.Optional[t.List[str]] = None,
+        **kwargs: t.Any,
+    ) -> None:
+        merge_impl = (
+            super().merge
+            if self._supports_native_merge(target_table)
+            else partial(logical_merge, self)
+        )
+        merge_impl(  # type: ignore
+            target_table,
+            source_table,
+            target_columns_to_types,
+            unique_key,
+            when_matched=when_matched,
+            merge_filter=merge_filter,
+            source_columns=source_columns,
+        )
+
+    def _prepare_merge_source_query(self, query: Query, unique_key: t.Sequence[exp.Expr]) -> Query:
+        # A native MERGE inserts every unmatched source row, so duplicate keys in the source would
+        # produce duplicate keys in the target. Keep one row per key, like the DELETE + INSERT path.
+        return exp.select("*").distinct(*unique_key).from_(query.subquery("_src"))  # type: ignore
+
+    def _merge(
+        self,
+        target_table: TableName,
+        query: Query,
+        on: exp.Expr,
+        whens: exp.Whens,
+    ) -> None:
+        # DuckDB rejects a qualified column on the left-hand side of UPDATE SET and in the INSERT
+        # column list. Conditions and right-hand sides may keep the qualifier.
+        # TODO: sqlglot's DuckDB generator doesn't apply `merge_without_target_sql` yet. Remove this
+        # once the pinned sqlglot version does.
+        whens = whens.copy()
+        for when in whens.expressions:
+            then = when.args.get("then")
+            targets: t.List[exp.Expr] = []
+            if isinstance(then, exp.Update):
+                targets = [e.this for e in then.expressions if isinstance(e, exp.EQ)]
+            elif isinstance(then, exp.Insert) and isinstance(then.this, exp.Tuple):
+                targets = then.this.expressions
+            for column in targets:
+                if isinstance(column, exp.Column) and column.table.upper() == MERGE_TARGET_ALIAS:
+                    column.set("table", None)
+        super()._merge(target_table, query, on, whens)
+
     def _df_to_source_queries(
         self,
         df: DF,

tests/core/engine_adapter/integration/test_integration.py

@@ -31,6 +31,7 @@
 from sqlmesh.core.environment import EnvironmentSuffixTarget
 from sqlmesh.core.dialect import select_from_values
 from sqlmesh.core.model import Model, load_sql_based_model
+from sqlmesh.core.engine_adapter.duckdb import DuckDBEngineAdapter
 from sqlmesh.core.engine_adapter.shared import DataObject, DataObjectType
 from sqlmesh.core.engine_adapter.mixins import RowDiffMixin, LogicalMergeMixin
 from sqlmesh.core.model.definition import create_sql_model
@@ -2857,11 +2858,16 @@ def test_incremental_by_unique_key_model_when_matched(ctx: TestContext):
     if not ctx.supports_merge:
         pytest.skip(f"{ctx.dialect} on {ctx.gateway} doesnt support merge")
 
-    # DuckDB and some other engines use logical_merge which doesn't support when_matched
+    # Engines that use logical_merge (e.g. MySQL) don't support when_matched
     if isinstance(ctx.engine_adapter, LogicalMergeMixin):
         pytest.skip(
             f"{ctx.dialect} on {ctx.gateway} uses logical merge which doesn't support when_matched"
         )
+    # DuckDB only merges natively from 1.4.0; older versions fall back to logical_merge
+    if isinstance(ctx.engine_adapter, DuckDBEngineAdapter) and not (
+        ctx.engine_adapter._duckdb_supports_merge
+    ):
+        pytest.skip("The installed duckdb doesn't support MERGE (requires >= 1.4.0)")
 
     def _mutate_config(current_gateway_name: str, config: Config):
         connection = config.gateways[current_gateway_name].connection