Add a new picking algorithm in fifo compaction (#14326)

Summary:
Add a new kv ratio based compaction picking algorithm in fifo compaction

Pull Request resolved: https://github.com/facebook/rocksdb/pull/14326

Test Plan: Unit test

Reviewed By: pdillinger

Differential Revision: D93257941

Pulled By: xingbowang

fbshipit-source-id: fd2d0e1356c7b54682a1197475a1bd26cb45c9d4

db/c.cc

@@ -7047,6 +7047,27 @@ uint64_t rocksdb_fifo_compaction_options_get_max_table_files_size(
   return fifo_opts->rep.max_table_files_size;
 }
 
+void rocksdb_fifo_compaction_options_set_max_data_files_size(
+    rocksdb_fifo_compaction_options_t* fifo_opts, uint64_t size) {
+  fifo_opts->rep.max_data_files_size = size;
+}
+
+uint64_t rocksdb_fifo_compaction_options_get_max_data_files_size(
+    rocksdb_fifo_compaction_options_t* fifo_opts) {
+  return fifo_opts->rep.max_data_files_size;
+}
+
+void rocksdb_fifo_compaction_options_set_use_kv_ratio_compaction(
+    rocksdb_fifo_compaction_options_t* fifo_opts,
+    unsigned char use_kv_ratio_compaction) {
+  fifo_opts->rep.use_kv_ratio_compaction = use_kv_ratio_compaction;
+}
+
+unsigned char rocksdb_fifo_compaction_options_get_use_kv_ratio_compaction(
+    rocksdb_fifo_compaction_options_t* fifo_opts) {
+  return fifo_opts->rep.use_kv_ratio_compaction;
+}
+
 void rocksdb_fifo_compaction_options_destroy(
     rocksdb_fifo_compaction_options_t* fifo_opts) {
   delete fifo_opts;

db/c_test.c

@@ -3596,6 +3596,14 @@ int main(int argc, char** argv) {
         100000 ==
         rocksdb_fifo_compaction_options_get_max_table_files_size(fco));
 
+    rocksdb_fifo_compaction_options_set_max_data_files_size(fco, 200000);
+    CheckCondition(
+        200000 == rocksdb_fifo_compaction_options_get_max_data_files_size(fco));
+
+    rocksdb_fifo_compaction_options_set_use_kv_ratio_compaction(fco, 1);
+    CheckCondition(
+        1 == rocksdb_fifo_compaction_options_get_use_kv_ratio_compaction(fco));
+
     rocksdb_fifo_compaction_options_destroy(fco);
   }
 

db/column_family.cc

@@ -401,7 +401,13 @@ ColumnFamilyOptions SanitizeCfOptions(const ImmutableDBOptions& db_options,
   }
 
   if (result.max_compaction_bytes == 0) {
-    result.max_compaction_bytes = result.target_file_size_base * 25;
+    // For FIFO with use_kv_ratio_compaction, leave max_compaction_bytes as 0
+    // to signal "auto-calculate target from capacity and SST/blob ratio."
+    // When explicitly set by the user, it overrides the auto-calculated target.
+    if (result.compaction_style != kCompactionStyleFIFO ||
+        !result.compaction_options_fifo.use_kv_ratio_compaction) {
+      result.max_compaction_bytes = result.target_file_size_base * 25;
+    }
   }
 
   bool is_block_based_table = (result.table_factory->IsInstanceOf(
@@ -1563,6 +1569,34 @@ Status ColumnFamilyData::ValidateOptions(
         "FIFO compaction only supported with max_open_files = -1.");
   }
 
+  if (cf_options.compaction_options_fifo.use_kv_ratio_compaction) {
+    if (cf_options.compaction_style != kCompactionStyleFIFO) {
+      return Status::InvalidArgument(
+          "use_kv_ratio_compaction is only supported with FIFO compaction "
+          "style.");
+    }
+    if (!cf_options.compaction_options_fifo.allow_compaction) {
+      return Status::InvalidArgument(
+          "use_kv_ratio_compaction requires allow_compaction = true. "
+          "allow_compaction enables intra-L0 compaction, and "
+          "use_kv_ratio_compaction selects the picking strategy.");
+    }
+    if (cf_options.compaction_options_fifo.max_data_files_size == 0) {
+      return Status::InvalidArgument(
+          "use_kv_ratio_compaction requires max_data_files_size > 0 to "
+          "compute the target compacted file size from data capacity.");
+    }
+  }
+
+  if (cf_options.compaction_options_fifo.max_data_files_size > 0 &&
+      cf_options.compaction_options_fifo.max_data_files_size <
+          cf_options.compaction_options_fifo.max_table_files_size) {
+    return Status::InvalidArgument(
+        "max_data_files_size (total data = SST + blob) must be >= "
+        "max_table_files_size (SST only) when non-zero, since total data "
+        "always includes SST data.");
+  }
+
   std::vector<uint32_t> supported{0, 1, 2, 4, 8};
   if (std::find(supported.begin(), supported.end(),
                 cf_options.memtable_protection_bytes_per_key) ==

db/compaction/compaction_picker.cc

@@ -27,12 +27,11 @@
 
 namespace ROCKSDB_NAMESPACE {
 
-bool FindIntraL0Compaction(const std::vector<FileMetaData*>& level_files,
+bool PickCostBasedIntraL0Compaction(
-                           size_t min_files_to_compact,
+    const std::vector<FileMetaData*>& level_files, size_t min_files_to_compact,
-                           uint64_t max_compact_bytes_per_del_file,
+    uint64_t max_compact_bytes_per_del_file, uint64_t max_compaction_bytes,
-                           uint64_t max_compaction_bytes,
+    CompactionInputFiles* comp_inputs) {
-                           CompactionInputFiles* comp_inputs) {
+  TEST_SYNC_POINT("PickCostBasedIntraL0Compaction");
-  TEST_SYNC_POINT("FindIntraL0Compaction");
 
   size_t start = 0;
 

db/compaction/compaction_picker.h

@@ -328,11 +328,10 @@ class NullCompactionPicker : public CompactionPicker {
 //                                        files. Cannot be nullptr.
 //
 // @return                                true iff compaction was found.
-bool FindIntraL0Compaction(const std::vector<FileMetaData*>& level_files,
+bool PickCostBasedIntraL0Compaction(
-                           size_t min_files_to_compact,
+    const std::vector<FileMetaData*>& level_files, size_t min_files_to_compact,
-                           uint64_t max_compact_bytes_per_del_file,
+    uint64_t max_compact_bytes_per_del_file, uint64_t max_compaction_bytes,
-                           uint64_t max_compaction_bytes,
+    CompactionInputFiles* comp_inputs);
-                           CompactionInputFiles* comp_inputs);
 
 CompressionType GetCompressionType(const VersionStorageInfo* vstorage,
                                    const MutableCFOptions& mutable_cf_options,

db/compaction/compaction_picker_fifo.cc

@@ -9,6 +9,7 @@
 
 #include "db/compaction/compaction_picker_fifo.h"
 
+#include <algorithm>
 #include <cinttypes>
 #include <string>
 #include <vector>
@@ -31,6 +32,29 @@ uint64_t GetTotalFilesSize(const std::vector<FileMetaData*>& files) {
   }
   return total_size;
 }
+
+// Compute effective data size and capacity limit for FIFO compaction.
+// When max_data_files_size > 0 (blob-aware mode), the effective size includes
+// both SST and blob file sizes, and the limit is max_data_files_size.
+// Otherwise, only SST sizes are used with max_table_files_size as the limit.
+void GetEffectiveSizeAndLimit(const CompactionOptionsFIFO& fifo_opts,
+                              uint64_t total_sst_size, uint64_t total_blob_size,
+                              uint64_t* effective_size,
+                              uint64_t* effective_max) {
+  *effective_size = total_sst_size;
+  *effective_max = fifo_opts.max_table_files_size;
+  if (fifo_opts.max_data_files_size > 0) {
+    *effective_size += total_blob_size;
+    *effective_max = fifo_opts.max_data_files_size;
+  }
+}
+
+// Return the effective capacity limit for FIFO compaction.
+// Convenience wrapper when only the limit is needed (e.g., PickTTLCompaction).
+uint64_t GetEffectiveMax(const CompactionOptionsFIFO& fifo_opts) {
+  return fifo_opts.max_data_files_size > 0 ? fifo_opts.max_data_files_size
+                                           : fifo_opts.max_table_files_size;
+}
 }  // anonymous namespace
 
 bool FIFOCompactionPicker::NeedsCompaction(
@@ -98,10 +122,43 @@ Compaction* FIFOCompactionPicker::PickTTLCompaction(
   // Return a nullptr and proceed to size-based FIFO compaction if:
   // 1. there are no files older than ttl OR
   // 2. there are a few files older than ttl, but deleting them will not bring
-  //    the total size to be less than max_table_files_size threshold.
+  //    the total size to be less than the size threshold.
-  if (inputs[0].files.empty() ||
+  uint64_t effective_max =
-      total_size >
+      GetEffectiveMax(mutable_cf_options.compaction_options_fifo);
-          mutable_cf_options.compaction_options_fifo.max_table_files_size) {
+  // Estimate the effective remaining data after dropping TTL-expired SSTs.
+  // Each dropped SST also frees a proportional share of blob data.
+  //
+  // In multi-level FIFO (migration), we must use total SST across ALL levels
+  // as the reference, because total_blob covers all levels. Using only L0
+  // SST would inflate the blob estimate.
+  uint64_t effective_remaining = total_size;
+  if (mutable_cf_options.compaction_options_fifo.max_data_files_size > 0) {
+    uint64_t total_blob = vstorage->GetBlobStats().total_file_size;
+    // Compute total SST across all levels so the reference scope matches
+    // total_blob's scope (all levels).
+    uint64_t total_sst_all_levels = GetTotalFilesSize(level_files);
+    for (int level = 1; level < vstorage->num_levels(); ++level) {
+      total_sst_all_levels += GetTotalFilesSize(vstorage->LevelFiles(level));
+    }
+    // remaining_sst_all = total_sst_all - dropped_l0_sst
+    // total_size is the remaining L0 SST after removing expired files;
+    // original L0 SST minus remaining L0 SST = dropped.
+    uint64_t original_l0_sst = GetTotalFilesSize(level_files);
+    uint64_t dropped_sst = original_l0_sst - total_size;
+    uint64_t remaining_sst_all = total_sst_all_levels - dropped_sst;
+    // Proportional blob estimate: each SST byte "owns" a proportional
+    // share of blob bytes. Both reference sizes must come from the same
+    // scope (all levels) to avoid inflated estimates.
+    if (total_sst_all_levels > 0 && total_blob > 0) {
+      effective_remaining =
+          remaining_sst_all +
+          static_cast<uint64_t>(static_cast<double>(remaining_sst_all) /
+                                total_sst_all_levels * total_blob);
+    } else {
+      effective_remaining = remaining_sst_all;
+    }
+  }
+  if (inputs[0].files.empty() || effective_remaining > effective_max) {
     return nullptr;
   }
 
@@ -151,7 +208,9 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
     const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
     const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
     LogBuffer* log_buffer) {
-  // compute the total size and identify the last non-empty level
+  const auto& fifo_opts = mutable_cf_options.compaction_options_fifo;
+
+  // compute the total SST size and identify the last non-empty level
   int last_level = 0;
   uint64_t total_size = 0;
   for (int level = 0; level < vstorage->num_levels(); ++level) {
@@ -164,52 +223,13 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
   const std::vector<FileMetaData*>& last_level_files =
       vstorage->LevelFiles(last_level);
 
-  if (last_level == 0 &&
+  // Compute effective size and limit for comparison.
-      total_size <=
+  uint64_t effective_size, effective_max;
-          mutable_cf_options.compaction_options_fifo.max_table_files_size) {
+  GetEffectiveSizeAndLimit(fifo_opts, total_size,
-    // total size not exceeded, try to find intra level 0 compaction if enabled
+                           vstorage->GetBlobStats().total_file_size,
-    const std::vector<FileMetaData*>& level0_files = vstorage->LevelFiles(0);
+                           &effective_size, &effective_max);
-    if (mutable_cf_options.compaction_options_fifo.allow_compaction &&
-        level0_files.size() > 0) {
-      CompactionInputFiles comp_inputs;
-      // try to prevent same files from being compacted multiple times, which
-      // could produce large files that may never TTL-expire. Achieve this by
-      // disallowing compactions with files larger than memtable (inflate its
-      // size by 10% to account for uncompressed L0 files that may have size
-      // slightly greater than memtable size limit).
-      size_t max_compact_bytes_per_del_file =
-          static_cast<size_t>(MultiplyCheckOverflow(
-              static_cast<uint64_t>(mutable_cf_options.write_buffer_size),
-              1.1));
-      if (FindIntraL0Compaction(
-              level0_files,
-              mutable_cf_options
-                  .level0_file_num_compaction_trigger /* min_files_to_compact */
-              ,
-              max_compact_bytes_per_del_file,
-              mutable_cf_options.max_compaction_bytes, &comp_inputs)) {
-        Compaction* c = new Compaction(
-            vstorage, ioptions_, mutable_cf_options, mutable_db_options,
-            {comp_inputs}, 0, 16 * 1024 * 1024 /* output file size limit */,
-            0 /* max compaction bytes, not applicable */,
-            0 /* output path ID */, mutable_cf_options.compression,
-            mutable_cf_options.compression_opts, Temperature::kUnknown,
-            0 /* max_subcompactions */, {},
-            /* earliest_snapshot */ std::nullopt,
-            /* snapshot_checker */ nullptr,
-            CompactionReason::kFIFOReduceNumFiles,
-            /* trim_ts */ "", vstorage->CompactionScore(0),
-            /* l0_files_might_overlap */ true);
-        return c;
-      }
-    }
 
-    ROCKS_LOG_BUFFER(
+  if (last_level == 0 && effective_size <= effective_max) {
-        log_buffer,
-        "[%s] FIFO compaction: nothing to do. Total size %" PRIu64
-        ", max size %" PRIu64 "\n",
-        cf_name.c_str(), total_size,
-        mutable_cf_options.compaction_options_fifo.max_table_files_size);
     return nullptr;
   }
 
@@ -227,11 +247,29 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
   inputs[0].level = last_level;
 
   if (last_level == 0) {
+    // When using blob-aware sizing, use proportional estimation (same
+    // principle as EstimateTotalDataForSST): each SST "owns"
+    // effective_size / num_files of total data. This is an approximation
+    // — individual SSTs may reference different amounts of blob data,
+    // but uniform distribution is a reasonable estimate for FIFO dropping.
+    uint64_t remaining_size = effective_size;
+    const uint64_t num_files = last_level_files.size();
+    // Proportional estimate of data per file (SST + blob).
+    // Use max(1) to prevent stalling when effective_size < num_files.
+    const uint64_t data_per_file =
+        (fifo_opts.max_data_files_size > 0 && num_files > 0)
+            ? std::max(effective_size / num_files, uint64_t{1})
+            : 0;
+
     // In L0, right-most files are the oldest files.
     for (auto ritr = last_level_files.rbegin(); ritr != last_level_files.rend();
          ++ritr) {
       auto f = *ritr;
-      total_size -= f->fd.file_size;
+      if (fifo_opts.max_data_files_size > 0) {
+        remaining_size -= std::min(remaining_size, data_per_file);
+      } else {
+        remaining_size -= std::min(remaining_size, f->fd.file_size);
+      }
       inputs[0].files.push_back(f);
       char tmp_fsize[16];
       AppendHumanBytes(f->fd.GetFileSize(), tmp_fsize, sizeof(tmp_fsize));
@@ -239,13 +277,11 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
                        "[%s] FIFO compaction: picking file %" PRIu64
                        " with size %s for deletion",
                        cf_name.c_str(), f->fd.GetNumber(), tmp_fsize);
-      if (total_size <=
+      if (remaining_size <= effective_max) {
-          mutable_cf_options.compaction_options_fifo.max_table_files_size) {
         break;
       }
     }
-  } else if (total_size >
+  } else if (effective_size > effective_max) {
-             mutable_cf_options.compaction_options_fifo.max_table_files_size) {
     // If the last level is non-L0, we actually don't know which file is
     // logically the oldest since the file creation time only represents
     // when this file was compacted to this level, which is independent
@@ -255,34 +291,36 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
     // file with the smallest key will be deleted first.  This design decision
     // better serves a major type of FIFO use cases where smaller keys are
     // associated with older data.
+    const uint64_t num_files = last_level_files.size();
+    // Proportional estimate of data per file (SST + blob), same as L0 path.
+    const uint64_t data_per_file =
+        (fifo_opts.max_data_files_size > 0 && num_files > 0)
+            ? std::max(effective_size / num_files, uint64_t{1})
+            : 0;
     for (const auto& f : last_level_files) {
       if (f->being_compacted) {
         continue;
       }
-      total_size -= f->fd.file_size;
+      if (fifo_opts.max_data_files_size > 0) {
+        effective_size -= std::min(effective_size, data_per_file);
+      } else {
+        effective_size -= std::min(effective_size, f->fd.file_size);
+      }
       inputs[0].files.push_back(f);
       char tmp_fsize[16];
       AppendHumanBytes(f->fd.GetFileSize(), tmp_fsize, sizeof(tmp_fsize));
-      ROCKS_LOG_BUFFER(
+      ROCKS_LOG_BUFFER(log_buffer,
-          log_buffer,
+                       "[%s] FIFO compaction: picking file %" PRIu64
-          "[%s] FIFO compaction: picking file %" PRIu64
+                       " with size %s for deletion under total size %" PRIu64
-          " with size %s for deletion under total size %" PRIu64
+                       " vs max size %" PRIu64,
-          " vs max table files size %" PRIu64,
+                       cf_name.c_str(), f->fd.GetNumber(), tmp_fsize,
-          cf_name.c_str(), f->fd.GetNumber(), tmp_fsize, total_size,
+                       effective_size, effective_max);
-          mutable_cf_options.compaction_options_fifo.max_table_files_size);
 
-      if (total_size <=
+      if (effective_size <= effective_max) {
-          mutable_cf_options.compaction_options_fifo.max_table_files_size) {
         break;
       }
     }
   } else {
-    ROCKS_LOG_BUFFER(
-        log_buffer,
-        "[%s] FIFO compaction: nothing to do. Total size %" PRIu64
-        ", max size %" PRIu64 "\n",
-        cf_name.c_str(), total_size,
-        mutable_cf_options.compaction_options_fifo.max_table_files_size);
     return nullptr;
   }
 
@@ -419,6 +457,249 @@ Compaction* FIFOCompactionPicker::PickTemperatureChangeCompaction(
   return c;
 }
 
+Compaction* FIFOCompactionPicker::PickIntraL0Compaction(
+    const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
+    const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
+    LogBuffer* log_buffer) {
+  const auto& fifo_opts = mutable_cf_options.compaction_options_fifo;
+
+  if (!fifo_opts.allow_compaction) {
+    return nullptr;
+  }
+
+  const std::vector<FileMetaData*>& level0_files = vstorage->LevelFiles(0);
+  if (level0_files.empty()) {
+    return nullptr;
+  }
+
+  if (fifo_opts.use_kv_ratio_compaction) {
+    return PickRatioBasedIntraL0Compaction(
+        cf_name, mutable_cf_options, mutable_db_options, vstorage, log_buffer);
+  }
+
+  // Old intra-L0 path: merge small files using PickCostBasedIntraL0Compaction.
+  // Minimum files to compact follows level0_file_num_compaction_trigger.
+  // Try to prevent same files from being compacted multiple times, which
+  // could produce large files that may never TTL-expire. Achieve this by
+  // disallowing compactions with files larger than memtable (inflate its
+  // size by 10% to account for uncompressed L0 files that may have size
+  // slightly greater than memtable size limit).
+
+  CompactionInputFiles comp_inputs;
+  size_t max_compact_bytes_per_del_file =
+      static_cast<size_t>(MultiplyCheckOverflow(
+          static_cast<uint64_t>(mutable_cf_options.write_buffer_size), 1.1));
+  if (PickCostBasedIntraL0Compaction(
+          level0_files,
+          mutable_cf_options
+              .level0_file_num_compaction_trigger /* min_files_to_compact */,
+          max_compact_bytes_per_del_file,
+          mutable_cf_options.max_compaction_bytes, &comp_inputs)) {
+    Compaction* c = new Compaction(
+        vstorage, ioptions_, mutable_cf_options, mutable_db_options,
+        {comp_inputs}, 0, 16 * 1024 * 1024 /* output file size limit */,
+        0 /* max compaction bytes, not applicable */, 0 /* output path ID */,
+        mutable_cf_options.compression, mutable_cf_options.compression_opts,
+        Temperature::kUnknown, 0 /* max_subcompactions */, {},
+        /* earliest_snapshot */ std::nullopt,
+        /* snapshot_checker */ nullptr, CompactionReason::kFIFOReduceNumFiles,
+        /* trim_ts */ "", vstorage->CompactionScore(0),
+        /* l0_files_might_overlap */ true);
+    return c;
+  }
+
+  return nullptr;
+}
+
+Compaction* FIFOCompactionPicker::PickRatioBasedIntraL0Compaction(
+    const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
+    const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
+    LogBuffer* log_buffer) {
+  const auto& fifo_opts = mutable_cf_options.compaction_options_fifo;
+  assert(fifo_opts.use_kv_ratio_compaction);
+  assert(fifo_opts.max_data_files_size > 0);
+
+  // During migration from level/universal compaction to FIFO, non-L0 levels
+  // may still contain files. The ratio-based algorithm only operates on L0,
+  // so skip it until PickSizeCompaction has drained all non-L0 levels.
+  // Once levels collapse to L0-only, this algorithm will kick in.
+  for (int level = 1; level < vstorage->num_levels(); ++level) {
+    if (!vstorage->LevelFiles(level).empty()) {
+      ROCKS_LOG_BUFFER(log_buffer,
+                       "[%s] FIFO kv-ratio compaction: skipping — non-L0 "
+                       "level %d still has %" ROCKSDB_PRIszt
+                       " files (migration in progress)",
+                       cf_name.c_str(), level,
+                       vstorage->LevelFiles(level).size());
+      return nullptr;
+    }
+  }
+
+  if (!level0_compactions_in_progress_.empty()) {
+    return nullptr;
+  }
+
+  const std::vector<FileMetaData*>& level0_files = vstorage->LevelFiles(0);
+  if (mutable_cf_options.level0_file_num_compaction_trigger <= 1) {
+    // trigger <= 0 is invalid; trigger == 1 means compact after every flush,
+    // which doesn't make sense for tiered merging (the tier boundary loop
+    // divides by trigger, so trigger == 1 would cause an infinite loop).
+    return nullptr;
+  }
+  const size_t trigger = static_cast<size_t>(
+      mutable_cf_options.level0_file_num_compaction_trigger);
+  if (level0_files.size() < trigger) {
+    return nullptr;
+  }
+
+  // Determine the target compacted file size.
+  //
+  // When max_compaction_bytes > 0 (explicitly set by user), use it directly
+  // as the target. This allows users to override the auto-calculated value.
+  //
+  // When max_compaction_bytes == 0 (default), auto-calculate from the data
+  // capacity and observed SST/blob ratio:
+  //   target = max_data_files_size * sst_ratio / trigger
+  //
+  // This is recomputed on every PickCompaction call. The computation is
+  // trivial (sum file sizes + arithmetic) and PickCompaction is only called
+  // once per flush or compaction completion, so no caching is needed.
+  uint64_t target = 0;
+  if (mutable_cf_options.max_compaction_bytes > 0) {
+    // User explicitly set max_compaction_bytes — use it as target
+    target = mutable_cf_options.max_compaction_bytes;
+  } else {
+    // Auto-calculate from capacity and observed SST/blob ratio
+    uint64_t total_sst = GetTotalFilesSize(level0_files);
+    uint64_t total_blob = vstorage->GetBlobStats().total_file_size;
+    uint64_t total_data = total_sst + total_blob;
+
+    if (total_data == 0 || total_sst == 0) {
+      return nullptr;
+    }
+
+    // Compute sst_ratio (inverse of EstimateTotalDataForSST's proportion):
+    // when no blob files exist, sst_ratio is 1.0 and the target becomes
+    // max_data_files_size / trigger, which is large. The algorithm will
+    // naturally not find small enough files to compact.
+    double sst_ratio =
+        (total_blob > 0) ? static_cast<double>(total_sst) / total_data : 1.0;
+
+    uint64_t total_sst_at_cap =
+        static_cast<uint64_t>(fifo_opts.max_data_files_size * sst_ratio);
+    target = total_sst_at_cap / trigger;
+
+    ROCKS_LOG_BUFFER(log_buffer,
+                     "[%s] FIFO ratio-based compaction: sst_ratio=%.4f, "
+                     "target_file_size=%" PRIu64,
+                     cf_name.c_str(), sst_ratio, target);
+  }
+  if (target == 0) {
+    return nullptr;
+  }
+
+  // Tiered size-based file selection.
+  //
+  // Tier boundaries form a geometric sequence descending from target:
+  //   ..., target/trigger^2, target/trigger, target
+  // For each boundary (smallest first), find contiguous L0 files with
+  // size < boundary. If their accumulated bytes >= boundary, merge them.
+  // The output (~boundary bytes) advances to the next tier. Files that
+  // reach target are "graduated" and never compacted again.
+  //
+  // Trade-off: write amplification vs L0 file count.
+  //
+  // Write amp: O(log(target/flush) / log(trigger)) per byte, instead of
+  //   O(target / (trigger * flush)) from flat merging. Each byte is
+  //   rewritten once per tier crossing.
+  //
+  // L0 file count: trigger + k * (trigger - 1) at steady state, where
+  //   k = ceil(log(target/flush) / log(trigger)). This is higher than
+  //   the original trigger target because intermediate tier files
+  //   accumulate while waiting for the next tier merge. The trade-off
+  //   is explicit: more L0 files in exchange for logarithmic (instead
+  //   of linear) write amplification.
+
+  // Build tier boundaries from smallest to largest.
+  // Stop at 10KB minimum — SST files of most workloads are larger than
+  // this, so lower boundaries would only waste CPU scanning L0 files.
+  // Files smaller than the lowest boundary simply merge at that boundary.
+  static constexpr uint64_t kMinTierBoundary = 10 * 1024;  // 10KB
+  std::vector<uint64_t> boundaries;
+  for (uint64_t b = target; b >= kMinTierBoundary; b /= trigger) {
+    boundaries.push_back(b);
+  }
+  if (boundaries.empty()) {
+    // target itself is below kMinTierBoundary — use target as the
+    // sole boundary so we can still compact at the target size.
+    boundaries.push_back(target);
+  }
+  std::reverse(boundaries.begin(), boundaries.end());
+
+  // For each tier boundary (smallest first), scan L0 for mergeable batches.
+  // L0 files are stored newest-first; oldest is at the end.
+  for (const uint64_t boundary : boundaries) {
+    for (size_t scan = level0_files.size(); scan > 0;) {
+      // Skip files >= boundary (they belong to higher tiers) or in-progress
+      if (level0_files[scan - 1]->fd.file_size >= boundary ||
+          level0_files[scan - 1]->being_compacted) {
+        --scan;
+        continue;
+      }
+
+      // Found a file < boundary — collect contiguous batch
+      std::vector<FileMetaData*> batch;
+      uint64_t accumulated = 0;
+      size_t pos = scan;
+      while (pos > 0 && level0_files[pos - 1]->fd.file_size < boundary &&
+             !level0_files[pos - 1]->being_compacted) {
+        // Don't let output exceed 2x boundary (prevent tier-skipping)
+        if (accumulated >= boundary &&
+            accumulated + level0_files[pos - 1]->fd.file_size > boundary * 2) {
+          break;
+        }
+        batch.push_back(level0_files[pos - 1]);
+        accumulated += level0_files[pos - 1]->fd.file_size;
+        --pos;
+      }
+
+      // Viable: >= 2 files and accumulated >= boundary
+      if (batch.size() >= 2 && accumulated >= boundary) {
+        CompactionInputFiles comp_inputs;
+        comp_inputs.level = 0;
+        comp_inputs.files = std::move(batch);
+
+        ROCKS_LOG_BUFFER(
+            log_buffer,
+            "[%s] FIFO kv-ratio compaction: picking %" ROCKSDB_PRIszt
+            " files (%" PRIu64 " bytes) at tier boundary %" PRIu64
+            " for intra-L0 compaction, target=%" PRIu64,
+            cf_name.c_str(), comp_inputs.files.size(), accumulated, boundary,
+            target);
+
+        Compaction* c = new Compaction(
+            vstorage, ioptions_, mutable_cf_options, mutable_db_options,
+            {comp_inputs}, 0, boundary /* output file size limit */,
+            0 /* max compaction bytes, not applicable */,
+            0 /* output path ID */, mutable_cf_options.compression,
+            mutable_cf_options.compression_opts, Temperature::kUnknown,
+            0 /* max_subcompactions */, {},
+            /* earliest_snapshot */ std::nullopt,
+            /* snapshot_checker */ nullptr,
+            CompactionReason::kFIFOReduceNumFiles,
+            /* trim_ts */ "", vstorage->CompactionScore(0),
+            /* l0_files_might_overlap */ true);
+        return c;
+      }
+
+      // This batch wasn't enough — advance past it
+      scan = pos;
+    }
+  }
+
+  return nullptr;
+}
+
 // The full_history_ts_low parameter is used to control bottommost file marking
 // for compaction when user-defined timestamps (UDT) are enabled.
 
@@ -441,10 +722,22 @@ Compaction* FIFOCompactionPicker::PickCompaction(
     c = PickSizeCompaction(cf_name, mutable_cf_options, mutable_db_options,
                            vstorage, log_buffer);
   }
+  // Intra-L0 compaction merges small files to reduce file count.
+  // It runs after size-based dropping: if PickSizeCompaction dropped files,
+  // it returned non-null and we skip this. Otherwise, we try to reduce
+  // L0 file count by merging small files together.
+  if (c == nullptr) {
+    c = PickIntraL0Compaction(cf_name, mutable_cf_options, mutable_db_options,
+                              vstorage, log_buffer);
+  }
   if (c == nullptr) {
     c = PickTemperatureChangeCompaction(
         cf_name, mutable_cf_options, mutable_db_options, vstorage, log_buffer);
   }
+  if (c == nullptr) {
+    ROCKS_LOG_BUFFER(log_buffer, "[%s] FIFO compaction: no compaction picked",
+                     cf_name.c_str());
+  }
   RegisterCompaction(c);
   return c;
 }

db/compaction/compaction_picker_fifo.h

@@ -55,6 +55,28 @@ class FIFOCompactionPicker : public CompactionPicker {
                                  VersionStorageInfo* version,
                                  LogBuffer* log_buffer);
 
+  // Intra-L0 compaction: merges small L0 files to reduce file count.
+  // Dispatches between two strategies based on configuration:
+  //   - use_kv_ratio_compaction = true: PickRatioBasedIntraL0Compaction
+  //   (BlobDB-optimized)
+  //   - use_kv_ratio_compaction = false: PickCostBasedIntraL0Compaction
+  //   (original)
+  // Only active when allow_compaction = true.
+  Compaction* PickIntraL0Compaction(const std::string& cf_name,
+                                    const MutableCFOptions& mutable_cf_options,
+                                    const MutableDBOptions& mutable_db_options,
+                                    VersionStorageInfo* vstorage,
+                                    LogBuffer* log_buffer);
+
+  // Capacity-derived intra-L0 compaction for BlobDB workloads.
+  // Uses the observed SST/blob ratio to compute a target file size,
+  // producing uniform files for predictable FIFO trimming.
+  // Called from PickIntraL0Compaction when use_kv_ratio_compaction = true.
+  Compaction* PickRatioBasedIntraL0Compaction(
+      const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
+      const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
+      LogBuffer* log_buffer);
+
   // Will pick one file to compact at a time, starting from the oldest file.
   Compaction* PickTemperatureChangeCompaction(
       const std::string& cf_name, const MutableCFOptions& mutable_cf_options,

db/compaction/compaction_picker_level.cc

@@ -914,10 +914,10 @@ bool LevelCompactionBuilder::PickIntraL0Compaction() {
     // resort to L0->L0 compaction yet.
     return false;
   }
-  return FindIntraL0Compaction(level_files, kMinFilesForIntraL0Compaction,
+  return PickCostBasedIntraL0Compaction(
-                               std::numeric_limits<uint64_t>::max(),
+      level_files, kMinFilesForIntraL0Compaction,
-                               mutable_cf_options_.max_compaction_bytes,
+      std::numeric_limits<uint64_t>::max(),
-                               &start_level_inputs_);
+      mutable_cf_options_.max_compaction_bytes, &start_level_inputs_);
 }
 
 bool LevelCompactionBuilder::PickSizeBasedIntraL0Compaction() {

db/db_compaction_test.cc

@@ -7840,7 +7840,7 @@ class DBCompactionTestL0FilesMisorderCorruption : public DBCompactionTest {
       options_.level0_file_num_compaction_trigger = 3;
 
       CompactionOptionsFIFO fifo_options;
-      if (compaction_path_to_test == "FindIntraL0Compaction" ||
+      if (compaction_path_to_test == "PickCostBasedIntraL0Compaction" ||
           compaction_path_to_test == "CompactRange") {
         fifo_options.allow_compaction = true;
       } else if (compaction_path_to_test == "CompactFile") {
@@ -7940,7 +7940,7 @@ class DBCompactionTestL0FilesMisorderCorruption : public DBCompactionTest {
 
   void SetupSyncPoints(const std::string& compaction_path_to_test) {
     compaction_path_sync_point_called_.store(false);
-    if (compaction_path_to_test == "FindIntraL0Compaction" &&
+    if (compaction_path_to_test == "PickCostBasedIntraL0Compaction" &&
         options_.compaction_style == CompactionStyle::kCompactionStyleLevel) {
       ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
           "PostPickFileToCompact", [&](void* arg) {
@@ -7950,7 +7950,7 @@ class DBCompactionTestL0FilesMisorderCorruption : public DBCompactionTest {
             *picked_file_to_compact = false;
           });
       ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
-          "FindIntraL0Compaction", [&](void* /*arg*/) {
+          "PickCostBasedIntraL0Compaction", [&](void* /*arg*/) {
             compaction_path_sync_point_called_.store(true);
           });
 
@@ -7986,12 +7986,12 @@ class DBCompactionTestL0FilesMisorderCorruption : public DBCompactionTest {
           "PickDeleteTriggeredCompactionReturnNonnullptr", [&](void* /*arg*/) {
             compaction_path_sync_point_called_.store(true);
           });
-    } else if ((compaction_path_to_test == "FindIntraL0Compaction" ||
+    } else if ((compaction_path_to_test == "PickCostBasedIntraL0Compaction" ||
                 compaction_path_to_test == "CompactRange") &&
                options_.compaction_style ==
                    CompactionStyle::kCompactionStyleFIFO) {
       ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
-          "FindIntraL0Compaction", [&](void* /*arg*/) {
+          "PickCostBasedIntraL0Compaction", [&](void* /*arg*/) {
             compaction_path_sync_point_called_.store(true);
           });
     }
@@ -8151,7 +8151,7 @@ TEST_F(DBCompactionTestL0FilesMisorderCorruption,
     IngestOneKeyValue(dbfull(), Key(i), "new", options_);
   }
 
-  SetupSyncPoints("FindIntraL0Compaction");
+  SetupSyncPoints("PickCostBasedIntraL0Compaction");
   ResumeCompactionThread();
 
   ASSERT_OK(dbfull()->TEST_WaitForCompact());
@@ -8284,7 +8284,8 @@ TEST_F(DBCompactionTestL0FilesMisorderCorruption,
 
 TEST_F(DBCompactionTestL0FilesMisorderCorruption,
        FlushAfterIntraL0FIFOCompactionWithIngestedFile) {
-  for (const std::string compaction_path_to_test : {"FindIntraL0Compaction"}) {
+  for (const std::string compaction_path_to_test :
+       {"PickCostBasedIntraL0Compaction"}) {
     SetupOptions(CompactionStyle::kCompactionStyleFIFO,
                  compaction_path_to_test);
     DestroyAndReopen(options_);

db/version_set.cc

@@ -3792,15 +3792,20 @@ void VersionStorageInfo::ComputeCompactionScore(
       }
 
       if (compaction_style_ == kCompactionStyleFIFO) {
-        auto max_table_files_size =
+        const auto& fifo_opts = mutable_cf_options.compaction_options_fifo;
-            mutable_cf_options.compaction_options_fifo.max_table_files_size;
+        uint64_t effective_size = total_size;
-        if (max_table_files_size == 0) {
+        uint64_t effective_max = fifo_opts.max_table_files_size;
-          // avoid divide 0
+        if (fifo_opts.max_data_files_size > 0) {
-          max_table_files_size = 1;
+          // Blob-aware: include blob file sizes in the total
+          effective_size += GetBlobStats().total_file_size;
+          effective_max = fifo_opts.max_data_files_size;
         }
-        score = static_cast<double>(total_size) / max_table_files_size;
+        if (effective_max == 0) {
-        if (score < 1 &&
+          // avoid divide 0
-            mutable_cf_options.compaction_options_fifo.allow_compaction) {
+          effective_max = 1;
+        }
+        score = static_cast<double>(effective_size) / effective_max;
+        if (score < 1 && fifo_opts.allow_compaction) {
           score = std::max(
               static_cast<double>(num_sorted_runs) /
                   mutable_cf_options.level0_file_num_compaction_trigger,

db_stress_tool/db_stress_common.h

@@ -161,6 +161,8 @@ DECLARE_uint64(periodic_compaction_seconds);
 DECLARE_string(daily_offpeak_time_utc);
 DECLARE_uint64(compaction_ttl);
 DECLARE_bool(fifo_allow_compaction);
+DECLARE_uint64(fifo_compaction_max_data_files_size_mb);
+DECLARE_bool(fifo_compaction_use_kv_ratio_compaction);
 DECLARE_bool(allow_concurrent_memtable_write);
 DECLARE_double(experimental_mempurge_threshold);
 DECLARE_bool(enable_write_thread_adaptive_yield);

db_stress_tool/db_stress_gflags.cc

@@ -415,6 +415,17 @@ DEFINE_bool(fifo_allow_compaction, false,
             "If true, set `Options::compaction_options_fifo.allow_compaction = "
             "true`. It only take effect when FIFO compaction is used.");
 
+DEFINE_uint64(fifo_compaction_max_data_files_size_mb, 0,
+              "If non-zero, set "
+              "`Options::compaction_options_fifo.max_data_files_size` to this "
+              "value (in MB). Only takes effect with FIFO compaction.");
+
+DEFINE_bool(fifo_compaction_use_kv_ratio_compaction, false,
+            "If true, set "
+            "`Options::compaction_options_fifo.use_kv_ratio_compaction = "
+            "true`. Requires fifo_allow_compaction and "
+            "fifo_compaction_max_data_files_size_mb > 0.");
+
 DEFINE_bool(allow_concurrent_memtable_write, false,
             "Allow multi-writers to update mem tables in parallel.");
 

db_stress_tool/db_stress_test_base.cc

@@ -4378,6 +4378,12 @@ void InitializeOptionsFromFlags(
       ROCKSDB_NAMESPACE::CompactionStyle::kCompactionStyleFIFO) {
     options.compaction_options_fifo.allow_compaction =
         FLAGS_fifo_allow_compaction;
+    if (FLAGS_fifo_compaction_max_data_files_size_mb > 0) {
+      options.compaction_options_fifo.max_data_files_size =
+          FLAGS_fifo_compaction_max_data_files_size_mb * 1024 * 1024;
+    }
+    options.compaction_options_fifo.use_kv_ratio_compaction =
+        FLAGS_fifo_compaction_use_kv_ratio_compaction;
   }
   options.compaction_pri =
       static_cast<ROCKSDB_NAMESPACE::CompactionPri>(FLAGS_compaction_pri);

include/rocksdb/advanced_options.h

@@ -128,6 +128,50 @@ struct CompactionOptionsFIFO {
   // not be used. The minmum buffer size must be at least 4KiB
   uint64_t trivial_copy_buffer_size = 4096;
 
+  // When non-zero, FIFO compaction uses the combined size of SST files and
+  // blob files for size-based trimming decisions. When the total data size
+  // (SST + blob) exceeds this limit, the oldest SST files are dropped along
+  // with their associated blob files.
+  //
+  // When non-zero, this takes precedence over max_table_files_size for all
+  // FIFO compaction decisions: size-based dropping, TTL threshold checks,
+  // and compaction score computation. max_table_files_size is ignored.
+  //
+  // When zero (default), FIFO compaction uses max_table_files_size which
+  // only considers SST file sizes, maintaining backward compatibility.
+  //
+  // This option is primarily intended for use with integrated BlobDB where
+  // blob files can represent a significant portion of the total data.
+  //
+  // Dynamically changeable through SetOptions() API.
+  // Default: 0 (use max_table_files_size behavior)
+  uint64_t max_data_files_size = 0;
+
+  // When true, enables a capacity-derived intra-L0 compaction strategy
+  // optimized for BlobDB workloads where SST files are much smaller than
+  // write_buffer_size. Uses the observed key/value size ratio (SST vs blob
+  // file sizes) to compute a target compacted file size, producing uniform
+  // files for predictable FIFO trimming.
+  //
+  // Uses level0_file_num_compaction_trigger as the target max L0 file count.
+  //
+  // When max_compaction_bytes is 0, the target is auto-calculated from the
+  // data capacity and observed SST/blob ratio. When max_compaction_bytes is
+  // explicitly set to a non-zero value, it overrides the auto-calculated
+  // target.
+  //
+  // Requires:
+  //   - allow_compaction = true (master switch for intra-L0 compaction)
+  //   - max_data_files_size > 0 (needed to compute the target file size)
+  // Setting this to true without these will fail option validation.
+  //
+  // When false, the old intra-L0 strategy is used if allow_compaction is
+  // true (PickCostBasedIntraL0Compaction with 1.1 * write_buffer_size guard).
+  //
+  // Dynamically changeable through SetOptions() API.
+  // Default: false
+  bool use_kv_ratio_compaction = false;
+
   CompactionOptionsFIFO() : max_table_files_size(1 * 1024 * 1024 * 1024) {}
   CompactionOptionsFIFO(uint64_t _max_table_files_size, bool _allow_compaction)
       : max_table_files_size(_max_table_files_size),
@@ -643,6 +687,15 @@ struct AdvancedColumnFamilyOptions {
   //
   // Default: target_file_size_base * 25
   //
+  // For FIFO compaction with use_kv_ratio_compaction=true:
+  // When set to 0 (and compaction_style is FIFO), the value is NOT sanitized
+  // to the default. Instead, the target compacted file size is automatically
+  // calculated from the data capacity (max_data_files_size) and observed
+  // SST/blob ratio. When explicitly set to a non-zero value, it overrides
+  // the auto-calculated target and is used directly as the max compaction
+  // input size. Note: for FIFO, this controls the output file size target,
+  // not a general compaction byte limit as in level/universal compaction.
+  //
   // Dynamically changeable through SetOptions() API
   uint64_t max_compaction_bytes = 0;
 

include/rocksdb/c.h

@@ -2804,6 +2804,19 @@ rocksdb_fifo_compaction_options_set_max_table_files_size(
 extern ROCKSDB_LIBRARY_API uint64_t
 rocksdb_fifo_compaction_options_get_max_table_files_size(
     rocksdb_fifo_compaction_options_t* fifo_opts);
+extern ROCKSDB_LIBRARY_API void
+rocksdb_fifo_compaction_options_set_max_data_files_size(
+    rocksdb_fifo_compaction_options_t* fifo_opts, uint64_t size);
+extern ROCKSDB_LIBRARY_API uint64_t
+rocksdb_fifo_compaction_options_get_max_data_files_size(
+    rocksdb_fifo_compaction_options_t* fifo_opts);
+extern ROCKSDB_LIBRARY_API void
+rocksdb_fifo_compaction_options_set_use_kv_ratio_compaction(
+    rocksdb_fifo_compaction_options_t* fifo_opts,
+    unsigned char use_kv_ratio_compaction);
+extern ROCKSDB_LIBRARY_API unsigned char
+rocksdb_fifo_compaction_options_get_use_kv_ratio_compaction(
+    rocksdb_fifo_compaction_options_t* fifo_opts);
 extern ROCKSDB_LIBRARY_API void rocksdb_fifo_compaction_options_destroy(
     rocksdb_fifo_compaction_options_t* fifo_opts);
 

java/rocksjni/compaction_options_fifo.cc

@@ -71,6 +71,54 @@ jboolean Java_org_rocksdb_CompactionOptionsFIFO_allowCompaction(JNIEnv*, jclass,
   return static_cast<jboolean>(opt->allow_compaction);
 }
 
+/*
+ * Class:     org_rocksdb_CompactionOptionsFIFO
+ * Method:    setMaxDataFilesSize
+ * Signature: (JJ)V
+ */
+void Java_org_rocksdb_CompactionOptionsFIFO_setMaxDataFilesSize(
+    JNIEnv*, jclass, jlong jhandle, jlong jmax_data_files_size) {
+  auto* opt =
+      reinterpret_cast<ROCKSDB_NAMESPACE::CompactionOptionsFIFO*>(jhandle);
+  opt->max_data_files_size = static_cast<uint64_t>(jmax_data_files_size);
+}
+
+/*
+ * Class:     org_rocksdb_CompactionOptionsFIFO
+ * Method:    maxDataFilesSize
+ * Signature: (J)J
+ */
+jlong Java_org_rocksdb_CompactionOptionsFIFO_maxDataFilesSize(JNIEnv*, jclass,
+                                                              jlong jhandle) {
+  auto* opt =
+      reinterpret_cast<ROCKSDB_NAMESPACE::CompactionOptionsFIFO*>(jhandle);
+  return static_cast<jlong>(opt->max_data_files_size);
+}
+
+/*
+ * Class:     org_rocksdb_CompactionOptionsFIFO
+ * Method:    setUseKvRatioCompaction
+ * Signature: (JZ)V
+ */
+void Java_org_rocksdb_CompactionOptionsFIFO_setUseKvRatioCompaction(
+    JNIEnv*, jclass, jlong jhandle, jboolean use_kv_ratio_compaction) {
+  auto* opt =
+      reinterpret_cast<ROCKSDB_NAMESPACE::CompactionOptionsFIFO*>(jhandle);
+  opt->use_kv_ratio_compaction = static_cast<bool>(use_kv_ratio_compaction);
+}
+
+/*
+ * Class:     org_rocksdb_CompactionOptionsFIFO
+ * Method:    useKvRatioCompaction
+ * Signature: (J)Z
+ */
+jboolean Java_org_rocksdb_CompactionOptionsFIFO_useKvRatioCompaction(
+    JNIEnv*, jclass, jlong jhandle) {
+  auto* opt =
+      reinterpret_cast<ROCKSDB_NAMESPACE::CompactionOptionsFIFO*>(jhandle);
+  return static_cast<jboolean>(opt->use_kv_ratio_compaction);
+}
+
 /*
  * Class:     org_rocksdb_CompactionOptionsFIFO
  * Method:    disposeInternal

java/src/main/java/org/rocksdb/CompactionOptionsFIFO.java

@@ -75,6 +75,51 @@ public class CompactionOptionsFIFO extends RocksObject {
     return allowCompaction(nativeHandle_);
   }
 
+  /**
+   * Combined SST + blob file size limit for FIFO compaction trimming.
+   * When non-zero, FIFO uses total_sst + total_blob for size-based dropping.
+   * When zero (default), uses max_table_files_size (SST-only).
+   *
+   * @param maxDataFilesSize the combined size limit in bytes
+   *
+   * @return the reference to the current options.
+   */
+  public CompactionOptionsFIFO setMaxDataFilesSize(final long maxDataFilesSize) {
+    setMaxDataFilesSize(nativeHandle_, maxDataFilesSize);
+    return this;
+  }
+
+  /**
+   * Get the combined SST + blob file size limit.
+   *
+   * @return max data files size in bytes, 0 means disabled
+   */
+  public long maxDataFilesSize() {
+    return maxDataFilesSize(nativeHandle_);
+  }
+
+  /**
+   * Enable capacity-derived intra-L0 compaction using the observed key/value
+   * size ratio. Requires maxDataFilesSize > 0.
+   *
+   * @param useKvRatioCompaction true to enable
+   *
+   * @return the reference to the current options.
+   */
+  public CompactionOptionsFIFO setUseKvRatioCompaction(final boolean useKvRatioCompaction) {
+    setUseKvRatioCompaction(nativeHandle_, useKvRatioCompaction);
+    return this;
+  }
+
+  /**
+   * Check if capacity-derived intra-L0 compaction is enabled.
+   *
+   * @return true if enabled
+   */
+  public boolean useKvRatioCompaction() {
+    return useKvRatioCompaction(nativeHandle_);
+  }
+
   private static native long newCompactionOptionsFIFO();
   @Override
   protected final void disposeInternal(final long handle) {
@@ -86,4 +131,9 @@ public class CompactionOptionsFIFO extends RocksObject {
   private static native long maxTableFilesSize(final long handle);
   private static native void setAllowCompaction(final long handle, final boolean allowCompaction);
   private static native boolean allowCompaction(final long handle);
+  private static native void setMaxDataFilesSize(final long handle, final long maxDataFilesSize);
+  private static native long maxDataFilesSize(final long handle);
+  private static native void setUseKvRatioCompaction(
+      final long handle, final boolean useKvRatioCompaction);
+  private static native boolean useKvRatioCompaction(final long handle);
 }

options/cf_options.cc

@@ -311,6 +311,14 @@ static std::unordered_map<std::string, OptionTypeInfo>
         {"trivial_copy_buffer_size",
          {offsetof(struct CompactionOptionsFIFO, trivial_copy_buffer_size),
           OptionType::kUInt64T, OptionVerificationType::kNormal,
+          OptionTypeFlags::kMutable}},
+        {"max_data_files_size",
+         {offsetof(struct CompactionOptionsFIFO, max_data_files_size),
+          OptionType::kUInt64T, OptionVerificationType::kNormal,
+          OptionTypeFlags::kMutable}},
+        {"use_kv_ratio_compaction",
+         {offsetof(struct CompactionOptionsFIFO, use_kv_ratio_compaction),
+          OptionType::kBoolean, OptionVerificationType::kNormal,
           OptionTypeFlags::kMutable}}};
 
 static std::unordered_map<std::string, OptionTypeInfo>
@@ -1260,6 +1268,10 @@ void MutableCFOptions::Dump(Logger* log) const {
                  compaction_options_fifo.max_table_files_size);
   ROCKS_LOG_INFO(log, "compaction_options_fifo.allow_compaction : %d",
                  compaction_options_fifo.allow_compaction);
+  ROCKS_LOG_INFO(log, "compaction_options_fifo.max_data_files_size : %" PRIu64,
+                 compaction_options_fifo.max_data_files_size);
+  ROCKS_LOG_INFO(log, "compaction_options_fifo.use_kv_ratio_compaction : %d",
+                 compaction_options_fifo.use_kv_ratio_compaction);
 
   // Blob file related options
   ROCKS_LOG_INFO(log, "                        enable_blob_files: %s",

options/options_settable_test.cc

@@ -678,7 +678,8 @@ TEST_F(OptionsSettableTest, ColumnFamilyOptionsAllFieldsSettable) {
       "preserve_internal_time_seconds=86400;"
       "compaction_options_fifo={max_table_files_size=3;allow_"
       "compaction=true;age_for_warm=0;file_temperature_age_thresholds={{"
-      "temperature=kCold;age=12345}};};"
+      "temperature=kCold;age=12345}};max_data_files_size=1073741824;"
+      "use_kv_ratio_compaction=false;};"
       "blob_cache=1M;"
       "memtable_protection_bytes_per_key=2;"
       "persist_user_defined_timestamps=true;"

tools/db_bench_tool.cc

@@ -1034,6 +1034,14 @@ DEFINE_uint64(fifo_compaction_ttl, 0, "TTL for the SST Files in seconds.");
 
 DEFINE_uint64(fifo_age_for_warm, 0, "age_for_warm for FIFO compaction.");
 
+DEFINE_uint64(fifo_compaction_max_data_files_size_mb, 0,
+              "Combined SST + blob file size limit for FIFO compaction "
+              "trimming. 0 means use max_table_files_size (SST-only).");
+
+DEFINE_bool(fifo_compaction_use_kv_ratio_compaction, false,
+            "Enable capacity-derived intra-L0 compaction for FIFO with "
+            "BlobDB. Requires fifo_compaction_max_data_files_size_mb > 0.");
+
 // Stacked BlobDB Options
 DEFINE_bool(use_blob_db, false, "[Stacked BlobDB] Open a BlobDB instance.");
 
@@ -4425,6 +4433,10 @@ class Benchmark {
         FLAGS_fifo_compaction_max_table_files_size_mb * 1024 * 1024,
         FLAGS_fifo_compaction_allow_compaction);
     options.compaction_options_fifo.age_for_warm = FLAGS_fifo_age_for_warm;
+    options.compaction_options_fifo.max_data_files_size =
+        FLAGS_fifo_compaction_max_data_files_size_mb * 1024 * 1024;
+    options.compaction_options_fifo.use_kv_ratio_compaction =
+        FLAGS_fifo_compaction_use_kv_ratio_compaction;
     options.prefix_extractor = prefix_extractor_;
     if (FLAGS_use_uint64_comparator) {
       options.comparator = test::Uint64Comparator();

tools/db_crashtest.py

@@ -265,6 +265,10 @@ default_params = {
     "stats_dump_period_sec": lambda: random.choice([0, 10, 600]),
     "compaction_ttl": lambda: random.choice([0, 0, 1, 2, 10, 100, 1000]),
     "fifo_allow_compaction": lambda: random.randint(0, 1),
+    "fifo_compaction_max_data_files_size_mb": lambda: random.choice(
+        [0, 100, 500]
+    ),
+    "fifo_compaction_use_kv_ratio_compaction": lambda: random.randint(0, 1),
     # Test small max_manifest_file_size in a smaller chance, as most of the
     # time we wnat manifest history to be preserved to help debug
     "max_manifest_file_size": lambda: random.choice(
@@ -969,9 +973,21 @@ def finalize_and_sanitize(src_params):
         # Disable irrelevant tiering options
         dest_params["preclude_last_level_data_seconds"] = 0
         dest_params["last_level_temperature"] = "kUnknown"
+        # use_kv_ratio_compaction requires allow_compaction and
+        # max_data_files_size > 0
+        if dest_params.get("fifo_compaction_use_kv_ratio_compaction", 0) == 1:
+            if (
+                dest_params.get("fifo_allow_compaction", 0) != 1
+                or dest_params.get("fifo_compaction_max_data_files_size_mb", 0)
+                == 0
+            ):
+                dest_params["fifo_compaction_use_kv_ratio_compaction"] = 0
     else:
         # Disable irrelevant tiering options
         dest_params["file_temperature_age_thresholds"] = ""
+        # Disable FIFO-specific options for non-FIFO compaction styles
+        dest_params["fifo_compaction_max_data_files_size_mb"] = 0
+        dest_params["fifo_compaction_use_kv_ratio_compaction"] = 0
     if dest_params["partition_filters"] == 1:
         if dest_params["index_type"] != 2:
             dest_params["partition_filters"] = 0

unreleased_history/new_features/fifo_kv_ratio_compaction.md

@@ -0,0 +1 @@
+Added `CompactionOptionsFIFO::max_data_files_size` to support FIFO compaction trimming based on combined SST and blob file sizes. Added `CompactionOptionsFIFO::use_kv_ratio_compaction` to enable a capacity-derived intra-L0 compaction strategy optimized for BlobDB workloads, producing uniform-sized compacted files for predictable FIFO trimming.

wiki/fifo_compaction.md

@@ -0,0 +1,672 @@
+# FIFO Compaction Strategy
+
+This document describes the FIFO compaction style in RocksDB, covering the
+file dropping strategies and both the old and new intra-L0 compaction
+picking strategies.
+
+## Overview
+
+FIFO compaction is designed for time-series and log-like workloads where data
+has a natural expiration. All data lives at L0. When total data exceeds a
+configured size limit, the oldest SST files are dropped — no merge, no rewrite,
+just deletion. This gives near-zero write amplification for the compaction layer.
+
+```
+L0 (all data lives here):
+  Newest                                                           Oldest
+    |                                                                |
+    v                                                                v
+  [SST_N] [SST_N-1] ... [SST_3] [SST_2] [SST_1]
+    ^                                       ^
+    |                                       |
+  new flushes added here          oldest files dropped here
+                                  (when over size limit)
+```
+
+Without intra-L0 compaction, every memtable flush creates a new small SST file.
+Over time, the number of L0 files grows, increasing read amplification (each
+point lookup must check every L0 file). Intra-L0 compaction addresses this by
+merging small files into fewer larger files.
+
+## Compaction Picking Priority Chain
+
+When compaction is triggered (score >= 1.0), the picker tries these strategies
+in order, returning the first non-null result:
+
+```
+PickCompaction():
+    |
+    |-- 1. PickTTLCompaction()               [File Dropping]
+    |       Drop files older than TTL.
+    |
+    |-- 2. PickSizeCompaction()              [File Dropping]
+    |       Drop oldest files when over size limit.
+    |
+    |-- 3. PickIntraL0Compaction()           [Intra-L0]
+    |       Dispatcher: merges small L0 files to reduce file count.
+    |       Requires allow_compaction=true. Dispatches to:
+    |         - PickRatioBasedIntraL0Compaction (use_kv_ratio_compaction=true)
+    |         - PickCostBasedIntraL0Compaction  (use_kv_ratio_compaction=false)
+    |
+    |-- 4. PickTemperatureChangeCompaction() [Temperature Migration]
+            Rewrite one file to change its temperature tier.
+            Lowest priority — runs only if nothing else needs to be done.
+```
+
+Steps 1 and 2 are **file dropping** — they delete old files to enforce size
+or TTL limits. Step 3 is **intra-L0 compaction** — it merges small files into
+fewer larger ones. `PickIntraL0Compaction` is the dispatcher that selects
+between the two strategies based on `use_kv_ratio_compaction`.
+
+Step 4 is **temperature migration** — it rewrites a single file to change its
+storage temperature (e.g., moving cold data to cheaper storage). It picks one
+file at a time, checking if the file's age exceeds a configured threshold but
+its current temperature doesn't match the target. It runs last because it's
+the lowest priority: disk space management (dropping) and read amplification
+(intra-L0) are more important than storage tiering. Since FIFO only allows
+one compaction at a time, running temperature change last ensures it never
+blocks more critical operations.
+
+Note: Intra-L0 compaction runs after size-based dropping. If `PickSizeCompaction`
+dropped files (returned non-null), `PickIntraL0Compaction` is skipped. This
+means intra-L0 only runs when the DB is under the size limit or when
+size-based compaction is already in progress.
+
+## Score Computation
+
+The compaction score determines when compaction should be triggered. For FIFO:
+
+```
+score = effective_total_size / effective_max_size
+```
+
+Where:
+- `effective_total_size` = total SST size (or SST + blob when
+  `max_data_files_size > 0`)
+- `effective_max_size` = `max_table_files_size` (or `max_data_files_size`
+  when set)
+
+Additional score contributions:
+- When `allow_compaction` is true (enables intra-L0 compaction):
+  `score = max(score, num_sorted_runs / level0_file_num_compaction_trigger)`
+- When `ttl > 0`: score is boosted by expired file count
+- When temperature thresholds are set: score is boosted if files need
+  temperature change
+
+---
+
+# Part 1: File Dropping Strategies
+
+These strategies delete old files to enforce data size or TTL limits.
+No data is rewritten — files are simply removed.
+
+## TTL-Based Dropping (`PickTTLCompaction`)
+
+**When**: `ttl > 0`
+
+Drops L0 files whose data is older than the TTL threshold. Iterates from
+oldest to newest, checking `newest_key_time` or `creation_time` from table
+properties against `current_time - ttl`.
+
+```
+Before TTL compaction (ttl = 3600s, files older than 1 hour):
+
+  L0: [F6:10m] [F5:20m] [F4:40m] [F3:50m] [F2:70m] [F1:80m]
+                                              ^^^^     ^^^^
+                                            older than TTL --> DROP
+
+After:
+  L0: [F6:10m] [F5:20m] [F4:40m] [F3:50m]
+```
+
+Returns `nullptr` if deleting expired files would still leave the total size
+above the size limit — in that case, size-based dropping handles it instead.
+
+**Config**: `MutableCFOptions::ttl` (in seconds)
+
+## Size-Based Dropping (`PickSizeCompaction`)
+
+**When**: Total size exceeds the configured limit.
+
+### SST-Only Mode (default)
+
+Compares sum of SST file sizes against `max_table_files_size`:
+
+```
+Before (total 1.2GB > max_table_files_size 1GB):
+
+  L0: [F8:200MB] [F7:200MB] [F6:200MB] [F5:200MB] [F4:200MB] [F3:200MB]
+                                                       total = 1.2GB
+
+  Drop oldest files until under limit:
+  Drop F3 (200MB) --> remaining = 1.0GB <= 1GB limit --> STOP
+
+After:
+  L0: [F8:200MB] [F7:200MB] [F6:200MB] [F5:200MB] [F4:200MB]
+                                                       total = 1.0GB
+```
+
+### Blob-Aware Mode (`max_data_files_size > 0`)
+
+When BlobDB is enabled, SST files are small (keys + blob references) and blob
+files hold the actual values. The total disk usage is dominated by blob files,
+so `max_table_files_size` (SST-only) cannot control total disk usage.
+
+`max_data_files_size` accounts for both SST and blob files:
+
+```
+effective_size = total_sst + total_blob
+
+Example: total_sst = 10MB, total_blob = 9.99GB
+  max_table_files_size = 1GB  --> sees 10MB, no dropping (WRONG!)
+  max_data_files_size = 10GB  --> sees 10GB, drops when exceeded (CORRECT)
+```
+
+When dropping files, proportional estimation is used to account for blob
+data freed per SST file:
+
+```
+data_per_file = effective_size / num_files
+```
+
+Blob files are automatically cleaned up when their linked SSTs are deleted
+(via `BlobFileMetaData::GetLinkedSsts()` reference counting).
+
+**Config**:
+- `CompactionOptionsFIFO::max_table_files_size` (default: 1GB)
+- `CompactionOptionsFIFO::max_data_files_size` (default: 0, disabled)
+
+## Temperature Migration (`PickTemperatureChangeCompaction`)
+
+**When**: `file_temperature_age_thresholds` is non-empty
+
+This is NOT file dropping — it **rewrites** a single SST file to assign it a
+new storage temperature (e.g., kWarm, kCold). This allows tiered storage
+systems to move aging data to cheaper/slower media. The file content is
+unchanged; only the temperature metadata is updated.
+
+Picks one file at a time, scanning from oldest to newest. For each file,
+checks if its age exceeds a configured threshold AND its current temperature
+doesn't match the target. Only one file is migrated per compaction to minimize
+impact on other operations. Only works with single-level FIFO
+(`num_levels == 1`).
+
+This runs as the **lowest priority** in the picking chain (step 4) because
+storage tiering is less urgent than disk space management (dropping) or read
+amplification (intra-L0 compaction). Since FIFO allows only one compaction at
+a time, this ensures temperature migration never blocks critical operations.
+
+```
+Config: file_temperature_age_thresholds = [{kWarm, 3600}, {kCold, 86400}]
+
+  [F6:5m,kUnk] [F5:30m,kUnk] [F4:2h,kUnk] [F3:5h,kUnk] [F2:2d,kUnk]
+                                                            ^^^^^^^^
+                                                            age > 86400s
+                                                            --> compact to kCold
+
+After:
+  [F6:5m,kUnk] [F5:30m,kUnk] [F4:2h,kUnk] [F3:5h,kUnk] [F2:2d,kCold]
+```
+
+**Config**: `CompactionOptionsFIFO::file_temperature_age_thresholds`
+
+---
+
+# Part 2: Intra-L0 Compaction
+
+Intra-L0 compaction merges multiple small L0 files into fewer larger files
+to reduce file count and read amplification. Unlike file dropping, this
+rewrites data — but only SST data (blob files are never rewritten).
+
+`allow_compaction = true` is the **master switch** for intra-L0 compaction.
+When enabled, `use_kv_ratio_compaction` selects which picking strategy to use:
+
+```
+  allow_compaction = true          (master switch for intra-L0)
+          |
+          +-- use_kv_ratio_compaction = false   (default)
+          |     Old Strategy: PickCostBasedIntraL0Compaction
+          |     Guard: 1.1 * write_buffer_size
+          |     Works when SST ~= write_buffer_size (non-BlobDB)
+          |
+          +-- use_kv_ratio_compaction = true
+                New Strategy: PickRatioBasedIntraL0Compaction
+                Guard: capacity-derived target from SST/blob ratio
+                Works when SST << write_buffer_size (BlobDB)
+                Requires: max_data_files_size > 0
+```
+
+## Old Strategy: `PickCostBasedIntraL0Compaction`
+
+**When**: `allow_compaction = true` AND `use_kv_ratio_compaction = false`.
+Called from `PickIntraL0Compaction` (which only runs when `PickSizeCompaction`
+returned nullptr, meaning the DB is under the size limit).
+
+This is the original intra-L0 compaction, implemented in
+`PickCostBasedIntraL0Compaction()`. It uses a greedy algorithm to pick files,
+with a `write_buffer_size`-based guard to prevent re-compacting large files.
+
+### Algorithm
+
+```
+1. Start from the newest L0 file (index 0)
+2. Greedily add older files while compact_bytes_per_del_file decreases
+3. Stop when:
+   - A file is being_compacted
+   - compact_bytes_per_del_file starts increasing (diminishing returns)
+   - Total exceeds max_compaction_bytes
+4. Check: enough files (>= trigger) AND per_del < 1.1 * write_buffer_size
+5. Output: always a single file
+```
+
+### Understanding `compact_bytes_per_del_file`
+
+`compact_bytes_per_del_file` measures the **cost per file eliminated**. When
+we compact N files into 1 output, we eliminate (N-1) files but must read and
+rewrite all N files' data. The metric is:
+
+```
+compact_bytes_per_del_file = total_input_bytes / (num_files - 1)
+```
+
+The algorithm greedily adds files as long as this ratio keeps **decreasing**
+(meaning each additional file is "cheap" to include). When adding a file
+causes the ratio to **increase**, we stop — it signals diminishing returns.
+
+```
+Example: scanning files from newest (left) to oldest (right)
+
+  Files:    [F5:32KB] [F4:64KB] [F3:48KB] [F2:96KB] [F1:128KB]
+
+  Step 1: Start with F5 (32KB). compact_bytes = 32KB.
+  Step 2: Add F4.  compact_bytes = 96KB.  per_del = 96/1 = 96KB.
+  Step 3: Add F3.  compact_bytes = 144KB. per_del = 144/2 = 72KB. (72 < 96, improving)
+  Step 4: Add F2.  compact_bytes = 240KB. per_del = 240/3 = 80KB. (80 > 72, WORSE!)
+          --> STOP. Adding F2 makes the ratio increase.
+
+  Result: pick [F5, F4, F3] (3 files), per_del = 72KB.
+```
+
+The ratio increases when a file is significantly larger than the average of
+files already selected. This naturally prevents including already-compacted
+files (which are larger than flush files) — IF the size gap is significant.
+
+### Example (uniform flush files)
+
+```
+Before (4 flush files of 64KB each, trigger=4):
+
+  L0: [F4:64KB] [F3:64KB] [F2:64KB] [F1:64KB]
+       newest                          oldest
+
+  PickCostBasedIntraL0Compaction:
+    Add F4: compact_bytes = 64KB
+    Add F3: compact_bytes = 128KB, per_del = 128/1 = 128KB
+    Add F2: compact_bytes = 192KB, per_del = 192/2 = 96KB  (96 < 128, better)
+    Add F1: compact_bytes = 256KB, per_del = 256/3 = 85KB  (85 < 96, better)
+    No more files. Check: 4 >= trigger(4) and 85KB < 70MB. OK.
+
+After:
+  L0: [C1:256KB]    (single compacted output)
+```
+
+### Example (flush + compacted, ratio detects size gap)
+
+```
+  L0: [F8:64KB] [F7:64KB] [F6:64KB] [F5:64KB] [C1:256KB]
+       newest                                     oldest (compacted)
+
+  PickCostBasedIntraL0Compaction:
+    Add F8: compact_bytes = 64KB
+    Add F7: compact_bytes = 128KB, per_del = 128/1 = 128KB
+    Add F6: compact_bytes = 192KB, per_del = 192/2 = 96KB  (improving)
+    Add F5: compact_bytes = 256KB, per_del = 256/3 = 85KB  (improving)
+    Add C1: compact_bytes = 512KB, per_del = 512/4 = 128KB (128 > 85, WORSE!)
+    --> STOP before C1.
+
+  Result: pick [F8, F7, F6, F5] — compacted file C1 is excluded.
+  This works because C1 (256KB) is 4x larger than flush files (64KB).
+```
+
+### Anti-Re-Compaction Guard
+
+The guard `compact_bytes_per_del_file < 1.1 * write_buffer_size` prevents
+picking files that are already near memtable size. The idea: compacted files
+should be ~write_buffer_size, so they'd push `per_del` above the guard.
+
+```
+Guard works when SST ~= write_buffer_size:
+
+  Files: [64MB, 64MB, 64MB, 64MB]   (SST ~= WBS = 64MB)
+  per_del = 256MB/3 = 85MB > 70MB   --> guard rejects --> no re-compaction
+```
+
+### Known Limitation with BlobDB
+
+With BlobDB, SST files are ~1000x smaller than `write_buffer_size`. The guard
+threshold (e.g., 70MB) is never reached by any L0 file. ALL files pass the
+guard, including previously compacted files:
+
+```
+Guard FAILS when SST << write_buffer_size (BlobDB):
+
+  write_buffer_size = 64MB, SST files ~64KB (1000x smaller)
+  Guard threshold: 1.1 * 64MB = 70.4MB
+
+  10 compacted files of 256KB each:
+    per_del = 2560KB/9 = 284KB << 70.4MB --> guard passes!
+    ALL 10 files re-compacted into 1 file of 2.56MB
+
+  Result: cascading re-compaction creates "monster files"
+
+  Round 1: [64KB, 64KB, 64KB, 64KB] --> compact --> [256KB]
+  Round 2: [64KB, 64KB, 64KB, 256KB] --> compact ALL --> [448KB]
+  Round 3: [64KB, 64KB, 64KB, 448KB] --> compact ALL --> [640KB]
+  ... files grow unboundedly
+```
+
+Use the KV-ratio strategy instead for BlobDB workloads.
+
+### Config
+
+- `CompactionOptionsFIFO::allow_compaction` (default: false)
+- Anti-re-compaction guard: `1.1 * write_buffer_size`
+- Min files: `level0_file_num_compaction_trigger`
+
+## New Strategy: `use_kv_ratio_compaction` (`PickRatioBasedIntraL0Compaction`)
+
+**When**: `allow_compaction = true` AND `use_kv_ratio_compaction = true`
+AND `max_data_files_size > 0`
+
+This strategy replaces the `write_buffer_size`-based guard with a
+**capacity-derived target** and uses **tiered size-based merging** to achieve
+logarithmic write amplification. It observes the actual SST/blob size ratio,
+computes a target graduated file size, and merges files incrementally through
+size tiers rather than directly to target.
+
+### Why a New Strategy?
+
+```
+Without BlobDB:  SST ~= write_buffer_size     --> old guard works
+With BlobDB:     SST ~= write_buffer_size/1000 --> old guard is useless
+```
+
+The new strategy derives the target from the **data capacity** and
+**observed key/value ratio**, not from `write_buffer_size`.
+
+### Algorithm
+
+**Step 1: Target Computation**
+
+The target graduated file size can be determined in two ways:
+
+```
+If max_compaction_bytes > 0 (explicitly set by user):
+  target = max_compaction_bytes      // user override
+
+If max_compaction_bytes == 0 (default, auto-calculate):
+  sst_ratio = total_l0_sst / (total_l0_sst + total_blob)
+  total_sst_at_cap = max_data_files_size * sst_ratio
+  target = total_sst_at_cap / level0_file_num_compaction_trigger
+```
+
+```
+Example (auto-calculated):
+  max_data_files_size = 10GB, sst_ratio = 0.001 (64KB SST / 64MB total)
+  total_sst_at_cap = 10GB * 0.001 = 10MB
+  trigger = 10
+  target = 10MB / 10 = 1MB
+```
+
+The `sst_ratio` is **recomputed on every `PickCompaction` call**. The
+computation is trivial (sum file sizes + arithmetic) and `PickCompaction`
+is only called once per flush or compaction completion, so no caching is
+needed. This also means the ratio naturally adapts when `SetOptions()`
+changes configuration.
+
+**Step 2: Tier Boundaries**
+
+Tier boundaries form a geometric sequence descending from the target,
+using `trigger` as the growth factor:
+
+```
+..., target/trigger^2, target/trigger, target
+```
+
+Example with target=1MB, trigger=10:
+  boundaries = [10KB, 100KB, 1MB]
+
+Boundaries below 10KB are not generated (SST files of most workloads
+are larger than this). If target itself is below 10KB, it is used as
+the sole boundary.
+
+Files >= target are "graduated" and never compacted again. They sit
+in L0 until FIFO drops them.
+
+**Step 3: Tiered File Selection**
+
+For each tier boundary (smallest first), scan L0 from oldest to newest:
+
+```
+For each boundary B (from smallest to largest):
+  1. Skip files >= B (they belong to higher tiers) and being_compacted files
+  2. Collect contiguous files < B
+  3. Stop when accumulated >= B (cap at 2*B to prevent tier-skipping)
+  4. If >= 2 files and accumulated >= B: merge them
+  5. Output (~B bytes) lands at the next tier
+```
+
+Processing boundaries smallest-first ensures bottom-up build: flush outputs
+are merged first, and higher-tier merges happen naturally as lower-tier
+outputs accumulate.
+
+```
+Example (target=1MB, trigger=10, flush~10KB):
+
+  Tier boundaries: [10KB, 100KB, 1MB]
+
+  L0: [1MB_grad] [1MB_grad] [100KB] [100KB] [10KB] [10KB] [F] [F] [F] [F]
+
+  Scan at boundary=10KB:
+    F,F,F,F (all < 10KB) --> accumulated >= 10KB? If yes, merge → ~10KB output
+
+  Scan at boundary=100KB:
+    10KB,10KB,... (all < 100KB) --> accumulated >= 100KB? merge → ~100KB
+
+  Scan at boundary=1MB:
+    100KB,100KB,... (all < 1MB) --> accumulated >= 1MB? merge → graduated!
+```
+
+### Trade-Off: Write Amp vs L0 File Count
+
+The tiered approach trades higher L0 file count for logarithmic write amp:
+
+```
+Write amp per byte:
+  k + 1 = ceil(log(target/flush) / log(trigger)) + 1
+  Each byte is rewritten once per tier crossing.
+
+L0 file count at steady state:
+  trigger + k * (trigger - 1)
+  More than the original trigger target, but bounded logarithmically.
+
+Example (target=1MB, flush=1KB, trigger=10):
+  k = 3 tiers, write amp = 4, file count ≈ 37
+  vs flat merging: write amp ≈ 57
+```
+
+### Anti-Re-Compaction Guard
+
+The guard is implicit in the tier boundaries:
+
+```
+Graduated files (>= target) are skipped at EVERY tier boundary.
+  1MB >= 1MB   --> skipped at 1MB boundary
+  1MB >= 100KB --> skipped at 100KB boundary
+  1MB >= 10KB  --> skipped at 10KB boundary
+
+Intermediate tier files are only merged at HIGHER tier boundaries.
+  A 100KB file (output of tier-0 merge) is:
+    >= 100KB --> skipped at 100KB boundary (won't be re-merged at same tier)
+    < 1MB    --> eligible at 1MB boundary (merges into graduated file)
+```
+
+Compare with the old strategy's guard:
+
+```
+Old: guard = 1.1 * write_buffer_size (breaks when SST << WBS)
+New: graduated files >= target always excluded; intermediate files
+     progress through tiers without cascading re-compaction
+```
+
+### Steady State
+
+```
+Steady state L0 (target=64MB, trigger=4, flush~1MB):
+
+  [64MB_grad, 64MB_grad, 64MB_grad, 64MB_grad,
+   16MB, 16MB, 16MB,
+   4MB, 4MB,
+   1MB, 1MB, 1MB]
+
+  - 4 graduated files at target size (frozen until FIFO drops them)
+  - Intermediate files at tier sizes (accumulating for next merge)
+  - Flush outputs (accumulating for first tier merge)
+
+When FIFO drops the oldest graduated file, it removes exactly
+1/trigger of the total SST data (predictable).
+```
+
+### Write Amplification
+
+```
+With BlobDB (SST ~1KB, blob ~1MB per flush, target=1MB, trigger=10):
+  - k = 3 tiers (1KB → 10KB → 100KB → 1MB)
+  - SST write amp: k+1 = 4x (flush + 3 tier crossings)
+  - Blob write amp: ~1x (never rewritten)
+  - Total write amp: ~1 + 1KB*4/(1KB+1MB) ≈ 1.004x
+
+Without BlobDB (SST ~64MB per flush):
+  - target = large, ratio = 1, k = 1 typically
+  - SST write amp: ~2x
+```
+
+### File Uniformity
+
+At steady state, all graduated files are close to the target size.
+Output is in [boundary, 2*boundary) at each tier. Variable flush sizes
+are handled naturally — the size-based merge rule produces consistent
+output regardless of individual file sizes.
+
+### Config
+
+- `CompactionOptionsFIFO::allow_compaction` (required: true)
+- `CompactionOptionsFIFO::use_kv_ratio_compaction` (default: false)
+- `CompactionOptionsFIFO::max_data_files_size` (required, > 0)
+- `level0_file_num_compaction_trigger` (target max L0 file count)
+- `max_compaction_bytes` (default: 0 = auto-calculate target from capacity;
+  when > 0, overrides auto-calculated target with this value)
+
+## Choosing Between Old and New Intra-L0 Strategies
+
+Both strategies require `allow_compaction = true`. The choice of strategy
+depends on whether BlobDB is used:
+
+```
+Decision tree:
+
+  Want intra-L0 compaction?
+    |
+    +-- NO:  allow_compaction = false (default)
+    |        No file merging, only dropping.
+    |
+    +-- YES: allow_compaction = true
+             |
+             +-- Using BlobDB (SST << write_buffer_size)?
+             |     |
+             |     +-- YES: use_kv_ratio_compaction = true
+             |     |        (also requires max_data_files_size > 0)
+             |     |
+             |     +-- NO:  use_kv_ratio_compaction = false (default)
+             |              Old strategy works fine.
+```
+
+| Criteria | Old (default) | New (`use_kv_ratio_compaction`) |
+|----------|------------------------|-------------------------------|
+| Guard mechanism | `1.1 * write_buffer_size` | capacity-derived target |
+| Works with BlobDB? | No (guard broken) | Yes (designed for it) |
+| File uniformity | Poor with BlobDB | Good (+/-25%) |
+| Re-compaction risk | High with BlobDB | None (tiered boundaries prevent it) |
+| Write amp (BlobDB) | Unpredictable | Logarithmic: (k+1)x SST, ~1x total |
+| Requires | `allow_compaction=true` | `allow_compaction=true` + `use_kv_ratio_compaction=true` + `max_data_files_size>0` |
+
+---
+
+# Configuration Examples
+
+## Basic FIFO (no intra-L0 compaction)
+
+```cpp
+options.compaction_style = kCompactionStyleFIFO;
+options.compaction_options_fifo.max_table_files_size = 1ULL * 1024 * 1024 * 1024;  // 1GB
+```
+
+## FIFO with old intra-L0 (non-BlobDB)
+
+```cpp
+options.compaction_style = kCompactionStyleFIFO;
+options.compaction_options_fifo.max_table_files_size = 1ULL * 1024 * 1024 * 1024;
+options.compaction_options_fifo.allow_compaction = true;
+options.level0_file_num_compaction_trigger = 4;
+```
+
+## FIFO with BlobDB and KV-ratio compaction
+
+```cpp
+options.compaction_style = kCompactionStyleFIFO;
+options.compaction_options_fifo.max_data_files_size = 10ULL * 1024 * 1024 * 1024;  // 10GB
+options.compaction_options_fifo.allow_compaction = true;   // master switch
+options.compaction_options_fifo.use_kv_ratio_compaction = true;  // select new strategy
+options.level0_file_num_compaction_trigger = 10;
+options.enable_blob_files = true;
+options.min_blob_size = 1024;
+```
+
+## FIFO with TTL + BlobDB
+
+```cpp
+options.compaction_style = kCompactionStyleFIFO;
+options.compaction_options_fifo.max_data_files_size = 10ULL * 1024 * 1024 * 1024;
+options.compaction_options_fifo.allow_compaction = true;
+options.compaction_options_fifo.use_kv_ratio_compaction = true;
+options.level0_file_num_compaction_trigger = 10;
+options.ttl = 86400;  // 24 hours
+options.enable_blob_files = true;
+options.min_blob_size = 1024;
+```
+
+---
+
+# Configuration Reference
+
+## CompactionOptionsFIFO
+
+| Option | Type | Default | Description |
+|--------|------|---------|-------------|
+| `max_table_files_size` | uint64_t | 1GB | SST-only size limit for FIFO dropping |
+| `max_data_files_size` | uint64_t | 0 | Combined SST+blob size limit (0=disabled) |
+| `allow_compaction` | bool | false | Master switch for intra-L0 compaction (required for both old and new strategies) |
+| `use_kv_ratio_compaction` | bool | false | Select capacity-derived intra-L0 strategy (requires allow_compaction=true AND max_data_files_size>0) |
+| `age_for_warm` | uint64_t | 0 | DEPRECATED |
+| `file_temperature_age_thresholds` | vector | empty | Age-based temperature migration |
+| `allow_trivial_copy_when_change_temperature` | bool | false | Allow trivial copy for temp change |
+| `trivial_copy_buffer_size` | uint64_t | 4096 | Buffer size for trivial copy |
+
+## Related CF Options
+
+| Option | Relevance to FIFO |
+|--------|-------------------|
+| `level0_file_num_compaction_trigger` | Target max L0 file count for KV-ratio; min files for old intra-L0 |
+| `ttl` | TTL-based file expiration (seconds) |
+| `write_buffer_size` | Guard threshold for old-style intra-L0 (1.1x) |
+| `max_compaction_bytes` | For KV-ratio: 0 = auto-calculate target from capacity; > 0 = use as target directly. For old intra-L0: cap on total input size. Default sanitized to target_file_size_base * 25 (except when use_kv_ratio_compaction=true, where 0 is preserved) |