diff --git a/src/mito2/src/engine/row_selector_test.rs b/src/mito2/src/engine/row_selector_test.rs index 0366211117d..602b5448ea7 100644 --- a/src/mito2/src/engine/row_selector_test.rs +++ b/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; diff --git a/src/mito2/src/read/scan_util.rs b/src/mito2/src/read/scan_util.rs index afd3de10d96..5d9d3543154 100644 --- a/src/mito2/src/read/scan_util.rs +++ b/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?