From b43360f86b6741c9e895d16c97874e888623c3a2 Mon Sep 17 00:00:00 2001 From: Andreas Motl Date: Fri, 26 Jun 2026 12:39:30 +0200 Subject: [PATCH] Connectors: Add source and target adapters for Delta Lake --- docs/changelog.md | 2 + pyproject.toml | 4 +- src/omniload/core/registry.py | 13 ++++++ src/omniload/source/deltalake/adapter.py | 54 +++++++++++++++++++++++ src/omniload/source/deltalake/api.py | 18 ++++++++ src/omniload/target/deltalake/__init__.py | 20 +++++++++ 6 files changed, 110 insertions(+), 1 deletion(-) create mode 100644 src/omniload/source/deltalake/adapter.py create mode 100644 src/omniload/source/deltalake/api.py create mode 100644 src/omniload/target/deltalake/__init__.py diff --git a/docs/changelog.md b/docs/changelog.md index a83a0c82..e815a5da 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -2,6 +2,8 @@ ## in progress +- Connectors: Added source and target adapters for the Delta Lake table format + ## 2026/08/04 v0.9.0 - Database: Load SQLite and DuckDB databases that live on a remote filesystem, diff --git a/pyproject.toml b/pyproject.toml index bb616fb4..20ef57fd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -83,7 +83,7 @@ dependencies = [ "databricks-sql-connector<5", "databricks-sqlalchemy<3", "dataclasses-json<0.7", - "dlt>=1.22,<1.30", + "dlt[az,deltalake,polars]>=1.22,<1.30", "dlt-cratedb<0.2", "duckdb<1.6", "duckdb-engine<0.18", @@ -98,6 +98,7 @@ dependencies = [ "google-cloud-bigquery-storage<3", "google-cloud-spanner<4", "influxdb-client<2", + "lakefs-spec", "mq-bridge-py>=0.3.2,<0.4,!=0.3.9", "mysql-connector-python<27", "omniload[filesystem]", @@ -135,6 +136,7 @@ dependencies = [ "trino[sqlalchemy]==0.338.0", "typer<0.28", "urllib3<2.8", + "yarl<2", ] optional-dependencies.develop = [ "pkginfo<2", diff --git a/src/omniload/core/registry.py b/src/omniload/core/registry.py index 4b38542d..fd188488 100644 --- a/src/omniload/core/registry.py +++ b/src/omniload/core/registry.py @@ -48,6 +48,7 @@ "attio": "omniload.source.attio.api:AttioSource", "aws+mqb": "omniload.source.mqbridge.api:MqBridgeSource", "az": "dlt_filesystem.source.impl.remote:AzureSource", + "az+delta": "omniload.source.deltalake.api:DeltaLakeSource", "bruin": "omniload.source.bruin.api:BruinSource", "chess": "omniload.source.chess.api:ChessSource", "clickup": "omniload.source.clickup.api:ClickupSource", @@ -63,6 +64,7 @@ "elasticsearch": "omniload.source.elasticsearch.api:ElasticsearchSource", "facebookads": "omniload.source.facebook_ads.api:FacebookAdsSource", "file": "dlt_filesystem.source.fsspec.local:LocalFilesystemSource", + "file+delta": "omniload.source.deltalake.api:DeltaLakeSource", "fireflies": "omniload.source.fireflies.api:FirefliesSource", "fluxx": "omniload.source.fluxx.api:FluxxSource", "frankfurter": "omniload.source.frankfurter.api:FrankfurterSource", @@ -75,8 +77,10 @@ "googleanalytics": "omniload.source.google_analytics.api:GoogleAnalyticsSource", "gorgias": "omniload.source.gorgias.api:GorgiasSource", "gs": "dlt_filesystem.source.impl.remote:GCSSource", + "gs+delta": "omniload.source.deltalake.api:DeltaLakeSource", "gsheets": "omniload.source.google_sheets.api:GoogleSheetsSource", "hdfs": "dlt_filesystem.source.fsspec.hdfs:HDFSSource", + "hdfs+delta": "omniload.source.deltalake.api:DeltaLakeSource", "hostaway": "omniload.source.hostaway.api:HostawaySource", "http": "omniload.source.http.api:HttpSource", "https": "omniload.source.http.api:HttpSource", @@ -93,6 +97,7 @@ "kafka+mqb": "omniload.source.mqbridge.api:MqBridgeSource", "kinesis": "omniload.source.kinesis.api:KinesisSource", "klaviyo": "omniload.source.klaviyo.api:KlaviyoSource", + "lakefs+delta": "omniload.source.deltalake.api:DeltaLakeSource", "linear": "omniload.source.linear.api:LinearSource", "linkedinads": "omniload.source.linkedin_ads.api:LinkedInAdsSource", "mailchimp": "omniload.source.mailchimp.api:MailchimpSource", @@ -122,6 +127,7 @@ "rsync": "omniload.source.rsync.api:RsyncSource", "rsync+ssh": "omniload.source.rsync.api:RsyncSource", "s3": "dlt_filesystem.source.impl.remote:S3Source", + "s3+delta": "omniload.source.deltalake.api:DeltaLakeSource", "salesforce": "omniload.source.salesforce.api:SalesforceSource", "sftp": "dlt_filesystem.source.impl.remote:SFTPSource", "sharepoint": "dlt_filesystem.source.fsspec.sharepoint:SharePointSource", @@ -135,6 +141,7 @@ "stripe": "omniload.source.stripe.api:StripeAnalyticsSource", "tiktok": "omniload.source.tiktok_ads.api:TikTokSource", "trustpilot": "omniload.source.trustpilot.api:TrustpilotSource", + "uc+delta": "omniload.source.deltalake.api:DeltaLakeSource", "webhdfs": "dlt_filesystem.source.fsspec.webhdfs:WebHDFSSource", "wise": "omniload.source.wise.api:WiseSource", "zendesk": "omniload.source.zendesk.api:ZendeskSource", @@ -149,6 +156,7 @@ "adls": "dlt_filesystem.target.api:AzureDestination", "athena": "omniload.target.athena:AthenaDestination", "az": "dlt_filesystem.target.api:AzureDestination", + "az+delta": "omniload.target.deltalake:DeltaLakeDestination", "bigquery": "omniload.target.bigquery:BigQueryDestination", "clickhouse": "omniload.target.clickhouse:ClickhouseDestination", "clickhouse+native": "omniload.target.clickhouse:ClickhouseDestination", @@ -158,7 +166,11 @@ "duckdb": "omniload.target.duckdb:DuckDBDestination", "elasticsearch": "omniload.target.elasticsearch.api:ElasticsearchDestination", "file": "dlt_filesystem.target.api:LocalFilesystemDestination", + "file+delta": "omniload.target.deltalake:DeltaLakeDestination", "gs": "dlt_filesystem.target.api:GCSDestination", + "gs+delta": "omniload.target.deltalake:DeltaLakeDestination", + "hdfs+delta": "omniload.target.deltalake:DeltaLakeDestination", + "lakefs+delta": "omniload.target.deltalake:DeltaLakeDestination", "md": "omniload.target.motherduck:MotherduckDestination", "mongodb": "omniload.target.mongodb:MongoDBDestination", "mongodb+srv": "omniload.target.mongodb:MongoDBDestination", @@ -173,6 +185,7 @@ "redshift+psycopg2": "omniload.target.redshift:RedshiftDestination", "redshift+redshift_connector": "omniload.target.redshift:RedshiftDestination", "s3": "dlt_filesystem.target.api:S3Destination", + "s3+delta": "omniload.target.deltalake:DeltaLakeDestination", "snowflake": "omniload.target.snowflake:SnowflakeDestination", "sqlite": "omniload.target.sqlite:SqliteDestination", "synapse": "omniload.target.synapse:SynapseDestination", diff --git a/src/omniload/source/deltalake/adapter.py b/src/omniload/source/deltalake/adapter.py new file mode 100644 index 00000000..585ef7ea --- /dev/null +++ b/src/omniload/source/deltalake/adapter.py @@ -0,0 +1,54 @@ +from typing import Iterable, Optional + +import dlt +import polars as pl +from dlt.extract import DltResource +from yarl import URL + + +@dlt.source(name="deltalake", max_table_nesting=0) +def deltalake_source( + uri: str, + table: str, + batch_size: Optional[int] = 75_000, +) -> Iterable[DltResource]: + """ + Read from Delta Lake tables. + + Args: + uri (str): A filesystem URI that addresses the Delta Lake catalog. + table (str): . that addresses the Delta Lake table. + batch_size (int): Batch size for Polars + + Returns: + Iterable[DltResource]: Resources with data in random order, + optimized for speed. + """ + + url = URL(uri) + storage_options = dict(url.query) + url = url.with_query(None) + + if url.scheme != "uc": + table_fields = table.split(".") + if len(table_fields) != 2: + raise ValueError("Table name must be in the format .
") + url = url.joinpath(table_fields[-2], table_fields[-1]) + + uri = str(url) + + with pl.Config(streaming_chunk_size=batch_size): + + def reader(): + frame = pl.scan_delta(uri, storage_options=storage_options) + for batch in frame.collect_batches( + engine="streaming", chunk_size=batch_size + ): + yield batch + + return dlt.resource( + reader, + name=table, + # TODO: Are other write dispositions possible? + write_disposition="replace", + )() diff --git a/src/omniload/source/deltalake/api.py b/src/omniload/source/deltalake/api.py new file mode 100644 index 00000000..769a162d --- /dev/null +++ b/src/omniload/source/deltalake/api.py @@ -0,0 +1,18 @@ +class DeltaLakeSource: + def handles_incrementality(self) -> bool: + return True + + def dlt_source(self, uri: str, table: str, **kwargs): + + uri = uri.replace("+delta://", "://") + + # TODO: Review! + if kwargs.get("incremental_key"): + raise ValueError( + "DeltaLake takes care of incrementality on its own, " + "you should not provide incremental_key" + ) + + from omniload.source.deltalake.adapter import deltalake_source + + return deltalake_source(uri, table) diff --git a/src/omniload/target/deltalake/__init__.py b/src/omniload/target/deltalake/__init__.py new file mode 100644 index 00000000..7377adff --- /dev/null +++ b/src/omniload/target/deltalake/__init__.py @@ -0,0 +1,20 @@ +import dlt +from yarl import URL + +from omniload.target.model import GenericSqlDestination + + +class DeltaLakeDestination(GenericSqlDestination): + def dlt_dest(self, uri: str, **kwargs): + kwargs.pop("dest_table", None) + kwargs.pop("staging_bucket", None) + uri = uri.replace("+delta://", "://") + # TODO: Why not prune query parameters from URL when conveyed via `credentials`? + url = URL(uri) + creds = url.query + return dlt.destinations.filesystem(bucket_url=uri, credentials=creds, **kwargs) + + def dlt_run_params(self, uri: str, table: str, **kwargs) -> dict: + params = super().dlt_run_params(uri, table, **kwargs) + params["table_format"] = "delta" + return params