perf(promql): push down last row for instant queries (#9034)

* perf(promql): push down last row for instant queries Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test: guard instant last row correctness Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test: update instant query explain results Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix: apply last row after source deduplication Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix: scope post-merge last row selection Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(perf): cover instant PromQL last row Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(perf): sort generated SST rows before writing Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(promql): cover instant last row selection in sqlness Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(promql): avoid last row hints for lossy timestamp casts Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(promql): preserve stale marker semantics across flushes Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(promql): decode dictionary labels in stale regression Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(promql): avoid reserved column name in stale fixture Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(promql): exercise LastRow hints and filtered results Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * refactor: keep after-merge mode in LastRow selector Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix: reject instant LastRow across residual filters Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test: expect after-merge selector in instant vector guards Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * docs: explain instant LastRow filter eligibility Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix: restrict instant LastRow to safe selector nodes Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * refactor: show LastRow merge mode directly in diagnostics Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test: refresh LastRow display in explain expectations Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
2026-09-10 23:42:16 +00:00 · 2026-09-09 12:38:52 +00:00
parent c65a4e545e
commit da5cb1a190
33 changed files with 3190 additions and 335 deletions
@@ -38,6 +38,7 @@ DEFAULT_CASES = [
    "tests/perf/query_cases/prom_remote_write_mixed_every/case.toml",
    "tests/perf/query_cases/prom_remote_write_integer_counter/case.toml",
    "tests/perf/query_cases/promql_range_boundary/case.toml",
+    "tests/perf/query_cases/promql_instant_last_row_9034/case.toml",
 ]

 HEAVY_CASES = [
@@ -32,6 +32,7 @@ use mito_codec::row_converter::{DensePrimaryKeyCodec, PrimaryKeyCodecExt, SortFi
 use mito2::access_layer::{FilePathProvider, Metrics, WriteType};
 use mito2::config::IndexConfig;
 use mito2::manifest::action::{RegionCheckpoint, RegionManifest, RemovedFilesRecord};
+use mito2::memtable::sort_primary_key_record_batch;
 use mito2::read::FlatSource;
 use mito2::sst::file::{FileMeta, RegionFileId};
 use mito2::sst::index::{Indexer, IndexerBuilder};
@@ -282,8 +283,10 @@ fn generate_record_batch(
    columns.push(Arc::new(pk_builder.finish()));
    columns.push(Arc::new(UInt64Array::from_value(sequence, rows)));
    columns.push(Arc::new(UInt8Array::from_value(OpType::Put as u8, rows)));
-    RecordBatch::try_new(flat_schema, columns)
-        .expect("generated fixture columns should match flat SST Arrow schema")
+    let batch = RecordBatch::try_new(flat_schema, columns)
+        .expect("generated fixture columns should match flat SST Arrow schema");
+    sort_primary_key_record_batch(&batch)
+        .expect("generated fixture batch should sort by primary key, timestamp, and sequence")
 }

 fn file_meta_from_sst_info(
@@ -532,3 +535,204 @@ pub(super) async fn run_direct_sst(args: DirectArgs) {
    fs::write(out_dir.join("summary.json"), serde_json::to_vec_pretty(&serde_json::json!({ "case": case_name, "seed": seed, "table_index": table_index, "table": table.name, "database": table.database, "region_id": region_id.as_u64(), "table_dir": table_dir, "region_dir": region_dir, "sst_format": format!("{format:?}"), "sst_count": scenario.layout.sst_count, "rows_per_sst": scenario.layout.rows_per_sst, "row_group_size": scenario.layout.row_group_size, "total_rows": scenario.layout.sst_count * scenario.layout.rows_per_sst, "checkpoint_path": checkpoint_path, "files_jsonl_path": files_jsonl_path, "readback_validated": false, "metadata_source": "synthetic" })).expect("failed to serialize fixture summary")).expect("failed to write fixture summary.json");
    println!("Done. wrote {} SST file entries", manifest.files.len());
 }
+
+#[cfg(test)]
+mod tests {
+    use std::num::NonZeroUsize;
+
+    use datatypes::arrow::array::{BinaryArray, DictionaryArray, StringArray};
+
+    use super::*;
+
+    type LogicalRow = (Vec<u8>, i64, u64, String, String, f64);
+
+    fn string_values(batch: &RecordBatch, column: usize) -> Vec<String> {
+        let dictionary = batch
+            .column(column)
+            .as_any()
+            .downcast_ref::<DictionaryArray<UInt32Type>>()
+            .expect("fixture tag column should be a string dictionary");
+        let values = dictionary
+            .values()
+            .as_any()
+            .downcast_ref::<StringArray>()
+            .expect("fixture tag dictionary should contain strings");
+        dictionary
+            .keys()
+            .values()
+            .iter()
+            .map(|key| values.value(*key as usize).to_string())
+            .collect()
+    }
+
+    fn encoded_primary_keys(batch: &RecordBatch) -> Vec<Vec<u8>> {
+        let dictionary = batch
+            .column(batch.num_columns() - 3)
+            .as_any()
+            .downcast_ref::<DictionaryArray<UInt32Type>>()
+            .expect("fixture primary key should be a binary dictionary");
+        let values = dictionary
+            .values()
+            .as_any()
+            .downcast_ref::<BinaryArray>()
+            .expect("fixture primary key dictionary should contain binary keys");
+        dictionary
+            .keys()
+            .values()
+            .iter()
+            .map(|key| values.value(*key as usize).to_vec())
+            .collect()
+    }
+
+    #[test]
+    fn direct_sst_generator_sorts_all_layouts_and_timestamp_units() {
+        for (series_layout, timestamp_type) in [
+            ("timestamp_major", "TIMESTAMP(9)"),
+            ("timestamp_major", "TIMESTAMP(3)"),
+            ("round_robin", "TIMESTAMP(9)"),
+            ("round_robin", "TIMESTAMP(3)"),
+            ("per_sst", "TIMESTAMP(9)"),
+            ("per_sst", "TIMESTAMP(3)"),
+        ] {
+            let mut case: CaseFile = toml::from_str(include_str!(
+                "../../../../../tests/perf/query_cases/promql_instant_last_row_9034/case.toml"
+            ))
+            .expect("existing query perf case should parse");
+            let scenario = match &mut case.scenario {
+                Scenario::DirectReadableSst(scenario) => scenario,
+                _ => unreachable!("included case is a direct SST case"),
+            };
+            scenario.layout.series_layout = series_layout.to_string();
+            scenario.layout.series_count = NonZeroUsize::new(12).expect("12 is nonzero");
+            scenario.layout.rows_per_sst = 36;
+            scenario.layout.sst_count = 1;
+            scenario.layout.row_group_size = 12;
+            let table = &mut scenario.tables[0];
+            let time_index = table.time_index.clone();
+            table
+                .columns
+                .iter_mut()
+                .find(|column| column.name == time_index)
+                .expect("fixture case has time index column")
+                .ty = timestamp_type.to_string();
+
+            let sst_idx = 10;
+            let sequence = 1010;
+            let metadata = Arc::new(build_region_metadata(table, RegionId::from(42)));
+            let batch =
+                generate_record_batch(table, &metadata, &scenario.layout, sst_idx, sequence);
+            assert_eq!(36, batch.num_rows(), "{series_layout}/{timestamp_type}");
+
+            let base_row = sst_idx * scenario.layout.rows_per_sst;
+            let host = table
+                .columns
+                .iter()
+                .find(|column| column.name == "host")
+                .expect("fixture case has host tag");
+            let instance = table
+                .columns
+                .iter()
+                .find(|column| column.name == "instance")
+                .expect("fixture case has instance tag");
+            let value = table
+                .columns
+                .iter()
+                .find(|column| column.name == "value")
+                .expect("fixture case has value field");
+            let (min, max) = match value.distribution.as_ref() {
+                Some(Distribution::DeterministicWave { min, max }) => (*min, *max),
+                _ => unreachable!("included case has deterministic wave values"),
+            };
+            let mut expected = (0..scenario.layout.rows_per_sst)
+                .map(|row| {
+                    let series = series_for_row(&scenario.layout, sst_idx, base_row, row);
+                    let mut tags = HashMap::new();
+                    tags.insert(host.name.clone(), tag_value(host, series));
+                    tags.insert(instance.name.clone(), tag_value(instance, series));
+                    let timestamp = timestamp_for_row(&scenario.layout, base_row, row);
+                    (
+                        encode_dense_primary_key(table, &tags),
+                        if timestamp_type == "TIMESTAMP(3)" {
+                            timestamp / 1_000_000
+                        } else {
+                            timestamp
+                        },
+                        sequence,
+                        tag_value(host, series),
+                        tag_value(instance, series),
+                        wave_value(min, max, base_row + row),
+                    )
+                })
+                .collect::<Vec<LogicalRow>>();
+            expected.sort_by(|left, right| {
+                left.0
+                    .cmp(&right.0)
+                    .then_with(|| left.1.cmp(&right.1))
+                    .then_with(|| right.2.cmp(&left.2))
+            });
+
+            let timestamps = if timestamp_type == "TIMESTAMP(9)" {
+                batch
+                    .column(batch.num_columns() - 4)
+                    .as_any()
+                    .downcast_ref::<TimestampNanosecondArray>()
+                    .expect("nanosecond fixture timestamp")
+                    .values()
+                    .to_vec()
+            } else {
+                batch
+                    .column(batch.num_columns() - 4)
+                    .as_any()
+                    .downcast_ref::<TimestampMillisecondArray>()
+                    .expect("millisecond fixture timestamp")
+                    .values()
+                    .to_vec()
+            };
+            let sequences = batch
+                .column(batch.num_columns() - 2)
+                .as_any()
+                .downcast_ref::<UInt64Array>()
+                .expect("fixture sequence column")
+                .values()
+                .to_vec();
+            let hosts = string_values(&batch, 0);
+            let instances = string_values(&batch, 1);
+            let values = batch
+                .column(2)
+                .as_any()
+                .downcast_ref::<Float64Array>()
+                .expect("fixture value column")
+                .values()
+                .to_vec();
+            let actual = encoded_primary_keys(&batch)
+                .into_iter()
+                .zip(timestamps)
+                .zip(sequences)
+                .zip(hosts)
+                .zip(instances)
+                .zip(values)
+                .map(
+                    |(((((primary_key, timestamp), sequence), host), instance), value)| {
+                        (primary_key, timestamp, sequence, host, instance, value)
+                    },
+                )
+                .collect::<Vec<LogicalRow>>();
+
+            assert_eq!(expected, actual, "{series_layout}/{timestamp_type}");
+            for pair in actual.windows(2) {
+                assert!(
+                    pair[0].0 < pair[1].0
+                        || (pair[0].0 == pair[1].0 && pair[0].1 < pair[1].1)
+                        || (pair[0].0 == pair[1].0
+                            && pair[0].1 == pair[1].1
+                            && pair[0].2 >= pair[1].2),
+                    "rows must be encoded-PK ascending, timestamp ascending, sequence descending: {series_layout}/{timestamp_type}"
+                );
+            }
+            if series_layout != "per_sst" {
+                assert!(actual.iter().any(|row| row.3 == "host2"));
+                assert!(actual.iter().any(|row| row.3 == "host10"));
+            }
+        }
+    }
+}
@@ -389,7 +389,7 @@ fn resolve_series_row_selector(
    scan_config: &ScanConfig,
 ) -> error::Result<Option<TimeSeriesRowSelector>> {
    match scan_config.series_row_selector.as_deref() {
-        Some("last_row") => Ok(Some(TimeSeriesRowSelector::LastRow)),
+        Some("last_row") => Ok(Some(TimeSeriesRowSelector::LastRow { after_merge: false })),
        Some(other) => Err(error::IllegalConfigSnafu {
            msg: format!("Unknown series_row_selector '{other}'"),
        }
@@ -318,7 +318,7 @@ mod test {
    use mito2::config::MitoConfig;
    use store_api::region_engine::{PrepareRequest, QueryScanContext};
    use store_api::region_request::{RegionFlushRequest, RegionPutRequest, RegionRequest};
-    use store_api::storage::TimeSeriesDistribution;
+    use store_api::storage::{TimeSeriesDistribution, TimeSeriesRowSelector};

    use super::*;
    use crate::config::EngineConfig;
@@ -429,6 +429,7 @@ mod test {
        let scan_req = ScanRequest {
            projection,
            filters: vec![],
+            series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: true }),
            ..Default::default()
        };

@@ -444,6 +445,10 @@ mod test {
            &[11, 10, 9, 8, 0, 1, 4]
        );
        assert_eq!(scan_req.filters.len(), 1);
+        assert_eq!(
+            scan_req.series_row_selector,
+            Some(TimeSeriesRowSelector::LastRow { after_merge: true })
+        );
        assert_eq!(
            scan_req.filters[0],
            logical_expr::col(DATA_SCHEMA_TABLE_ID_COLUMN_NAME)
@@ -2621,7 +2621,7 @@ mod tests {
        let key = SelectorResultKey {
            file_id,
            row_group_idx: 0,
-            selector: TimeSeriesRowSelector::LastRow,
+            selector: TimeSeriesRowSelector::LastRow { after_merge: false },
        };
        assert!(cache.get_selector_result(&key).is_none());
        let result = Arc::new(SelectorResultValue::new(
@@ -12,20 +12,25 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use api::v1::Rows;
+use api::v1::value::ValueData;
+use api::v1::{Rows, Value};
 use common_base::readable_size::ReadableSize;
+use common_query::prometheus::PROMETHEUS_STALE_NAN_BITS;
 use common_recordbatch::RecordBatches;
 use datafusion_common::ScalarValue;
 use datafusion_expr::{col, lit};
+use datatypes::arrow::array::{Float64Array, StringArray, TimestampMillisecondArray};
 use store_api::region_engine::RegionEngine;
 use store_api::region_request::RegionRequest;
 use store_api::storage::{RegionId, ScanRequest, TimeSeriesRowSelector};

 use crate::config::MitoConfig;
 use crate::engine::MitoEngine;
+use crate::read::scan_region::Scanner;
 use crate::test_util::batch_util::sort_batches_and_print;
 use crate::test_util::{
-    CreateRequestBuilder, TestEnv, build_rows_for_key, flush_region, put_rows, rows_schema,
+    CreateRequestBuilder, TestEnv, build_delete_rows_for_key, build_rows_for_key, delete_rows,
+    flush_region, put_rows, rows_schema,
 };

 async fn test_last_row(append_mode: bool, flat_format: bool) {
@@ -98,7 +103,7 @@ async fn test_last_row(append_mode: bool, flat_format: bool) {
        .scanner(
            region_id,
            ScanRequest {
-                series_row_selector: Some(TimeSeriesRowSelector::LastRow),
+                series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: false }),
                ..Default::default()
            },
        )
@@ -121,7 +126,7 @@ async fn scan_last_row(
            region_id,
            ScanRequest {
                filters,
-                series_row_selector: Some(TimeSeriesRowSelector::LastRow),
+                series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: false }),
                ..Default::default()
            },
        )
@@ -193,6 +198,276 @@ const LAST_ROW_AT_TEN: &str = "\
 | a     | 10.0    | 1970-01-01T00:00:10 |
 +-------+---------+---------------------+";

+async fn new_merge_last_row_engine(
+    flat_format: bool,
+) -> (
+    TestEnv,
+    MitoEngine,
+    RegionId,
+    Vec<api::v1::ColumnSchema>,
+    Vec<api::v1::ColumnSchema>,
+) {
+    let mut env = TestEnv::new().await;
+    let engine = env.create_engine(MitoConfig::default()).await;
+    let region_id = RegionId::new(1, 1);
+    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 sst_format = if flat_format { "flat" } else { "primary_key" };
+    let request = CreateRequestBuilder::new()
+        .insert_option("sst_format", sst_format)
+        .build();
+    let schema = rows_schema(&request);
+    let delete_schema = crate::test_util::delete_rows_schema(&request);
+    engine
+        .handle_request(region_id, RegionRequest::Create(request))
+        .await
+        .unwrap();
+    (env, engine, region_id, schema, delete_schema)
+}
+
+fn value_row(key: &str, value: f64, timestamp: i64) -> api::v1::Row {
+    api::v1::Row {
+        values: vec![
+            Value {
+                value_data: Some(ValueData::StringValue(key.to_string())),
+            },
+            Value {
+                value_data: Some(ValueData::F64Value(value)),
+            },
+            Value {
+                value_data: Some(ValueData::TimestampMillisecondValue(timestamp)),
+            },
+        ],
+    }
+}
+
+async fn last_row_scanner(engine: &MitoEngine, region_id: RegionId) -> Scanner {
+    engine
+        .scanner(
+            region_id,
+            ScanRequest {
+                series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: false }),
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap()
+}
+
+async fn scan_last_row_batches(scanner: &Scanner) -> RecordBatches {
+    RecordBatches::try_collect(scanner.scan().await.unwrap())
+        .await
+        .unwrap()
+}
+
+fn last_row_values(batches: &RecordBatches) -> Vec<(String, f64, i64)> {
+    let mut rows = Vec::new();
+    for batch in batches.iter() {
+        let batch = batch.df_record_batch();
+        let tags = batch
+            .column_by_name("tag_0")
+            .unwrap()
+            .as_any()
+            .downcast_ref::<StringArray>()
+            .unwrap();
+        let fields = batch
+            .column_by_name("field_0")
+            .unwrap()
+            .as_any()
+            .downcast_ref::<Float64Array>()
+            .unwrap();
+        let timestamps = batch
+            .column_by_name("ts")
+            .unwrap()
+            .as_any()
+            .downcast_ref::<TimestampMillisecondArray>()
+            .unwrap();
+        for index in 0..batch.num_rows() {
+            rows.push((
+                tags.value(index).to_string(),
+                fields.value(index),
+                timestamps.value(index),
+            ));
+        }
+    }
+    rows.sort_by(|left, right| left.0.cmp(&right.0));
+    rows
+}
+
+#[tokio::test]
+async fn test_last_row_merge_deduplicates_same_timestamp_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: schema.clone(),
+                rows: vec![value_row("a", 1.0, 1000)],
+            },
+        )
+        .await;
+        flush_region(&engine, region_id, None).await;
+        put_rows(
+            &engine,
+            region_id,
+            Rows {
+                schema,
+                rows: vec![value_row("a", 2.0, 1000)],
+            },
+        )
+        .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(), 2.0, 1000)],
+            last_row_values(&scan_last_row_batches(&scanner).await)
+        );
+    }
+}
+
+#[tokio::test]
+async fn test_last_row_returns_stale_marker_and_preserves_ordinary_nan() {
+    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: schema.clone(),
+                rows: vec![value_row("a", 1.0, 1000), value_row("b", f64::NAN, 1000)],
+            },
+        )
+        .await;
+        flush_region(&engine, region_id, None).await;
+        put_rows(
+            &engine,
+            region_id,
+            Rows {
+                schema,
+                rows: vec![value_row(
+                    "a",
+                    f64::from_bits(PROMETHEUS_STALE_NAN_BITS),
+                    1000,
+                )],
+            },
+        )
+        .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());
+        let values = last_row_values(&scan_last_row_batches(&scanner).await);
+        assert_eq!(2, values.len(), "unexpected LastRow values: {values:?}");
+        assert_eq!("a", values[0].0);
+        assert_eq!(PROMETHEUS_STALE_NAN_BITS, values[0].1.to_bits());
+        assert_eq!(1000, values[0].2);
+        assert_eq!("b", values[1].0);
+        assert!(values[1].1.is_nan());
+        assert_ne!(PROMETHEUS_STALE_NAN_BITS, values[1].1.to_bits());
+    }
+}
+
+#[tokio::test]
+async fn test_last_row_delete_wins_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)],
+            },
+        )
+        .await;
+        flush_region(&engine, region_id, None).await;
+        delete_rows(
+            &engine,
+            region_id,
+            Rows {
+                schema: delete_schema,
+                rows: build_delete_rows_for_key("a", 1, 2),
+            },
+        )
+        .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!(last_row_values(&scan_last_row_batches(&scanner).await).is_empty());
+    }
+}
+
+#[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 = engine
+            .scanner(
+                region_id,
+                ScanRequest {
+                    series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: true }),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+        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;
+    let scanner = last_row_scanner(&engine, region_id).await;
+    assert!(last_row_values(&scan_last_row_batches(&scanner).await).is_empty());
+}
+
 #[tokio::test]
 async fn test_last_row_append_mode_disabled() {
    test_last_row(false, false).await;
@@ -2197,7 +2197,7 @@ async fn test_exact_sequence_read_with_last_row_selector_keeps_in_range_rows() {
            memtable_min_sequence: Some(0),
            memtable_max_sequence: Some(1),
            exact_sequence_range: true,
-            series_row_selector: Some(TimeSeriesRowSelector::LastRow),
+            series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: false }),
            ..Default::default()
        })
        .await
@@ -2213,7 +2213,7 @@ async fn test_exact_sequence_read_with_last_row_selector_keeps_in_range_rows() {
 | series | 2.0     | 1970-01-01T00:00:02 |
 +--------+---------+---------------------+",
        scan(ScanRequest {
-            series_row_selector: Some(TimeSeriesRowSelector::LastRow),
+            series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: false }),
            ..Default::default()
        })
        .await
@@ -2583,7 +2583,9 @@ async fn test_non_preserve_compaction_sequence_collision_with_format(flat_format
            .scan_to_stream(
                region_id,
                ScanRequest {
-                    series_row_selector: Some(TimeSeriesRowSelector::LastRow),
+                    series_row_selector: Some(TimeSeriesRowSelector::LastRow {
+                        after_merge: false,
+                    }),
                    ..Default::default()
                },
            )
@@ -144,7 +144,7 @@ impl FlatRowGroupLastRowCachedReader {
        let key = SelectorResultKey {
            file_id,
            row_group_idx,
-            selector: TimeSeriesRowSelector::LastRow,
+            selector: TimeSeriesRowSelector::LastRow { after_merge: false },
        };

        if let Some(value) = cache_strategy.get_selector_result(&key) {
@@ -1441,7 +1441,27 @@ mod tests {
    }

    #[test]
-    fn normalizes_and_clears_time_filters() {
+    fn selector_after_merge_changes_fingerprint() {
+        let ordinary = test_scan_fingerprint(
+            vec!["k0 = 'foo'".to_string()],
+            vec![],
+            Some(TimeSeriesRowSelector::LastRow { after_merge: false }),
+            true,
+            0,
+        );
+        let after_merge = test_scan_fingerprint(
+            vec!["k0 = 'foo'".to_string()],
+            vec![],
+            Some(TimeSeriesRowSelector::LastRow { after_merge: true }),
+            true,
+            0,
+        );
+
+        assert_ne!(ordinary, after_merge);
+    }
+
+    #[test]
+    fn true_selector_after_merge_is_preserved_by_fingerprint_transforms() {
        let normalized =
            test_scan_fingerprint(vec!["k0 = 'foo'".to_string()], vec![], None, true, 0);

@@ -1450,18 +1470,32 @@ mod tests {
        let fingerprint = test_scan_fingerprint(
            vec!["k0 = 'foo'".to_string()],
            vec!["ts >= 1000".to_string()],
-            Some(TimeSeriesRowSelector::LastRow),
+            Some(TimeSeriesRowSelector::LastRow { after_merge: true }),
            true,
            7,
        );

        let reset = fingerprint.without_time_filters();
+        let candidate = fingerprint.for_candidate_series();
+        let series_data = fingerprint.for_series_data(SeriesRange::new(0, 1).unwrap());

        assert_eq!(reset.read_columns(), fingerprint.read_columns());
        assert_eq!(reset.read_column_types(), fingerprint.read_column_types());
        assert_eq!(reset.filters(), fingerprint.filters());
        assert!(reset.time_filters().is_empty());
        assert_eq!(reset.series_row_selector, fingerprint.series_row_selector);
+        assert_eq!(
+            fingerprint.series_row_selector,
+            Some(TimeSeriesRowSelector::LastRow { after_merge: true })
+        );
+        assert_eq!(
+            candidate.series_row_selector,
+            fingerprint.series_row_selector
+        );
+        assert_eq!(
+            series_data.series_row_selector,
+            fingerprint.series_row_selector
+        );
        assert_eq!(reset.append_mode, fingerprint.append_mode);
        assert_eq!(reset.filter_deleted, fingerprint.filter_deleted);
        assert_eq!(reset.merge_mode, fingerprint.merge_mode);
@@ -2503,7 +2503,7 @@ mod tests {
        )
        .await
        .with_distribution(Some(TimeSeriesDistribution::PerSeries))
-        .with_series_row_selector(Some(TimeSeriesRowSelector::LastRow))
+        .with_series_row_selector(Some(TimeSeriesRowSelector::LastRow { after_merge: true }))
        .with_merge_mode(MergeMode::LastNonNull)
        .with_filter_deleted(false)
        .build();
@@ -2528,7 +2528,7 @@ mod tests {
                col("v0").gt(lit(1)).to_string(),
            ],
            time_filters: vec![col("ts").gt_eq(ts_lit(1000)).to_string()],
-            series_row_selector: Some(TimeSeriesRowSelector::LastRow),
+            series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: true }),
            append_mode: false,
            filter_deleted: false,
            merge_mode: MergeMode::LastNonNull,
@@ -2537,6 +2537,10 @@ mod tests {
        }
        .build();
        assert_eq!(&expected, fingerprint);
+        assert_eq!(
+            input.series_row_selector,
+            Some(TimeSeriesRowSelector::LastRow { after_merge: true })
+        );
    }

    #[tokio::test]
@@ -30,7 +30,7 @@ use futures::Stream;
 use prometheus::IntGauge;
 use smallvec::SmallVec;
 use snafu::ResultExt;
-use store_api::storage::{RegionId, SequenceRange};
+use store_api::storage::{RegionId, SequenceRange, TimeSeriesRowSelector};

 use crate::error::{ComputeArrowSnafu, Result};
 use crate::memtable::MemScanMetrics;
@@ -1571,18 +1571,13 @@ 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
+                    match stream_ctx.input.series_row_selector {
+                        Some(TimeSeriesRowSelector::LastRow { after_merge: false })
+                            if stream_ctx.input.sequence_range.is_none() =>
+                        {
+                            stream_ctx.input.series_row_selector
+                        }
+                        _ => None,
                    },
                    fetch_metrics.as_deref(),
                )
@@ -263,7 +263,7 @@ impl SeqScan {
        };

        let reader = match &stream_ctx.input.series_row_selector {
-            Some(TimeSeriesRowSelector::LastRow) => {
+            Some(TimeSeriesRowSelector::LastRow { .. }) => {
                Box::pin(FlatLastRowReader::new(reader).into_stream()) as _
            }
            None => reader,
@@ -180,7 +180,7 @@ impl SeriesScan {
    }

    fn supports_two_phase(input: &ScanInput) -> bool {
-        if !is_sparse_metric_metadata(input.region_metadata()) {
+        if input.sequence_range.is_some() || !is_sparse_metric_metadata(input.region_metadata()) {
            return false;
        }
        #[cfg(feature = "enterprise")]
@@ -1214,6 +1214,39 @@ mod tests {
    use datatypes::arrow::record_batch::RecordBatch;

    use super::*;
+    use crate::read::flat_projection::FlatProjectionMapper;
+    use crate::read::scan_region::PredicateGroup;
+    use crate::test_util::scheduler_util::SchedulerEnv;
+    use crate::test_util::sst_util::sst_region_metadata_with_encoding;
+
+    #[tokio::test]
+    async fn two_phase_eligibility_rejects_exact_sequence_range() {
+        let env = SchedulerEnv::new().await;
+        let metadata = Arc::new(sst_region_metadata_with_encoding(
+            store_api::codec::PrimaryKeyEncoding::Sparse,
+        ));
+        let predicate = PredicateGroup::default();
+
+        let eligible = ScanInput::builder(
+            env.access_layer.clone(),
+            FlatProjectionMapper::new(&metadata, [0]).unwrap(),
+        )
+        .with_predicate(predicate.clone())
+        .build();
+        assert!(SeriesScan::supports_two_phase(&eligible));
+
+        let exact_sequence = ScanInput::builder(
+            env.access_layer.clone(),
+            FlatProjectionMapper::new(&metadata, [0]).unwrap(),
+        )
+        .with_predicate(predicate)
+        .with_sequence_range(Some(store_api::storage::SequenceRange::GtLtEq {
+            min: 1,
+            max: 2,
+        }))
+        .build();
+        assert!(!SeriesScan::supports_two_phase(&exact_sequence));
+    }

    #[test]
    fn candidate_distributor_stops_after_all_receivers_close() {
@@ -202,10 +202,10 @@ impl FileRange {
            ))
            .await?;

-        let use_last_row_reader = if selector
-            .map(|s| s == TimeSeriesRowSelector::LastRow)
-            .unwrap_or(false)
-        {
+        let use_last_row_reader = if matches!(
+            selector,
+            Some(TimeSeriesRowSelector::LastRow { after_merge: false })
+        ) {
            // Only use LastRowReader if row group does not contain DELETE, all
            // rows are selected, and filters that still run after this reader
            // cannot change which row is last. Tag filters are safe because a
@@ -231,6 +231,11 @@ impl InstantManipulate {
        "InstantManipulate"
    }

+    /// Returns whether this node evaluates a single timestamp rather than a range.
+    pub fn is_single_evaluation(&self) -> bool {
+        self.start == self.end
+    }
+
    fn staleness_field_columns(&self) -> impl Iterator<Item = &str> {
        let [field, companion] = mixed_sample_fields(self.field_column.as_deref());
        [
@@ -200,6 +200,11 @@ impl SeriesNormalize {
        "SeriesNormalize"
    }

+    /// Returns whether this plan removes Prometheus stale markers.
+    pub const fn filter_stale_markers(&self) -> bool {
+        self.filter_stale_markers
+    }
+
    pub fn to_execution_plan(&self, exec_input: Arc<dyn ExecutionPlan>) -> Arc<dyn ExecutionPlan> {
        Arc::new(SeriesNormalizeExec {
            offset: self.offset,
@@ -286,6 +286,17 @@ impl DummyTableProvider {
        self.scan_request.lock().unwrap().series_row_selector = Some(selector);
    }

+    /// Clones this provider for one logical table-scan use-site.
+    ///
+    /// The optimizer may attach different hints to scans that share the same
+    /// catalog provider, so the per-scan request must not share its mutex.
+    pub fn clone_for_scan(&self) -> Self {
+        Self {
+            scan_request: Arc::new(Mutex::new(self.scan_request.lock().unwrap().clone())),
+            ..self.clone()
+        }
+    }
+
    pub fn with_vector_search_hint(&self, hint: VectorSearchRequest) {
        self.scan_request.lock().unwrap().vector_search = Some(hint);
    }
@@ -410,12 +410,13 @@ mod tests {
    use datafusion_expr::expr::ScalarFunction;
    use datafusion_expr::logical_plan::JoinType;
    use datafusion_expr::{
-        Expr, LogicalPlan, LogicalPlanBuilder, Signature, Subquery, Volatility, col, lit,
+        Expr, Extension, LogicalPlan, LogicalPlanBuilder, Signature, Subquery, Volatility, col, lit,
    };
    use datafusion_optimizer::{OptimizerContext, OptimizerRule};
    use datatypes::schema::ColumnSchema;
+    use promql::extension_plan::InstantManipulate;
    use store_api::metadata::{ColumnMetadata, RegionMetadataBuilder};
-    use store_api::storage::{ConcreteDataType, VectorDistanceMetric};
+    use store_api::storage::{ConcreteDataType, VectorDistanceMetric, VectorSearchRequest};

    use super::VectorSearchState;
    use crate::dummy_catalog::DummyTableProvider;
@@ -469,16 +470,42 @@ mod tests {
    }

    fn vec_distance_expr(function_name: &'static str) -> Expr {
+        vec_distance_expr_with_query(function_name, "[1.0, 2.0]")
+    }
+
+    fn vec_distance_expr_with_query(function_name: &'static str, query_vector: &str) -> Expr {
        let udf = create_udf(Arc::new(TestVectorFunction::new(function_name)));
        Expr::ScalarFunction(ScalarFunction::new_udf(
            Arc::new(udf),
            vec![
                col("v"),
-                lit(ScalarValue::Utf8(Some("[1.0, 2.0]".to_string()))),
+                lit(ScalarValue::Utf8(Some(query_vector.to_string()))),
            ],
        ))
    }

+    fn scan_request_from_plan(plan: &LogicalPlan) -> store_api::storage::ScanRequest {
+        let mut request = None;
+        plan.apply_with_subqueries(|node| {
+            if let LogicalPlan::TableScan(scan) = node
+                && let Some(source) = scan.source.as_any().downcast_ref::<DefaultTableSource>()
+                && let Some(provider) = source
+                    .table_provider
+                    .as_any()
+                    .downcast_ref::<DummyTableProvider>()
+            {
+                request = Some(provider.scan_request());
+            }
+            Ok(datafusion_common::tree_node::TreeNodeRecursion::Continue)
+        })
+        .unwrap();
+        request.unwrap()
+    }
+
+    fn vector_hint_from_plan(plan: &LogicalPlan) -> Option<VectorSearchRequest> {
+        scan_request_from_plan(plan).vector_search
+    }
+
    fn build_dummy_provider(column_id: u32) -> Arc<DummyTableProvider> {
        build_dummy_provider_with_nullable(column_id, false)
    }
@@ -565,15 +592,147 @@ mod tests {
            .unwrap();

        let context = OptimizerContext::default();
-        let _ = ScanHintRule.rewrite(plan, &context).unwrap();
+        let rewritten = ScanHintRule.rewrite(plan, &context).unwrap().data;

-        let hint = dummy_provider.get_vector_search_hint().unwrap();
+        let hint = vector_hint_from_plan(&rewritten).unwrap();
        assert_eq!(hint.column_id, 10);
        assert_eq!(hint.k, 5);
        assert_eq!(hint.metric, VectorDistanceMetric::L2sq);
        assert_eq!(hint.query_vector, vec![1.0, 2.0]);
    }

+    #[test]
+    fn test_vector_hint_is_applied_when_single_evaluation_limit_blocks_last_row() {
+        let provider = build_dummy_provider(10);
+        let source = Arc::new(DefaultTableSource::new(provider));
+        let distance = vec_distance_expr(VEC_L2SQ_DISTANCE);
+        let scan = LogicalPlanBuilder::scan_with_filters("t", source, None, vec![])
+            .unwrap()
+            .sort(vec![distance.sort(true, false)])
+            .unwrap()
+            .limit(0, Some(5))
+            .unwrap()
+            .build()
+            .unwrap();
+        let plan = LogicalPlan::Extension(Extension {
+            node: Arc::new(InstantManipulate::new(
+                1000,
+                1000,
+                1000,
+                1000,
+                "ts".to_string(),
+                vec![],
+                Some("v".to_string()),
+                scan,
+            )),
+        });
+
+        let rewritten = ScanHintRule
+            .rewrite(plan, &OptimizerContext::default())
+            .unwrap()
+            .data;
+        let request = scan_request_from_plan(&rewritten);
+        assert_eq!(request.series_row_selector, None);
+        assert_eq!(
+            request.vector_search,
+            Some(VectorSearchRequest {
+                column_id: 10,
+                query_vector: vec![1.0, 2.0],
+                k: 5,
+                metric: VectorDistanceMetric::L2sq,
+            })
+        );
+    }
+
+    #[test]
+    fn test_single_evaluation_limit_keeps_vector_hints_on_both_branches() {
+        let provider_a = build_dummy_provider(10);
+        let provider_b = build_dummy_provider(20);
+        let distance_a = vec_distance_expr_with_query(VEC_L2SQ_DISTANCE, "[1.0, 2.0]");
+        let distance_b = vec_distance_expr_with_query(VEC_L2SQ_DISTANCE, "[3.0, 4.0]");
+
+        let branch_a = LogicalPlanBuilder::scan_with_filters(
+            "t",
+            Arc::new(DefaultTableSource::new(provider_a)),
+            None,
+            vec![],
+        )
+        .unwrap()
+        .sort(vec![distance_a.sort(true, false)])
+        .unwrap()
+        .limit(0, Some(5))
+        .unwrap()
+        .build()
+        .unwrap();
+        let branch_a = LogicalPlan::Extension(Extension {
+            node: Arc::new(InstantManipulate::new(
+                1000,
+                1000,
+                1000,
+                1000,
+                "ts".to_string(),
+                vec![],
+                Some("v".to_string()),
+                branch_a,
+            )),
+        });
+
+        let branch_b = LogicalPlanBuilder::scan_with_filters(
+            "t",
+            Arc::new(DefaultTableSource::new(provider_b)),
+            None,
+            vec![],
+        )
+        .unwrap()
+        .sort(vec![distance_b.sort(true, false)])
+        .unwrap()
+        .limit(0, Some(9))
+        .unwrap()
+        .build()
+        .unwrap();
+
+        let plan = LogicalPlanBuilder::from(branch_a)
+            .union(branch_b)
+            .unwrap()
+            .build()
+            .unwrap();
+        let rewritten = ScanHintRule
+            .rewrite(plan, &OptimizerContext::default())
+            .unwrap()
+            .data;
+
+        let LogicalPlan::Union(union) = rewritten else {
+            panic!("expected union plan")
+        };
+        let request_a = scan_request_from_plan(&union.inputs[0]);
+        let request_b = scan_request_from_plan(&union.inputs[1]);
+
+        assert_eq!(request_a.series_row_selector, None);
+        assert_eq!(
+            request_a.vector_search,
+            Some(VectorSearchRequest {
+                column_id: 10,
+                query_vector: vec![1.0, 2.0],
+                k: 5,
+                metric: VectorDistanceMetric::L2sq,
+            })
+        );
+        assert_eq!(request_b.series_row_selector, None);
+        assert_eq!(
+            request_b.vector_search,
+            Some(VectorSearchRequest {
+                column_id: 20,
+                query_vector: vec![3.0, 4.0],
+                k: 9,
+                metric: VectorDistanceMetric::L2sq,
+            })
+        );
+
+        // The catalog providers are not mutated; the requests above belong to
+        // the two rewritten scan use-sites.
+        assert_ne!(request_a.vector_search, request_b.vector_search);
+    }
+
    #[test]
    fn test_limit_offset_for_vector_hint() {
        let dummy_provider = build_dummy_provider(10);
@@ -589,9 +748,9 @@ mod tests {
            .unwrap();

        let context = OptimizerContext::default();
-        let _ = ScanHintRule.rewrite(plan, &context).unwrap();
+        let rewritten = ScanHintRule.rewrite(plan, &context).unwrap().data;

-        let hint = dummy_provider.get_vector_search_hint().unwrap();
+        let hint = vector_hint_from_plan(&rewritten).unwrap();
        assert_eq!(hint.k, 15);
    }

@@ -621,8 +780,8 @@ mod tests {
            .unwrap()
            .build()
            .unwrap();
-        let _ = ScanHintRule.rewrite(plan, &context).unwrap();
-        let hint = dummy_provider.get_vector_search_hint().unwrap();
+        let rewritten = ScanHintRule.rewrite(plan, &context).unwrap().data;
+        let hint = vector_hint_from_plan(&rewritten).unwrap();
        assert_eq!(hint.metric, VectorDistanceMetric::InnerProduct);
    }

@@ -681,9 +840,9 @@ mod tests {
            .unwrap();

        let context = OptimizerContext::default();
-        let _ = ScanHintRule.rewrite(plan, &context).unwrap();
+        let rewritten = ScanHintRule.rewrite(plan, &context).unwrap().data;

-        let hint = dummy_provider.get_vector_search_hint().unwrap();
+        let hint = vector_hint_from_plan(&rewritten).unwrap();
        assert_eq!(hint.column_id, 10);
        assert_eq!(hint.k, 5);
    }
@@ -701,9 +860,9 @@ mod tests {
            .unwrap();

        let context = OptimizerContext::default();
-        let _ = ScanHintRule.rewrite(plan, &context).unwrap();
+        let rewritten = ScanHintRule.rewrite(plan, &context).unwrap().data;

-        let hint = dummy_provider.get_vector_search_hint().unwrap();
+        let hint = vector_hint_from_plan(&rewritten).unwrap();
        assert_eq!(hint.k, 4);
    }

@@ -828,10 +987,10 @@ mod tests {
            .unwrap();

        let context = OptimizerContext::default();
-        let _ = ScanHintRule.rewrite(plan, &context).unwrap();
+        let rewritten = ScanHintRule.rewrite(plan, &context).unwrap().data;

        // Hint propagates through simple subquery
-        let hint = provider.get_vector_search_hint().unwrap();
+        let hint = vector_hint_from_plan(&rewritten).unwrap();
        assert_eq!(hint.k, 5);
    }

@@ -901,8 +1060,8 @@ mod tests {
        let _ = ScanHintRule.rewrite(t1_plan, &context).unwrap();
        assert!(t1_provider.get_vector_search_hint().is_none());

-        let _ = ScanHintRule.rewrite(t2_plan, &context).unwrap();
-        let hint = t2_provider.get_vector_search_hint().unwrap();
+        let rewritten = ScanHintRule.rewrite(t2_plan, &context).unwrap().data;
+        let hint = vector_hint_from_plan(&rewritten).unwrap();
        assert_eq!(hint.column_id, 20);
        assert_eq!(hint.k, 5);
    }
@@ -82,7 +82,11 @@ pub trait VectorIndexEngine: Send + Sync {
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Display)]
 pub enum TimeSeriesRowSelector {
    /// Only keep the last row of each time-series.
-    LastRow,
+    #[strum(to_string = "LastRow {{ after_merge: {after_merge} }}")]
+    LastRow {
+        /// Whether selection runs after cross-source merge and deduplication.
+        after_merge: bool,
+    },
 }

 /// A hint on how to distribute time-series data on the scan output.
@@ -323,13 +327,19 @@ mod tests {
        );

        let request = ScanRequest {
+            series_row_selector: Some(TimeSeriesRowSelector::LastRow { after_merge: true }),
            snapshot_on_scan: true,
            exact_sequence_range: true,
            ..Default::default()
        };
        assert_eq!(
            request.to_string(),
-            "ScanRequest { snapshot_on_scan: true, exact_sequence_range: true }"
+            "ScanRequest { series_row_selector: LastRow { after_merge: true }, snapshot_on_scan: true, exact_sequence_range: true }"
+        );
+
+        assert_eq!(
+            TimeSeriesRowSelector::LastRow { after_merge: false }.to_string(),
+            "LastRow { after_merge: false }"
        );

        let request = ScanRequest {
@@ -15,9 +15,16 @@
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};

+use api::v1::value::ValueData;
+use api::v1::{
+    ColumnDataType, ColumnSchema, Row, RowInsertRequest, RowInsertRequests, Rows, SemanticType,
+};
+use common_query::prometheus::PROMETHEUS_STALE_NAN_BITS;
 use common_query::{Output, OutputData};
 use common_recordbatch::util::collect_batches;
-use datatypes::arrow::array::{Float64Array, Int64Array};
+use datatypes::arrow::array::{Float64Array, Int64Array, StringArray, TimestampMillisecondArray};
+use datatypes::arrow::compute::cast;
+use datatypes::arrow::datatypes::DataType;
 use frontend::instance::Instance;
 use query::parser::{PromQuery, QueryLanguageParser, QueryStatement};
 use rstest::rstest;
@@ -945,3 +952,198 @@ async fn anon_promql_ratio_repro(instance: Arc<dyn MockInstance>) {
        whole[0].pretty_print()
    );
 }
+
+#[apply(both_instances_cases)]
+async fn promql_stale_marker_excludes_series_across_flushes(instance: Arc<dyn MockInstance>) {
+    let ins = instance.frontend();
+    let table = "promql_stale_marker";
+    execute_all(
+        &ins,
+        &format!(
+            "CREATE TABLE {table} (\
+                series STRING, \
+                ts TIMESTAMP TIME INDEX, \
+                val DOUBLE, \
+                PRIMARY KEY (series)\
+            ) WITH (sst_format = 'flat')"
+        ),
+        QueryContext::arc(),
+    )
+    .await;
+
+    let rows = |samples: &[(&str, f64, i64)]| Rows {
+        schema: vec![
+            ColumnSchema {
+                column_name: "series".to_string(),
+                datatype: ColumnDataType::String as i32,
+                semantic_type: SemanticType::Tag as i32,
+                ..Default::default()
+            },
+            ColumnSchema {
+                column_name: "ts".to_string(),
+                datatype: ColumnDataType::TimestampMillisecond as i32,
+                semantic_type: SemanticType::Timestamp as i32,
+                ..Default::default()
+            },
+            ColumnSchema {
+                column_name: "val".to_string(),
+                datatype: ColumnDataType::Float64 as i32,
+                semantic_type: SemanticType::Field as i32,
+                ..Default::default()
+            },
+        ],
+        rows: samples
+            .iter()
+            .map(|(series, value, timestamp)| Row {
+                values: vec![
+                    ValueData::StringValue((*series).to_string()).into(),
+                    ValueData::TimestampMillisecondValue(*timestamp).into(),
+                    ValueData::F64Value(*value).into(),
+                ],
+            })
+            .collect(),
+    };
+    let insert = |samples: &[(&str, f64, i64)]| RowInsertRequests {
+        inserts: vec![RowInsertRequest {
+            table_name: table.to_string(),
+            rows: Some(rows(samples)),
+        }],
+    };
+
+    ins.handle_row_inserts(
+        insert(&[("stale", 10.0, 1_000), ("ordinary", 20.0, 1_000)]),
+        QueryContext::arc(),
+        false,
+        false,
+    )
+    .await
+    .unwrap();
+    execute_all(
+        &ins,
+        &format!("ADMIN FLUSH_TABLE('{table}')"),
+        QueryContext::arc(),
+    )
+    .await;
+
+    let ordinary_nan = f64::from_bits(0x7ff8_0000_0000_0000);
+    ins.handle_row_inserts(
+        insert(&[
+            ("stale", f64::from_bits(PROMETHEUS_STALE_NAN_BITS), 2_000),
+            ("ordinary", ordinary_nan, 2_000),
+        ]),
+        QueryContext::arc(),
+        false,
+        false,
+    )
+    .await
+    .unwrap();
+
+    let extract_samples = |batches: common_recordbatch::RecordBatches| {
+        let mut samples = batches
+            .iter()
+            .flat_map(|batch| {
+                let series = cast(
+                    batch.column_by_name("series").unwrap().as_ref(),
+                    &DataType::Utf8,
+                )
+                .unwrap();
+                let series = series.as_any().downcast_ref::<StringArray>().unwrap();
+                let values = batch
+                    .column_by_name("val")
+                    .unwrap()
+                    .as_any()
+                    .downcast_ref::<Float64Array>()
+                    .unwrap();
+                let timestamps = batch
+                    .column_by_name("ts")
+                    .unwrap()
+                    .as_any()
+                    .downcast_ref::<TimestampMillisecondArray>()
+                    .unwrap();
+                (0..batch.num_rows())
+                    .map(|row| {
+                        (
+                            series.value(row).to_string(),
+                            values.value(row),
+                            timestamps.value(row),
+                        )
+                    })
+                    .collect::<Vec<_>>()
+            })
+            .collect::<Vec<_>>();
+        samples.sort_by(|left, right| left.0.cmp(&right.0));
+        samples
+    };
+
+    let raw_output = ins
+        .do_query(
+            &format!("SELECT series, val, ts FROM {table} WHERE ts = 2000"),
+            QueryContext::arc(),
+        )
+        .await
+        .remove(0)
+        .unwrap();
+    let raw_batches = match raw_output.data {
+        OutputData::Stream(stream) => collect_batches(stream).await.unwrap(),
+        OutputData::RecordBatches(recordbatches) => recordbatches,
+        _ => unreachable!(),
+    };
+    let raw_samples = extract_samples(raw_batches);
+    assert_eq!(raw_samples.len(), 2, "{raw_samples:?}");
+    let stale_marker = raw_samples
+        .iter()
+        .find(|(series, _, _)| series == "stale")
+        .unwrap();
+    assert_eq!(stale_marker.1.to_bits(), PROMETHEUS_STALE_NAN_BITS);
+    let ordinary_marker = raw_samples
+        .iter()
+        .find(|(series, _, _)| series == "ordinary")
+        .unwrap();
+    assert_eq!(ordinary_marker.1.to_bits(), ordinary_nan.to_bits());
+
+    // Keep the finite 1000ms sample eligible so a stale marker cannot fall back to it.
+    let query_at = |at| {
+        let time = UNIX_EPOCH.checked_add(Duration::from_millis(at)).unwrap();
+        promql_query_as_batches(
+            ins.clone(),
+            table,
+            None,
+            QueryContext::arc(),
+            time,
+            time,
+            Duration::from_secs(1),
+            Duration::from_secs(3),
+        )
+    };
+    let assert_nan_series = |samples: Vec<(String, f64, i64)>, at| {
+        assert_eq!(samples.len(), 1, "{samples:?}");
+        assert_eq!(samples[0].0, "ordinary", "{samples:?}");
+        assert_eq!(samples[0].2, at, "{samples:?}");
+        assert_eq!(
+            samples[0].1.to_bits(),
+            ordinary_nan.to_bits(),
+            "{samples:?}"
+        );
+    };
+
+    for flush_markers in [false, true] {
+        if flush_markers {
+            execute_all(
+                &ins,
+                &format!("ADMIN FLUSH_TABLE('{table}')"),
+                QueryContext::arc(),
+            )
+            .await;
+        }
+
+        assert_eq!(
+            extract_samples(query_at(1_500).await),
+            vec![
+                ("ordinary".to_string(), 20.0, 1_500),
+                ("stale".to_string(), 10.0, 1_500)
+            ]
+        );
+        assert_nan_series(extract_samples(query_at(2_000).await), 2_000);
+        assert_nan_series(extract_samples(query_at(2_500).await), 2_500);
+    }
+}
@@ -429,7 +429,7 @@ tql analyze sum(aggr_optimize_not);
 |_|_|_ProjectionExec: expr=[greptime_timestamp@4 as greptime_timestamp, greptime_value@5 as greptime_value] REDACTED
 |_|_|_PromInstantManipulateExec: range=[0..0], lookback=[300000], interval=[300000], time index=[greptime_timestamp] REDACTED
 |_|_|_PromSeriesDivideExec: tags=["a", "b", "c", "d"] REDACTED
-|_|_|_SeriesScan: region=REDACTED, "partition_count":{"count":0, "mem_ranges":0, "files":0, "file_ranges":0}, "distribution":"PerSeries", "mode":"legacy" REDACTED
+|_|_|_SeriesScan: region=REDACTED, "partition_count":{"count":0, "mem_ranges":0, "files":0, "file_ranges":0}, "selector":"LastRow { after_merge: true }", "distribution":"PerSeries", "mode":"legacy" REDACTED
 |_|_|_|
 | 1_| 1_|_AggregateExec: mode=FinalPartitioned, gby=[greptime_timestamp@0 as greptime_timestamp], aggr=[__sum_state(aggr_optimize_not.greptime_value)] REDACTED
 |_|_|_RepartitionExec: partitioning=REDACTED
@@ -437,7 +437,7 @@ tql analyze sum(aggr_optimize_not);
 |_|_|_ProjectionExec: expr=[greptime_timestamp@4 as greptime_timestamp, greptime_value@5 as greptime_value] REDACTED
 |_|_|_PromInstantManipulateExec: range=[0..0], lookback=[300000], interval=[300000], time index=[greptime_timestamp] REDACTED
 |_|_|_PromSeriesDivideExec: tags=["a", "b", "c", "d"] REDACTED
-|_|_|_SeriesScan: region=REDACTED, "partition_count":{"count":0, "mem_ranges":0, "files":0, "file_ranges":0}, "distribution":"PerSeries", "mode":"legacy" REDACTED
+|_|_|_SeriesScan: region=REDACTED, "partition_count":{"count":0, "mem_ranges":0, "files":0, "file_ranges":0}, "selector":"LastRow { after_merge: true }", "distribution":"PerSeries", "mode":"legacy" REDACTED
 |_|_|_|
 |_|_| Total rows: 0_|
 +-+-+-+
@@ -113,7 +113,7 @@ explain analyze
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@1 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":0, "files":1, "file_ranges":1}, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":0, "files":1, "file_ranges":1}, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 |_|_| Total rows: 4_|
 +-+-+-+
@@ -165,7 +165,7 @@ explain analyze
 | 1_| 0_|_AggregateExec: mode=Final, gby=[], aggr=[last_value(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[last_value(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":0, "files":1, "file_ranges":1}, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":0, "files":1, "file_ranges":1}, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 |_|_| Total rows: 1_|
 +-+-+-+
@@ -277,7 +277,7 @@ order by ordered_host;
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@1 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 1_|_ProjectionExec: expr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST]@0 as ordered_host, last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST]@1 as last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]@2 as last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_SortPreservingMergeExec: [last_value(t.host) ORDER BY [t.ts ASC NULLS LAST]@0 ASC NULLS LAST] REDACTED
@@ -286,7 +286,7 @@ order by ordered_host;
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@1 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 2_|_ProjectionExec: expr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST]@0 as ordered_host, last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST]@1 as last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]@2 as last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_SortPreservingMergeExec: [last_value(t.host) ORDER BY [t.ts ASC NULLS LAST]@0 ASC NULLS LAST] REDACTED
@@ -295,7 +295,7 @@ order by ordered_host;
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@1 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 |_|_| Total rows: 4_|
 +-+-+-+
@@ -355,17 +355,17 @@ explain analyze
 | 1_| 0_|_AggregateExec: mode=Final, gby=[], aggr=[__last_value_state(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__last_value_state(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 1_|_AggregateExec: mode=Final, gby=[], aggr=[__last_value_state(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__last_value_state(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 2_|_AggregateExec: mode=Final, gby=[], aggr=[__last_value_state(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__last_value_state(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 |_|_| Total rows: 1_|
 +-+-+-+
@@ -559,17 +559,17 @@ explain analyze
 | 1_| 0_|_AggregateExec: mode=Final, gby=[], aggr=[__last_value_state(t1.ts) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__last_value_state(t1.ts) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 1_|_AggregateExec: mode=Final, gby=[], aggr=[__last_value_state(t1.ts) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__last_value_state(t1.ts) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 2_|_AggregateExec: mode=Final, gby=[], aggr=[__last_value_state(t1.ts) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__last_value_state(t1.ts) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 |_|_| Total rows: 1_|
 +-+-+-+
@@ -648,7 +648,7 @@ order by ordered_host;
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST], last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@0 as host], aggr=[last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST], last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 1_|_ProjectionExec: expr=[last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST]@0 as ordered_host, last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]@1 as last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
 |_|_|_SortPreservingMergeExec: [last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST]@0 ASC NULLS LAST] REDACTED
@@ -657,7 +657,7 @@ order by ordered_host;
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST], last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@0 as host], aggr=[last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST], last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 2_|_ProjectionExec: expr=[last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST]@0 as ordered_host, last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]@1 as last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
 |_|_|_SortPreservingMergeExec: [last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST]@0 ASC NULLS LAST] REDACTED
@@ -666,7 +666,7 @@ order by ordered_host;
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST], last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@0 as host], aggr=[last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST], last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 |_|_| Total rows: 4_|
 +-+-+-+
@@ -0,0 +1,548 @@
+-- Correctness coverage for the instant-selector last-row optimization.
+-- Each TQL EVAL below has distinct values so its chosen physical sample is visible.
+-- The default instant lookback is (T - 300s, T]: exclude its lower bound,
+-- include a sample just inside it and at T, and exclude a future sample.
+CREATE TABLE instant_last_lookback (
+    ts TIMESTAMP TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+Affected Rows: 0
+
+INSERT INTO instant_last_lookback VALUES
+    (700000, 10, 'lower_excluded'),
+    (700001, 11, 'just_inside'),
+    (1000000, 12, 'upper_included'),
+    (1000000, 14, 'future_has_prior'),
+    (1000001, 13, 'future_has_prior');
+
+Affected Rows: 5
+
+ADMIN FLUSH_TABLE('instant_last_lookback');
+
+--------------------------------------------+
+| ADMIN FLUSH_TABLE('instant_last_lookback') |
+--------------------------------------------+
+| 0                                          |
+--------------------------------------------+
+
+-- Expected pairs: just_inside=11, upper_included=12, future_has_prior=14.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_lookback;
+
+---------------------+------+------------------+
+| ts                  | val  | series           |
+---------------------+------+------------------+
+| 1970-01-01T00:16:40 | 11.0 | just_inside      |
+| 1970-01-01T00:16:40 | 12.0 | upper_included   |
+| 1970-01-01T00:16:40 | 14.0 | future_has_prior |
+---------------------+------+------------------+
+
+-- An empty table must remain an empty instant vector.
+CREATE TABLE instant_last_empty (
+    ts TIMESTAMP TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+Affected Rows: 0
+
+-- Expected: no rows.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_empty;
+
++
++
+
+DROP TABLE instant_last_empty;
+
+Affected Rows: 0
+
+DROP TABLE instant_last_lookback;
+
+Affected Rows: 0
+
+-- Positive offset reads earlier data; negative offset reads later data.  Repeat
+-- the earlier evaluation after a later one to catch a cached instant window.
+CREATE TABLE instant_last_offset (
+    ts TIMESTAMP TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+Affected Rows: 0
+
+INSERT INTO instant_last_offset VALUES
+    (940000, 21, 'offset'),
+    (1000000, 22, 'offset'),
+    (1060000, 23, 'offset');
+
+Affected Rows: 3
+
+ADMIN FLUSH_TABLE('instant_last_offset');
+
+------------------------------------------+
+| ADMIN FLUSH_TABLE('instant_last_offset') |
+------------------------------------------+
+| 0                                        |
+------------------------------------------+
+
+-- Expected value: 22.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_offset;
+
+---------------------+------+--------+
+| ts                  | val  | series |
+---------------------+------+--------+
+| 1970-01-01T00:16:40 | 22.0 | offset |
+---------------------+------+--------+
+
+-- Expected value: 21 (evaluation time shifted back 60s).
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_offset offset 60s;
+
+---------------------+------+--------+
+| ts                  | val  | series |
+---------------------+------+--------+
+| 1970-01-01T00:16:40 | 21.0 | offset |
+---------------------+------+--------+
+
+-- Expected value: 23 (evaluation time shifted forward 60s).
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_offset offset -60s;
+
+---------------------+------+--------+
+| ts                  | val  | series |
+---------------------+------+--------+
+| 1970-01-01T00:16:40 | 23.0 | offset |
+---------------------+------+--------+
+
+-- Expected value: 23.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1060, 1060, '1s') instant_last_offset;
+
+---------------------+------+--------+
+| ts                  | val  | series |
+---------------------+------+--------+
+| 1970-01-01T00:17:40 | 23.0 | offset |
+---------------------+------+--------+
+
+-- Expected value: 22 again, not the later evaluation's value.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_offset;
+
+---------------------+------+--------+
+| ts                  | val  | series |
+---------------------+------+--------+
+| 1970-01-01T00:16:40 | 22.0 | offset |
+---------------------+------+--------+
+
+DROP TABLE instant_last_offset;
+
+Affected Rows: 0
+
+-- Put t1 and t2 in one SST. Deleting newest t2 must reveal t1 whether the
+-- Delete is still in the memtable or has been flushed to a newer SST.
+CREATE TABLE instant_last_delete (
+    ts TIMESTAMP TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+Affected Rows: 0
+
+INSERT INTO instant_last_delete VALUES
+    (900000, 31, 'delete'),
+    (1000000, 32, 'delete');
+
+Affected Rows: 2
+
+ADMIN FLUSH_TABLE('instant_last_delete');
+
+------------------------------------------+
+| ADMIN FLUSH_TABLE('instant_last_delete') |
+------------------------------------------+
+| 0                                        |
+------------------------------------------+
+
+DELETE FROM instant_last_delete WHERE series = 'delete' AND ts = 1000000;
+
+Affected Rows: 1
+
+-- Expected value: 31; the newest point is a memtable Delete.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_delete;
+
+---------------------+------+--------+
+| ts                  | val  | series |
+---------------------+------+--------+
+| 1970-01-01T00:16:40 | 31.0 | delete |
+---------------------+------+--------+
+
+ADMIN FLUSH_TABLE('instant_last_delete');
+
+------------------------------------------+
+| ADMIN FLUSH_TABLE('instant_last_delete') |
+------------------------------------------+
+| 0                                        |
+------------------------------------------+
+
+-- Expected value: 31; the Delete is now in an SST.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_delete;
+
+---------------------+------+--------+
+| ts                  | val  | series |
+---------------------+------+--------+
+| 1970-01-01T00:16:40 | 31.0 | delete |
+---------------------+------+--------+
+
+-- A same-timestamp memtable overwrite of t1 must win over the old SST value.
+INSERT INTO instant_last_delete VALUES (900000, 33, 'delete');
+
+Affected Rows: 1
+
+-- Expected value: 33.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_delete;
+
+---------------------+------+--------+
+| ts                  | val  | series |
+---------------------+------+--------+
+| 1970-01-01T00:16:40 | 33.0 | delete |
+---------------------+------+--------+
+
+-- A newer memtable point must win, and retain that identity after its flush.
+INSERT INTO instant_last_delete VALUES (1010000, 34, 'delete');
+
+Affected Rows: 1
+
+-- Expected value: 34.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1010, 1010, '1s') instant_last_delete;
+
+---------------------+------+--------+
+| ts                  | val  | series |
+---------------------+------+--------+
+| 1970-01-01T00:16:50 | 34.0 | delete |
+---------------------+------+--------+
+
+ADMIN FLUSH_TABLE('instant_last_delete');
+
+------------------------------------------+
+| ADMIN FLUSH_TABLE('instant_last_delete') |
+------------------------------------------+
+| 0                                        |
+------------------------------------------+
+
+-- Expected value: 33 after flush; it overwrites t1 in an older SST.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_delete;
+
+---------------------+------+--------+
+| ts                  | val  | series |
+---------------------+------+--------+
+| 1970-01-01T00:16:40 | 33.0 | delete |
+---------------------+------+--------+
+
+-- Expected value: 34 after flush.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1010, 1010, '1s') instant_last_delete;
+
+---------------------+------+--------+
+| ts                  | val  | series |
+---------------------+------+--------+
+| 1970-01-01T00:16:50 | 34.0 | delete |
+---------------------+------+--------+
+
+DROP TABLE instant_last_delete;
+
+Affected Rows: 0
+
+-- Ordinary IEEE NaN is a valid PromQL sample and must not be suppressed.
+CREATE TABLE instant_last_nan (
+    ts TIMESTAMP TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+Affected Rows: 0
+
+INSERT INTO instant_last_nan VALUES (1000000, 'NaN'::DOUBLE, 'ordinary_nan');
+
+Affected Rows: 1
+
+ADMIN FLUSH_TABLE('instant_last_nan');
+
+---------------------------------------+
+| ADMIN FLUSH_TABLE('instant_last_nan') |
+---------------------------------------+
+| 0                                     |
+---------------------------------------+
+
+-- Expected value: NaN, with the ordinary_nan series present.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_nan;
+
+---------------------+-----+--------------+
+| ts                  | val | series       |
+---------------------+-----+--------------+
+| 1970-01-01T00:16:40 | NaN | ordinary_nan |
+---------------------+-----+--------------+
+
+DROP TABLE instant_last_nan;
+
+Affected Rows: 0
+
+-- PromQL normalizes to milliseconds; these distinct raw timestamps collide at
+-- 1s. LastRow must preserve the unhinted path's selected samples: 42 and 52.
+CREATE TABLE instant_last_sec (
+    ts TIMESTAMP(0) TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+Affected Rows: 0
+
+INSERT INTO instant_last_sec VALUES
+    (0, 61, 'sec'),
+    (1, 62, 'sec'),
+    (2, 63, 'sec');
+
+Affected Rows: 3
+
+ADMIN FLUSH_TABLE('instant_last_sec');
+
+---------------------------------------+
+| ADMIN FLUSH_TABLE('instant_last_sec') |
+---------------------------------------+
+| 0                                     |
+---------------------------------------+
+
+-- Expected value: 62 at the native seconds boundary.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1, 1, '1s') instant_last_sec;
+
+------+--------+---------------------+
+| val  | series | ts                  |
+------+--------+---------------------+
+| 62.0 | sec    | 1970-01-01T00:00:01 |
+------+--------+---------------------+
+
+CREATE TABLE instant_last_micro (
+    ts TIMESTAMP(6) TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+Affected Rows: 0
+
+INSERT INTO instant_last_micro VALUES
+    (999999, 41, 'micro'),
+    (1000000, 42, 'micro'),
+    (1000001, 43, 'micro');
+
+Affected Rows: 3
+
+ADMIN FLUSH_TABLE('instant_last_micro');
+
+-----------------------------------------+
+| ADMIN FLUSH_TABLE('instant_last_micro') |
+-----------------------------------------+
+| 0                                       |
+-----------------------------------------+
+
+-- Expected value: 42, matching the unhinted millisecond-normalized path.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1, 1, '1s') instant_last_micro;
+
+------+--------+---------------------+
+| val  | series | ts                  |
+------+--------+---------------------+
+| 42.0 | micro  | 1970-01-01T00:00:01 |
+------+--------+---------------------+
+
+CREATE TABLE instant_last_nano (
+    ts TIMESTAMP(9) TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+Affected Rows: 0
+
+INSERT INTO instant_last_nano VALUES
+    (999999999, 51, 'nano'),
+    (1000000000, 52, 'nano'),
+    (1000000001, 53, 'nano');
+
+Affected Rows: 3
+
+ADMIN FLUSH_TABLE('instant_last_nano');
+
+----------------------------------------+
+| ADMIN FLUSH_TABLE('instant_last_nano') |
+----------------------------------------+
+| 0                                      |
+----------------------------------------+
+
+-- Expected value: 52, matching the unhinted millisecond-normalized path.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1, 1, '1s') instant_last_nano;
+
+------+--------+---------------------+
+| val  | series | ts                  |
+------+--------+---------------------+
+| 52.0 | nano   | 1970-01-01T00:00:01 |
+------+--------+---------------------+
+
+DROP TABLE instant_last_sec;
+
+Affected Rows: 0
+
+DROP TABLE instant_last_micro;
+
+Affected Rows: 0
+
+DROP TABLE instant_last_nano;
+
+Affected Rows: 0
+
+-- A field selector chooses the val column; it does not filter val. PromQL must
+-- apply the comparison after selecting the latest eligible sample, rather than
+-- falling back to an older sample that satisfies the comparison.
+CREATE TABLE instant_last_field_filter (
+    ts TIMESTAMP TIME INDEX,
+    val DOUBLE,
+    host STRING,
+    instance STRING,
+    PRIMARY KEY (host, instance)
+) ENGINE=mito;
+
+Affected Rows: 0
+
+-- Keep the older matching sample in an SST and the latest non-matching sample
+-- in the memtable. Both timestamps are eligible at 1s.
+INSERT INTO instant_last_field_filter VALUES (900, 10, 'host-a', 'instance-a');
+
+Affected Rows: 1
+
+ADMIN FLUSH_TABLE('instant_last_field_filter');
+
+------------------------------------------------+
+| ADMIN FLUSH_TABLE('instant_last_field_filter') |
+------------------------------------------------+
+| 0                                              |
+------------------------------------------------+
+
+INSERT INTO instant_last_field_filter VALUES (1000, 1, 'host-a', 'instance-a');
+
+Affected Rows: 1
+
+-- Expected: no rows. The selected latest value is 1, so it must not fall back
+-- to the older value 10 merely because that value is greater than 5.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1, 1, '1s') instant_last_field_filter{__field__="val"} > 5;
+
++
++
+
+-- A value matcher filters the scan before selecting a sample: the older 10
+-- remains eligible, unlike the post-selection comparison above.
+TQL EVAL (1, 1, '1s') instant_last_field_filter{val="10.0"};
+
+---------------------+------+--------+------------+
+| ts                  | val  | host   | instance   |
+---------------------+------+--------+------------+
+| 1970-01-01T00:00:01 | 10.0 | host-a | instance-a |
+---------------------+------+--------+------------+
+
+-- SQL filters rows before aggregation, so the older matching value remains.
+-- Expected value: 10.
+SELECT last_value(val ORDER BY ts) FROM instant_last_field_filter WHERE val > 5;
+
+--------------------------------------------------------------------------------------------------+
+| last_value(instant_last_field_filter.val) ORDER BY [instant_last_field_filter.ts ASC NULLS LAST] |
+--------------------------------------------------------------------------------------------------+
+| 10.0                                                                                             |
+--------------------------------------------------------------------------------------------------+
+
+DROP TABLE instant_last_field_filter;
+
+Affected Rows: 0
+
+-- Range evaluation is a control: range functions need their complete windows,
+-- not a single last row.  At 10s and 20s, [11s] contains two samples.
+CREATE TABLE instant_last_range_control (
+    ts TIMESTAMP TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+Affected Rows: 0
+
+INSERT INTO instant_last_range_control VALUES
+    (0, 0, 'range'),
+    (10000, 10, 'range'),
+    (20000, 20, 'range');
+
+Affected Rows: 3
+
+ADMIN FLUSH_TABLE('instant_last_range_control');
+
+-------------------------------------------------+
+| ADMIN FLUSH_TABLE('instant_last_range_control') |
+-------------------------------------------------+
+| 0                                               |
+-------------------------------------------------+
+
+-- Expected instant values at 0s, 10s, and 20s: 0, 10, and 20.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (0, 20, '10s') instant_last_range_control;
+
+---------------------+------+--------+
+| ts                  | val  | series |
+---------------------+------+--------+
+| 1970-01-01T00:00:00 | 0.0  | range  |
+| 1970-01-01T00:00:10 | 10.0 | range  |
+| 1970-01-01T00:00:20 | 20.0 | range  |
+---------------------+------+--------+
+
+-- Expected last values at 0s, 10s, and 20s: 0, 10, and 20.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (0, 20, '10s') last_over_time(instant_last_range_control[11s]);
+
+---------------------+-----------------------------------+--------+
+| ts                  | prom_last_over_time(ts_range,val) | series |
+---------------------+-----------------------------------+--------+
+| 1970-01-01T00:00:00 | 0.0                               | range  |
+| 1970-01-01T00:00:10 | 10.0                              | range  |
+| 1970-01-01T00:00:20 | 20.0                              | range  |
+---------------------+-----------------------------------+--------+
+
+-- At 10s, rate is 10/11 because counter-zero extrapolation stops at 0s;
+-- at 20s it is 1. The 0s window has only one sample.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (0, 20, '10s') rate(instant_last_range_control[11s]);
+
+---------------------+-----------------------------------------+--------+
+| ts                  | prom_rate(ts_range,val,ts,Int64(11000)) | series |
+---------------------+-----------------------------------------+--------+
+| 1970-01-01T00:00:10 | 0.9090909090909092                      | range  |
+| 1970-01-01T00:00:20 | 1.0                                     | range  |
+---------------------+-----------------------------------------+--------+
+
+-- Expected rate at 20s: 1 from both points in the [11s] window.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (20, 20, '1s') rate(instant_last_range_control[11s]);
+
+---------------------+-----------------------------------------+--------+
+| ts                  | prom_rate(ts_range,val,ts,Int64(11000)) | series |
+---------------------+-----------------------------------------+--------+
+| 1970-01-01T00:00:20 | 1.0                                     | range  |
+---------------------+-----------------------------------------+--------+
+
+DROP TABLE instant_last_range_control;
+
+Affected Rows: 0
+
@@ -0,0 +1,258 @@
+-- Correctness coverage for the instant-selector last-row optimization.
+-- Each TQL EVAL below has distinct values so its chosen physical sample is visible.
+
+-- The default instant lookback is (T - 300s, T]: exclude its lower bound,
+-- include a sample just inside it and at T, and exclude a future sample.
+CREATE TABLE instant_last_lookback (
+    ts TIMESTAMP TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+INSERT INTO instant_last_lookback VALUES
+    (700000, 10, 'lower_excluded'),
+    (700001, 11, 'just_inside'),
+    (1000000, 12, 'upper_included'),
+    (1000000, 14, 'future_has_prior'),
+    (1000001, 13, 'future_has_prior');
+ADMIN FLUSH_TABLE('instant_last_lookback');
+
+-- Expected pairs: just_inside=11, upper_included=12, future_has_prior=14.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_lookback;
+
+-- An empty table must remain an empty instant vector.
+CREATE TABLE instant_last_empty (
+    ts TIMESTAMP TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+-- Expected: no rows.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_empty;
+
+DROP TABLE instant_last_empty;
+DROP TABLE instant_last_lookback;
+
+-- Positive offset reads earlier data; negative offset reads later data.  Repeat
+-- the earlier evaluation after a later one to catch a cached instant window.
+CREATE TABLE instant_last_offset (
+    ts TIMESTAMP TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+INSERT INTO instant_last_offset VALUES
+    (940000, 21, 'offset'),
+    (1000000, 22, 'offset'),
+    (1060000, 23, 'offset');
+ADMIN FLUSH_TABLE('instant_last_offset');
+
+-- Expected value: 22.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_offset;
+
+-- Expected value: 21 (evaluation time shifted back 60s).
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_offset offset 60s;
+
+-- Expected value: 23 (evaluation time shifted forward 60s).
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_offset offset -60s;
+
+-- Expected value: 23.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1060, 1060, '1s') instant_last_offset;
+
+-- Expected value: 22 again, not the later evaluation's value.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_offset;
+
+DROP TABLE instant_last_offset;
+
+-- Put t1 and t2 in one SST. Deleting newest t2 must reveal t1 whether the
+-- Delete is still in the memtable or has been flushed to a newer SST.
+CREATE TABLE instant_last_delete (
+    ts TIMESTAMP TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+INSERT INTO instant_last_delete VALUES
+    (900000, 31, 'delete'),
+    (1000000, 32, 'delete');
+ADMIN FLUSH_TABLE('instant_last_delete');
+
+DELETE FROM instant_last_delete WHERE series = 'delete' AND ts = 1000000;
+
+-- Expected value: 31; the newest point is a memtable Delete.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_delete;
+
+ADMIN FLUSH_TABLE('instant_last_delete');
+
+-- Expected value: 31; the Delete is now in an SST.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_delete;
+
+-- A same-timestamp memtable overwrite of t1 must win over the old SST value.
+INSERT INTO instant_last_delete VALUES (900000, 33, 'delete');
+
+-- Expected value: 33.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_delete;
+
+-- A newer memtable point must win, and retain that identity after its flush.
+INSERT INTO instant_last_delete VALUES (1010000, 34, 'delete');
+
+-- Expected value: 34.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1010, 1010, '1s') instant_last_delete;
+
+ADMIN FLUSH_TABLE('instant_last_delete');
+
+-- Expected value: 33 after flush; it overwrites t1 in an older SST.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_delete;
+
+-- Expected value: 34 after flush.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1010, 1010, '1s') instant_last_delete;
+
+DROP TABLE instant_last_delete;
+
+-- Ordinary IEEE NaN is a valid PromQL sample and must not be suppressed.
+CREATE TABLE instant_last_nan (
+    ts TIMESTAMP TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+INSERT INTO instant_last_nan VALUES (1000000, 'NaN'::DOUBLE, 'ordinary_nan');
+ADMIN FLUSH_TABLE('instant_last_nan');
+
+-- Expected value: NaN, with the ordinary_nan series present.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1000, 1000, '1s') instant_last_nan;
+
+DROP TABLE instant_last_nan;
+
+-- PromQL normalizes to milliseconds; these distinct raw timestamps collide at
+-- 1s. LastRow must preserve the unhinted path's selected samples: 42 and 52.
+CREATE TABLE instant_last_sec (
+    ts TIMESTAMP(0) TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+INSERT INTO instant_last_sec VALUES
+    (0, 61, 'sec'),
+    (1, 62, 'sec'),
+    (2, 63, 'sec');
+ADMIN FLUSH_TABLE('instant_last_sec');
+
+-- Expected value: 62 at the native seconds boundary.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1, 1, '1s') instant_last_sec;
+
+CREATE TABLE instant_last_micro (
+    ts TIMESTAMP(6) TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+INSERT INTO instant_last_micro VALUES
+    (999999, 41, 'micro'),
+    (1000000, 42, 'micro'),
+    (1000001, 43, 'micro');
+ADMIN FLUSH_TABLE('instant_last_micro');
+
+-- Expected value: 42, matching the unhinted millisecond-normalized path.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1, 1, '1s') instant_last_micro;
+
+CREATE TABLE instant_last_nano (
+    ts TIMESTAMP(9) TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+INSERT INTO instant_last_nano VALUES
+    (999999999, 51, 'nano'),
+    (1000000000, 52, 'nano'),
+    (1000000001, 53, 'nano');
+ADMIN FLUSH_TABLE('instant_last_nano');
+
+-- Expected value: 52, matching the unhinted millisecond-normalized path.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1, 1, '1s') instant_last_nano;
+
+DROP TABLE instant_last_sec;
+DROP TABLE instant_last_micro;
+DROP TABLE instant_last_nano;
+
+-- A field selector chooses the val column; it does not filter val. PromQL must
+-- apply the comparison after selecting the latest eligible sample, rather than
+-- falling back to an older sample that satisfies the comparison.
+CREATE TABLE instant_last_field_filter (
+    ts TIMESTAMP TIME INDEX,
+    val DOUBLE,
+    host STRING,
+    instance STRING,
+    PRIMARY KEY (host, instance)
+) ENGINE=mito;
+
+-- Keep the older matching sample in an SST and the latest non-matching sample
+-- in the memtable. Both timestamps are eligible at 1s.
+INSERT INTO instant_last_field_filter VALUES (900, 10, 'host-a', 'instance-a');
+ADMIN FLUSH_TABLE('instant_last_field_filter');
+INSERT INTO instant_last_field_filter VALUES (1000, 1, 'host-a', 'instance-a');
+
+-- Expected: no rows. The selected latest value is 1, so it must not fall back
+-- to the older value 10 merely because that value is greater than 5.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1, 1, '1s') instant_last_field_filter{__field__="val"} > 5;
+
+-- A value matcher filters the scan before selecting a sample: the older 10
+-- remains eligible, unlike the post-selection comparison above.
+TQL EVAL (1, 1, '1s') instant_last_field_filter{val="10.0"};
+
+-- SQL filters rows before aggregation, so the older matching value remains.
+-- Expected value: 10.
+SELECT last_value(val ORDER BY ts) FROM instant_last_field_filter WHERE val > 5;
+
+DROP TABLE instant_last_field_filter;
+
+-- Range evaluation is a control: range functions need their complete windows,
+-- not a single last row.  At 10s and 20s, [11s] contains two samples.
+CREATE TABLE instant_last_range_control (
+    ts TIMESTAMP TIME INDEX,
+    val DOUBLE,
+    series STRING PRIMARY KEY
+) ENGINE=mito;
+
+INSERT INTO instant_last_range_control VALUES
+    (0, 0, 'range'),
+    (10000, 10, 'range'),
+    (20000, 20, 'range');
+ADMIN FLUSH_TABLE('instant_last_range_control');
+
+-- Expected instant values at 0s, 10s, and 20s: 0, 10, and 20.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (0, 20, '10s') instant_last_range_control;
+
+-- Expected last values at 0s, 10s, and 20s: 0, 10, and 20.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (0, 20, '10s') last_over_time(instant_last_range_control[11s]);
+
+-- At 10s, rate is 10/11 because counter-zero extrapolation stops at 0s;
+-- at 20s it is 1. The 0s window has only one sample.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (0, 20, '10s') rate(instant_last_range_control[11s]);
+
+-- Expected rate at 20s: 1 from both points in the [11s] window.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (20, 20, '1s') rate(instant_last_range_control[11s]);
+
+DROP TABLE instant_last_range_control;
@@ -75,7 +75,7 @@ TQL ANALYZE VERBOSE (0, 0, '1s') test{host=~".*"};
 | 1_| 0_|_PromInstantManipulateExec: range=[0..0], lookback=[300000], interval=[1000], time index=[ts] REDACTED
 |_|_|_PromSeriesDivideExec: tags=["host"] REDACTED
 |_|_|_CooperativeExec REDACTED
-|_|_|_SeriesScan: region=REDACTED, {"partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "distribution":"PerSeries", "projection": ["ts", "host", "val"], "filters": ["ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(0, None)"], "REDACTED
+|_|_|_SeriesScan: region=REDACTED, {"partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "selector":"LastRow { after_merge: true }", "distribution":"PerSeries", "projection": ["ts", "host", "val"], "filters": ["ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(0, None)"], "REDACTED
 |_|_|_|
 |_|_| Total rows: 2_|
 +-+-+-+
@@ -99,7 +99,7 @@ TQL ANALYZE VERBOSE (0, 0, '1s') test{host=~".+"};
 | 1_| 0_|_PromInstantManipulateExec: range=[0..0], lookback=[300000], interval=[1000], time index=[ts] REDACTED
 |_|_|_PromSeriesDivideExec: tags=["host"] REDACTED
 |_|_|_CooperativeExec REDACTED
-|_|_|_SeriesScan: region=REDACTED, {"partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "distribution":"PerSeries", "projection": ["ts", "host", "val"], "filters": ["host != Dictionary(UInt32, Utf8(\"\"))", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(0, None)"], "REDACTED
+|_|_|_SeriesScan: region=REDACTED, {"partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "selector":"LastRow { after_merge: true }", "distribution":"PerSeries", "projection": ["ts", "host", "val"], "filters": ["host != Dictionary(UInt32, Utf8(\"\"))", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(0, None)"], "REDACTED
 |_|_|_|
 |_|_| Total rows: 2_|
 +-+-+-+
@@ -145,7 +145,7 @@ TQL ANALYZE VERBOSE (0, 0, '1s') test{host!~".+"};
 | 1_| 0_|_PromInstantManipulateExec: range=[0..0], lookback=[300000], interval=[1000], time index=[ts] REDACTED
 |_|_|_PromSeriesDivideExec: tags=["host"] REDACTED
 |_|_|_CooperativeExec REDACTED
-|_|_|_SeriesScan: region=REDACTED, {"partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "distribution":"PerSeries", "projection": ["ts", "host", "val"], "filters": ["host = Dictionary(UInt32, Utf8(\"\"))", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(0, None)"], "REDACTED
+|_|_|_SeriesScan: region=REDACTED, {"partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "selector":"LastRow { after_merge: true }", "distribution":"PerSeries", "projection": ["ts", "host", "val"], "filters": ["host = Dictionary(UInt32, Utf8(\"\"))", "ts >= TimestampMillisecond(-299999, None)", "ts <= TimestampMillisecond(0, None)"], "REDACTED
 |_|_|_|
 |_|_| Total rows: 0_|
 +-+-+-+
@@ -87,7 +87,7 @@ explain analyze
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@1 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 |_|_| Total rows: 4_|
 +-+-+-+
@@ -139,7 +139,7 @@ explain analyze
 | 1_| 0_|_AggregateExec: mode=Final, gby=[], aggr=[last_value(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[last_value(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 |_|_| Total rows: 1_|
 +-+-+-+
@@ -113,7 +113,7 @@ explain analyze
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@1 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":0, "files":1, "file_ranges":1}, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":0, "files":1, "file_ranges":1}, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 |_|_| Total rows: 4_|
 +-+-+-+
@@ -165,7 +165,7 @@ explain analyze
 | 1_| 0_|_AggregateExec: mode=Final, gby=[], aggr=[last_value(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[last_value(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":0, "files":1, "file_ranges":1}, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":{"count":1, "mem_ranges":0, "files":1, "file_ranges":1}, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 |_|_| Total rows: 1_|
 +-+-+-+
@@ -276,7 +276,7 @@ order by ordered_host;
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@1 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 1_|_SortPreservingMergeExec: [ordered_host@0 ASC NULLS LAST] REDACTED
 |_|_|_SortExec: expr=[ordered_host@0 ASC NULLS LAST], preserve_REDACTED
@@ -284,7 +284,7 @@ order by ordered_host;
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@1 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 2_|_SortPreservingMergeExec: [ordered_host@0 ASC NULLS LAST] REDACTED
 |_|_|_SortExec: expr=[ordered_host@0 ASC NULLS LAST], preserve_REDACTED
@@ -292,7 +292,7 @@ order by ordered_host;
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@1 as host], aggr=[last_value(t.host) ORDER BY [t.ts ASC NULLS LAST], last_value(t.not_pk) ORDER BY [t.ts ASC NULLS LAST], last_value(t.val) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 |_|_| Total rows: 4_|
 +-+-+-+
@@ -352,17 +352,17 @@ explain analyze
 | 1_| 0_|_AggregateExec: mode=Final, gby=[], aggr=[__last_value_state(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__last_value_state(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 1_|_AggregateExec: mode=Final, gby=[], aggr=[__last_value_state(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__last_value_state(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 2_|_AggregateExec: mode=Final, gby=[], aggr=[__last_value_state(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__last_value_state(t.ts) ORDER BY [t.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 |_|_| Total rows: 1_|
 +-+-+-+
@@ -556,17 +556,17 @@ explain analyze
 | 1_| 0_|_AggregateExec: mode=Final, gby=[], aggr=[__last_value_state(t1.ts) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__last_value_state(t1.ts) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 1_|_AggregateExec: mode=Final, gby=[], aggr=[__last_value_state(t1.ts) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__last_value_state(t1.ts) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 2_|_AggregateExec: mode=Final, gby=[], aggr=[__last_value_state(t1.ts) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
 |_|_|_CoalescePartitionsExec REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__last_value_state(t1.ts) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 |_|_| Total rows: 1_|
 +-+-+-+
@@ -644,7 +644,7 @@ order by ordered_host;
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST], last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@0 as host], aggr=[last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST], last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 1_|_SortPreservingMergeExec: [ordered_host@0 ASC NULLS LAST] REDACTED
 |_|_|_SortExec: expr=[ordered_host@0 ASC NULLS LAST], preserve_REDACTED
@@ -652,7 +652,7 @@ order by ordered_host;
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST], last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@0 as host], aggr=[last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST], last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 | 1_| 2_|_SortPreservingMergeExec: [ordered_host@0 ASC NULLS LAST] REDACTED
 |_|_|_SortExec: expr=[ordered_host@0 ASC NULLS LAST], preserve_REDACTED
@@ -660,7 +660,7 @@ order by ordered_host;
 |_|_|_AggregateExec: mode=FinalPartitioned, gby=[host@0 as host], aggr=[last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST], last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
 |_|_|_RepartitionExec: REDACTED
 |_|_|_AggregateExec: mode=Partial, gby=[host@0 as host], aggr=[last_value(t1.host) ORDER BY [t1.ts ASC NULLS LAST], last_value(t1.val) ORDER BY [t1.ts ASC NULLS LAST]] REDACTED
-|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow" REDACTED
+|_|_|_SeqScan: region=REDACTED, "partition_count":REDACTED, "selector":"LastRow { after_merge: false }" REDACTED
 |_|_|_|
 |_|_| Total rows: 4_|
 +-+-+-+
@@ -297,7 +297,7 @@ TQL ANALYZE sum(test2);
 |_|_|_ProjectionExec: expr=[greptime_timestamp@0 as greptime_timestamp, greptime_value@1 as greptime_value] REDACTED
 |_|_|_PromInstantManipulateExec: range=[0..0], lookback=[300000], interval=[300000], time index=[greptime_timestamp] REDACTED
 |_|_|_PromSeriesDivideExec: tags=["shard"] REDACTED
-|_|_|_SeriesScan: region=REDACTED, "partition_count":{"count":0, "mem_ranges":0, "files":0, "file_ranges":0}, "distribution":"PerSeries", "mode":"legacy" REDACTED
+|_|_|_SeriesScan: region=REDACTED, "partition_count":{"count":0, "mem_ranges":0, "files":0, "file_ranges":0}, "selector":"LastRow { after_merge: true }", "distribution":"PerSeries", "mode":"legacy" REDACTED
 |_|_|_|
 | 1_| 1_|_AggregateExec: mode=FinalPartitioned, gby=[greptime_timestamp@0 as greptime_timestamp], aggr=[__sum_state(test2.greptime_value)] REDACTED
 |_|_|_RepartitionExec: partitioning=REDACTED
@@ -305,7 +305,7 @@ TQL ANALYZE sum(test2);
 |_|_|_ProjectionExec: expr=[greptime_timestamp@0 as greptime_timestamp, greptime_value@1 as greptime_value] REDACTED
 |_|_|_PromInstantManipulateExec: range=[0..0], lookback=[300000], interval=[300000], time index=[greptime_timestamp] REDACTED
 |_|_|_PromSeriesDivideExec: tags=["shard"] REDACTED
-|_|_|_SeriesScan: region=REDACTED, "partition_count":{"count":0, "mem_ranges":0, "files":0, "file_ranges":0}, "distribution":"PerSeries", "mode":"legacy" REDACTED
+|_|_|_SeriesScan: region=REDACTED, "partition_count":{"count":0, "mem_ranges":0, "files":0, "file_ranges":0}, "selector":"LastRow { after_merge: true }", "distribution":"PerSeries", "mode":"legacy" REDACTED
 |_|_|_|
 |_|_| Total rows: 0_|
 +-+-+-+
@@ -204,7 +204,7 @@ case for issue #7913. It writes 8192 series × 20160 samples through remote-writ
 in 1440-sample daily time chunks, flushing after each chunk before running 1d/7d/14d
 TQL selectors. It is not included in the default `all` case set because ingestion
 cost dominates routine CI validation. Commenting `/query-regression heavy` runs
-only this case; `/query-regression` runs the six routine default cases. Manual
+only this case; `/query-regression` runs the seven routine default cases. Manual
 workflow dispatch accepts the `heavy` token to select this case.

 ## OTLP trace load scenario
@@ -393,12 +393,12 @@ parquetbench/scanbench` as the read-bench tool against each target's data direct

 The workflow runs when an allowlisted repository admin comments
 `/query-regression` on a non-draft PR. It does not rerun on pushes,
-ready-for-review, or reopen events. `/query-regression` runs the six routine
-default cases; `/query-regression heavy` runs only the high-cardinality
-remote-write #7913 case. PR runs build base/candidate once and
-use `--allow-large-fixture`. Manual `workflow_dispatch` runs can pass `all`,
-`heavy`, one case path, or a comma/whitespace-separated list of case paths, and
-can override refs.
+ready-for-review, or reopen events. `/query-regression` runs the seven routine
+default cases, including `promql_instant_last_row_9034`;
+`/query-regression heavy` runs only the high-cardinality remote-write #7913 case.
+PR runs build base/candidate once and use `--allow-large-fixture`. Manual
+`workflow_dispatch` runs can pass `all`, `heavy`, one case path, or a
+comma/whitespace-separated list of case paths, and can override refs.

 Comment admission is two workflows. `slash-command-dispatch.yml` uses
 [peter-evans/slash-command-dispatch](https://github.com/peter-evans/slash-command-dispatch)
@@ -77,7 +77,12 @@ cycles series labels across rows. `series_layout = "timestamp_major"` writes all
 series for one timestamp before advancing to the next timestamp; use it for
 Prometheus-like high-cardinality scrape fixtures where short query windows should
 still contain many raw samples. `timestamp_major` requires `rows_per_sst` to be
-divisible by `series_count`.
+divisible by `series_count`. `per_sst` uses one series selected from the SST
+index. These describe logical generation order. Before either flat or primary-key
+SST format is written, each completed generated batch is physically sorted by
+encoded primary key ascending, timestamp ascending, and sequence descending, as
+required by the Mito Parquet writer; sorting preserves every generated column and
+row, including deterministic-wave values.

 `[scenario]` is required. Other scenario variants are intentionally unsupported
 for now, but `scenario.kind` leaves room for future `write_then_query` and
@@ -0,0 +1,92 @@
+# PromQL instant-selector latency regression coverage for #9034.
+#
+# Instant, range, and aggregate controls cover LastRow latency behavior. The
+# candidate instant query must exercise the existing LastRow selector; the range
+# query must not. This is latency regression coverage only: optimizer and
+# sqlness tests own plan and correctness coverage. Evaluation is 0.7s after the
+# final generated sample (last timestamp = start + 1023 * 0.1s =
+# 1704067302.3) and remains inside the default lookback window.
+
+[case]
+name = "promql_instant_last_row_9034"
+description = "PromQL instant selector LastRow latency regression for #9034"
+
+[scenario]
+kind = "direct_readable_sst"
+seed = 9034
+
+[[scenario.tables]]
+database = "public"
+name = "promql_instant_last_row_9034"
+engine = "mito"
+append_mode = true
+sst_format = "flat"
+primary_key = ["host", "instance"]
+time_index = "ts"
+
+[[scenario.tables.columns]]
+name = "host"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 16, prefix = "host" }
+
+[[scenario.tables.columns]]
+name = "instance"
+type = "STRING"
+semantic = "tag"
+distribution = { kind = "cardinality", values = 256, prefix = "instance" }
+
+[[scenario.tables.columns]]
+name = "value"
+type = "DOUBLE"
+semantic = "field"
+distribution = { kind = "deterministic_wave", min = 0.0, max = 1000.0 }
+
+[[scenario.tables.columns]]
+name = "ts"
+type = "TIMESTAMP(3)"
+semantic = "timestamp"
+
+[scenario.layout]
+regions = 1
+sst_count = 64
+rows_per_sst = 4096
+row_group_size = 1024
+series_count = 256
+start_unix_nanos = 1704067200000000000
+step_nanos = 100000000
+time_range_layout = "non_overlapping_per_sst"
+series_layout = "timestamp_major"
+
+# Candidate instant query must exercise the existing LastRow selector.
+[[scenario.queries]]
+name = "instant_selector_last_row_candidate"
+kind = "tql"
+query = "TQL ANALYZE VERBOSE (1704067303, 1704067303, '1s') promql_instant_last_row_9034{host=~'host.*'}"
+warmup = 3
+iterations = 9
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 25
+
+# True range control: this must not enable the LastRow selector.
+[[scenario.queries]]
+name = "range_selector_last_row_control"
+kind = "tql"
+query = "TQL ANALYZE VERBOSE (1704067293, 1704067303, '1s') promql_instant_last_row_9034{host=~'host.*'}"
+warmup = 3
+iterations = 9
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 25
+
+# This pre-existing aggregate-hint control should return one row per 256 series; it is latency regression coverage, not a result oracle.
+[[scenario.queries]]
+name = "sql_aggregate_last_row_control"
+kind = "sql"
+query = "SELECT host, instance, last_value(value ORDER BY ts ASC) FROM promql_instant_last_row_9034 GROUP BY host, instance"
+warmup = 3
+iterations = 9
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 25
@@ -40,6 +40,7 @@ class QueryRegressionCaseSelectionTest(unittest.TestCase):
                "tests/perf/query_cases/prom_remote_write_mixed_every/case.toml",
                "tests/perf/query_cases/prom_remote_write_integer_counter/case.toml",
                "tests/perf/query_cases/promql_range_boundary/case.toml",
+                "tests/perf/query_cases/promql_instant_last_row_9034/case.toml",
            ],
        )