fix: add 'ducklake prefix if not present (#4941)

Author: eakmanrq

sqlmesh/core/config/connection.py

@@ -238,7 +238,10 @@ def to_sql(self, alias: str) -> str:
             options.append("READ_ONLY")
 
         # DuckLake specific options
+        path = self.path
         if self.type == "ducklake":
+            if not path.startswith("ducklake:"):
+                path = f"ducklake:{path}"
             if self.data_path is not None:
                 options.append(f"DATA_PATH '{self.data_path}'")
             if self.encrypted:
@@ -254,7 +257,7 @@ def to_sql(self, alias: str) -> str:
         alias_sql = (
             f" AS {alias}" if not (self.type == "motherduck" or self.path.startswith("md:")) else ""
         )
-        return f"ATTACH IF NOT EXISTS '{self.path}'{alias_sql}{options_sql}"
+        return f"ATTACH IF NOT EXISTS '{path}'{alias_sql}{options_sql}"
 
 
 class BaseDuckDBConnectionConfig(ConnectionConfig):

tests/core/test_connection_config.py

@@ -742,9 +742,29 @@ def test_duckdb_attach_ducklake_catalog(make_config):
     assert ducklake_catalog.encrypted is True
     assert ducklake_catalog.data_inlining_row_limit == 10
     # 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")
-    assert "DATA_INLINING_ROW_LIMIT 10" in ducklake_catalog.to_sql("ducklake")
+    generated_sql = ducklake_catalog.to_sql("ducklake")
+    assert "DATA_PATH '/tmp/ducklake_data'" in generated_sql
+    assert "ENCRYPTED" in generated_sql
+    assert "DATA_INLINING_ROW_LIMIT 10" in generated_sql
+    # Check that the ducklake: prefix is automatically added
+    assert "ATTACH IF NOT EXISTS 'ducklake:catalog.ducklake'" in generated_sql
+
+    # Test that a path with existing ducklake: prefix is preserved
+    config_with_prefix = make_config(
+        type="duckdb",
+        catalogs={
+            "ducklake": DuckDBAttachOptions(
+                type="ducklake",
+                path="ducklake:catalog.ducklake",
+                data_path="/tmp/ducklake_data",
+            ),
+        },
+    )
+    ducklake_catalog_with_prefix = config_with_prefix.catalogs.get("ducklake")
+    generated_sql_with_prefix = ducklake_catalog_with_prefix.to_sql("ducklake")
+    assert "ATTACH IF NOT EXISTS 'ducklake:catalog.ducklake'" in generated_sql_with_prefix
+    # Ensure we don't have double prefixes
+    assert "'ducklake:catalog.ducklake" in generated_sql_with_prefix
 
 
 def test_duckdb_attach_options():
@@ -762,6 +782,22 @@ def test_duckdb_attach_options():
     assert options.to_sql(alias="db") == "ATTACH IF NOT EXISTS 'test.db' AS db"
 
 
+def test_ducklake_attach_add_ducklake_prefix():
+    # Test that ducklake: prefix is automatically added when missing
+    options = DuckDBAttachOptions(type="ducklake", path="catalog.ducklake")
+    assert (
+        options.to_sql(alias="my_ducklake")
+        == "ATTACH IF NOT EXISTS 'ducklake:catalog.ducklake' AS my_ducklake"
+    )
+
+    # Test that ducklake: prefix is preserved when already present
+    options = DuckDBAttachOptions(type="ducklake", path="ducklake:catalog.ducklake")
+    assert (
+        options.to_sql(alias="my_ducklake")
+        == "ATTACH IF NOT EXISTS 'ducklake:catalog.ducklake' AS my_ducklake"
+    )
+
+
 def test_duckdb_config_json_strings(make_config):
     config = make_config(
         type="duckdb",