Release highlights: 1.30

Breaking changes

• Failed load packages no longer auto-abort. auto_abort_on_terminal_error now defaults to False. With raise_on_failed_jobs=True a terminally failed job is retried and raises LoadClientJobTerminalRetry instead of LoadClientJobFailed, and the package stays pending. drop_pending_packages is deprecated in favour of abort_packages. Set the flag back to True to restore the old behavior (#3557).
• Filesystem layouts without {ext} now append the extension. A layout that omits {ext} previously wrote extension-less files because the documented fallback was unreachable. dlt now appends the job extension, so a table writes mocked-table.jsonl and .jsonl.gz stays intact for compressed jobs. Pipelines already running an ext-less layout will start writing different file names (#4220).
• Table prefix keeps its separator. With layout="{table_name}" the table prefix is now event. rather than event, which stops a replace of event from deleting events. Delta and iceberg tables move to the new location. A new warn_unsafe_layout_separators option (default True) warns when a separator around the table name can occur inside table names (#4283).

Join datasets across destinations

dlt can now join two datasets that live on different destinations (#4286). It attaches the foreign dataset into the query engine of the destination you read from, so the join runs in one engine. Supported for duckdb, motherduck, ducklake, lance, lancedb and filesystem, with eager and lazy materialization. A cross-destination join needs an explicit on.

import dlt

crm = dlt.pipeline("crm", destination="duckdb", dataset_name="crm")
events = dlt.pipeline("events", destination="filesystem", dataset_name="events")

# dlt attaches the filesystem dataset into duckdb and runs the join there
joined = crm.dataset().table("users").join(
    events.dataset().table("events"),
    on="users.id = events.user_id",
)
df = joined.df()

Native nested types on Snowflake

Set use_nested_types on the Snowflake destination to store nested (json) columns as native ARRAY, OBJECT and MAP types instead of a single VARIANT (#4276). dlt derives the structured type from the column's Arrow type, automatically for parquet or from a pyarrow schema you declare for jsonl. Schema and type evolution both work. The interface is experimental for now.

import dlt
import pyarrow as pa

@dlt.resource(
    columns=pa.schema([
        pa.field("payload", pa.struct([("tags", pa.list_(pa.int64())), ("city", pa.string())])),
    ])
)
def items():
    yield {"id": 1, "payload": {"tags": [1, 2], "city": "Berlin"}}

snow = dlt.destinations.snowflake(use_nested_types=True)
pipeline = dlt.pipeline("my_pipeline", destination=snow)
pipeline.run(items(), loader_file_format="jsonl")

Inspect and recover failed load packages

A terminally failed job no longer silently deletes its package (#3557). abort_packages records what happened and lets you inspect or retry a job first, and new Pipeline methods list_pending_retry_jobs_in_package, fail_pending_job and retry_failed_job plus a dlt pipeline <name> abort-packages CLI command drive recovery. Every retry's exception is saved in the package's .exceptions folder.

import os
import dlt

pipeline = dlt.attach("my_pipeline")
load_id = pipeline.list_normalized_load_packages()[0]
jobs = pipeline.list_pending_retry_jobs_in_package(load_id)
job_file, folder = jobs[0]
# give up on a job, keep its exception, then retry the rest
pipeline.fail_pending_job(load_id, os.path.basename(job_file))
pipeline.load()

Retry racing schema migrations with tenacity

Parallel runs against one dataset can race on CREATE TABLE and ALTER TABLE and fail. The new retry_schema_update helper marks that failure as retryable (#4266), so wrapping the run in tenacity retries the schema migration with backoff and jitter instead of failing the load. dlt re-reads the destination on each attempt and applies only what is still missing.

import dlt
from tenacity import retry, retry_if_exception, stop_after_attempt, wait_random_exponential
from dlt.pipeline.helpers import retry_schema_update

pipeline = dlt.pipeline(pipeline_name="chess", destination="duckdb", dataset_name="games")

@retry(
    stop=stop_after_attempt(5),
    wait=wait_random_exponential(multiplier=1, max=30),
    retry=retry_if_exception(retry_schema_update()),
    reraise=True,
)
def load():
    return pipeline.run([{"id": 1}], table_name="games")

Input and output lineage in traces

Traces now carry input and output lineage, in a relational form close to OpenLineage but ingestible by dlt itself (#4289). The sql_database, rest_api and filesystem sources record what each resource read onto the extract step, and dlt derives the tables written per resource onto the load step, so a run's read and write sides are captured automatically. Custom sources can record their own reads with resource.add_input().

Shout-out to new contributors

Big thanks to our newest contributors:

• @sohamwaghe
• @anxkhn
• @maxtaran2010
• @mattfaltyn
• @richacode007-byte
• @axelray-dev
• @chuenchen309
• @bartcode
• @AkhilTrivediX

Full release notes

View the 1.30.0 release notes