Repository: vercel/next.js
Pull request: Turbopack: cleanup db log and add verbose option

### What?

Prints nice graphs into the database LOG file.

Example:

```
  2 | 00000405 | Compaction:
  2 | 00000405 | MERGE (1311566 keys):
  2 | 00000405 | 00000349 INPUT  |                                                                            [======]                  | c00004c1488ef798-d40e1767c1ee6786 
  2 | 00000405 | 00000353 INPUT  |                                                                                   [========]         | d40e1a134691c633-e91bfab1b0dab95a 
  2 | 00000405 | 00000357 INPUT  |                                                                            [===========]             | c00084722c196aa3-e0d85966f3086922 
  2 | 00000405 | 00000358 INPUT  |                                                                                        [===========] | e0d9aaacdbea0b12-ffffbaef025822aa 
  2 | 00000405 | 00000365 INPUT  |                                                                                            [===]     | e91bfe6c3e8c62ea-f5c19d448c217d89 
  2 | 00000405 | 00000366 INPUT  |                                                                                                [===] | f5c1a023c1b77b86-fffffe0d895c3b37 
  2 | 00000405 | 00000372 INPUT  |                                                                            [=======================] | c00084722c196aa3-ffffbaef025822aa 
  2 | 00000405 | 00000374 INPUT  |                                                                            [=======================] | c0250dd8b533294a-ffb01c4e6519a470 
  2 | 00000405 | 00000384 OUTPUT |                                                                            [======]                  | c00004c1488ef798-d40e1767c1ee6786
  2 | 00000405 | 00000388 OUTPUT |                                                                                   [========]         | d40e1a134691c633-e91bfab1b0dab95a
  2 | 00000405 | 00000393 OUTPUT |                                                                            [===========]             | c00084722c196aa3-e0d85966f3086922
  2 | 00000405 | 00000394 OUTPUT |                                                                                        [===========] | e0d9aaacdbea0b12-ffffbaef025822aa
  2 | 00000405 | 00000401 OUTPUT |                                                                                            [===]     | e91bfe6c3e8c62ea-f5c19d448c217d89
  2 | 00000405 | 00000402 OUTPUT |                                                                                                [===] | f5c1a023c1b77b86-fffffe0d895c3b37
  2 | 00000405 | MERGE (1277134 keys):
  2 | 00000405 | 00000347 INPUT  |                          [=======]                                                                   | 4000034716c961a1-549678c058a07106 
  2 | 00000405 | 00000351 INPUT  |                                  [=======]                                                           | 54967af27cc1ec1d-6979dbc1b9969add 
  2 | 00000405 | 00000354 INPUT  |                          [=======================]                                                   | 400174554370f74b-7fff39f8ec73f8b7 
  2 | 00000405 | 00000361 INPUT  |                                          [====]                                                      | 6979dd7b4df64f96-75e162f9f837a260 
  2 | 00000405 | 00000362 INPUT  |                                               [==]                                                   | 75e1648525c2154a-7ffffea14ef427f3 
  2 | 00000405 | 00000371 INPUT  |                          [=======================]                                                   | 400174554370f74b-7fff39f8ec73f8b7 
  2 | 00000405 | 00000373 INPUT  |                          [=======================]                                                   | 40038dd87f963da6-7fff0105911c9647 
  2 | 00000405 | 00000382 OUTPUT |                          [=======]                                                                   | 4000034716c961a1-549678c058a07106
  2 | 00000405 | 00000386 OUTPUT |                                  [=======]                                                           | 54967af27cc1ec1d-6979dbc1b9969add
  2 | 00000405 | 00000390 OUTPUT |                          [=======================]                                                   | 400174554370f74b-7fff39f8ec73f8b7
  2 | 00000405 | 00000391 OUTPUT |                                          [====]                                                      | 6979dd7b4df64f96-75e162f9f837a260
  2 | 00000405 | 00000392 OUTPUT |                                               [==]                                                   | 75e1648525c2154a-7ffffea14ef427f3
  2 | 00000405 | MERGE (1389734 keys):
  2 | 00000405 | 00000348 INPUT  | [=======]                                                                                            | 00000da30f0f1507-15bdf81f772b13ab 
  2 | 00000405 | 00000352 INPUT  |         [======]                                                                                     | 15bdfdf36416a0c5-27bb1097c991db6c 
  2 | 00000405 | 00000359 INPUT  | [===========]                                                                                        | 0000bed76899fc20-20f90e531b79c213 
  2 | 00000405 | 00000360 INPUT  |             [===========]                                                                            | 20fa2f7e9952973f-3fff4d10047dbfc4 
  2 | 00000405 | 00000367 INPUT  |                [====]                                                                                | 27bb1d45d57871a5-33d7f6c31d6a87f2 
  2 | 00000405 | 00000368 INPUT  |                     [===]                                                                            | 33d7f95aefeb6ca0-3ffff899c293a0b4 
  2 | 00000405 | 00000376 INPUT  | [=======================]                                                                            | 0000bed76899fc20-3fff4d10047dbfc4 
  2 | 00000405 | 00000385 OUTPUT | [=======]                                                                                            | 00000da30f0f1507-15bdf81f772b13ab
  2 | 00000405 | 00000389 OUTPUT |         [======]                                                                                     | 15bdfdf36416a0c5-27bb1097c991db6c
  2 | 00000405 | 00000397 OUTPUT | [===========]                                                                                        | 0000bed76899fc20-20f90e531b79c213
  2 | 00000405 | 00000398 OUTPUT |             [===========]                                                                            | 20fa2f7e9952973f-3fff4d10047dbfc4
  2 | 00000405 | 00000403 OUTPUT |                [====]                                                                                | 27bb1d45d57871a5-33d7f6c31d6a87f2
  2 | 00000405 | 00000404 OUTPUT |                     [===]                                                                            | 33d7f95aefeb6ca0-3ffff899c293a0b4
  2 | 00000405 | MERGE (1362988 keys):
  2 | 00000405 | 00000346 INPUT  |                                                   [=======]                                          | 8000063004f10520-96da09f7fbce981e 
  2 | 00000405 | 00000350 INPUT  |                                                           [======]                                   | 96da108f86d302c8-a6f4e6fc64d75324 
  2 | 00000405 | 00000355 INPUT  |                                                   [============]                                     | 8000b4968b026f12-a18f4d0889cdaa27 
  2 | 00000405 | 00000356 INPUT  |                                                                [==========]                          | a190533ade79230a-bfff4bd5dbeb964b 
  2 | 00000405 | 00000363 INPUT  |                                                                  [====]                              | a6f4e8052ab381ac-b37e203806567e5a 
  2 | 00000405 | 00000364 INPUT  |                                                                       [===]                          | b37e2083733ac276-bffffe42b4a8071b 
  2 | 00000405 | 00000375 INPUT  |                                                   [=======================]                          | 8000b4968b026f12-bfff4bd5dbeb964b 
  2 | 00000405 | 00000383 OUTPUT |                                                   [=======]                                          | 8000063004f10520-96da09f7fbce981e
  2 | 00000405 | 00000387 OUTPUT |                                                           [======]                                   | 96da108f86d302c8-a6f4e6fc64d75324
  2 | 00000405 | 00000395 OUTPUT |                                                   [============]                                     | 8000b4968b026f12-a18f4d0889cdaa27
  2 | 00000405 | 00000396 OUTPUT |                                                                [==========]                          | a190533ade79230a-bfff4bd5dbeb964b
  2 | 00000405 | 00000399 OUTPUT |                                                                  [====]                              | a6f4e8052ab381ac-b37e203806567e5a
  2 | 00000405 | 00000400 OUTPUT |                                                                       [===]                          | b37e2083733ac276-bffffe42b4a8071b
Time 2025-11-10T11:05:02.041664Z
Commit 00000406 5341422 keys in 333ms 302µs
FAM | META SEQ | SST SEQ        | RANGE
  2 | 00000405 | 00000384 SST    |                                                                            [======]                  | c00004c1488ef798-d40e1767c1ee6786 (96 MiB, cold)
  2 | 00000405 | 00000388 SST    |                                                                                   [========]         | d40e1a134691c633-e91bfab1b0dab95a (96 MiB, cold)
  2 | 00000405 | 00000393 SST    |                                                                            [===========]             | c00084722c196aa3-e0d85966f3086922 (34 MiB, warm)
  2 | 00000405 | 00000394 SST    |                                                                                        [===========] | e0d9aaacdbea0b12-ffffbaef025822aa (48 MiB, warm)
  2 | 00000405 | 00000401 SST    |                                                                                            [===]     | e91bfe6c3e8c62ea-f5c19d448c217d89 (72 MiB, cold)
  2 | 00000405 | 00000402 SST    |                                                                                                [===] | f5c1a023c1b77b86-fffffe0d895c3b37 (77 MiB, cold)
  2 | 00000405 | 00000382 SST    |                          [=======]                                                                   | 4000034716c961a1-549678c058a07106 (95 MiB, cold)
  2 | 00000405 | 00000386 SST    |                                  [=======]                                                           | 54967af27cc1ec1d-6979dbc1b9969add (96 MiB, cold)
  2 | 00000405 | 00000390 SST    |                          [=======================]                                                   | 400174554370f74b-7fff39f8ec73f8b7 (82 MiB, warm)
  2 | 00000405 | 00000391 SST    |                                          [====]                                                      | 6979dd7b4df64f96-75e162f9f837a260 (70 MiB, cold)
  2 | 00000405 | 00000392 SST    |                                               [==]                                                   | 75e1648525c2154a-7ffffea14ef427f3 (76 MiB, cold)
  2 | 00000405 | 00000385 SST    | [=======]                                                                                            | 00000da30f0f1507-15bdf81f772b13ab (95 MiB, cold)
  2 | 00000405 | 00000389 SST    |         [======]                                                                                     | 15bdfdf36416a0c5-27bb1097c991db6c (95 MiB, cold)
  2 | 00000405 | 00000397 SST    | [===========]                                                                                        | 0000bed76899fc20-20f90e531b79c213 (41 MiB, warm)
  2 | 00000405 | 00000398 SST    |             [===========]                                                                            | 20fa2f7e9952973f-3fff4d10047dbfc4 (39 MiB, warm)
  2 | 00000405 | 00000403 SST    |                [====]                                                                                | 27bb1d45d57871a5-33d7f6c31d6a87f2 (82 MiB, cold)
  2 | 00000405 | 00000404 SST    |                     [===]                                                                            | 33d7f95aefeb6ca0-3ffff899c293a0b4 (89 MiB, cold)
  2 | 00000405 | 00000383 SST    |                                                   [=======]                                          | 8000063004f10520-96da09f7fbce981e (95 MiB, cold)
  2 | 00000405 | 00000387 SST    |                                                           [======]                                   | 96da108f86d302c8-a6f4e6fc64d75324 (94 MiB, cold)
  2 | 00000405 | 00000395 SST    |                                                   [============]                                     | 8000b4968b026f12-a18f4d0889cdaa27 (40 MiB, warm)
  2 | 00000405 | 00000396 SST    |                                                                [==========]                          | a190533ade79230a-bfff4bd5dbeb964b (44 MiB, warm)
  2 | 00000405 | 00000399 SST    |                                                                  [====]                              | a6f4e8052ab381ac-b37e203806567e5a (83 MiB, cold)
  2 | 00000405 | 00000400 SST    |                                                                       [===]                          | b37e2083733ac276-bffffe42b4a8071b (88 MiB, cold)
  2 | 00000405 | 00000346 00000347 00000348 00000349 00000350 00000351 00000352 00000353 00000354 00000355 00000356 00000357 00000358 00000359 00000360 OBSOLETE SST
  2 | 00000405 | 00000361 00000362 00000363 00000364 00000365 00000366 00000367 00000368 00000371 00000372 00000373 00000374 00000375 00000376 OBSOLETE SST
    |          | 00000346 00000347 00000348 00000349 00000350 00000351 00000352 00000353 00000354 00000355 00000356 00000357 00000358 00000359 00000360 SST DELETED
    |          | 00000361 00000362 00000363 00000364 00000365 00000366 00000367 00000368 00000371 00000372 00000373 00000374 00000375 00000376 SST DELETED
    |          | 00000381 META DELETED
```

Changed files:
filename: turbopack/crates/turbo-persistence/Cargo.toml
patch: @@ -10,6 +10,7 @@ verify_sst_content = []
 strict_checks = []
 stats = ["quick_cache/stats"]
 print_stats = ["stats"]
+verbose_log = []
 
 [dependencies]
 anyhow = { workspace = true }

filename: turbopack/crates/turbo-persistence/src/db.rs
patch: @@ -14,6 +14,7 @@ use byteorder::{BE, ReadBytesExt, WriteBytesExt};
 use jiff::Timestamp;
 use memmap2::Mmap;
 use parking_lot::{Mutex, RwLock};
+use smallvec::SmallVec;
 
 pub use crate::compaction::selector::CompactConfig;
 use crate::{
@@ -476,7 +477,7 @@ impl<S: ParallelScheduler> TurboPersistence<S> {
         &self,
         CommitOptions {
             mut new_meta_files,
-            mut new_sst_files,
+            new_sst_files,
             mut new_blob_files,
             mut sst_seq_numbers_to_delete,
             mut blob_seq_numbers_to_delete,
@@ -611,35 +612,88 @@ impl<S: ParallelScheduler> TurboPersistence<S> {
                 writeln!(log, "Time {time}")?;
                 let span = time.until(Timestamp::now())?;
                 writeln!(log, "Commit {seq:08} {keys_written} keys in {span:#}")?;
-                for (seq, family, ssts, obsolete) in new_meta_info {
-                    writeln!(log, "{seq:08} META family:{family}",)?;
+                writeln!(log, "FAM | META SEQ | SST SEQ        | RANGE")?;
+                for (meta_seq, family, ssts, obsolete) in new_meta_info {
                     for (seq, min, max, size) in ssts {
                         writeln!(
                             log,
-                            "  {seq:08} SST  {min:016x}-{max:016x} {} MiB",
+                            "{family:3} | {meta_seq:08} | {seq:08} SST    | {} ({} MiB)",
+                            range_to_str(min, max),
                             size / 1024 / 1024
                         )?;
                     }
-                    for seq in obsolete {
-                        writeln!(log, "  {seq:08} OBSOLETE SST")?;
+                    for obsolete in obsolete.chunks(15) {
+                        write!(log, "{family:3} | {meta_seq:08} |")?;
+                        for seq in obsolete {
+                            write!(log, " {seq:08}")?;
+                        }
+                        writeln!(log, " OBSOLETE SST")?;
                     }
                 }
-                new_sst_files.sort_unstable_by_key(|(seq, _)| *seq);
-                for (seq, _) in new_sst_files.iter() {
-                    writeln!(log, "{seq:08} NEW SST")?;
+
+                fn write_seq_numbers<W: std::io::Write, T, I>(
+                    log: &mut W,
+                    items: I,
+                    label: &str,
+                    extract_seq: fn(&T) -> u32,
+                ) -> std::io::Result<()>
+                where
+                    I: IntoIterator<Item = T>,
+                {
+                    let items: Vec<T> = items.into_iter().collect();
+                    for chunk in items.chunks(15) {
+                        write!(log, "    |          |")?;
+                        for item in chunk {
+                            write!(log, " {:08}", extract_seq(item))?;
+                        }
+                        writeln!(log, " {}", label)?;
+                    }
+                    Ok(())
                 }
+
                 new_blob_files.sort_unstable_by_key(|(seq, _)| *seq);
-                for (seq, _) in new_blob_files.iter() {
-                    writeln!(log, "{seq:08} NEW BLOB")?;
-                }
-                for seq in sst_seq_numbers_to_delete.iter() {
-                    writeln!(log, "{seq:08} SST DELETED")?;
-                }
-                for seq in meta_seq_numbers_to_delete.iter() {
-                    writeln!(log, "{seq:08} META DELETED")?;
-                }
-                for seq in blob_seq_numbers_to_delete.iter() {
-                    writeln!(log, "{seq:08} BLOB DELETED")?;
+                write_seq_numbers(&mut log, new_blob_files, "NEW BLOB", |&(seq, _)| seq)?;
+                write_seq_numbers(
+                    &mut log,
+                    blob_seq_numbers_to_delete,
+                    "BLOB DELETED",
+                    |&seq| seq,
+                )?;
+                write_seq_numbers(&mut log, sst_seq_numbers_to_delete, "SST DELETED", |&seq| {
+                    seq
+                })?;
+                write_seq_numbers(
+                    &mut log,
+                    meta_seq_numbers_to_delete,
+                    "META DELETED",
+                    |&seq| seq,
+                )?;
+                #[cfg(feature = "verbose_log")]
+                {
+                    writeln!(log, "New database state:")?;
+                    writeln!(log, "FAM | META SEQ | SST SEQ        | RANGE")?;
+                    let inner = self.inner.read();
+                    let families = inner.meta_files.iter().map(|meta| meta.family()).filter({
+                        let mut set = HashSet::new();
+                        move |family| set.insert(*family)
+                    });
+                    for family in families {
+                        for meta in inner.meta_files.iter() {
+                            if meta.family() != family {
+                                continue;
+                            }
+                            let meta_seq = meta.sequence_number();
+                            for entry in meta.entries().iter() {
+                                let seq = entry.sequence_number();
+                                let range = entry.range();
+                                writeln!(
+                                    log,
+                                    "{family:3} | {meta_seq:08} | {seq:08}        | {}",
+                                    range_to_str(range.min_hash, range.max_hash)
+                                )?;
+                            }
+                        }
+                    }
                 }
             }
             anyhow::Ok(())
@@ -835,30 +889,7 @@ impl<S: ParallelScheduler> TurboPersistence<S> {
                         });
                     }
 
-                    self.parallel_scheduler.block_in_place(|| {
-                        let metrics = compute_metrics(&ssts_with_ranges, 0..=u64::MAX);
-                        let guard = log_mutex.lock();
-                        let mut log = self.open_log()?;
-                        writeln!(
-                            log,
-                            "Compaction for family {family} (coverage: {}, overlap: {}, \
-                             duplication: {} / {} MiB):",
-                            metrics.coverage,
-                            metrics.overlap,
-                            metrics.duplication,
-                            metrics.duplicated_size / 1024 / 1024
-                        )?;
-                        for job in merge_jobs.iter() {
-                            writeln!(log, "  merge")?;
-                            for i in job.iter() {
-                                let seq = ssts_with_ranges[*i].seq;
-                                let (min, max) = ssts_with_ranges[*i].range().into_inner();
-                                writeln!(log, "    {seq:08} {min:016x}-{max:016x}")?;
-                            }
-                        }
-                        drop(guard);
-                        anyhow::Ok(())
-                    })?;
+                    let metrics = compute_metrics(&ssts_with_ranges, 0..=u64::MAX);
 
                     // Later we will remove the merged files
                     let sst_seq_numbers_to_delete = merge_jobs
@@ -875,6 +906,7 @@ impl<S: ParallelScheduler> TurboPersistence<S> {
                             new_sst_files: Vec<(u32, File, StaticSortedFileBuilderMeta<'static>)>,
                             blob_seq_numbers_to_delete: Vec<u32>,
                             keys_written: u64,
+                            indicies: SmallVec<[usize; 1]>,
                         },
                         Move {
                             seq: u32,
@@ -883,185 +915,193 @@ impl<S: ParallelScheduler> TurboPersistence<S> {
                     }
                     let merge_result = self
                         .parallel_scheduler
-                        .parallel_map_collect_owned::<_, _, Result<Vec<_>>>(merge_jobs, |indices| {
-                            let _span = span.clone().entered();
-                            if indices.len() == 1 {
-                                // If we only have one file, we can just move it
-                                let index = indices[0];
-                                let meta_index = ssts_with_ranges[index].meta_index;
-                                let index_in_meta = ssts_with_ranges[index].index_in_meta;
-                                let meta_file = &meta_files[meta_index];
-                                let entry = meta_file.entry(index_in_meta);
-                                let amqf = Cow::Borrowed(entry.raw_amqf(meta_file.amqf_data()));
-                                let meta = StaticSortedFileBuilderMeta {
-                                    min_hash: entry.min_hash(),
-                                    max_hash: entry.max_hash(),
-                                    amqf,
-                                    key_compression_dictionary_length: entry
-                                        .key_compression_dictionary_length(),
-                                    block_count: entry.block_count(),
-                                    size: entry.size(),
-                                    entries: 0,
-                                };
-                                return Ok(PartialMergeResult::Move {
-                                    seq: entry.sequence_number(),
-                                    meta,
-                                });
-                            }
-
-                            fn create_sst_file<'l, S: ParallelScheduler>(
-                                parallel_scheduler: &S,
-                                entries: &[LookupEntry<'l>],
-                                total_key_size: usize,
-                                path: &Path,
-                                seq: u32,
-                            ) -> Result<(u32, File, StaticSortedFileBuilderMeta<'static>)>
-                            {
-                                let _span = tracing::trace_span!("write merged sst file").entered();
-                                let (meta, file) = parallel_scheduler.block_in_place(|| {
-                                    write_static_stored_file(
-                                        entries,
-                                        total_key_size,
-                                        &path.join(format!("{seq:08}.sst")),
-                                    )
-                                })?;
-                                Ok((seq, file, meta))
-                            }
-
-                            let mut new_sst_files = Vec::new();
-
-                            // Iterate all SST files
-                            let iters = indices
-                                .iter()
-                                .map(|&index| {
+                        .parallel_map_collect_owned::<_, _, Result<Vec<_>>>(
+                            merge_jobs,
+                            |indicies| {
+                                let _span = span.clone().entered();
+                                if indicies.len() == 1 {
+                                    // If we only have one file, we can just move it
+                                    let index = indicies[0];
                                     let meta_index = ssts_with_ranges[index].meta_index;
                                     let index_in_meta = ssts_with_ranges[index].index_in_meta;
-                                    let meta = &meta_files[meta_index];
-                                    meta.entry(index_in_meta)
-                                        .sst(meta)?
-                                        .iter(key_block_cache, value_block_cache)
-                                })
-                                .collect::<Result<Vec<_>>>()?;
-
-                            let iter = MergeIter::new(iters.into_iter())?;
-
-                            // TODO figure out how to delete blobs when they are no longer
-                            // referenced
-                            let blob_seq_numbers_to_delete: Vec<u32> = Vec::new();
-
-                            let mut keys_written = 0;
-
-                            let mut total_key_size = 0;
-                            let mut total_value_size = 0;
-                            let mut current: Option<LookupEntry<'_>> = None;
-                            let mut entries = Vec::new();
-                            let mut last_entries = Vec::new();
-                            let mut last_entries_total_key_size = 0;
-                            for entry in iter {
-                                let entry = entry?;
-
-                                // Remove duplicates
-                                if let Some(current) = current.take() {
-                                    if current.key != entry.key {
-                                        let key_size = current.key.len();
-                                        let value_size = current.value.uncompressed_size_in_sst();
-                                        total_key_size += key_size;
-                                        total_value_size += value_size;
-
-                                        if total_key_size + total_value_size
-                                            > DATA_THRESHOLD_PER_COMPACTED_FILE
-                                            || entries.len() >= MAX_ENTRIES_PER_COMPACTED_FILE
-                                        {
-                                            let selected_total_key_size =
-                                                last_entries_total_key_size;
-                                            swap(&mut entries, &mut last_entries);
-                                            last_entries_total_key_size = total_key_size - key_size;
-                                            total_key_size = key_size;
-                                            total_value_size = value_size;
-
-                                            if !entries.is_empty() {
-                                                let seq = sequence_number
-                                                    .fetch_add(1, Ordering::SeqCst)
-                                                    + 1;
-
-                                                keys_written += entries.len() as u64;
-                                                new_sst_files.push(create_sst_file(
-                                                    &self.parallel_scheduler,
-                                                    &entries,
-                                                    selected_total_key_size,
-                                                    path,
-                                                    seq,
-                                                )?);
-
-                                                entries.clear();
+                                    let meta_file = &meta_files[meta_index];
+                                    let entry = meta_file.entry(index_in_meta);
+                                    let amqf = Cow::Borrowed(entry.raw_amqf(meta_file.amqf_data()));
+                                    let meta = StaticSortedFileBuilderMeta {
+                                        min_hash: entry.min_hash(),
+                                        max_hash: entry.max_hash(),
+                                        amqf,
+                                        key_compression_dictionary_length: entry
+                                            .key_compression_dictionary_length(),
+                                        block_count: entry.block_count(),
+                                        size: entry.size(),
+                                        entries: 0,
+                                    };
+                                    return Ok(PartialMergeResult::Move {
+                                        seq: entry.sequence_number(),
+                                        meta,
+                                    });
+                                }
+
+                                fn create_sst_file<'l, S: ParallelScheduler>(
+                                    parallel_scheduler: &S,
+                                    entries: &[LookupEntry<'l>],
+                                    total_key_size: usize,
+                                    path: &Path,
+                                    seq: u32,
+                                ) -> Result<(u32, File, StaticSortedFileBuilderMeta<'static>)>
+                                {
+                                    let _span =
+                                        tracing::trace_span!("write merged sst file").entered();
+                                    let (meta, file) = parallel_scheduler.block_in_place(|| {
+                                        write_static_stored_file(
+                                            entries,
+                                            total_key_size,
+                                            &path.join(format!("{seq:08}.sst")),
+                                        )
+                                    })?;
+                                    Ok((seq, file, meta))
+                                }
+
+                                let mut new_sst_files = Vec::new();
+
+                                // Iterate all SST files
+                                let iters = indicies
+                                    .iter()
+                                    .map(|&index| {
+                                        let meta_index = ssts_with_ranges[index].meta_index;
+                                        let index_in_meta = ssts_with_ranges[index].index_in_meta;
+                                        let meta = &meta_files[meta_index];
+                                        meta.entry(index_in_meta)
+                                            .sst(meta)?
+                                            .iter(key_block_cache, value_block_cache)
+                                    })
+                                    .collect::<Result<Vec<_>>>()?;
+
+                                let iter = MergeIter::new(iters.into_iter())?;
+
+                                // TODO figure out how to delete blobs when they are no longer
+                                // referenced
+                                let blob_seq_numbers_to_delete: Vec<u32> = Vec::new();
+
+                                let mut keys_written = 0;
+
+                                let mut total_key_size = 0;
+                                let mut total_value_size = 0;
+                                let mut current: Option<LookupEntry<'_>> = None;
+                                let mut entries = Vec::new();
+                                let mut last_entries = Vec::new();
+                                let mut last_entries_total_key_size = 0;
+                                for entry in iter {
+                                    let entry = entry?;
+
+                                    // Remove duplicates
+                                    if let Some(current) = current.take() {
+                                        if current.key != entry.key {
+                                            let key_size = current.key.len();
+                                            let value_size =
+                                                current.value.uncompressed_size_in_sst();
+                                            total_key_size += key_size;
+                                            total_value_size += value_size;
+
+                                            if total_key_size + total_value_size
+                                                > DATA_THRESHOLD_PER_COMPACTED_FILE
+                                                || entries.len() >= MAX_ENTRIES_PER_COMPACTED_FILE
+                                            {
+                                                let selected_total_key_size =
+                                                    last_entries_total_key_size;
+                                                swap(&mut entries, &mut last_entries);
+                                                last_entries_total_key_size =
+                                                    total_key_size - key_size;
+                                                total_key_size = key_size;
+                                                total_value_size = value_size;
+
+                                                if !entries.is_empty() {
+                                                    let seq = sequence_number
+                                                        .fetch_add(1, Ordering::SeqCst)
+                                                        + 1;
+
+                                                    keys_written += entries.len() as u64;
+                                                    new_sst_files.push(create_sst_file(
+                                                        &self.parallel_scheduler,
+                                                        &entries,
+                                                        selected_total_key_size,
+                                                        path,
+                                                        seq,
+                                                    )?);
+
+                                                    entries.clear();
+                                                }
                                             }
-                                        }
 
-                                        entries.push(current);
-                                    } else {
-                                        // Override value
+                                            entries.push(current);
+                                        } else {
+                                            // Override value
+                                        }
                                     }
+                                    current = Some(entry);
+                                }
+                                if let Some(entry) = current {
+                                    total_key_size += entry.key.len();
+                                    // Obsolete as we no longer need total_value_size
+                                    // total_value_size += entry.value.uncompressed_size_in_sst();
+                                    entries.push(entry);
                                 }
-                                current = Some(entry);
-                            }
-                            if let Some(entry) = current {
-                                total_key_size += entry.key.len();
-                                // Obsolete as we no longer need total_value_size
-                                // total_value_size += entry.value.uncompressed_size_in_sst();
-                                entries.push(entry);
-                            }
 
-                            // If we have one set of entries left, write them to a new SST file
-                            if last_entries.is_empty() && !entries.is_empty() {
-                                let seq = sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
-
-                                keys_written += entries.len() as u64;
-                                new_sst_files.push(create_sst_file(
-                                    &self.parallel_scheduler,
-                                    &entries,
-                                    total_key_size,
-                                    path,
-                                    seq,
-                                )?);
-                            } else
-                            // If we have two sets of entries left, merge them and
-                            // split it into two SST files, to avoid having a
-                            // single SST file that is very small.
-                            if !last_entries.is_empty() {
-                                last_entries.append(&mut entries);
-
-                                last_entries_total_key_size += total_key_size;
-
-                                let (part1, part2) = last_entries.split_at(last_entries.len() / 2);
-
-                                let seq1 = sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
-                                let seq2 = sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
-
-                                keys_written += part1.len() as u64;
-                                new_sst_files.push(create_sst_file(
-                                    &self.parallel_scheduler,
-                                    part1,
-                                    // We don't know the exact sizes so we estimate them
-                                    last_entries_total_key_size / 2,
-                                    path,
-                                    seq1,
-                                )?);
-
-                                keys_written += part2.len() as u64;
-                                new_sst_files.push(create_sst_file(
-                                    &self.parallel_scheduler,
-                                    part2,
-                                    last_entries_total_key_size / 2,
-                                    path,
-                                    seq2,
-                                )?);
-                            }
-                            Ok(PartialMergeResult::Merged {
-                                new_sst_files,
-                                blob_seq_numbers_to_delete,
-                                keys_written,
-                            })
-                        })
+                                // If we have one set of entries left, write them to a new SST file
+                                if last_entries.is_empty() && !entries.is_empty() {
+                                    let seq = sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
+
+                                    keys_written += entries.len() as u64;
+                                    new_sst_files.push(create_sst_file(
+                                        &self.parallel_scheduler,
+                                        &entries,
+                                        total_key_size,
+                                        path,
+                                        seq,
+                                    )?);
+                                } else
+                                // If we have two sets of entries left, merge them and
+                                // split it into two SST files, to avoid having a
+                                // single SST file that is very small.
+                                if !last_entries.is_empty() {
+                                    last_entries.append(&mut entries);
+
+                                    last_entries_total_key_size += total_key_size;
+
+                                    let (part1, part2) =
+                                        last_entries.split_at(last_entries.len() / 2);
+
+                                    let seq1 = sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
+                                    let seq2 = sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
+
+                                    keys_written += part1.len() as u64;
+                                    new_sst_files.push(create_sst_file(
+                                        &self.parallel_scheduler,
+                                        part1,
+                                        // We don't know the exact sizes so we estimate them
+                                        last_entries_total_key_size / 2,
+                                        path,
+                                        seq1,
+                                    )?);
+
+                                    keys_written += part2.len() as u64;
+                                    new_sst_files.push(create_sst_file(
+                                        &self.parallel_scheduler,
+                                        part2,
+                                        last_entries_total_key_size / 2,
+                                        path,
+                                        seq2,
+                                    )?);
+                                }
+                                Ok(PartialMergeResult::Merged {
+                                    new_sst_files,
+                                    blob_seq_numbers_to_delete,
+                                    keys_written,
+                                    indicies,
+                                })
+                            },
+                        )
                         .with_context(|| {
                             format!("Failed to merge database files for family {family}")
                         })?;
@@ -1072,6 +1112,7 @@ impl<S: ParallelScheduler> TurboPersistence<S> {
                             if let PartialMergeResult::Merged {
                                 new_sst_files,
                                 blob_seq_numbers_to_delete,
+                                indicies: _,
                                 keys_written: _,
                             } = r
                             {
@@ -1088,43 +1129,91 @@ impl<S: ParallelScheduler> TurboPersistence<S> {
                     let mut new_sst_files = Vec::with_capacity(sst_files_len);
                     let mut blob_seq_numbers_to_delete = Vec::with_capacity(blob_delete_len);
 
+                    let meta_seq = sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
                     let mut meta_file_builder = MetaFileBuilder::new(family);
 
                     let mut keys_written = 0;
-                    for result in merge_result {
-                        match result {
-                            PartialMergeResult::Merged {
-                                new_sst_files: merged_new_sst_files,
-                                blob_seq_numbers_to_delete: merged_blob_seq_numbers_to_delete,
-                                keys_written: merged_keys_written,
-                            } => {
-                                for (seq, file, meta) in merged_new_sst_files {
+                    self.parallel_scheduler.block_in_place(|| {
+                        let guard = log_mutex.lock();
+                        let mut log = self.open_log()?;
+                        writeln!(
+                            log,
+                            "{family:3} | {meta_seq:08} | Compaction (coverage: {}, overlap: {}, \
+                             duplication: {} / {} MiB):",
+                            metrics.coverage,
+                            metrics.overlap,
+                            metrics.duplication,
+                            metrics.duplicated_size / 1024 / 1024
+                        )?;
+                        for result in merge_result {
+                            match result {
+                                PartialMergeResult::Merged {
+                                    new_sst_files: merged_new_sst_files,
+                                    blob_seq_numbers_to_delete: merged_blob_seq_numbers_to_delete,
+                                    keys_written: merged_keys_written,
+                                    indicies,
+                                } => {
+                                    writeln!(
+                                        log,
+                                        "{family:3} | {meta_seq:08} | MERGE \
+                                         ({merged_keys_written} keys):"
+                                    )?;
+                                    for i in indicies.iter() {
+                                        let seq = ssts_with_ranges[*i].seq;
+                                        let (min, max) = ssts_with_ranges[*i].range().into_inner();
+                                        writeln!(
+                                            log,
+                                            "{family:3} | {meta_seq:08} | {seq:08} INPUT  | {} ",
+                                            range_to_str(min, max)
+                                        )?;
+                                    }
+                                    for (seq, file, meta) in merged_new_sst_files {
+                                        let min = meta.min_hash;
+                                        let max = meta.max_hash;
+                                        writeln!(
+                                            log,
+                                            "{family:3} | {meta_seq:08} | {seq:08} OUTPUT | {}",
+                                            range_to_str(min, max)
+                                        )?;
+
+                                        meta_file_builder.add(seq, meta);
+                                        new_sst_files.push((seq, file));
+                                    }
+                                    blob_seq_numbers_to_delete
+                                        .extend(merged_blob_seq_numbers_to_delete);
+                                    keys_written += merged_keys_written;
+                                }
+                                PartialMergeResult::Move { seq, meta } => {
+                                    let min = meta.min_hash;
+                                    let max = meta.max_hash;
+                                    writeln!(
+                                        log,
+                                        "{family:3} | {meta_seq:08} | {seq:08} MOVED  | {}",
+                                        range_to_str(min, max)
+                                    )?;
+
                                     meta_file_builder.add(seq, meta);
-                                    new_sst_files.push((seq, file));
                                 }
-                                blob_seq_numbers_to_delete
-                                    .extend(merged_blob_seq_numbers_to_delete);
-                                keys_written += merged_keys_written;
-                            }
-                            PartialMergeResult::Move { seq, meta } => {
-                                meta_file_builder.add(seq, meta);
                             }
                         }
-                    }
+                        drop(log);
+                        drop(guard);
+
+                        anyhow::Ok(())
+                    })?;
 
                     for &seq in sst_seq_numbers_to_delete.iter() {
                         meta_file_builder.add_obsolete_sst_file(seq);
                     }
 
-                    let seq = sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
                     let meta_file = {
                         let _span = tracing::trace_span!("write meta file").entered();
                         self.parallel_scheduler
-                            .block_in_place(|| meta_file_builder.write(&self.path, seq))?
+                            .block_in_place(|| meta_file_builder.write(&self.path, meta_seq))?
                     };
 
                     Ok(PartialResultPerFamily {
-                        new_meta_file: Some((seq, meta_file)),
+                        new_meta_file: Some((meta_seq, meta_file)),
                         new_sst_files,
                         sst_seq_numbers_to_delete,
                         blob_seq_numbers_to_delete,
@@ -1273,6 +1362,30 @@ impl<S: ParallelScheduler> TurboPersistence<S> {
     }
 }
 
+fn range_to_str(min: u64, max: u64) -> String {
+    use std::fmt::Write;
+    const DISPLAY_SIZE: usize = 100;
+    const TOTAL_SIZE: u64 = u64::MAX;
+    let start_pos = (min as u128 * DISPLAY_SIZE as u128 / TOTAL_SIZE as u128) as usize;
+    let end_pos = (max as u128 * DISPLAY_SIZE as u128 / TOTAL_SIZE as u128) as usize;
+    let mut range_str = String::new();
+    for i in 0..DISPLAY_SIZE {
+        if i == start_pos && i == end_pos {
+            range_str.push('O');
+        } else if i == start_pos {
+            range_str.push('[');
+        } else if i == end_pos {
+            range_str.push(']');
+        } else if i > start_pos && i < end_pos {
+            range_str.push('=');
+        } else {
+            range_str.push(' ');
+        }
+    }
+    write!(range_str, " | {min:016x}-{max:016x}").unwrap();
+    range_str
+}
+
 pub struct MetaFileInfo {
     pub sequence_number: u32,
     pub family: u32,