feat: DumpMergeWriter + merge-aware compaction

[JustMaier]

Summary

• DumpMergeWriter: Direct read-modify-write into data.bin during dump phases, bypassing ops log + compaction entirely. Uses 1024-bucket striped locks for rayon thread safety.
• merge-aware compaction: Fixes correctness bug where cold/hot compact used LWW, silently dropping earlier Merge ops for the same key. DocSiloAdapter now sets a merge function that concatenates Mi arrays.
• dump_processor wiring: Auto-detects whether DumpMergeWriter is available (data.bin exists from images phase) and uses it; falls back to ops log for the first phase.

How it works

1. Images phase writes doc ops to ops log → compact_after_dumps creates data.bin + index.bin
2. Tags/tools/techniques/resources phases: prepare_dump_merge() returns DumpMergeWriter
3. Per-slot batched Mi values are merge_put() directly into data.bin (read existing → decode → concat Mi → re-encode → write in-place)
4. No ops log writes, no compaction step needed for subsequent phases

Files changed

File	What
crates/datasilo/src/lib.rs	DumpMergeWriter struct, prepare_dump_merge, set_merge_fn, compact_cold_merge, hot compact merge
crates/datasilo/src/hash_index.rs	update_existing_concurrent (thread-safe entry update)
src/silos/doc_format.rs	merge_encoded_docs, merge_encoded_docs_into
src/silos/doc_silo_adapter.rs	prepare_dump_merge wrapper, set_merge_fn on open
src/sync/dump_processor.rs	DumpMergeWriter integration, collect_doc_op merge_writer param

Tests

• 3 DumpMergeWriter tests (basic, overflow, 1000-entry concurrent rayon)
• 2 merge_encoded_docs tests (Mi concatenation, non-Mi replace)
• 3 compaction tests (cold merge, hot merge, LWW backwards compat)

Test plan

• All datasilo crate tests pass (33 pass, 3 pre-existing key=0 sentinel failures)
• All doc_format tests pass (8/8)
• All doc_silo_adapter tests pass (2/2)
• All hash_index tests pass (12/12)
• Small-scale 14M dump validation (Scarlet running)
• Full-scale 107M dump validation (pending small-scale results)

🤖 Generated with Claude Code

[JustMaier]

PR #148 Review: DumpMergeWriter + merge-aware compaction

Overall this is well-structured -- the dump merge writer is a sound optimization (bypass ops log for subsequent phases), and the merge-aware compaction fixes a real correctness bug. The test coverage is solid for the happy paths. I found several issues that need attention before merge, ranging from potential data corruption to silent data loss.

---

1. Critical: Read/write mmap aliasing on the same file (UB risk)

prepare_dump_merge creates a second MmapMut (the write mmap) on the same data.bin file that self.data_mmap (the read mmap) already maps. The DumpMergeWriter then reads from read_ptr (the DataSilo read-only mmap) and writes to write_ptr (the new writable mmap).

Two mmaps on the same file from the same process is not undefined behavior per se, but the OS makes no guarantees about visibility timing between them. On Linux, both mappings typically share the same page cache, so writes through write_ptr are visible through read_ptr immediately. On Windows, this is less guaranteed -- separate MapViewOfFile calls may or may not share pages.

This matters because merge_put reads existing data from read_ptr, merges, and writes to write_ptr. If a prior merge_put wrote to the same slot (which should not happen in normal dump flow but could happen with retry logic or duplicate keys in CSV), the read would see stale data from the read mmap.

Recommendation: Document the single-write-per-key invariant explicitly in the safety contract. Consider whether the read path should use write_ptr instead of read_ptr to guarantee read-your-own-writes, which would eliminate the aliasing concern entirely -- the read mmap would only be needed for its length.

2. High: update_existing_concurrent writes partial fields without atomicity

The update_existing_concurrent method writes offset, length, and allocated as three separate copy_nonoverlapping calls. If a concurrent reader (e.g., a get() call from another thread) reads the entry between writes, it could see an inconsistent state -- new length with old offset, or vice versa.

During dump phases this is likely safe because queries are not running, but this method is pub and its safety contract only says "no two threads call this with the same key simultaneously" -- it says nothing about concurrent readers. A get() on a key being updated could observe torn reads.

Recommendation: Either (a) write all three fields in a single copy_nonoverlapping of 16 bytes (offset+length+allocated are contiguous at +8..+24), or (b) document in the safety contract that concurrent reads of the same key are also unsafe.

3. Medium: merge_encoded_docs error silently falls back to new.to_vec()

Throughout dump_processor.rs, the merge callback is:

|existing, new| {
    merge_encoded_docs(existing, new)
        .unwrap_or_else(|_| new.to_vec())
}

If decoding fails (corrupted existing data, truncated write, etc.), this silently drops the existing doc and replaces it with just the new fragment. For a tags phase, this means the base image doc fields (userId, sortAt, etc.) would be silently lost and replaced with just Mi([tagId]).

Same pattern in doc_silo_adapter.rs line ~798 for the compaction merge function.

Recommendation: At minimum, log the error with the key so it is diagnosable. Better: count decode failures in an atomic counter on DumpMergeWriter (like overflow_count) so the post-phase summary reports them.

4. Medium: Hot compaction merge order -- correct but needs a comment

In compact_hot_from (around line 1024-1033), after merging ops together, the code merges with existing data as:

*new_value = merge_fn(existing_bytes, new_value);

This calls merge_fn(existing_from_data_file, merged_ops_value). In merge_encoded_docs, new_data fields replace existing fields (except Mi which concatenates). So the argument order matters: the "new" ops should be in the new_data position, and the data file entry should be in the existing position. This is correct.

However, the ops-internal merge also uses the same argument convention: first op is "existing", second op is "new". If both ops set the same non-Mi field, the second op wins (correct LWW within an ops batch). Then the full merge treats the data file as "existing" and the ops-merged value as "new", so ops correctly override data file values. This is correct but subtle -- a comment clarifying the argument order semantics in compact_hot_from would prevent future mistakes.

5. Medium: compact_cold_merge calls write_batch_parallel which replaces the entire data file

compact_cold_merge collects all merged ops into a HashMap, then calls write_batch_parallel to write them. write_batch_parallel creates a new data.bin and index.bin from scratch. This is correct for cold compaction (no existing data file), but if called when there IS an existing data file (which should not happen because compact() routes to compact_hot_from when data exists), it would overwrite the data file with only the ops entries, losing all other keys.

This is protected by the routing logic in compact(), but it is a latent footgun. Consider adding a debug assertion in compact_cold_merge that self.data_mmap.is_none().

6. Low: flush() is a no-op

DumpMergeWriter::flush() returns Ok(()) with a comment saying "the mmap will flush on drop." This is misleading -- callers might expect flush() to actually persist data. The MmapMut is stored in _write_mmap but flush() cannot access it through the raw pointer.

Recommendation: Either make flush() actually call self._write_mmap.flush() (change _write_mmap to not be prefixed with _ and access it directly), or remove the method and document that data is persisted on drop.

7. Low: Linear scan in merge_encoded_docs field lookup

if let Some(entry) = fields.iter_mut().find(|(f, _)| *f == field_idx) {

This is O(n) per field in the new record, making the merge O(n*m) where n = existing fields and m = new fields. For doc records with ~15 fields this is negligible. Fine as-is, just noting it.

8. Nit: merge_encoded_docs_into is declared but never called

The zero-allocation variant merge_encoded_docs_into that writes into a caller-provided buffer is defined but all call sites use merge_encoded_docs (which allocates). Was this intended for future use from the DumpMergeWriter hot path?

---

Summary

The architecture is sound -- DumpMergeWriter for direct in-place writes during dump phases is a clear win over the ops log + compaction path. The merge-aware compaction fix is necessary for correctness. Main concerns:

1. Use write_ptr for reads too (or document the aliasing invariant)
2. Single 16-byte write in update_existing_concurrent (torn read risk)
3. Do not silently swallow decode errors in the merge callback
4. flush() should actually flush or be removed

Tests are thorough for the new code paths. The fallback to ops log when DumpMergeWriter is None looks correct -- all existing collect_doc_op call sites pass None for the new parameter in the non-merge path.

[JustMaier]

Code Review: DumpMergeWriter + merge-aware compaction

Good structural approach — bypassing the ops log for subsequent dump phases is the right move for throughput, and the striped-lock design is straightforward. Several issues below, ranging from a potential soundness problem to minor nits.

---

1. Dual-mmap aliasing in DumpMergeWriter (soundness concern)

prepare_dump_merge() opens a second MmapMut on the same data.bin file that the DataSilo already holds a read-only Mmap for (self.data_mmap). The DumpMergeWriter then:

• Reads from read_ptr (the DataSilo's existing read mmap)
• Writes to write_ptr (the new writable mmap)

Both point to the same file. On Linux, the kernel maps the same physical pages for both, so writes through write_ptr are visible through read_ptr — this works. On Windows, behavior is less guaranteed with independent mmap handles on the same file.

More importantly: there is no memory barrier between the write via write_ptr and the subsequent read via read_ptr in merge_put. If thread A writes slot X, then thread B reads slot X (different stripe bucket because slot X hashes differently for B's merge), the write may not be visible through the read mmap. In practice the stripe lock provides ordering for same-key access, and different-key access reads different offsets, so this is likely safe — but the aliasing of two separate mmaps over the same file is technically UB under Rust's aliasing rules.

Suggestion: Consider using a single MmapMut and reading from the same pointer (cast to *const u8 for reads). This eliminates the aliasing question entirely. The read mmap in DataSilo isn't needed during dump since no queries run concurrently.

---

2. merge_encoded_docs slot extraction assumes Merge op layout

let slot = if existing.len() >= 5 {
    u32::from_le_bytes(existing[1..5].try_into().unwrap())
} else {
    0
};

This hard-codes that byte 0 is the op tag and bytes 1..5 are the slot. It works for OP_TAG_MERGE records, but:

• No validation of byte 0: If existing is a Create op or corrupted data, slot extraction silently returns a wrong value.
• Fallback to slot=0: If existing.len() < 5, slot becomes 0 — but slot_to_key maps slot 0 to key 1 (via SLOT_KEY_OFFSET), so this wouldn't cause a key=0 sentinel collision. However, a record claiming to be slot 0 when the real slot is something else is still a silent data corruption.

Suggestion: Either assert/validate that existing[0] == OP_TAG_MERGE, or extract slot from the decode_doc_fields path (which already parses the op). The perf cost of one extra byte check is negligible.

---

3. merge_encoded_docs_into allocates despite the name

The doc comment says "Zero allocation except for the field Vec decode" — but decode_doc_fields calls decode_doc_op which allocates Vec<(u16, PackedValue)> and potentially allocates for string/Mi values. Then encode_merge_fields_into allocates the encoded output. In the hot path (dump_processor calling merge_put), the merge_fn closure calls merge_encoded_docs (not _into), which also allocates the return Vec<u8>.

This is not a bug, but the naming is misleading. Consider either:

• Actually making _into zero-alloc (reuse a thread-local field vec), or
• Dropping the "zero allocation" claim from the doc

Also: merge_encoded_docs_into is defined but never called — all callsites use merge_encoded_docs. Dead code?

---

4. Hot compact merge ordering — verified correct

In compact_hot_from, ops are accumulated left-to-right (merge_fn(accumulated, new_op)), then the accumulated result is merged with the data file (merge_fn(data_file, accumulated)). For Mi concatenation this yields [data_file_values, op1_values, op2_values, ...] — chronologically correct. The test confirms: "base+add1+add2". Good.

---

5. flush() is a no-op (misleading API)

pub fn flush(&self) -> io::Result<()> {
    Ok(())
}

This silently succeeds without flushing anything. Callers who call flush() expecting data to be durable will be misled. The mmap is stored as _write_mmap — you could expose a method that calls self._write_mmap.flush() instead.

Suggestion: Either make it actually flush, or remove the method and document that callers must drop the writer to trigger the mmap flush.

---

6. Overflow handling: silent data loss in dump_processor

When merge_put returns false (overflow — merged data exceeds allocated buffer), the dump_processor silently continues:

mw.merge_put(key, &doc_encode_buf, |existing, new| {
    crate::silos::doc_format::merge_encoded_docs(existing, new)
        .unwrap_or_else(|_| new.to_vec())
});

The return value is discarded at every callsite. The overflow count is logged at the end of the phase, but by then the data is lost — those slots will have incomplete Mi arrays.

Suggestion: At minimum, collect overflow keys into a Vec and fall back to ops-log writes for those entries. Or, if the overflow rate is expected to be zero in practice (because buffer_ratio provides enough headroom), add an assertion or at least a per-row warning log so the first overflow is immediately visible.

---

7. unwrap_or_else in merge_fn swallows decode errors

This pattern appears 5+ times:

doc_format::merge_encoded_docs(existing, new)
    .unwrap_or_else(|_| new.to_vec())

If decode_doc_fields fails on the existing record (corrupt data, truncated write), the merge silently replaces the entire existing record with just the new data. This loses all previously accumulated fields for that slot.

For dump phases where data integrity matters, consider at least logging when this fallback triggers. A decode failure on an existing record in data.bin suggests something went wrong in a prior phase.

---

8. collect_doc_op allocates on the merge_writer path

The merge_writer branch in collect_doc_op uses encode_merge_fields (allocating) instead of encode_merge_fields_into (reusing doc_buf):

if let Some(mw) = merge_writer {
    let bytes = crate::silos::doc_format::encode_merge_fields(slot, &fields);
    mw.merge_put(key, &bytes, ...);
}

The parallel-ops-writer path reuses doc_buf via encode_merge_fields_into. For consistency and throughput, the merge writer path should do the same — especially since this runs in a rayon hot loop at 100M+ rows.

---

9. Minor: compact_cold_merge uses write_batch_parallel

let batch: Vec<(u64, Vec<u8>)> = entries.into_iter().collect();
let count = self.write_batch_parallel(&batch)?;

write_batch_parallel rebuilds the entire data file and index from scratch. For cold compaction that is fine (there is no existing data file, or it is being replaced). But the doc comment doesn't make it clear that cold-merge is a full rewrite. Worth a one-line comment.

Also: the entries HashMap iteration order is nondeterministic. If write_batch_parallel cares about key ordering (e.g., for sequential I/O during reads), this could cause suboptimal layout. Probably fine for dumps, but worth noting.

---

10. Things done well

• Striped lock design is clean and well-documented. 1024 stripes with key % N is a solid choice for rayon workloads.
• Test coverage is good: basic, overflow, 1000-entry concurrent, cold merge, hot merge, and LWW backwards compat. The hot merge test validates the non-obvious three-way merge ordering.
• Fallback strategy (merge writer when data.bin exists, ops log otherwise) handles the images-phase bootstrap correctly.
• Code organization: keeping merge_encoded_docs in doc_format.rs and the merge plumbing in datasilo is the right separation of concerns.
• The encode_dump_merge hoisting in dump_processor (encoding once, then choosing write path) simplifies the control flow vs the old duplicated encode calls.

---

Summary

Must fix before merge:

• Issue #6 (overflow return value silently discarded — data loss)
• Issue #5 (flush() is a no-op — either implement or remove)

Should fix:

• Issue #2 (validate op tag before slot extraction)
• Issue #7 (log when decode fallback triggers)
• Issue #8 (use encode_merge_fields_into for merge writer path)

Consider:

• Issue #1 (single mmap to avoid aliasing question)
• Issue #3 (remove dead merge_encoded_docs_into or use it)