Repository: TobikoData/sqlmesh
Pull request #4913 (2025-07-04 17:19:00)

fix: don't include type when using ducklake
Prior to this PR, would get the following error (confirmed in two different environments):
```
duckdb.duckdb.NotImplementedException: Not implemented Error: Failed to initialize DuckLake:PRIMARY KEY/UNIQUE constraints are not supported in DuckLake
```

Confirmed that if I remove this type then the error goes away and Ducklake works as expected. I don't see anywhere in Ducklake docs that it says to include TYPE when using ducklake. Tested using Ducklake with DuckDB and Postgres.


Patch:
diff --git a/docs/integrations/engines/duckdb.md b/docs/integrations/engines/duckdb.md
index 19d8d6e1..bc0af4f2 100644
--- a/docs/integrations/engines/duckdb.md
+++ b/docs/integrations/engines/duckdb.md
@@ -10,15 +10,15 @@
 
 ### Connection options
 
-| Option             | Description                                                                                                                                                                                                                                     |  Type  | Required |
-|--------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:------:|:--------:|
-| `type`             | Engine type name - must be `duckdb`                                                                                                                                                                                                             | string |    Y     |
-| `database`         | The optional database name. If not specified, the in-memory database is used. Cannot be defined if using `catalogs`.                                                                                                                            | string |    N     |
-| `catalogs`         | Mapping to define multiple catalogs. Can [attach DuckDB catalogs](#duckdb-catalogs-example) or [catalogs for other connections](#other-connection-catalogs-example). First entry is the default catalog. Cannot be defined if using `database`. |  dict  |    N     |
-| `extensions`       | Extension to load into duckdb. Only autoloadable extensions are supported.                                                                                                                                                                      |  list  |    N     |
-| `connector_config` | Configuration to pass into the duckdb connector.                                                                                                                                                                                                |  dict  |    N     |
-| `secrets`   | Configuration for authenticating external sources (e.g., S3) using DuckDB secrets.                                                                                                                                                                                                |  dict  |    N     |
-| `filesystems`   | Configuration for registering `fsspec` filesystems to the DuckDB connection.                                                                                                                                                                                                |  dict  |    N     |
+| Option             | Description                                                                                                                                                                                                                                     |   Type    | Required |
+|--------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:---------:|:--------:|
+| `type`             | Engine type name - must be `duckdb`                                                                                                                                                                                                             |  string   |    Y     |
+| `database`         | The optional database name. If not specified, the in-memory database is used. Cannot be defined if using `catalogs`.                                                                                                                            |  string   |    N     |
+| `catalogs`         | Mapping to define multiple catalogs. Can [attach DuckDB catalogs](#duckdb-catalogs-example) or [catalogs for other connections](#other-connection-catalogs-example). First entry is the default catalog. Cannot be defined if using `database`. |   dict    |    N     |
+| `extensions`       | Extension to load into duckdb. Only autoloadable extensions are supported.                                                                                                                                                                      |   list    |    N     |
+| `connector_config` | Configuration to pass into the duckdb connector.                                                                                                                                                                                                |   dict    |    N     |
+| `secrets`          | Configuration for authenticating external sources (e.g., S3) using DuckDB secrets. Can be a list of secret configurations or a dictionary with custom secret names.                                                                             | list/dict |    N     |
+| `filesystems`      | Configuration for registering `fsspec` filesystems to the DuckDB connection.                                                                                                                                                                    |   dict    |    N     |
 
 #### DuckDB Catalogs Example
 
@@ -194,9 +194,18 @@ DuckDB can read data directly from cloud services via extensions (e.g., [httpfs]
 
 The `secrets` option allows you to configure DuckDB's [Secrets Manager](https://duckdb.org/docs/configuration/secrets_manager.html) to authenticate with external services like S3. This is the recommended approach for cloud storage authentication in DuckDB v0.10.0 and newer, replacing the [legacy authentication method](https://duckdb.org/docs/stable/extensions/httpfs/s3api_legacy_authentication.html) via variables.
 
-##### Secrets Configuration Example for S3
+##### Secrets Configuration
 
-The `secrets` accepts a list of secret configurations, each defining the necessary authentication parameters for the specific service:
+The `secrets` option supports two formats:
+
+1. **List format** (default secrets): A list of secret configurations where each secret uses DuckDB's default naming
+2. **Dictionary format** (named secrets): A dictionary where keys are custom secret names and values are the secret configurations
+
+This flexibility allows you to organize multiple secrets of the same type or reference specific secrets by name in your SQL queries.
+
+##### List Format Example (Default Secrets)
+
+Using a list creates secrets with DuckDB's default naming:
 
 === "YAML"
 
@@ -253,6 +262,75 @@ The `secrets` accepts a list of secret configurations, each defining the necessa
     )
     ```
 
+##### Dictionary Format Example (Named Secrets)
+
+Using a dictionary allows you to assign custom names to your secrets for better organization and reference:
+
+=== "YAML"
+
+    ```yaml linenums="1"
+    gateways:
+      duckdb:
+        connection:
+          type: duckdb
+          catalogs:
+            local: local.db
+            remote: "s3://bucket/data/remote.duckdb"
+          extensions:
+            - name: httpfs
+          secrets:
+            my_s3_secret:
+              type: s3
+              region: "YOUR_AWS_REGION"
+              key_id: "YOUR_AWS_ACCESS_KEY"
+              secret: "YOUR_AWS_SECRET_KEY"
+            my_azure_secret:
+              type: azure
+              account_name: "YOUR_AZURE_ACCOUNT"
+              account_key: "YOUR_AZURE_KEY"
+    ```
+
+=== "Python"
+
+    ```python linenums="1"
+    from sqlmesh.core.config import (
+        Config,
+        ModelDefaultsConfig,
+        GatewayConfig,
+        DuckDBConnectionConfig
+    )
+
+    config = Config(
+        model_defaults=ModelDefaultsConfig(dialect="duckdb"),
+        gateways={
+            "duckdb": GatewayConfig(
+                connection=DuckDBConnectionConfig(
+                    catalogs={
+                        "local": "local.db",
+                        "remote": "s3://bucket/data/remote.duckdb"
+                    },
+                    extensions=[
+                        {"name": "httpfs"},
+                    ],
+                    secrets={
+                        "my_s3_secret": {
+                            "type": "s3",
+                            "region": "YOUR_AWS_REGION",
+                            "key_id": "YOUR_AWS_ACCESS_KEY",
+                            "secret": "YOUR_AWS_SECRET_KEY"
+                        },
+                        "my_azure_secret": {
+                            "type": "azure",
+                            "account_name": "YOUR_AZURE_ACCOUNT",
+                            "account_key": "YOUR_AZURE_KEY"
+                        }
+                    }
+                )
+            ),
+        }
+    )
+    ```
+
 After configuring the secrets, you can directly reference S3 paths in your catalogs or in SQL queries without additional authentication steps.
 
 Refer to the official DuckDB documentation for the full list of [supported S3 secret parameters](https://duckdb.org/docs/stable/extensions/httpfs/s3api.html#overview-of-s3-secret-parameters) and for more information on the [Secrets Manager configuration](https://duckdb.org/docs/configuration/secrets_manager.html).
@@ -273,9 +351,9 @@ The `filesystems` accepts a list of file systems to register in the DuckDB conne
           type: duckdb
           catalogs:
             ducklake:
-            type: ducklake
-            path: myducklakecatalog.duckdb
-            data_path: abfs://MyFabricWorkspace/MyFabricLakehouse.Lakehouse/Files/DuckLake.Files
+              type: ducklake
+              path: myducklakecatalog.duckdb
+              data_path: abfs://MyFabricWorkspace/MyFabricLakehouse.Lakehouse/Files/DuckLake.Files
         extensions:
           - ducklake
         filesystems:
diff --git a/sqlmesh/core/config/connection.py b/sqlmesh/core/config/connection.py
index 2c897cd8..f6e12a37 100644
--- a/sqlmesh/core/config/connection.py
+++ b/sqlmesh/core/config/connection.py
@@ -232,7 +232,7 @@ class DuckDBAttachOptions(BaseConfig):
         options = []
         # 'duckdb' is actually not a supported type, but we'd like to allow it for
         # fully qualified attach options or integration testing, similar to duckdb-dbt
-        if self.type not in ("duckdb", "motherduck"):
+        if self.type not in ("duckdb", "ducklake", "motherduck"):
             options.append(f"TYPE {self.type.upper()}")
         if self.read_only:
             options.append("READ_ONLY")
@@ -277,7 +277,7 @@ class BaseDuckDBConnectionConfig(ConnectionConfig):
     catalogs: t.Optional[t.Dict[str, t.Union[str, DuckDBAttachOptions]]] = None
     extensions: t.List[t.Union[str, t.Dict[str, t.Any]]] = []
     connector_config: t.Dict[str, t.Any] = {}
-    secrets: t.List[t.Dict[str, t.Any]] = []
+    secrets: t.Union[t.List[t.Dict[str, t.Any]], t.Dict[str, t.Dict[str, t.Any]]] = []
     filesystems: t.List[t.Dict[str, t.Any]] = []
 
     concurrent_tasks: int = 1
@@ -362,14 +362,22 @@ class BaseDuckDBConnectionConfig(ConnectionConfig):
                         "More info: https://duckdb.org/docs/stable/extensions/httpfs/s3api_legacy_authentication.html"
                     )
                 else:
-                    for secrets in self.secrets:
+                    if isinstance(self.secrets, list):
+                        secrets_items = [(secret_dict, "") for secret_dict in self.secrets]
+                    else:
+                        secrets_items = [
+                            (secret_dict, secret_name)
+                            for secret_name, secret_dict in self.secrets.items()
+                        ]
+
+                    for secret_dict, secret_name in secrets_items:
                         secret_settings: t.List[str] = []
-                        for field, setting in secrets.items():
+                        for field, setting in secret_dict.items():
                             secret_settings.append(f"{field} '{setting}'")
                         if secret_settings:
                             secret_clause = ", ".join(secret_settings)
                             try:
-                                cursor.execute(f"CREATE SECRET ({secret_clause});")
+                                cursor.execute(f"CREATE SECRET {secret_name} ({secret_clause});")
                             except Exception as e:
                                 raise ConfigError(f"Failed to create secret: {e}")
 


Test patch:
diff --git a/tests/core/test_connection_config.py b/tests/core/test_connection_config.py
index 0d7df3d7..6d91c72f 100644
--- a/tests/core/test_connection_config.py
+++ b/tests/core/test_connection_config.py
@@ -4,7 +4,7 @@ import typing as t
 
 import pytest
 from _pytest.fixtures import FixtureRequest
-from unittest.mock import patch
+from unittest.mock import patch, MagicMock
 
 from sqlmesh.core.config.connection import (
     BigQueryConnectionConfig,
@@ -455,6 +455,110 @@ def test_duckdb(make_config):
     assert not config.is_recommended_for_state_sync
 
 
+@patch("duckdb.connect")
+def test_duckdb_multiple_secrets(mock_connect, make_config):
+    """Test that multiple secrets are correctly converted to CREATE SECRET SQL statements."""
+    mock_cursor = MagicMock()
+    mock_connection = MagicMock()
+    mock_connection.cursor.return_value = mock_cursor
+    mock_connection.execute = mock_cursor.execute
+    mock_connect.return_value = mock_connection
+
+    # Create config with 2 secrets
+    config = make_config(
+        type="duckdb",
+        secrets=[
+            {
+                "type": "s3",
+                "region": "us-east-1",
+                "key_id": "my_aws_key",
+                "secret": "my_aws_secret",
+            },
+            {
+                "type": "azure",
+                "account_name": "myaccount",
+                "account_key": "myaccountkey",
+            },
+        ],
+    )
+
+    assert isinstance(config, DuckDBConnectionConfig)
+    assert len(config.secrets) == 2
+
+    # Create cursor which triggers _cursor_init
+    cursor = config.create_engine_adapter().cursor
+
+    execute_calls = [call[0][0] for call in mock_cursor.execute.call_args_list]
+    create_secret_calls = [call for call in execute_calls if call.startswith("CREATE SECRET")]
+
+    # Should have exactly 2 CREATE SECRET calls
+    assert len(create_secret_calls) == 2
+
+    # Verify the SQL for the first secret (S3)
+    assert (
+        create_secret_calls[0]
+        == "CREATE SECRET  (type 's3', region 'us-east-1', key_id 'my_aws_key', secret 'my_aws_secret');"
+    )
+
+    # Verify the SQL for the second secret (Azure)
+    assert (
+        create_secret_calls[1]
+        == "CREATE SECRET  (type 'azure', account_name 'myaccount', account_key 'myaccountkey');"
+    )
+
+
+@patch("duckdb.connect")
+def test_duckdb_named_secrets(mock_connect, make_config):
+    """Test that named secrets are correctly converted to CREATE SECRET SQL statements."""
+    mock_cursor = MagicMock()
+    mock_connection = MagicMock()
+    mock_connection.cursor.return_value = mock_cursor
+    mock_connection.execute = mock_cursor.execute
+    mock_connect.return_value = mock_connection
+
+    # Create config with named secrets using dictionary format
+    config = make_config(
+        type="duckdb",
+        secrets={
+            "my_s3_secret": {
+                "type": "s3",
+                "region": "us-east-1",
+                "key_id": "my_aws_key",
+                "secret": "my_aws_secret",
+            },
+            "my_azure_secret": {
+                "type": "azure",
+                "account_name": "myaccount",
+                "account_key": "myaccountkey",
+            },
+        },
+    )
+
+    assert isinstance(config, DuckDBConnectionConfig)
+    assert len(config.secrets) == 2
+
+    # Create cursor which triggers _cursor_init
+    cursor = config.create_engine_adapter().cursor
+
+    execute_calls = [call[0][0] for call in mock_cursor.execute.call_args_list]
+    create_secret_calls = [call for call in execute_calls if call.startswith("CREATE SECRET")]
+
+    # Should have exactly 2 CREATE SECRET calls
+    assert len(create_secret_calls) == 2
+
+    # Verify the SQL for the first secret (S3) includes the secret name
+    assert (
+        create_secret_calls[0]
+        == "CREATE SECRET my_s3_secret (type 's3', region 'us-east-1', key_id 'my_aws_key', secret 'my_aws_secret');"
+    )
+
+    # Verify the SQL for the second secret (Azure) includes the secret name
+    assert (
+        create_secret_calls[1]
+        == "CREATE SECRET my_azure_secret (type 'azure', account_name 'myaccount', account_key 'myaccountkey');"
+    )
+
+
 @pytest.mark.parametrize(
     "kwargs1, kwargs2, shared_adapter",
     [