Feat: add support for METADATA_SCHEMA in ducklake config.yml (#5616)

Author: georgesittas

docs/integrations/engines/duckdb.md

@@ -81,8 +81,9 @@ SQLMesh will place models with the explicit catalog "ephemeral", such as `epheme
               data_path: data/ducklake
               encrypted: True
               data_inlining_row_limit: 10
+              metadata_schema: main
     ```
-    
+
 === "Python"
 
     ```python linenums="1"
@@ -106,6 +107,7 @@ SQLMesh will place models with the explicit catalog "ephemeral", such as `epheme
                             data_path="data/ducklake",
                             encrypted=True,
                             data_inlining_row_limit=10,
+                            metadata_schema="main",
                         ),
                     }
                 )
@@ -114,6 +116,14 @@ SQLMesh will place models with the explicit catalog "ephemeral", such as `epheme
     )
     ```
 
+**DuckLake Configuration Options:**
+
+- `path`: Path to the DuckLake catalog file
+- `data_path`: Path where DuckLake data files are stored
+- `encrypted`: Whether to enable encryption for the catalog (default: `False`)
+- `data_inlining_row_limit`: Maximum number of rows to inline in the catalog (default: `0`)
+- `metadata_schema`: The schema in the catalog server in which to store the DuckLake metadata tables (default: `main`)
+
 #### 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

@@ -238,6 +238,7 @@ class DuckDBAttachOptions(BaseConfig):
     data_path: t.Optional[str] = None
     encrypted: bool = False
     data_inlining_row_limit: t.Optional[int] = None
+    metadata_schema: t.Optional[str] = None
 
     def to_sql(self, alias: str) -> str:
         options = []
@@ -259,6 +260,8 @@ def to_sql(self, alias: str) -> str:
                 options.append("ENCRYPTED")
             if self.data_inlining_row_limit is not None:
                 options.append(f"DATA_INLINING_ROW_LIMIT {self.data_inlining_row_limit}")
+            if self.metadata_schema is not None:
+                options.append(f"METADATA_SCHEMA '{self.metadata_schema}'")
 
         options_sql = f" ({', '.join(options)})" if options else ""
         alias_sql = ""

tests/core/test_connection_config.py

@@ -822,6 +822,37 @@ def test_ducklake_attach_add_ducklake_prefix():
     )
 
 
+def test_ducklake_metadata_schema():
+    # Test that metadata_schema parameter is included when specified
+    options = DuckDBAttachOptions(
+        type="ducklake", path="catalog.ducklake", metadata_schema="custom_schema"
+    )
+    assert (
+        options.to_sql(alias="my_ducklake")
+        == "ATTACH IF NOT EXISTS 'ducklake:catalog.ducklake' AS my_ducklake (METADATA_SCHEMA 'custom_schema')"
+    )
+
+    # Test that metadata_schema is not included when not specified (default behavior)
+    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 metadata_schema with other ducklake options
+    options = DuckDBAttachOptions(
+        type="ducklake",
+        path="catalog.ducklake",
+        data_path="/path/to/data",
+        encrypted=True,
+        metadata_schema="workspace_schema",
+    )
+    assert (
+        options.to_sql(alias="my_ducklake")
+        == "ATTACH IF NOT EXISTS 'ducklake:catalog.ducklake' AS my_ducklake (DATA_PATH '/path/to/data', ENCRYPTED, METADATA_SCHEMA 'workspace_schema')"
+    )
+
+
 def test_duckdb_config_json_strings(make_config):
     config = make_config(
         type="duckdb",