fix(promql): preserve native timestamps through sample selection

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

src/promql/src/extension_plan.rs

@@ -28,8 +28,13 @@ mod union_distinct_on;
 pub use absent::{Absent, AbsentExec, AbsentStream};
 use common_query::native_histogram::{SUM_FIELD, native_histogram_value_type};
 use common_query::prometheus::is_prometheus_stale_nan;
-use datafusion::arrow::array::{Array, Float64Array, StructArray};
-use datafusion::arrow::datatypes::{ArrowPrimitiveType, TimestampMillisecondType};
+use datafusion::arrow::array::{
+    Array, Float64Array, StructArray, TimestampMicrosecondArray, TimestampMillisecondArray,
+    TimestampNanosecondArray, TimestampSecondArray,
+};
+use datafusion::arrow::datatypes::{
+    ArrowPrimitiveType, DataType, TimeUnit, TimestampMillisecondType,
+};
 use datafusion::common::DFSchemaRef;
 use datafusion::error::{DataFusionError, Result as DataFusionResult};
 use datatypes::data_type::DataType as _;
@@ -47,6 +52,50 @@ pub use union_distinct_on::{UnionDistinctOn, UnionDistinctOnExec, UnionDistinctO
 
 pub type Millisecond = <TimestampMillisecondType as ArrowPrimitiveType>::Native;
 
+/// Returns a timestamp value without reducing its Arrow storage precision.
+pub(crate) fn native_timestamp_values(array: &dyn Array) -> datafusion::error::Result<Vec<i64>> {
+    let value = match array.data_type() {
+        DataType::Timestamp(TimeUnit::Second, _) => array
+            .as_any()
+            .downcast_ref::<TimestampSecondArray>()
+            .map(|a| a.values().to_vec()),
+        DataType::Timestamp(TimeUnit::Millisecond, _) => array
+            .as_any()
+            .downcast_ref::<TimestampMillisecondArray>()
+            .map(|a| a.values().to_vec()),
+        DataType::Timestamp(TimeUnit::Microsecond, _) => array
+            .as_any()
+            .downcast_ref::<TimestampMicrosecondArray>()
+            .map(|a| a.values().to_vec()),
+        DataType::Timestamp(TimeUnit::Nanosecond, _) => array
+            .as_any()
+            .downcast_ref::<TimestampNanosecondArray>()
+            .map(|a| a.values().to_vec()),
+        _ => None,
+    };
+    value.ok_or_else(|| {
+        datafusion::error::DataFusionError::Execution("Time index column is not a timestamp".into())
+    })
+}
+
+pub(crate) fn timestamp_unit(data_type: &DataType) -> datafusion::error::Result<TimeUnit> {
+    match data_type {
+        DataType::Timestamp(unit, _) => Ok(*unit),
+        _ => Err(datafusion::error::DataFusionError::Execution(
+            "Time index column is not a timestamp".into(),
+        )),
+    }
+}
+
+pub(crate) fn native_per_nanosecond(unit: TimeUnit) -> i128 {
+    match unit {
+        TimeUnit::Second => 1_000_000_000,
+        TimeUnit::Millisecond => 1_000_000,
+        TimeUnit::Microsecond => 1_000,
+        TimeUnit::Nanosecond => 1,
+    }
+}
+
 const METRIC_NUM_SERIES: &str = "num_series";
 
 fn prometheus_stale_sample_column(column: &dyn Array) -> Option<(&dyn Array, &Float64Array)> {

src/promql/src/extension_plan/instant_manipulate.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 use std::any::Any;
-use std::cmp::Ordering;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
@@ -29,6 +28,7 @@ use datafusion::execution::context::TaskContext;
 use datafusion::logical_expr::{
     EmptyRelation, Expr, Extension, LogicalPlan, UserDefinedLogicalNodeCore,
 };
+use datafusion::physical_expr::EquivalenceProperties;
 use datafusion::physical_plan::metrics::{
     BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder, MetricValue, MetricsSet,
 };
@@ -46,8 +46,9 @@ use snafu::ResultExt;
 use crate::error::{DeserializeSnafu, Result};
 use crate::extension_plan::series_divide::SeriesDivide;
 use crate::extension_plan::{
-    METRIC_NUM_SERIES, Millisecond, is_prometheus_stale_sample, prometheus_stale_sample_column,
-    resolve_column_name, serialize_column_index,
+    METRIC_NUM_SERIES, Millisecond, is_prometheus_stale_sample, native_per_nanosecond,
+    native_timestamp_values, prometheus_stale_sample_column, resolve_column_name,
+    serialize_column_index, timestamp_unit,
 };
 use crate::metrics::PROMQL_SERIES_COUNT;
 
@@ -67,7 +68,7 @@ fn mixed_sample_fields(field: Option<&str>) -> [Option<&str>; 2] {
 /// This plan will try to align the input time series, for every timestamp between
 /// `start` and `end` with step `interval`. Find in the `lookback` range if data
 /// is missing at the given timestamp.
-#[derive(Debug, PartialEq, Eq, Hash, PartialOrd)]
+#[derive(Debug, PartialEq, Eq, Hash)]
 pub struct InstantManipulate {
     start: Millisecond,
     end: Millisecond,
@@ -79,9 +80,35 @@ pub struct InstantManipulate {
     /// Primary sample column used to derive the columns checked for staleness.
     field_column: Option<String>,
     input: LogicalPlan,
+    output_schema: DFSchemaRef,
     unfix: Option<UnfixIndices>,
 }
 
+impl PartialOrd for InstantManipulate {
+    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
+        (
+            self.start,
+            self.end,
+            self.lookback_delta,
+            self.interval,
+            &self.time_index_column,
+            &self.tag_columns,
+            &self.field_column,
+            &self.input,
+        )
+            .partial_cmp(&(
+                other.start,
+                other.end,
+                other.lookback_delta,
+                other.interval,
+                &other.time_index_column,
+                &other.tag_columns,
+                &other.field_column,
+                &other.input,
+            ))
+    }
+}
+
 #[derive(Debug, PartialEq, Eq, Hash, PartialOrd)]
 struct UnfixIndices {
     pub time_index_idx: u64,
@@ -98,7 +125,7 @@ impl UserDefinedLogicalNodeCore for InstantManipulate {
     }
 
     fn schema(&self) -> &DFSchemaRef {
-        self.input.schema()
+        &self.output_schema
     }
 
     fn expressions(&self) -> Vec<Expr> {
@@ -180,6 +207,7 @@ impl UserDefinedLogicalNodeCore for InstantManipulate {
                 end: self.end,
                 lookback_delta: self.lookback_delta,
                 interval: self.interval,
+                output_schema: Self::calculate_output_schema(&input, &time_index_column)?,
                 time_index_column,
                 tag_columns: Self::resolve_tag_columns(&input, &self.tag_columns),
                 field_column,
@@ -195,6 +223,7 @@ impl UserDefinedLogicalNodeCore for InstantManipulate {
                 time_index_column: self.time_index_column.clone(),
                 tag_columns: Self::resolve_tag_columns(&input, &self.tag_columns),
                 field_column: self.field_column.clone(),
+                output_schema: Self::calculate_output_schema(&input, &self.time_index_column)?,
                 input,
                 unfix: None,
             })
@@ -203,6 +232,38 @@ impl UserDefinedLogicalNodeCore for InstantManipulate {
 }
 
 impl InstantManipulate {
+    fn calculate_output_schema(
+        input: &LogicalPlan,
+        time_index_column: &str,
+    ) -> DataFusionResult<DFSchemaRef> {
+        let input_schema = input.schema();
+        let time_index = input_schema
+            .index_of_column_by_name(None, time_index_column)
+            .ok_or_else(|| {
+                DataFusionError::Internal(format!(
+                    "InstantManipulate time index {time_index_column} not found"
+                ))
+            })?;
+        let mut fields = (0..input_schema.fields().len())
+            .map(|index| {
+                let (qualifier, field) = input_schema.qualified_field(index);
+                (qualifier.cloned(), field.clone())
+            })
+            .collect::<Vec<_>>();
+        let (qualifier, field) = input_schema.qualified_field(time_index);
+        fields[time_index] = (
+            qualifier.cloned(),
+            Arc::new(field.as_ref().clone().with_data_type(DataType::Timestamp(
+                datafusion::arrow::datatypes::TimeUnit::Millisecond,
+                None,
+            ))),
+        );
+        Ok(Arc::new(DFSchema::new_with_metadata(
+            fields,
+            input_schema.metadata().clone(),
+        )?))
+    }
+
     #[allow(clippy::too_many_arguments)]
     pub fn new(
         start: Millisecond,
@@ -219,6 +280,8 @@ impl InstantManipulate {
             end,
             lookback_delta,
             interval,
+            output_schema: Self::calculate_output_schema(&input, &time_index_column)
+                .unwrap_or_else(|_| input.schema().clone()),
             time_index_column,
             tag_columns,
             field_column,
@@ -274,6 +337,25 @@ impl InstantManipulate {
     pub fn to_execution_plan(&self, exec_input: Arc<dyn ExecutionPlan>) -> Arc<dyn ExecutionPlan> {
         let reuse_tsid_column = matches!(self.tag_columns.as_slice(), [tag] if tag == "__tsid");
 
+        let mut fields = exec_input.schema().fields().to_vec();
+        let time_index = exec_input
+            .schema()
+            .index_of(&self.time_index_column)
+            .expect("time index column not found");
+        fields[time_index] = Arc::new(fields[time_index].as_ref().clone().with_data_type(
+            DataType::Timestamp(datafusion::arrow::datatypes::TimeUnit::Millisecond, None),
+        ));
+        let output_schema = Arc::new(datafusion::arrow::datatypes::Schema::new_with_metadata(
+            fields,
+            exec_input.schema().metadata().clone(),
+        ));
+        let input_properties = exec_input.properties();
+        let properties = Arc::new(PlanProperties::new(
+            EquivalenceProperties::new(output_schema.clone()),
+            input_properties.partitioning.clone(),
+            input_properties.emission_type,
+            input_properties.boundedness,
+        ));
         Arc::new(InstantManipulateExec {
             start: self.start,
             end: self.end,
@@ -283,6 +365,8 @@ impl InstantManipulate {
             field_column: self.field_column.clone(),
             reuse_tsid_column,
             input: exec_input,
+            output_schema,
+            properties,
             metric: ExecutionPlanMetricsSet::new(),
         })
     }
@@ -311,9 +395,10 @@ impl InstantManipulate {
     pub fn deserialize(bytes: &[u8]) -> Result<Self> {
         let pb_instant_manipulate =
             pb::InstantManipulate::decode(bytes).context(DeserializeSnafu)?;
+        let empty_schema = Arc::new(DFSchema::empty());
         let placeholder_plan = LogicalPlan::EmptyRelation(EmptyRelation {
             produce_one_row: false,
-            schema: Arc::new(DFSchema::empty()),
+            schema: empty_schema.clone(),
         });
 
         let unfix = UnfixIndices {
@@ -329,6 +414,7 @@ impl InstantManipulate {
             time_index_column: String::new(),
             tag_columns: Vec::new(),
             field_column: None,
+            output_schema: empty_schema,
             input: placeholder_plan,
             unfix: Some(unfix),
         })
@@ -346,6 +432,8 @@ pub struct InstantManipulateExec {
     reuse_tsid_column: bool,
 
     input: Arc<dyn ExecutionPlan>,
+    output_schema: SchemaRef,
+    properties: Arc<PlanProperties>,
     metric: ExecutionPlanMetricsSet,
 }
 
@@ -355,11 +443,11 @@ impl ExecutionPlan for InstantManipulateExec {
     }
 
     fn schema(&self) -> SchemaRef {
-        self.input.schema()
+        self.output_schema.clone()
     }
 
     fn properties(&self) -> &Arc<PlanProperties> {
-        self.input.properties()
+        &self.properties
     }
 
     fn required_input_distribution(&self) -> Vec<Distribution> {
@@ -380,6 +468,14 @@ impl ExecutionPlan for InstantManipulateExec {
         children: Vec<Arc<dyn ExecutionPlan>>,
     ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
         assert!(!children.is_empty());
+        let input = children[0].clone();
+        let input_properties = input.properties();
+        let properties = Arc::new(PlanProperties::new(
+            EquivalenceProperties::new(self.output_schema.clone()),
+            input_properties.partitioning.clone(),
+            input_properties.emission_type,
+            input_properties.boundedness,
+        ));
         Ok(Arc::new(Self {
             start: self.start,
             end: self.end,
@@ -388,7 +484,9 @@ impl ExecutionPlan for InstantManipulateExec {
             time_index_column: self.time_index_column.clone(),
             field_column: self.field_column.clone(),
             reuse_tsid_column: self.reuse_tsid_column,
-            input: children[0].clone(),
+            input,
+            output_schema: self.output_schema.clone(),
+            properties,
             metric: self.metric.clone(),
         }))
     }
@@ -413,6 +511,7 @@ impl ExecutionPlan for InstantManipulateExec {
             .column_with_name(&self.time_index_column)
             .expect("time index column not found")
             .0;
+        let time_unit = timestamp_unit(schema.field(time_index).data_type())?;
         let field_indices = mixed_sample_fields(self.field_column.as_deref()).map(|field| {
             field.and_then(|field| schema.column_with_name(field).map(|(index, _)| index))
         });
@@ -426,10 +525,11 @@ impl ExecutionPlan for InstantManipulateExec {
             lookback_delta: self.lookback_delta,
             interval: self.interval,
             time_index,
+            time_unit,
             field_indices,
             tsid_index,
             reuse_tsid_column: self.reuse_tsid_column && tsid_index.is_some(),
-            schema,
+            schema: self.output_schema.clone(),
             input,
             metric: baseline_metric,
             num_series,
@@ -493,6 +593,7 @@ pub struct InstantManipulateStream {
     interval: Millisecond,
     // Column index of TIME INDEX column's position in schema
     time_index: usize,
+    time_unit: datafusion::arrow::datatypes::TimeUnit,
     field_indices: [Option<usize>; 2],
     tsid_index: Option<usize>,
     reuse_tsid_column: bool,
@@ -516,9 +617,6 @@ impl Stream for InstantManipulateStream {
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let poll = match ready!(self.input.poll_next_unpin(cx)) {
             Some(Ok(batch)) => {
-                if batch.num_rows() == 0 {
-                    return Poll::Pending;
-                }
                 let timer = std::time::Instant::now();
                 self.num_series.add(1);
                 let result = Ok(batch).and_then(|batch| self.manipulate(batch));
@@ -543,22 +641,11 @@ impl InstantManipulateStream {
     /// lookback window `(eval_ts - lookback_delta, eval_ts]`; a sample at exactly
     /// `eval_ts - lookback_delta` is too old.
     pub fn manipulate(&self, input: RecordBatch) -> DataFusionResult<RecordBatch> {
-        let ts_column = input
-            .column(self.time_index)
-            .as_any()
-            .downcast_ref::<TimestampMillisecondArray>()
-            .ok_or_else(|| {
-                DataFusionError::Execution(
-                    "Time index Column downcast to TimestampMillisecondArray failed".into(),
-                )
-            })?;
-
-        // Early return for empty input
+        let ts_column = input.column(self.time_index);
         if ts_column.is_empty() {
-            return Ok(input);
+            return Ok(RecordBatch::new_empty(self.schema.clone()));
         }
-
-        // Field columns for staleness checks, classified once per batch.
+        let scale = native_per_nanosecond(self.time_unit);
         let stale_sample_columns = self.field_indices.map(|index| {
             index.and_then(|index| prometheus_stale_sample_column(input.column(index).as_ref()))
         });
@@ -568,101 +655,71 @@ impl InstantManipulateStream {
                 .flatten()
                 .any(|column| is_prometheus_stale_sample(*column, row))
         };
-
-        // Optimize iteration range based on actual data bounds
-        let first_ts = ts_column.value(0);
-        let last_ts = ts_column.value(ts_column.len() - 1);
-        // A sample at `t` is eligible for eval time `eval_ts` iff:
-        //   t > eval_ts - lookback_delta  <=>  eval_ts < t + lookback_delta.
-        // Therefore the last eval timestamp for which the last sample is still eligible is:
-        //   last_ts + lookback_delta - 1 (millisecond granularity).
-        let last_useful = if self.lookback_delta > 0 {
-            last_ts + self.lookback_delta - 1
+        let timestamps = native_timestamp_values(ts_column.as_ref())?;
+        let len = timestamps.len();
+        let to_nanoseconds = |timestamp: i64| (timestamp as i128) * scale;
+        let first_ns = to_nanoseconds(timestamps[0]);
+        let last_ns = to_nanoseconds(timestamps[len - 1]);
+        // An exact sample remains useful with zero lookback. Otherwise the lower
+        // boundary is exclusive, so subtract one nanosecond from its final window.
+        let last_useful = if self.lookback_delta == 0 {
+            last_ns
         } else {
-            last_ts
+            last_ns + (self.lookback_delta as i128) * 1_000_000 - 1
+        };
+        let first_ms = (first_ns + 999_999).div_euclid(1_000_000);
+        let last_ms = last_useful.div_euclid(1_000_000);
+        let query_start = self.start as i128;
+        let query_end = self.end as i128;
+        let interval = self.interval as i128;
+        let max_start = first_ms.max(query_start);
+        let min_end = last_ms.min(query_end);
+        let (aligned_start, aligned_end) = if max_start > min_end {
+            (1, 0)
+        } else {
+            (
+                query_start + (max_start - query_start) / interval * interval,
+                query_end - (query_end - min_end) / interval * interval,
+            )
         };
-
-        let max_start = first_ts.max(self.start);
-        let min_end = last_useful.min(self.end);
-
-        let aligned_start = self.start + (max_start - self.start) / self.interval * self.interval;
-        let aligned_end = self.end - (self.end - min_end) / self.interval * self.interval;
-
         let estimated_points = if aligned_end >= aligned_start {
-            ((aligned_end - aligned_start) / self.interval).saturating_add(1) as usize
+            (aligned_end - aligned_start) / interval + 1
         } else {
             0
         };
-        if estimated_points > MAX_INSTANT_MANIPULATE_OUTPUT_POINTS {
+        if estimated_points > MAX_INSTANT_MANIPULATE_OUTPUT_POINTS as i128 {
             return Err(DataFusionError::Execution(format!(
                 "InstantManipulate output points exceed limit: {estimated_points} > {MAX_INSTANT_MANIPULATE_OUTPUT_POINTS}"
             )));
         }
+        let estimated_points = estimated_points as usize;
+        let aligned_start = aligned_start as i64;
+        let aligned_end = aligned_end as i64;
         let mut take_indices = Vec::with_capacity(estimated_points);
-
-        let mut cursor = 0;
-
-        let aligned_ts_iter = (aligned_start..=aligned_end).step_by(self.interval as usize);
         let mut aligned_ts = Vec::with_capacity(estimated_points);
-
-        // calculate the offsets to take
-        'next: for expected_ts in aligned_ts_iter {
-            // first, search toward end to see if there is matched timestamp
-            while cursor < ts_column.len() {
-                let curr = ts_column.value(cursor);
-                match curr.cmp(&expected_ts) {
-                    Ordering::Equal => {
-                        if is_stale(cursor) {
-                            // Ignore the stale marker.
-                        } else {
-                            take_indices.push(cursor as u64);
-                            aligned_ts.push(expected_ts);
-                        }
-                        continue 'next;
-                    }
-                    Ordering::Greater => break,
-                    Ordering::Less => {}
+        let mut cursor = 0;
+        for expected_ms in (aligned_start..=aligned_end).step_by(self.interval as usize) {
+            let expected = (expected_ms as i128) * 1_000_000;
+            let mut exact_candidate = None;
+            while cursor < len && to_nanoseconds(timestamps[cursor]) <= expected {
+                if to_nanoseconds(timestamps[cursor]) == expected && exact_candidate.is_none() {
+                    exact_candidate = Some(cursor);
                 }
                 cursor += 1;
             }
-            if cursor == ts_column.len() {
-                cursor -= 1;
-                // short cut this loop
-                if ts_column.value(cursor) + self.lookback_delta <= expected_ts {
-                    break;
-                }
-            }
-
-            // then examine the value
-            let curr_ts = ts_column.value(cursor);
-            if curr_ts + self.lookback_delta <= expected_ts {
+            let Some(candidate) = exact_candidate.or_else(|| cursor.checked_sub(1)) else {
                 continue;
-            }
-            if curr_ts > expected_ts {
-                // exceeds current expected timestamp, examine the previous value
-                if let Some(prev_cursor) = cursor.checked_sub(1) {
-                    let prev_ts = ts_column.value(prev_cursor);
-                    if prev_ts + self.lookback_delta > expected_ts {
-                        // only use the point in the time range
-                        if is_stale(prev_cursor) {
-                            // Do not use a stale marker as the newest value.
-                            continue;
-                        }
-                        // use this point
-                        take_indices.push(prev_cursor as u64);
-                        aligned_ts.push(expected_ts);
-                    }
-                }
-            } else if is_stale(cursor) {
-                // Do not use a stale marker as the newest value.
-            } else {
-                // use this point
-                take_indices.push(cursor as u64);
-                aligned_ts.push(expected_ts);
+            };
+            let candidate_ts = to_nanoseconds(timestamps[candidate]);
+            let lower = expected - (self.lookback_delta as i128) * 1_000_000;
+            if (candidate_ts == expected || candidate_ts > lower)
+                && candidate_ts <= expected
+                && !is_stale(candidate)
+            {
+                take_indices.push(candidate as u64);
+                aligned_ts.push(expected_ms);
             }
         }
-
-        // take record batch and replace the time index column
         self.take_record_batch_optional(input, take_indices, aligned_ts)
     }
 
@@ -696,7 +753,7 @@ impl InstantManipulateStream {
             arrays.push(compute::take(array, indices_array, None)?);
         }
 
-        let result = RecordBatch::try_new(record_batch.schema(), arrays)
+        let result = RecordBatch::try_new(self.schema.clone(), arrays)
             .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))?;
         Ok(result)
     }
@@ -718,9 +775,11 @@ fn reuse_constant_column(array: &Arc<dyn Array>, len: usize) -> DataFusionResult
 mod test {
     use common_query::native_histogram::build_histogram_array;
     use common_query::prometheus::PROMETHEUS_STALE_NAN_BITS;
-    use datafusion::arrow::array::Float64Array;
+    use datafusion::arrow::array::{
+        Float64Array, TimestampMicrosecondArray, TimestampNanosecondArray,
+    };
     use datafusion::arrow::buffer::NullBuffer;
-    use datafusion::arrow::datatypes::{DataType, Field, Schema};
+    use datafusion::arrow::datatypes::{DataType, Field, Schema, TimeUnit};
     use datafusion::common::ToDFSchema;
     use datafusion::datasource::memory::MemorySourceConfig;
     use datafusion::datasource::source::DataSourceExec;
@@ -753,6 +812,8 @@ mod test {
             time_index_column: TIME_INDEX_COLUMN.to_string(),
             field_column: Some("value".to_string()),
             reuse_tsid_column: false,
+            output_schema: memory_exec.schema(),
+            properties: memory_exec.properties().clone(),
             input: memory_exec,
             metric: ExecutionPlanMetricsSet::new(),
         });
@@ -767,6 +828,136 @@ mod test {
         assert_eq!(result_literal, expected);
     }
 
+    #[tokio::test]
+    async fn native_timestamps_select_exact_samples_and_keep_ms_output() {
+        for (unit, ticks_per_ms) in [
+            (TimeUnit::Microsecond, 1_000_i64),
+            (TimeUnit::Nanosecond, 1_000_000_i64),
+        ] {
+            let lower = 1_000 * ticks_per_ms;
+            let upper = 1_001 * ticks_per_ms;
+            let stale = f64::from_bits(PROMETHEUS_STALE_NAN_BITS);
+            for (name, timestamps, values, expected_timestamps, expected_values) in [
+                (
+                    "exact upper sample",
+                    vec![lower + 1, upper],
+                    vec![1.0, 2.0],
+                    vec![1_001],
+                    vec![2.0],
+                ),
+                (
+                    "exclusive lower boundary and future sample",
+                    vec![lower, upper + 1],
+                    vec![1.0, 2.0],
+                    vec![1_000],
+                    vec![1.0],
+                ),
+                (
+                    "one native tick above lower boundary",
+                    vec![lower + 1, upper + 1],
+                    vec![1.0, 2.0],
+                    vec![1_001],
+                    vec![1.0],
+                ),
+                (
+                    "future stale marker does not suppress",
+                    vec![lower + 1, upper + 1],
+                    vec![1.0, stale],
+                    vec![1_001],
+                    vec![1.0],
+                ),
+                (
+                    "latest in-window stale marker suppresses",
+                    vec![lower + 1, lower + 2, upper + 1],
+                    vec![1.0, stale, 3.0],
+                    vec![],
+                    vec![],
+                ),
+            ] {
+                let schema = Arc::new(Schema::new(vec![
+                    Field::new(TIME_INDEX_COLUMN, DataType::Timestamp(unit, None), false),
+                    Field::new("value", DataType::Float64, true),
+                ]));
+                let time: Arc<dyn Array> = match unit {
+                    TimeUnit::Microsecond => Arc::new(TimestampMicrosecondArray::from(timestamps)),
+                    TimeUnit::Nanosecond => Arc::new(TimestampNanosecondArray::from(timestamps)),
+                    _ => unreachable!(),
+                };
+                let batch = RecordBatch::try_new(
+                    schema.clone(),
+                    vec![time, Arc::new(Float64Array::from(values))],
+                )
+                .unwrap();
+                let logical_input = LogicalPlan::EmptyRelation(EmptyRelation {
+                    produce_one_row: false,
+                    schema: schema.clone().to_dfschema_ref().unwrap(),
+                });
+                let plan = InstantManipulate::new(
+                    1_000,
+                    1_001,
+                    1,
+                    1,
+                    TIME_INDEX_COLUMN.to_string(),
+                    Vec::new(),
+                    Some("value".to_string()),
+                    logical_input.clone(),
+                );
+                let output_schema = Arc::new(Schema::new(vec![
+                    Field::new(
+                        TIME_INDEX_COLUMN,
+                        DataType::Timestamp(TimeUnit::Millisecond, None),
+                        false,
+                    ),
+                    Field::new("value", DataType::Float64, true),
+                ]));
+                assert_eq!(plan.schema().as_arrow(), output_schema.as_ref());
+
+                let rebuilt = InstantManipulate::deserialize(&plan.serialize())
+                    .unwrap()
+                    .with_exprs_and_inputs(vec![], vec![logical_input])
+                    .unwrap();
+                assert_eq!(rebuilt.schema(), plan.schema());
+                assert_eq!(rebuilt.input.schema().as_arrow(), schema.as_ref());
+
+                let input = Arc::new(DataSourceExec::new(Arc::new(
+                    MemorySourceConfig::try_new(&[vec![batch]], schema.clone(), None).unwrap(),
+                )));
+                let exec = rebuilt.to_execution_plan(input);
+                assert_eq!(exec.schema(), output_schema);
+                assert_eq!(exec.children()[0].schema(), schema);
+
+                let batches =
+                    datafusion::physical_plan::collect(exec, SessionContext::default().task_ctx())
+                        .await
+                        .unwrap();
+                assert_eq!(batches.len(), 1, "{unit:?}: {name}");
+                let output = &batches[0];
+                assert_eq!(output.schema(), output_schema);
+                let timestamps = output
+                    .column(0)
+                    .as_any()
+                    .downcast_ref::<TimestampMillisecondArray>()
+                    .unwrap();
+                let values = output
+                    .column(1)
+                    .as_any()
+                    .downcast_ref::<Float64Array>()
+                    .unwrap();
+                assert_eq!(
+                    timestamps.values().as_ref(),
+                    expected_timestamps.as_slice(),
+                    "{unit:?}: {name}"
+                );
+                assert_eq!(
+                    values.values().as_ref(),
+                    expected_values.as_slice(),
+                    "{unit:?}: {name}"
+                );
+                assert_eq!(values.null_count(), 0, "{unit:?}: {name}");
+            }
+        }
+    }
+
     #[test]
     fn pruning_should_keep_time_and_field_columns_for_exec() {
         let df_schema = prepare_test_data().schema().to_dfschema_ref().unwrap();
@@ -934,6 +1125,8 @@ mod test {
             time_index_column: TIME_INDEX_COLUMN.to_string(),
             field_column: Some("value".to_string()),
             reuse_tsid_column: true,
+            output_schema: input.schema(),
+            properties: input.properties().clone(),
             input,
             metric: ExecutionPlanMetricsSet::new(),
         });
@@ -995,6 +1188,8 @@ mod test {
             time_index_column: TIME_INDEX_COLUMN.to_string(),
             field_column: Some("value".to_string()),
             reuse_tsid_column: true,
+            output_schema: input.schema(),
+            properties: input.properties().clone(),
             input,
             metric: ExecutionPlanMetricsSet::new(),
         });
@@ -1049,6 +1244,8 @@ mod test {
             time_index_column: TIME_INDEX_COLUMN.to_string(),
             field_column: Some("value".to_string()),
             reuse_tsid_column: false,
+            output_schema: input.schema(),
+            properties: input.properties().clone(),
             input,
             metric: ExecutionPlanMetricsSet::new(),
         });
@@ -1336,6 +1533,159 @@ mod test {
         .await;
     }
 
+    #[test]
+    fn exact_ties_select_first_and_lookback_uses_latest() {
+        for (values, expected_timestamp) in [
+            (vec![42.0, f64::from_bits(PROMETHEUS_STALE_NAN_BITS)], 1_000),
+            (vec![f64::from_bits(PROMETHEUS_STALE_NAN_BITS), 42.0], 1_050),
+        ] {
+            let schema = Arc::new(Schema::new(vec![
+                Field::new(
+                    TIME_INDEX_COLUMN,
+                    DataType::Timestamp(TimeUnit::Millisecond, None),
+                    false,
+                ),
+                Field::new("value", DataType::Float64, true),
+            ]));
+            let input = RecordBatch::try_new(
+                schema.clone(),
+                vec![
+                    Arc::new(TimestampMillisecondArray::from(vec![1_000, 1_000])),
+                    Arc::new(Float64Array::from(values)),
+                ],
+            )
+            .unwrap();
+            let stream = InstantManipulateStream {
+                start: 1_000,
+                end: 1_050,
+                lookback_delta: 100,
+                interval: 50,
+                time_index: 0,
+                time_unit: TimeUnit::Millisecond,
+                field_indices: [Some(1), None],
+                tsid_index: None,
+                reuse_tsid_column: false,
+                schema: schema.clone(),
+                input: Box::pin(
+                    datafusion::physical_plan::memory::MemoryStream::try_new(vec![], schema, None)
+                        .unwrap(),
+                ),
+                metric: BaselineMetrics::new(&ExecutionPlanMetricsSet::new(), 0),
+                num_series: Count::new(),
+            };
+
+            let output = stream.manipulate(input).unwrap();
+            let timestamps = output
+                .column(0)
+                .as_any()
+                .downcast_ref::<TimestampMillisecondArray>()
+                .unwrap();
+            let values = output
+                .column(1)
+                .as_any()
+                .downcast_ref::<Float64Array>()
+                .unwrap();
+            assert_eq!(timestamps.values(), &[expected_timestamp]);
+            assert_eq!(values.values(), &[42.0]);
+        }
+    }
+
+    #[test]
+    fn empty_batch_uses_declared_output_schema() {
+        let input_schema = Arc::new(Schema::new(vec![Field::new(
+            TIME_INDEX_COLUMN,
+            DataType::Timestamp(TimeUnit::Second, None),
+            false,
+        )]));
+        let output_schema = Arc::new(Schema::new(vec![Field::new(
+            TIME_INDEX_COLUMN,
+            DataType::Timestamp(TimeUnit::Millisecond, None),
+            false,
+        )]));
+        let input = RecordBatch::new_empty(input_schema.clone());
+        let stream = InstantManipulateStream {
+            start: 0,
+            end: 0,
+            lookback_delta: 0,
+            interval: 1,
+            time_index: 0,
+            time_unit: TimeUnit::Second,
+            field_indices: [None, None],
+            tsid_index: None,
+            reuse_tsid_column: false,
+            schema: output_schema.clone(),
+            input: Box::pin(
+                datafusion::physical_plan::memory::MemoryStream::try_new(
+                    vec![],
+                    input_schema,
+                    None,
+                )
+                .unwrap(),
+            ),
+            metric: BaselineMetrics::new(&ExecutionPlanMetricsSet::new(), 0),
+            num_series: Count::new(),
+        };
+
+        let output = stream.manipulate(input).unwrap();
+        assert_eq!(output.schema(), output_schema);
+        assert_eq!(
+            output.schema().field(0).data_type(),
+            &DataType::Timestamp(TimeUnit::Millisecond, None)
+        );
+    }
+
+    #[test]
+    fn extreme_alignment_retains_exact_sample() {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new(
+                TIME_INDEX_COLUMN,
+                DataType::Timestamp(TimeUnit::Millisecond, None),
+                false,
+            ),
+            Field::new("value", DataType::Float64, true),
+        ]));
+        let input = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(TimestampMillisecondArray::from(vec![i64::MAX])),
+                Arc::new(Float64Array::from(vec![7.0])),
+            ],
+        )
+        .unwrap();
+        let stream = InstantManipulateStream {
+            start: i64::MIN + 1,
+            end: i64::MAX,
+            lookback_delta: 0,
+            interval: i64::MAX,
+            time_index: 0,
+            time_unit: TimeUnit::Millisecond,
+            field_indices: [Some(1), None],
+            tsid_index: None,
+            reuse_tsid_column: false,
+            schema: schema.clone(),
+            input: Box::pin(
+                datafusion::physical_plan::memory::MemoryStream::try_new(vec![], schema, None)
+                    .unwrap(),
+            ),
+            metric: BaselineMetrics::new(&ExecutionPlanMetricsSet::new(), 0),
+            num_series: Count::new(),
+        };
+
+        let output = stream.manipulate(input).unwrap();
+        let timestamps = output
+            .column(0)
+            .as_any()
+            .downcast_ref::<TimestampMillisecondArray>()
+            .unwrap();
+        let values = output
+            .column(1)
+            .as_any()
+            .downcast_ref::<Float64Array>()
+            .unwrap();
+        assert_eq!(timestamps.values(), &[i64::MAX]);
+        assert_eq!(values.values(), &[7.0]);
+    }
+
     #[tokio::test]
     async fn ordinary_nan_is_selected_for_exact_and_lookback() {
         let schema = Arc::new(Schema::new(vec![
@@ -1365,6 +1715,8 @@ mod test {
             time_index_column: TIME_INDEX_COLUMN.to_string(),
             field_column: Some("value".to_string()),
             reuse_tsid_column: false,
+            output_schema: input.schema(),
+            properties: input.properties().clone(),
             input,
             metric: ExecutionPlanMetricsSet::new(),
         });
@@ -1444,6 +1796,8 @@ mod test {
             time_index_column: TIME_INDEX_COLUMN.to_string(),
             field_column: Some("value".to_string()),
             reuse_tsid_column: false,
+            output_schema: input.schema(),
+            properties: input.properties().clone(),
             input,
             metric: ExecutionPlanMetricsSet::new(),
         });
@@ -1507,6 +1861,8 @@ mod test {
             time_index_column: TIME_INDEX_COLUMN.to_string(),
             field_column: Some("value".to_string()),
             reuse_tsid_column: false,
+            output_schema: input.schema(),
+            properties: input.properties().clone(),
             input,
             metric: ExecutionPlanMetricsSet::new(),
         });
@@ -1554,6 +1910,8 @@ mod test {
             time_index_column: TIME_INDEX_COLUMN.to_string(),
             field_column: Some("value".to_string()),
             reuse_tsid_column: false,
+            output_schema: input.schema(),
+            properties: input.properties().clone(),
             input,
             metric: ExecutionPlanMetricsSet::new(),
         });

src/promql/src/extension_plan/normalize.rs

@@ -33,8 +33,14 @@ use datafusion::physical_plan::{
     SendableRecordBatchStream,
 };
 use datafusion_expr::col;
-use datatypes::arrow::array::TimestampMillisecondArray;
-use datatypes::arrow::datatypes::{SchemaRef, TimestampMillisecondType};
+use datatypes::arrow::array::{
+    TimestampMicrosecondArray, TimestampMillisecondArray, TimestampNanosecondArray,
+    TimestampSecondArray,
+};
+use datatypes::arrow::datatypes::{
+    SchemaRef, TimeUnit, TimestampMicrosecondType, TimestampMillisecondType,
+    TimestampNanosecondType, TimestampSecondType,
+};
 use datatypes::arrow::record_batch::RecordBatch;
 use futures::{Stream, StreamExt, ready};
 use greptime_proto::substrait_extension as pb;
@@ -44,7 +50,7 @@ use snafu::ResultExt;
 use crate::error::{DeserializeSnafu, Result};
 use crate::extension_plan::{
     METRIC_NUM_SERIES, Millisecond, is_prometheus_stale_sample, prometheus_stale_sample_column,
-    resolve_column_name, serialize_column_index,
+    resolve_column_name, serialize_column_index, timestamp_unit,
 };
 use crate::metrics::PROMQL_SERIES_COUNT;
 
@@ -393,27 +399,81 @@ pub struct SeriesNormalizeStream {
 
 impl SeriesNormalizeStream {
     pub fn normalize(&self, input: RecordBatch) -> DataFusionResult<RecordBatch> {
-        let ts_column = input
-            .column(self.time_index)
-            .as_any()
-            .downcast_ref::<TimestampMillisecondArray>()
-            .ok_or_else(|| {
-                DataFusionError::Execution(
-                    "Time index Column downcast to TimestampMillisecondArray failed".into(),
-                )
-            })?;
-
+        let time_unit = timestamp_unit(input.column(self.time_index).data_type())?;
+        let offset = match time_unit {
+            TimeUnit::Second => (self.offset % 1_000 == 0).then_some(self.offset / 1_000),
+            TimeUnit::Millisecond => Some(self.offset),
+            TimeUnit::Microsecond => self.offset.checked_mul(1_000),
+            TimeUnit::Nanosecond => self.offset.checked_mul(1_000_000),
+        }
+        .ok_or_else(|| {
+            DataFusionError::Execution("SeriesNormalize: timestamp offset overflow".into())
+        })?;
         let bias_timestamp = |timestamp: i64| {
-            timestamp.checked_add(self.offset).ok_or_else(|| {
+            timestamp.checked_add(offset).ok_or_else(|| {
                 DataFusionError::Execution("SeriesNormalize: timestamp offset overflow".into())
             })
         };
 
-        // bias the timestamp column by offset
-        let ts_column_biased = if self.offset == 0 {
-            Arc::new(ts_column.clone()) as _
-        } else {
-            Arc::new(ts_column.try_unary::<_, TimestampMillisecondType, _>(&bias_timestamp)?)
+        // Bias timestamps in their native Arrow unit; histogram start timestamps
+        // intentionally remain millisecond payloads below.
+        let ts_column_biased: Arc<dyn Array> = match time_unit {
+            TimeUnit::Second => {
+                let column = input
+                    .column(self.time_index)
+                    .as_any()
+                    .downcast_ref::<TimestampSecondArray>()
+                    .ok_or_else(|| {
+                        DataFusionError::Execution("Time index column downcast failed".into())
+                    })?;
+                if offset == 0 {
+                    Arc::new(column.clone())
+                } else {
+                    Arc::new(column.try_unary::<_, TimestampSecondType, _>(&bias_timestamp)?)
+                }
+            }
+            TimeUnit::Millisecond => {
+                let column = input
+                    .column(self.time_index)
+                    .as_any()
+                    .downcast_ref::<TimestampMillisecondArray>()
+                    .ok_or_else(|| {
+                        DataFusionError::Execution("Time index column downcast failed".into())
+                    })?;
+                if offset == 0 {
+                    Arc::new(column.clone())
+                } else {
+                    Arc::new(column.try_unary::<_, TimestampMillisecondType, _>(&bias_timestamp)?)
+                }
+            }
+            TimeUnit::Microsecond => {
+                let column = input
+                    .column(self.time_index)
+                    .as_any()
+                    .downcast_ref::<TimestampMicrosecondArray>()
+                    .ok_or_else(|| {
+                        DataFusionError::Execution("Time index column downcast failed".into())
+                    })?;
+                if offset == 0 {
+                    Arc::new(column.clone())
+                } else {
+                    Arc::new(column.try_unary::<_, TimestampMicrosecondType, _>(&bias_timestamp)?)
+                }
+            }
+            TimeUnit::Nanosecond => {
+                let column = input
+                    .column(self.time_index)
+                    .as_any()
+                    .downcast_ref::<TimestampNanosecondArray>()
+                    .ok_or_else(|| {
+                        DataFusionError::Execution("Time index column downcast failed".into())
+                    })?;
+                if offset == 0 {
+                    Arc::new(column.clone())
+                } else {
+                    Arc::new(column.try_unary::<_, TimestampNanosecondType, _>(&bias_timestamp)?)
+                }
+            }
         };
         let mut columns = input.columns().to_vec();
         columns[self.time_index] = ts_column_biased;
@@ -444,7 +504,11 @@ impl SeriesNormalizeStream {
                         if timestamp == 0 {
                             Ok(0)
                         } else {
-                            bias_timestamp(timestamp)
+                            timestamp.checked_add(self.offset).ok_or_else(|| {
+                                DataFusionError::Execution(
+                                    "SeriesNormalize: histogram timestamp offset overflow".into(),
+                                )
+                            })
                         }
                     })?;
                 // Replace only the start timestamp child to preserve the histogram payload and
@@ -720,64 +784,95 @@ mod test {
         regular.start_timestamp = Some(500);
         let mut ordinary_nan = native_histogram(f64::NAN);
         ordinary_nan.start_timestamp = Some(0);
+        let mut unknown_start = native_histogram(7.0);
+        unknown_start.start_timestamp = None;
         let histograms = build_histogram_array(&[
             Some(regular),
             Some(native_histogram(f64::from_bits(PROMETHEUS_STALE_NAN_BITS))),
             Some(ordinary_nan),
+            Some(unknown_start),
             None,
         ]);
-        let schema = Arc::new(Schema::new(vec![
-            Field::new(
-                TIME_INDEX_COLUMN,
-                TimestampMillisecondType::DATA_TYPE,
-                false,
-            ),
-            Field::new("value", histograms.data_type().clone(), true),
-        ]));
-        let batch = RecordBatch::try_new(
-            schema.clone(),
-            vec![
-                Arc::new(TimestampMillisecondArray::from(vec![
-                    1_000, 2_000, 3_000, 4_000,
-                ])),
-                histograms,
-            ],
-        )
-        .unwrap();
-        let input = Arc::new(DataSourceExec::new(Arc::new(
-            MemorySourceConfig::try_new(&[vec![batch]], schema, None).unwrap(),
-        )));
-        let exec = Arc::new(SeriesNormalizeExec {
-            offset: 1_000,
-            time_index_column_name: TIME_INDEX_COLUMN.to_string(),
-            filter_stale_markers: true,
-            tag_columns: Vec::new(),
-            input,
-            metric: ExecutionPlanMetricsSet::new(),
-        });
-
-        let context = SessionContext::default();
-        let batches = datafusion::physical_plan::collect(exec, context.task_ctx())
-            .await
-            .unwrap();
-        let batch = batches.iter().find(|batch| batch.num_rows() == 3).unwrap();
-        let values = batch
-            .column(1)
-            .as_any()
-            .downcast_ref::<datafusion::arrow::array::StructArray>()
-            .unwrap();
-
-        let timestamps = batch
-            .column(0)
-            .as_any()
-            .downcast_ref::<TimestampMillisecondArray>()
-            .unwrap();
-        assert_eq!(timestamps.values(), &[2_000, 4_000, 5_000]);
-        let regular = read_histogram(values, 0).unwrap().unwrap();
-        assert_eq!((regular.sum, regular.start_timestamp), (42.0, Some(1_500)));
-        let ordinary_nan = read_histogram(values, 1).unwrap().unwrap();
-        assert!(ordinary_nan.sum.is_nan());
-        assert_eq!(ordinary_nan.start_timestamp, Some(0));
-        assert!(read_histogram(values, 2).unwrap().is_none());
+        for (unit, ticks_per_ms) in [
+            (TimeUnit::Millisecond, 1_i64),
+            (TimeUnit::Microsecond, 1_000),
+            (TimeUnit::Nanosecond, 1_000_000),
+        ] {
+            let timestamp_array = |values: Vec<i64>| -> Arc<dyn Array> {
+                match unit {
+                    TimeUnit::Millisecond => Arc::new(TimestampMillisecondArray::from(values)),
+                    TimeUnit::Microsecond => Arc::new(TimestampMicrosecondArray::from(values)),
+                    TimeUnit::Nanosecond => Arc::new(TimestampNanosecondArray::from(values)),
+                    TimeUnit::Second => unreachable!(),
+                }
+            };
+            for offset in [-1_i64, 1] {
+                let timestamps = timestamp_array(
+                    [1_000, 2_000, 3_000, 4_000, 5_000]
+                        .into_iter()
+                        .map(|timestamp| timestamp * ticks_per_ms)
+                        .collect(),
+                );
+                let schema = Arc::new(Schema::new(vec![
+                    Field::new(TIME_INDEX_COLUMN, timestamps.data_type().clone(), false),
+                    Field::new("value", histograms.data_type().clone(), true),
+                ]));
+                let batch =
+                    RecordBatch::try_new(schema.clone(), vec![timestamps, histograms.clone()])
+                        .unwrap();
+                let input = Arc::new(DataSourceExec::new(Arc::new(
+                    MemorySourceConfig::try_new(&[vec![batch]], schema, None).unwrap(),
+                )));
+                let exec = Arc::new(SeriesNormalizeExec {
+                    offset,
+                    time_index_column_name: TIME_INDEX_COLUMN.to_string(),
+                    filter_stale_markers: true,
+                    tag_columns: Vec::new(),
+                    input,
+                    metric: ExecutionPlanMetricsSet::new(),
+                });
+                let context = SessionContext::default();
+                let batches = datafusion::physical_plan::collect(exec, context.task_ctx())
+                    .await
+                    .unwrap();
+                assert_eq!(
+                    batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
+                    4,
+                    "unit={unit:?}, offset={offset}"
+                );
+                let batch = batches.iter().find(|batch| batch.num_rows() == 4).unwrap();
+                let expected_timestamps = timestamp_array(
+                    [1_000, 3_000, 4_000, 5_000]
+                        .into_iter()
+                        .map(|timestamp| (timestamp + offset) * ticks_per_ms)
+                        .collect(),
+                );
+                assert_eq!(
+                    batch.column(0).to_data(),
+                    expected_timestamps.to_data(),
+                    "unit={unit:?}, offset={offset}"
+                );
+                let values = batch
+                    .column(1)
+                    .as_any()
+                    .downcast_ref::<datafusion::arrow::array::StructArray>()
+                    .unwrap();
+                let regular = read_histogram(values, 0).unwrap().unwrap();
+                assert_eq!(
+                    (regular.sum, regular.start_timestamp),
+                    (42.0, Some(500 + offset)),
+                    "unit={unit:?}, offset={offset}"
+                );
+                let ordinary_nan = read_histogram(values, 1).unwrap().unwrap();
+                assert!(ordinary_nan.sum.is_nan());
+                assert_eq!(ordinary_nan.start_timestamp, Some(0));
+                let unknown_start = read_histogram(values, 2).unwrap().unwrap();
+                assert_eq!(
+                    (unknown_start.sum, unknown_start.start_timestamp),
+                    (7.0, None)
+                );
+                assert!(read_histogram(values, 3).unwrap().is_none());
+            }
+        }
     }
 }

src/promql/src/extension_plan/range_manipulate.rs

@@ -21,7 +21,7 @@ use std::task::{Context, Poll};
 use common_telemetry::{debug, warn};
 use datafusion::arrow::array::{Array, ArrayRef, Int64Array, TimestampMillisecondArray};
 use datafusion::arrow::compute;
-use datafusion::arrow::datatypes::{Field, SchemaRef};
+use datafusion::arrow::datatypes::{DataType, Field, SchemaRef, TimeUnit};
 use datafusion::arrow::error::ArrowError;
 use datafusion::arrow::record_batch::RecordBatch;
 use datafusion::common::stats::Precision;
@@ -46,7 +46,8 @@ use snafu::ResultExt;
 
 use crate::error::{DeserializeSnafu, Result};
 use crate::extension_plan::{
-    METRIC_NUM_SERIES, Millisecond, resolve_column_name, serialize_column_index,
+    METRIC_NUM_SERIES, Millisecond, native_per_nanosecond, native_timestamp_values,
+    resolve_column_name, serialize_column_index, timestamp_unit,
 };
 use crate::metrics::PROMQL_SERIES_COUNT;
 use crate::range_array::RangeArray;
@@ -142,9 +143,21 @@ impl RangeManipulate {
             ));
         };
         let ts_col_field = &columns[ts_col_index];
+        let output_time_field = Arc::new(
+            ts_col_field
+                .as_ref()
+                .clone()
+                .with_data_type(DataType::Timestamp(TimeUnit::Millisecond, None)),
+        );
+        new_columns[ts_col_index] = (
+            input_schema.qualified_field(ts_col_index).0.cloned(),
+            output_time_field.clone(),
+        );
         let timestamp_range_field = Field::new(
             Self::build_timestamp_range_name(time_index),
-            RangeArray::convert_field(ts_col_field).data_type().clone(),
+            RangeArray::convert_field(output_time_field.as_ref())
+                .data_type()
+                .clone(),
             ts_col_field.is_nullable(),
         );
         new_columns.push((None, Arc::new(timestamp_range_field)));
@@ -515,6 +528,7 @@ impl ExecutionPlan for RangeManipulateExec {
                     .0
             })
             .collect();
+        let time_unit = timestamp_unit(schema.field(time_index).data_type())?;
         let aligned_ts_array =
             RangeManipulateStream::build_aligned_ts_array(self.start, self.end, self.interval);
         Ok(Box::pin(RangeManipulateStream {
@@ -523,6 +537,7 @@ impl ExecutionPlan for RangeManipulateExec {
             interval: self.interval,
             range: self.range,
             time_index,
+            time_unit,
             field_columns,
             aligned_ts_array,
             output_schema: self.output_schema.clone(),
@@ -584,6 +599,7 @@ pub struct RangeManipulateStream {
     interval: Millisecond,
     range: Millisecond,
     time_index: usize,
+    time_unit: TimeUnit,
     field_columns: Vec<usize>,
     aligned_ts_array: ArrayRef,
 
@@ -656,10 +672,13 @@ impl RangeManipulateStream {
         }
 
         // push timestamp range column
-        let ts_range_column =
-            RangeArray::from_ranges(input.column(self.time_index).clone(), ranges.clone())
-                .map_err(|e| ArrowError::InvalidArgumentError(e.to_string()))?
-                .into_dict();
+        let timestamp_values = compute::cast(
+            input.column(self.time_index),
+            &DataType::Timestamp(TimeUnit::Millisecond, None),
+        )?;
+        let ts_range_column = RangeArray::from_ranges(timestamp_values, ranges.clone())
+            .map_err(|e| ArrowError::InvalidArgumentError(e.to_string()))?
+            .into_dict();
         new_columns.push(Arc::new(ts_range_column));
 
         // truncate other columns
@@ -694,52 +713,58 @@ impl RangeManipulateStream {
         &self,
         input: &RecordBatch,
     ) -> DataFusionResult<(Vec<(u32, u32)>, (i64, i64))> {
-        let ts_column = input
-            .column(self.time_index)
-            .as_any()
-            .downcast_ref::<TimestampMillisecondArray>()
-            .ok_or_else(|| {
-                DataFusionError::Execution(
-                    "Time index Column downcast to TimestampMillisecondArray failed".into(),
-                )
-            })?;
-
-        let len = ts_column.len();
+        let ts_column = input.column(self.time_index);
+        let scale = native_per_nanosecond(self.time_unit);
+        let timestamps = native_timestamp_values(ts_column.as_ref())?;
+        let timestamp = |index| (timestamps[index] as i128) * scale;
+        let len = timestamps.len();
         if len == 0 {
             return Ok((vec![], (self.start, self.end)));
         }
 
-        // shorten the range to calculate
-        let first_ts = ts_column.value(0);
-        // Preserve the query's alignment pattern when optimizing start time
-        let remainder = (first_ts - self.start).rem_euclid(self.interval);
-        let first_ts_aligned = if remainder == 0 {
-            first_ts
-        } else {
-            first_ts + (self.interval - remainder)
-        };
-        let last_ts = ts_column.value(ts_column.len() - 1);
-        let last_ts_with_range = last_ts + self.range;
-        let remainder = (last_ts_with_range - self.start).rem_euclid(self.interval);
+        // Shorten the range using wide arithmetic so timestamps near the native
+        // type limits retain every query-aligned evaluation point.
+        let query_start = self.start as i128;
+        let query_end = self.end as i128;
+        let interval = self.interval as i128;
+        let first_ts = timestamp(0).div_euclid(1_000_000);
+        // Preserve the query's alignment pattern when optimizing start time.
+        let remainder = (first_ts - query_start).rem_euclid(interval);
+        let first_ts_aligned = first_ts + (interval - remainder).rem_euclid(interval);
+        let last_ts_with_range =
+            (timestamp(len - 1) + (self.range as i128) * 1_000_000).div_euclid(1_000_000);
+        let remainder = (last_ts_with_range - query_start).rem_euclid(interval);
         let last_ts_aligned = last_ts_with_range - remainder;
-        let start = self.start.max(first_ts_aligned);
-        let end = self.end.min(last_ts_aligned);
+        let start = query_start.max(first_ts_aligned);
+        let end = query_end.min(last_ts_aligned);
         if start > end {
-            return Ok((vec![], (start, end)));
+            let bounds = if start >= i64::MIN as i128
+                && start <= i64::MAX as i128
+                && end >= i64::MIN as i128
+                && end <= i64::MAX as i128
+            {
+                (start as i64, end as i64)
+            } else {
+                (self.start, self.end)
+            };
+            return Ok((vec![], bounds));
         }
-        let mut ranges = Vec::with_capacity(((self.end - self.start) / self.interval + 1) as usize);
+        // The intersection is within the declared i64 query bounds.
+        let start = start as i64;
+        let end = end as i64;
+        let mut ranges = Vec::new();
 
         // calculate for every aligned timestamp (`curr_ts`), assume the ts column is ordered.
         let mut left = 0usize;
         let mut right = 0usize;
         for curr_ts in (start..=end).step_by(self.interval as _) {
-            let start_ts = curr_ts - self.range;
+            let start_ts = (curr_ts as i128) * 1_000_000 - (self.range as i128) * 1_000_000;
 
-            while left < len && ts_column.value(left) <= start_ts {
+            while left < len && timestamp(left) <= start_ts {
                 left += 1;
             }
             right = right.max(left);
-            while right < len && ts_column.value(right) <= curr_ts {
+            while right < len && timestamp(right) <= (curr_ts as i128) * 1_000_000 {
                 right += 1;
             }
 
@@ -756,7 +781,10 @@ impl RangeManipulateStream {
 
 #[cfg(test)]
 mod test {
-    use datafusion::arrow::array::{ArrayRef, DictionaryArray, Float64Array, StringArray};
+    use datafusion::arrow::array::{
+        ArrayRef, DictionaryArray, Float64Array, StringArray, TimestampMicrosecondArray,
+        TimestampNanosecondArray,
+    };
     use datafusion::arrow::datatypes::{
         ArrowPrimitiveType, DataType, Field, Int64Type, Schema, TimestampMillisecondType,
     };
@@ -888,6 +916,130 @@ mod test {
         assert_eq!(result_literal, expected);
     }
 
+    #[tokio::test]
+    async fn native_timestamps_preserve_range_membership_and_ms_payload() {
+        for (unit, ticks_per_ms) in [
+            (TimeUnit::Microsecond, 1_000_i64),
+            (TimeUnit::Nanosecond, 1_000_000_i64),
+        ] {
+            let lower = 1_000 * ticks_per_ms;
+            let upper = 1_001 * ticks_per_ms;
+            // Exclude the lower boundary and future sample; retain both native
+            // samples in the same millisecond bucket and the exact upper sample.
+            let timestamps = vec![lower, lower + 1, lower + 2, upper, upper + 1];
+            let time: ArrayRef = match unit {
+                TimeUnit::Microsecond => Arc::new(TimestampMicrosecondArray::from(timestamps)),
+                TimeUnit::Nanosecond => Arc::new(TimestampNanosecondArray::from(timestamps)),
+                _ => unreachable!(),
+            };
+            let schema = Arc::new(Schema::new(vec![
+                Field::new(TIME_INDEX_COLUMN, DataType::Timestamp(unit, None), false),
+                Field::new("value", DataType::Float64, true),
+            ]));
+            let batch = RecordBatch::try_new(
+                schema.clone(),
+                vec![
+                    time,
+                    Arc::new(Float64Array::from(vec![10.0, 20.0, 30.0, 40.0, 50.0])),
+                ],
+            )
+            .unwrap();
+            let logical_input = LogicalPlan::EmptyRelation(EmptyRelation {
+                produce_one_row: false,
+                schema: schema.clone().to_dfschema_ref().unwrap(),
+            });
+            let plan = RangeManipulate::new(
+                1_001,
+                1_001,
+                1,
+                1,
+                TIME_INDEX_COLUMN.to_string(),
+                vec!["value".to_string()],
+                logical_input.clone(),
+            )
+            .unwrap();
+            let output_time = Field::new(
+                TIME_INDEX_COLUMN,
+                DataType::Timestamp(TimeUnit::Millisecond, None),
+                false,
+            );
+            let output_schema = Arc::new(Schema::new(vec![
+                output_time.clone(),
+                RangeArray::convert_field(&Field::new("value", DataType::Float64, true)),
+                Field::new(
+                    RangeManipulate::build_timestamp_range_name(TIME_INDEX_COLUMN),
+                    RangeArray::convert_field(&output_time).data_type().clone(),
+                    false,
+                ),
+            ]));
+            assert_eq!(plan.schema().as_arrow(), output_schema.as_ref());
+
+            let rebuilt = RangeManipulate::deserialize(&plan.serialize())
+                .unwrap()
+                .with_exprs_and_inputs(vec![], vec![logical_input])
+                .unwrap();
+            assert_eq!(rebuilt.schema(), plan.schema());
+            assert_eq!(rebuilt.input.schema().as_arrow(), schema.as_ref());
+
+            let input = Arc::new(DataSourceExec::new(Arc::new(
+                MemorySourceConfig::try_new(&[vec![batch]], schema.clone(), None).unwrap(),
+            )));
+            let exec = rebuilt.to_execution_plan(input);
+            assert_eq!(exec.schema(), output_schema);
+            assert_eq!(exec.children()[0].schema(), schema);
+
+            let batches =
+                datafusion::physical_plan::collect(exec, SessionContext::default().task_ctx())
+                    .await
+                    .unwrap();
+            assert_eq!(batches.len(), 1, "{unit:?}");
+            let output = &batches[0];
+            assert_eq!(output.schema(), output_schema);
+            assert_eq!(output.num_rows(), 1);
+            assert_eq!(
+                output
+                    .column(0)
+                    .as_any()
+                    .downcast_ref::<TimestampMillisecondArray>()
+                    .unwrap()
+                    .values()
+                    .as_ref(),
+                &[1_001]
+            );
+
+            // RangeArray packs offset/length into dictionary keys; Arrow dictionary
+            // equality treats those packed keys as indices and cannot compare them.
+            let values = RangeArray::try_new(
+                output
+                    .column(1)
+                    .as_any()
+                    .downcast_ref::<DictionaryArray<Int64Type>>()
+                    .unwrap()
+                    .clone(),
+            )
+            .unwrap();
+            assert_eq!(values.get_offset_length(0), Some((1, 3)));
+            assert_eq!(
+                values.get(0).unwrap().to_data(),
+                Float64Array::from(vec![20.0, 30.0, 40.0]).to_data()
+            );
+            let timestamps = RangeArray::try_new(
+                output
+                    .column(2)
+                    .as_any()
+                    .downcast_ref::<DictionaryArray<Int64Type>>()
+                    .unwrap()
+                    .clone(),
+            )
+            .unwrap();
+            assert_eq!(timestamps.get_offset_length(0), Some((1, 3)));
+            assert_eq!(
+                timestamps.get(0).unwrap().to_data(),
+                TimestampMillisecondArray::from(vec![1_000, 1_000, 1_001]).to_data()
+            );
+        }
+    }
+
     #[tokio::test]
     async fn pruning_should_keep_time_and_value_columns_for_exec() {
         let schema = Arc::new(Schema::new(vec![
@@ -1047,6 +1199,7 @@ mod test {
             interval: 30000,      // 30s step
             range: 60000,         // 60s lookback
             time_index: 0,
+            time_unit: TimeUnit::Millisecond,
             field_columns: vec![],
             aligned_ts_array: Arc::new(TimestampMillisecondArray::from(vec![0i64; 0])),
             output_schema: schema.clone(),
@@ -1111,6 +1264,7 @@ mod test {
             interval,
             range,
             time_index: 0,
+            time_unit: TimeUnit::Millisecond,
             field_columns: vec![],
             aligned_ts_array: Arc::new(TimestampMillisecondArray::from(vec![0i64; 0])),
             output_schema: schema.clone(),
@@ -1227,6 +1381,15 @@ mod test {
         }
     }
 
+    #[test]
+    fn calculate_range_keeps_extreme_range_tail() {
+        let (ranges, bounds) =
+            calculate_range_for_test(i64::MAX - 1, i64::MAX, 1, i64::MAX, &[i64::MAX]);
+
+        assert_eq!(bounds, (i64::MAX, i64::MAX));
+        assert_eq!(ranges, vec![(0, 1)]);
+    }
+
     #[test]
     fn calculate_range_matches_bruteforce_oracle_for_deterministic_cases() {
         let cases = vec![

src/query/src/promql/planner.rs

@@ -1974,8 +1974,32 @@ impl PromPlanner {
                     DfExpr::Column(Column::new(qualifier.cloned(), field.name().clone()))
                 })
                 .collect::<Vec<_>>();
-            project_exprs
-                .push(build_special_time_expr(&time_index_column).alias(&timestamp_value_column));
+            // `timestamp()` carries the sample timestamp as a value. Cast native
+            // input here because the helper consumes millisecond ticks.
+            let sample_time = col(&time_index_column);
+            let sample_time = if sample_time
+                .get_type(normalize.schema())
+                .context(DataFusionPlanningSnafu)?
+                == ArrowDataType::Timestamp(ArrowTimeUnit::Millisecond, None)
+            {
+                sample_time
+            } else {
+                DfExpr::Cast(Cast {
+                    expr: Box::new(sample_time),
+                    data_type: ArrowDataType::Timestamp(ArrowTimeUnit::Millisecond, None),
+                })
+            };
+            let sample_time = sample_time
+                .cast_to(&ArrowDataType::Int64, normalize.schema())
+                .context(DataFusionPlanningSnafu)?
+                .cast_to(&ArrowDataType::Float64, normalize.schema())
+                .context(DataFusionPlanningSnafu)?;
+            let sample_time = DfExpr::BinaryExpr(BinaryExpr {
+                left: Box::new(sample_time),
+                op: Operator::Divide,
+                right: Box::new(lit(1000.0)),
+            });
+            project_exprs.push(sample_time.alias(&timestamp_value_column));
             let normalize = LogicalPlanBuilder::from(normalize)
                 .project(project_exprs)
                 .context(DataFusionPlanningSnafu)?
@@ -2339,14 +2363,18 @@ impl PromPlanner {
             None => 0,
         };
         let mut scan_filters = Self::matchers_to_expr(label_matchers.clone(), table_schema)?;
-        if let Some(time_index_filter) = self.build_time_index_filter(offset_duration)? {
+        if let Some(time_index_filter) =
+            self.build_time_index_filter(offset_duration, table_schema)?
+        {
             scan_filters.push(time_index_filter);
         }
-        table_scan = LogicalPlanBuilder::from(table_scan)
-            .filter(conjunction(scan_filters).unwrap()) // Safety: `scan_filters` is not empty.
-            .context(DataFusionPlanningSnafu)?
-            .build()
-            .context(DataFusionPlanningSnafu)?;
+        if let Some(filter) = conjunction(scan_filters) {
+            table_scan = LogicalPlanBuilder::from(table_scan)
+                .filter(filter)
+                .context(DataFusionPlanningSnafu)?
+                .build()
+                .context(DataFusionPlanningSnafu)?;
+        }
 
         // make a projection plan if there is any `__field__` matcher
         if let Some(field_matchers) = &self.ctx.field_column_matcher {
@@ -2718,72 +2746,83 @@ impl PromPlanner {
         Ok(table_ref)
     }
 
-    fn build_time_index_filter(&self, offset_duration: i64) -> Result<Option<DfExpr>> {
+    fn build_time_index_filter(
+        &self,
+        offset_duration: i64,
+        schema: &DFSchemaRef,
+    ) -> Result<Option<DfExpr>> {
         let start = self.ctx.start;
         let end = self.ctx.end;
         if end < start {
             return InvalidTimeRangeSnafu { start, end }.fail();
         }
-        let lookback_delta = self.ctx.lookback_delta;
-        let range = self.ctx.range.unwrap_or_default();
-        let interval = self.ctx.interval;
         let time_index_expr = self.create_time_index_column_expr()?;
-        let num_points = (end - start) / interval;
-
-        // Prometheus semantics:
-        // - Instant selector lookback: (eval_ts - lookback_delta, eval_ts]
-        // - Range selector:           (eval_ts - range, eval_ts]
-        //
-        // So samples positioned exactly at the lower boundary must be excluded. We align the scan
-        // lower bound with Prometheus by shifting it forward by 1ms (millisecond granularity),
-        // while still using a `>=` filter.
-        let selector_window = if range == 0 { lookback_delta } else { range };
-        let lower_exclusive_adjustment = if selector_window > 0 { 1 } else { 0 };
-
-        // Scan a continuous time range
-        if (end - start) / interval > MAX_SCATTER_POINTS || interval <= INTERVAL_1H {
-            let single_time_range = time_index_expr
-                .clone()
-                .gt_eq(DfExpr::Literal(
-                    ScalarValue::TimestampMillisecond(
-                        Some(
-                            self.ctx.start - offset_duration - selector_window
-                                + lower_exclusive_adjustment,
-                        ),
-                        None,
-                    ),
-                    None,
-                ))
-                .and(time_index_expr.lt_eq(DfExpr::Literal(
-                    ScalarValue::TimestampMillisecond(Some(self.ctx.end - offset_duration), None),
-                    None,
-                )));
-            return Ok(Some(single_time_range));
-        }
-
-        // Otherwise scan scatter ranges separately
-        let mut filters = Vec::with_capacity(num_points as usize + 1);
-        for timestamp in (start..=end).step_by(interval as usize) {
-            filters.push(
+        let time_index_name = self.ctx.time_index_column.as_ref().unwrap();
+        let unit = schema
+            .index_of_column_by_name(None, time_index_name)
+            .and_then(|index| match schema.field(index).data_type() {
+                ArrowDataType::Timestamp(unit, _) => Some(*unit),
+                _ => None,
+            })
+            .unwrap_or(ArrowTimeUnit::Millisecond);
+        let scalar = |milliseconds: i64| -> Option<ScalarValue> {
+            let value = match unit {
+                ArrowTimeUnit::Second => milliseconds.div_euclid(1_000),
+                ArrowTimeUnit::Millisecond => milliseconds,
+                ArrowTimeUnit::Microsecond => milliseconds.checked_mul(1_000)?,
+                ArrowTimeUnit::Nanosecond => milliseconds.checked_mul(1_000_000)?,
+            };
+            Some(match unit {
+                ArrowTimeUnit::Second => ScalarValue::TimestampSecond(Some(value), None),
+                ArrowTimeUnit::Millisecond => ScalarValue::TimestampMillisecond(Some(value), None),
+                ArrowTimeUnit::Microsecond => ScalarValue::TimestampMicrosecond(Some(value), None),
+                ArrowTimeUnit::Nanosecond => ScalarValue::TimestampNanosecond(Some(value), None),
+            })
+        };
+        let window = self.ctx.range.unwrap_or(self.ctx.lookback_delta);
+        let filter = |lower_ms: i64, upper_ms: i64| -> Option<DfExpr> {
+            let lower = DfExpr::Literal(scalar(lower_ms)?, None);
+            let lower_filter = if window == 0 {
+                time_index_expr.clone().gt_eq(lower)
+            } else if unit == ArrowTimeUnit::Millisecond
+                && let Some(inclusive_lower) = lower_ms.checked_add(1)
+            {
                 time_index_expr
                     .clone()
-                    .gt_eq(DfExpr::Literal(
-                        ScalarValue::TimestampMillisecond(
-                            Some(
-                                timestamp - offset_duration - selector_window
-                                    + lower_exclusive_adjustment,
-                            ),
-                            None,
-                        ),
-                        None,
-                    ))
-                    .and(time_index_expr.clone().lt_eq(DfExpr::Literal(
-                        ScalarValue::TimestampMillisecond(Some(timestamp - offset_duration), None),
-                        None,
-                    ))),
+                    .gt_eq(DfExpr::Literal(scalar(inclusive_lower)?, None))
+            } else {
+                time_index_expr.clone().gt(lower)
+            };
+            Some(
+                lower_filter.and(
+                    time_index_expr
+                        .clone()
+                        .lt_eq(DfExpr::Literal(scalar(upper_ms)?, None)),
+                ),
             )
+        };
+        let bounds = |timestamp: i64| {
+            timestamp
+                .checked_sub(offset_duration)
+                .and_then(|upper| upper.checked_sub(window).map(|lower| (lower, upper)))
+        };
+        let num_points = (end as i128 - start as i128) / self.ctx.interval as i128;
+        if num_points > MAX_SCATTER_POINTS as i128 || self.ctx.interval <= INTERVAL_1H {
+            return Ok(bounds(start)
+                .zip(bounds(end))
+                .and_then(|((lower, _), (_, upper))| filter(lower, upper)));
+        }
+        let mut filters = Vec::new();
+        for timestamp in (start..=end).step_by(self.ctx.interval as usize) {
+            let Some((lower, upper)) = bounds(timestamp) else {
+                // An unrepresentable envelope must not discard samples.
+                return Ok(None);
+            };
+            let Some(filter) = filter(lower, upper) else {
+                return Ok(None);
+            };
+            filters.push(filter);
         }
-
         Ok(filters.into_iter().reduce(DfExpr::or))
     }
 
@@ -2884,14 +2923,14 @@ impl PromPlanner {
             self.ctx.tag_columns.clone()
         };
 
-        let is_time_index_ms = scan_table
+        let is_time_index_second = scan_table
             .schema()
             .timestamp_column()
             .with_context(|| TimeIndexNotFoundSnafu {
                 table: maybe_phy_table_ref.to_quoted_string(),
             })?
             .data_type
-            == ConcreteDataType::timestamp_millisecond_datatype();
+            == ConcreteDataType::timestamp_second_datatype();
 
         let scan_projection = if table_id_filter.is_some() {
             let mut required_columns = HashSet::new();
@@ -2944,8 +2983,8 @@ impl PromPlanner {
                 .context(DataFusionPlanningSnafu)?;
         }
 
-        if !is_time_index_ms {
-            // cast to ms if time_index not in Millisecond precision
+        if is_time_index_second {
+            // Promote seconds so millisecond offsets remain exact; retain finer precision.
             let expr: Vec<_> = self
                 .create_field_column_exprs()?
                 .into_iter()
@@ -9051,7 +9090,17 @@ mod test {
 
         let manipulate = find_instant_manipulate(&plan).unwrap();
         let exec = manipulate.to_execution_plan(Arc::new(DataSourceExec::new(Arc::new(
-            MemorySourceConfig::try_new(&[], Arc::new(ArrowSchema::empty()), None).unwrap(),
+            MemorySourceConfig::try_new(
+                &[],
+                Arc::new(
+                    datafusion_expr::UserDefinedLogicalNodeCore::inputs(manipulate)[0]
+                        .schema()
+                        .as_arrow()
+                        .clone(),
+                ),
+                None,
+            )
+            .unwrap(),
         ))));
         assert!(format!("{exec:?}").contains("reuse_tsid_column: true"));
     }
@@ -12135,6 +12184,57 @@ mod test {
         }
     }
 
+    #[tokio::test]
+    async fn native_scan_bounds_preserve_zero_lookback_and_overflow() {
+        let table_provider = build_test_table_provider(
+            &[(DEFAULT_SCHEMA_NAME.to_string(), "some_metric".to_string())],
+            1,
+            1,
+        )
+        .await;
+        let mut planner = PromPlanner {
+            table_provider,
+            ctx: PromPlannerContext::from_eval_stmt(&build_eval_stmt("some_metric")),
+            promql_annotations: None,
+        };
+        planner.ctx.time_index_column = Some("timestamp".to_string());
+        planner.ctx.start = 1_000;
+        planner.ctx.lookback_delta = 0;
+        let schema = Arc::new(
+            DFSchema::try_from(ArrowSchema::new(vec![Field::new(
+                "timestamp",
+                ArrowDataType::Timestamp(ArrowTimeUnit::Nanosecond, None),
+                false,
+            )]))
+            .unwrap(),
+        );
+        for (end, interval, windows) in [
+            (1_000, 1_000, 1),
+            (2_000, 1_000, 1),
+            (7_201_000, 7_200_000, 2),
+        ] {
+            planner.ctx.end = end;
+            planner.ctx.interval = interval;
+            let filter = planner
+                .build_time_index_filter(0, &schema)
+                .unwrap()
+                .unwrap()
+                .to_string();
+            assert_eq!(filter.matches(">=").count(), windows, "{filter}");
+            assert!(
+                filter.contains("TimestampNanosecond(1000000000, None)"),
+                "{filter}"
+            );
+        }
+        planner.ctx.end = i64::MAX;
+        assert!(
+            planner
+                .build_time_index_filter(0, &schema)
+                .unwrap()
+                .is_none()
+        );
+    }
+
     #[tokio::test]
     async fn test_non_ms_precision() {
         let catalog_list = MemoryCatalogManager::with_default_setup();
@@ -12205,12 +12305,7 @@ mod test {
         .unwrap();
         assert_eq!(
             plan.display_indent_schema().to_string(),
-            "PromInstantManipulate: range=[0..100000000], lookback=[1000], interval=[5000], time index=[timestamp] [field:Float64;N, tag:Utf8, timestamp:Timestamp(ms)]\
-            \n  PromSeriesDivide: tags=[\"tag\"] [field:Float64;N, tag:Utf8, timestamp:Timestamp(ms)]\
-            \n    Sort: metrics.tag ASC NULLS FIRST, metrics.timestamp ASC NULLS FIRST [field:Float64;N, tag:Utf8, timestamp:Timestamp(ms)]\
-            \n      Filter: metrics.tag = Utf8(\"1\") AND metrics.timestamp >= TimestampMillisecond(-999, None) AND metrics.timestamp <= TimestampMillisecond(100000000, None) [field:Float64;N, tag:Utf8, timestamp:Timestamp(ms)]\
-            \n        Projection: metrics.field, metrics.tag, CAST(metrics.timestamp AS Timestamp(ms)) AS timestamp [field:Float64;N, tag:Utf8, timestamp:Timestamp(ms)]\
-            \n          TableScan: metrics [tag:Utf8, timestamp:Timestamp(ns), field:Float64;N]"
+            "PromInstantManipulate: range=[0..100000000], lookback=[1000], interval=[5000], time index=[timestamp] [tag:Utf8, timestamp:Timestamp(ms), field:Float64;N]\n  PromSeriesDivide: tags=[\"tag\"] [tag:Utf8, timestamp:Timestamp(ns), field:Float64;N]\n    Sort: metrics.tag ASC NULLS FIRST, metrics.timestamp ASC NULLS FIRST [tag:Utf8, timestamp:Timestamp(ns), field:Float64;N]\n      Filter: metrics.tag = Utf8(\"1\") AND metrics.timestamp > TimestampNanosecond(-1000000000, None) AND metrics.timestamp <= TimestampNanosecond(100000000000000, None) [tag:Utf8, timestamp:Timestamp(ns), field:Float64;N]\n        TableScan: metrics [tag:Utf8, timestamp:Timestamp(ns), field:Float64;N]"
         );
         let plan = PromPlanner::stmt_to_plan(
             DfTableSourceProvider::new(
@@ -12235,15 +12330,7 @@ mod test {
         .unwrap();
         assert_eq!(
             plan.display_indent_schema().to_string(),
-            "Filter: prom_avg_over_time(timestamp_range,field) IS NOT NULL [timestamp:Timestamp(ms), prom_avg_over_time(timestamp_range,field):Float64;N, tag:Utf8]\
-            \n  Projection: metrics.timestamp, prom_avg_over_time(timestamp_range, field) AS prom_avg_over_time(timestamp_range,field), metrics.tag [timestamp:Timestamp(ms), prom_avg_over_time(timestamp_range,field):Float64;N, tag:Utf8]\
-            \n    PromRangeManipulate: req range=[0..100000000], interval=[5000], eval range=[5000], time index=[timestamp], values=[\"field\"] [field:Dictionary(Int64, Float64);N, tag:Utf8, timestamp:Timestamp(ms), timestamp_range:Dictionary(Int64, Timestamp(ms))]\
-            \n      PromSeriesNormalize: offset=[0], time index=[timestamp], filter NaN: [true] [field:Float64;N, tag:Utf8, timestamp:Timestamp(ms)]\
-            \n        PromSeriesDivide: tags=[\"tag\"] [field:Float64;N, tag:Utf8, timestamp:Timestamp(ms)]\
-            \n          Sort: metrics.tag ASC NULLS FIRST, metrics.timestamp ASC NULLS FIRST [field:Float64;N, tag:Utf8, timestamp:Timestamp(ms)]\
-            \n            Filter: metrics.tag = Utf8(\"1\") AND metrics.timestamp >= TimestampMillisecond(-4999, None) AND metrics.timestamp <= TimestampMillisecond(100000000, None) [field:Float64;N, tag:Utf8, timestamp:Timestamp(ms)]\
-            \n              Projection: metrics.field, metrics.tag, CAST(metrics.timestamp AS Timestamp(ms)) AS timestamp [field:Float64;N, tag:Utf8, timestamp:Timestamp(ms)]\
-            \n                TableScan: metrics [tag:Utf8, timestamp:Timestamp(ns), field:Float64;N]"
+            "Filter: prom_avg_over_time(timestamp_range,field) IS NOT NULL [timestamp:Timestamp(ms), prom_avg_over_time(timestamp_range,field):Float64;N, tag:Utf8]\n  Projection: metrics.timestamp, prom_avg_over_time(timestamp_range, field) AS prom_avg_over_time(timestamp_range,field), metrics.tag [timestamp:Timestamp(ms), prom_avg_over_time(timestamp_range,field):Float64;N, tag:Utf8]\n    PromRangeManipulate: req range=[0..100000000], interval=[5000], eval range=[5000], time index=[timestamp], values=[\"field\"] [tag:Utf8, timestamp:Timestamp(ms), field:Dictionary(Int64, Float64);N, timestamp_range:Dictionary(Int64, Timestamp(ms))]\n      PromSeriesNormalize: offset=[0], time index=[timestamp], filter NaN: [true] [tag:Utf8, timestamp:Timestamp(ns), field:Float64;N]\n        PromSeriesDivide: tags=[\"tag\"] [tag:Utf8, timestamp:Timestamp(ns), field:Float64;N]\n          Sort: metrics.tag ASC NULLS FIRST, metrics.timestamp ASC NULLS FIRST [tag:Utf8, timestamp:Timestamp(ns), field:Float64;N]\n            Filter: metrics.tag = Utf8(\"1\") AND metrics.timestamp > TimestampNanosecond(-5000000000, None) AND metrics.timestamp <= TimestampNanosecond(100000000000000, None) [tag:Utf8, timestamp:Timestamp(ns), field:Float64;N]\n              TableScan: metrics [tag:Utf8, timestamp:Timestamp(ns), field:Float64;N]"
         );
     }
 

tests/cases/standalone/common/promql/native_time_selection.result

@@ -0,0 +1,361 @@
+-- Regression coverage for instant and range selection on native microsecond and
+-- nanosecond time indexes.
+CREATE TABLE native_time_us (
+    ts TIMESTAMP(6) TIME INDEX,
+    series STRING PRIMARY KEY,
+    val DOUBLE,
+);
+
+Affected Rows: 0
+
+INSERT INTO native_time_us VALUES
+    (1000001, 'future', 101),
+    (1000000, 'exact', 201),
+    (1000001, 'exact', 202),
+    (-299000000, 'lowerbound', 301),
+    (-298999999, 'lowerplus', 302),
+    (1000000, 'positive_lowerbound', 701),
+    (1000001, 'positive_lowerplus', 702),
+    (1000001, 'multi', 401),
+    (-1000000, 'offset', 501),
+    (0, 'offset', 502),
+    (1000000, 'offset', 503),
+    (999999, 'past', 602),
+    (999001, 'past', 601),
+    (0, 'window', 1),
+    (1, 'window', 2),
+    (2, 'window', 5),
+    (1000000, 'window', 3),
+    (1000001, 'window', 4);
+
+Affected Rows: 18
+
+-- Future-only selection is empty before flushing, exercising the memtable path.
+TQL EVAL (1, 1, '1s', '300s') native_time_us{series="future"};
+
+++
+++
+
+ADMIN FLUSH_TABLE('native_time_us');
+
++-------------------------------------+
+| ADMIN FLUSH_TABLE('native_time_us') |
++-------------------------------------+
+| 0                                   |
++-------------------------------------+
+
+-- At 1s, selection keeps an exact native timestamp.
+TQL EVAL (1, 1, '1s', '300s') native_time_us{series="exact"};
+
++---------------------+--------+-------+
+| ts                  | series | val   |
++---------------------+--------+-------+
+| 1970-01-01T00:00:01 | exact  | 201.0 |
++---------------------+--------+-------+
+
+TQL EVAL (1, 1, '1s', '300s') timestamp(native_time_us{series="future"});
+
+++
+++
+
+TQL EVAL (1, 1, '1s', '300s') timestamp(native_time_us{series="exact"});
+
++---------------------+-------+--------+
+| ts                  | value | series |
++---------------------+-------+--------+
+| 1970-01-01T00:00:01 | 1.0   | exact  |
++---------------------+-------+--------+
+
+-- Instant lookback bounds are exclusive: these return only 302 and 702.
+TQL EVAL (1, 1, '1s', '300s') native_time_us{series=~"lower.*"};
+
++---------------------+-----------+-------+
+| ts                  | series    | val   |
++---------------------+-----------+-------+
+| 1970-01-01T00:00:01 | lowerplus | 302.0 |
++---------------------+-----------+-------+
+
+TQL EVAL (301, 301, '1s', '300s') native_time_us{series=~"positive_lower.*"};
+
++---------------------+--------------------+-------+
+| ts                  | series             | val   |
++---------------------+--------------------+-------+
+| 1970-01-01T00:05:01 | positive_lowerplus | 702.0 |
++---------------------+--------------------+-------+
+
+-- The sub-millisecond point belongs only to the 2s evaluation step.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1, 2, '1s', '300s') native_time_us{series="multi"};
+
++---------------------+--------+-------+
+| ts                  | series | val   |
++---------------------+--------+-------+
+| 1970-01-01T00:00:02 | multi  | 401.0 |
++---------------------+--------+-------+
+
+-- The latest native timestamp below 1s is retained even when inserts are unordered.
+TQL EVAL (1, 1, '1s', '300s') native_time_us{series="past"};
+
++---------------------+--------+-------+
+| ts                  | series | val   |
++---------------------+--------+-------+
+| 1970-01-01T00:00:01 | past   | 602.0 |
++---------------------+--------+-------+
+
+-- Offsets select native timestamps, including stored negative time.
+TQL EVAL (0, 0, '1s', '300s') native_time_us{series="offset"};
+
++---------------------+--------+-------+
+| ts                  | series | val   |
++---------------------+--------+-------+
+| 1970-01-01T00:00:00 | offset | 502.0 |
++---------------------+--------+-------+
+
+TQL EVAL (0, 0, '1s', '300s') native_time_us{series="offset"} offset 1s;
+
++---------------------+--------+-------+
+| ts                  | series | val   |
++---------------------+--------+-------+
+| 1970-01-01T00:00:00 | offset | 501.0 |
++---------------------+--------+-------+
+
+TQL EVAL (0, 0, '1s', '300s') native_time_us{series="offset"} offset -1s;
+
++---------------------+--------+-------+
+| ts                  | series | val   |
++---------------------+--------+-------+
+| 1970-01-01T00:00:00 | offset | 503.0 |
++---------------------+--------+-------+
+
+-- [1s] at 1s excludes 0 and 1s+tick, retaining 0+tick, 0+2ticks, and 1s.
+TQL EVAL (1, 1, '1s', '300s') count_over_time(native_time_us{series="window"}[1s]);
+
++---------------------+------------------------------------+--------+
+| ts                  | prom_count_over_time(ts_range,val) | series |
++---------------------+------------------------------------+--------+
+| 1970-01-01T00:00:01 | 3.0                                | window |
++---------------------+------------------------------------+--------+
+
+TQL EVAL (1, 1, '1s', '300s') sum_over_time(native_time_us{series="window"}[1s]);
+
++---------------------+----------------------------------+--------+
+| ts                  | prom_sum_over_time(ts_range,val) | series |
++---------------------+----------------------------------+--------+
+| 1970-01-01T00:00:01 | 10.0                             | window |
++---------------------+----------------------------------+--------+
+
+TQL EVAL (1, 1, '1s', '300s') last_over_time(native_time_us{series="window"}[1s]);
+
++---------------------+-----------------------------------+--------+
+| ts                  | prom_last_over_time(ts_range,val) | series |
++---------------------+-----------------------------------+--------+
+| 1970-01-01T00:00:01 | 3.0                               | window |
++---------------------+-----------------------------------+--------+
+
+-- The inner selector consumes native time; the subquery consumes ms evaluations.
+TQL EVAL (1, 1, '1s') last_over_time((native_time_us{series="exact"})[1s:1s]);
+
++---------------------+-----------------------------------+--------+
+| ts                  | prom_last_over_time(ts_range,val) | series |
++---------------------+-----------------------------------+--------+
+| 1970-01-01T00:00:01 | 201.0                             | exact  |
++---------------------+-----------------------------------+--------+
+
+DROP TABLE native_time_us;
+
+Affected Rows: 0
+
+CREATE TABLE native_time_ns (
+    ts TIMESTAMP(9) TIME INDEX,
+    series STRING PRIMARY KEY,
+    val DOUBLE,
+);
+
+Affected Rows: 0
+
+INSERT INTO native_time_ns VALUES
+    (1000000001, 'future', 101),
+    (1000000000, 'exact', 201),
+    (1000000001, 'exact', 202),
+    (-299000000000, 'lowerbound', 301),
+    (-298999999999, 'lowerplus', 302),
+    (1000000000, 'positive_lowerbound', 701),
+    (1000000001, 'positive_lowerplus', 702),
+    (1000000001, 'multi', 401),
+    (-1000000000, 'offset', 501),
+    (0, 'offset', 502),
+    (1000000000, 'offset', 503),
+    (999999000, 'past', 602),
+    (999001000, 'past', 601),
+    (0, 'window', 1),
+    (1, 'window', 2),
+    (2, 'window', 5),
+    (1000000000, 'window', 3),
+    (1000000001, 'window', 4);
+
+Affected Rows: 18
+
+-- Future-only selection is empty before flushing, exercising the memtable path.
+TQL EVAL (1, 1, '1s', '300s') native_time_ns{series="future"};
+
+++
+++
+
+ADMIN FLUSH_TABLE('native_time_ns');
+
++-------------------------------------+
+| ADMIN FLUSH_TABLE('native_time_ns') |
++-------------------------------------+
+| 0                                   |
++-------------------------------------+
+
+-- At 1s, selection keeps an exact native timestamp.
+TQL EVAL (1, 1, '1s', '300s') native_time_ns{series="exact"};
+
++---------------------+--------+-------+
+| ts                  | series | val   |
++---------------------+--------+-------+
+| 1970-01-01T00:00:01 | exact  | 201.0 |
++---------------------+--------+-------+
+
+TQL EVAL (1, 1, '1s', '300s') timestamp(native_time_ns{series="future"});
+
+++
+++
+
+TQL EVAL (1, 1, '1s', '300s') timestamp(native_time_ns{series="exact"});
+
++---------------------+-------+--------+
+| ts                  | value | series |
++---------------------+-------+--------+
+| 1970-01-01T00:00:01 | 1.0   | exact  |
++---------------------+-------+--------+
+
+-- Instant lookback bounds are exclusive: these return only 302 and 702.
+TQL EVAL (1, 1, '1s', '300s') native_time_ns{series=~"lower.*"};
+
++---------------------+-----------+-------+
+| ts                  | series    | val   |
++---------------------+-----------+-------+
+| 1970-01-01T00:00:01 | lowerplus | 302.0 |
++---------------------+-----------+-------+
+
+TQL EVAL (301, 301, '1s', '300s') native_time_ns{series=~"positive_lower.*"};
+
++---------------------+--------------------+-------+
+| ts                  | series             | val   |
++---------------------+--------------------+-------+
+| 1970-01-01T00:05:01 | positive_lowerplus | 702.0 |
++---------------------+--------------------+-------+
+
+-- The sub-millisecond point belongs only to the 2s evaluation step.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1, 2, '1s', '300s') native_time_ns{series="multi"};
+
++---------------------+--------+-------+
+| ts                  | series | val   |
++---------------------+--------+-------+
+| 1970-01-01T00:00:02 | multi  | 401.0 |
++---------------------+--------+-------+
+
+-- The latest native timestamp below 1s is retained even when inserts are unordered.
+TQL EVAL (1, 1, '1s', '300s') native_time_ns{series="past"};
+
++---------------------+--------+-------+
+| ts                  | series | val   |
++---------------------+--------+-------+
+| 1970-01-01T00:00:01 | past   | 602.0 |
++---------------------+--------+-------+
+
+-- Offsets select native timestamps, including stored negative time.
+TQL EVAL (0, 0, '1s', '300s') native_time_ns{series="offset"};
+
++---------------------+--------+-------+
+| ts                  | series | val   |
++---------------------+--------+-------+
+| 1970-01-01T00:00:00 | offset | 502.0 |
++---------------------+--------+-------+
+
+TQL EVAL (0, 0, '1s', '300s') native_time_ns{series="offset"} offset 1s;
+
++---------------------+--------+-------+
+| ts                  | series | val   |
++---------------------+--------+-------+
+| 1970-01-01T00:00:00 | offset | 501.0 |
++---------------------+--------+-------+
+
+TQL EVAL (0, 0, '1s', '300s') native_time_ns{series="offset"} offset -1s;
+
++---------------------+--------+-------+
+| ts                  | series | val   |
++---------------------+--------+-------+
+| 1970-01-01T00:00:00 | offset | 503.0 |
++---------------------+--------+-------+
+
+-- [1s] at 1s excludes 0 and 1s+tick, retaining 0+tick, 0+2ticks, and 1s.
+TQL EVAL (1, 1, '1s', '300s') count_over_time(native_time_ns{series="window"}[1s]);
+
++---------------------+------------------------------------+--------+
+| ts                  | prom_count_over_time(ts_range,val) | series |
++---------------------+------------------------------------+--------+
+| 1970-01-01T00:00:01 | 3.0                                | window |
++---------------------+------------------------------------+--------+
+
+TQL EVAL (1, 1, '1s', '300s') sum_over_time(native_time_ns{series="window"}[1s]);
+
++---------------------+----------------------------------+--------+
+| ts                  | prom_sum_over_time(ts_range,val) | series |
++---------------------+----------------------------------+--------+
+| 1970-01-01T00:00:01 | 10.0                             | window |
++---------------------+----------------------------------+--------+
+
+TQL EVAL (1, 1, '1s', '300s') last_over_time(native_time_ns{series="window"}[1s]);
+
++---------------------+-----------------------------------+--------+
+| ts                  | prom_last_over_time(ts_range,val) | series |
++---------------------+-----------------------------------+--------+
+| 1970-01-01T00:00:01 | 3.0                               | window |
++---------------------+-----------------------------------+--------+
+
+-- The inner selector consumes native time; the subquery consumes ms evaluations.
+TQL EVAL (1, 1, '1s') last_over_time((native_time_ns{series="exact"})[1s:1s]);
+
++---------------------+-----------------------------------+--------+
+| ts                  | prom_last_over_time(ts_range,val) | series |
++---------------------+-----------------------------------+--------+
+| 1970-01-01T00:00:01 | 201.0                             | exact  |
++---------------------+-----------------------------------+--------+
+
+DROP TABLE native_time_ns;
+
+Affected Rows: 0
+
+-- Second precision is promoted before applying fractional-second offsets.
+CREATE TABLE native_time_sec (ts TIMESTAMP(0) TIME INDEX, val DOUBLE);
+
+Affected Rows: 0
+
+INSERT INTO native_time_sec VALUES (0, 10), (1, 11), (2, 12);
+
+Affected Rows: 3
+
+TQL EVAL (1, 1, '1s', '1s') native_time_sec offset 500ms;
+
++------+---------------------+
+| val  | ts                  |
++------+---------------------+
+| 10.0 | 1970-01-01T00:00:01 |
++------+---------------------+
+
+TQL EVAL (1, 1, '1s', '1s') native_time_sec offset -500ms;
+
++------+---------------------+
+| val  | ts                  |
++------+---------------------+
+| 11.0 | 1970-01-01T00:00:01 |
++------+---------------------+
+
+DROP TABLE native_time_sec;
+
+Affected Rows: 0
+

tests/cases/standalone/common/promql/native_time_selection.sql

@@ -0,0 +1,133 @@
+-- Regression coverage for instant and range selection on native microsecond and
+-- nanosecond time indexes.
+
+CREATE TABLE native_time_us (
+    ts TIMESTAMP(6) TIME INDEX,
+    series STRING PRIMARY KEY,
+    val DOUBLE,
+);
+
+INSERT INTO native_time_us VALUES
+    (1000001, 'future', 101),
+    (1000000, 'exact', 201),
+    (1000001, 'exact', 202),
+    (-299000000, 'lowerbound', 301),
+    (-298999999, 'lowerplus', 302),
+    (1000000, 'positive_lowerbound', 701),
+    (1000001, 'positive_lowerplus', 702),
+    (1000001, 'multi', 401),
+    (-1000000, 'offset', 501),
+    (0, 'offset', 502),
+    (1000000, 'offset', 503),
+    (999999, 'past', 602),
+    (999001, 'past', 601),
+    (0, 'window', 1),
+    (1, 'window', 2),
+    (2, 'window', 5),
+    (1000000, 'window', 3),
+    (1000001, 'window', 4);
+
+-- Future-only selection is empty before flushing, exercising the memtable path.
+TQL EVAL (1, 1, '1s', '300s') native_time_us{series="future"};
+
+ADMIN FLUSH_TABLE('native_time_us');
+
+-- At 1s, selection keeps an exact native timestamp.
+TQL EVAL (1, 1, '1s', '300s') native_time_us{series="exact"};
+TQL EVAL (1, 1, '1s', '300s') timestamp(native_time_us{series="future"});
+TQL EVAL (1, 1, '1s', '300s') timestamp(native_time_us{series="exact"});
+
+-- Instant lookback bounds are exclusive: these return only 302 and 702.
+TQL EVAL (1, 1, '1s', '300s') native_time_us{series=~"lower.*"};
+TQL EVAL (301, 301, '1s', '300s') native_time_us{series=~"positive_lower.*"};
+
+-- The sub-millisecond point belongs only to the 2s evaluation step.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1, 2, '1s', '300s') native_time_us{series="multi"};
+
+-- The latest native timestamp below 1s is retained even when inserts are unordered.
+TQL EVAL (1, 1, '1s', '300s') native_time_us{series="past"};
+
+-- Offsets select native timestamps, including stored negative time.
+TQL EVAL (0, 0, '1s', '300s') native_time_us{series="offset"};
+TQL EVAL (0, 0, '1s', '300s') native_time_us{series="offset"} offset 1s;
+TQL EVAL (0, 0, '1s', '300s') native_time_us{series="offset"} offset -1s;
+
+-- [1s] at 1s excludes 0 and 1s+tick, retaining 0+tick, 0+2ticks, and 1s.
+TQL EVAL (1, 1, '1s', '300s') count_over_time(native_time_us{series="window"}[1s]);
+TQL EVAL (1, 1, '1s', '300s') sum_over_time(native_time_us{series="window"}[1s]);
+TQL EVAL (1, 1, '1s', '300s') last_over_time(native_time_us{series="window"}[1s]);
+
+-- The inner selector consumes native time; the subquery consumes ms evaluations.
+TQL EVAL (1, 1, '1s') last_over_time((native_time_us{series="exact"})[1s:1s]);
+
+DROP TABLE native_time_us;
+
+CREATE TABLE native_time_ns (
+    ts TIMESTAMP(9) TIME INDEX,
+    series STRING PRIMARY KEY,
+    val DOUBLE,
+);
+
+INSERT INTO native_time_ns VALUES
+    (1000000001, 'future', 101),
+    (1000000000, 'exact', 201),
+    (1000000001, 'exact', 202),
+    (-299000000000, 'lowerbound', 301),
+    (-298999999999, 'lowerplus', 302),
+    (1000000000, 'positive_lowerbound', 701),
+    (1000000001, 'positive_lowerplus', 702),
+    (1000000001, 'multi', 401),
+    (-1000000000, 'offset', 501),
+    (0, 'offset', 502),
+    (1000000000, 'offset', 503),
+    (999999000, 'past', 602),
+    (999001000, 'past', 601),
+    (0, 'window', 1),
+    (1, 'window', 2),
+    (2, 'window', 5),
+    (1000000000, 'window', 3),
+    (1000000001, 'window', 4);
+
+-- Future-only selection is empty before flushing, exercising the memtable path.
+TQL EVAL (1, 1, '1s', '300s') native_time_ns{series="future"};
+
+ADMIN FLUSH_TABLE('native_time_ns');
+
+-- At 1s, selection keeps an exact native timestamp.
+TQL EVAL (1, 1, '1s', '300s') native_time_ns{series="exact"};
+TQL EVAL (1, 1, '1s', '300s') timestamp(native_time_ns{series="future"});
+TQL EVAL (1, 1, '1s', '300s') timestamp(native_time_ns{series="exact"});
+
+-- Instant lookback bounds are exclusive: these return only 302 and 702.
+TQL EVAL (1, 1, '1s', '300s') native_time_ns{series=~"lower.*"};
+TQL EVAL (301, 301, '1s', '300s') native_time_ns{series=~"positive_lower.*"};
+
+-- The sub-millisecond point belongs only to the 2s evaluation step.
+-- SQLNESS SORT_RESULT 3 1
+TQL EVAL (1, 2, '1s', '300s') native_time_ns{series="multi"};
+
+-- The latest native timestamp below 1s is retained even when inserts are unordered.
+TQL EVAL (1, 1, '1s', '300s') native_time_ns{series="past"};
+
+-- Offsets select native timestamps, including stored negative time.
+TQL EVAL (0, 0, '1s', '300s') native_time_ns{series="offset"};
+TQL EVAL (0, 0, '1s', '300s') native_time_ns{series="offset"} offset 1s;
+TQL EVAL (0, 0, '1s', '300s') native_time_ns{series="offset"} offset -1s;
+
+-- [1s] at 1s excludes 0 and 1s+tick, retaining 0+tick, 0+2ticks, and 1s.
+TQL EVAL (1, 1, '1s', '300s') count_over_time(native_time_ns{series="window"}[1s]);
+TQL EVAL (1, 1, '1s', '300s') sum_over_time(native_time_ns{series="window"}[1s]);
+TQL EVAL (1, 1, '1s', '300s') last_over_time(native_time_ns{series="window"}[1s]);
+
+-- The inner selector consumes native time; the subquery consumes ms evaluations.
+TQL EVAL (1, 1, '1s') last_over_time((native_time_ns{series="exact"})[1s:1s]);
+
+DROP TABLE native_time_ns;
+
+-- Second precision is promoted before applying fractional-second offsets.
+CREATE TABLE native_time_sec (ts TIMESTAMP(0) TIME INDEX, val DOUBLE);
+INSERT INTO native_time_sec VALUES (0, 10), (1, 11), (2, 12);
+TQL EVAL (1, 1, '1s', '1s') native_time_sec offset 500ms;
+TQL EVAL (1, 1, '1s', '1s') native_time_sec offset -500ms;
+DROP TABLE native_time_sec;