Idempotent Transformation

Pipeline produksi jarang berjalan sekali lalu selesai. Jaringan putus setelah data dikirim tetapi sebelum konfirmasi diterima, orchestrator menjalankan ulang job yang gagal, dan sistem messaging menjamin pengiriman at-least-once sehingga pesan yang sama bisa dibaca dua kali. Di dunia seperti itu, transformasi yang menambah data setiap kali dijalankan adalah sumber duplikasi yang pasti muncul. Idempotent transformation dirancang agar eksekusi berkali-kali menghasilkan keadaan akhir yang sama dengan eksekusi sekali — syarat yang membuat retry menjadi strategi pemulihan yang aman.

Dua Semantik Penulisan: Replace vs Append #

Idempotency sebuah transformasi ditentukan oleh semantik penulisannya. Pola replace menulis ulang hasil untuk cakupan tertentu — satu partisi tanggal atau satu kunci natural — sehingga eksekusi kedua menimpa hasil eksekusi pertama. Pola append hanya menambah baris baru tanpa memeriksa isi tabel, sehingga eksekusi kedua melipatgandakan hasil. Perhatikan perbedaan keduanya:

Kapan eksekusi ulang tersebut terjadi? Setidaknya pada tiga skenario lazim: orchestrator menjalankan ulang job yang statusnya gagal padahal sebagian tulisan sudah ter-commit; sistem messaging dengan jaminan at-least-once mengirim ulang pesan yang sama karena acknowledgment hilang; dan backfill yang rentang waktunya tumpang tindih dengan eksekusi reguler. Ketiganya tidak bisa dicegah sepenuhnya — satu-satunya pertahanan adalah transformasi yang memang aman diulang.

Jebakan Pola Append #

Skrip berikut memindahkan pesanan ke tabel agregat harian dengan pola append. Eksekusi pertama tampak benar; masalah baru terlihat saat job dijalankan ulang — persis skenario retry yang lazim terjadi di produksi.

import sqlite3
from pathlib import Path

DB = Path("toko.db")
if DB.exists():
    DB.unlink()

koneksi = sqlite3.connect(DB)
koneksi.executescript("""
CREATE TABLE pesanan (
    id INTEGER PRIMARY KEY,
    tanggal TEXT NOT NULL,
    produk TEXT NOT NULL,
    qty INTEGER NOT NULL,
    harga INTEGER NOT NULL
);
INSERT INTO pesanan VALUES
    (1, '2026-01-10', 'Kemeja Batik', 1, 250000),
    (2, '2026-01-10', 'Kain Songket', 2, 180000),
    (3, '2026-01-11', 'Blouse Katun', 1, 120000),
    (4, '2026-01-11', 'Seluar Linen', 2, 210000);
CREATE TABLE penjualan_harian (
    tanggal TEXT,
    produk TEXT,
    jumlah_pesanan INTEGER,
    total_pendapatan INTEGER
);
""")


def transform_append(koneksi):
    koneksi.execute("""
        INSERT INTO penjualan_harian
        SELECT tanggal, produk, COUNT(*), SUM(qty * harga)
        FROM pesanan
        GROUP BY tanggal, produk
    """)
    koneksi.commit()


def ringkas(koneksi):
    return koneksi.execute(
        """SELECT COUNT(*), COALESCE(SUM(jumlah_pesanan), 0),
                  COALESCE(SUM(total_pendapatan), 0)
           FROM penjualan_harian"""
    ).fetchone()


print("=== RUN 1 (pola append) ===")
transform_append(koneksi)
print("  (baris, jumlah pesanan, total pendapatan):", ringkas(koneksi))

print("=== RUN 2: job dijalankan ulang ===")
transform_append(koneksi)
print("  (baris, jumlah pesanan, total pendapatan):", ringkas(koneksi))

koneksi.close()

Output yang Dihasilkan:

=== RUN 1 (pola append) ===
  (baris, jumlah pesanan, total pendapatan): (4, 4, 1150000)
=== RUN 2: job dijalankan ulang ===
  (baris, jumlah pesanan, total pendapatan): (8, 8, 2300000)

Eksekusi kedua menduplikasi seluruh hasil: 4 baris menjadi 8, total pendapatan Rp1150000 menjadi Rp2300000. Tabel penjualan_harian tidak memiliki kunci primer, sehingga INSERT INTO ... SELECT menambah empat baris baru tanpa perlawanan dari database. Laporan keuangan yang dibaca setelah retry tersebut akan melaporkan pendapatan dua kali lipat.

Solusi: Full Rebuild per Partisi #

Versi idempotent membatasi cakupan tulis pada satu partisi tanggal: hapus partisi tersebut, lalu isi ulang dari sumber, keduanya dalam satu transaksi. Karena hasil akhir partisi ditentukan penuh oleh isi sumber — bukan oleh berapa kali fungsi dipanggil — eksekusi ulang tidak mengubah apa pun.

import sqlite3
from pathlib import Path

DB = Path("toko.db")
if DB.exists():
    DB.unlink()

koneksi = sqlite3.connect(DB)
koneksi.executescript("""
CREATE TABLE pesanan (
    id INTEGER PRIMARY KEY,
    tanggal TEXT NOT NULL,
    produk TEXT NOT NULL,
    qty INTEGER NOT NULL,
    harga INTEGER NOT NULL
);
INSERT INTO pesanan VALUES
    (1, '2026-01-10', 'Kemeja Batik', 1, 250000),
    (2, '2026-01-10', 'Kain Songket', 2, 180000),
    (3, '2026-01-11', 'Blouse Katun', 1, 120000),
    (4, '2026-01-11', 'Seluar Linen', 2, 210000);
CREATE TABLE penjualan_harian (
    tanggal TEXT PRIMARY KEY,
    jumlah_pesanan INTEGER NOT NULL,
    total_pendapatan INTEGER NOT NULL
);
""")


def transform_partisi(koneksi, tanggal):
    koneksi.execute("DELETE FROM penjualan_harian WHERE tanggal = ?", (tanggal,))
    koneksi.execute(
        """INSERT INTO penjualan_harian (tanggal, jumlah_pesanan, total_pendapatan)
           SELECT tanggal, COUNT(*), SUM(qty * harga)
           FROM pesanan
           WHERE tanggal = ?
           GROUP BY tanggal""",
        (tanggal,),
    )
    koneksi.commit()


def dump(koneksi):
    return koneksi.execute(
        """SELECT tanggal, jumlah_pesanan, total_pendapatan
           FROM penjualan_harian ORDER BY tanggal"""
    ).fetchall()


partisi = ["2026-01-10", "2026-01-11"]

print("=== RUN 1: rebuild per partisi ===")
for tanggal in partisi:
    transform_partisi(koneksi, tanggal)
dump_1 = dump(koneksi)
for baris in dump_1:
    print(" ", baris)

print("=== RUN 2: seluruh job dijalankan ulang ===")
for tanggal in partisi:
    transform_partisi(koneksi, tanggal)
dump_2 = dump(koneksi)

print("  hasil run 1 == hasil run 2:", dump_1 == dump_2)
print("  isi tabel akhir:")
for baris in dump_2:
    print(" ", baris)

koneksi.close()

Output yang Dihasilkan:

=== RUN 1: rebuild per partisi ===
  ('2026-01-10', 2, 610000)
  ('2026-01-11', 2, 540000)
=== RUN 2: seluruh job dijalankan ulang ===
  hasil run 1 == hasil run 2: True
  isi tabel akhir:
  ('2026-01-10', 2, 610000)
  ('2026-01-11', 2, 540000)

Mari kita telaah baris-baris kode di atas:

1. Baris 30-31 (transform_partisi, penghapusan): DELETE menghapus baris partisi tanggal tertentu lebih dulu, sehingga sisa eksekusi sebelumnya tidak pernah menumpuk.
2. Baris 32-40 (penulisan ulang): INSERT ... SELECT mengisi ulang partisi dari sumber dengan filter WHERE tanggal = ?, lalu commit() mengikat hapus-dan-isi menjadi satu transaksi atomik — pembaca lain tidak pernah melihat partisi setengah kosong.
3. Baris 50-62 (pembuktian): seluruh job dijalankan dua kali, lalu kedua dump tabel dibandingkan dengan ==; hasil True membuktikan tabel akhir tidak terpengaruh jumlah eksekusi.

Perhatikan pula kolom tanggal yang kini menjadi PRIMARY KEY di tabel agregat — batasan ini menjadi jaring pengaman terakhir: kunci duplikat akan ditolak database alih-alih lolos diam-diam.

Dua catatan melengkapi pola tersebut. Pertama, idempotency bergantung pada determinisme terhadap sumber: transform_partisi membaca ulang seluruh baris partisi dari pesanan, sehingga hasil akhir ditentukan isi sumber — bukan riwayat eksekusi. Kedua, untuk tabel kecil yang tidak dipartisi, varian full-table (DELETE tanpa filter lalu INSERT penuh) memberikan jaminan yang sama dengan cakupan seluruh tabel; biayanya sebanding ukuran tabel, sehingga pola per-partisi tetap pilihan utama untuk data besar.

Partisi yang aman dibangun ulang adalah modal penting untuk masalah berikutnya: data pesanan yang datang setelah partisi tanggal selesai dihitung. Chapter Late Arriving Data menggunakan pola rebuild yang sama untuk memperbaiki agregat hari-hari yang sudah terlanjur diterbitkan.