feat: expose region min/max timestamp in region_statistics (#9060)

* feat: expose region min/max timestamp in region_statistics

Signed-off-by: Sainath Singineedi <44405294+sainad2222@users.noreply.github.com>

* test: cover region_statistic time range assembly and projected values

Signed-off-by: Sainath Singineedi <44405294+sainad2222@users.noreply.github.com>

---------

Signed-off-by: Sainath Singineedi <44405294+sainad2222@users.noreply.github.com>

src/catalog/src/system_schema/information_schema/region_statistics.rs

@@ -20,13 +20,18 @@ use common_error::ext::BoxedError;
 use common_meta::datanode::RegionStat;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{DfSendableRecordBatchStream, RecordBatch, SendableRecordBatchStream};
+use common_time::timestamp::TimeUnit;
 use datafusion::execution::TaskContext;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
 use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
+use datatypes::timestamp::TimestampMillisecond;
 use datatypes::value::Value;
-use datatypes::vectors::{StringVectorBuilder, UInt32VectorBuilder, UInt64VectorBuilder};
+use datatypes::vectors::{
+    StringVectorBuilder, TimestampMillisecondVectorBuilder, UInt32VectorBuilder,
+    UInt64VectorBuilder,
+};
 use snafu::ResultExt;
 use store_api::storage::{ScanRequest, TableId};
 
@@ -51,6 +56,8 @@ const SST_NUM: &str = "sst_num";
 const INDEX_SIZE: &str = "index_size";
 const ENGINE: &str = "engine";
 const REGION_ROLE: &str = "region_role";
+const MIN_TIMESTAMP: &str = "min_timestamp";
+const MAX_TIMESTAMP: &str = "max_timestamp";
 
 const INIT_CAPACITY: usize = 42;
 
@@ -109,6 +116,16 @@ impl InformationSchemaRegionStatistics {
             ColumnSchema::new(INDEX_SIZE, ConcreteDataType::uint64_datatype(), true),
             ColumnSchema::new(ENGINE, ConcreteDataType::string_datatype(), true),
             ColumnSchema::new(REGION_ROLE, ConcreteDataType::string_datatype(), true),
+            ColumnSchema::new(
+                MIN_TIMESTAMP,
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                true,
+            ),
+            ColumnSchema::new(
+                MAX_TIMESTAMP,
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                true,
+            ),
         ]))
     }
 
@@ -175,6 +192,8 @@ struct InformationSchemaRegionStatisticsBuilder {
     index_sizes: UInt64VectorBuilder,
     engines: StringVectorBuilder,
     region_roles: StringVectorBuilder,
+    min_timestamps: TimestampMillisecondVectorBuilder,
+    max_timestamps: TimestampMillisecondVectorBuilder,
 }
 
 impl InformationSchemaRegionStatisticsBuilder {
@@ -197,6 +216,8 @@ impl InformationSchemaRegionStatisticsBuilder {
             index_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             engines: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             region_roles: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            min_timestamps: TimestampMillisecondVectorBuilder::with_capacity(INIT_CAPACITY),
+            max_timestamps: TimestampMillisecondVectorBuilder::with_capacity(INIT_CAPACITY),
         }
     }
 
@@ -261,6 +282,20 @@ impl InformationSchemaRegionStatisticsBuilder {
         self.index_sizes.push(Some(region_stat.index_size));
         self.engines.push(Some(&region_stat.engine));
         self.region_roles.push(Some(&region_stat.role.to_string()));
+        // Floor the min and ceil the max so the window stays a superset: a narrower
+        // one would hide regions from a time-bounded lookup.
+        self.min_timestamps.push(
+            region_stat
+                .min_timestamp
+                .and_then(|ts| ts.convert_to(TimeUnit::Millisecond))
+                .map(TimestampMillisecond),
+        );
+        self.max_timestamps.push(
+            region_stat
+                .max_timestamp
+                .and_then(|ts| ts.convert_to_ceil(TimeUnit::Millisecond))
+                .map(TimestampMillisecond),
+        );
     }
 
     fn finish(&mut self) -> Result<RecordBatch> {
@@ -280,6 +315,8 @@ impl InformationSchemaRegionStatisticsBuilder {
             Arc::new(self.index_sizes.finish()),
             Arc::new(self.engines.finish()),
             Arc::new(self.region_roles.finish()),
+            Arc::new(self.min_timestamps.finish()),
+            Arc::new(self.max_timestamps.finish()),
         ];
 
         RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)

src/common/meta/src/datanode.rs

@@ -16,7 +16,7 @@ use std::collections::{HashMap, HashSet};
 use std::str::FromStr;
 
 use api::v1::meta::{DatanodeWorkloads, HeartbeatRequest, RequestHeader};
-use common_time::util as time_util;
+use common_time::{Timestamp, util as time_util};
 use lazy_static::lazy_static;
 use regex::Regex;
 use serde::{Deserialize, Serialize};
@@ -117,6 +117,12 @@ pub struct RegionStat {
     /// **Only used by remote WAL prune.**
     /// In mito engine, this is the same as `data_topic_latest_entry_id`.
     pub metadata_topic_latest_entry_id: u64,
+    /// The earliest timestamp of the region's current data, if it holds any.
+    #[serde(default)]
+    pub min_timestamp: Option<Timestamp>,
+    /// The latest timestamp of the region's current data, if it holds any.
+    #[serde(default)]
+    pub max_timestamp: Option<Timestamp>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -331,6 +337,8 @@ impl From<&api::v1::meta::RegionStat> for RegionStat {
             query_scanned_bytes: region_stat.query_scanned_bytes,
             data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id,
             metadata_topic_latest_entry_id: region_stat.metadata_topic_latest_entry_id,
+            min_timestamp: region_stat.min_timestamp,
+            max_timestamp: region_stat.max_timestamp,
         }
     }
 }
@@ -593,6 +601,8 @@ mod tests {
                 query_scanned_bytes: 20,
                 data_topic_latest_entry_id: 0,
                 metadata_topic_latest_entry_id: 0,
+                min_timestamp: None,
+                max_timestamp: None,
             }],
             ..Default::default()
         };

src/meta-srv/src/gc/handler.rs

@@ -499,5 +499,7 @@ fn dropped_region_stat(region_id: RegionId) -> RegionStat {
         query_scanned_bytes: 0,
         data_topic_latest_entry_id: 0,
         metadata_topic_latest_entry_id: 0,
+        min_timestamp: None,
+        max_timestamp: None,
     }
 }

src/meta-srv/src/gc/mock.rs

@@ -450,6 +450,8 @@ fn mock_region_stat(
         index_size: 0,
         data_topic_latest_entry_id: 0,
         metadata_topic_latest_entry_id: 0,
+        min_timestamp: None,
+        max_timestamp: None,
         written_bytes: 0,
         query_cpu_time: 0,
         query_scanned_bytes: 0,

src/meta-srv/src/handler/collect_leader_region_handler.rs

@@ -88,6 +88,8 @@ mod tests {
             index_size: 0,
             data_topic_latest_entry_id: 0,
             metadata_topic_latest_entry_id: 0,
+            min_timestamp: None,
+            max_timestamp: None,
             written_bytes: 0,
             query_cpu_time: 0,
             query_scanned_bytes: 0,

src/meta-srv/src/handler/failure_handler.rs

@@ -109,6 +109,8 @@ mod tests {
                 written_bytes: 0,
                 query_cpu_time: 0,
                 query_scanned_bytes: 0,
+                min_timestamp: None,
+                max_timestamp: None,
             }
         }
         acc.stat = Some(Stat {

src/meta-srv/src/handler/persist_stats_handler.rs

@@ -302,6 +302,8 @@ mod tests {
             query_scanned_bytes: 0,
             data_topic_latest_entry_id: 200,
             metadata_topic_latest_entry_id: 200,
+            min_timestamp: None,
+            max_timestamp: None,
         }
     }
 

src/meta-srv/src/handler/region_lease_handler.rs

@@ -182,6 +182,8 @@ mod test {
             written_bytes: 0,
             query_cpu_time: 0,
             query_scanned_bytes: 0,
+            min_timestamp: None,
+            max_timestamp: None,
         }
     }
 

src/meta-srv/src/selector/weight_compute.rs

@@ -202,6 +202,8 @@ mod tests {
                 written_bytes: 0,
                 query_cpu_time: 0,
                 query_scanned_bytes: 0,
+                min_timestamp: None,
+                max_timestamp: None,
             }],
             ..Default::default()
         }
@@ -234,6 +236,8 @@ mod tests {
                 written_bytes: 0,
                 query_cpu_time: 0,
                 query_scanned_bytes: 0,
+                min_timestamp: None,
+                max_timestamp: None,
             }],
             ..Default::default()
         }
@@ -266,6 +270,8 @@ mod tests {
                 written_bytes: 0,
                 query_cpu_time: 0,
                 query_scanned_bytes: 0,
+                min_timestamp: None,
+                max_timestamp: None,
             }],
             ..Default::default()
         }

src/metric-engine/src/utils.rs

@@ -84,6 +84,10 @@ fn merge_region_statistic(
         query_scanned_bytes: data_stat.query_scanned_bytes,
         data_topic_latest_entry_id: data_stat.data_topic_latest_entry_id,
         metadata_topic_latest_entry_id: metadata_stat.metadata_topic_latest_entry_id,
+        // Metadata rows are written at timestamp 0, so merging their bounds would
+        // pin every metric region's minimum to the epoch.
+        min_timestamp: data_stat.min_timestamp,
+        max_timestamp: data_stat.max_timestamp,
     }
 }
 
@@ -120,6 +124,8 @@ pub(crate) fn encode_manifest_info_to_extensions(
 #[cfg(test)]
 mod tests {
 
+    use common_time::Timestamp;
+
     use super::*;
 
     #[test]
@@ -144,6 +150,76 @@ mod tests {
         assert_eq!(to_data_region_id(region_id), expected_region_id);
     }
 
+    #[test]
+    fn merge_region_statistic_ignores_metadata_timestamps() {
+        // Metadata rows are written at timestamp 0; folding them in would report
+        // 1970 as the minimum for a region holding only modern samples.
+        let metadata_stat = RegionStatistic {
+            min_timestamp: Some(Timestamp::new_millisecond(0)),
+            max_timestamp: Some(Timestamp::new_millisecond(0)),
+            ..Default::default()
+        };
+        let data_stat = RegionStatistic {
+            min_timestamp: Some(Timestamp::new_millisecond(1_700_000_000_000)),
+            max_timestamp: Some(Timestamp::new_millisecond(1_700_000_001_000)),
+            ..Default::default()
+        };
+
+        let statistic = merge_region_statistic(&metadata_stat, &data_stat);
+
+        assert_eq!(
+            statistic.min_timestamp,
+            Some(Timestamp::new_millisecond(1_700_000_000_000))
+        );
+        assert_eq!(
+            statistic.max_timestamp,
+            Some(Timestamp::new_millisecond(1_700_000_001_000))
+        );
+    }
+
+    #[test]
+    fn merge_region_statistic_reports_no_range_for_an_empty_data_region() {
+        // Metadata present but no samples: the region has no data to bound, so
+        // both ends must stay NULL rather than collapsing onto the epoch.
+        let metadata_stat = RegionStatistic {
+            min_timestamp: Some(Timestamp::new_millisecond(0)),
+            max_timestamp: Some(Timestamp::new_millisecond(0)),
+            ..Default::default()
+        };
+
+        let statistic = merge_region_statistic(&metadata_stat, &RegionStatistic::default());
+
+        assert_eq!(statistic.min_timestamp, None);
+        assert_eq!(statistic.max_timestamp, None);
+    }
+
+    #[test]
+    fn merge_region_statistic_keeps_pre_epoch_samples() {
+        // Samples entirely before 1970 must not have their maximum pulled up to
+        // the metadata timestamp.
+        let metadata_stat = RegionStatistic {
+            min_timestamp: Some(Timestamp::new_millisecond(0)),
+            max_timestamp: Some(Timestamp::new_millisecond(0)),
+            ..Default::default()
+        };
+        let data_stat = RegionStatistic {
+            min_timestamp: Some(Timestamp::new_millisecond(-2_000)),
+            max_timestamp: Some(Timestamp::new_millisecond(-1_000)),
+            ..Default::default()
+        };
+
+        let statistic = merge_region_statistic(&metadata_stat, &data_stat);
+
+        assert_eq!(
+            statistic.min_timestamp,
+            Some(Timestamp::new_millisecond(-2_000))
+        );
+        assert_eq!(
+            statistic.max_timestamp,
+            Some(Timestamp::new_millisecond(-1_000))
+        );
+    }
+
     #[test]
     fn merge_region_statistic_uses_data_region_query_stats() {
         let metadata_stat = RegionStatistic {

src/mito2/src/engine/basic_test.rs

@@ -24,6 +24,7 @@ use common_base::readable_size::ReadableSize;
 use common_error::ext::ErrorExt;
 use common_error::status_code::StatusCode;
 use common_recordbatch::RecordBatches;
+use common_time::Timestamp;
 use common_wal::options::WAL_OPTIONS_KEY;
 use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::ColumnSchema;
@@ -808,6 +809,93 @@ async fn test_region_usage_with_format(flat_format: bool) {
     assert!(region_stat.estimated_disk_size() > 3000);
 }
 
+#[tokio::test]
+async fn test_region_usage_time_range() {
+    test_region_usage_time_range_with_format(false).await;
+    test_region_usage_time_range_with_format(true).await;
+}
+
+async fn test_region_usage_time_range_with_format(flat_format: bool) {
+    let mut env = TestEnv::with_prefix("region_usage_time_range").await;
+    let engine = env
+        .create_engine(MitoConfig {
+            default_flat_format: flat_format,
+            ..Default::default()
+        })
+        .await;
+
+    let region_id = RegionId::new(1, 1);
+    let request = CreateRequestBuilder::new().build();
+    let column_schemas = rows_schema(&request);
+    engine
+        .handle_request(region_id, RegionRequest::Create(request))
+        .await
+        .unwrap();
+    let region = engine.get_region(region_id).unwrap();
+
+    // An empty region has nothing to bound.
+    let region_stat = region.region_statistic();
+    assert_eq!(region_stat.min_timestamp, None);
+    assert_eq!(region_stat.max_timestamp, None);
+
+    // Unflushed data lives only in the memtable, so the range must come from there.
+    put_rows(
+        &engine,
+        region_id,
+        Rows {
+            schema: column_schemas.clone(),
+            rows: build_rows_for_key("a", 10, 20, 0),
+        },
+    )
+    .await;
+
+    let region_stat = region.region_statistic();
+    assert_eq!(
+        region_stat.min_timestamp,
+        Some(Timestamp::new_millisecond(10_000))
+    );
+    assert_eq!(
+        region_stat.max_timestamp,
+        Some(Timestamp::new_millisecond(19_000))
+    );
+
+    // After a flush the same range is served by the SST instead.
+    flush_region(&engine, region_id, None).await;
+
+    let region_stat = region.region_statistic();
+    assert_eq!(
+        region_stat.min_timestamp,
+        Some(Timestamp::new_millisecond(10_000))
+    );
+    assert_eq!(
+        region_stat.max_timestamp,
+        Some(Timestamp::new_millisecond(19_000))
+    );
+
+    // A later write lands in the memtable below the SST range: the minimum must
+    // come from the memtable and the maximum from the SST, so a merge that drops
+    // either side is caught.
+    put_rows(
+        &engine,
+        region_id,
+        Rows {
+            schema: column_schemas.clone(),
+            rows: build_rows_for_key("b", 0, 5, 0),
+        },
+    )
+    .await;
+
+    let region_stat = region.region_statistic();
+    assert_eq!(
+        region_stat.min_timestamp,
+        Some(Timestamp::new_millisecond(0))
+    );
+    assert_eq!(
+        region_stat.max_timestamp,
+        Some(Timestamp::new_millisecond(19_000))
+    );
+}
+
 #[tokio::test]
 async fn test_engine_with_write_cache() {
     test_engine_with_write_cache_with_format(false).await;

src/mito2/src/memtable/version.rs

@@ -17,6 +17,7 @@
 use std::sync::Arc;
 use std::time::Duration;
 
+use common_time::Timestamp;
 use smallvec::SmallVec;
 use store_api::metadata::RegionMetadataRef;
 use store_api::storage::SequenceNumber;
@@ -162,6 +163,17 @@ impl MemtableVersion {
             + self.mutable.num_rows()
     }
 
+    /// Returns the time range covered by the memtables, if any hold data.
+    pub(crate) fn time_range(&self) -> Option<(Timestamp, Timestamp)> {
+        let mut mutables = Vec::new();
+        self.mutable.list_memtables(&mut mutables);
+        self.immutables
+            .iter()
+            .chain(mutables.iter())
+            .filter_map(|mem| mem.stats().time_range())
+            .reduce(|(min_a, max_a), (min_b, max_b)| (min_a.min(min_b), max_a.max(max_b)))
+    }
+
     /// Returns true if the memtable version is empty.
     ///
     /// The version is empty when mutable memtable is empty and there is no

src/mito2/src/region.rs

@@ -701,6 +701,13 @@ impl MitoRegion {
         let manifest_version = self.stats.manifest_version();
         let file_removed_cnt = self.stats.file_removed_cnt();
 
+        let time_range = match (version.ssts.time_range(), version.memtables.time_range()) {
+            (Some((sst_min, sst_max)), Some((mem_min, mem_max))) => {
+                Some((sst_min.min(mem_min), sst_max.max(mem_max)))
+            }
+            (range, None) | (None, range) => range,
+        };
+
         let topic_latest_entry_id = self.topic_latest_entry_id.load(Ordering::Relaxed);
         let written_bytes = self.region_stats.written_bytes.load(Ordering::Relaxed);
         let query_cpu_time = self.region_stats.query_cpu_time.load(Ordering::Relaxed);
@@ -727,6 +734,8 @@ impl MitoRegion {
             written_bytes,
             query_cpu_time,
             query_scanned_bytes,
+            min_timestamp: time_range.map(|(min, _)| min),
+            max_timestamp: time_range.map(|(_, max)| max),
         }
     }
 

src/mito2/src/sst/version.rs

@@ -20,7 +20,7 @@ use std::sync::Arc;
 use common_time::{TimeToLive, Timestamp};
 use store_api::storage::{FileId, RegionId};
 
-use crate::sst::file::{FileHandle, FileMeta, Level, MAX_LEVEL};
+use crate::sst::file::{FileHandle, FileMeta, FileTimeRange, Level, MAX_LEVEL};
 use crate::sst::file_purger::FilePurgerRef;
 
 /// A version of all SSTs in a region.
@@ -152,6 +152,16 @@ impl SstVersion {
             .sum()
     }
 
+    /// Returns the time range covered by every file in this version, including
+    /// files referenced from other regions after a repartition.
+    pub(crate) fn time_range(&self) -> Option<FileTimeRange> {
+        self.levels
+            .iter()
+            .flat_map(|level_meta| level_meta.files.values())
+            .map(|file_handle| file_handle.time_range())
+            .reduce(|(min_a, max_a), (min_b, max_b)| (min_a.min(min_b), max_a.max(max_b)))
+    }
+
     /// Returns the space occupied by SST data files owned by `region_id`.
     pub(crate) fn owned_sst_usage(&self, region_id: RegionId) -> u64 {
         self.levels
@@ -257,6 +267,74 @@ mod tests {
     use super::*;
     use crate::test_util::new_noop_file_purger;
 
+    #[test]
+    fn time_range_spans_files_referenced_from_other_regions() {
+        let purger = new_noop_file_purger();
+        let owned = FileMeta {
+            file_id: FileId::random(),
+            region_id: RegionId::new(1, 1),
+            time_range: (
+                Timestamp::new_millisecond(200),
+                Timestamp::new_millisecond(300),
+            ),
+            ..Default::default()
+        };
+        let referenced = FileMeta {
+            file_id: FileId::random(),
+            region_id: RegionId::new(2, 1),
+            time_range: (
+                Timestamp::new_millisecond(50),
+                Timestamp::new_millisecond(100),
+            ),
+            ..Default::default()
+        };
+
+        let mut version = SstVersion::new();
+        version.add_files(purger, [owned, referenced].into_iter());
+
+        assert_eq!(
+            version.time_range(),
+            Some((
+                Timestamp::new_millisecond(50),
+                Timestamp::new_millisecond(300)
+            ))
+        );
+    }
+
+    #[test]
+    fn time_range_compares_across_units() {
+        let purger = new_noop_file_purger();
+        let seconds = FileMeta {
+            file_id: FileId::random(),
+            time_range: (Timestamp::new_second(1), Timestamp::new_second(2)),
+            ..Default::default()
+        };
+        let millis = FileMeta {
+            file_id: FileId::random(),
+            time_range: (
+                Timestamp::new_millisecond(500),
+                Timestamp::new_millisecond(2500),
+            ),
+            ..Default::default()
+        };
+
+        let mut version = SstVersion::new();
+        version.add_files(purger, [seconds, millis].into_iter());
+
+        assert_eq!(
+            version.time_range(),
+            Some((
+                Timestamp::new_millisecond(500),
+                Timestamp::new_millisecond(2500)
+            ))
+        );
+    }
+
+    #[test]
+    fn time_range_is_none_without_files() {
+        assert_eq!(SstVersion::new().time_range(), None);
+    }
+
     #[test]
     fn test_add_files() {
         let purger = new_noop_file_purger();

src/standalone/src/information_extension.rs

@@ -139,6 +139,8 @@ impl InformationExtension for StandaloneInformationExtension {
                     written_bytes: region_stat.written_bytes,
                     query_cpu_time: region_stat.query_cpu_time,
                     query_scanned_bytes: region_stat.query_scanned_bytes,
+                    min_timestamp: region_stat.min_timestamp,
+                    max_timestamp: region_stat.max_timestamp,
                 }
             })
             .collect::<Vec<_>>();

src/store-api/src/region_engine.rs

@@ -576,6 +576,14 @@ pub struct RegionStatistic {
     /// The total scanned bytes of the region since region opened.
     #[serde(default)]
     pub query_scanned_bytes: u64,
+    /// The earliest timestamp of the region's current data, or `None` if it holds
+    /// none. Unlike the size and row counters this covers files referenced from
+    /// other regions too, since a time range cannot double count.
+    #[serde(default)]
+    pub min_timestamp: Option<Timestamp>,
+    /// The latest timestamp of the region's current data. See [`Self::min_timestamp`].
+    #[serde(default)]
+    pub max_timestamp: Option<Timestamp>,
     /// The latest entry id of the region's remote WAL since last flush.
     /// For metric engine, there're two latest entry ids, one for data and one for metadata.
     /// TODO(weny): remove this two fields and use single instead.

tests/cases/standalone/common/information_schema/region_statistics.result

@@ -49,3 +49,60 @@ DROP TABLE test;
 
 Affected Rows: 0
 
+-- The projected columns are millisecond-typed. A nanosecond time index proves
+-- min_timestamp floors and max_timestamp ceils, so the reported window stays a
+-- superset of the data and cannot hide a region from a time-bounded lookup.
+CREATE TABLE precise (
+    a int primary key,
+    ts timestamp(9) time index,
+);
+
+Affected Rows: 0
+
+-- 1500000ns = 1.5ms floors to 1ms; 3999999ns = 3.999999ms ceils to 4ms.
+INSERT INTO precise VALUES
+       (1, 1500000),
+       (2, 2500000),
+       (3, 3999999);
+
+Affected Rows: 3
+
+-- SQLNESS SLEEP 3s
+SELECT min_timestamp, max_timestamp
+       FROM INFORMATION_SCHEMA.REGION_STATISTICS WHERE table_id
+       IN (SELECT TABLE_ID FROM INFORMATION_SCHEMA.TABLES WHERE table_name = 'precise' and table_schema = 'public');
+
++-------------------------+-------------------------+
+| min_timestamp           | max_timestamp           |
++-------------------------+-------------------------+
+| 1970-01-01T00:00:00.001 | 1970-01-01T00:00:00.004 |
++-------------------------+-------------------------+
+
+DROP TABLE precise;
+
+Affected Rows: 0
+
+-- A region with no rows has nothing to bound, so both ends stay NULL rather
+-- than collapsing onto the epoch.
+CREATE TABLE empty_region (
+    a int primary key,
+    ts timestamp time index,
+);
+
+Affected Rows: 0
+
+-- SQLNESS SLEEP 3s
+SELECT min_timestamp, max_timestamp
+       FROM INFORMATION_SCHEMA.REGION_STATISTICS WHERE table_id
+       IN (SELECT TABLE_ID FROM INFORMATION_SCHEMA.TABLES WHERE table_name = 'empty_region' and table_schema = 'public');
+
++---------------+---------------+
+| min_timestamp | max_timestamp |
++---------------+---------------+
+|               |               |
++---------------+---------------+
+
+DROP TABLE empty_region;
+
+Affected Rows: 0
+

tests/cases/standalone/common/information_schema/region_statistics.sql

@@ -29,3 +29,38 @@ SELECT SUM(region_rows), SUM(written_bytes_since_open), SUM(query_cpu_time_milli
 SELECT data_length, index_length, avg_row_length, table_rows FROM INFORMATION_SCHEMA.TABLES WHERE table_name = 'test';
 
 DROP TABLE test;
+
+-- The projected columns are millisecond-typed. A nanosecond time index proves
+-- min_timestamp floors and max_timestamp ceils, so the reported window stays a
+-- superset of the data and cannot hide a region from a time-bounded lookup.
+CREATE TABLE precise (
+    a int primary key,
+    ts timestamp(9) time index,
+);
+
+-- 1500000ns = 1.5ms floors to 1ms; 3999999ns = 3.999999ms ceils to 4ms.
+INSERT INTO precise VALUES
+       (1, 1500000),
+       (2, 2500000),
+       (3, 3999999);
+
+-- SQLNESS SLEEP 3s
+SELECT min_timestamp, max_timestamp
+       FROM INFORMATION_SCHEMA.REGION_STATISTICS WHERE table_id
+       IN (SELECT TABLE_ID FROM INFORMATION_SCHEMA.TABLES WHERE table_name = 'precise' and table_schema = 'public');
+
+DROP TABLE precise;
+
+-- A region with no rows has nothing to bound, so both ends stay NULL rather
+-- than collapsing onto the epoch.
+CREATE TABLE empty_region (
+    a int primary key,
+    ts timestamp time index,
+);
+
+-- SQLNESS SLEEP 3s
+SELECT min_timestamp, max_timestamp
+       FROM INFORMATION_SCHEMA.REGION_STATISTICS WHERE table_id
+       IN (SELECT TABLE_ID FROM INFORMATION_SCHEMA.TABLES WHERE table_name = 'empty_region' and table_schema = 'public');
+
+DROP TABLE empty_region;

tests/cases/standalone/common/system/information_schema.result

@@ -380,7 +380,9 @@ order by table_schema, table_name, column_name;
 | greptime      | information_schema | region_statistics                     | engine                            | 14               | 2147483647               | 2147483647             |                   |               |                    | utf8               | utf8_bin       |            |       | select,insert |                       | String               | string              | FIELD         |                | YES         | string              |                |        |
 | greptime      | information_schema | region_statistics                     | index_size                        | 13               |                          |                        | 20                | 0             |                    |                    |                |            |       | select,insert |                       | UInt64               | bigint unsigned     | FIELD         |                | YES         | bigint unsigned     |                |        |
 | greptime      | information_schema | region_statistics                     | manifest_size                     | 10               |                          |                        | 20                | 0             |                    |                    |                |            |       | select,insert |                       | UInt64               | bigint unsigned     | FIELD         |                | YES         | bigint unsigned     |                |        |
+| greptime      | information_schema | region_statistics                     | max_timestamp                     | 17               |                          |                        |                   |               | 3                  |                    |                |            |       | select,insert |                       | TimestampMillisecond | timestamp(3)        | FIELD         |                | YES         | timestamp(3)        |                |        |
 | greptime      | information_schema | region_statistics                     | memtable_size                     | 9                |                          |                        | 20                | 0             |                    |                    |                |            |       | select,insert |                       | UInt64               | bigint unsigned     | FIELD         |                | YES         | bigint unsigned     |                |        |
+| greptime      | information_schema | region_statistics                     | min_timestamp                     | 16               |                          |                        |                   |               | 3                  |                    |                |            |       | select,insert |                       | TimestampMillisecond | timestamp(3)        | FIELD         |                | YES         | timestamp(3)        |                |        |
 | greptime      | information_schema | region_statistics                     | query_cpu_time_millis             | 6                |                          |                        | 20                | 0             |                    |                    |                |            |       | select,insert |                       | UInt64               | bigint unsigned     | FIELD         |                | YES         | bigint unsigned     |                |        |
 | greptime      | information_schema | region_statistics                     | query_scanned_bytes               | 7                |                          |                        | 20                | 0             |                    |                    |                |            |       | select,insert |                       | UInt64               | bigint unsigned     | FIELD         |                | YES         | bigint unsigned     |                |        |
 | greptime      | information_schema | region_statistics                     | region_id                         | 1                |                          |                        | 20                | 0             |                    |                    |                |            |       | select,insert |                       | UInt64               | bigint unsigned     | FIELD         |                | NO          | bigint unsigned     |                |        |