SFTP, EDI & Flat-File Ingestion: File Landing, Schema Drift, Late & Partial Files

flat-file ingestion is the un-glamorous, load-bearing pipeline that still moves payroll runs, bank settlement files, insurance claims, and partner EDI documents into your warehouse every night — and it is the pipeline senior data engineers under-invest in until the night a half-written file gets loaded and the finance close is wrong by a day. A partner drops a file onto an SFTP server, your job wakes up, and everything that can go wrong upstream of a SELECT now lives on your plate: a file that is still uploading when you grab it, a CSV whose customer name contains an un-escaped comma, a fixed-width layout that shifted by two bytes, a header that gained a column overnight, a "daily" file that shows up two days late, and a truncated transfer that looks complete until you count the rows. None of these are query problems; they are ingestion problems, and they are where real pipelines break.

This guide is the senior-data-engineering walkthrough for building a flat-file intake you can trust. It works through the four things that decide whether a dropped file becomes clean rows or a 3 a.m. page: how the file lands (the file landing zone, atomic rename, and idempotent SFTP pickup), how it parses (CSV parsing quoting hazards, fixed-width byte offsets, and EDI X12/EDIFACT envelopes), how you survive schema drift (fingerprinting the header and routing added, reordered, and type-changed columns to evolve, quarantine, or fail), and how you handle late-arriving files and partial files (trailer-count and checksum completeness gates, watermark reopen windows, and dedupe by file hash). Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the ETL practice library →, sharpen your parsers on the CSV parsing practice library →, and harden your input checks on the data-validation practice library →.

---

On this page

• Why flat-file ingestion is where pipelines quietly break
• File landing zones and SFTP transport
• Parsing CSV, fixed-width, and EDI formats
• Schema drift detection and handling
• Late-arriving and partial files
• Cheat sheet — flat-file ingestion recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why flat-file ingestion is where pipelines quietly break

Four axes, four different failure modes — the file is untrusted until you prove otherwise

The one-sentence invariant: flat-file ingestion is the discipline of turning an untrusted file that a partner dropped — over which you have zero schema guarantees, zero write-atomicity guarantees, and zero delivery-time guarantees — into rows you can safely load, and every design decision reduces to four axes: how the file lands (transport and atomicity), how it parses (format and encoding), whether its shape matches what you expected (schema stability), and whether it is actually complete and on time (completeness and timing). Unlike an API you call or a stream you subscribe to, a flat file is a one-way, fire-and-forget artifact: the sender has already gone home. There is no retry handshake, no schema negotiation, no backpressure. If the file is wrong, half-written, or late, you absorb it — which is exactly why interviewers use flat-file ingestion to separate engineers who have run production intake from those who have only written pandas.read_csv against a clean sample.

The four axes interviewers actually probe.

• Transport and landing. How does the file arrive, and how do you know it is fully written before you touch it? SFTP is still the dominant B2B transport in 2026 (banks, payroll processors, healthcare clearinghouses, EDI VANs). The hazard is reading a file mid-upload. The senior answer names atomic landing — the sender writes file.csv.part and renames to file.csv only when complete, or drops a separate file.ok / manifest trigger — and an idempotent pickup so re-running the job never double-loads.
• Format and parsing. CSV, TSV, pipe-delimited, fixed-width, JSON-lines, and EDI (X12 / EDIFACT) each have a distinct failure surface. "CSV" alone hides quoting rules, embedded delimiters and newlines, encoding and byte-order marks, and ragged rows. Fixed-width has no delimiter to lean on — a one-byte shift silently corrupts every downstream column. EDI is a nested-envelope grammar, not a table. Naming the specific hazard per format is the signal.
• Schema stability. The partner controls the schema and will change it without telling you — add a column, reorder two, rename amt to amount, or start sending a string where you expected an integer. This is schema drift, and the design question is not "will it happen" (it will) but "what does your pipeline do when it does": fail loudly, quarantine for review, or evolve automatically. The wrong default silently loads garbage.
• Completeness and timing. Is the file whole, and is it on time? A truncated transfer, a still-uploading file, or a partner who sent only the first of three expected files are all incompleteness. A daily file that lands two days late is a timing problem that must reopen a window without double-counting. The senior answer reaches for control/trailer records, manifest row counts, and checksums to prove completeness before loading, and a file-hash ledger to make reprocessing idempotent.

The 2026 reality — streams get the headlines, flat files still move the money.

• SFTP + flat files remain the B2B lingua franca. Payroll, ACH/wire settlement, card networks, EDI trading partners, insurance, logistics, and government feeds overwhelmingly exchange fixed-width and EDI files over SFTP. These systems predate REST and will outlive it; "just ask them for an API" is not on the table for a Fortune 500 partner integration.
• Managed connectors help but do not remove the problem. Fivetran, Airbyte, AWS Transfer Family, and Azure Data Factory can move and parse files, but the semantics — atomic landing contract, drift policy, completeness proof, late-window handling — are still yours to design. The tool grabs the bytes; you own whether loading them is safe.
• The lakehouse pattern normalises "land raw, then process." The modern default is an immutable raw landing zone (object storage) partitioned by arrival date, with parsing and validation as downstream steps — so a bad file is quarantined, not silently merged. Auto-loading tools (Databricks Auto Loader, Snowpipe, COPY INTO) still need you to define the drift and completeness behaviour.
• Contracts are shifting left. Mature teams ship a data contract per feed — expected columns, types, delimiter, encoding, row-count tolerance, delivery SLA — as versioned data, and the pipeline diffs each arrival against it. This is the flat-file analogue of a schema registry.

What interviewers listen for.

• Do you say "I never read a file until I know it is completely written" and name the atomic-rename / manifest mechanism? — required answer.
• Do you make pickup idempotent — a processed-file ledger keyed on file name and content hash, not just "move it after loading"? — senior signal.
• Do you treat schema drift as a policy (fail / quarantine / evolve) rather than assuming the header never changes? — senior signal.
• Do you prove completeness with a trailer count or checksum instead of trusting that the file that appeared is the whole file? — senior signal.
• Do you describe the raw file as untrusted input and land it immutably before parsing, rather than parsing straight into the warehouse? — required framing.

Worked example — the four-axis file-ingestion checklist

Detailed explanation. The single most useful artifact for a flat-file interview is a four-axis checklist you run against any new feed before writing a line of parsing code. Every senior file-intake design converges on these four questions; having them memorised turns a vague "how would you ingest this?" into a structured answer. Walk through building the checklist for a concrete feed: a payroll processor dropping a daily employee-earnings file.

• The feed. acme_payroll_earnings_YYYYMMDD.csv, dropped to /inbound/acme/ on your SFTP server around 02:00, comma-delimited with a header row and a trailer record.
• Transport. SFTP, key-based auth, PGP-encrypted payload, one file per day.
• Downstream. A raw.payroll_earnings warehouse table feeding the finance close.
• The four questions. Land safely? Parse correctly? Detect drift? Prove completeness and timing?

Question. Build the four-axis intake checklist for the payroll feed and state the concrete control for each axis.

Input.

Axis	Question to answer	Control for this feed
Transport & landing	Is the file fully written before I read it?	wait for .ok trigger; copy to immutable raw by dated key
Format & parsing	What breaks the parser?	RFC-4180 quoting, UTF-8 + BOM strip, PGP decrypt first
Schema stability	Did the columns change?	fingerprint header; compare to contract
Completeness & timing	Is it whole and on time?	trailer row count == data rows; SLA 02:00–06:00

Code.

Flat-file intake checklist (run before writing parser code)
===========================================================

1. TRANSPORT & LANDING
   [ ] How does the file arrive?              -> SFTP /inbound/acme/
   [ ] How do I know it is complete on disk?  -> wait for acme_*.ok trigger
   [ ] Is pickup idempotent?                  -> ledger on (name, sha256)
   [ ] Is raw immutable?                       -> copy to s3://raw/acme/2026/08/18/

2. FORMAT & PARSING
   [ ] Exact format?                          -> CSV, comma, double-quote, CRLF
   [ ] Encoding?                              -> UTF-8, may carry a BOM
   [ ] Quoting / embedded delimiters?         -> RFC-4180, quotes around names
   [ ] Pre-processing?                        -> PGP decrypt before anything else

3. SCHEMA STABILITY
   [ ] Expected columns + types?              -> contract v3 (14 columns)
   [ ] Header present?                        -> yes; fingerprint & diff it
   [ ] Drift policy?                          -> additive=evolve, else quarantine

4. COMPLETENESS & TIMING
   [ ] Completeness proof?                    -> trailer 'T|<rowcount>'
   [ ] Delivery SLA?                          -> land by 06:00, alert if missing
   [ ] Late / partial handling?               -> reopen window; dedupe by hash

Step-by-step explanation.

1. Axis 1 (transport & landing) is answered first because it gates everything else — there is no point parsing a file that is still uploading. For this feed the sender drops a companion acme_payroll_earnings_20260818.ok file only after the data file is fully written, so the trigger's existence is the "fully written" signal. The job copies the raw bytes to an immutable, date-partitioned key so re-runs read the same input.
2. Axis 2 (format & parsing) pins the exact dialect before code is written. "CSV" is under-specified: this feed is comma-delimited, double-quoted per RFC-4180, CRLF line endings, UTF-8 that may carry a byte-order mark, and — crucially — PGP-encrypted, so decryption is step zero of parsing. Writing these down prevents the classic "it worked on the sample, broke in prod" failure.
3. Axis 3 (schema stability) records the expected shape as a contract (14 columns, named and typed at version 3) and a drift policy: an added trailing column is additive and safe to evolve; anything else (drop, reorder, type change) is quarantined for a human. The header is fingerprinted so drift is detected on arrival, not three tables downstream.
4. Axis 4 (completeness & timing) is what turns "a file appeared" into "the file is safe to load." The trailer record T|<rowcount> lets you assert the data-row count matches; the 06:00 SLA turns a missing file into a page instead of a silent gap; and the late/partial plan (reopen the window, dedupe by content hash) means a late or re-sent file never double-loads the close.
5. The order matters: land → parse → drift → complete. Each axis assumes the previous one passed. Presenting the answer in this order signals you understand ingestion as a pipeline of gates, each of which can reject the file, not a single read_csv call.

Output.

Axis	Failure it prevents	If the control fires
Transport & landing	reading a half-written file	wait / retry; never load a partial
Format & parsing	mangled rows, encoding corruption	reject to rejected/; alert owner
Schema stability	silently loading the wrong columns	quarantine; open a contract ticket
Completeness & timing	missing rows, double-load, late close	hold, reopen window, dedupe

Rule of thumb. Never open a parser before you have answered all four axes on paper. The file is untrusted input; land it, prove it is complete, confirm its shape, and only then parse. Skipping an axis is how a "simple CSV load" becomes a finance incident.

Worked example — what interviewers actually probe

Detailed explanation. The senior flat-file interview has a predictable arc: an ambiguous opener ("a vendor is going to drop us a daily file — how do you ingest it?"), then progressive narrowing to test whether you know the failure modes. Candidates who immediately reach for atomicity, idempotency, and completeness score highest; candidates who describe "a cron job that runs read_csv and inserts" score lowest. Walk through the grading rubric.

• Ambiguous opener. "A partner will SFTP us a daily file — design the ingestion." — invites the landing/parse/drift/complete framing.
• Follow-up 1. "What if the job runs while the file is still uploading?" — probes the atomic-landing axis.
• Follow-up 2. "The partner adds a column next quarter — what happens?" — probes schema drift.
• Follow-up 3. "How do you know you got the whole file?" — probes completeness.
• Follow-up 4. "The file is two days late, then they re-send it — now what?" — probes late/partial + idempotency.

Question. Draft a five-minute senior answer that covers all four axes without waiting for the follow-ups.

Input.

Interview signal	Weak answer	Senior answer
Landing	"cron reads the file at 3am"	"wait for the .ok trigger; copy raw to an immutable dated key"
Idempotency	"move the file after loading"	"ledger on (filename, sha256); skip if seen"
Parsing	"pandas read_csv"	"RFC-4180 reader, explicit encoding, reject ragged rows"
Drift	"the schema is fixed"	"fingerprint header; additive=evolve else quarantine"
Completeness	"if the file is there, load it"	"assert trailer count == data rows and checksum matches"

Code.

Senior flat-file ingestion answer template (5 minutes)
======================================================

Minute 1 — land it safely
  "The file is untrusted. I never read it until I know it is fully
   written — either the sender writes .part and renames, or drops a
   separate .ok/manifest trigger. I copy the raw bytes to an immutable,
   date-partitioned landing zone before parsing anything."

Minute 2 — make pickup idempotent
  "Pickup is idempotent: a processed-file ledger keyed on filename AND
   content hash. If I have already loaded this exact file, I skip it.
   Re-running the job or a partner re-send never double-loads."

Minute 3 — parse defensively
  "I pin the exact dialect: delimiter, quote char, encoding, line
   ending. I use an RFC-4180-compliant reader, strip a BOM if present,
   and reject ragged rows to a rejected/ prefix instead of silently
   dropping fields. For fixed-width I slice by byte offset; for EDI I
   parse the ISA/GS/ST envelope."

Minute 4 — handle schema drift by policy
  "The partner owns the schema and will change it. I fingerprint the
   header on arrival and diff it against a versioned contract. Additive
   changes evolve automatically; drops, reorders, and type changes are
   quarantined for review, not loaded."

Minute 5 — prove completeness, handle late/partial
  "Before loading I prove completeness: trailer record count equals
   data rows, checksum matches the manifest, byte size above a floor.
   A late file reopens its watermark window; because pickup is
   idempotent by hash, a re-send is deduped, never double-counted."

Step-by-step explanation.

1. Minute 1 is the framing that scores. Leading with "the file is untrusted, I never read it until it is fully written" signals you have been burned by a mid-upload read and designed against it. Weak candidates start with the parser and never mention atomicity.
2. Minute 2 pre-empts the idempotency probe. Saying "ledger on filename and content hash" is the detail that matters — a re-sent file with the same name but different content must be treated as new, and a re-sent identical file must be skipped. "Move the file after loading" is fragile because a crash between load and move double-loads.
3. Minute 3 shows parsing maturity: pinning the dialect, using an RFC-4180 reader, handling BOM and encoding, and rejecting rather than silently mangling malformed rows. Naming fixed-width byte offsets and EDI envelopes in the same breath shows range across formats.
4. Minute 4 treats drift as an inevitability with a policy, not an accident. The evolve/quarantine/fail split, keyed on the kind of change, is the senior distinction — "the schema is fixed" is the answer that ships a silent data-corruption bug the first quarter the partner adds a column.
5. Minute 5 closes on completeness and timing, the axes juniors forget entirely. Trailer counts and checksums prove the file is whole; the idempotent-by-hash ledger is what makes late files and re-sends safe. Covering this unprompted is the difference between a task-runner and a pipeline owner.

Output.

Grading criterion	Weak score	Senior score
Names atomic landing in minute 1	rare	mandatory
Idempotent pickup by content hash	rare	required
Defensive, dialect-pinned parsing	occasional	expected
Drift as evolve/quarantine/fail policy	rare	senior signal
Completeness proof (trailer/checksum)	rare	senior signal

Rule of thumb. The senior flat-file answer is a five-minute monologue: land it immutably, make pickup idempotent by hash, parse defensively, handle drift by policy, and prove completeness before loading. Rehearse it once; it survives every follow-up.

Worked example — the "is this file safe to load?" decision tree

Detailed explanation. Given a file that has appeared in the landing zone, the senior engineer runs a short decision tree before a single row reaches the warehouse. Codifying the tree makes the answer reproducible: any interviewer can hand you a scenario and you can walk it out loud. Walk through the tree with three files: a clean on-time file, a still-uploading file, and a re-sent duplicate.

• Q1. Is the file fully written (trigger present / stable size)? → no = wait/retry; yes = go to Q2.
• Q2. Have I already processed this exact content (hash in ledger)? → yes = skip (idempotent); no = go to Q3.
• Q3. Does the header fingerprint match the contract (or an allowed evolution)? → no = quarantine; yes = go to Q4.
• Q4. Does completeness hold (trailer count + checksum + size floor)? → no = hold as partial, retry; yes = load.

Question. Walk the decision tree for the three files and record where each one exits.

Input.

File	Q1 written?	Q2 seen before?	Q3 schema ok?	Q4 complete?
clean on-time file	yes	no	yes	yes
still-uploading file	no	—	—	—
re-sent duplicate	yes	yes	—	—

Code.

# Decision-tree helper (illustrative)
def is_safe_to_load(fully_written: bool,
                    already_processed: bool,
                    schema_ok: bool,
                    complete: bool) -> str:
    """Return the ingestion verdict for one landed file."""
    if not fully_written:
        return "WAIT"            # still uploading; retry later
    if already_processed:
        return "SKIP"            # idempotent: exact content seen before
    if not schema_ok:
        return "QUARANTINE"      # drift; needs review
    if not complete:
        return "HOLD_PARTIAL"    # truncated / short; retry or backfill
    return "LOAD"                # all gates passed


print(is_safe_to_load(True,  False, True,  True))   # -> LOAD
print(is_safe_to_load(False, False, False, False))  # -> WAIT
print(is_safe_to_load(True,  True,  False, False))  # -> SKIP

Step-by-step explanation.

1. The clean on-time file passes every gate: it is fully written (Q1), never seen before (Q2), matches the contract (Q3), and its trailer count and checksum verify (Q4). Verdict: LOAD. This is the happy path — and it is the only path that reaches the warehouse.
2. The still-uploading file fails Q1: the .ok trigger is absent (or the file size is still changing between two stats). The tree short-circuits to WAIT without ever touching the bytes. This is the single most important gate — reading here is the mid-upload bug.
3. The re-sent duplicate passes Q1 (it is fully written) but fails Q2: its content hash is already in the processed-file ledger. Verdict: SKIP. This is idempotency in action — a partner who re-sends yesterday's identical file must not double-load it.
4. The gates are ordered by cost and blast radius: cheap-and-catastrophic first (mid-upload read, double-load), then shape (drift), then completeness. A file must clear every gate; failing any one diverts it to a non-loading outcome (WAIT, SKIP, QUARANTINE, HOLD_PARTIAL) that is safe by construction.
5. Note that already_processed is checked on content hash, not filename — a partner who re-sends a corrected file under the same name has a different hash, so it is (correctly) not skipped; it proceeds to the schema and completeness gates like any new file.

Output.

File	Exit gate	Verdict
clean on-time file	passes all four	LOAD
still-uploading file	Q1 (not written)	WAIT
re-sent duplicate	Q2 (hash seen)	SKIP
drifted header	Q3 (schema)	QUARANTINE
truncated file	Q4 (completeness)	HOLD_PARTIAL

Rule of thumb. Four gates, in order: written? seen? shaped right? complete? Only a file that clears all four reaches the warehouse; every other outcome is a safe non-load. Draw this tree on the whiteboard and the interviewer can hand you any file scenario.

Senior interview question on flat-file ingestion design

A senior interviewer often opens with: "A new partner will SFTP us a daily fixed-width settlement file that feeds the finance close. Design the end-to-end ingestion: how the file lands, how you guarantee you never read a half-written file, how pickup stays idempotent across retries and re-sends, how you detect a layout change, and how you prove the file is complete before it touches the ledger."

Solution Using an immutable landing zone + manifest gate + content-hash ledger

# ingest_settlement.py — the safe-to-load pipeline skeleton
import hashlib
import os
from datetime import datetime, timezone

RAW_PREFIX = "s3://raw/acme/settlement"      # immutable, date-partitioned

def sha256_of(path: str) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()

def is_fully_written(data_path: str) -> bool:
    # Sender drops a companion .ok trigger only after the data file closes.
    return os.path.exists(data_path + ".ok")

def already_loaded(conn, filename: str, digest: str) -> bool:
    with conn.cursor() as cur:
        cur.execute("""
            SELECT 1 FROM ingest_ledger
            WHERE  filename = %s AND content_sha256 = %s
        """, (filename, digest))
        return cur.fetchone() is not None

def record_loaded(conn, filename: str, digest: str, rows: int):
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO ingest_ledger(filename, content_sha256, row_count, loaded_at)
            VALUES (%s, %s, %s, now())
            ON CONFLICT (filename, content_sha256) DO NOTHING
        """, (filename, digest, rows))

def ingest(conn, local_path: str) -> str:
    filename = os.path.basename(local_path)

    # GATE 1 — never read a half-written file
    if not is_fully_written(local_path):
        return "WAIT"

    digest = sha256_of(local_path)

    # GATE 2 — idempotent pickup by (name, content hash)
    if already_loaded(conn, filename, digest):
        return "SKIP"

    # Land raw immutably BEFORE parsing (date-partitioned key)
    day = datetime.now(timezone.utc).strftime("%Y/%m/%d")
    raw_key = f"{RAW_PREFIX}/{day}/{filename}"
    put_object(raw_key, local_path)          # write-once; never overwritten

    # GATE 3 + GATE 4 — schema + completeness (see sections 4 and 5)
    layout = load_contract("acme_settlement", version="v3")
    rows = parse_fixed_width(raw_key, layout)         # raises on bad slice
    assert_schema_ok(rows, layout)                    # else QUARANTINE
    assert_complete(rows, raw_key)                    # trailer + checksum

    load_to_warehouse(rows, target="raw.settlement")
    record_loaded(conn, filename, digest, len(rows))
    return "LOAD"

-- Idempotency ledger — the durable memory of what has been loaded
CREATE TABLE ingest_ledger (
    filename        TEXT        NOT NULL,
    content_sha256  CHAR(64)    NOT NULL,
    row_count       BIGINT      NOT NULL,
    loaded_at       TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (filename, content_sha256)   -- exact-content dedupe
);

Step-by-step trace.

Gate	Mechanism	Result if it fails
1 — fully written	wait for .ok trigger	WAIT (retry next cycle)
2 — seen before	ledger on (filename, sha256)	SKIP (idempotent)
land raw	copy to immutable dated key	(write-once; re-runs read same bytes)
3 — schema	fingerprint vs contract v3	QUARANTINE
4 — completeness	trailer count + checksum	HOLD_PARTIAL
load	COPY into raw.settlement	LOAD + record in ledger

After deployment, a normal night runs: the .ok trigger appears, the file hashes to a value not in the ledger, the raw bytes are copied to s3://raw/acme/settlement/2026/08/18/, the fixed-width layout matches contract v3, the trailer count matches the parsed data rows, and the file loads — with its (filename, sha256) recorded so a retry or re-send is a no-op. A crash after the warehouse load but before the ledger insert is safe because the load target is idempotent by the same key; the re-run re-loads the same rows into the same partition and records the ledger row.

Output:

Scenario	Verdict	Warehouse effect
normal on-time file	LOAD	rows appear once
job runs mid-upload	WAIT	nothing loaded; retried
partner re-sends identical file	SKIP	no change (deduped)
partner sends corrected file (same name)	LOAD	new hash; loads as new
layout shifted by 2 bytes	QUARANTINE	held; owner alerted

Why this works — concept by concept:

• Untrusted-input framing — the file is treated as adversarial until proven safe. Every gate can reject it; only a file that clears all four reaches the warehouse. This inverts the naive "read then hope" flow into "prove then load."
• Atomic landing via trigger — the .ok companion file (or a temp-name-then-rename by the sender) is the "fully written" signal. Reading before it exists is the mid-upload bug that corrupts a load with a half-file; the trigger removes the race.
• Immutable raw before parse — copying the raw bytes to a write-once, date-partitioned key means every re-run reads identical input and you always have the original to reprocess. Parsing straight into the warehouse throws the evidence away.
• Content-hash idempotency ledger — keying on (filename, sha256) makes pickup idempotent: an identical re-send is skipped, a corrected re-send (new hash) is processed, and a crash-then-retry never double-loads. Filename alone is not enough; hash is what makes "exactly once" real.
• Cost — one full-file hash read (O(bytes)) and one ledger lookup per file — negligible against the cost of a double-loaded finance close or a silently mangled layout. The gates run in O(1) index lookups plus one O(rows) parse that would happen anyway. The eliminated cost is the 3 a.m. incident.

ETL
Topic — etl
ETL problems on batch file ingestion

Practice →

Validation Topic — data-validation Data-validation problems on untrusted input

Practice →

---

2. File landing zones and SFTP transport

The landing zone is a contract, not a folder — atomic drop, manifest gate, immutable raw, idempotent pickup

The mental model in one line: a file landing zone is a contract between the sender and your pipeline that specifies where files arrive, how you know one is fully written, and what happens to it after you read it — the senior design lands raw bytes immutably under a dated key, refuses to read any file until an atomic signal (a temp-name-then-rename, or a separate manifest/trigger file) says it is complete, and records every processed file in a content-hash ledger so SFTP pickup is idempotent across retries and re-sends. The folder is the easy part; the contract is what stops a half-written file from wrecking a load.

The four axes for landing and transport.

• Transport. SFTP (SSH File Transfer Protocol) is the 2026 B2B default — key-based auth, encrypted channel, firewall-friendly single port. FTPS and object-storage drop (S3/GCS/Azure) are the alternatives. The transport decides auth (SSH keys vs IAM) and notification (poll a directory vs an event like S3 ObjectCreated).
• Atomicity. How do you know the file is fully written? Three mechanisms: the sender writes file.csv.filepart and renames to file.csv only on completion (rename is atomic on the same filesystem); the sender drops a separate file.ok / manifest trigger after the data file closes; or you poll file size and wait for it to stop changing across two intervals. Never trust "the file exists" as "the file is done."
• Immutability. Raw bytes land write-once under a date-partitioned key (raw/<partner>/<yyyy>/<mm>/<dd>/<filename>). You never parse straight into the warehouse — you land, then process — so a bad file is quarantined, the original is always available to reprocess, and re-runs are deterministic.
• Idempotency. Pickup must be safe to re-run. A processed-file ledger keyed on (filename, content_sha256) means an identical re-send is skipped and a crash-then-retry never double-loads. The lifecycle is inbound → raw → processed | rejected, and moving/deleting the inbound file is a convenience, not the correctness mechanism — the ledger is.

The atomic-landing mechanisms — pick one and enforce it.

• Temp-name + rename. Sender uploads to settlement.csv.part, renames to settlement.csv when done. Your poller ignores *.part. Rename is atomic on POSIX filesystems, so you never observe a partial settlement.csv. This is the cleanest mechanism when you can dictate the sender's behaviour.
• Manifest / trigger file. Sender drops settlement.csv, then settlement.csv.ok (or a manifest.json listing files, sizes, and checksums). Your poller waits for the trigger. This also carries completeness metadata — expected row count and checksum — which the completeness gate (section 5) needs.
• Size-stability poll. When you control neither the sender nor a trigger, stat the file twice N seconds apart; if size and mtime are unchanged, treat it as complete. This is the weakest mechanism (a slow uploader can pause mid-transfer) and is a last resort.

SFTP mechanics senior engineers get right.

• Key-based auth, not passwords. Provision an SSH keypair per partner; rotate on a schedule; never embed passwords in DAG code. Store the private key in a secrets manager, not the repo.
• PGP before parse. Financial and healthcare files are usually PGP-encrypted on top of SFTP's transport encryption (defense in depth; the file is encrypted at rest on the SFTP box). Decrypt to a temp path before hashing/parsing — the decrypted bytes are what you hash for the ledger.
• Connection reliability. SFTP sessions drop. Wrap pickup in bounded retries with backoff; verify the downloaded byte count against the remote stat size; resume or re-download on mismatch. A truncated download is as dangerous as a truncated upload.
• Do not delete on the server blindly. Archive the remote file (move to processed/ on the SFTP box) or leave it and rely on your ledger. Deleting immediately means a failed load has no source to retry from.

Common interview probes on landing and SFTP.

• "How do you avoid reading a file that is still uploading?" — required answer: temp-name-then-rename or a .ok trigger; never trust existence alone.
• "How is pickup idempotent?" — ledger on (filename, content hash); skip if seen.
• "Where do you decrypt PGP?" — to a temp file before hashing and parsing; hash the plaintext.
• "Why land raw immutably?" — deterministic reprocessing, quarantine of bad files, an audit trail of exactly what arrived.

Worked example — polling an SFTP inbox with a trigger-file gate

Detailed explanation. The canonical SFTP pickup: poll the inbound directory, ignore any data file whose .ok trigger is missing, download matched pairs over a retried connection, verify the byte count, and hand off to landing. Build the poller.

• Directory. /inbound/acme/ on the partner SFTP server.
• Pairing. A data file X is ready only when X.ok also exists.
• Verification. Downloaded size must equal the remote stat size.
• Handoff. Ready files go to the landing/hash/ledger flow from section 1.

Question. Write an SFTP poller that returns only fully-written, verified-download files.

Input.

Parameter	Value
Host / dir	sftp.acme.com : /inbound/acme/
Auth	SSH key from secrets manager
Ready signal	companion <name>.ok trigger
Download check	local bytes == remote stat size

Code.

# sftp_poller.py — returns only complete, verified files
import os
import time
import paramiko

def open_sftp(host: str, user: str, key_path: str) -> paramiko.SFTPClient:
    key = paramiko.Ed25519Key.from_private_key_file(key_path)
    transport = paramiko.Transport((host, 22))
    transport.connect(username=user, pkey=key)
    return paramiko.SFTPClient.from_transport(transport)

def list_ready_files(sftp, remote_dir: str) -> list[str]:
    """A data file is ready iff its companion <name>.ok trigger exists."""
    names = set(sftp.listdir(remote_dir))
    ready = []
    for n in names:
        if n.endswith(".ok"):
            continue
        if f"{n}.ok" in names:          # trigger present -> fully written
            ready.append(n)
    return sorted(ready)

def download_verified(sftp, remote_dir: str, name: str, local_dir: str,
                      retries: int = 3) -> str:
    remote_path = f"{remote_dir}/{name}"
    local_path  = os.path.join(local_dir, name)
    remote_size = sftp.stat(remote_path).st_size

    for attempt in range(1, retries + 1):
        sftp.get(remote_path, local_path)
        local_size = os.path.getsize(local_path)
        if local_size == remote_size:               # download not truncated
            # mirror the trigger locally so downstream sees "fully written"
            open(local_path + ".ok", "w").close()
            return local_path
        time.sleep(2 ** attempt)                     # backoff, then retry
    raise IOError(f"{name}: download truncated after {retries} attempts "
                  f"({local_size} != {remote_size} bytes)")

def poll(host, user, key_path, remote_dir, local_dir) -> list[str]:
    sftp = open_sftp(host, user, key_path)
    try:
        return [download_verified(sftp, remote_dir, n, local_dir)
                for n in list_ready_files(sftp, remote_dir)]
    finally:
        sftp.close()

Step-by-step explanation.

1. open_sftp authenticates with an Ed25519 key loaded from a secrets-manager path — never a password, never a key checked into the repo. The transport is a single SSH channel, so it traverses partner firewalls cleanly.
2. list_ready_files is the atomic-landing gate: it walks the directory, skips trigger files themselves, and returns a data file only if its <name>.ok companion is present. A file mid-upload has no trigger yet, so it is invisible to the poller — the mid-upload read is impossible by construction.
3. download_verified guards the download side of atomicity. It records the remote size via stat, downloads, and compares the local byte count. A truncated download (dropped SFTP session) fails the equality check and retries with exponential backoff, so a network blip never yields a short local file that then parses as "complete."
4. On a verified download it writes a local .ok companion, propagating the "fully written" signal to the landing stage (section 1's is_fully_written). The whole pipeline speaks one atomicity dialect: a data file is real only when its trigger sits beside it.
5. Files are processed in sorted order for determinism, and the SFTP session is always closed in finally. Note what the poller deliberately does not do: it does not delete the remote file. Archiving/cleanup happens only after the ledger records a successful load, so a downstream failure always has a source to retry from.

Output.

Remote state	Poller behaviour
X present, X.ok absent	ignored (still uploading)
X + X.ok present	downloaded + verified + .ok mirrored
download truncated	retried with backoff; raises if persistent
X already downloaded earlier	re-listed; deduped later by hash ledger

Rule of thumb. Gate every SFTP pickup on an explicit "fully written" signal (trigger file or temp-rename), verify the downloaded byte count against the remote size, and never delete the remote file before your ledger confirms a successful load. Existence is not completeness.

Worked example — decrypting PGP before hashing and landing

Detailed explanation. Financial and healthcare feeds arrive PGP-encrypted on top of SFTP. The order of operations matters: decrypt to a temp path first, hash the plaintext for the idempotency ledger, then land the plaintext immutably. Hashing the ciphertext would break idempotency because PGP encryption is non-deterministic (a fresh session key per encryption), so the same plaintext yields different ciphertext each time.

• Order. download ciphertext → PGP decrypt → hash plaintext → land plaintext raw → parse.
• Key. Your private key decrypts; the partner encrypted with your public key.
• Why hash plaintext. PGP ciphertext of identical input differs run-to-run; only the plaintext hash is stable.

Question. Write the decrypt-then-hash step and explain why hashing the ciphertext would break the idempotency ledger.

Input.

Step	Input	Output
download	settlement.csv.pgp	local ciphertext
decrypt	ciphertext + private key	settlement.csv plaintext
hash	plaintext bytes	stable sha256 for ledger
land	plaintext	immutable raw key

Code.

# decrypt_and_land.py
import gnupg
import hashlib
import os

def decrypt_pgp(cipher_path: str, plain_path: str, gpg_home: str, passphrase: str) -> str:
    gpg = gnupg.GPG(gnupghome=gpg_home)
    with open(cipher_path, "rb") as f:
        result = gpg.decrypt_file(f, passphrase=passphrase, output=plain_path)
    if not result.ok:
        raise ValueError(f"PGP decrypt failed for {cipher_path}: {result.status}")
    return plain_path

def sha256_of(path: str) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()

def decrypt_hash_land(cipher_path: str, work_dir: str, raw_prefix: str) -> tuple[str, str]:
    base = os.path.basename(cipher_path).removesuffix(".pgp")
    plain_path = os.path.join(work_dir, base)

    # 1. Decrypt ciphertext -> plaintext
    decrypt_pgp(cipher_path, plain_path, gpg_home="/secrets/gpg", passphrase=os.environ["PGP_PASSPHRASE"])

    # 2. Hash the PLAINTEXT (stable across re-encryptions of the same data)
    digest = sha256_of(plain_path)

    # 3. Land plaintext immutably (this is what the parser + ledger use)
    raw_key = f"{raw_prefix}/{base}"
    put_object(raw_key, plain_path)          # write-once
    return raw_key, digest

Step-by-step explanation.

1. decrypt_pgp uses your private key (in an isolated gnupghome under /secrets) to turn the partner's ciphertext into plaintext, failing loudly if decryption does not succeed — a wrong key or a corrupted file must stop the pipeline, not produce garbage plaintext.
2. Hashing happens on the plaintext, and this is the crux. PGP uses a random session key per encryption, so encrypting the identical settlement data twice produces two different ciphertexts. If you hashed the ciphertext, every re-send — even of byte-identical data — would look new to the ledger and double-load. The plaintext hash is stable and is therefore the correct idempotency key.
3. The plaintext is landed immutably under the raw prefix; this is the artifact the parser and the completeness gate read. The ciphertext is transient working state and can be discarded after successful decryption (or archived for audit, but it is never the source of truth).
4. The passphrase comes from the environment/secrets manager, never the code. The gnupghome is isolated so the pipeline's keyring cannot be polluted by other processes.
5. The returned (raw_key, digest) feeds directly into section 1's ledger check: already_loaded(conn, filename, digest). Because digest is the plaintext hash, idempotency holds across the partner's non-deterministic re-encryptions.

Output.

Encryption run of identical data	Ciphertext hash	Plaintext hash
first send	a1b2...	9f3c...
re-send (re-encrypted)	d4e5... (different!)	9f3c... (same)
ledger verdict if hashing ciphertext	new → double-load bug	—
ledger verdict if hashing plaintext	—	seen → SKIP (correct)

Rule of thumb. Decrypt first, hash the plaintext, then land. Hashing PGP ciphertext silently breaks idempotency because encryption is non-deterministic — the same data re-encrypted looks new every time. The plaintext is your source of truth for both the ledger and the parser.

Worked example — the immutable raw zone and processed/rejected lifecycle

Detailed explanation. The landing zone has three logical areas — raw (immutable, everything that arrives), processed (successfully loaded), and rejected (failed a gate) — and a date-partitioned key layout that makes reprocessing and auditing trivial. Build the lifecycle and show how a file moves through it.

• Raw. raw/<partner>/<yyyy>/<mm>/<dd>/<filename> — write-once, never modified, retained per policy.
• Processed. A pointer/marker (not a copy) recording that a raw key loaded successfully.
• Rejected. Raw keys that failed schema or completeness, with a reason, awaiting review.

Question. Design the key layout and the state transitions, and show why raw stays immutable even on reprocessing.

Input.

Zone	Purpose	Mutable?
raw	exactly what arrived	no (write-once)
processed	audit of successful loads	append-only
rejected	quarantine + reason	append-only

Code.

# lifecycle.py — raw is write-once; state lives in a table, not by moving bytes
from datetime import datetime, timezone

def raw_key(partner: str, filename: str, arrived: datetime) -> str:
    d = arrived.strftime("%Y/%m/%d")
    return f"raw/{partner}/{d}/{filename}"

def land_raw(local_path: str, key: str) -> None:
    if object_exists(key):
        # write-once: a second landing of the same key is a no-op, not overwrite
        return
    put_object(key, local_path)

def set_state(conn, raw_key: str, state: str, reason: str = None) -> None:
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO file_state(raw_key, state, reason, at)
            VALUES (%s, %s, %s, now())
        """, (raw_key, state, reason))   # append-only history, not an UPDATE

-- File-state history: append-only; the latest row per raw_key is current state
CREATE TABLE file_state (
    raw_key   TEXT        NOT NULL,
    state     TEXT        NOT NULL,      -- 'landed' | 'loaded' | 'rejected'
    reason    TEXT,                      -- populated for 'rejected'
    at        TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX idx_file_state_key ON file_state (raw_key, at DESC);

-- Current state of every file that ever arrived
SELECT DISTINCT ON (raw_key) raw_key, state, reason, at
FROM   file_state
ORDER  BY raw_key, at DESC;

Step-by-step explanation.

1. raw_key embeds the arrival date, giving a natural, query-friendly partition (raw/acme/2026/08/18/...). Date-partitioning makes "reprocess everything from last Tuesday" a prefix scan and keeps object listings small.
2. land_raw is write-once: if the key already exists (a reprocess of the same arrival), it is a no-op rather than an overwrite. The bytes that arrived are frozen forever, so any later reprocessing reads exactly what the partner sent — no "someone patched the raw file" ambiguity.
3. State is tracked in an append-only file_state table, not by physically moving objects between raw/, processed/, and rejected/ folders. Moving bytes is slow, non-atomic across a crash, and destroys the "what actually arrived" record. A row per transition is atomic and gives a full audit history.
4. The DISTINCT ON (raw_key) ... ORDER BY at DESC query collapses the history into current state. A file that landed, was rejected for drift, then reprocessed and loaded after a contract update shows all three rows — the timeline is preserved, which auditors and on-call both need.
5. Rejection carries a reason (e.g. schema_drift: unexpected column, incomplete: trailer count mismatch), so a human reviewing the quarantine sees why without re-running the parser. The raw bytes are untouched, so once the underlying issue is fixed, reprocessing is a re-run against the same immutable key.

Output.

Event	file_state row	raw bytes
file lands	('...','landed',NULL)	written once
load succeeds	('...','loaded',NULL)	unchanged
fails completeness	('...','rejected','incomplete')	unchanged
reprocessed after fix	('...','loaded',NULL)	same bytes re-read

Rule of thumb. Keep raw write-once and track lifecycle as append-only state rows, not by moving files between folders. The bytes that arrived are evidence; freeze them, and let a state table — not the filesystem — hold "landed / loaded / rejected." Reprocessing then means re-reading identical input.

Senior interview question on landing zones and SFTP

A senior interviewer might ask: "Design the landing zone for a bank that SFTPs us PGP-encrypted settlement files, sometimes twice (a morning file and a corrected afternoon re-send under the same name). Cover the atomic-landing signal, PGP handling, the immutable raw layout, and how pickup stays idempotent so the identical morning file is never loaded twice but the corrected afternoon file is loaded."

Solution Using trigger-gated pickup + plaintext-hash ledger + immutable dated raw

# bank_settlement_landing.py — end-to-end landing with correct re-send semantics
import os
from datetime import datetime, timezone

def process_inbound(conn, sftp, remote_dir: str, work_dir: str) -> list[tuple[str, str]]:
    results = []
    for name in list_ready_files(sftp, remote_dir):          # trigger-gated (sec 2)
        cipher = download_verified(sftp, remote_dir, name, work_dir)

        # 1. Decrypt; hash the PLAINTEXT (stable across re-encryption)
        raw_prefix = f"raw/bank/{datetime.now(timezone.utc):%Y/%m/%d}"
        raw_key, digest = decrypt_hash_land(cipher, work_dir, raw_prefix)
        filename = os.path.basename(raw_key)

        # 2. Idempotent decision on (filename, plaintext hash)
        if already_loaded(conn, filename, digest):
            set_state(conn, raw_key, "skipped", reason="duplicate content")
            results.append((filename, "SKIP"))
            continue

        # 3. New content (first send OR corrected re-send) -> gates 3 & 4, then load
        set_state(conn, raw_key, "landed")
        rows = parse_and_validate(raw_key)                   # schema + completeness
        load_to_warehouse(rows, target="raw.bank_settlement", partition=digest[:12])
        record_loaded(conn, filename, digest, len(rows))
        set_state(conn, raw_key, "loaded")
        results.append((filename, "LOAD"))
    return results

-- The ledger PK is (filename, content_sha256): same name, different content = new load
-- so the corrected afternoon re-send (different plaintext) loads, while an
-- identical re-send (same plaintext) is skipped.
SELECT filename, content_sha256, row_count, loaded_at
FROM   ingest_ledger
WHERE  filename = 'bank_settlement_20260818.csv'
ORDER  BY loaded_at;
-- Two rows: morning (hash A) and corrected afternoon (hash B).

Step-by-step trace.

File	Trigger?	Plaintext hash	In ledger?	Verdict
morning file	yes	A	no	LOAD (records A)
identical re-poll of morning	yes	A	yes	SKIP
corrected afternoon (same name)	yes	B	no	LOAD (records B)
identical re-send of afternoon	yes	B	yes	SKIP

After deployment, the morning file lands (trigger present), decrypts, hashes to A, is not in the ledger, and loads — recording (name, A). If the poller re-sees it before cleanup, hash A is now in the ledger, so it is skipped. The corrected afternoon file arrives under the same name but decrypts to different plaintext, hashing to B; B is not in the ledger, so it correctly loads as a new version. Any identical re-send of the afternoon file hashes to B, is found, and is skipped. Filename-only dedupe would have wrongly skipped the correction; content-hash dedupe gets both cases right.

Output:

Requirement	Mechanism	Result
never read half-written file	.ok trigger gate	mid-upload impossible
identical re-send not double-loaded	ledger on plaintext hash	SKIP
corrected re-send is loaded	different plaintext hash	LOAD
PGP non-determinism handled	hash plaintext, not ciphertext	stable dedupe
deterministic reprocessing	immutable dated raw key	same bytes re-read

Why this works — concept by concept:

• Trigger-gated pickup — the .ok companion (mirrored on download) is the single source of truth for "fully written," making a mid-upload read structurally impossible rather than merely unlikely.
• Plaintext-hash idempotency — hashing after decryption defeats PGP's per-session-key non-determinism, so the ledger sees identical data as identical. This is what lets the same-name correction load (new hash) while the identical re-send is skipped (seen hash).
• Composite ledger key (filename, sha256) — filename groups the feed; the hash distinguishes versions. This exact pairing is what makes "skip duplicates but accept corrections" a two-line lookup instead of a fragile heuristic.
• Immutable dated raw + append-only state — the bytes are frozen and the lifecycle is a state history, so a correction, a rejection, and a reprocess are all recorded without ever mutating what arrived.
• Cost — one decrypt (O(bytes)), one plaintext hash (O(bytes)), one indexed ledger lookup per file. Trivial against a mis-loaded bank settlement. The design turns "did we already load this?" from an unanswerable question into an index probe.

File I/O
Topic — file-io
File I/O and landing-zone problems

Practice →

ETL Topic — etl ETL problems on batch ingestion pipelines

Practice →

---

3. Parsing CSV, fixed-width, and EDI formats

CSV is not simple, fixed-width has no delimiter to lean on, and EDI is a grammar — three formats, three failure surfaces

The mental model in one line: flat-file parsing is three genuinely different problems wearing the word "file" — CSV parsing is a quoting-and-encoding minefield where an embedded comma or newline inside a quoted field silently shifts every downstream column if you split(","); fixed-width has no delimiter at all, so a single-byte offset error corrupts every field to its right; and EDI (X12 / EDIFACT) is a nested-envelope grammar of segments and elements, not a table — and a senior parser respects the exact rules of whichever one it faces instead of reaching for the naive default. The bug is never "the file was weird"; it is "we assumed the format was simpler than it is."

The four axes for parsing.

• Delimiter / structure. CSV/TSV/pipe rely on a delimiter you must respect inside quotes; fixed-width relies on byte offsets with no delimiter; EDI relies on delimiter characters declared in the file's own header (X12 announces its element and segment separators in the ISA segment). You must discover the structure, not assume it.
• Quoting and escaping. RFC-4180 CSV wraps fields containing the delimiter, a quote, or a newline in double quotes, and escapes an embedded quote by doubling it (""). A parser that ignores quoting will split "Smith, Jr." into two columns. Fixed-width has no quoting; EDI escapes with a release character.
• Encoding. UTF-8 (possibly with a BOM), Latin-1/Windows-1252 (common from legacy mainframes), UTF-16, and EBCDIC (still alive on mainframe fixed-width feeds) all appear. Mis-decoding turns £ into £ or raises mid-file. Encoding must be pinned per feed, and a BOM stripped from the first field.
• Row shape / raggedness. Rows with too few or too many fields (ragged CSV), rows shorter than the fixed-width record length, or EDI segments out of expected order are structural errors. The senior choice is to reject the offending record to a rejected/ sink with its line number, not silently pad, truncate, or drop it.

CSV — the hazards that bite in production.

• Embedded delimiters and newlines. A quoted field can legally contain the delimiter ("Smith, Jr.") and even a literal newline ("123 Main St\nApt 4"). line.split(",") and reading line-by-line both corrupt these. Use a real CSV reader that understands quoting and multi-line fields.
• BOM and encoding. A UTF-8 BOM () prepended to the file attaches to the first header name, so id becomes id and your header lookup misses. Read with utf-8-sig (which strips the BOM) or strip it explicitly.
• Doubled-quote escaping. "He said ""hi""" is the single value He said "hi". Naive splitting mangles it; RFC-4180 readers handle it.
• Ragged rows and trailing delimiters. A row with a missing trailing field, or an extra empty field from a trailing comma, must be caught by asserting field count against the header — not papered over.

Fixed-width — no delimiter means no forgiveness.

• Byte offsets, not character offsets. With multibyte encodings, "column 10–20" may mean bytes 10–20, not characters. Legacy fixed-width is usually single-byte (Latin-1/EBCDIC), so byte == char; confirm which.
• Padding. Numeric fields are often zero-padded or space-padded; text is space-padded to the field width. Strip padding after slicing, and know whether right-justified (numbers) or left-justified (text).
• A one-byte shift is catastrophic. If the layout says amount is bytes 40–52 and the file inserted one extra byte upstream, every field from 40 on is off by one — and it will parse without error, just wrong. Validate a known-format field (e.g. a date column that must match YYYYMMDD) as a shift canary.

EDI (X12 / EDIFACT) — read the envelope, not the rows.

• The envelope hierarchy. X12 nests: ISA (interchange) → GS (functional group) → ST (transaction set, e.g. an 850 purchase order) → segments → SE/GE/IEA closers. EDIFACT uses UNB/UNG/UNH. You parse the tree, not flat lines.
• Self-describing delimiters. X12's ISA segment declares the element separator, sub-element separator, and segment terminator in fixed byte positions, so you read the delimiters from the file rather than assuming * and ~.
• Segments and elements. Each segment starts with a segment ID (N1, PO1), followed by delimiter-separated elements. A parser maps segment IDs to meaning per the transaction-set spec.
• Control numbers. ISA/GS/ST carry control numbers used for de-duplication and acknowledgement (the 997/999 functional ack). These are your completeness and idempotency hooks for EDI.

Common interview probes on parsing.

• "Why not split(',') for CSV?" — embedded delimiters/newlines inside quoted fields; use an RFC-4180 reader.
• "How do you parse fixed-width?" — slice by byte offset, strip padding, validate a canary field to catch shifts.
• "What is the ISA segment in X12?" — the interchange envelope that declares the delimiters and control numbers.
• "How do you handle a ragged row?" — reject it with its line number to a rejected sink; never silently pad or drop.

Worked example — a defensive CSV reader that rejects bad rows

Detailed explanation. The canonical defensive CSV read: use the standard library's RFC-4180 reader (handles quoting, embedded delimiters, and multi-line fields), read with a BOM-stripping encoding, validate each row's field count against the header, and route bad rows to a rejected sink with their line number instead of dropping or padding them. Build it.

• Reader. csv.reader (or csv.DictReader) — RFC-4180-compliant, not str.split.
• Encoding. utf-8-sig to strip a BOM transparently.
• Validation. Field count must equal header length; else reject with line number.

Question. Write a CSV reader that yields clean rows and collects rejects with reasons.

Input.

Line	Raw content	Outcome
header	id,name,amount	3 columns
1	1,"Smith, Jr.",100	OK (quoted comma)
2	2,Doe,	OK (empty amount)
3	3,Roe,50,EXTRA	reject (4 fields)

Code.

# defensive_csv.py — RFC-4180 read, BOM-safe, ragged rows rejected (not dropped)
import csv
from dataclasses import dataclass, field

@dataclass
class ParseResult:
    rows: list[dict] = field(default_factory=list)
    rejects: list[tuple[int, str, str]] = field(default_factory=list)  # (lineno, raw, reason)

def read_csv(path: str, delimiter: str = ",") -> ParseResult:
    result = ParseResult()
    # utf-8-sig transparently strips a leading BOM from the first header cell
    with open(path, "r", encoding="utf-8-sig", newline="") as f:
        reader = csv.reader(f, delimiter=delimiter, quotechar='"',
                            doublequote=True, strict=True)
        try:
            header = next(reader)
        except StopIteration:
            result.rejects.append((0, "", "empty file"))
            return result
        header = [h.strip() for h in header]
        width = len(header)

        for lineno, cells in enumerate(reader, start=2):   # data starts at line 2
            if len(cells) != width:
                raw = delimiter.join(cells)
                result.rejects.append(
                    (lineno, raw, f"expected {width} fields, got {len(cells)}"))
                continue
            result.rows.append(dict(zip(header, (c.strip() for c in cells))))
    return result

Step-by-step explanation.

1. csv.reader with quotechar='"' and doublequote=True implements RFC-4180: line 1's "Smith, Jr." is correctly read as the single value Smith, Jr. rather than split on the internal comma. A hand-rolled split(",") would have produced four fields and shifted amount into the name. strict=True makes malformed quoting raise instead of silently guessing.
2. Opening with encoding="utf-8-sig" strips a UTF-8 BOM if present, so the first header cell is id, not id. Without this, header[0] would carry the BOM and every dictionary key lookup on id downstream would fail with a KeyError that is maddening to diagnose.
3. Field-count validation is the ragged-row guard: line 3 has four fields against a three-column header, so it is rejected with its line number and reason, not padded to three or dropped silently. Silent handling of ragged rows is exactly how a column-shift corruption enters the warehouse unnoticed.
4. Empty fields are legal data, not errors: line 2's empty amount yields {"amount": ""}, which downstream typing can turn into NULL. The parser distinguishes "missing value" (fine) from "wrong number of fields" (structural error) — only the latter is a reject.
5. The result separates clean rows from rejects, so the pipeline can load the good rows and surface the bad ones for the feed owner with enough context (line number + raw text + reason) to fix the source. Nothing is thrown away.

Output.

Line	Parsed as	Bucket
1	{id:1, name:Smith, Jr., amount:100}	rows
2	{id:2, name:Doe, amount:}	rows
3	expected 3 fields, got 4	rejects

Rule of thumb. Never split(',') a CSV. Use an RFC-4180 reader with explicit quote handling, read as utf-8-sig to defuse the BOM, and reject ragged rows with their line number rather than padding or dropping them. Empty fields are data; wrong field counts are structural errors.

Worked example — a fixed-width slicer with a shift canary

Detailed explanation. Fixed-width parsing slices each record by byte offset per a layout spec, strips padding, and — because a one-byte upstream shift parses without error but corrupts everything — validates a known-format "canary" field to detect misalignment. Build the slicer.

• Layout. A list of (name, start, length, justify) slices.
• Padding. Strip after slicing; numbers right-justified, text left-justified.
• Canary. A field with a strict format (a YYYYMMDD date) that must validate, or the record is flagged as shifted.

Question. Write a fixed-width parser that slices by the layout and rejects records where the canary field fails, indicating a byte shift.

Input.

Field	Start	Len	Justify
emp_id	0	6	right (zero-pad)
name	6	20	left (space-pad)
pay_date	26	8	canary YYYYMMDD
amount_cents	34	10	right (zero-pad)

Code.

# fixed_width.py — offset slicing with a canary to catch byte shifts
from datetime import datetime

LAYOUT = [
    ("emp_id",       0,  6,  "right"),
    ("name",         6,  20, "left"),
    ("pay_date",     26, 8,  "left"),    # canary: must be YYYYMMDD
    ("amount_cents", 34, 10, "right"),
]
RECORD_LEN = 44

def parse_record(line: str, lineno: int) -> dict:
    # A short line means truncation or a wrong layout — never silently pad.
    if len(line) < RECORD_LEN:
        raise ValueError(f"line {lineno}: length {len(line)} < expected {RECORD_LEN}")

    rec = {}
    for name, start, length, justify in LAYOUT:
        raw = line[start:start + length]
        rec[name] = raw.rstrip() if justify == "left" else raw.lstrip("0 ")

    # Canary: a shift of even one byte breaks the date format.
    try:
        datetime.strptime(rec["pay_date"], "%Y%m%d")
    except ValueError:
        raise ValueError(
            f"line {lineno}: canary pay_date={rec['pay_date']!r} not YYYYMMDD "
            f"-> likely byte-offset shift")
    return rec

def parse_fixed_width(lines: list[str]) -> tuple[list[dict], list[tuple[int, str]]]:
    rows, rejects = [], []
    for lineno, line in enumerate(lines, start=1):
        try:
            rows.append(parse_record(line, lineno))
        except ValueError as e:
            rejects.append((lineno, str(e)))
    return rows, rejects

Step-by-step explanation.

1. Each field is sliced by absolute byte offset (line[start:start+length]). Fixed-width has no delimiter, so the layout is the schema; get an offset wrong and the field is silently wrong. The RECORD_LEN check rejects any line shorter than the layout expects — a truncated record must never be padded into looking valid.
2. Padding is stripped according to justification: text is left-justified and space-padded (rstrip), numbers are right-justified and zero/space-padded (lstrip("0 ")). Stripping the wrong side would turn 000042 into 000042 kept or 42 — you must know the convention per field.
3. The canary is the shift detector. pay_date must parse as YYYYMMDD; if an upstream process inserted or dropped a byte, the date slice now contains part of the name or the amount and fails strptime. Without a canary, a shifted file loads perfectly cleanly and wrong — the most dangerous fixed-width failure.
4. A failed canary raises with the offending value and line number, routing the record to rejects with a message that names the likely cause ("byte-offset shift"). This turns an invisible corruption into an actionable alert.
5. parse_fixed_width collects clean rows and rejects separately, mirroring the CSV reader's contract: load the good, surface the bad with context. The canary check runs per record, so a shift that starts partway through a file (mixed valid/invalid) is caught precisely at the first bad record.

Output.

Record	pay_date slice	Verdict
aligned record	20260818	row
one-byte shift	0260818X	reject (canary fails)
truncated line	(< 44 chars)	reject (short record)

Rule of thumb. Slice fixed-width by explicit byte offsets, strip padding by justification, and always validate a strict-format canary field (a date or a check digit). A byte shift parses without error and loads wrong — the canary is the only thing standing between you and silent corruption.

Worked example — an X12 EDI envelope parser (ISA/GS/ST)

Detailed explanation. X12 EDI is a nested envelope, and — critically — it declares its own delimiters in the ISA segment, so you read the element separator, sub-element separator, and segment terminator from the file before splitting anything. Build a parser that reads the delimiters from ISA and walks the ISA → GS → ST → segments hierarchy.

• ISA is fixed-position. The element separator is byte 3; the sub-element separator and segment terminator sit at the end of the 106-byte ISA.
• Hierarchy. ISA (interchange) → GS (group) → ST (transaction set) → segments → SE/GE/IEA closers.
• Control numbers. ISA13 / GS06 / ST02 are used for dedupe and acknowledgement.

Question. Write an X12 parser that discovers delimiters from ISA and returns the envelope structure with control numbers.

Input.

Segment	Meaning	Key element
ISA	interchange header	ISA13 = control number
GS	functional group	GS06 = group control
ST	transaction set	ST01 = type (e.g. 850)
SE/GE/IEA	closers	counts + control echoes

Code.

# x12_parser.py — delimiters are declared in ISA; read them from the file
from dataclasses import dataclass, field

@dataclass
class Interchange:
    control_number: str
    element_sep: str
    segment_term: str
    groups: list["Group"] = field(default_factory=list)

@dataclass
class Group:
    control_number: str
    transactions: list["Transaction"] = field(default_factory=list)

@dataclass
class Transaction:
    set_type: str            # e.g. "850" purchase order
    control_number: str
    segments: list[list[str]] = field(default_factory=list)

def parse_x12(raw: str) -> Interchange:
    if not raw.startswith("ISA"):
        raise ValueError("not an X12 interchange (no ISA)")

    # The ISA segment is fixed-length; delimiters are AT KNOWN BYTE POSITIONS.
    element_sep  = raw[3]            # byte 3 = element separator
    segment_term = raw[105]         # byte 105 = segment terminator (end of ISA)
    segments = [s for s in raw.split(segment_term) if s.strip()]

    isa = segments[0].split(element_sep)
    interchange = Interchange(control_number=isa[13].strip(),
                              element_sep=element_sep, segment_term=segment_term)

    group = txn = None
    for seg in segments[1:]:
        els = seg.split(element_sep)
        tag = els[0].strip()
        if tag == "GS":
            group = Group(control_number=els[6].strip())
            interchange.groups.append(group)
        elif tag == "ST":
            txn = Transaction(set_type=els[1].strip(), control_number=els[2].strip())
            group.transactions.append(txn)
        elif tag in ("SE", "GE", "IEA"):
            txn = None if tag == "SE" else txn      # close the transaction set
        elif txn is not None:
            txn.segments.append(els)                # a data segment inside the ST
    return interchange

Step-by-step explanation.

1. The parser refuses anything that does not start with ISA — the interchange header is mandatory and fixed-position. Because X12 lets each sender choose delimiters, you cannot hard-code * and ~; you read the element separator from byte 3 and the segment terminator from byte 105 of the ISA. This "discover the delimiter from the file" step is the senior EDI move.
2. Splitting on the discovered segment_term yields the flat list of segments; splitting each on the element_sep yields its elements. The ISA's element 13 (isa[13]) is the interchange control number — the top-level dedupe/ack key.
3. The walk maintains the envelope hierarchy with a small state machine: GS opens a functional group, ST opens a transaction set (recording its type, e.g. 850 for a purchase order, and its control number), and SE/GE/IEA close the respective levels. This mirrors the nested grammar rather than treating segments as flat rows.
4. Data segments (anything between ST and SE, like PO1 line items) are appended to the current transaction. The structure returned is a tree — interchange → groups → transactions → segments — which is what a downstream mapper needs to turn an 850 into order rows.
5. The control numbers (ISA13, GS06, ST02) captured at each level are the EDI equivalent of a file hash: they de-duplicate re-sent interchanges and are echoed in the 997/999 functional acknowledgement you send back. They are the completeness/idempotency hooks for EDI specifically.

Output.

Level	Field captured	Example
Interchange (ISA)	control number, delimiters	000000123, *, ~
Group (GS)	group control number	1
Transaction (ST)	set type + control number	850, 0001
Segments	data rows under the ST	PO1, N1, ...

Rule of thumb. For X12, read the delimiters from the ISA segment before splitting anything, then walk the ISA → GS → ST → SE/GE/IEA envelope as a tree. Capture the control numbers — they are your dedupe key and the basis of the functional acknowledgement. EDI is a grammar, not a table.

Senior interview question on flat-file parsing

A senior interviewer might ask: "You inherit an ingestion job that does line.split(',') on partner CSVs and it keeps corrupting rows where customer names contain commas, plus a separate fixed-width feed that occasionally shifts by a byte and loads silently wrong. Redesign both parsers to be defensive, and explain how you would detect the fixed-width shift before the bad data reaches the warehouse."

Solution Using an RFC-4180 reader + offset slicer with a validating canary + reject sink

# robust_ingest.py — both formats, defensively, with a shared reject sink
import csv
from datetime import datetime

def parse_csv_strict(path: str, expected_cols: list[str]) -> tuple[list[dict], list[tuple]]:
    rows, rejects = [], []
    with open(path, encoding="utf-8-sig", newline="") as f:
        reader = csv.reader(f, quotechar='"', doublequote=True, strict=True)
        header = [h.strip() for h in next(reader)]
        if header != expected_cols:                      # schema check (see sec 4)
            raise ValueError(f"header {header} != contract {expected_cols}")
        for lineno, cells in enumerate(reader, start=2):
            if len(cells) != len(header):
                rejects.append((lineno, ",".join(cells),
                                f"got {len(cells)} fields, want {len(header)}"))
                continue
            rows.append(dict(zip(header, (c.strip() for c in cells))))
    return rows, rejects

FW_LAYOUT = [("acct", 0, 12, "right"), ("txn_date", 12, 8, "left"),
             ("amount_cents", 20, 12, "right")]
FW_LEN = 32

def parse_fixed_strict(lines: list[str]) -> tuple[list[dict], list[tuple]]:
    rows, rejects = [], []
    for lineno, line in enumerate(lines, start=1):
        if len(line.rstrip("\r\n")) < FW_LEN:
            rejects.append((lineno, "short record")); continue
        rec = {n: (line[s:s+l].rstrip() if j == "left" else line[s:s+l].lstrip("0 "))
               for n, s, l, j in FW_LAYOUT}
        try:
            datetime.strptime(rec["txn_date"], "%Y%m%d")   # canary
        except ValueError:
            rejects.append((lineno, f"canary txn_date={rec['txn_date']!r} -> shift"))
            continue
        rows.append(rec)
    return rows, rejects

-- Reject sink: every bad record is captured with context, never dropped
CREATE TABLE ingest_rejects (
    feed        TEXT        NOT NULL,
    raw_key     TEXT        NOT NULL,
    line_no     INT         NOT NULL,
    raw_text    TEXT,
    reason      TEXT        NOT NULL,
    rejected_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- Alert when a feed's reject rate crosses a threshold (drift or shift signal)
SELECT feed, count(*) AS rejects
FROM   ingest_rejects
WHERE  rejected_at > now() - INTERVAL '1 day'
GROUP  BY feed HAVING count(*) > 100;

Step-by-step trace.

Input hazard	Naive result	Defensive result
1,"Smith, Jr.",100	4 fields (corrupt)	3 fields (correct)
BOM on header	id key misses	stripped by utf-8-sig
CSV row with extra field	silently shifted	rejected with line no
fixed-width byte shift	loads wrong, no error	canary fails -> rejected
truncated fixed record	padded silently	rejected (short record)

After deployment, the CSV parser reads quoted commas correctly, strips the BOM, and diverts ragged rows to ingest_rejects with their line numbers; the fixed-width parser slices by offset and catches any byte shift via the txn_date canary before a single wrong amount reaches the ledger. A rising reject count on a feed becomes an alert — the early-warning signal that the partner changed something.

Output:

Metric	Before (naive)	After (defensive)
quoted-comma rows	corrupted	correct
BOM handling	broken key lookups	transparent strip
ragged/short rows	silently mangled	rejected with context
fixed-width shift	silent corruption	caught by canary
observability	none	reject-rate alerting

Why this works — concept by concept:

• RFC-4180 reader — the standard CSV reader honours quoting, doubled-quote escaping, and multi-line fields, so embedded delimiters no longer shift columns. This single change eliminates the entire class of split(',') corruption bugs.
• BOM-safe decoding — reading as utf-8-sig strips the byte-order mark from the first header cell, preventing the silent KeyError-on-first-column failure that plagues naive readers.
• Field-count and length gates — asserting field count (CSV) and record length (fixed-width) turns structural errors into explicit rejects with line numbers instead of silent padding, truncation, or drops.
• Validating canary for fixed-width — a strict-format field (YYYYMMDD date) that must validate is the only reliable detector of a byte-offset shift, which otherwise parses cleanly and loads wrong. It converts an invisible corruption into a targeted reject.
• Cost — the RFC-4180 reader is the same O(bytes) scan as a naive split; the canary is one strptime per record. Negligible cost for eliminating silent column corruption. The reject sink adds observability (reject-rate alerts) that pays for itself the first time a partner quietly changes a layout.

CSV
Topic — csv-parsing
CSV parsing and quoting-edge-case problems

Practice →

Parsing Topic — parsing Parsing problems on delimited and fixed-width data

Practice →

---

4. Schema drift detection and handling

The partner owns the schema and will change it — fingerprint the header, classify the change, route by policy

The mental model in one line: schema drift is the inevitability that the sender changes the file's shape — adds a column, drops one, reorders two, renames a field, or starts sending a string where you expected an integer — without telling you, and the senior design does not prevent drift (you cannot) but detects it on arrival by fingerprinting the header against a versioned contract and routes by policy: additive changes evolve automatically, breaking changes fail loudly, and ambiguous changes quarantine for a human. The failure mode is not drift itself; it is drift that loads silently because nobody checked.

The four axes for schema drift.

• Detection. How do you notice the shape changed? A schema fingerprint — a hash of the ordered (column_name, declared_type) list, or of the header row — compared against the contract on every arrival. Cheap, deterministic, and it catches drift before any row loads.
• Classification. What kind of change is it? The six kinds — add, drop, reorder, rename, type change, width/cardinality change — have wildly different risk. An added trailing column is usually safe; a reorder in a headerless file is catastrophic (every column silently maps to the wrong target).
• Policy. What do you do about each kind? The policy matrix maps change-kind → action: additive → evolve, drop/type-change/reorder → quarantine or fail. The default for anything unrecognised must be "do not load."
• Contract source of truth. Where does "expected" live? In a versioned data contract (a table or a checked-in spec) — expected columns, order, types, delimiter, encoding — not hard-coded in the parser. Onboarding a schema change becomes updating the contract, reviewed like code.

The six kinds of drift, ranked by danger.

• Add a column (usually safe). A new trailing column. If your parser is header-driven and selects by name, this is additive and evolvable. If your parser is positional, a mid-row insert shifts everything after it — dangerous.
• Reorder columns (danger depends on header). With a header and name-based mapping, reorder is harmless — you map by name. In a headerless fixed-width or positional CSV, a reorder is invisible and corrupts every affected column. This is why headerless feeds need a stricter contract.
• Rename a column (breaks name-based mapping). amt → amount breaks a name-keyed parser (the old key vanishes) but is invisible to a positional one. Neither should auto-evolve; a rename needs a human to update the mapping.
• Type change (silent corruption risk). A column that was always integer starts carrying "N/A" or a decimal. It may still parse as text and load, then break every downstream cast. Type drift is quarantine-worthy because it usually indicates an upstream data-quality change.
• Drop a column (breaks downstream). A column your warehouse table and downstream models depend on disappears. This must fail or quarantine — evolving by silently filling NULLs hides a real upstream problem.
• Width / cardinality change (fixed-width specific). A fixed-width field grows from 10 to 12 bytes, shifting the record layout. Effectively a reorder for every field after it; must fail against the contract's record length.

The drift policy matrix — the senior artifact.

• Additive (new nullable column) → evolve. Add the column to the target (nullable), record the contract bump, load. Safe because existing consumers ignore the new column.
• Reorder with header → tolerate. Map by name; the physical order does not matter. Assert the set of names matches; ignore order.
• Reorder without header, rename, drop, type-change → quarantine or fail. These change meaning. Hold the file, alert the feed owner, and require a contract update (reviewed) before loading. Never auto-evolve a meaning-changing drift.
• The default. Any header the fingerprint does not recognise as the contract or an allowed evolution is quarantined. Deny-by-default is the safe posture.

Common interview probes on schema drift.

• "How do you detect drift?" — fingerprint the ordered (name, type) list; compare to the contract on arrival.
• "A partner adds a column — load or fail?" — if additive/trailing and you map by name, evolve; otherwise quarantine.
• "Why is a reorder dangerous?" — only in headerless/positional files, where it silently mis-maps columns.
• "Where does the expected schema live?" — a versioned contract as data, reviewed like code; not hard-coded in the parser.

Worked example — a schema fingerprint and drift classifier

Detailed explanation. The core mechanism: compute a deterministic fingerprint of the arriving header, diff it against the contract, and classify the difference into one of the six kinds so the policy layer can route it. Build the fingerprint and classifier.

• Fingerprint. A hash over the ordered list of column names (and types if known).
• Diff. Set difference for add/drop; order comparison for reorder; position-value comparison for rename.
• Output. A classification the policy matrix consumes.

Question. Write a classifier that, given the contract columns and the arriving columns, returns the drift kind.

Input.

Case	Contract	Arrived	Expected class
A	id,name,amount	id,name,amount	none
B	id,name,amount	id,name,amount,region	added
C	id,name,amount	id,amount,name	reordered
D	id,name,amount	id,name	dropped

Code.

# drift.py — fingerprint + classify header drift against a contract
import hashlib

def fingerprint(columns: list[str]) -> str:
    joined = "\x1f".join(c.strip().lower() for c in columns)   # unit-separator
    return hashlib.sha256(joined.encode()).hexdigest()[:16]

def classify_drift(contract: list[str], arrived: list[str]) -> dict:
    c_set, a_set = set(contract), set(arrived)
    added   = [c for c in arrived  if c not in c_set]
    dropped = [c for c in contract if c not in a_set]

    if fingerprint(contract) == fingerprint(arrived):
        kind = "none"
    elif added and not dropped and arrived[:len(contract)] == contract:
        kind = "added"                      # new trailing column(s), prefix intact
    elif not added and not dropped and set(arrived) == set(contract):
        kind = "reordered"                  # same names, different order
    elif dropped and not added:
        kind = "dropped"
    elif added and dropped and len(added) == len(dropped):
        kind = "renamed_or_swapped"         # ambiguous: needs a human
    else:
        kind = "mixed"

    return {"kind": kind, "added": added, "dropped": dropped,
            "contract_fp": fingerprint(contract), "arrived_fp": fingerprint(arrived)}

Step-by-step explanation.

1. fingerprint normalises each column name (trim + lowercase) and joins with a unit-separator that cannot appear in a name, then hashes. Normalising means a cosmetic case change (ID vs id) does not falsely trip drift, while the order-preserving join means a reorder does change the fingerprint (order is part of the identity).
2. The classifier computes set differences first: added (in arrival, not contract) and dropped (in contract, not arrival). These two lists drive most of the classification.
3. added and not dropped and arrived[:len(contract)] == contract is the precise test for a safe additive change: new column(s) appear, nothing was removed, and the original columns remain in their original positions as a prefix. This is the only case the policy layer will auto-evolve.
4. Same names in a different order (set equal, sequence not) is reordered — safe if the parser maps by name, dangerous if positional, so the policy layer decides based on whether the feed is headered. Equal added and dropped counts is flagged renamed_or_swapped — genuinely ambiguous (did amt become amount, or was one dropped and another added?), so it is never auto-resolved.
5. The classifier returns both fingerprints alongside the kind, so the decision is auditable: the contract's fingerprint and the arrival's fingerprint are logged with the verdict, giving on-call a stable identifier for "this exact drift" across files.

Output.

Case	added	dropped	kind
A	[]	[]	none
B	[region]	[]	added
C	[]	[]	reordered
D	[]	[name]	dropped

Rule of thumb. Fingerprint the ordered, normalised header and classify the diff into add/drop/reorder/rename before loading a row. Only a strict "new trailing columns, original prefix intact" pattern is safe to auto-evolve; everything else is at best name-mapped, at worst quarantined.

Worked example — the evolve/quarantine/fail policy router

Detailed explanation. With drift classified, the policy router maps the kind (and the feed's header-ness) to an action: evolve the target schema, quarantine for review, or fail the load. Build the router and the schema-evolution step for the additive case.

• Evolve. Additive + header-mapped → ALTER TABLE ADD COLUMN (nullable), bump contract, load.
• Quarantine. Rename, drop, type change, or reorder-without-header → hold + alert.
• Fail. Contract violation with no safe action (e.g. fixed-width record-length change) → reject.

Question. Write the router that turns a drift classification into an action, and the evolve step for additive drift.

Input.

Drift kind	Headered feed?	Action
none	any	load
added	yes	evolve then load
reordered	yes	load (name-mapped)
reordered	no	quarantine
dropped / renamed / type	any	quarantine

Code.

# policy.py — route a drift classification to an action
def decide_action(drift: dict, headered: bool) -> str:
    kind = drift["kind"]
    if kind == "none":
        return "LOAD"
    if kind == "added" and headered:
        return "EVOLVE"                 # additive + name-mapped -> safe
    if kind == "reordered":
        return "LOAD" if headered else "QUARANTINE"
    if kind in ("dropped", "renamed_or_swapped", "type_change"):
        return "QUARANTINE"             # meaning changed -> human review
    return "QUARANTINE"                 # mixed / unknown -> deny by default

def evolve_target(conn, table: str, new_cols: list[str]) -> None:
    """Additive evolution: add each new column as NULLABLE (safe for consumers)."""
    with conn.cursor() as cur:
        for col in new_cols:
            cur.execute(f'ALTER TABLE {table} ADD COLUMN IF NOT EXISTS "{col}" TEXT NULL')
    conn.commit()

def handle_arrival(conn, drift: dict, headered: bool, table: str, raw_key: str) -> str:
    action = decide_action(drift, headered)
    if action == "EVOLVE":
        evolve_target(conn, table, drift["added"])
        bump_contract_version(conn, table, drift["arrived_fp"])   # record the new shape
        return "LOAD"                                             # then load
    if action == "QUARANTINE":
        set_state(conn, raw_key, "rejected", reason=f"schema_drift:{drift['kind']}")
        alert_feed_owner(raw_key, drift)
    return action

Step-by-step explanation.

1. decide_action is the policy matrix in code. The only two paths to a load without human involvement are none (fingerprint matches) and additive-on-a-headered-feed (EVOLVE). Everything meaning-changing routes to QUARANTINE, and the final catch-all is also quarantine — deny-by-default for any classification the matrix does not explicitly bless.
2. Reorder is the one kind whose action depends on the feed: with a header, columns map by name so order is irrelevant and it loads; without a header (positional/fixed-width), a reorder silently mis-maps and must be quarantined. Encoding this conditional is the senior nuance juniors miss.
3. evolve_target implements safe additive evolution: each new column is added as NULLABLE TEXT with IF NOT EXISTS, so existing downstream consumers that do not reference the column are unaffected, and re-running the evolution is idempotent. New columns are typed permissively (TEXT) on arrival; tightening the type is a later, deliberate migration.
4. On evolve, the contract version is bumped to the arrival's fingerprint, recording that this new shape is now the expected one. The next file with this shape will fingerprint-match and take the none fast path — so an approved evolution becomes the new normal without a code change.
5. On quarantine, the file's state is set to rejected with a reason that names the drift kind, and the feed owner is alerted. The raw bytes are untouched (immutable), so once a human confirms the change and updates the contract, the same file reprocesses cleanly. Nothing is lost; the load is merely paused until the shape is understood.

Output.

Arrival	kind	headered	action
adds region (trailing)	added	yes	EVOLVE → LOAD
swaps two columns	reordered	yes	LOAD
swaps two columns	reordered	no	QUARANTINE
amt→amount	renamed_or_swapped	any	QUARANTINE

Rule of thumb. Encode the drift policy as an explicit matrix: only "fingerprint match" and "additive on a headered feed" load without a human; everything meaning-changing quarantines. Evolve additive columns as nullable, bump the contract to the new fingerprint, and let deny-by-default handle everything the matrix does not recognise.

Worked example — a headerless positional feed needs a stricter contract

Detailed explanation. Headerless feeds (fixed-width, positional CSV without a header row) are the highest-risk drift surface because there is no column name to map by — the position is the only identity, so any reorder, insert, or width change silently corrupts. The mitigation is a stricter contract: the expected column order and (for fixed-width) exact offsets are the contract, and a per-file structural check plus the section-3 canary detect drift that a headerless file cannot announce.

• The risk. No header means drift is invisible; position is meaning.
• The contract. Expected ordered fields + offsets + record length, versioned as data.
• The checks. Record-length assertion + canary field(s) + optional value-domain checks per column.

Question. Design the drift defenses for a headerless fixed-width feed where the partner might silently insert a field.

Input.

Defense	Catches
record-length assertion	width/insert changes
canary field validation	byte shifts from inserts
per-column domain check	type/format drift
contract version pin	any layout change

Code.

# headerless_contract.py — positional feeds validate structure, not names
CONTRACT = {
    "version": "v4",
    "record_len": 32,
    "fields": [
        {"name": "acct",   "start": 0,  "len": 12, "domain": r"^\d{12}$"},
        {"name": "date",   "start": 12, "len": 8,  "domain": r"^\d{8}$"},   # canary
        {"name": "amount", "start": 20, "len": 12, "domain": r"^\d{1,12}$"},
    ],
}

import re
from datetime import datetime

def validate_positional(line: str, contract: dict) -> list[str]:
    """Return a list of violations; empty list = record matches the contract."""
    problems = []
    body = line.rstrip("\r\n")
    if len(body) != contract["record_len"]:
        problems.append(f"record_len {len(body)} != {contract['record_len']}")
        return problems                       # length wrong -> offsets meaningless

    for f in contract["fields"]:
        val = body[f["start"]:f["start"] + f["len"]].strip()
        if not re.match(f["domain"], val):
            problems.append(f"{f['name']}={val!r} violates {f['domain']}")
    # date canary: format AND a plausibility check
    date_val = body[12:20]
    try:
        datetime.strptime(date_val, "%Y%m%d")
    except ValueError:
        problems.append(f"canary date={date_val!r} -> layout shift")
    return problems

Step-by-step explanation.

1. For a headerless feed the contract is the ordered field list with exact offsets and a record length — there are no names in the file to trust, so the contract carries all the meaning. Pinning record_len first is deliberate: if the total length is wrong, every offset is suspect and the per-field checks would produce noise, so the function returns early.
2. Each field is validated against a domain regex, not just sliced. acct must be 12 digits, amount must be numeric — these domain checks are how a headerless feed detects the type/format drift that a headered feed would catch by name. A partner who starts padding accounts with letters trips the ^\d{12}$ check immediately.
3. The date field doubles as a canary (section 3): a byte insert upstream shifts the date slice so strptime fails, catching a structural drift that the record-length check alone might miss if the insert is offset by a compensating trim elsewhere. Two independent structural checks (length + canary) make silent misalignment very hard.
4. The function returns a list of violations rather than a boolean, so a quarantined record carries a precise, human-readable reason set ("amount violates ^\d{1,12}$", "canary date -> shift"). This is what the feed owner needs to diagnose whether the partner changed the layout or sent bad data.
5. The contract is versioned (v4); a deliberate layout change is a reviewed contract bump, after which the new offsets are authoritative. Because the file has no header to fingerprint, the contract version is the schema identity for this feed — the whole drift-detection story rests on it.

Output.

Record	length	canary	verdict
well-formed	32	valid date	load
partner inserted a field	34	fails	reject (record_len)
letters in acct	32	valid	reject (domain)
one-byte shift	32	fails	reject (canary)

Rule of thumb. Headerless feeds are the riskiest drift surface — position is meaning and there is no name to map by. Compensate with a stricter contract: pin the record length, validate every field's domain, and keep a date/check-digit canary. Two independent structural checks catch the silent misalignment a headerless file cannot announce.

Senior interview question on schema drift

A senior interviewer might ask: "You own a nightly CSV feed from a partner who, historically, has added columns without warning, occasionally renamed one, and once shipped a file where an integer column contained 'N/A'. Design drift detection and handling so additive changes flow through automatically, renames and type changes are caught and reviewed, and nothing corrupt ever silently loads. Cover the contract, the fingerprint, and the policy."

Solution Using a versioned contract + header fingerprint + evolve/quarantine/fail router

# drift_pipeline.py — full drift handling for a headered CSV feed
def ingest_with_drift_control(conn, raw_key: str, feed: str, target: str) -> str:
    contract = load_contract(conn, feed)                 # versioned: columns + types
    header, sample = read_header_and_sample(raw_key)     # first row + a few data rows

    drift = classify_drift(contract["columns"], header)
    action = decide_action(drift, headered=True)

    if action == "QUARANTINE":
        set_state(conn, raw_key, "rejected", reason=f"schema_drift:{drift['kind']}")
        alert_feed_owner(raw_key, drift)
        return "QUARANTINE"

    if action == "EVOLVE":
        evolve_target(conn, target, drift["added"])
        bump_contract_version(conn, feed, drift["arrived_fp"])

    # Type-drift guard: even when the header matches, values can drift.
    type_violations = check_types(sample, contract["types"])
    if type_violations:                                  # e.g. 'N/A' in an int column
        set_state(conn, raw_key, "rejected", reason=f"type_drift:{type_violations}")
        alert_feed_owner(raw_key, {"type": type_violations})
        return "QUARANTINE"

    rows, rejects = parse_csv_strict(raw_key, expected_cols=header)
    load_to_warehouse(rows, target)
    save_rejects(conn, feed, raw_key, rejects)
    return "LOAD"

-- Versioned contract as data (reviewed like code on every bump)
CREATE TABLE feed_contract (
    feed        TEXT        NOT NULL,
    version     INT         NOT NULL,
    columns     JSONB       NOT NULL,   -- ordered column names
    types       JSONB       NOT NULL,   -- name -> expected type
    fingerprint CHAR(16)    NOT NULL,
    active      BOOLEAN     NOT NULL DEFAULT true,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (feed, version)
);

Step-by-step trace.

Arrival	header check	type check	action
unchanged	match	pass	LOAD
adds promo_code (trailing)	added	pass	EVOLVE → LOAD
amt → amount	renamed	(not reached)	QUARANTINE
int column has 'N/A'	match	fails	QUARANTINE

After deployment, an unchanged file fast-paths to load; a partner adding a trailing promo_code triggers an additive ALTER TABLE ADD COLUMN promo_code TEXT NULL, a contract bump, and a load — no human needed. A rename fingerprints as renamed_or_swapped and quarantines with an alert. The nasty case — a header that matches but an integer column carrying 'N/A' — is caught by the separate type check on sampled values, because header-level drift detection alone cannot see value-level type drift. Nothing corrupt loads; every quarantine keeps the immutable raw for reprocessing after review.

Output:

Requirement	Mechanism	Result
additive changes auto-flow	fingerprint + evolve router	EVOLVE → LOAD
renames caught	classifier → quarantine	held + alert
type drift caught	value-level type check	held + alert
nothing corrupt loads	deny-by-default policy	safe
reviewable history	versioned feed_contract	audit trail

Why this works — concept by concept:

• Versioned contract as data — the expected columns, types, and fingerprint live in feed_contract, reviewed on every bump like code. "Expected schema" is no longer hidden in the parser; it is an auditable, queryable record.
• Header fingerprint detection — a 16-char hash of the normalised, ordered header detects any structural change on arrival, before a row loads, and gives on-call a stable identifier for each distinct shape.
• Evolve/quarantine/fail router — the policy matrix auto-loads only fingerprint-matches and additive changes; renames, drops, reorders-without-header, and type changes quarantine. Deny-by-default is the backstop.
• Separate value-level type check — header drift detection is blind to a matching header with drifted values ('N/A' in an int column), so an independent type check on sampled rows catches the type drift that the fingerprint cannot. Two layers, two failure surfaces.
• Cost — one header hash and a small sampled type check per file (O(1) in the row count for the sample), plus an occasional ALTER TABLE ADD COLUMN on genuine additive drift. Trivial against the cost of silently loading mismatched or mistyped columns into a warehouse other teams trust.

Validation
Topic — data-validation
Data-validation and schema-check problems

Practice →

Transformation Topic — data-transformation Data-transformation problems on evolving schemas

Practice →

---

5. Late-arriving and partial files

A file is not "done" because it exists — prove completeness, detect truncation, reopen late windows without double-loading

The mental model in one line: a file's mere presence in the landing zone proves nothing — partial files (truncated transfers, still-uploading files, or a partner who sent one of three expected files) and late-arriving files (a "daily" feed that lands two days after its SLA) are the two completeness-and-timing failures, and the senior design proves a file is whole before loading (trailer-record count, manifest checksum, byte-size floor), detects truncation explicitly, and reopens a late file's processing window while an idempotent file-hash ledger guarantees a re-send is deduped, never double-counted. Completeness is asserted, not assumed; lateness is absorbed, not ignored.

The four axes for late and partial files.

• Completeness proof. How do you prove the file is whole? A control/trailer record carrying the data-row count, a manifest checksum (sha256), and a minimum byte-size floor. All three are independent evidence; the strongest designs require the trailer count and the checksum to agree before loading.
• Partial / truncation detection. How do you catch a short file? An EOF before the trailer record, a data-row count below the trailer's claim, or a checksum mismatch. A still-uploading file (no trigger) is caught earlier by the landing gate; a truncated download is caught by the byte-count check; a truncated upload with a trailer is caught by the count mismatch.
• Timing / SLA. When should the file have arrived, and what happens if it does not? An expected-arrival window turns a missing file into an alert (not a silent gap) and a late file into a reopen of the processing window rather than a dropped batch.
• Idempotent reprocessing. How do you reprocess a late or re-sent file without double-loading? The content-hash ledger (section 1) plus a load target keyed idempotently, so reopening a window and reloading is a no-op for already-seen content.

Completeness mechanisms — require more than one.

• Trailer / control record. The last line is a control record (T|<row_count> or an EDI SE/GE/IEA with counts). Assert parsed_data_rows == trailer_count. This catches a truncated file that lost rows before the trailer — if the trailer itself survived; combine with a checksum for the case where the trailer is also lost.
• Manifest checksum. A companion manifest (file.manifest.json) lists the expected filename, byte size, row count, and sha256. Recompute the hash on the landed file and compare. A checksum mismatch means the bytes are not what the sender intended — truncated, corrupted, or wrong file.
• Byte-size floor. The smallest plausible size (header + trailer + one row). A file at or near the floor with a claimed count of thousands is obviously truncated. Cheap first-pass sanity check.
• Row-count tolerance band. For feeds without a trailer, an expected count range (e.g. "10k–200k rows for a normal day") flags a suspiciously tiny or huge file for review. Weaker than a trailer but better than nothing.

Partial-file detection — the truncation cases.

• Still uploading. No completion trigger / size still changing → caught by the landing gate (section 2). Never reaches the completeness check.
• Truncated download. Local byte count < remote size → caught by the SFTP verify step (section 2), retried.
• Truncated upload with trailer intact. Data rows < trailer count → caught here by the count assertion. HOLD_PARTIAL, retry/backfill.
• Truncated upload, trailer lost too. No trailer record found where expected, or checksum mismatch → HOLD_PARTIAL. This is why you need both a trailer and a checksum — either alone has a blind spot.

Late files and windowing.

• SLA window. Each feed declares an expected-by time. A file absent past the window fires a "missing feed" alert — the failure you must surface, because a silently missing daily file is invisible until a report is wrong.
• Watermark reopen. Late data belongs to its logical date (the business date in the file/filename), not its arrival date. A file for 2026-08-16 landing on the 18th reopens the 2026-08-16 partition, reprocesses it, and lets downstream incremental jobs pick up the correction via their watermark.
• Idempotent by hash. Reopening and reloading must be safe. Because the ledger dedupes on content hash and the load target is keyed idempotently (partition + business key), a late file that overlaps an already-loaded batch loads only the genuinely new content.

Common interview probes on late/partial files.

• "How do you know you got the whole file?" — trailer row count and manifest checksum must agree; plus a byte-size floor.
• "How do you detect a truncated file?" — data rows < trailer count, or checksum mismatch, or missing trailer.
• "A daily file is two days late — now what?" — reopen the file's logical-date window; downstream watermarks absorb the correction.
• "How do you avoid double-loading a re-sent file?" — content-hash ledger + idempotent load key.

Worked example — a completeness gate (trailer count + checksum + floor)

Detailed explanation. The canonical completeness gate: parse the file, split off the trailer record, assert the data-row count matches the trailer, recompute and compare the manifest checksum, and enforce a byte-size floor — loading only when all three agree. Build the gate.

• Trailer. Last line T|<count>; assert count == len(data_rows).
• Checksum. Recompute sha256; compare to manifest.
• Floor. Reject files smaller than header+trailer+1 row.

Question. Write a completeness gate that returns COMPLETE or a specific failure.

Input.

Check	Source of truth	Failure verdict
row count	trailer `T	`
checksum	manifest sha256	checksum mismatch
size floor	min plausible bytes	too small

Code.

# completeness.py — prove a file is whole before loading
import hashlib, json, os

def sha256_of(path: str) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()

def check_complete(data_path: str, manifest_path: str, size_floor: int = 128) -> tuple[str, str]:
    # 0. Byte-size floor — cheapest sanity check
    if os.path.getsize(data_path) < size_floor:
        return ("INCOMPLETE", f"size {os.path.getsize(data_path)} < floor {size_floor}")

    manifest = json.load(open(manifest_path))

    # 1. Checksum — the bytes are exactly what the sender intended
    actual = sha256_of(data_path)
    if actual != manifest["sha256"]:
        return ("INCOMPLETE", f"checksum {actual[:12]} != manifest {manifest['sha256'][:12]}")

    # 2. Trailer row count — the record count agrees
    with open(data_path, encoding="utf-8-sig") as f:
        lines = [ln.rstrip("\r\n") for ln in f if ln.strip()]
    if not lines or not lines[-1].startswith("T|"):
        return ("INCOMPLETE", "missing trailer record")
    trailer_count = int(lines[-1].split("|", 1)[1])
    data_rows = len(lines) - 2                       # minus header, minus trailer
    if data_rows != trailer_count:
        return ("INCOMPLETE", f"data rows {data_rows} != trailer {trailer_count}")

    return ("COMPLETE", f"{data_rows} rows verified")

Step-by-step explanation.

1. The byte-size floor runs first because it is the cheapest check and catches the most obvious truncation — a file smaller than header + trailer + one row cannot possibly be a full day's feed. It short-circuits before any hashing or parsing.
2. The checksum comparison is the strongest single check: recomputing the sha256 of the landed file and comparing it to the manifest proves the bytes are exactly what the sender hashed. A truncated transfer, a corrupted byte, or a wrong file all fail here — and this check works even when the trailer record was itself lost to truncation.
3. The trailer-count check is independent evidence: it splits off the T|<count> control record and asserts the parsed data-row count matches the claimed count. This catches the case where a file's content was truncated but the sender's manifest checksum is stale or absent. Requiring both checksum and trailer closes the blind spot each has alone.
4. A missing trailer (not lines[-1].startswith("T|")) is itself an incompleteness signal — if the file was truncated mid-transfer, the trailer (which is last) is the first thing lost, so its absence is a strong truncation indicator.
5. Only when the size floor, the checksum, and the trailer count all agree does the gate return COMPLETE. Any failure returns a specific reason so the pipeline can HOLD_PARTIAL with an actionable message. Completeness is proven by convergent independent evidence, not assumed from the file's existence.

Output.

File state	floor	checksum	trailer	verdict
whole file	ok	match	count ok	COMPLETE
truncated (rows lost)	ok	mismatch	count low	INCOMPLETE
truncated (no trailer)	ok	mismatch	missing	INCOMPLETE
wrong file sent	ok	mismatch	maybe ok	INCOMPLETE

Rule of thumb. Prove completeness with independent evidence — a byte-size floor, a manifest checksum, and a trailer row count — and load only when all three agree. A trailer alone misses a lost-trailer truncation; a checksum alone misses a stale manifest. Together they leave no blind spot.

Worked example — reopening a late file's logical-date window

Detailed explanation. A late file belongs to its business date (the logical date in the file or filename), not its arrival date. Loading it into the arrival date's partition would misattribute the data; the correct move is to reopen the business-date partition, reprocess, and let downstream incremental jobs pick up the change via their watermark. Build the late-handling logic.

• Logical date. Extracted from the filename/trailer, not now().
• Reopen. Load into the logical-date partition, marking it for downstream re-read.
• Watermark. Downstream jobs re-process partitions whose updated_at advanced.

Question. Write the late-file handler that attributes a file to its business date and reopens that window.

Input.

File	Business date	Arrival date	Action
acme_20260816.csv	2026-08-16	2026-08-18	reopen 08-16
acme_20260818.csv	2026-08-18	2026-08-18	normal load

Code.

# late_window.py — attribute to business date; reopen the partition
import re
from datetime import datetime, date

def business_date_from_name(filename: str) -> date:
    m = re.search(r"(\d{8})", filename)
    if not m:
        raise ValueError(f"no YYYYMMDD business date in {filename}")
    return datetime.strptime(m.group(1), "%Y%m%d").date()

def load_to_partition(conn, rows: list[dict], target: str, business_day: date) -> None:
    with conn.cursor() as cur:
        # Idempotent: delete-then-insert this file's rows within the partition,
        # or MERGE on the business key. Reloading the same content is a no-op.
        cur.execute(f"""
            DELETE FROM {target}
            WHERE  business_date = %s AND source_file_hash = %s
        """, (business_day, rows[0]["_file_hash"]))
        insert_rows(cur, target, rows)
        # Bump the partition watermark so downstream incremental jobs re-read it.
        cur.execute("""
            INSERT INTO partition_watermark(target, business_date, updated_at)
            VALUES (%s, %s, now())
            ON CONFLICT (target, business_date)
              DO UPDATE SET updated_at = now()
        """, (target, business_day))
    conn.commit()

def handle_file(conn, raw_key: str, rows: list[dict], target: str) -> str:
    business_day = business_date_from_name(raw_key)
    today = datetime.utcnow().date()
    load_to_partition(conn, rows, target, business_day)
    return "LATE_REOPEN" if business_day < today else "ON_TIME"

Step-by-step explanation.

1. business_date_from_name extracts the logical date from the filename's YYYYMMDD — the file for the 16th belongs to the 16th's partition even if it lands on the 18th. Attributing by arrival date would put the 16th's settlements in the 18th's numbers, silently corrupting daily totals.
2. load_to_partition writes into the business-date partition and does so idempotently: it deletes any existing rows for this partition-and-source-file-hash, then inserts. Reprocessing the identical late file is therefore a no-op (delete removes exactly what insert re-adds), and a corrected re-send (different hash) replaces cleanly. This is what makes reopening safe.
3. The partition watermark is bumped on every load. Downstream incremental jobs read partition_watermark and reprocess any partition whose updated_at advanced — so reopening the 16th automatically propagates the correction through aggregates, models, and reports without a manual backfill trigger.
4. handle_file classifies the load as LATE_REOPEN (business date in the past) or ON_TIME (today). The classification feeds monitoring: a spike in LATE_REOPEN for a feed signals an upstream delivery problem worth chasing with the partner.
5. Crucially, the on-time path and the late path use the same idempotent partition load — lateness is not a special corrupt path but the normal path pointed at an older partition. Because the ledger (section 1) already deduped by content hash upstream, and the partition load is idempotent by hash, a late file that overlaps previously-loaded data contributes only genuinely new or corrected rows.

Output.

File	business date	verdict	partition touched
acme_20260816.csv (late)	2026-08-16	LATE_REOPEN	2026-08-16
acme_20260818.csv	2026-08-18	ON_TIME	2026-08-18
re-send of 08-16 (same hash)	2026-08-16	LATE_REOPEN	no-op (deduped)

Rule of thumb. Attribute a file to its business date, not its arrival date, and reopen that partition idempotently (delete-by-file-hash then insert, or MERGE). Bump a partition watermark so downstream jobs re-read the correction automatically. Lateness is the normal load pointed at an older window — never a second, divergent code path.

Worked example — a missing-feed SLA alert

Detailed explanation. The failure juniors never guard against is the file that never arrives. A silently missing daily feed is invisible until a downstream report is wrong days later. The fix is an expected-arrival monitor per feed: if no file for the expected business date has landed by the SLA time, fire an alert. Build the monitor.

• Expectation. Each feed declares expected_by (a time) and a schedule (daily/weekdays).
• Check. At/after expected_by, is there a loaded file for today's business date?
• Alert. If not, page the on-call and the feed owner.

Question. Write the SLA monitor that alerts on a missing feed.

Input.

Feed	Schedule	expected_by (UTC)
acme_payroll	weekdays	06:00
globex_settle	daily	04:30

Code.

# sla_monitor.py — alert when an expected file has not landed by its SLA
from datetime import datetime, date, time, timezone

def business_dates_due(feed: dict, today: date) -> bool:
    if feed["schedule"] == "weekdays":
        return today.weekday() < 5          # Mon-Fri
    return True                             # daily

def feed_loaded_for(conn, feed_name: str, business_day: date) -> bool:
    with conn.cursor() as cur:
        cur.execute("""
            SELECT 1 FROM ingest_ledger
            WHERE  filename LIKE %s
              AND  loaded_at::date >= %s
        """, (f"{feed_name}%", business_day))
        return cur.fetchone() is not None

def check_slas(conn, feeds: list[dict]) -> list[dict]:
    now = datetime.now(timezone.utc)
    today = now.date()
    breaches = []
    for feed in feeds:
        if not business_dates_due(feed, today):
            continue
        sla = datetime.combine(today, time.fromisoformat(feed["expected_by"]), tzinfo=timezone.utc)
        if now >= sla and not feed_loaded_for(conn, feed["name"], today):
            breaches.append({"feed": feed["name"], "expected_by": feed["expected_by"],
                             "status": "MISSING"})
    for b in breaches:
        page_oncall(f"Feed {b['feed']} missing past SLA {b['expected_by']} UTC")
    return breaches

Step-by-step explanation.

1. business_dates_due encodes each feed's calendar — a weekday-only payroll feed is not expected on Saturday, so the monitor must not alert then. Getting the schedule right prevents alert fatigue that trains on-call to ignore the monitor.
2. feed_loaded_for checks the ledger (the source of truth for "did we load it"), not the landing directory — a file that landed but failed a gate is not "loaded" and should still count as missing for SLA purposes. Keying on the filename prefix and today's date answers "has today's file for this feed been loaded?"
3. The SLA time is combined with today's date into a timezone-aware instant. The monitor only evaluates a feed once now >= sla — before the deadline, a not-yet-arrived file is normal, not a breach.
4. A feed that is due, past its SLA, and not loaded is a MISSING breach and pages on-call and (in a fuller implementation) the feed owner. This is the alert that converts a silent gap into an actionable incident while there is still time to chase the partner before the business close.
5. The monitor returns the breach list for dashboards and reporting, so a feed that is chronically late shows up as a pattern, not just a one-off page — feeding the conversation with the partner about their delivery reliability.

Output.

Feed	due today?	past SLA?	loaded?	verdict
acme_payroll (Tue)	yes	yes	no	MISSING → page
acme_payroll (Sat)	no	—	—	skip
globex_settle	yes	yes	yes	OK

Rule of thumb. Monitor for the file that never comes. Per feed, declare an expected-by SLA and a schedule, check the ledger (not the directory) after the deadline, and page when a due feed is unloaded. A silently missing daily file is invisible until a report is wrong — the SLA alert is what makes absence loud.

Senior interview question on late and partial files

A senior interviewer might ask: "Your daily settlement feed sometimes arrives truncated, sometimes two days late, and sometimes the partner re-sends a corrected version. Design completeness verification, partial-file detection, and late-window handling so a truncated file is never loaded, a late file lands in the correct business-date partition, a re-send never double-counts, and a feed that never arrives pages someone."

Solution Using a trailer+checksum gate + business-date reopen + hash ledger + SLA monitor

# late_partial_pipeline.py — completeness, lateness, and idempotency together
def ingest_settlement_file(conn, raw_key: str, manifest_path: str, target: str) -> str:
    filename = raw_key.rsplit("/", 1)[-1]
    digest = sha256_of_landed(raw_key)

    # 1. Idempotency — identical re-send is a no-op
    if already_loaded(conn, filename, digest):
        return "SKIP"

    # 2. Completeness — trailer count + checksum + floor must all agree
    status, detail = check_complete(local_copy(raw_key), manifest_path)
    if status != "COMPLETE":
        set_state(conn, raw_key, "rejected", reason=f"incomplete:{detail}")
        return "HOLD_PARTIAL"                     # retry / backfill later

    # 3. Parse and stamp each row with the file hash (for idempotent partition load)
    rows, rejects = parse_csv_strict(local_copy(raw_key), expected_cols=contract_cols(conn))
    for r in rows:
        r["_file_hash"] = digest
        r["business_date"] = business_date_from_name(filename)
        r["source_file_hash"] = digest

    # 4. Late-aware, idempotent partition load (business date, not arrival date)
    load_to_partition(conn, rows, target, business_date_from_name(filename))
    record_loaded(conn, filename, digest, len(rows))
    save_rejects(conn, "settlement", raw_key, rejects)
    return "LATE_REOPEN" if business_date_from_name(filename) < today_utc() else "LOAD"

-- Idempotent partition load target: unique on (business_date, natural key)
-- so a reopened/late/re-sent file MERGEs rather than appends duplicates.
CREATE TABLE raw.settlement (
    business_date    DATE     NOT NULL,
    txn_id           TEXT     NOT NULL,
    amount_cents     BIGINT   NOT NULL,
    source_file_hash CHAR(64) NOT NULL,
    loaded_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (business_date, txn_id)         -- dedupe on reopen
);

Step-by-step trace.

Scenario	completeness	idempotency	partition	verdict
whole, on-time	passes	new hash	today	LOAD
truncated	fails (count/checksum)	—	—	HOLD_PARTIAL
two days late	passes	new hash	business date	LATE_REOPEN
identical re-send	(not reached)	seen hash	—	SKIP
corrected re-send	passes	new hash	business date	LATE_REOPEN (MERGE)
never arrives	—	—	—	SLA page (monitor)

After deployment, a whole on-time file passes the trailer+checksum gate, is new to the ledger, and loads into today's partition. A truncated file fails completeness and is held as partial for retry — never loaded. A file for the 16th arriving on the 18th passes completeness and loads into the 16th's partition, reopening it (PRIMARY KEY (business_date, txn_id) dedupes overlap on MERGE); the partition watermark bump propagates the correction downstream. An identical re-send is skipped by the hash ledger; a corrected re-send (new hash) MERGEs into the business-date partition, replacing prior rows by key. And the section-5 SLA monitor pages if the file never shows.

Output:

Requirement	Mechanism	Result
truncated never loads	trailer + checksum + floor gate	HOLD_PARTIAL
late → correct partition	business-date attribution	LATE_REOPEN
re-send never double-counts	hash ledger + PK MERGE	SKIP / clean replace
downstream picks up fix	partition watermark bump	auto reprocess
missing feed is loud	SLA monitor page	incident raised

Why this works — concept by concept:

• Convergent completeness proof — the trailer count, the manifest checksum, and the size floor are three independent witnesses; requiring all three closes the blind spots each has alone (a lost trailer, a stale manifest, an obviously tiny file), so a truncated file cannot slip through.
• Business-date attribution — attributing by the logical date in the filename, not arrival time, keeps a late file's data in the right window; a two-day-late settlement lands in its own day's totals, not today's.
• Idempotent partition load — PRIMARY KEY (business_date, txn_id) plus delete-by-file-hash/MERGE makes reopening a window safe: overlapping rows dedupe, corrected rows replace, and reprocessing identical content is a no-op.
• Content-hash ledger + watermark bump — the ledger dedupes whole-file re-sends, while the partition watermark propagates any reopen to downstream incremental jobs automatically, so a correction flows through without a manual backfill.
• SLA monitor for absence — completeness and idempotency handle the files that arrive; the SLA monitor handles the file that never does. Together they cover both "the file is wrong" and "the file is missing" — the two failure classes juniors forget.
• Cost — one checksum (O(bytes)), one trailer parse (already parsing anyway), one indexed ledger lookup, and one MERGE keyed on the primary key. All cheap; the eliminated cost is a truncated file corrupting a close, a late file landing in the wrong day, or a re-send double-counting revenue.

Processing
Topic — data-processing
Data-processing problems on completeness and dedupe

Practice →

Validation
Topic — validation
Validation problems on file completeness checks

Practice →

---

Cheat sheet — flat-file ingestion recipes

• The four axes, one sentence. Flat-file ingestion is turning an untrusted dropped file into safe rows across four axes: transport & landing (how it arrives, how you know it is fully written), format & parsing (CSV quoting, fixed-width offsets, EDI envelopes), schema stability (drift detection + evolve/quarantine/fail policy), and completeness & timing (trailer count + checksum, late-window reopen). Answer all four on paper before writing a parser; the file is untrusted until every gate passes.
• Atomic landing template. Never read a file until it is fully written — the sender writes file.part and renames on completion, or drops a separate file.ok/manifest trigger; last resort is polling until size + mtime stop changing. Land raw bytes write-once under a dated key (raw/<partner>/<yyyy>/<mm>/<dd>/<file>); track lifecycle (landed/loaded/rejected) as append-only state rows, never by moving files between folders.
• Idempotent pickup. Key the processed-file ledger on (filename, content_sha256) — an identical re-send is SKIP, a corrected re-send under the same name (different hash) is a new LOAD, and a crash-then-retry never double-loads. For PGP feeds, decrypt first and hash the plaintext (ciphertext is non-deterministic and would break dedupe). Verify SFTP download byte count against the remote stat size.
• Defensive CSV reader flags. Use an RFC-4180 reader (csv.reader, not str.split) with quotechar='"', doublequote=True, strict=True; open as utf-8-sig to strip a BOM from the first header cell; assert len(cells) == len(header) and reject ragged rows to a sink with their line number — never pad, truncate, or silently drop. Empty fields are data; wrong field counts are structural errors.
• Fixed-width crib. Slice by absolute byte offset per a versioned layout; strip padding by justification (numbers right/zero-padded, text left/space-padded); reject any line shorter than the record length; and validate a strict-format canary field (a YYYYMMDD date or a check digit) — a one-byte shift parses cleanly and loads wrong, so the canary is your only shift detector.
• EDI (X12) envelope crib. Read the delimiters from the file: the ISA segment declares the element separator (byte 3) and segment terminator (byte 105). Walk the envelope tree ISA → GS → ST → segments → SE/GE/IEA; capture control numbers (ISA13, GS06, ST02) — they are your dedupe key and the basis of the 997/999 functional acknowledgement. EDIFACT uses UNB/UNG/UNH. EDI is a grammar, not a table.
• Schema-drift fingerprint + policy matrix. Fingerprint the normalised, ordered header (sha256 of name\x1ftype list) and diff against a versioned contract stored as data. Policy: fingerprint-match → LOAD; new trailing columns on a headered feed → EVOLVE (add nullable) → LOAD; reorder with header → LOAD (map by name); reorder without header / rename / drop / type-change → QUARANTINE. Default for anything unrecognised = QUARANTINE (deny-by-default). Add a separate value-level type check — a matching header can still carry drifted values ('N/A' in an int column).
• Headerless feeds need a stricter contract. Position is meaning and there is no name to map by, so pin the exact record length and byte offsets in the contract, validate every field's domain (regex), and keep a canary. Two independent structural checks (length + canary) catch the silent misalignment a headerless file cannot announce.
• Completeness gate. Prove wholeness with independent evidence: a byte-size floor (cheap first pass), a manifest checksum (bytes are exactly what the sender intended — survives a lost trailer), and a trailer/control-record row count (data_rows == trailer_count — survives a stale manifest). Load only when all agree; a missing trailer is itself a truncation signal.
• Partial-file detection. Still-uploading → caught by the landing trigger; truncated download → caught by the byte-count verify; truncated upload with trailer → caught by count mismatch; truncated upload without trailer → caught by checksum mismatch. This is why you need both a trailer and a checksum — either alone has a blind spot. Failed completeness → HOLD_PARTIAL, retry/backfill; never load.
• Late files + windowing. Attribute a file to its business date (the YYYYMMDD in the name/trailer), not its arrival date; reopen that partition idempotently (delete-by-file-hash then insert, or MERGE on (business_date, natural_key)); bump a partition watermark so downstream incremental jobs re-read the correction automatically. Lateness is the normal load pointed at an older window — never a second, divergent code path.
• Monitor for absence. The file that never arrives is invisible until a report is wrong. Per feed, declare an expected-by SLA and a schedule (daily/weekdays/holidays), check the ledger (not the directory) after the deadline, and page on-call + the feed owner when a due feed is unloaded. Track chronic lateness as a pattern for the partner conversation.

Frequently asked questions

What is flat-file ingestion in one sentence?

Flat-file ingestion is the process of turning an untrusted file that a partner drops on you — typically over SFTP, in CSV, fixed-width, or EDI format — into rows you can safely load into a warehouse, by proving the file is fully written, correctly parsed, structurally as expected, and actually complete before a single row lands. Unlike an API or a stream, a flat file is a one-way, fire-and-forget artifact with no schema negotiation, no delivery handshake, and no retry protocol, so the receiver absorbs every failure mode: half-written files, quoting hazards, schema drift, truncation, and late delivery. The senior discipline is to treat the file as adversarial input and gate it through landing, parsing, drift, and completeness checks — the file is not "data" until it has passed all four. It is one of the most under-invested-in yet load-bearing pipelines in data engineering, because it still moves payroll, banking, insurance, and B2B EDI traffic that predates and outlasts REST.

Why is CSV parsing so error-prone?

CSV parsing looks trivial and is not, because "comma-separated values" hides a real grammar. A field can legally contain the delimiter itself ("Smith, Jr."), a literal newline ("123 Main St\nApt 4"), or a double-quote escaped by doubling it ("He said ""hi"""), so any parser built on line.split(",") or line-by-line reading corrupts these rows — usually by silently shifting every column after the offending field, which then loads wrong without raising an error. On top of the quoting rules sit encoding hazards: a UTF-8 byte-order mark attaches to the first header name (turning id into id and breaking key lookups), and legacy feeds arrive in Latin-1 or Windows-1252 that mis-decode under a UTF-8 assumption. The fix is to use an RFC-4180-compliant reader (Python's csv module, not string splitting), read with utf-8-sig to strip the BOM, and validate each row's field count against the header — rejecting ragged rows with their line number rather than padding or dropping them. CSV is error-prone precisely because the happy-path sample always works and the edge cases only appear in production.

How do you make SFTP file pickup idempotent?

Idempotent SFTP pickup means re-running the job — or a partner re-sending a file — never double-loads. The mechanism is a processed-file ledger keyed on both the filename and the content hash (sha256): before loading, you look up (filename, hash); if it is present you skip, otherwise you load and record it. Keying on filename alone is insufficient because a partner often re-sends a corrected file under the same name — that must load — while an identical re-send must be skipped; the content hash distinguishes the two. For PGP-encrypted feeds you must decrypt first and hash the plaintext, because PGP uses a fresh session key per encryption, so the same data re-encrypted produces different ciphertext that would look new every time. Two further guards complete the picture: verify the downloaded byte count against the remote stat size (so a truncated download is not mistaken for a complete file), and land raw bytes immutably under a dated key so every reprocess reads identical input. With these, a crash between load and ledger-write is safe because the load target itself is keyed idempotently.

What is schema drift and how do you handle it?

Schema drift is the inevitability that the sender changes the file's shape without telling you — adds a column, drops one, reorders two, renames a field, or starts sending a string where you expected an integer. You cannot prevent it (the partner owns the schema), so the discipline is to detect it on arrival and handle it by policy. Detection is a schema fingerprint: hash the normalised, ordered header (and types if known) and compare it to a versioned contract stored as data, on every file, before any row loads. Handling is a policy matrix keyed on the kind of change: a new trailing column on a header-mapped feed is additive and can evolve automatically (add the column as nullable, bump the contract); a reorder is safe only if you map by name (headered feeds) and dangerous otherwise; and renames, drops, type changes, and reorders on headerless feeds are quarantined for human review rather than loaded. The default for anything the matrix does not explicitly bless is quarantine — deny-by-default. A separate value-level type check catches the nasty case where the header matches but the values drifted (an 'N/A' in an integer column), which header fingerprinting alone cannot see.

How do you detect a partial or still-uploading file?

There are four distinct partial-file cases and each has its own detector. A still-uploading file is caught at the landing gate — you never read a data file until its completion signal exists (a temp-name-then-rename by the sender, or a separate .ok/manifest trigger), so a file mid-upload is invisible to the poller. A truncated download (a dropped SFTP session) is caught by verifying the downloaded byte count against the remote stat size and retrying on mismatch. A truncated upload whose trailer survived is caught by a control/trailer record: assert the parsed data-row count equals the trailer's claimed count. And a truncated upload that lost the trailer too is caught by a manifest checksum mismatch (or by the trailer simply being absent where it should be). This is exactly why the strongest completeness gate requires both a trailer count and a checksum plus a byte-size floor — each mechanism has a blind spot the others cover. A file that fails any completeness check is held as partial (HOLD_PARTIAL) and retried or backfilled; it is never loaded.

How do you handle late-arriving files without double-loading?

A late-arriving file is a "daily" feed that lands after its SLA — sometimes days late, sometimes as a correction to an earlier file. The two rules are: attribute by business date, and reprocess idempotently. Attribution means the file belongs to the logical date encoded in its name or trailer (YYYYMMDD), not its arrival time — so a file for the 16th that lands on the 18th loads into the 16th's partition, keeping daily totals correct rather than misattributing the data to today. Idempotency means reopening that partition is safe to repeat: load via a MERGE (or delete-by-file-hash then insert) on a natural key like (business_date, txn_id), so an overlapping or re-sent file dedupes rather than appends duplicates, and reprocessing identical content is a no-op. Bumping a partition watermark on the reopen lets downstream incremental jobs re-read the corrected window automatically, so the fix propagates through aggregates and reports without a manual backfill. Underpinning both is the content-hash ledger from landing: an identical re-send is skipped outright, while a corrected re-send (new hash) flows through and replaces prior rows by key. Separately, an SLA monitor pages when a due file never arrives — the failure that is otherwise invisible until a report is wrong.

Practice on PipeCode

• Drill the ETL practice library → for the batch-ingestion, landing-zone, watermark, and idempotent-load problems that senior file-intake interviews probe.
• Sharpen your parsers on the CSV parsing practice library → for the quoting, embedded-delimiter, encoding, and ragged-row edge cases that break naive readers.
• Harden your input checks on the data-validation practice library → for the schema-drift, completeness, and type-check patterns that keep corrupt files out of the warehouse.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the four-axis file-ingestion checklist against real graded inputs.

Lock in flat-file ingestion muscle memory

Docs explain formats. PipeCode drills explain the decision — when a file is safe to read, why CSV quoting breaks a naive parser, when schema drift should evolve versus quarantine, and how a trailer count plus a checksum prove a file is whole before it touches the warehouse. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs senior data engineers actually face.

Practice ETL problems →
Practice CSV parsing problems →