DuckLake: partitioned_by on INCREMENTAL_BY_UNIQUE_KEY results in unpartitioned files on initial build

[nathantapsas]

When using SQLMesh with DuckDB + DuckLake, models configured with partitioned_by (including INCREMENTAL_BY_UNIQUE_KEY) are not physically partitioned on initial table creation.

SQLMesh logs show:

CREATE OR REPLACE TABLE ... AS SELECT ...
ALTER TABLE ... SET PARTITIONED BY (...)
After this sequence, the initial data file remains unpartitioned (partition_id = NULL in DuckLake metadata). Only subsequent writes (INSERT / MERGE) produce partitioned files and partition paths.

SQLMesh / environment
SQLMesh: 0.231.1
Engine: DuckDB 1.4.4 with DuckLake extension
Connection: DuckDB catalog attached as DuckLake (TYPE ducklake, DATA_PATH set)
Model config example
MODEL (
name silver.accounts_snapshot,
kind INCREMENTAL_BY_UNIQUE_KEY (
unique_key (account_number, __data_snapshot_date),
),
partitioned_by __data_snapshot_date,
);
Expected behavior
If partitioned_by is configured, initial table build should physically write partitioned files (or docs should clearly state initial CTAS files will be unpartitioned and only future writes will partition).

Actual behavior
Initial build creates unpartitioned file(s).

Why this matters
On full rebuilds / first deployments, large snapshot tables are not physically partitioned as configured, which can impact pruning, storage layout, and performance until later incremental writes occur.

Possible fix direction
For DuckLake targets with partitioned_by, consider creating table schema + partition spec first, then inserting data (instead of CTAS then ALTER TABLE ... SET PARTITIONED BY), or perform a post-create rewrite when needed.

[nathantapsas]

I prototyped a local fix in DuckDBEngineAdapter._create_table for DuckLake + partitioned_by tables.

What I changed
For DuckLake, when partitioned_by is set and a CTAS expression is provided:

1. Avoid CTAS-first data write.
2. Create schema-only table from the query (SELECT * FROM () WHERE FALSE LIMIT 0).
3. Apply ALTER TABLE ... SET PARTITIONED BY (...).
4. Insert data with INSERT INTO ... .

Why
Current flow is CTAS first, then ALTER TABLE ... SET PARTITIONED BY, which leaves the initial file unpartitioned (partition_id = NULL).
This change ensures the first real write occurs after partitioning is configured.

Important note
In this path, target_columns_to_types is often None for non-annotated SQL models, so the fix should not depend on it.

if catalog_type == "ducklake":
    partitioned_by_exps = kwargs.pop("partitioned_by", None)
    if partitioned_by_exps and expression is not None:
        should_insert_after_create = True
        insert_expression = expression.copy()

        # schema-only create (no data written yet)
        expression = (
            exp.select("*")
            .from_(expression.subquery("_sqlmesh_schema_only", copy=False))
            .where(exp.false())
            .limit(0)
        )

super()._create_table(...)

if partitioned_by_exps:
    self.execute(f"ALTER TABLE {table_name_str} SET PARTITIONED BY ({partitioned_by_str});")
    if should_insert_after_create and insert_expression is not None:
        self.execute(exp.insert(insert_expression, exp.to_table(table_name)))