Fix CompactRange with kForceOptimized in range split scenarios (#12702)

db/compaction/compaction_picker.cc

@@ -782,15 +782,35 @@ Compaction* CompactionPicker::PickCompactionForCompactRange(
       }
     }
     if (inputs_shrunk.empty()) {
-      return nullptr;
+      if (covering_the_whole_range) {
+        // All files in the entire range were filtered out, nothing to do
+        return nullptr;
+      } else {
+        // All files in current batch were filtered out, but there are more
+        // files beyond the size limit. We do not compact this batch due to
+        // max_file_num_to_ignore, but
+        // we should signal the caller to continue with the remaining files.
+        if (skip_input_index < inputs.size()) {
+          // Set compaction_end to the smallest key of the first file we skipped
+          InternalKey* next_start = &inputs.files[skip_input_index]->smallest;
+          **compaction_end = *next_start;
+        } else {
+          // Shouldn't happen, but handle gracefully
+          *compaction_end = nullptr;
+        }
+        return nullptr;
+      }
     }
+
     if (inputs.size() != inputs_shrunk.size()) {
       inputs.files.swap(inputs_shrunk);
     }
-    // set covering_the_whole_range to false if there is any file that need to
-    // be compacted in the range of inputs[skip_input_index+1, inputs.size())
-    for (size_t i = skip_input_index + 1; i < inputs.size(); ++i) {
-      if (inputs[i]->fd.GetNumber() < max_file_num_to_ignore) {
+
+    // Check if there are any files in the range beyond skip_input_index that
+    // still need to be compacted. After the swap above, inputs_shrunk contains
+    // the original file list.
+    for (size_t i = skip_input_index + 1; i < inputs_shrunk.size(); ++i) {
+      if (inputs_shrunk[i]->fd.GetNumber() < max_file_num_to_ignore) {
         covering_the_whole_range = false;
       }
     }

db/db_compaction_test.cc

@@ -7285,6 +7285,90 @@ TEST_F(DBCompactionTest, ManualCompactionBottomLevelOptimized) {
   ASSERT_EQ(num, 0);
 }
 
+// Test for GitHub issue #12702:
+// CompactRange with kForceOptimized should work correctly when:
+// 1. max_compaction_bytes limit causes input resizing (covering_the_whole_range
+// = false)
+// 2. kForceOptimized filtering removes all selected files (inputs_shrunk =
+// empty) This scenario happens in range split situations where some files were
+// recently recompacted (high file numbers) while others were not (low file
+// numbers).
+TEST_F(DBCompactionTest, ManualCompactionBottomLevelOptimizedWithSizeLimit) {
+  Options opts = CurrentOptions();
+  opts.num_levels = 3;
+  opts.compression = kNoCompression;
+  opts.target_file_size_base = 10 << 20;  // 10MB
+  opts.disable_auto_compactions = true;
+  DestroyAndReopen(opts);
+
+  Random rnd(301);
+
+  // Step 1: Create initial files across the full key range [0, 200)
+  // This simulates the initial state before a partition split
+  for (int i = 0; i < 200; i++) {
+    ASSERT_OK(Put(Key(i), rnd.RandomString(100)));
+  }
+  ASSERT_OK(Flush());
+  MoveFilesToLevel(2);  // Move to bottommost level
+
+  // Step 2: Simulate writes only to left partition [0, 100)
+  // This causes recompaction of files in the left range
+  for (int i = 0; i < 100; i++) {
+    ASSERT_OK(Put(Key(i), rnd.RandomString(100)));
+  }
+  ASSERT_OK(Flush());
+
+  // Compact to merge with bottommost level
+  // This creates NEW files for [0, 100) range with HIGH file numbers
+  CompactRangeOptions cro_initial;
+  cro_initial.bottommost_level_compaction =
+      BottommostLevelCompaction::kForceOptimized;
+  std::string start_key = Key(0);
+  std::string end_key = Key(100);
+  Slice start_slice(start_key);
+  Slice end_slice(end_key);
+  ASSERT_OK(dbfull()->CompactRange(cro_initial, &start_slice, &end_slice));
+
+  // At this point:
+  // - Files covering [0, 100) have HIGH file numbers (recently compacted)
+  // - Files covering [100, 200) have LOW file numbers (not recompacted)
+
+  // Step 3: Set a small max_compaction_bytes to trigger the bug
+  // We want the limit to cause input resizing (covering_the_whole_range =
+  // false) while kForceOptimized filters out all files in the first batch
+  std::string prop;
+  EXPECT_TRUE(dbfull()->GetProperty(DB::Properties::kLiveSstFilesSize, &prop));
+  uint64_t total_size = atoi(prop.c_str());
+  // Set limit to about 30% of total size to ensure we hit the bug scenario
+  uint64_t max_compaction_bytes = total_size / 3;
+  ASSERT_OK(dbfull()->SetOptions(
+      {{"max_compaction_bytes", std::to_string(max_compaction_bytes)}}));
+
+  // Step 4: Try to compact the left partition [0, 100)
+  // Before the fix: This would return nullptr and fail to compact
+  // After the fix: This should succeed even though first batch is filtered out
+  CompactRangeOptions cro;
+  cro.bottommost_level_compaction = BottommostLevelCompaction::kForceOptimized;
+  ASSERT_OK(dbfull()->CompactRange(cro, &start_slice, &end_slice));
+
+  // Verify the compaction succeeded (data is still readable)
+  for (int i = 0; i < 100; i++) {
+    ASSERT_NE("NOT_FOUND", Get(Key(i)));
+  }
+
+  // Step 5: Also verify right partition [100, 200) works
+  // (This should have always worked, even before the fix)
+  std::string start_key_right = Key(100);
+  std::string end_key_right = Key(200);
+  Slice start_slice_right(start_key_right);
+  Slice end_slice_right(end_key_right);
+  ASSERT_OK(dbfull()->CompactRange(cro, &start_slice_right, &end_slice_right));
+
+  for (int i = 100; i < 200; i++) {
+    ASSERT_NE("NOT_FOUND", Get(Key(i)));
+  }
+}
+
 TEST_F(DBCompactionTest, ManualCompactionMax) {
   uint64_t l1_avg_size = 0, l2_avg_size = 0;
   auto generate_sst_func = [&]() {