Transaction.Overwrite drops rows under concurrent writes — partitionedFanoutWriter race suspected

## Summary

Under concurrent producer load, `Transaction.Overwrite` produces a staged snapshot whose total row count is **less than** the input snapshot's row count, even though the operation should preserve rows. The losses are non-deterministic and small (typically 1k–10k rows on tables with millions of rows). They reproduce repeatably when a parallel Iceberg writer is committing into the same table while compaction runs.

The transaction reports success — the row loss is silent unless the caller does a post-stage row-count consistency check.

## Versions

- iceberg-go: v0.5.0
- arrow-go: v18.5.2-0.20260220015023
- Go: 1.24
- OS: macOS 25.2.0 (also reproduced on Linux)
- Storage backends tested: local fileblob (`gocloud.dev/blob/fileblob`)

## Symptom

A janitor-style compactor runs:

```go
tbl, _ := cat.LoadTable(ctx, ident)

scan := tbl.Scan(table.WithRowFilter(filter)) // partition-equality filter
schema, recIter, _ := scan.ToArrowRecords(ctx)
reader := newStreamingRecordReader(schema, recIter) // adapts iter.Seq2 → array.RecordReader
defer reader.Release()

tx := tbl.NewTransaction()
if err := tx.Overwrite(ctx, reader, nil, table.WithOverwriteFilter(filter)); err != nil {
return err
}
staged, _ := tx.StagedTable()

// Verify row preservation:
beforeRows := sumRowsAcrossManifests(tbl) // walks tbl.CurrentSnapshot() manifests
stagedRows := sumRowsAcrossManifests(staged.Table)
if beforeRows != stagedRows {
return fmt.Errorf("input rows %d != staged rows %d", beforeRows, stagedRows)
}
```

While a parallel goroutine commits new data to the same table at ~10 commits/sec via a separate `cat.LoadTable` + `tx.Append`, this verification fails on roughly 60% of compaction attempts:

```
input rows 5900000 != staged rows 5890013 (Δ -9987)
input rows 6500000 != staged rows 6499023 (Δ -977)
input rows 8000000 != staged rows 7989862 (Δ -10138)
input rows 8800000 != staged rows 8797027 (Δ -2973)
input rows 9800000 != staged rows 9789864 (Δ -10136)
input rows 13000000 != staged rows 12990059 (Δ -9941)
input rows 15600000 != staged rows 15596013 (Δ -3987)
input rows 18000000 != staged rows 17989981 (Δ -10019)
input rows 19000000 != staged rows 18996026 (Δ -3974)
input rows 24600000 != staged rows 24596019 (Δ -3981)
input rows 34200000 != staged rows 34192221 (Δ -7779)
```

`staged < before` in every case. Rows are being dropped, not duplicated.

Both `beforeRows` and `stagedRows` are computed by walking `Snapshot.Manifests(fs)` and summing `DataFile.Count()` across every entry — i.e., they read the immutable on-disk manifest list. The discrepancy means the staged snapshot's manifest list **actually contains data files whose total row count is less than the input snapshot's data files**. This is not a stat skew or a master-check arithmetic bug; it's fewer rows in the resulting table.

The same compaction code path on the same table with **no concurrent writer** passes the verification cleanly every time (29200 in / 29200 out, 600000 in / 600000 out across many runs). The race only fires under concurrent producer load.

## Suspect: `partitionedFanoutWriter` row-loss race

`tx.Overwrite` → `recordsToDataFiles` (table/arrow_utils.go:1300) → `partitionedFanoutWriter.Write` (table/partitioned_fanout_writer.go:73).

The fanout writer spawns N workers (default `config.EnvConfig.MaxWorkers = 5`) all reading from a single `inputRecordsCh`:

```go
// table/partitioned_fanout_writer.go:73-87
func (p *partitionedFanoutWriter) Write(ctx context.Context, workers int) iter.Seq2[iceberg.DataFile, error] {
inputRecordsCh := make(chan arrow.RecordBatch, workers)
outputDataFilesCh := make(chan iceberg.DataFile, workers)

fanoutWorkers, ctx := errgroup.WithContext(ctx)
startRecordFeeder(ctx, p.itr, fanoutWorkers, inputRecordsCh)

for range workers {
fanoutWorkers.Go(func() error {
return p.fanout(ctx, inputRecordsCh, outputDataFilesCh)
})
}

return p.yieldDataFiles(fanoutWorkers, outputDataFilesCh)
}
```

Each worker, on each batch:

```go
// table/partitioned_fanout_writer.go:112-153
func (p *partitionedFanoutWriter) fanout(ctx context.Context, inputRecordsCh <-chan arrow.RecordBatch, dataFilesChannel chan<- iceberg.DataFile) error {
for {
select {
case <-ctx.Done():
return context.Cause(ctx)

case record, ok := <-inputRecordsCh:
if !ok {
return nil
}
defer record.Release() // ← line 122: defer inside a for loop

partitions, err := p.getPartitions(record)
...
for _, val := range partitions {
...
rollingDataWriter, err := p.writerFactory.getOrCreateRollingDataWriter(
ctx, p.concurrentDataFileWriter, partitionPath, val.partitionValues, dataFilesChannel)
if err != nil {
return err
}

err = rollingDataWriter.Add(partitionRecord)
if err != nil {
return err
}
}
}
}
}
```

Two concrete concerns:

**1. `defer record.Release()` inside a `for` loop (line 122).** Defer is function-scoped, so releases accumulate until `fanout` returns rather than firing per-iteration. At minimum a memory smell. Under early-exit paths it could mean the released-on-defer record is still being read by another worker that pulled the same `*RecordBatch` (Arrow record batches are reference-counted; if Retain happens elsewhere this might be benign, but it's worth auditing).

**2. Concurrent `writerFactory.getOrCreateRollingDataWriter` for the same partition path.** N workers all racing to read from a single channel will frequently pull adjacent batches that hash to the same partition. If `getOrCreate` isn't strictly mutually exclusive across workers (or if the per-writer `Add` isn't synchronized), two workers can:

- Both observe the partition is missing, both create a rolling writer, only one ends up wired into the data-files channel → the other writer's flushed Parquet file is **never appended to the staged manifest list**, so its rows vanish from the staged total. This produces *exactly* the failure shape we see: a small, non-deterministic number of rows missing, biased toward partitions that received bursts during the compaction window.
- Or: both register but interleave `Add` calls in a way that drops some row-index slices.

I have not yet read `writer_factory.go` / `concurrent_data_file_writer.go` end-to-end to confirm whether the synchronization is sufficient — that would be the next step.

## What's confirmed vs. hypothesized

**Confirmed:**
- The symptom: `staged.totalRows < tbl.totalRows` after `tx.Overwrite` returns success.
- It is reproducible against fileblob with the workload described below.
- It does not reproduce against a quiescent table (same code, same filter, same sequence of operations).
- The transaction's snapshot pinning at the metadata level looks correct: `t.meta.currentSnapshot()` is captured at `NewTransaction` time, `Snapshot.dataFiles(fs, nil)` reads the immutable on-disk manifest list, and `classifyFilesForDeletions` correctly uses that pinned snapshot. So this does NOT appear to be the classifier mis-classifying files; the classifier sees the correct file set.

**Hypothesized:**
- The row loss happens on the writer side (fanout) rather than the reader/classifier side.
- The specific mechanism is one of the two `partitionedFanoutWriter` concerns above.
- Setting `config.EnvConfig.MaxWorkers = 1` should make the symptom vanish entirely (single-worker fanout has no concurrent `getOrCreate`).

I have a diagnostic build running with `MaxWorkers = 1` right now and will post a follow-up comment with the result. If the failures vanish at MaxWorkers=1 and reappear at MaxWorkers=5, the writer-race hypothesis is confirmed and the bug is upstream of any caller workaround.

## Reproduction

The full reproduction harness lives at https://github.com/mystictraveler/iceberg-janitor (branch `feature/go-rewrite-mvp`). The relevant pieces:

- `go/cmd/janitor-streamer/main.go` — TPC-DS streaming workload writer using iceberg-go's `SqlCatalog` (committing ~10/sec/fact at 200k–50k rows per commit, 50 partitions per fact)
- `go/pkg/janitor/compact.go` — the compaction code path shown above
- `go/pkg/safety/verify.go` — the row-count verification (master check)
- `go/test/bench/bench-tpcds.sh` — driver

```bash
DURATION_SECONDS=180 \
COMMITS_PER_MINUTE=600 \
STORE_SALES_PER_BATCH=200000 \
STORE_RETURNS_PER_BATCH=50000 \
CATALOG_SALES_PER_BATCH=100000 \
QUERY_INTERVAL_SECONDS=45 \
MAINTENANCE_INTERVAL_SECONDS=20 \
./go/test/bench/bench-tpcds.sh
```

11/18 compactions trip the verification on the post-#1-fix iceberg-janitor build.

## Severity

The transaction returns success on a row-losing operation. Any caller that doesn't independently row-count-check the staged snapshot will silently lose data. Concretely: a maintenance/compaction job using `Transaction.Overwrite` against any table receiving concurrent writes will accumulate row loss.

We are working around this on our side by replacing `tx.Overwrite` with `Transaction.ReplaceDataFilesWithDataFiles` (table/transaction.go:599), which takes explicit `filesToDelete []iceberg.DataFile` and `filesToAdd []iceberg.DataFile` lists with no row-filter classification and no partitionedFanoutWriter on the write side. Happy to share that as a separate issue/PR once it's landed and stable, in case it's useful for upstream tests.

## Adjacent

- mystictraveler/iceberg-janitor#1 — different concurrency-failure mode (CAS retry on "branch main has changed" requirement validation). Already fixed on our side; not the same bug.
- mystictraveler/iceberg-janitor#4 — our internal tracker for this bug, which links back here.

Read table/partitioned_fanout_writer.go, writer_factory.go, and concurrent_data_file_writer.go, then run the linked iceberg-janitor benchmark with concurrent writes and varying MaxWorkers. Trace how partition writers are created, flushed, and added to the staged manifest list; done means reproducing the race, identifying the loss mechanism, and demonstrating preserved row counts with a regression test.

Written by the indexing model from the issue text.