Delta-Erweiterung

Die Erweiterung delta ergänzt Unterstützung für das offene Speicherformat Delta Lake. Sie basiert auf dem Delta Kernel. Die Erweiterung bietet Lese- und Schreibunterstützung für Delta-Tabellen, lokal und remote.

Implementierungsdetails stehen im Ankündigungs-Blogbeitrag.

Warnung Uns ist eine Regression in Azure Onelake bekannt, die offenbar eine Folge einer Änderung in delta-kernel-rs ist. Sie können das Issue auf GitHub verfolgen.

Für die Verbindung zum Unity Catalog hat DuckDB die Erweiterung unity_catalog.

Installation und Laden

Die Erweiterung delta wird beim ersten Einsatz transparent aus dem offiziellen Erweiterungs-Repository automatisch geladen. Wenn Sie sie manuell installieren und laden möchten, führen Sie aus:

INSTALL delta;
LOAD delta;

Verwendung

Um eine lokale Delta-Tabelle zu scannen, führen Sie aus:

SELECT *
FROM delta_scan('file:///some/path/on/local/machine');

Aus einem S3-Bucket lesen

Um eine Delta-Tabelle in einem S3-Bucket zu scannen, führen Sie aus:

SELECT *
FROM delta_scan('s3://some/delta/table');

Zur Authentifizierung an S3-Buckets werden DuckDB-Secrets unterstützt:

CREATE SECRET (
    TYPE s3,
    PROVIDER credential_chain
);
SELECT *
FROM delta_scan('s3://some/delta/table/with/auth');

Um öffentliche Buckets auf S3 zu scannen, müssen Sie möglicherweise die richtige Region übergeben, indem Sie ein Secret mit der Region Ihres öffentlichen S3-Buckets anlegen:

CREATE SECRET (
    TYPE s3,
    REGION 'my-region'
);
SELECT *
FROM delta_scan('s3://some/public/table/in/my-region');

Aus Azure Blob Storage lesen

Um eine Delta-Tabelle in einem Azure-Blob-Storage-Bucket zu scannen, führen Sie aus:

SELECT *
FROM delta_scan('az://my-container/my-table');

Zur Authentifizierung an Azure Blob Storage werden DuckDB-Secrets unterstützt:

CREATE SECRET (
    TYPE azure,
    PROVIDER credential_chain
);
SELECT *
FROM delta_scan('az://my-container/my-table-with-auth');

Aus Google Cloud Storage lesen

Um eine Delta-Tabelle in einem GCS-Bucket zu scannen, verwenden Sie HMAC-Schlüssel und legen Sie ein Secret an:

CREATE SECRET (
    TYPE gcs,
    KEY_ID '⟨hmac-key-id⟩',
    SECRET '⟨hmac-secret⟩'
);
SELECT *
FROM delta_scan('gs://my-bucket/my-delta-table');

Daten anhängen

Um Zeilen an eine Delta-Tabelle anzuhängen, hängen Sie sie an und verwenden Sie INSERT INTO:

ATTACH 's3://my-bucket/my-delta-table' AS my_table (TYPE delta);
INSERT INTO my_table SELECT * FROM other_table;

Time Travel

Um eine bestimmte Version einer Delta-Tabelle zu lesen, hängen Sie sie an und verwenden Sie die Klausel AT (VERSION => n):

ATTACH 's3://my-bucket/my-delta-table' AS my_table (TYPE delta);
SELECT * FROM my_table AT (VERSION => 5);

Alternativ können Sie eine Version schon beim Anhängen festnageln:

ATTACH 's3://my-bucket/my-delta-table' AS my_table (TYPE delta, VERSION 5);

Checkpointing

Um das Delta-Log einer angehängten Tabelle in eine Checkpoint-Datei zu verdichten:

ATTACH 'path/to/my-delta-table' AS my_table (TYPE delta);
CHECKPOINT my_table;

Idempotente Appends

Die Erweiterung delta stellt eine API für idempotente Appends bereit, die Deltas transaktionale Versionsnummern pro Anwendung nutzt, um Exactly-Once-Semantik zu geben: Ein Append, der mit einer app_id und einer Version markiert ist, wird nur committed, wenn die aktuell für diese app_id aufgezeichnete Version der Tabelle der erwarteten vorherigen Version entspricht. So kann ein Producer einen Batch sicher erneut versuchen, ohne ihn zu verdoppeln.

Markieren Sie den Append innerhalb einer Transaktion mit delta_set_transaction_version(⟨table⟩, ⟨app_id⟩, ⟨new_version⟩, ⟨expected_previous_version⟩):

ATTACH 'path/to/my-delta-table' AS my_table (TYPE delta);

BEGIN TRANSACTION;
CALL delta_set_transaction_version('my_table', 'my_app_id', 1::UBIGINT, NULL::UBIGINT);
INSERT INTO my_table VALUES (1);
COMMIT;

Beim COMMIT wird die Version per Compare-and-Swap verglichen: Hat ein anderer Prozess die Version für my_app_id inzwischen erhöht, schlägt der Commit fehl. Ein Abbruch der Transaktion lässt die Version unverändert. Lesen Sie die aktuelle Version mit delta_get_transaction_version(⟨table⟩, ⟨app_id⟩); die Funktion gibt NULL zurück, wenn noch keine Version aufgezeichnet wurde.

Attach-Optionen

Beim Anhängen einer Delta-Tabelle können Sie die folgenden Optionen an ATTACH übergeben:

Option	Type	Default	Description
VERSION	UBIGINT	latest	Die angehängte Tabelle auf eine bestimmte Tabellenversion festnageln.
PIN_SNAPSHOT	BOOLEAN	false	Den Tabellen-Snapshot einmal beim Anhängen auflösen und wiederverwenden, statt bei jeder Abfrage die neueste Version neu aufzulösen.
PUSHDOWN_PARTITION_INFO	BOOLEAN	true	Partitionsinformationen nach unten durchreichen, sodass ganze Dateien anhand von Partitionswerten übersprungen werden können.
PUSHDOWN_FILTERS	VARCHAR	all	Filter-Pushdown-Modus zum Überspringen von Dateien. Einer von none, all, constant_only, dynamic_only.

ATTACH 's3://my-bucket/my-delta-table' AS my_table (
    TYPE delta,
    PIN_SNAPSHOT true,
    PUSHDOWN_FILTERS 'constant_only'
);

Gescannte Dateien prüfen

delta_list_files gibt die Datendateien zurück, die ein Scan für eine Tabelle lesen würde, zusammen mit ihrer Kardinalität, den Partitionswerten und ob sie Deletion Vectors tragen. Das hilft, die Wirkung von Data Skipping zu verstehen:

SELECT * FROM delta_list_files('file:///some/path/on/local/machine');

Column	Type	Description
data_file	VARCHAR	Pfad zur Parquet-Datendatei.
cardinality	UBIGINT	Anzahl der Zeilen in der Datei.
partitions	MAP(VARCHAR, VARCHAR)	Partitionsspaltenwerte der Datei.
have_deletes	BOOLEAN	Ob die Datei einen zugehörigen Deletion Vector hat.

Credential Chains in Delta

DuckDB Delta verwendet delta-kernel-rs und object_store für einige Netzwerkvorgänge. Diese Systeme haben eine andere Reihenfolge (und andere Einschluss-Standards) für Credential Chains. Wenn Ihr System mehrere Credential-Quellen bereithält, z. B. sowohl Service Principal über die Umgebung als auch eine CLI-basierte Option, kann das Laden der Zugangsdaten inkonsistent sein.

Um Mehrdeutigkeiten zu vermeiden, empfehlen wir, in Ihren Produktions-Chain-Secrets genau einen verfügbaren Credential-Typ zu konfigurieren.

Einstellungen

Die Erweiterung delta fügt die folgenden Einstellungen hinzu:

Setting	Type	Default	Description
delta_kernel_logging	BOOLEAN	false	Das interne Logging des Delta Kernel an den DuckDB-Logger weiterleiten. Kann die Leistung beeinträchtigen, auch wenn DuckDB-Logging deaktiviert ist.
delta_scan_explain_files_filtered	BOOLEAN	true	Die gefilterten Dateien zur EXPLAIN-Ausgabe hinzufügen. Kann die Leistung von delta_scan bei Abfragen mit EXPLAIN ANALYZE beeinträchtigen.

Funktionen

Die Erweiterung delta unterstützt:

• multithreaded Scans und das Lesen von Parquet-Metadaten
• Data Skipping / Filter-Pushdown
  • Überspringen von Row Groups in einer Datei (anhand von Parquet-Metadaten)
  • Überspringen ganzer Dateien (anhand von Delta-Partitionsinformationen)
• Projektions-Pushdown
• Scannen von Tabellen mit Deletion Vectors
• alle primitiven Typen
• Structs
• den Typ VARIANT
• Blind Appends (INSERT INTO)
• Cloud Storage (AWS S3, Azure, GCS) mit Secrets

Unterstützte Plattformen

Die Erweiterung delta unterstützt derzeit nur die folgenden Plattformen:

• Linux AMD64 (x86_64 und ARM64): linux_amd64 und linux_arm64
• macOS Intel und Apple Silicon: osx_amd64 und osx_arm64
• Windows AMD64: windows_amd64

Unterstützung für die anderen DuckDB-Plattformen ist in Arbeit.

delta-rs mit DuckDB verwenden

In diesem Beispiel legen wir eine Delta-Tabelle mit dem Python-Binding delta-rs an und lesen sie anschließend mit der DuckDB-Erweiterung delta. Außerdem zeigen wir weitere Leseoperationen mit DuckDB, etwa das Lesen des Change Data Feed über die Arrow-Zero-Copy-Integration. Dieser Vorgang kann bei größeren Daten auch lazy sein, wenn Sie Arrow Datasets verwenden.

Klicken Sie hier, um das vollständige Beispiel zu sehen.

import deltalake as dl
import pyarrow as pa

# Create a delta table and read it with DuckDB Delta extension
dl.write_deltalake(
    "tmp/some_table",
    pa.table({
        "id": [1, 2, 3],
        "value": ["a", "b", "c"]
    })
)
with duckdb.connect() as conn:
    conn.execute("""
        INSTALL delta;
        LOAD delta;
    """)
    conn.sql("""
        SELECT * FROM delta_scan('tmp/some_table')
    """).show()

# Append some data and read the data change feed using the PyArrow integration
dl.write_deltalake(
    "tmp/some_table",
    pa.table({
        "id": [4, 5],
        "value": ["d", "e"]
    }),
    mode="append"
)
table = dl.DeltaTable("tmp/some_table").load_cdf(starting_version=1, ending_version=2)
with duckdb.connect() as conn:
    conn.register("t", table)
    conn.sql("SELECT * FROM t").show()