DuckLake support

[adrianbr]

Specs from rudolfix

(written by zilto, adapted from: #2709 (comment))

Tasks

Create a new ducklake destination; it should inherit a lot of features from duckdbdestination.

Configuration

To allow to configure the catalog (a database) and storage (a filesystem), DuckLakeCredentials should derive from duckdb configuration and have the following signature

@configspec(init=False)
class DuckLakeCredentials(DuckDbBaseCredentials):
    drivername: Final[str] = dataclasses.field(  # type: ignore
        default="ducklake", init=False, repr=False, compare=False
    )
    username: str
    password: TSecretStrValue = None
    database: str  # the name of the ducklake; required by DuckLakeSqlClient
    catalog: ConnectionStringCredentials  # for catalog; like postgres
    storage: FilesystemConfiguration  # for data;

Users will be able to configure the catalog and the storage from their config.toml and secrets.toml

Resolve catalog and storage secrets

The duckdb connection needs credentials to the catalog (postgres example) and the storage (supported storage)

• for storage, the duckdb destination already has a feature to get secrets from FilesystemConfiguration (i.e., the mechanism that allows to query S3 with duckdb)
• for catalog, we need to implement the function to get secrets from ConnectionStringCredentials

Configure DuckDB instance to support DuckLake

The class DuckDbBaseCredentials allows to set extensions, pragmas, global config, and local config. This allows to load the ducklake extension, but not to install it.

• DuckLakeCredentials should inherit from DuckDbBaseCredentials, but enforce the ducklake extension to be installed
• Set the current database to the ducklake name (details here).
• A lot of filesystem are supported as duckdb extensions. httpfs supports all S3-compliant APIs. Also, it supports python fsspec. We can progressively add support here

Out-of-scope

• ducklake table maintenance; this should be done by the user directly against their ducklake instance

Future work

• Add ducklake to TTableFormat
• DuckLakeClient should implement SupportsOpenTables to allow users to get authenticated catalog and table relation from the pipeline/destination ie. to do table maintenance. this is how delta and iceberg work.

Original issue

Feature description

we should support ducklake

User reported it currently doesn't work with our duckdb destination and we likely need to make some adjustments.

https://www.linkedin.com/feed/update/urn:li:activity:7335310253997096963?commentUrn=urn%3Ali%3Acomment%3A%28activity%3A7335310253997096963%2C7335375393492733952%29&dashCommentUrn=urn%3Ali%3Afsd_comment%3A%287335375393492733952%2Curn%3Ali%3Aactivity%3A7335310253997096963%29

[mattiasthalen]

Here's an example of the code I'm using:

import duckdb
import dlt

def attach_ducklake(db__name: str = '') -> duckdb.DuckDBPyConnection:
    """DuckDB connection for DuckLake."""

    # Initialize DuckDB connection
    db = duckdb.connect(db__name)

    # Setup postgres as catalog backend for DuckLake
    db.install_extension("postgres")
    db.execute(f"""
        CREATE SECRET(
            TYPE postgres,
            HOST '{os.getenv("PG__HOST")}',
            PORT {os.getenv("PG__PORT")},
            DATABASE '{os.getenv("PG__DATABASE")}',
            USER '{os.getenv("PG__USER")}',
            PASSWORD '{os.getenv("PG__PASSWORD")}'
        );
    """)

    # Setup R2 as storage backend for DuckLake
    db.install_extension("httpfs")
    db.load_extension("httpfs")
    db.execute(f"""
        CREATE SECRET(
            TYPE r2,
            KEY_ID '{os.getenv("R2__ACCESS_KEY_ID")}',
            SECRET '{os.getenv("R2__SECRET_ACCESS_KEY")}',
            ACCOUNT_ID '{os.getenv("R2__ACCOUNT_ID")}'
        );
    """)

    # Attach the DuckLake
    db.install_extension("ducklake")
    db.execute(f"ATTACH 'ducklake:postgres:dbname={os.getenv("PG__DATABASE")} host={os.getenv("PG__HOST")}' AS ducklake (DATA_PATH 'r2://{os.getenv("R2__BUCKET")}', ENCRYPTED);")
    db.execute("USE ducklake;")

    return db

ducklake = attach_ducklake()

pipeline = dlt.pipeline(
    pipeline_name="adventure_works",
    destination=dlt.destinations.duckdb(ducklake),
    dataset_name="das"
)

source = adventure_works_source()

load_info = pipeline.run(source)
print(load_info)

ducklake.sql("SELECT current_database();").show()
ducklake.sql("SHOW ALL TABLES;").show()
ducklake.close()

Pipeline ducklake load step completed in 2.70 seconds
1 load package(s) were loaded to destination duckdb and into dataset das_20250603092645
The duckdb destination used duckdb:///:external: location to store data
Load package 1748942806.2843683 is LOADED and contains no failed jobs
┌────────────────────┐
│ current_database() │
│      varchar       │
├────────────────────┤
│ ducklake           │
└────────────────────┘

┌──────────────────────┬──────────────────────┬──────────────────────┬──────────────────────┬───────────────────────────────────────────────────────┬───────────┐
│       database       │        schema        │         name         │     column_names     │                     column_types                      │ temporary │
│       varchar        │       varchar        │       varchar        │      varchar[]       │                       varchar[]                       │  boolean  │
├──────────────────────┼──────────────────────┼──────────────────────┼──────────────────────┼───────────────────────────────────────────────────────┼───────────┤
│ memory               │ das_20250603092645   │ _dlt_loads           │ [load_id, schema_n…  │ [VARCHAR, VARCHAR, BIGINT, TIMESTAMP WITH TIME ZONE…  │ false     │
│ memory               │ das_20250603092645   │ _dlt_pipeline_state  │ [version, engine_v…  │ [BIGINT, BIGINT, VARCHAR, VARCHAR, TIMESTAMP WITH T…  │ false     │
│ memory               │ das_20250603092645   │ _dlt_version         │ [version, engine_v…  │ [BIGINT, BIGINT, TIMESTAMP WITH TIME ZONE, VARCHAR,…  │ false     │
│ memory               │ das_20250603092645   │ raw__adventure_wor…  │ [department_id, na…  │ [BIGINT, VARCHAR, VARCHAR, VARCHAR, VARCHAR, VARCHAR] │ false     │
├──────────────────────┴──────────────────────┴──────────────────────┴──────────────────────┴───────────────────────────────────────────────────────┴───────────┤
│ 33 rows (20 shown)                                                                                                                                  6 columns │
└───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘

My theory is that dlt.destinations.duckdb ignores USE ducklake;

[esadek]

Here's a more concise example. DuckDB is able to create a table in the correct my_ducklake database after it is selected with a USE statement. However, even though the same connection is used in the dlt pipeline destination, dlt loads data into memory. In my mind there are two potential solutions: dlt could automatically use the selected (default) database, or the database name could be a parameter for the pipeline or destination, similarly to how dataset_name can be used to specify the schema.

import dlt
import duckdb

con = duckdb.connect()
con.install_extension("ducklake")
con.execute("ATTACH 'ducklake:metadata.ducklake' AS my_ducklake;")
con.execute("USE my_ducklake;")
con.execute("CREATE TABLE people AS SELECT 1 AS id, 'Alice' AS name;")


@dlt.resource()
def events():
    yield [{"event_id": 1, "event_tstamp": "2024-07-30T10:00:00.123"}]


pipeline = dlt.pipeline(
    pipeline_name="ducklake_example",
    destination=dlt.destinations.duckdb(con),
)

load_info = pipeline.run(events)
print(load_info)

con.sql("SHOW ALL TABLES;").show()
con.close()

Pipeline ducklake_example load step completed in 0.02 seconds
1 load package(s) were loaded to destination duckdb and into dataset ducklake_example_dataset
The duckdb destination used duckdb:///:external: location to store data
Load package 1749079863.409554 is LOADED and contains no failed jobs
┌──────────────────────┬──────────────────────┬──────────────────────┬──────────────────────┬──────────────────────────────────────────────────────┬───────────┐
│       database       │        schema        │         name         │     column_names     │                     column_types                     │ temporary │
│       varchar        │       varchar        │       varchar        │      varchar[]       │                      varchar[]                       │  boolean  │
├──────────────────────┼──────────────────────┼──────────────────────┼──────────────────────┼──────────────────────────────────────────────────────┼───────────┤
│ __ducklake_metadat…  │ main                 │ ducklake_column      │ [column_id, begin_…  │ [BIGINT, BIGINT, BIGINT, BIGINT, BIGINT, VARCHAR, …  │ false     │
│ __ducklake_metadat…  │ main                 │ ducklake_column_tag  │ [table_id, column_…  │ [BIGINT, BIGINT, BIGINT, BIGINT, VARCHAR, VARCHAR]   │ false     │
│ __ducklake_metadat…  │ main                 │ ducklake_data_file   │ [data_file_id, tab…  │ [BIGINT, BIGINT, BIGINT, BIGINT, BIGINT, VARCHAR, …  │ false     │
│ __ducklake_metadat…  │ main                 │ ducklake_delete_file │ [delete_file_id, t…  │ [BIGINT, BIGINT, BIGINT, BIGINT, BIGINT, VARCHAR, …  │ false     │
│ __ducklake_metadat…  │ main                 │ ducklake_file_colu…  │ [data_file_id, tab…  │ [BIGINT, BIGINT, BIGINT, BIGINT, BIGINT, BIGINT, V…  │ false     │
│ __ducklake_metadat…  │ main                 │ ducklake_file_part…  │ [data_file_id, tab…  │ [BIGINT, BIGINT, BIGINT, VARCHAR]                    │ false     │
│ __ducklake_metadat…  │ main                 │ ducklake_files_sch…  │ [data_file_id, pat…  │ [BIGINT, VARCHAR, BOOLEAN, TIMESTAMP WITH TIME ZONE] │ false     │
│ __ducklake_metadat…  │ main                 │ ducklake_inlined_d…  │ [table_id, table_n…  │ [BIGINT, VARCHAR, BIGINT]                            │ false     │
│ __ducklake_metadat…  │ main                 │ ducklake_metadata    │ [key, value]         │ [VARCHAR, VARCHAR]                                   │ false     │
│ __ducklake_metadat…  │ main                 │ ducklake_partition…  │ [partition_id, tab…  │ [BIGINT, BIGINT, BIGINT, BIGINT, VARCHAR]            │ false     │
│          ·           │  ·                   │       ·              │          ·           │                     ·                                │   ·       │
│          ·           │  ·                   │       ·              │          ·           │                     ·                                │   ·       │
│          ·           │  ·                   │       ·              │          ·           │                     ·                                │   ·       │
│ __ducklake_metadat…  │ main                 │ ducklake_table       │ [table_id, table_u…  │ [BIGINT, UUID, BIGINT, BIGINT, BIGINT, VARCHAR]      │ false     │
│ __ducklake_metadat…  │ main                 │ ducklake_table_col…  │ [table_id, column_…  │ [BIGINT, BIGINT, BOOLEAN, BOOLEAN, VARCHAR, VARCHAR] │ false     │
│ __ducklake_metadat…  │ main                 │ ducklake_table_stats │ [table_id, record_…  │ [BIGINT, BIGINT, BIGINT, BIGINT]                     │ false     │
│ __ducklake_metadat…  │ main                 │ ducklake_tag         │ [object_id, begin_…  │ [BIGINT, BIGINT, BIGINT, VARCHAR, VARCHAR]           │ false     │
│ __ducklake_metadat…  │ main                 │ ducklake_view        │ [view_id, view_uui…  │ [BIGINT, UUID, BIGINT, BIGINT, BIGINT, VARCHAR, VA…  │ false     │
│ memory               │ ducklake_example_d…  │ _dlt_loads           │ [load_id, schema_n…  │ [VARCHAR, VARCHAR, BIGINT, TIMESTAMP WITH TIME ZON…  │ false     │
│ memory               │ ducklake_example_d…  │ _dlt_pipeline_state  │ [version, engine_v…  │ [BIGINT, BIGINT, VARCHAR, VARCHAR, TIMESTAMP WITH …  │ false     │
│ memory               │ ducklake_example_d…  │ _dlt_version         │ [version, engine_v…  │ [BIGINT, BIGINT, TIMESTAMP WITH TIME ZONE, VARCHAR…  │ false     │
│ memory               │ ducklake_example_d…  │ events               │ [event_id, event_t…  │ [BIGINT, TIMESTAMP WITH TIME ZONE, VARCHAR, VARCHAR] │ false     │
│ my_ducklake          │ main                 │ people               │ [id, name]           │ [INTEGER, VARCHAR]                                   │ false     │
├──────────────────────┴──────────────────────┴──────────────────────┴──────────────────────┴──────────────────────────────────────────────────────┴───────────┤
│ 24 rows (20 shown)                                                                                                                                 6 columns │
└──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘

[rudolfix]

@esadek @mattiasthalen loading step in dlt is multithreaded and we use conn.cursor() to get a clone of the original connection. several settings are reset for such cursor (ie. all/some? OPTIONS) I bet USE is also not applied. You could try to hack the dlt code to see if that is true:

@configspec(init=False)
class DuckDbBaseCredentials(ConnectionStringCredentials):
    read_only: bool = False  # open database read/write

    def borrow_conn(self, read_only: bool) -> Any:
        import duckdb

        if not hasattr(self, "_conn_lock"):
            self._conn_lock = threading.Lock()

        config = self._get_conn_config()
        # obtain a lock because duck releases the GIL and we have refcount concurrency
        with self._conn_lock:
            if not hasattr(self, "_conn"):
                self._conn = duckdb.connect(
                    database=self._conn_str(), read_only=read_only, config=config
                )
                self._conn_owner = True
                self._conn_borrows = 0

            # track open connections to properly close it
            self._conn_borrows += 1
            # print(f"getting conn refcnt {self._conn_borrows} at {id(self)}")
            return self._conn.cursor()

borrow_conn is used by all the dlt so you can run "init" scripts on returned cursor and they will be applied in loader.

the quick hack would be to add "on open" and "on borrow" script to duckdb config and run them accordingly. a real implementation of ducklake would be similar to Motherduck. in particular we'd derive from DuckDbClientConfiguration to add FilesystemCredentials for bucket and ConnectionStringCredentials for the catalog.

if someone is interested we can take contribution here. In that case I can write better spec.

I hope duckdb tests could be re-used. like we enable motherduck tests for duckdb tests.

[mattiasthalen]

@rudolfix, I'd be up to contribute. I've done some experementation and also went down the MD route to see how it's solved there.

[rudolfix]

@mattiasthalen tell me if this is helpful or you need something more detailed. I'm pretty sure you can hack a working destination pretty quickly and then we can do a few iterations of reviews.

Let's do it as a separate destination. Let's name it ducklake

1. prerequisites
   We need to merge fixes leaking datasets tests #2730 first:

• allows to easy setup extensions, configs and pragmas
• allows to create secrets for storages

Look how Motherduck is implemented. It derives from various duckdb classes

1. configuration
   DuckLakeConfiguration should derive from duckdb configuration

• adds catalog field of type ConnectionStringCredentials (ie. postgres) to create catalog secret
• adds filesystem of type FilesysytemConfiguration - to create storage secret
• credentials field expect database name. same way as Motherduck

2. Secrets:

• we already have a method in DuckdbSqlClient that creates secret from FilesystemConfiguration
• add similar method for ConnectionStringCredentials to create database secret (ie. postgres, mysql, skip for duckdb obviously)

3. you'll need a customized DuckLakeSqlClient which will setup secrets and run 6additional scripts on first open connection. again see MotherduckSqlClient

• we are placing current database and schema in search_path so user can query tables without qualification
• I'm not sure if we should install extensions seamlessly. better to request users to do that. WDYT?

4. destination capabilities (factory module)

• it looks to me we need another table format! let's name it ducklake. set it here:

preferred_table_format: TTableFormat = None
    supported_table_formats: Sequence[TTableFormat] = None

• you can also switch preferred loader file format to parquet

5. type mapper: I think duckdb type mapper

6. Future extensions:

• DuckLakeClient should implement SupportsOpenTables to allow users to get authenticated catalog and table relation from the pipeline/destination ie. to do table maintenance. this is how delta and iceberg work.

[matsonj]

@rudolfix any way we can help move this forward?

[rudolfix]

@matsonj all the building blocks to assemble this destination are +- ready in existing code base. it there anything special that could help motherduck? ie. using md:// as catalog? maybe that could be interesting

@mattiasthalen are you still interested in implementing this? if you are busy with something else we can take over :)

[mattiasthalen]

@rudolfix, sorry, completely forgot about this!

Have been busy with clients and getting sqlglot/sqlmesh into fabric 😅

And it doesn't look like I'll have much time in the near future either.

[matsonj]

@rudolfix yeah I think we could definitely do md: as a catalog! That would be really awesome. I'm flagging this issue internally for some feedback as well.

[matsonj]

Ok so some good news: The existing MotherDuck connector works on Managed DuckLake that we announced today. You just have to create the DuckLake in advance.

Config in my secrets.toml:

pipeline config in my script:

    pipeline = dlt.pipeline(
        pipeline_name='stock_data',
        destination='motherduck',
        dataset_name='stock_data'
    )

and of course, the data in my s3 bucket

I'm fuzzy on if this will work for self-hosted DuckLake enjoyers, however.

[richard-gyiko]

Anyone has a workaround for self-hosted DuckLake users?

[deschman]

Anyone has a workaround for self-hosted DuckLake users?

For writing from external systems, we prefer to go to flat files instead of loading direct to DuckLake. That way, its just a matter of reading from those flat files in any properly configured DuckDB, DuckLake, or something else.

For reverse ETL from DuckLake to other systems, I got very close this week but need to turn my attention elsewhere. I think the workaround would be to utilize engine_adapter_callback. My callback function has these few lines in it to ensure my cloud filesystem is registered and DuckLake attaches whenever I connect.

from textwrap import dedent

from fsspec import filesystem
from sqlalchemy import Engine, create_engine, event

engine = create_engine(
    engine.url,
    connect_args={
        'register_filesystems': [filesystem('gcs')],
        'preload_extensions': [
            'delta',
            'ducklake',
        ],
    },
)

event.listen(
    engine,
    'connect',
    lambda conn, record: conn.execute(dedent(f"""
        ATTACH 'ducklake:{metadata_path}' AS local_ducklake
        (DATA_PATH '{data_path}');
        USE local_ducklake;
    """)),
)

While I have verified that this callback does what it is intended to do, and connections are able to read from DuckLake, I'm still getting the error below in my traceback.

ProgrammingError: (duckdb.duckdb.ParserException) Parser Error: SELECT clause without selection list
[SQL: SELECT 
FROM customer.loyalty]
(Background on this error at: https://sqlalche.me/e/20/f405)

Would love to know how to get this over the finish line if anyone picks up where I left off.