Delta Lake

[amotl]

About

Delta Lake is an open-source storage framework that enables building a format agnostic Lakehouse architecture with many compute engines. For Python, native Delta Lake bindings exist based on delta-rs with Pandas integration. It will be sweet to support Delta Lake like CrateDB Toolkit is doing it.

Evaluation

It looks like dlt supports the format already, see dlt » Delta table format.

References

• https://delta.io/
• https://pypi.org/project/deltalake/

[amotl]

Microsoft Fabric's OneLake also stores tables in Delta Parquet format.

[amotl]

I haven't been able to find any example how to use dlt to write data into a Delta Lake table.

• https://dlthub.com/docs/examples
• https://github.com/dlt-hub/dlt_demos
• https://github.com/dlt-hub/dlt/tree/devel/docs/examples

I am currently figuring that it's all implemented per dlt.destinations.filesystem already, and I would just add "table_format": "delta" to dlt_run_params?

@coderabbitai to the rescue: Can you confirm this, or even better come up with a ready-to-go implementation similar to CsvDestination in omniload/target/csv/__init__.py, but using dlt with deltalake instead?

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== Repository root files =="
fd -d 2 .

echo
echo "== Target subtree =="
tree -a -L 3 omniload/target || true

echo
echo "== Outline: omniload/target/csv/__init__.py =="
ast-grep outline omniload/target/csv/__init__.py --view expanded || true

echo
echo "== File: omniload/target/csv/__init__.py =="
sed -n '1,260p' omniload/target/csv/__init__.py

echo
echo "== Search: dlt-related code in repository =="
rg -n -C3 --type=py '\bdlt\b|filesystem|table_format|delta|Destination|DestinationResponse' omniload pyproject.toml README.md || true

echo
echo "== Candidate dependency/config files =="
fd -a 'pyproject.toml|requirements.*|poetry.lock|uv.lock|Pipfile|README.md' .

echo
echo "== pyproject.toml (first 260 lines, if present) =="
if [ -f pyproject.toml ]; then
  sed -n '1,260p' pyproject.toml
fi

Length of output: 149387

---

🌐 Web query:

dlt filesystem destination Delta table format table_format delta documentation

Length of output: 14908

---

🌐 Web query:

dlt destinations filesystem table_format delta example python

Length of output: 15808

---

🌐 Web query:

dlthub Delta table format filesystem destination write to Delta Lake

Length of output: 18052

---

---

@amotl, yes — your hunch is right.

For omniload, the minimal Delta Lake target is:

• keep using dlt.destinations.filesystem(...)
• return "table_format": "delta" from dlt_run_params()

That is the right pattern here.

Why I’m confident:

• dlt docs explicitly support Delta tables on the filesystem destination via table_format="delta".
• omniload/target/athena/__init__.py already uses the same pattern for Iceberg with "table_format": "iceberg".
• omniload/target/csv/__init__.py is only complicated because CSV needs a temp-dir + post-copy hack. Delta tables are directory-native, so no such hack is needed.

Minimal implementation:

import dlt

from omniload.target.model import GenericSqlDestination


class DeltaLakeDestination(GenericSqlDestination):
    def dlt_dest(self, uri: str, **kwargs):
        return dlt.destinations.filesystem(bucket_url=uri, **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

That should work for local paths and object stores supported by filesystem, e.g. file://, s3://, gs://, az://, subject to the credential shape dlt expects.

Two important notes:

1. pyproject.toml already has pyarrow>=18,<25, which is fine.
2. The repo does not currently depend on deltalake, so you need to add it. dlt uses that library under the hood for Delta writes.

Also useful:

• you do not need to force loader_file_format="parquet"; for Delta, dlt uses Parquet internally anyway.
• merge is supported for Delta with upsert and insert-only strategies when primary_key / merge_key is set.

Relevant docs:

• https://dlthub.com/docs/dlt-ecosystem/destinations/delta-iceberg

• https://dlthub.com/docs/dlt-ecosystem/destinations/filesystem

• https://dlthub.com/docs/hub/ecosystem/delta

• Add a new Delta Lake destination wrapper that reuses dlt.destinations.filesystem

• Register a delta/deltalake destination scheme in the destination registry

• Add the missing deltalake dependency

• Add a small test/docs example showing table_format="delta" behavior

---

💡 For code changes across multiple files, use @coderabbitai plan in an issue comment

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

@amotl Yes — your assumption is correct.

For Delta Lake with dlt, the minimal implementation is:

• use dlt.destinations.filesystem(...)
• set table_format = "delta" in dlt_run_params()

You do not need the CsvDestination temp-dir/copy logic. Delta tables are directory-native.

omniload/target/delta/__init__.py

import dlt

from omniload.target.model import GenericSqlDestination


class DeltaLakeDestination(GenericSqlDestination):
    def dlt_dest(self, uri: str, **kwargs):
        return dlt.destinations.filesystem(bucket_url=uri, **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

Registry entry

Wherever destinations are registered, add:

"delta": "omniload.target.delta:DeltaLakeDestination",
"deltalake": "omniload.target.delta:DeltaLakeDestination",

Dependency

Add the missing runtime dependency:

deltalake = "*"

pyarrow is already present, so that part looks fine.

Why this is the right shape

This mirrors the existing Athena/Iceberg pattern:

• Athena uses filesystem
• Iceberg is enabled by setting table_format="iceberg"
• Delta works the same way with table_format="delta"

Effective run params

For a destination table like analytics.events, this should end up as:

{
    "dataset_name": "analytics",
    "table_name": "events",
    "table_format": "delta",
}

Usage

omniload ingest \
  --source-uri postgresql://... \
  --dest-uri s3://bucket/path \
  --dest-table analytics.events

If you want, I can turn this into a concrete patch plan for #104 with the exact files to add/update.

[amotl]

@coderabbitai: This works well, thank you very much. Based on how other object store source pipeline components are defined in omniload/source/blobstorage/api.py, can you provide an implementation for Delta Lake as well?

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== Find relevant source files =="
fd . omniload/source -t f

echo
echo "== Outline: omniload/source/blobstorage/api.py =="
ast-grep outline omniload/source/blobstorage/api.py --view expanded || true

echo
echo "== File: omniload/source/blobstorage/api.py =="
sed -n '1,260p' omniload/source/blobstorage/api.py

echo
echo "== Find Delta / deltalake references =="
rg -n -C3 --type=py '\bdelta\b|\bdeltalake\b|DeltaTable|write_deltalake|blobstorage|filesystem|DuckDB' omniload pyproject.toml README.md || true

echo
echo "== Outline all source Python files =="
fd . omniload/source -e py -x sh -c 'echo "--- $1 ---"; ast-grep outline "$1" --view expanded || true; echo' sh {}

echo
echo "== Read likely companion files =="
for f in \
  omniload/source/model.py \
  omniload/source/blobstorage/__init__.py \
  omniload/source/blobstorage/reader.py \
  omniload/source/sql.py \
  omniload/core/registry.py \
  pyproject.toml
do
  if [ -f "$f" ]; then
    echo "--- FILE: $f ---"
    sed -n '1,260p' "$f"
    echo
  fi
done

Length of output: 154411

---

🌐 Web query:

deltalake python DeltaTable storage_options S3 GCS to_pyarrow_dataset to_batches documentation

Length of output: 18688

---

---

@amotl Yes — I’d add a dedicated Delta source, not extend omniload/source/blobstorage/api.py’s file-extension readers.

Why: blob storage readers are file-based; Delta tables are directory-based and must be opened via deltalake.DeltaTable(...), then streamed with to_pyarrow_dataset().to_batches().

• Add omniload/source/deltalake/api.py
• Register delta+s3 and delta+gs source schemes
• Add deltalake dependency
• Start with full-snapshot reads only; no incremental/CDF yet

If you want, I can turn this into a tighter patch plan for #104 with exact file diffs and a minimal test case.

🐇✨

---

💡 For code changes across multiple files, use @coderabbitai plan in an issue comment

[amotl]

Support for addressing Delta tables per Unity Catalog will also be nice.

• https://docs.unitycatalog.io/usage/tables/deltalake/
• https://delta-io.github.io/delta-rs/usage/loading-table/#load-table-from-unity-catalog
• https://github.com/benfleis/duckdb-unitycatalog-playground/
• https://docs.databricks.com/aws/en/tables/managed
• https://delta.io/blog/2026-05-06-delta-grows-up-writes-time-travel-and-unity-catalog/
• https://duckdb.org/2026/05/07/delta-uc-updates

[amotl]

@coderabbitai: We've submitted a draft patch, still lacking documentation and software tests. wdyt?

• Connectors: Add source and target adapters for Delta Lake #109

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

repo="$(gh repo view --json nameWithOwner --jq .nameWithOwner)"
echo "== Repo =="
echo "$repo"

echo
echo "== GH-109 details =="
gh issue view 109 --repo "$repo" --comments || true

echo
echo "== Search Delta-related files and registrations =="
fd -i 'delta|deltalake' omniload pyproject.toml README.md docs || true
rg -n -C3 --type=py --type=toml --type=md \
  'DeltaLake|deltalake|table_format|delta\+|filesystem\(|omniload\.source\.deltalake|omniload\.target\.delta|registry' \
  omniload pyproject.toml README.md docs || true

echo
echo "== Outline likely changed files =="
for f in \
  omniload/source/deltalake/api.py \
  omniload/target/delta/__init__.py \
  omniload/core/registry.py \
  omniload/source/blobstorage/api.py \
  omniload/target/csv/__init__.py \
  omniload/target/athena/__init__.py \
  pyproject.toml \
  README.md
do
  if [ -f "$f" ]; then
    echo "--- OUTLINE: $f ---"
    ast-grep outline "$f" --view expanded || true
    echo
  fi
done

echo
echo "== Read likely changed files =="
for f in \
  omniload/source/deltalake/api.py \
  omniload/target/delta/__init__.py \
  omniload/core/registry.py \
  pyproject.toml \
  README.md
do
  if [ -f "$f" ]; then
    echo "--- FILE: $f ---"
    sed -n '1,260p' "$f"
    echo
  fi
done

echo
echo "== Existing tests related to destinations/sources =="
fd -a . tests omniload -e py | sed 's#^\./##' | rg -n 'test|delta|blobstorage|registry|athena|csv' || true
rg -n -C2 --type=py 'DeltaLake|deltalake|table_format|delta\+|filesystem\(' tests omniload || true

Length of output: 35481

---

🏁 Script executed:

#!/bin/bash
set -euo pipefail

repo="$(gh repo view --json nameWithOwner --jq .nameWithOwner)"

echo "== PR `#109` metadata =="
gh pr view 109 --repo "$repo" --json number,title,state,isDraft,author,body,headRefName,baseRefName,url

echo
echo "== PR `#109` changed files =="
gh pr view 109 --repo "$repo" --json files --jq '.files[] | [.path, .additions, .deletions] | `@tsv`'

echo
echo "== PR `#109` diff stats =="
gh pr diff 109 --repo "$repo" --stat

echo
echo "== PR `#109` diff (Delta-related files only, first 260 lines each where possible) =="
for f in \
  omniload/source/deltalake/api.py \
  omniload/target/delta/__init__.py \
  omniload/core/registry.py \
  pyproject.toml \
  README.md \
  docs/changelog.md \
  tests/main/test_factory.py \
  tests/main/test_sources.py \
  tests/main/test_targets.py \
  tests/warehouse/storage/test_filesystem.py \
  tests/warehouse/storage/test_s3.py
do
  echo "--- DIFF FOR: $f ---"
  gh pr diff 109 --repo "$repo" -- "$f" | sed -n '1,260p' || true
  echo
done

Length of output: 2544

---

🏁 Script executed:

#!/bin/bash
set -euo pipefail

repo="$(gh repo view --json nameWithOwner --jq .nameWithOwner)"

echo "== PR `#109` unified diff =="
gh pr diff 109 --repo "$repo" --patch | sed -n '1,520p'

echo
echo "== Raw PR files metadata =="
gh api "repos/$repo/pulls/109/files?per_page=100" | jq -r '.[] | "FILE\t\(.filename)\t+\(.additions)\t-\(.deletions)\nPATCH\n\(.patch // "<no patch>")\n----"'

Length of output: 19617

---

---

@amotl This looks good as a draft. The shape of #109 is right.

What I like:

• adds both source and target adapters
• wires the registry comprehensively
• uses table_format="delta" on the target side
• includes uc+delta on the source side
• pulls in the needed dlt extras

So: good direction, good file layout, good scope.

Before merge, I’d tighten these points:

1. Incrementality contract

   • omniload/source/deltalake/api.py returns handles_incrementality() == True
   • but dlt_source() rejects incremental_key

   That reads inconsistent. If the source currently does snapshot reads only, I would either:

   • return False, or
   • document clearly that Delta handles versioned snapshots internally, but omniload-level incremental_key is unsupported.

2. Target URI/query handling

   • omniload/target/deltalake/__init__.py passes:
     • bucket_url=uri
     • credentials=url.query

   I would review that carefully. Right now the query params seem to go in twice: once embedded in the URL, once as credentials. The TODO there is valid. I’d prefer to strip the query from bucket_url and pass credentials/storage options separately.

3. Unity Catalog scope

   • source registry includes uc+delta
   • destination registry does not

   That is fine for now, but docs should say this explicitly so users do not assume UC writes are supported yet.

4. kwargs passthrough

   • omniload/source/deltalake/api.py currently ignores all kwargs except incremental_key

   If you want batch_size, version selection, or future projection/filter knobs, this is the place to thread them through.

Docs I’d add:

• one example each for:
  • file+delta://...
  • s3+delta://...
  • uc+delta://...
• explain that URI query parameters become Delta storage options / credentials
• explain current read semantics:
  • latest snapshot by default
  • no omniload incremental_key
  • row order is not guaranteed
• explain current UC support is read-only, if that is the intended scope

Tests I’d add:

• registry resolves s3+delta, gs+delta, file+delta, uc+delta
• target dlt_run_params() adds table_format == "delta"
• target strips +delta:// correctly
• source rejects incremental_key
• source path composition:
  • non-UC: schema.table becomes <uri>/<schema>/<table>
  • UC: table name is passed through without path joining logic changing the URI
• query params are handled as intended on both source and target

Net: solid draft, close to mergeable once docs/tests land and the incrementality/query semantics are clarified.