Data Ingestion Patterns — Full Load, Incremental, CDC

Full Load — Read Everything, Every Time

BEGIN;
TRUNCATE TABLE silver.store_master;
INSERT INTO silver.store_master
SELECT store_id, store_name, city, region, is_active, manager_id FROM source.stores;
COMMIT;
-- other queries see either all-old or all-new, never empty (MVCC) — but only
-- while this single transaction is what they're reading against

CREATE TABLE silver.store_master_new AS
SELECT store_id, store_name, city, region, is_active, manager_id FROM source.stores;

BEGIN;
ALTER TABLE silver.store_master RENAME TO store_master_old;
ALTER TABLE silver.store_master_new RENAME TO store_master;
COMMIT;

DROP TABLE silver.store_master_old;
-- during load: store_master_old serves queries. after rename: store_master (new) does.
-- zero seconds where the table is empty or has partial data

def full_load_with_swap(source_conn, dest_conn, table: str) -> int:
    df = pd.read_sql(f"SELECT * FROM {table}", source_conn)
    staging = f"{table}_staging"
    df.to_sql(staging, dest_conn, if_exists='replace', index=False)
    with dest_conn.cursor() as cur:
        cur.execute(f"ALTER TABLE {table} RENAME TO {table}_old")
        cur.execute(f"ALTER TABLE {staging} RENAME TO {table}")
        cur.execute(f"DROP TABLE {table}_old")
    dest_conn.commit()
    return len(df)

>>> full_load_with_swap(source_conn, dest_conn, 'store_master')
40
# 40 stores reloaded, zero downtime — analysts querying store_master mid-swap
# saw either the complete old table or the complete new one

1. TABLE GROWS TOO LARGE
   orders: 500M rows, full load takes 6h, SLA is 6 AM → barely fits.
   Signal to switch: full load duration > 20% of the run interval.

2. SOURCE LOAD DURING EXTRACTION
   A full table scan fills the buffer pool, evicting hot pages —
   the application slows down for 30-60 min afterward.
   Fix: extract from a read replica, never the primary.

3. DESTINATION INCONSISTENCY WINDOW
   TRUNCATE-then-INSERT (Variant A) leaves the table empty mid-transaction
   for any query outside that transaction. Fix: staging swap (Variant B).

4. RELOAD OVERWRITES LATE-ARRIVING CORRECTIONS
   A manual data fix in the destination gets silently overwritten by the
   next full load. Expected behavior — but teams get surprised by it.
   If destination edits must survive: use incremental or CDC instead.

Incremental — Only What Changed

import json, logging
from datetime import datetime, timezone
from pathlib import Path

log = logging.getLogger('incremental_ingestion')
CHECKPOINT_FILE = Path('/data/checkpoints/orders_watermark.json')

def load_watermark() -> datetime:
    if CHECKPOINT_FILE.exists():
        wm = datetime.fromisoformat(json.loads(CHECKPOINT_FILE.read_text())['watermark'])
        log.info('Loaded watermark: %s', wm.isoformat())
        return wm
    default = datetime(2020, 1, 1, tzinfo=timezone.utc)
    log.info('No checkpoint found — starting from %s', default.isoformat())
    return default

def save_watermark(watermark: datetime) -> None:
    tmp = CHECKPOINT_FILE.with_suffix('.tmp')
    tmp.write_text(json.dumps({'watermark': watermark.isoformat()}))
    tmp.rename(CHECKPOINT_FILE)   # atomic on POSIX

def extract_changed_orders(conn, since: datetime, until: datetime) -> pd.DataFrame:
    """since is exclusive, until is inclusive — no boundary row re-processed or skipped."""
    df = pd.read_sql("""
        SELECT order_id, customer_id, store_id, order_amount, status, created_at, updated_at
        FROM orders WHERE updated_at > %s AND updated_at <= %s ORDER BY updated_at ASC
    """, conn, params=(since, until))
    log.info('Extracted %d rows (updated %s to %s)', len(df), since.isoformat(), until.isoformat())
    return df

def upsert_orders(df: pd.DataFrame, dest_conn) -> int:
    if df.empty:
        return 0
    with dest_conn.cursor() as cur:
        for _, row in df.iterrows():
            cur.execute("""
                INSERT INTO silver.orders (order_id, customer_id, store_id, order_amount, status, created_at, updated_at, ingested_at)
                VALUES (%s, %s, %s, %s, %s, %s, %s, NOW())
                ON CONFLICT (order_id) DO UPDATE SET status = EXCLUDED.status, updated_at = EXCLUDED.updated_at, ingested_at = NOW()
                WHERE silver.orders.updated_at < EXCLUDED.updated_at
            """, (row.order_id, row.customer_id, row.store_id, row.order_amount, row.status, row.created_at, row.updated_at))
    dest_conn.commit()
    return len(df)

def run_incremental(source_conn, dest_conn) -> dict:
    since = load_watermark()
    until = pd.read_sql("SELECT NOW() AT TIME ZONE 'UTC'", source_conn).iloc[0, 0].to_pydatetime()

    df = extract_changed_orders(source_conn, since, until)
    if df.empty:
        return {'rows_processed': 0, 'new_watermark': since.isoformat()}

    written = upsert_orders(df, dest_conn)
    save_watermark(until)   # only AFTER the write succeeded
    return {'rows_processed': written, 'new_watermark': until.isoformat()}

INFO Loaded watermark: 2026-03-17T05:45:00+00:00
INFO Extracted 1,842 rows (updated 2026-03-17T05:45:00+00:00 to 2026-03-17T06:00:00+00:00)
>>> run_incremental(source_conn, dest_conn)
{'rows_processed': 1842, 'new_watermark': '2026-03-17T06:00:00+00:00'}

# PITFALL 1: HARD DELETES ARE INVISIBLE
# A deleted row produces no result from 'WHERE updated_at > checkpoint' —
# there's nothing left to return. Destination silently diverges from source.
# Fix A: use CDC (captures DELETE explicitly)
# Fix B: soft-delete column (deleted_at / is_deleted) — updates updated_at, so it's seen
# Fix C: periodic full-load reconciliation (weekly) if deletes are rare

# PITFALL 2: NO updated_at COLUMN
# Fix A: use max(primary_key) as watermark — ONLY safe if rows are insert-only
# Fix B: CDC (doesn't depend on an application-maintained timestamp)
# Fix C: full load, if the table is small enough

# PITFALL 3: CLOCK SKEW BETWEEN SOURCE AND PIPELINE SERVER
# pipeline clock 06:00:00, source clock 06:00:02 (2s ahead) —
# a row inserted at 06:00:01 on the source's clock looks like "the future" and gets excluded
# Fix: always use the SOURCE database's NOW() as the upper bound, never the pipeline server's

# PITFALL 4: LATE-ARRIVING UPDATES
# row.updated_at = 11:58:00, but it doesn't actually reach the source table until
# 12:03:00 (a delayed application retry) — by then the checkpoint has already moved past 12:00:00
# Fix: extend the LOWER bound back by a safe margin (e.g. 30 min) and rely on
# upsert to make the resulting re-processed overlap rows harmless

Change Data Capture — The Complete Picture

-- Application writes:
UPDATE orders SET status = 'delivered' WHERE order_id = 9284751;

-- PostgreSQL WAL records (simplified):
-- {LSN: 0/1A3F2B8, op: UPDATE, table: orders,
--  old: {order_id: 9284751, status: 'confirmed'}, new: {..., status: 'delivered'}}

-- Debezium decodes the WAL and publishes to Kafka topic 'prod.public.orders':
{
  "before": {"order_id": 9284751, "status": "confirmed"},
  "after":  {"order_id": 9284751, "status": "delivered"},
  "op": "u",   // c=create, u=update, d=delete, r=read/snapshot
  "source": {"lsn": 28437128, "txId": 847291}
}

-- A DELETE looks like: {"before": {...}, "after": null, "op": "d"}

CDC captures everything:
✓ INSERT → op: "c"   ✓ UPDATE → op: "u" (before+after)   ✓ DELETE → op: "d" (before image)
✓ Schema changes (with schema registry)   ✓ Transaction boundaries (atomic groups)

# postgresql.conf — must restart PostgreSQL after this
wal_level = logical
max_replication_slots = 10
max_wal_senders = 10

CREATE USER debezium_user REPLICATION LOGIN PASSWORD 'strong_password';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium_user;

SELECT pg_create_logical_replication_slot('debezium_slot', 'pgoutput');

// POST http://kafka-connect:8083/connectors
{
  "name": "freshcart-orders-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres-primary",
    "database.dbname": "freshcart_prod",
    "table.include.list": "public.orders,public.customers,public.payments",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
    "snapshot.mode": "initial",
    "topic.prefix": "freshcart.cdc"
  }
}
// creates Kafka topics: freshcart.cdc.public.{orders,customers,payments}

consumer = Consumer({'bootstrap.servers': 'kafka:9092', 'group.id': 'freshcart-cdc-pipeline',
                      'enable.auto.commit': False})   # manual commit — at-least-once
consumer.subscribe(['freshcart.cdc.public.orders'])

while True:
    msg = consumer.poll(timeout=1.0)
    if msg is None or msg.error():
        continue
    event = json.loads(msg.value())
    if event['op'] in ('c', 'u', 'r'):
        upsert_to_silver(event['after'])
    elif event['op'] == 'd':
        soft_delete_in_silver(event['before']['order_id'])
    consumer.commit()   # only after the write succeeds

# snapshot.mode options: initial (default, full read then stream) | never (stream only,
# misses everything before connector start) | schema_only (schema only, no data) | always (dev only)

# PRACTICAL BOOTSTRAP for a 500M-row table:
# 1. pg_dump → S3 (parallel, 1-2 hours)
# 2. Bulk load the S3 dump into the destination
# 3. Start Debezium with snapshot.mode=schema_only, from the WAL LSN at dump time
# 4. Apply WAL events from that LSN forward — catches up during/after the bulk load
# → reduces bootstrap from 8 hours to ~2 hours

-- A stuck consumer means WAL accumulates on the SOURCE forever until it's read.
-- Monitor:
SELECT slot_name, pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) AS lag_bytes
FROM pg_replication_slots;
-- Alert when lag_bytes > 10 GB. If the consumer is unrecoverable: DROP the slot
-- (accepting data loss) rather than let the source disk fill and crash the database.

CDC LATENCY (Debezium + Kafka + consumer), end to end:
Source write → Kafka event:        50-200ms
Kafka event → consumer processing:  10-100ms
Consumer → destination write:       50-500ms
Total: 200ms - 1s — fine for near-real-time dashboards, NOT for synchronous app flow

1. Row count and growth rate?
   < 1M rows, grows slowly  → Full Load is viable
   > 1M rows or grows fast  → Incremental or CDC required

2. Reliable updated_at column?
   Yes → Incremental is viable, continue to Q3
   No, insert-only → use created_at or auto-increment PK
   No, has updates/deletes → CDC or Full Load only

3. Do hard deletes matter for the destination?
   No (rare, or soft-deleted) → Incremental is sufficient
   Yes → CDC required — incremental cannot see hard deletes

4. Latency requirement?
   > 15 min acceptable → Incremental on a schedule
   < 15 min           → CDC, or 5-minute micro-batch incremental
   < 1 min             → CDC only

product_categories    (500 rows, rarely changes)             → Full Load
orders                (500M rows, updated frequently)        → Incremental
customers             (10M rows, hard deletes for GDPR)      → CDC
payment_transactions  (1B rows, financial accuracy critical) → CDC
delivery_events       (append-only, no deletes)              → Incremental (created_at)
inventory             (updates + deletes frequently)         → CDC

FULL LOAD (nightly, 5 min total):
  reference.store_master, reference.product_categories, reference.city_tier_mapping

INCREMENTAL (every 15 min, updated_at watermark):
  orders (500M rows), delivery_events (2B rows, created_at), customer_reviews (created_at)

CDC (continuous, sub-second latency):
  customers (GDPR deletes), payments (financial), merchant_accounts, inventory_live

TOTAL INFRASTRUCTURE:
  Full load: 2 cron jobs. Incremental: 3 Airflow tasks.
  CDC: 1 Debezium connector, 4 Kafka topics, 1 consumer group.
  → Most data volume is incremental. Most operational complexity is CDC — for only 4 tables.

The customer success team reports that cancelled orders are still showing up as “active” on the dashboard. Orders customers cancelled yesterday appear as “placed” in the Silver layer.

SELECT order_id, status, updated_at FROM production.orders WHERE order_id = 9284751;
-- {status: 'cancelled', updated_at: '2026-03-17 14:32:00'}
SELECT order_id, status, updated_at FROM silver.orders WHERE order_id = 9284751;
-- {status: 'placed', updated_at: '2026-03-17 08:14:00'}   ← 6-hour gap

-- checkpoint file: {"watermark": "2026-03-17T08:00:00+00:00"} — hasn't moved in 6 hours

$ tail -100 /var/log/airflow/orders_incremental_20260317.log | grep ERROR
08:15:42 ERROR Connection to source database timed out
08:15:42 ERROR Pipeline failed — checkpoint NOT advanced
14:00:00 INFO  Database connection restored

14:00:02 INFO Loaded watermark: 2026-03-17T08:00:00+00:00
14:00:03 INFO Extracted 284,721 rows (updated 08:00 to 14:00)
14:00:47 INFO 284,721 rows upserted successfully
14:00:47 INFO Saved watermark: 2026-03-17T14:00:00+00:00

SELECT status FROM silver.orders WHERE order_id = 9284751;
-- 'cancelled' ← correct now

This was not a bug in the ingestion pattern — it was a 6-hour source database outage. The incremental pattern with checkpointing recovered perfectly: it resumed exactly where it stopped, processed the backlog, and Silver was correct within minutes of the database recovering. A full load would have needed a full 6-hour table scan to recover the same ground; CDC would have needed Kafka retention to have covered the whole 6-hour gap. Incremental just needed its next scheduled run.