feat: Dead Letter Queue with replay — rejected rows recovery

[malon64]

Context

Today, rejected rows are written to a CSV file and forgotten. In a production data platform this is a critical operational gap:

• A source fixes their data format → how to reprocess rows rejected last month?
• A Floe validation rule had a bug and is now corrected → how to replay without re-receiving source files?
• A transient infrastructure failure caused rejections → how to retry only the affected rows?

Without a replay mechanism, every rejected row represents permanent data loss. This issue introduces a Dead Letter Queue (DLQ) — a structured, queryable store for rejected rows — and a floe replay command to reinject them into the pipeline.

Design

Rejected rows today vs DLQ

	Today	DLQ
Format	CSV	Delta or Iceberg table
Queryability	None (flat file)	Full SQL via catalog
Metadata	Filename only	Run ID, entity, timestamp, rejection rules
Replay	Manual re-delivery	floe replay command
Retention	Until overwritten	Configurable retention policy

DLQ table schema

The DLQ is a Delta or Iceberg table with the original row columns plus system metadata columns:

__floe_run_id         string     -- Floe run identifier
__floe_entity         string     -- Entity name
__floe_source_file    string     -- Original source file URI
__floe_rejected_at    timestamp  -- Rejection timestamp
__floe_errors         string     -- JSON array of validation errors (existing field)
__floe_replay_status  string     -- null | "pending" | "replayed" | "failed"
__floe_replayed_at    timestamp  -- When the row was successfully replayed
<original columns>               -- All source row columns, unmasked

Config

sink:
  rejected:
    format: delta          # existing: "csv" | new: "delta" | "iceberg"
    storage: s3_out
    path: warehouse/_floe_dlq/orders
    dlq:
      enabled: true
      retention_days: 90   # auto-expire old rejected rows
      catalog: glue_main   # optional catalog registration

The DLQ sink is opt-in — existing CSV rejected sink is unchanged for backward compatibility.

floe replay command

# Replay all pending rows for an entity since a date
floe replay \
  --entity orders \
  --since 2026-04-01 \
  --config floe.yaml

# Replay only rows rejected by a specific rule
floe replay \
  --entity orders \
  --rule cast_error \
  --config floe.yaml

# Dry-run: show what would be replayed without writing
floe replay --entity orders --dry-run --config floe.yaml

Replay pipeline

The replay command reads from the DLQ table, runs the full Floe validation pipeline on the rows, and:

• Rows that now pass → written to the accepted sink (Delta/Iceberg), __floe_replay_status updated to "replayed"
• Rows that still fail → __floe_replay_status updated to "failed", new __floe_errors recorded
• Atomicity: uses Delta/Iceberg transactions — replay is all-or-nothing per batch

DLQ table (filter: replay_status IS NULL)
    ↓
validate (same rules as original run, or updated rules)
    ↓
accepted → write to target sink + update DLQ status = "replayed"
rejected → update DLQ status = "failed" + new error

Implementation plan

Phase 1 — DLQ sink (Delta/Iceberg rejected output)

• Extend RejectedSinkAdapter with Delta/Iceberg implementations
• Add __floe_* metadata columns to rejected rows before writing
• Add dlq config block to SinkTarget

Phase 2 — floe replay CLI command

• New replay subcommand in floe-cli
• Reads DLQ table via scan_delta / scan_iceberg
• Filters on __floe_replay_status IS NULL and optional --since / --rule
• Runs standard validation pipeline on the result
• Updates DLQ rows atomically after replay

Phase 3 — Retention policy

• Background cleanup: delete rows where __floe_rejected_at < NOW() - retention_days
• Implemented as a Delta VACUUM or Iceberg expire_snapshots + row delete

Dependency

Builds naturally on existing Delta/Iceberg write infrastructure. No new external dependencies.

DLQ rows being in a Delta/Iceberg table also means they are immediately visible in Unity Catalog / Glue (via issues #260 / #261) and traceable via OpenLineage (issue #262).

Estimated effort

~3 weeks:

• DLQ Delta/Iceberg sink + metadata columns: 1 week
• floe replay command + pipeline integration: 1 week
• Retention policy + tests: 1 week