Columnar results, and the ingest bottleneck that wasn’t the network

Part 3 of a series on the internals of a distributed image-processing platform.

This post tells two stories that turn out to teach the same lesson. Measure the real cost, then let the measurement pick the design. One story is about how results leave the system. The other is about how data comes into it, and a slow spot that turned out to be a lock nobody had thought to look at.

Story one. Results are stored as Parquet, and making them so costs almost nothing

Task results get consumed by two very different audiences. Machine-learning training tools want one shape of data, and SQL query engines want another. If you pick a single generic format like JSON, every consumer has to convert it first, and you throw away the speed that comes from a format each tool reads natively. Training tools like tf.data want TFRecord. Query engines like Spark, Athena, and DuckDB want columnar Parquet.

So the committed result path writes Parquet. Each task stages a small single-row-group Parquet object, and the coordinator promotes it into the results tree using the exactly-once commit from Part 1. The nice outcome is that a finished job is queryable right away with no extra load step. You can point a query engine straight at it.

SELECT shard, images_processed, bytes_read
FROM 'results/job-abc/*.parquet'
WHERE images_processed > 0;

No conversion job, no separate warehouse import. The bytes the worker staged are the thing you query.

Why this change was so safe to make

Remember from Part 1 that the staged object is write-only as far as the platform is concerned. Nothing in the commit path ever reads it back and tries to understand it. The coordinator just copies it, byte for byte, from the staging spot to its final spot. It does not care what is inside.

That single fact is what made switching the staged format from JSON to columnar Parquet almost risk-free. The worker now builds a one-row Parquet file instead of a JSON blob, and the commit path did not change at all, because it never looked inside the object in the first place. Here is the staging write.

// internal/worker, the staging write, simplified
var body bytes.Buffer
_ = formats.WriteResultsParquet(&body, []formats.ResultRow{{
    TaskID: a.TaskID,
    Shard: a.Shard,
    OutputKey: finalKey,
    ImagesProcessed: processed,
    BytesRead: totalBytes,
}})
_ = w.store.Put(ctx, stagingKey, bytes.NewReader(body.Bytes()),
    "application/vnd.apache.parquet")

One row group per task matches the “one result object per task” design exactly, and the query engines happily read across the whole tree of Parquet files. The result names even carry the format in their extension now, ending in .parquet, and nothing broke, because the tests refer to the name helpers by calling them rather than hard-coding a .json string to compare against.

The honest bit

The formats package also has working, tested writers for TFRecord, for WebDataset tar shards, and for Apache Arrow batches. But only Parquet is actually wired into a live path right now. The others are ready to plug into the same stage-then-commit path as their consumers show up, for example a real tf.data training job or a PyTorch data loader.

I want the docs to say that plainly rather than let “supports five formats” imply five live producers. It is one live producer and four honest, tested writers. The value of the setup is that adding the next live format is a worker-side change against a commit path that already does not care what the bytes are. But the map is not the territory, and the writeup says so.

Story two. 91 percent of ingest time was a lock

Ingest walks a local folder, and for each image it does four things. It opens and stats the file. It checksums the file with SHA-256. It uploads the file to the object store. And it records a metadata row in a small SQLite database. The natural assumption is that the upload dominates the time, since it is the one step that goes over the network and everything else is local.

That assumption was wrong. The only reason we know is that the pipeline is instrumented to attribute busy time to each of the four steps. Here is the struct that carries those numbers back.

// internal/ingestion/pipeline.go
type Progress struct {
    Processed int64
    Failed int64
    Bytes int64
    // Total busy time per step across all workers, in nanoseconds. This is the
    // relative cost breakdown of open+stat vs checksum vs upload vs DB insert,
    // which answers "where does ingest time actually go" without guessing.
    OpenNanos int64
    ChecksumNanos int64
    UploadNanos int64
    IndexNanos int64
}

The measurement was blunt and surprising. The SQLite insert step was about 91 percent of all ingest busy time. Not the upload. Not the hashing. The database write.

Two root causes, one obvious and one hidden in a driver

Cause one was a wrong pragma string in the database setup. The metadata store uses the modernc.org/sqlite driver, which expects settings written in a specific form. The code had been written using a different, older driver's form, which this driver silently ignores. The effect was that write-ahead logging never actually turned on, and the busy timeout was left at zero. So when two ingest workers tried to write at the same time, one got an instant "database is busy" error and dropped its row, instead of waiting a moment and retrying. The fix is one line, but you only find it by looking.

// internal/metadata/index.go
db, err := sql.Open("sqlite",
    path+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(30000)")

Write-ahead logging lets readers and a writer work at the same time, and a 30-second busy timeout means a bulk ingest of millions of files waits out write contention instead of dropping the occasional row.

Cause two was the design itself. Even with the settings fixed, SQLite allows only one writer at a time. If every one of the many upload workers writes its own row directly, they all fight over that single writer lock, and each one pays the cost of forcing its write to disk. The uploads run in parallel beautifully, but the writes line up one behind another, and that lining up was hidden behind a lock nobody was profiling.

The fix. Split the parallel part from the serial part

Redesign the pipeline so the two halves are separate. The upload workers do the part that parallelizes well, which is open, checksum, and upload, and then hand the finished record off to a single writer goroutine. They never touch SQLite themselves. Here is the shape.

// internal/ingestion/pipeline.go, inside IngestDir
for i := 0; i < p.workers; i++ {
    go func() {
        for job := range jobs {
            rec, n, t, err := p.uploadFile(ctx, job)
            // ... fold each step's timing into shared atomic counters ...
            if err != nil { failed.Add(1); continue }
            records <- indexReq{rec: rec, size: n}  // hand off, never write SQLite here
        }
    }()
}

// The single writer. This is the whole reason ingest stopped being 91 percent
// lock-wait. It drains records and commits them in batches.
go func() {
    batch := make([]metadata.DataRecord, 0, p.batchSize)
    flush := func() {
        // ... one transaction, one prepared statement, one disk flush per batch ...
        _ = p.idx.InsertBatch(ctx, batch)
        batch = batch[:0]
    }
    for req := range records {
        batch = append(batch, req.rec)
        if len(batch) >= p.batchSize { flush() }
    }
    flush()  // final partial batch
}()

Two good things happen here.

First, the single writer is now used by exactly one caller. SQLite’s one-writer rule stops being a fight, because there is only ever one goroutine holding the lock. The lock-wait that was 91 percent of the time is simply not in the path anymore.

Second, inserts are batched. InsertBatch groups a batch of records, 500 by default, into one transaction with one prepared statement, so the writer takes its lock and forces to disk once per batch instead of once per image. Spreading that one disk flush across 500 rows is the difference between a per-row and a per-batch durability cost. Here is that batch write.

// internal/metadata/index.go, one lock, one disk flush, whole batch atomic
func (idx *Index) InsertBatch(ctx context.Context, recs []DataRecord) error {
    tx, _ := idx.db.BeginTx(ctx, nil)
    stmt, _ := tx.PrepareContext(ctx, `INSERT OR REPLACE INTO records (...) VALUES (...)`)
    for _, r := range recs {
        if _, err := stmt.ExecContext(ctx /* ...12 columns... */); err != nil {
            tx.Rollback()
            return err
        }
    }
    return tx.Commit()  // the whole batch commits or rolls back together
}

The channel that carries records to the writer is buffered generously, so an upload worker never has to wait to hand off a finished record while the writer is mid-commit. And the shutdown order matters. First close the jobs channel, then wait for all uploads to finish, so no more records can be sent, then close the records channel so the writer drains what is left and flushes the final partial batch, then wait for the writer to report it is done. Get that order wrong and you either deadlock or lose the tail of the last batch.

The lesson both stories share

The result-format story and the ingest story are the same move made twice.

On the way out, do not pick a format from a gut feeling about what consumers want. Pick the one the query engines read natively, which is Parquet, and design the commit path so it never depends on the format, so the choice is cheap to make and cheap to change later.

On the way in, do not pick a bottleneck from a gut feeling about what is slow, which everyone assumed was the network. Instrument every step, and let the measurement point at the real culprit, which was a silently-ignored setting and a single-writer lock. The fix that follows, parallel uploads feeding one batched writer, is only obvious once the 91 percent number exists.

Both are the same discipline. Make the system tell you where the cost really is, then let that answer drive the design instead of your assumptions.

Links: Part 1: Exactly-once results in an AP system: two-phase commit, Raft, and the window you can’t close

Part 2: Bounded leases and bounded queues

Codebase: https://github.com/AbhiramDodda/distributed-image-processing-system