Connectors: Add source and target adapters for Delta Lake

docs/changelog.md

@@ -3,6 +3,7 @@
 ## in progress
 
 - Connectors: Updated to Asana client v5
+- Connectors: Added source and target adapters for the Delta Lake table format
 
 ## 2026/06/25 v0.4.0
 

omniload/core/registry.py

@@ -42,6 +42,7 @@
         "appstore": "omniload.source.appstore.api:AppleAppStoreSource",
         "asana": "omniload.source.asana.api:AsanaSource",
         "attio": "omniload.source.attio.api:AttioSource",
+        "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",
@@ -54,6 +55,7 @@
         "dynamodb": "omniload.source.dynamodb.api:DynamoDBSource",
         "elasticsearch": "omniload.source.elasticsearch.api:ElasticsearchSource",
         "facebookads": "omniload.source.facebook_ads.api:FacebookAdsSource",
+        "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",
@@ -64,7 +66,9 @@
         "googleanalytics": "omniload.source.google_analytics.api:GoogleAnalyticsSource",
         "gorgias": "omniload.source.gorgias.api:GorgiasSource",
         "gs": "omniload.source.blobstorage.api:GCSSource",
+        "gs+delta": "omniload.source.deltalake.api:DeltaLakeSource",
         "gsheets": "omniload.source.google_sheets.api:GoogleSheetsSource",
+        "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",
@@ -77,6 +81,7 @@
         "kafka": "omniload.source.kafka.api:KafkaSource",
         "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",
@@ -96,6 +101,7 @@
         "redditads": "omniload.source.reddit_ads.api:RedditAdsSource",
         "revenuecat": "omniload.source.revenuecat.api:RevenueCatSource",
         "s3": "omniload.source.blobstorage.api:S3Source",
+        "s3+delta": "omniload.source.deltalake.api:DeltaLakeSource",
         "salesforce": "omniload.source.salesforce.api:SalesforceSource",
         "sftp": "omniload.source.sftp.api:SFTPSource",
         "shopify": "omniload.source.shopify.api:ShopifySource",
@@ -107,6 +113,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",
         "wise": "omniload.source.wise.api:WiseSource",
         "zendesk": "omniload.source.zendesk.api:ZendeskSource",
         "zoom": "omniload.source.zoom.api:ZoomSource",
@@ -115,6 +122,7 @@
 
 destinations: LazyRegistry = LazyRegistry(
     {
+        "az+delta": "omniload.target.deltalake:DeltaLakeDestination",
         "athena": "omniload.target.athena:AthenaDestination",
         "bigquery": "omniload.target.bigquery:BigQueryDestination",
         "clickhouse": "omniload.target.clickhouse:ClickhouseDestination",
@@ -124,7 +132,11 @@
         "databricks": "omniload.target.databricks:DatabricksDestination",
         "duckdb": "omniload.target.duckdb:DuckDBDestination",
         "elasticsearch": "omniload.target.elasticsearch.api:ElasticsearchDestination",
+        "file+delta": "omniload.target.deltalake:DeltaLakeDestination",
         "gs": "omniload.target.blobstorage: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",
@@ -139,6 +151,7 @@
         "redshift+psycopg2": "omniload.target.redshift:RedshiftDestination",
         "redshift+redshift_connector": "omniload.target.redshift:RedshiftDestination",
         "s3": "omniload.target.blobstorage:S3Destination",
+        "s3+delta": "omniload.target.deltalake:DeltaLakeDestination",
         "snowflake": "omniload.target.snowflake:SnowflakeDestination",
         "sqlite": "omniload.target.sqlite:SqliteDestination",
         "synapse": "omniload.target.synapse:SynapseDestination",

omniload/source/deltalake/adapter.py

@@ -0,0 +1,53 @@
+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): <schema>.<table> 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 <schema>.<table>")
+        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 i, batch in enumerate(
+                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",
+        )()

omniload/source/deltalake/api.py

@@ -0,0 +1,17 @@
+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)

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: Review: 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

pyproject.toml

@@ -82,7 +82,7 @@ dependencies = [
   "databricks-sql-connector<5",
   "databricks-sqlalchemy<3",
   "dataclasses-json<0.7",
-  "dlt<1.29",
+  "dlt[az,deltalake,polars]<1.29",
   "dlt-cratedb<0.2",
   "duckdb<1.6",
   "duckdb-engine<0.18",
@@ -97,6 +97,7 @@ dependencies = [
   "google-cloud-bigquery-storage<3",
   "google-cloud-spanner<4",
   "influxdb-client<2",
+  "lakefs-spec",
   "mysql-connector-python<10",
   "oracledb<4",
   "orjson<4",
@@ -131,6 +132,7 @@ dependencies = [
   "trino[sqlalchemy]==0.337.0",
   "typer<0.27",
   "urllib3<2.8",
+  "yarl<2",
 ]
 optional-dependencies.develop = [
   "pkginfo<2",