Federated Query

Chapter Trino memperlihatkan federasi query di skala klaster; sekarang kita mempraktikkannya di meja kita sendiri. Kondisinya khas TokoKita: data pelanggan hidup di database OLTP SQLite, pesanan terakumulasi sebagai Parquet di penyimpanan analitik, dan tim katalog menyimpan referensi produk dalam CSV. Laporan lintas ketiganya bisa diurus dua cara: memindahkan data (pipeline ETL menyalin semuanya ke satu tempat) atau memindahkan query (engine membaca tiap sumber di tempatnya). DuckDB mampu berperan sebagai federated query engine lokal untuk skenario kedua — dan seluruh contoh chapter ini dieksekusi nyata, termasuk ATTACH database SQLite yang berjalan tanpa internet di mesin penulisan.

Pindahkan Data atau Pindahkan Query #

Federasi bukan pengganti ETL: laporan harian yang di-query berulang kali lebih murah lewat pipeline yang menyalin sekali. Federasi bersinar untuk eksplorasi dan data yang tidak layak disalin. Dalam praktik dua pendekatan ini hidup berdampingan — layer federasi menjawab pertanyaan mendesak hari ini, sementara pipeline ETL membangun tabel analitik untuk beban rutin mingguan.

Perlu dicatat pula bahwa federasi di sini berbeda dari query lintas katalog Trino di chapter sebelumnya. Trino memfederasikan di level klaster: banyak worker membaca banyak sumber secara paralel. DuckDB memfederasikan di level proses: satu mesin membuka beberapa sumber sekaligus. Pola permintaannya sama — data tinggal di tempat, query yang berjalan — hanya skalanya yang berbeda.

Demo: Satu Query, Tiga Sumber #

Skrip berikut menyiapkan tiga sumber dalam tiga format berbeda, lalu memasang database SQLite lewat ATTACH dan menjahit semuanya dalam satu SELECT:

"""Ch9: Federated query — satu query SQL atas tiga sumber berbeda.

Sumber: SQLite (snapshot OLTP: data pelanggan), Parquet (warehouse: pesanan),
CSV (referensi produk dari tim lain). DuckDB menjadi federated query engine
lokal: ketiganya di-join dalam SATU query tanpa memindahkan data ke tabel
tunggal lebih dulu.
"""
import os
import sqlite3

import duckdb

SQLITE_DB = "oltp_fed.db"
PARQUET = "pesanan_analitik.parquet"
CSV = "produk_lengkap.csv"

for f in (SQLITE_DB, PARQUET, CSV):
    if os.path.exists(f):
        os.remove(f)

# --- sumber 1: SQLite (sistem OLTP) ---
oltp = sqlite3.connect(SQLITE_DB)
oltp.execute("""
    CREATE TABLE pelanggan (
        id INTEGER PRIMARY KEY, nama TEXT, kota TEXT, segmen TEXT,
        email TEXT, telp TEXT, alamat TEXT, created_at TEXT
    )
""")
oltp.executemany(
    "INSERT INTO pelanggan VALUES (?,?,?,?,?,?,?,?)",
    [
        (1, "Budi Santoso", "Jakarta", "reguler", "[email protected]",
         "0811-1111-111", "Jl. Merdeka 1", "2024-01-05"),
        (2, "Siti Rahayu", "Bandung", "vip", "[email protected]",
         "0812-2222-222", "Jl. Braga 2", "2024-01-12"),
        (3, "Dewi Lestari", "Surabaya", "reguler", "[email protected]",
         "0813-3333-333", "Jl. Pemuda 3", "2024-02-01"),
    ],
)
oltp.commit()
oltp.close()

# --- sumber 2: Parquet (warehouse analitik) ---
con = duckdb.connect()
con.execute(f"""
    COPY (
        SELECT * FROM (VALUES
            (1, 101, 1, 12500000.0),
            (2, 102, 1,  3500000.0),
            (3, 101, 2, 12500000.0),
            (4, 103, 1,   750000.0),
            (5, 102, 2,  3500000.0),
            (6, 103, 2,   180000.0),
            (7, 101, 3,   950000.0)
        ) t(pesanan_id, produk_id, pelanggan_id, nilai)
    ) TO '{PARQUET}' (FORMAT parquet)
""")

# --- sumber 3: CSV (referensi dari tim lain) ---
with open(CSV, "w", encoding="utf-8") as berkas:
    berkas.write(
        "produk_id,nama_produk,kategori\n"
        "101,Laptop Pro 14,elektronik\n"
        "102,Smartphone X1,elektronik\n"
        "103,Mouse Ergonomis,aksesoris\n"
    )

# --- federasi: ATTACH sqlite, lalu join tiga sumber dalam satu query ---
con.execute(f"ATTACH '{SQLITE_DB}' AS oltp (TYPE sqlite)")
print("== SHOW ALL TABLES (setelah ATTACH) ==")
for baris in con.execute("SHOW ALL TABLES").fetchall():
    print(f"  database={baris[0]}, tabel={baris[2]}, kolom={baris[3][:4]} ...")

print()
print("== bukti pushdown: hanya 3 dari 8 kolom SQLite yang dibaca ==")
plan = con.execute("""
    EXPLAIN SELECT nama FROM oltp.pelanggan WHERE segmen = 'vip'
""").fetchall()[0][1]
lines = [l.replace("│", "").strip() for l in plan.splitlines()]
mulai = next(i for i, l in enumerate(lines) if "SQLITE_SCAN" in l)
for l in lines[mulai:mulai + 14]:
    if l:
        print(l)

print()
print("== query federasi: pesanan (Parquet) x pelanggan (SQLite) x produk (CSV) ==")
for baris in con.execute("""
    SELECT p.kota, pr.kategori,
           count(*)               AS jumlah_pesanan,
           round(sum(f.nilai), 0) AS total_nilai
    FROM read_parquet('pesanan_analitik.parquet') f
    JOIN oltp.pelanggan           p ON p.id = f.pelanggan_id
    JOIN read_csv_auto('produk_lengkap.csv') pr ON pr.produk_id = f.produk_id
    GROUP BY p.kota, pr.kategori
    ORDER BY p.kota, pr.kategori
""").fetchall():
    print(baris)

Output yang Dihasilkan:

== SHOW ALL TABLES (setelah ATTACH) ==
  database=oltp, tabel=pelanggan, kolom=['id', 'nama', 'kota', 'segmen'] ...

== bukti pushdown: hanya 3 dari 8 kolom SQLite yang dibaca ==
SQLITE_SCAN
────────────────────
Table: pelanggan
File: oltp_fed.db
Projections:
segmen
nama
~2 rows
└───────────────────────────┘

== query federasi: pesanan (Parquet) x pelanggan (SQLite) x produk (CSV) ==
('Bandung', 'aksesoris', 1, Decimal('180000'))
('Bandung', 'elektronik', 2, Decimal('16000000'))
('Jakarta', 'aksesoris', 1, Decimal('750000'))
('Jakarta', 'elektronik', 2, Decimal('16000000'))
('Surabaya', 'elektronik', 1, Decimal('950000'))

Mari kita telaah bagian-bagian penting skrip di atas:

1. Baris 21-56 (tiga sumber): SQLite dibangun lewat pustaka sqlite3 standar dengan delapan kolom — meniru database OLTP yang kolomnya banyak dan penuh detail pribadi. Parquet dan CSV meniru penyimpanan analitik dan berkas tim lain.
2. Baris 68-74 (ATTACH ... TYPE sqlite): database SQLite dipasang sebagai schema oltp di koneksi DuckDB; SHOW ALL TABLES membuktikan tabel oltp.pelanggan kini terlihat dari satu koneksi yang sama.
3. Baris 76-86 (bukti pushdown): query hanya meminta kolom nama dengan filter segmen = 'vip'; rencana eksekusi menunjukkan SQLITE_SCAN dengan Projections: segmen, nama — DuckDB hanya memindahkan tiga kolom dari delapan yang ada di berkas SQLite.
4. Baris 88-99 (query federasi): satu SELECT merujuk read_parquet(...), oltp.pelanggan, dan read_csv_auto(...) sekaligus; hasilnya laporan nilai pesanan per kota per kategori tanpa satu pun proses salin-menyalin.

Membaca Hasil dan Batasnya #

Hasil federasi menunjukkan pola yang masuk akal: Jakarta dan Bandung masing-masing menyumbang dua pesanan elektronik bernilai 16 juta, disusul satu pesanan aksesoris di tiap kota (750 ribu di Jakarta, 180 ribu di Bandung), dan Surabaya satu pesanan elektronik. Nilai tampil sebagai Decimal karena round() di DuckDB mempertahankan tipe numerik presisi — perilaku yang sama dengan yang kita temui di chapter DuckDB.

Perhatikan pula bagian ~2 rows pada kartu SQLITE_SCAN: itu perkiraan kardinalitas dari statistik tabel. Optimizer memakai perkiraan ini untuk memilih urutan join dan metode agregasi, persis mekanisme cost-based yang dibedah di chapter Query Optimization — federasi tidak mengubah cara optimizer bekerja, hanya menambah jenis sumber yang bisa discan. Kolom kolom=['id', 'nama', 'kota', 'segmen'] ... pada SHOW ALL TABLES membuktikan schema SQLite terbawa utuh saat di-ATTACH, sehingga tipe data dan kunci primer tidak perlu didefinisikan ulang.

Dari sini peta lengkap sudah di tangan kita: columnar storage, vectorized execution, tiga engine dengan tiga arsitektur, optimizer, eksekusi terdistribusi, dan federasi. Satu pertanyaan tersisa untuk menutup section: dari sekian banyak engine ini, mana yang dipilih untuk pekerjaan mana? Chapter penutup, Pemilihan Query Engine, merangkum keputusan itu menjadi tabel dan diagram yang bisa dipakai berulang.