fix: ensures logical and physical region have the same timestamp unit (#6041)

* fix: check time unit of logical region

* test: enlarge ttl for alter test to avoid data expired during test

* chore: fix unused

src/metric-engine/src/engine/create.rs

@@ -146,6 +146,23 @@ impl MetricEngineInner {
             .iter()
             .map(|metadata| (metadata.column_schema.name.clone(), metadata.column_id))
             .collect::<HashMap<_, _>>();
+        let time_index_unit = create_data_region_request
+            .column_metadatas
+            .iter()
+            .find_map(|metadata| {
+                if metadata.semantic_type == SemanticType::Timestamp {
+                    metadata
+                        .column_schema
+                        .data_type
+                        .as_timestamp()
+                        .map(|data_type| data_type.unit())
+                } else {
+                    None
+                }
+            })
+            .context(UnexpectedRequestSnafu {
+                reason: "No time index column found",
+            })?;
         self.mito
             .handle_request(
                 data_region_id,
@@ -170,6 +187,7 @@ impl MetricEngineInner {
             physical_columns,
             primary_key_encoding,
             physical_region_options,
+            time_index_unit,
         );
 
         Ok(())
@@ -184,15 +202,38 @@ impl MetricEngineInner {
     ) -> Result<()> {
         let data_region_id = utils::to_data_region_id(physical_region_id);
 
-        ensure!(
-            self.state
-                .read()
-                .unwrap()
-                .exist_physical_region(data_region_id),
-            PhysicalRegionNotFoundSnafu {
+        let unit = self
+            .state
+            .read()
+            .unwrap()
+            .physical_region_time_index_unit(physical_region_id)
+            .context(PhysicalRegionNotFoundSnafu {
                 region_id: data_region_id,
-            }
-        );
+            })?;
+        // Checks the time index unit of each request.
+        for (_, request) in &requests {
+            // Safety: verify_region_create_request() ensures that the request is valid.
+            let time_index_column = request
+                .column_metadatas
+                .iter()
+                .find(|col| col.semantic_type == SemanticType::Timestamp)
+                .unwrap();
+            let request_unit = time_index_column
+                .column_schema
+                .data_type
+                .as_timestamp()
+                .unwrap()
+                .unit();
+            ensure!(
+                request_unit == unit,
+                UnexpectedRequestSnafu {
+                    reason: format!(
+                        "Metric has differenttime unit ({:?}) than the physical region ({:?})",
+                        request_unit, unit
+                    ),
+                }
+            );
+        }
 
         // Filters out the requests that the logical region already exists
         let requests = {

src/metric-engine/src/engine/open.rs

@@ -16,6 +16,7 @@
 
 use std::collections::HashSet;
 
+use api::v1::SemanticType;
 use common_telemetry::info;
 use mito2::engine::MITO_ENGINE_NAME;
 use object_store::util::join_dir;
@@ -223,6 +224,21 @@ impl MetricEngineInner {
         {
             let mut state = self.state.write().unwrap();
             // recover physical column names
+            // Safety: The physical columns are loaded from the data region, which always
+            // has a time index.
+            let time_index_unit = physical_columns
+                .iter()
+                .find_map(|col| {
+                    if col.semantic_type == SemanticType::Timestamp {
+                        col.column_schema
+                            .data_type
+                            .as_timestamp()
+                            .map(|data_type| data_type.unit())
+                    } else {
+                        None
+                    }
+                })
+                .unwrap();
             let physical_columns = physical_columns
                 .into_iter()
                 .map(|col| (col.column_schema.name, col.column_id))
@@ -232,6 +248,7 @@ impl MetricEngineInner {
                 physical_columns,
                 primary_key_encoding,
                 physical_region_options,
+                time_index_unit,
             );
             // recover logical regions
             for logical_region_id in &logical_regions {

src/metric-engine/src/engine/state.rs

@@ -16,6 +16,7 @@
 
 use std::collections::{HashMap, HashSet};
 
+use common_time::timestamp::TimeUnit;
 use snafu::OptionExt;
 use store_api::codec::PrimaryKeyEncoding;
 use store_api::metadata::ColumnMetadata;
@@ -31,6 +32,7 @@ pub struct PhysicalRegionState {
     physical_columns: HashMap<String, ColumnId>,
     primary_key_encoding: PrimaryKeyEncoding,
     options: PhysicalRegionOptions,
+    time_index_unit: TimeUnit,
 }
 
 impl PhysicalRegionState {
@@ -38,12 +40,14 @@ impl PhysicalRegionState {
         physical_columns: HashMap<String, ColumnId>,
         primary_key_encoding: PrimaryKeyEncoding,
         options: PhysicalRegionOptions,
+        time_index_unit: TimeUnit,
     ) -> Self {
         Self {
             logical_regions: HashSet::new(),
             physical_columns,
             primary_key_encoding,
             options,
+            time_index_unit,
         }
     }
 
@@ -89,11 +93,17 @@ impl MetricEngineState {
         physical_columns: HashMap<String, ColumnId>,
         primary_key_encoding: PrimaryKeyEncoding,
         options: PhysicalRegionOptions,
+        time_index_unit: TimeUnit,
     ) {
         let physical_region_id = to_data_region_id(physical_region_id);
         self.physical_regions.insert(
             physical_region_id,
-            PhysicalRegionState::new(physical_columns, primary_key_encoding, options),
+            PhysicalRegionState::new(
+                physical_columns,
+                primary_key_encoding,
+                options,
+                time_index_unit,
+            ),
         );
     }
 
@@ -178,6 +188,15 @@ impl MetricEngineState {
         self.physical_regions.contains_key(&physical_region_id)
     }
 
+    pub fn physical_region_time_index_unit(
+        &self,
+        physical_region_id: RegionId,
+    ) -> Option<TimeUnit> {
+        self.physical_regions
+            .get(&physical_region_id)
+            .map(|state| state.time_index_unit)
+    }
+
     pub fn get_primary_key_encoding(
         &self,
         physical_region_id: RegionId,