From aec7ba34c9b5dfc482e98565484fd1583f45151a Mon Sep 17 00:00:00 2001
From: "Lei, HUANG" <6406592+v0y4g3r@users.noreply.github.com>
Date: Sat, 1 Aug 2026 12:24:49 +0800
Subject: [PATCH] fix(mito2): prioritize newer compaction windows (#8714)

* fix(mito2): prioritize newer compaction windows

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* test(mito2): adapt window priority test after rebase

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): update compaction picker test

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

---------

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
---
 src/mito2/src/compaction/twcs.rs | 45 +++++++++++++++++++++++++++++---
 1 file changed, 41 insertions(+), 4 deletions(-)

diff --git a/src/mito2/src/compaction/twcs.rs b/src/mito2/src/compaction/twcs.rs
index abe6a81cd9..bbfc117198 100644
--- a/src/mito2/src/compaction/twcs.rs
+++ b/src/mito2/src/compaction/twcs.rs
@@ -73,6 +73,7 @@ impl TwcsPicker {
         let mut output = vec![];
         let windows = time_windows
             .values()
+            .rev()
             .filter(|window| {
                 !window.files.is_empty()
                     && self.time_range.as_ref().is_none_or(|time_range| {
@@ -1021,11 +1022,11 @@ mod tests {
             .to_vec(),
             expected_outputs: vec![
                 ExpectedOutput {
-                    input_files: vec![0, 1],
+                    input_files: vec![2, 3],
                     output_level: 1,
                 },
                 ExpectedOutput {
-                    input_files: vec![2, 3],
+                    input_files: vec![0, 1],
                     output_level: 1,
                 },
             ],
@@ -1052,11 +1053,11 @@ mod tests {
             .to_vec(),
             expected_outputs: vec![
                 ExpectedOutput {
-                    input_files: vec![0, 1],
+                    input_files: vec![2, 4],
                     output_level: 1,
                 },
                 ExpectedOutput {
-                    input_files: vec![2, 4],
+                    input_files: vec![0, 1],
                     output_level: 1,
                 },
             ],
@@ -1455,6 +1456,42 @@ mod tests {
         assert_eq!(output[0].inputs.len(), 32);
     }
 
+    #[tokio::test]
+    async fn test_newer_windows_have_priority() {
+        let older_file_ids = [FileId::random(), FileId::random()];
+        let newer_file_ids = [FileId::random(), FileId::random()];
+        let files = [
+            new_file_handle_with_sequence(older_file_ids[0], 1_000, 1_999, 0, 1),
+            new_file_handle_with_sequence(older_file_ids[1], 1_000, 1_999, 0, 2),
+            new_file_handle_with_sequence(newer_file_ids[0], 7_000, 7_999, 0, 3),
+            new_file_handle_with_sequence(newer_file_ids[1], 7_000, 7_999, 0, 4),
+        ];
+        let windows = assign_to_windows(files.iter(), 3);
+        let picker = TwcsPicker {
+            trigger_file_num: 2,
+            time_window_seconds: Some(3),
+            max_output_file_size: None,
+            append_mode: false,
+            max_background_tasks: Some(1),
+            time_range: None,
+        };
+
+        let output = picker
+            .build_output_with_time_range(RegionId::from_u64(123), windows, Some(9), None)
+            .await
+            .unwrap();
+
+        assert_eq!(1, output.len());
+        assert_eq!(
+            newer_file_ids.into_iter().collect::<HashSet<_>>(),
+            output[0]
+                .inputs
+                .iter()
+                .map(|file| file.file_id().file_id())
+                .collect::<HashSet<_>>()
+        );
+    }
+
     #[test]
     fn test_filter_time_windows_by_time_range() {
         let time_range = TimestampRange::new(