diff --git a/src/mito2/src/compaction/twcs.rs b/src/mito2/src/compaction/twcs.rs
index bbfc117198..342426bbef 100644
--- a/src/mito2/src/compaction/twcs.rs
+++ b/src/mito2/src/compaction/twcs.rs
@@ -169,7 +169,7 @@ impl TwcsPicker {
         let found_runs = sorted_runs.len();
         // We only remove deletion markers if we found less than 2 runs and not in append mode.
         // because after compaction there will be no overlapping files.
-        let filter_deleted =
+        let mut filter_deleted =
             found_runs <= 2 && !self.append_mode && !window_has_overlap(files, windows);
         if found_runs == 0 {
             return (vec![], filter_deleted);
@@ -210,6 +210,14 @@ impl TwcsPicker {
             );
         }
 
+        // The inputs are only a subset of the window: `reduce_runs` and `merge_seq_files` both
+        // narrow the selection down and the file num limit above narrows it further. A deletion
+        // marker may only be dropped when every file that can hold a row it masks is compacted
+        // along with it, so re-check the final selection against what stays in the window.
+        if filter_deleted && overlaps_files_left_behind(&inputs, &files_to_merge) {
+            filter_deleted = false;
+        }
+
         if inputs.len() > 1 {
             // If we have more than one group to compact.
             log_pick_result(
@@ -227,6 +235,20 @@ impl TwcsPicker {
     }
 }
 
+/// Returns whether any file group of `window_files` that is not part of `inputs` overlaps
+/// `inputs`. Such a group keeps rows a deletion marker in `inputs` masks, so the compaction
+/// must not filter deleted rows out.
+///
+/// Overlapping is inclusive on both boundaries here: two files that only share a boundary
+/// timestamp can still hold the same row.
+fn overlaps_files_left_behind(inputs: &[FileGroup], window_files: &[FileGroup]) -> bool {
+    let picked: HashSet<_> = inputs.iter().flat_map(|fg| fg.file_ids()).collect();
+    window_files
+        .iter()
+        .filter(|fg| !fg.file_ids().iter().any(|id| picked.contains(id)))
+        .any(|fg| inputs.iter().any(|input| input.overlap_inclusive(fg)))
+}
+
 #[allow(clippy::too_many_arguments)]
 fn log_pick_result(
     region_id: RegionId,
@@ -1456,6 +1478,99 @@ mod tests {
         assert_eq!(output[0].inputs.len(), 32);
     }
 
+    #[tokio::test]
+    async fn test_limit_max_input_files_keeps_deletion_markers() {
+        common_telemetry::init_default_ut_logging();
+
+        // One large file group spanning the whole window plus 32 small ones nested inside
+        // it. That is exactly 2 runs, so the window on its own allows filtering deletions.
+        let mut files = vec![new_file_handle_with_size_and_sequence(
+            FileId::random(),
+            0,
+            3_000_000,
+            0,
+            1,
+            1024 * 1024 * 1024,
+        )];
+        files.extend((0..32).map(|idx: i64| {
+            new_file_handle_with_size_and_sequence(
+                FileId::random(),
+                (idx + 1) * 10_000,
+                (idx + 1) * 10_000 + 1_000,
+                0,
+                (idx + 2) as u64,
+                1024,
+            )
+        }));
+
+        let windows = assign_to_windows(files.iter(), 3600);
+
+        let picker = TwcsPicker {
+            trigger_file_num: 4,
+            time_window_seconds: Some(3600),
+            max_output_file_size: None,
+            append_mode: false,
+            max_background_tasks: None,
+            time_range: None,
+        };
+
+        let active_window = find_latest_window_in_seconds(files.iter(), 3600);
+        let output = picker
+            .build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
+            .await
+            .unwrap();
+
+        assert_eq!(1, output.len());
+        // The input file num limit picks the smallest groups first, so the large group is
+        // left behind while the small groups that overlap it are compacted.
+        assert_eq!(32, output[0].inputs.len());
+        assert!(
+            !output[0].filter_deleted,
+            "deletion markers must be kept once the file num limit drops files they may mask"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_limit_max_input_files_still_filters_without_overlap() {
+        common_telemetry::init_default_ut_logging();
+
+        // 40 file groups with disjoint time ranges, i.e. a single run. Nothing the file num
+        // limit leaves behind can hold a row masked by a deletion marker we compact.
+        let files: Vec<_> = (0..40i64)
+            .map(|idx| {
+                new_file_handle_with_size_and_sequence(
+                    FileId::random(),
+                    (idx + 1) * 10_000,
+                    (idx + 1) * 10_000 + 1_000,
+                    0,
+                    (idx + 1) as u64,
+                    1024,
+                )
+            })
+            .collect();
+
+        let windows = assign_to_windows(files.iter(), 3600);
+
+        let picker = TwcsPicker {
+            trigger_file_num: 4,
+            time_window_seconds: Some(3600),
+            max_output_file_size: Some(1024 * 1024 * 1024),
+            append_mode: false,
+            max_background_tasks: None,
+            time_range: None,
+        };
+
+        let active_window = find_latest_window_in_seconds(files.iter(), 3600);
+        let output = picker
+            .build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
+            .await
+            .unwrap();
+
+        assert_eq!(1, output.len());
+        assert_eq!(32, output[0].inputs.len());
+        assert!(output[0].filter_deleted);
+    }
+
     #[tokio::test]
     async fn test_newer_windows_have_priority() {
         let older_file_ids = [FileId::random(), FileId::random()];
diff --git a/src/mito2/src/engine/compaction_test.rs b/src/mito2/src/engine/compaction_test.rs
index 4404b8d46a..1fc3a8f785 100644
--- a/src/mito2/src/engine/compaction_test.rs
+++ b/src/mito2/src/engine/compaction_test.rs
@@ -1342,6 +1342,155 @@ async fn test_compaction_region_with_overlapping_delete_all_with_format(flat_for
     assert!(vec.is_empty());
 }
 
+#[tokio::test]
+async fn test_compaction_input_limit_keeps_rows_deleted() {
+    test_compaction_input_limit_keeps_rows_deleted_with_format(false).await;
+    test_compaction_input_limit_keeps_rows_deleted_with_format(true).await;
+}
+
+/// Creates a region that only compacts when asked, so that a test can build an exact file
+/// layout with `put_and_flush` and `delete_and_flush`.
+async fn env_for_manual_compaction(
+    env: &mut TestEnv,
+    region_id: RegionId,
+    flat_format: bool,
+) -> (MitoEngine, Vec<ColumnSchema>) {
+    let engine = env
+        .create_engine(MitoConfig {
+            default_flat_format: flat_format,
+            min_compaction_interval: Duration::from_secs(3600),
+            ..Default::default()
+        })
+        .await;
+
+    env.get_schema_metadata_manager()
+        .register_region_table_info(
+            region_id.table_id(),
+            "test_table",
+            "test_catalog",
+            "test_schema",
+            None,
+            env.get_kv_backend(),
+        )
+        .await;
+
+    let request = CreateRequestBuilder::new()
+        .insert_option("compaction.type", "twcs")
+        .insert_option("compaction.twcs.time_window", "1h")
+        .build();
+    let column_schemas = request
+        .column_metadatas
+        .iter()
+        .map(column_metadata_to_column_schema)
+        .collect::<Vec<_>>();
+    engine
+        .handle_request(region_id, RegionRequest::Create(request))
+        .await
+        .unwrap();
+
+    (engine, column_schemas)
+}
+
+/// The picker caps a compaction at 32 input files and drops the largest file groups to get
+/// there. A deletion marker among the picked files must not be filtered out while the file
+/// holding the rows it masks stays behind, otherwise those rows become visible again.
+async fn test_compaction_input_limit_keeps_rows_deleted_with_format(flat_format: bool) {
+    common_telemetry::init_default_ut_logging();
+    let mut env = TestEnv::new().await;
+    let region_id = RegionId::new(1, 1);
+    let (engine, column_schemas) =
+        env_for_manual_compaction(&mut env, region_id, flat_format).await;
+
+    // One large file spanning the whole time window.
+    put_and_flush(&engine, region_id, &column_schemas, 0..3000).await;
+    // Deletes 6 rows of that file. The markers land in a tiny file that overlaps it.
+    delete_and_flush(&engine, region_id, &column_schemas, 10..16).await;
+    // 31 more tiny files that overlap the large one but not each other, so the window holds
+    // 33 file groups forming 2 runs.
+    for i in 2..33 {
+        put_and_flush(&engine, region_id, &column_schemas, i * 10..i * 10 + 6).await;
+    }
+
+    let scanner = engine
+        .scanner(region_id, ScanRequest::default())
+        .await
+        .unwrap();
+    assert_eq!(
+        33,
+        scanner.num_files(),
+        "unexpected files: {:?}",
+        scanner.file_ids()
+    );
+
+    compact(&engine, region_id).await;
+
+    let scanner = engine
+        .scanner(region_id, ScanRequest::default())
+        .await
+        .unwrap();
+    // The 32 tiny files are merged into one; the large file exceeds the input file num limit
+    // and is left behind.
+    assert_eq!(
+        2,
+        scanner.num_files(),
+        "unexpected files: {:?}",
+        scanner.file_ids()
+    );
+    let stream = scanner.scan().await.unwrap();
+    let vec = collect_stream_ts(stream).await;
+    assert!(
+        !(10..16).any(|ts| vec.contains(&(ts * 1000))),
+        "deleted rows are visible again after compaction"
+    );
+    assert_eq!(2994, vec.len());
+}
+
+#[tokio::test]
+async fn test_compaction_of_part_of_a_run_keeps_rows_deleted() {
+    test_compaction_of_part_of_a_run_keeps_rows_deleted_with_format(false).await;
+    test_compaction_of_part_of_a_run_keeps_rows_deleted_with_format(true).await;
+}
+
+/// A run is supposed to hold no overlapping files, which is what lets `merge_seq_files`
+/// compact part of a run and still filter deleted rows. Run detection compares time ranges
+/// exclusively though, so a file covering a single timestamp lands in the same run as the
+/// file it deletes rows from. The rows must stay deleted when only one of the two is picked.
+async fn test_compaction_of_part_of_a_run_keeps_rows_deleted_with_format(flat_format: bool) {
+    common_telemetry::init_default_ut_logging();
+    let mut env = TestEnv::new().await;
+    let region_id = RegionId::new(1, 1);
+    let (engine, column_schemas) =
+        env_for_manual_compaction(&mut env, region_id, flat_format).await;
+
+    // One large file spanning the whole time window.
+    put_and_flush(&engine, region_id, &column_schemas, 0..3000).await;
+    // Deletes a single row of that file, so the marker lands in a file covering one timestamp.
+    delete_and_flush(&engine, region_id, &column_schemas, 1..2).await;
+    // 31 more single row files, each holding another key at its own timestamp.
+    for ts in 1..32 {
+        let rows = Rows {
+            schema: column_schemas.clone(),
+            rows: build_rows_for_key("b", ts * 10, ts * 10 + 1, 0),
+        };
+        put_rows(&engine, region_id, rows).await;
+        flush(&engine, region_id).await;
+    }
+
+    compact(&engine, region_id).await;
+
+    let scanner = engine
+        .scanner(region_id, ScanRequest::default())
+        .await
+        .unwrap();
+    let stream = scanner.scan().await.unwrap();
+    let vec = collect_stream_ts(stream).await;
+    assert!(
+        !vec.contains(&1000),
+        "deleted row is visible again after compaction"
+    );
+    assert_eq!(3030, vec.len());
+}
+
 // For issue https://github.com/GreptimeTeam/greptimedb/issues/3633
 #[tokio::test]
 async fn test_readonly_during_compaction() {