Add data path and encrypted to duckdb catalog (#4600)

Author: mattiasthalen

docs/integrations/engines/duckdb.md

@@ -64,6 +64,53 @@ SQLMesh will place models with the explicit catalog "ephemeral", such as `epheme
     )
     ```
 
+#### DuckLake Catalog Example
+
+=== "YAML"
+
+    ```yaml linenums="1"
+    gateways:
+      my_gateway:
+        connection:
+          type: duckdb
+          catalogs:
+            ducklake:
+              type: ducklake
+              path: 'catalog.ducklake'
+              data_path: data/ducklake
+              encrypted: True
+    ```
+    
+=== "Python"
+
+    ```python linenums="1"
+    from sqlmesh.core.config import (
+        Config,
+        ModelDefaultsConfig,
+        GatewayConfig,
+        DuckDBConnectionConfig
+    )
+    from sqlmesh.core.config.connection import DuckDBAttachOptions
+
+    config = Config(
+        model_defaults=ModelDefaultsConfig(dialect=<dialect>),
+        gateways={
+            "my_gateway": GatewayConfig(
+                connection=DuckDBConnectionConfig(
+                    catalogs={
+                        "ducklake": DuckDBAttachOptions(
+                            type="ducklake",
+                            path="catalog.ducklake",
+                            data_path="data/ducklake",
+                            encrypted=True
+                        ),
+                    }
+                )
+            ),
+        }
+    )
+    ```
+
 #### Other Connection Catalogs Example
 
 Catalogs can also be defined to connect to anything that [DuckDB can be attached to](https://duckdb.org/docs/sql/statements/attach.html).

sqlmesh/core/config/connection.py

@@ -221,6 +221,10 @@ class DuckDBAttachOptions(BaseConfig):
     path: str
     read_only: bool = False
 
+    # DuckLake specific options
+    data_path: t.Optional[str] = None
+    encrypted: bool = False
+
     def to_sql(self, alias: str) -> str:
         options = []
         # 'duckdb' is actually not a supported type, but we'd like to allow it for
@@ -229,6 +233,14 @@ def to_sql(self, alias: str) -> str:
             options.append(f"TYPE {self.type.upper()}")
         if self.read_only:
             options.append("READ_ONLY")
+
+        # DuckLake specific options
+        if self.type == "ducklake":
+            if self.data_path:
+                options.append(f"DATA_PATH '{self.data_path}'")
+            if self.encrypted:
+                options.append("ENCRYPTED")
+
         options_sql = f" ({', '.join(options)})" if options else ""
         alias_sql = ""
         # TODO: Add support for Postgres schema. Currently adding it blocks access to the information_schema

tests/core/test_connection_config.py

@@ -602,6 +602,30 @@ def test_duckdb_attach_catalog(make_config):
     assert not config.is_recommended_for_state_sync
 
 
+def test_duckdb_attach_ducklake_catalog(make_config):
+    config = make_config(
+        type="duckdb",
+        catalogs={
+            "ducklake": DuckDBAttachOptions(
+                type="ducklake",
+                path="catalog.ducklake",
+                data_path="/tmp/ducklake_data",
+                encrypted=True,
+            ),
+        },
+    )
+    assert isinstance(config, DuckDBConnectionConfig)
+    ducklake_catalog = config.catalogs.get("ducklake")
+    assert ducklake_catalog is not None
+    assert ducklake_catalog.type == "ducklake"
+    assert ducklake_catalog.path == "catalog.ducklake"
+    assert ducklake_catalog.data_path == "/tmp/ducklake_data"
+    assert ducklake_catalog.encrypted is True
+    # Check that the generated SQL includes DATA_PATH
+    assert "DATA_PATH '/tmp/ducklake_data'" in ducklake_catalog.to_sql("ducklake")
+    assert "ENCRYPTED" in ducklake_catalog.to_sql("ducklake")
+
+
 def test_duckdb_attach_options():
     options = DuckDBAttachOptions(
         type="postgres", path="dbname=postgres user=postgres host=127.0.0.1", read_only=True