fix: apply last row after source deduplication

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

src/mito2/src/engine/row_selector_test.rs

@@ -416,6 +416,42 @@ async fn test_last_row_delete_wins_across_ssts() {
     }
 }
 
+#[tokio::test]
+async fn test_last_row_returns_older_put_after_newest_delete_across_ssts() {
+    for flat_format in [false, true] {
+        let (_env, engine, region_id, schema, delete_schema) =
+            new_merge_last_row_engine(flat_format).await;
+        put_rows(
+            &engine,
+            region_id,
+            Rows {
+                schema,
+                rows: vec![value_row("a", 1.0, 1000), value_row("a", 2.0, 2000)],
+            },
+        )
+        .await;
+        flush_region(&engine, region_id, None).await;
+        delete_rows(
+            &engine,
+            region_id,
+            Rows {
+                schema: delete_schema,
+                rows: build_delete_rows_for_key("a", 2, 3),
+            },
+        )
+        .await;
+        flush_region(&engine, region_id, None).await;
+
+        let scanner = last_row_scanner(&engine, region_id).await;
+        assert_eq!(2, scanner.num_files());
+        assert_eq!(0, scanner.num_memtables());
+        assert_eq!(
+            vec![("a".to_string(), 1.0, 1000)],
+            last_row_values(&scan_last_row_batches(&scanner).await)
+        );
+    }
+}
+
 #[tokio::test]
 async fn test_last_row_empty_region_returns_empty() {
     let (_env, engine, region_id, _schema, _delete_schema) = new_merge_last_row_engine(false).await;

src/mito2/src/read/scan_util.rs

@@ -1571,19 +1571,8 @@ pub fn build_flat_file_range_scan_stream(
             let build_reader_start = Instant::now();
             let Some(mut reader) = range
                 .flat_reader(
-                    // In exact `sequence_range` mode the row-group-level LastRow
-                    // shortcut would reduce each row group to its last-timestamp
-                    // row *before* the row-level sequence filter runs, silently
-                    // dropping in-range rows (a series with seq 1 at t1 and seq 2
-                    // at t2 under `(0, 1]` keeps only seq 2 and then filters it
-                    // out). Bypass the shortcut so the final per-row selector
-                    // (`FlatLastRowReader`, applied after source merging and the
-                    // sequence filter) selects on the filtered rows instead.
-                    if stream_ctx.input.sequence_range.is_some() {
-                        None
-                    } else {
-                        stream_ctx.input.series_row_selector
-                    },
+                    // LastRow must run after cross-source merge and deduplication.
+                    None,
                     fetch_metrics.as_deref(),
                 )
                 .await?