feat: allow customized time index unit for metric engine table

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

@@ -234,8 +234,9 @@ impl MetricEngineInner {
                 request_unit == unit,
                 UnexpectedRequestSnafu {
                     reason: format!(
-                        "Metric has differenttime unit ({:?}) than the physical region ({:?})",
-                        request_unit, unit
+                        "Metric has different time unit ({request_unit:?}) than the physical \
+                         table ({unit:?}); the time index unit of a logical table must match \
+                         the physical table's",
                     ),
                 }
             );
@@ -1109,6 +1110,37 @@ mod test {
         assert!(is_tsid_col(tsid_metadata));
     }
 
+    #[tokio::test]
+    async fn test_create_logical_regions_time_index_unit_mismatch() {
+        let env = TestEnv::new().await;
+        let engine = env.metric();
+        let physical_region_id = env.default_physical_region_id();
+        let logical_region_id = env.default_logical_region_id();
+        env.create_physical_region_with_ts_type(
+            physical_region_id,
+            &TestEnv::default_table_dir(),
+            vec![],
+            ConcreteDataType::timestamp_microsecond_datatype(),
+        )
+        .await;
+
+        // A logical region request with a millisecond time index is rejected
+        // with the mismatch error naming both units.
+        let request = create_logical_region_request(
+            &["job"],
+            physical_region_id,
+            &store_api::path_utils::table_dir("test", logical_region_id.table_id()),
+        );
+        let err = engine
+            .handle_request(logical_region_id, RegionRequest::Create(request))
+            .await
+            .unwrap_err();
+        let message = err.to_string();
+        assert!(message.contains("Millisecond"), "{message}");
+        assert!(message.contains("Microsecond"), "{message}");
+        assert!(message.contains("different time unit"), "{message}");
+    }
+
     #[tokio::test]
     async fn test_create_logical_regions() {
         let env = TestEnv::new().await;

src/metric-engine/src/engine/put.rs

@@ -791,7 +791,9 @@ mod tests {
     use common_function::utils::partition_expr_version;
     use common_query::prelude::{greptime_native_histogram, greptime_timestamp, greptime_value};
     use common_recordbatch::RecordBatches;
-    use datatypes::arrow::array::{Float64Array, TimestampMillisecondArray};
+    use datatypes::arrow::array::{
+        Float64Array, TimestampMicrosecondArray, TimestampMillisecondArray,
+    };
     use datatypes::prelude::ConcreteDataType;
     use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema};
     use datatypes::value::Value as PartitionValue;
@@ -1116,6 +1118,91 @@ mod tests {
         expr.as_json_str().unwrap()
     }
 
+    #[tokio::test]
+    async fn test_put_and_scan_microsecond_physical_region() {
+        let env = TestEnv::new().await;
+        let engine = env.metric();
+        let physical_region_id = env.default_physical_region_id();
+        let logical_region_id = env.default_logical_region_id();
+        env.create_physical_region_with_ts_type(
+            physical_region_id,
+            &TestEnv::default_table_dir(),
+            vec![],
+            ConcreteDataType::timestamp_microsecond_datatype(),
+        )
+        .await;
+
+        // A logical region with a matching microsecond time index is accepted.
+        let region_create_request = test_util::create_logical_region_request_with_ts_type(
+            &["job"],
+            physical_region_id,
+            &table_dir("test", logical_region_id.table_id()),
+            ConcreteDataType::timestamp_microsecond_datatype(),
+        );
+        engine
+            .handle_request(
+                logical_region_id,
+                RegionRequest::Create(region_create_request),
+            )
+            .await
+            .unwrap();
+
+        // Writing microsecond rows works.
+        let affected_rows = engine
+            .handle_request(
+                logical_region_id,
+                RegionRequest::Put(RegionPutRequest {
+                    rows: Rows {
+                        schema: test_util::row_schema_with_tags_and_ts_datatype(
+                            &["job"],
+                            ColumnDataType::TimestampMicrosecond,
+                        ),
+                        rows: test_util::build_rows_with_ts_datatype(
+                            1,
+                            2,
+                            ColumnDataType::TimestampMicrosecond,
+                        ),
+                    },
+                    hint: None,
+                    partition_expr_version: None,
+                    skip_wal: false,
+                }),
+            )
+            .await
+            .unwrap();
+        assert_eq!(affected_rows.affected_rows, 2);
+
+        // The scan returns timestamps in the physical region's unit.
+        let stream = engine
+            .scan_to_stream(logical_region_id, ScanRequest::default())
+            .await
+            .unwrap();
+        let batches = RecordBatches::try_collect(stream).await.unwrap();
+        let mut rows = Vec::new();
+        for batch in batches.iter() {
+            let batch = batch.df_record_batch();
+            let timestamps = batch
+                .column(batch.schema().index_of(greptime_timestamp()).unwrap())
+                .as_any()
+                .downcast_ref::<TimestampMicrosecondArray>()
+                .unwrap();
+            let values = batch
+                .column(batch.schema().index_of(greptime_value()).unwrap())
+                .as_any()
+                .downcast_ref::<Float64Array>()
+                .unwrap();
+            rows.extend(
+                timestamps
+                    .values()
+                    .iter()
+                    .copied()
+                    .zip(values.values().iter().copied()),
+            );
+        }
+        rows.sort_unstable_by_key(|(timestamp, _)| *timestamp);
+        assert_eq!(rows, vec![(0, 0.0), (1, 1.0)]);
+    }
+
     async fn create_logical_region_with_tags(
         env: &TestEnv,
         physical_region_id: RegionId,

src/metric-engine/src/test_util.rs

@@ -158,6 +158,24 @@ impl TestEnv {
         physical_region_id: RegionId,
         table_dir: &str,
         options: Vec<(String, String)>,
+    ) {
+        self.create_physical_region_with_ts_type(
+            physical_region_id,
+            table_dir,
+            options,
+            ConcreteDataType::timestamp_millisecond_datatype(),
+        )
+        .await
+    }
+
+    /// Create regions in [MetricEngine] with specific `physical_region_id`
+    /// and time index column type.
+    pub async fn create_physical_region_with_ts_type(
+        &self,
+        physical_region_id: RegionId,
+        table_dir: &str,
+        options: Vec<(String, String)>,
+        ts_type: ConcreteDataType,
     ) {
         let region_create_request = RegionCreateRequest {
             engine: METRIC_ENGINE_NAME.to_string(),
@@ -165,11 +183,7 @@ impl TestEnv {
                 ColumnMetadata {
                     column_id: 0,
                     semantic_type: SemanticType::Timestamp,
-                    column_schema: ColumnSchema::new(
-                        greptime_timestamp(),
-                        ConcreteDataType::timestamp_millisecond_datatype(),
-                        false,
-                    ),
+                    column_schema: ColumnSchema::new(greptime_timestamp(), ts_type, false),
                 },
                 ColumnMetadata {
                     column_id: 1,
@@ -330,16 +344,28 @@ pub fn create_logical_region_request(
     tags: &[&str],
     physical_region_id: RegionId,
     table_dir: &str,
+) -> RegionCreateRequest {
+    create_logical_region_request_with_ts_type(
+        tags,
+        physical_region_id,
+        table_dir,
+        ConcreteDataType::timestamp_millisecond_datatype(),
+    )
+}
+
+/// Generate a [RegionCreateRequest] for logical region with the given time
+/// index column type.
+pub fn create_logical_region_request_with_ts_type(
+    tags: &[&str],
+    physical_region_id: RegionId,
+    table_dir: &str,
+    ts_type: ConcreteDataType,
 ) -> RegionCreateRequest {
     let mut column_metadatas = vec![
         ColumnMetadata {
             column_id: 0,
             semantic_type: SemanticType::Timestamp,
-            column_schema: ColumnSchema::new(
-                greptime_timestamp(),
-                ConcreteDataType::timestamp_millisecond_datatype(),
-                false,
-            ),
+            column_schema: ColumnSchema::new(greptime_timestamp(), ts_type, false),
         },
         ColumnMetadata {
             column_id: 1,
@@ -407,10 +433,20 @@ pub fn alter_logical_region_request(tags: &[&str]) -> RegionAlterRequest {
 ///
 /// The result will also contains default timestamp and value column at beginning.
 pub fn row_schema_with_tags(tags: &[&str]) -> Vec<PbColumnSchema> {
+    row_schema_with_tags_and_ts_datatype(tags, ColumnDataType::TimestampMillisecond)
+}
+
+/// Generate a row schema with given tag columns and time index datatype.
+///
+/// The result will also contains default timestamp and value column at beginning.
+pub fn row_schema_with_tags_and_ts_datatype(
+    tags: &[&str],
+    ts_datatype: ColumnDataType,
+) -> Vec<PbColumnSchema> {
     let mut schema = vec![
         PbColumnSchema {
             column_name: greptime_timestamp().to_string(),
-            datatype: ColumnDataType::TimestampMillisecond as i32,
+            datatype: ts_datatype as i32,
             semantic_type: SemanticType::Timestamp as _,
             datatype_extension: None,
             options: None,
@@ -440,11 +476,33 @@ pub fn row_schema_with_tags(tags: &[&str]) -> Vec<PbColumnSchema> {
 /// The schema is generated by [row_schema_with_tags]. `num_tags` doesn't need to be precise,
 /// it's used to determine the column id for new columns.
 pub fn build_rows(num_tags: usize, num_rows: usize) -> Vec<Row> {
+    build_rows_with_ts_datatype(num_tags, num_rows, ColumnDataType::TimestampMillisecond)
+}
+
+/// Build [Row]s whose time index values use the given timestamp datatype.
+///
+/// The schema is generated by [row_schema_with_tags_and_ts_datatype].
+/// `num_tags` doesn't need to be precise, it's used to determine the column id
+/// for new columns.
+pub fn build_rows_with_ts_datatype(
+    num_tags: usize,
+    num_rows: usize,
+    ts_datatype: ColumnDataType,
+) -> Vec<Row> {
+    let ts_value = |i: usize| -> ValueData {
+        match ts_datatype {
+            ColumnDataType::TimestampSecond => ValueData::TimestampSecondValue(i as _),
+            ColumnDataType::TimestampMillisecond => ValueData::TimestampMillisecondValue(i as _),
+            ColumnDataType::TimestampMicrosecond => ValueData::TimestampMicrosecondValue(i as _),
+            ColumnDataType::TimestampNanosecond => ValueData::TimestampNanosecondValue(i as _),
+            _ => unreachable!("not a timestamp datatype"),
+        }
+    };
     let mut rows = vec![];
     for i in 0..num_rows {
         let mut values = vec![
             Value {
-                value_data: Some(ValueData::TimestampMillisecondValue(i as _)),
+                value_data: Some(ts_value(i)),
             },
             Value {
                 value_data: Some(ValueData::F64Value(i as f64)),

src/operator/src/insert.rs

@@ -22,6 +22,7 @@ use api::v1::region::{
     InsertRequest as RegionInsertRequest, InsertRequests as RegionInsertRequests,
     RegionRequestHeader,
 };
+use api::v1::value::ValueData;
 use api::v1::{
     AlterTableExpr, ColumnDataType, ColumnSchema, CreateTableExpr, InsertRequests,
     RowInsertRequest, RowInsertRequests, Rows, SemanticType,
@@ -46,6 +47,8 @@ use common_query::native_histogram::{is_native_histogram_value_type, native_hist
 use common_query::prelude::{greptime_timestamp, greptime_value};
 use common_telemetry::tracing_context::TracingContext;
 use common_telemetry::{debug, error, warn};
+use common_time::Timestamp;
+use common_time::timestamp::TimeUnit;
 use datatypes::schema::SkippingIndexOptions;
 use futures_util::future;
 use meter_macros::write_meter;
@@ -436,9 +439,25 @@ impl Inserter {
         validate_column_count_match(&requests)?;
 
         // check and create physical table
-        self.create_physical_table_on_demand(&ctx, physical_table.clone(), statement_executor)
+        let physical_table_ref = self
+            .create_physical_table_on_demand(&ctx, physical_table.clone(), statement_executor)
             .await?;
 
+        // Align the requests' time index unit with the physical table's before
+        // creating logical tables or writing: the metric engine requires the
+        // logical tables (hence the requests) to use the same time index unit
+        // as the physical table. Ingestion endpoints encode timestamps in a
+        // fixed unit (Prometheus remote write always uses millisecond, OTLP
+        // uses nanosecond on the metric engine path), so rewrite the schema
+        // and values to the physical table's unit here. Converting to a
+        // coarser unit truncates the sub-unit part, following
+        // `Timestamp::convert_to`'s floor semantics.
+        //
+        // This reuses the table lookup from `create_physical_table_on_demand`
+        // (no extra catalog access) and is a no-op apart from a schema scan
+        // when the units already match.
+        align_row_insert_requests_time_unit(&physical_table_ref, &mut requests)?;
+
         // check and create logical tables
         let CreateAlterTableResult {
             instant_table_ids,
@@ -1052,17 +1071,16 @@ impl Inserter {
         ctx: &QueryContextRef,
         physical_table: String,
         statement_executor: &StatementExecutor,
-    ) -> Result<()> {
+    ) -> Result<TableRef> {
         let catalog_name = ctx.current_catalog();
         let schema_name = ctx.current_schema();
 
         // check if exist
-        if self
+        if let Some(table) = self
             .get_table(catalog_name, &schema_name, &physical_table)
             .await?
-            .is_some()
         {
-            return Ok(());
+            return Ok(table);
         }
 
         // Gate here too, otherwise a disabled switch would still leak the physical table.
@@ -1114,7 +1132,7 @@ impl Inserter {
             .await;
 
         match res {
-            Ok(_) => Ok(()),
+            Ok(table) => Ok(table),
             Err(err) => {
                 error!(err; "Failed to create table {table_reference}");
                 Err(err)
@@ -1388,6 +1406,124 @@ fn request_is_native_histogram(request_schema: &[ColumnSchema]) -> bool {
         )
 }
 
+/// Rewrites the time index column of every request in `requests` to the given
+/// physical metric table's time index unit.
+///
+/// This only rewrites requests whose time index unit differs from the
+/// physical table's; converting to a coarser unit truncates the sub-unit part
+/// (floor, see `Timestamp::convert_to`). A metric table without a timestamp
+/// column is left alone: the metric engine rejects it anyway.
+fn align_row_insert_requests_time_unit(
+    physical_table: &TableRef,
+    requests: &mut RowInsertRequests,
+) -> Result<()> {
+    let Some(target_unit) = physical_table
+        .table_info()
+        .meta
+        .schema
+        .timestamp_column()
+        .and_then(|col| col.data_type.as_timestamp().map(|ts| ts.unit()))
+    else {
+        return Ok(());
+    };
+    convert_row_insert_requests_time_unit(requests, target_unit)
+}
+
+/// Converts the time index column of each request to `target_unit`.
+fn convert_row_insert_requests_time_unit(
+    requests: &mut RowInsertRequests,
+    target_unit: TimeUnit,
+) -> Result<()> {
+    for request in &mut requests.inserts {
+        let Some(rows) = request.rows.as_mut() else {
+            continue;
+        };
+        convert_rows_time_unit(rows, target_unit)?;
+    }
+    Ok(())
+}
+
+fn convert_rows_time_unit(rows: &mut Rows, target_unit: TimeUnit) -> Result<()> {
+    let Some(ts_index) = rows
+        .schema
+        .iter()
+        .position(|col| col.semantic_type == SemanticType::Timestamp as i32)
+    else {
+        return Ok(());
+    };
+    let Some(source_unit) = timestamp_unit_from_datatype(rows.schema[ts_index].datatype) else {
+        return Ok(());
+    };
+    if source_unit == target_unit {
+        return Ok(());
+    }
+
+    rows.schema[ts_index].datatype = column_datatype_from_time_unit(target_unit) as i32;
+    // Timestamp columns never carry a datatype extension.
+    rows.schema[ts_index].datatype_extension = None;
+
+    for row in &mut rows.rows {
+        let Some(value) = row.values.get_mut(ts_index) else {
+            continue;
+        };
+        let Some(value_data) = value.value_data.take() else {
+            continue;
+        };
+        value.value_data =
+            convert_timestamp_value_data(value_data, source_unit, target_unit, ts_index)?;
+    }
+    Ok(())
+}
+
+fn timestamp_unit_from_datatype(datatype: i32) -> Option<TimeUnit> {
+    ColumnDataType::try_from(datatype)
+        .ok()
+        .and_then(|datatype| match datatype {
+            ColumnDataType::TimestampSecond => Some(TimeUnit::Second),
+            ColumnDataType::TimestampMillisecond => Some(TimeUnit::Millisecond),
+            ColumnDataType::TimestampMicrosecond => Some(TimeUnit::Microsecond),
+            ColumnDataType::TimestampNanosecond => Some(TimeUnit::Nanosecond),
+            _ => None,
+        })
+}
+
+fn column_datatype_from_time_unit(unit: TimeUnit) -> ColumnDataType {
+    match unit {
+        TimeUnit::Second => ColumnDataType::TimestampSecond,
+        TimeUnit::Millisecond => ColumnDataType::TimestampMillisecond,
+        TimeUnit::Microsecond => ColumnDataType::TimestampMicrosecond,
+        TimeUnit::Nanosecond => ColumnDataType::TimestampNanosecond,
+    }
+}
+
+fn convert_timestamp_value_data(
+    value_data: ValueData,
+    source_unit: TimeUnit,
+    target_unit: TimeUnit,
+    column_index: usize,
+) -> Result<Option<ValueData>> {
+    let timestamp = match value_data {
+        ValueData::TimestampSecondValue(v) => Timestamp::new_second(v),
+        ValueData::TimestampMillisecondValue(v) => Timestamp::new_millisecond(v),
+        ValueData::TimestampMicrosecondValue(v) => Timestamp::new_microsecond(v),
+        ValueData::TimestampNanosecondValue(v) => Timestamp::new_nanosecond(v),
+        // Null or non-timestamp value; nothing to convert.
+        other => return Ok(Some(other)),
+    };
+    let converted = timestamp.convert_to(target_unit).context(InvalidInsertRequestSnafu {
+        reason: format!(
+            "timestamp column {column_index} value {} in unit {source_unit:?} overflows when converting to unit {target_unit:?}",
+            timestamp.value()
+        ),
+    })?;
+    Ok(Some(match target_unit {
+        TimeUnit::Second => ValueData::TimestampSecondValue(converted.value()),
+        TimeUnit::Millisecond => ValueData::TimestampMillisecondValue(converted.value()),
+        TimeUnit::Microsecond => ValueData::TimestampMicrosecondValue(converted.value()),
+        TimeUnit::Nanosecond => ValueData::TimestampNanosecondValue(converted.value()),
+    }))
+}
+
 fn table_is_native_histogram(table: &TableRef) -> bool {
     let mut fields = table.field_columns();
     let Some(col) = fields.next() else {
@@ -1705,7 +1841,7 @@ mod tests {
 
     use api::helper::ColumnDataTypeWrapper;
     use api::v1::helper::{field_column_schema, time_index_column_schema};
-    use api::v1::{RowInsertRequest, Rows, Value};
+    use api::v1::{Row, RowInsertRequest, Rows, Value};
     use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
     use common_meta::cache::new_table_flownode_set_cache;
     use common_meta::ddl::test_util::datanode_handler::NaiveDatanodeHandler;
@@ -1770,6 +1906,204 @@ mod tests {
         ))
     }
 
+    fn make_metric_physical_table_ref_with_time_unit(unit: TimeUnit) -> TableRef {
+        let schema = datatypes::schema::SchemaBuilder::try_from_columns(vec![
+            ColumnSchema::new(
+                greptime_timestamp(),
+                ConcreteDataType::timestamp_datatype(unit),
+                false,
+            )
+            .with_time_index(true),
+            ColumnSchema::new(greptime_value(), ConcreteDataType::float64_datatype(), true),
+        ])
+        .unwrap()
+        .build()
+        .unwrap();
+        let meta = TableMetaBuilder::empty()
+            .schema(Arc::new(schema))
+            .primary_key_indices(vec![])
+            .value_indices(vec![1])
+            .engine("metric")
+            .next_column_id(0)
+            .options(Default::default())
+            .created_on(Default::default())
+            .build()
+            .unwrap();
+        let info = Arc::new(
+            TableInfoBuilder::default()
+                .table_id(1)
+                .table_version(0)
+                .name("greptime_physical_table")
+                .schema_name(DEFAULT_SCHEMA_NAME)
+                .catalog_name(DEFAULT_CATALOG_NAME)
+                .desc(None)
+                .table_type(TableType::Base)
+                .meta(meta)
+                .build()
+                .unwrap(),
+        );
+        Arc::new(table::Table::new(
+            info,
+            table::metadata::FilterPushDownType::Unsupported,
+            Arc::new(DummyDataSource),
+        ))
+    }
+
+    fn ms_row_insert_request(timestamp_ms: i64) -> RowInsertRequest {
+        RowInsertRequest {
+            table_name: "my_metric".to_string(),
+            rows: Some(Rows {
+                schema: vec![
+                    time_index_column_schema(
+                        greptime_timestamp(),
+                        ColumnDataType::TimestampMillisecond,
+                    ),
+                    field_column_schema(greptime_value(), ColumnDataType::Float64),
+                ],
+                rows: vec![Row {
+                    values: vec![
+                        Value {
+                            value_data: Some(ValueData::TimestampMillisecondValue(timestamp_ms)),
+                        },
+                        Value {
+                            value_data: Some(ValueData::F64Value(1.0)),
+                        },
+                    ],
+                }],
+            }),
+        }
+    }
+
+    #[test]
+    fn test_convert_row_insert_requests_time_unit_noop_when_matching() {
+        let mut requests = RowInsertRequests {
+            inserts: vec![ms_row_insert_request(123)],
+        };
+        convert_row_insert_requests_time_unit(&mut requests, TimeUnit::Millisecond).unwrap();
+        let rows = requests.inserts[0].rows.as_ref().unwrap();
+        assert_eq!(
+            rows.schema[0].datatype,
+            ColumnDataType::TimestampMillisecond as i32
+        );
+        assert!(matches!(
+            rows.rows[0].values[0].value_data,
+            Some(ValueData::TimestampMillisecondValue(123))
+        ));
+    }
+
+    #[test]
+    fn test_convert_row_insert_requests_time_unit_widens_losslessly() {
+        let mut requests = RowInsertRequests {
+            inserts: vec![ms_row_insert_request(123)],
+        };
+        convert_row_insert_requests_time_unit(&mut requests, TimeUnit::Microsecond).unwrap();
+        let rows = requests.inserts[0].rows.as_ref().unwrap();
+        assert_eq!(
+            rows.schema[0].datatype,
+            ColumnDataType::TimestampMicrosecond as i32
+        );
+        assert!(matches!(
+            rows.rows[0].values[0].value_data,
+            Some(ValueData::TimestampMicrosecondValue(123_000))
+        ));
+    }
+
+    #[test]
+    fn test_convert_row_insert_requests_time_unit_truncates_on_narrowing() {
+        // 123_456_789 ns floors to 123_456 us and 123 ms; negative values
+        // floor towards negative infinity, matching `Timestamp::convert_to`.
+        let requests = |value_ns: i64| RowInsertRequests {
+            inserts: vec![RowInsertRequest {
+                table_name: "my_metric".to_string(),
+                rows: Some(Rows {
+                    schema: vec![
+                        time_index_column_schema(
+                            greptime_timestamp(),
+                            ColumnDataType::TimestampNanosecond,
+                        ),
+                        field_column_schema(greptime_value(), ColumnDataType::Float64),
+                    ],
+                    rows: vec![Row {
+                        values: vec![
+                            Value {
+                                value_data: Some(ValueData::TimestampNanosecondValue(value_ns)),
+                            },
+                            Value {
+                                value_data: Some(ValueData::F64Value(1.0)),
+                            },
+                        ],
+                    }],
+                }),
+            }],
+        };
+
+        let mut reqs = requests(123_456_789);
+        convert_row_insert_requests_time_unit(&mut reqs, TimeUnit::Microsecond).unwrap();
+        let rows = reqs.inserts[0].rows.as_ref().unwrap();
+        assert!(matches!(
+            rows.rows[0].values[0].value_data,
+            Some(ValueData::TimestampMicrosecondValue(123_456))
+        ));
+
+        let mut reqs = requests(123_456_789);
+        convert_row_insert_requests_time_unit(&mut reqs, TimeUnit::Millisecond).unwrap();
+        let rows = reqs.inserts[0].rows.as_ref().unwrap();
+        assert!(matches!(
+            rows.rows[0].values[0].value_data,
+            Some(ValueData::TimestampMillisecondValue(123))
+        ));
+
+        let mut reqs = requests(-123_456_789);
+        convert_row_insert_requests_time_unit(&mut reqs, TimeUnit::Microsecond).unwrap();
+        let rows = reqs.inserts[0].rows.as_ref().unwrap();
+        assert!(matches!(
+            rows.rows[0].values[0].value_data,
+            Some(ValueData::TimestampMicrosecondValue(-123_457))
+        ));
+    }
+
+    #[test]
+    fn test_convert_row_insert_requests_time_unit_overflow() {
+        let mut requests = RowInsertRequests {
+            inserts: vec![ms_row_insert_request(i64::MAX)],
+        };
+        let err =
+            convert_row_insert_requests_time_unit(&mut requests, TimeUnit::Nanosecond).unwrap_err();
+        assert!(err.to_string().contains("overflows"), "{err}");
+    }
+
+    #[test]
+    fn test_align_row_insert_requests_time_unit() {
+        // Microsecond physical table, millisecond request (prometheus remote
+        // write shape): the request is rewritten to the physical unit.
+        let table = make_metric_physical_table_ref_with_time_unit(TimeUnit::Microsecond);
+        let mut requests = RowInsertRequests {
+            inserts: vec![ms_row_insert_request(123)],
+        };
+        align_row_insert_requests_time_unit(&table, &mut requests).unwrap();
+        let rows = requests.inserts[0].rows.as_ref().unwrap();
+        assert_eq!(
+            rows.schema[0].datatype,
+            ColumnDataType::TimestampMicrosecond as i32
+        );
+        assert!(matches!(
+            rows.rows[0].values[0].value_data,
+            Some(ValueData::TimestampMicrosecondValue(123_000))
+        ));
+
+        // Millisecond physical table (the auto-created default): no-op.
+        let table = make_metric_physical_table_ref_with_time_unit(TimeUnit::Millisecond);
+        let mut requests = RowInsertRequests {
+            inserts: vec![ms_row_insert_request(123)],
+        };
+        align_row_insert_requests_time_unit(&table, &mut requests).unwrap();
+        let rows = requests.inserts[0].rows.as_ref().unwrap();
+        assert_eq!(
+            rows.schema[0].datatype,
+            ColumnDataType::TimestampMillisecond as i32
+        );
+    }
+
     #[tokio::test]
     async fn test_accommodate_existing_schema_and_reject_kind_changes() {
         let ts_name = "my_ts";

src/servers/src/otlp/metrics.rs

@@ -971,9 +971,15 @@ fn write_timestamp(
     table: &mut TableData,
     row: &mut Vec<Value>,
     time_nano: i64,
-    legacy_mode: bool,
+    metric_ctx: &OtlpMetricCtx,
 ) -> Result<()> {
-    if legacy_mode {
+    // Keep the full nanosecond precision whenever the request is headed for
+    // the metric engine: `Inserter::handle_metric_row_inserts` converts the
+    // timestamps to the physical table's time index unit (which may be
+    // micro/nanosecond). Only the non-metric prometheus-compatible path is
+    // fixed to milliseconds, to keep auto-created mito tables on the
+    // millisecond time index.
+    if metric_ctx.is_legacy || metric_ctx.with_metric_engine {
         row_writer::write_ts_to_nanos(
             table,
             greptime_timestamp(),
@@ -1075,7 +1081,7 @@ fn write_tags_and_timestamp(
         )?;
     }
 
-    write_timestamp(table, row, timestamp_nanos, metric_ctx.is_legacy)?;
+    write_timestamp(table, row, timestamp_nanos, metric_ctx)?;
 
     Ok(())
 }
@@ -1512,6 +1518,7 @@ fn encode_summary(
 
 #[cfg(test)]
 mod tests {
+    use api::v1::ColumnDataType;
     use common_query::prelude::set_default_prefix;
     use otel_arrow_rust::proto::opentelemetry::common::v1::AnyValue;
     use otel_arrow_rust::proto::opentelemetry::common::v1::any_value::Value as Val;
@@ -2497,6 +2504,81 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_metric_engine_path_keeps_nanosecond_precision() {
+        let time_unix_nano = 1_704_067_200_123_456_789u64;
+        let request = metrics_request(vec![Metric {
+            name: "my_gauge".to_string(),
+            data: Some(metric::Data::Gauge(Gauge {
+                data_points: vec![NumberDataPoint {
+                    time_unix_nano,
+                    value: Some(Value::AsDouble(1.0)),
+                    ..Default::default()
+                }],
+            })),
+            ..Default::default()
+        }]);
+
+        // The metric engine path keeps the full nanosecond precision here;
+        // `Inserter::handle_metric_row_inserts` converts the timestamps to
+        // the physical table's time index unit afterwards.
+        let mut metric_ctx = OtlpMetricCtx {
+            with_metric_engine: true,
+            ..Default::default()
+        };
+        let MetricsConversion { requests, .. } =
+            to_grpc_insert_requests(request, &mut metric_ctx).unwrap();
+
+        let rows = requests.inserts[0].rows.as_ref().unwrap();
+        let ts_index = rows
+            .schema
+            .iter()
+            .position(|column| column.column_name == greptime_timestamp())
+            .unwrap();
+        assert_eq!(
+            rows.schema[ts_index].datatype,
+            ColumnDataType::TimestampNanosecond as i32
+        );
+        assert!(matches!(
+            rows.rows[0].values[ts_index].value_data,
+            Some(ValueData::TimestampNanosecondValue(
+                1_704_067_200_123_456_789
+            ))
+        ));
+
+        // The non-metric prometheus-compatible path stays millisecond so
+        // auto-created mito tables keep the millisecond time index.
+        let mut compat_ctx = OtlpMetricCtx::default();
+        let request = metrics_request(vec![Metric {
+            name: "my_gauge".to_string(),
+            data: Some(metric::Data::Gauge(Gauge {
+                data_points: vec![NumberDataPoint {
+                    time_unix_nano,
+                    value: Some(Value::AsDouble(1.0)),
+                    ..Default::default()
+                }],
+            })),
+            ..Default::default()
+        }]);
+        let MetricsConversion { requests, .. } =
+            to_grpc_insert_requests(request, &mut compat_ctx).unwrap();
+
+        let rows = requests.inserts[0].rows.as_ref().unwrap();
+        let ts_index = rows
+            .schema
+            .iter()
+            .position(|column| column.column_name == greptime_timestamp())
+            .unwrap();
+        assert_eq!(
+            rows.schema[ts_index].datatype,
+            ColumnDataType::TimestampMillisecond as i32
+        );
+        assert!(matches!(
+            rows.rows[0].values[ts_index].value_data,
+            Some(ValueData::TimestampMillisecondValue(1_704_067_200_123))
+        ));
+    }
+
     #[test]
     fn test_exponential_histogram_gate_and_partial_outcome() {
         let request = metrics_request(vec![

src/servers/src/prom_store.rs

@@ -23,9 +23,10 @@ use api::prom_store::remote::label_matcher::Type as MatcherType;
 use api::prom_store::remote::{Label, Query, Sample, TimeSeries, WriteRequest};
 use api::v1::RowInsertRequests;
 use arrow::array::{
-    Array, AsArray, DictionaryArray, LargeStringArray, StringArray, StringViewArray,
+    Array, ArrayRef, AsArray, DictionaryArray, LargeStringArray, StringArray, StringViewArray,
 };
-use arrow::datatypes::{Float64Type, TimestampMillisecondType, UInt32Type};
+use arrow::compute::kernels::cast as casts;
+use arrow::datatypes::{DataType, Float64Type, TimeUnit, TimestampMillisecondType, UInt32Type};
 use common_grpc::precision::Precision;
 use common_query::prelude::{greptime_timestamp, greptime_value};
 use common_recordbatch::{RecordBatch, RecordBatches};
@@ -383,6 +384,35 @@ fn recordbatch_to_timeseries(
         .with_context(|| error::InvalidPromRemoteReadQueryResultSnafu {
             msg: format!("missing timestamp column '{timestamp_column_name}' in query result"),
         })?;
+    // The Prometheus remote read wire format carries millisecond timestamps,
+    // while the table's time index can use any time unit (e.g. a metric
+    // physical table created with TIMESTAMP(6)). Narrowing casts truncate the
+    // sub-millisecond part.
+    let ts_column: ArrayRef = match ts_column.data_type() {
+        DataType::Timestamp(TimeUnit::Millisecond, _) => ts_column.clone(),
+        DataType::Timestamp(_, _) => casts::cast(
+            ts_column,
+            &DataType::Timestamp(TimeUnit::Millisecond, None),
+        )
+        .map_err(|e| {
+            error::InvalidPromRemoteReadQueryResultSnafu {
+                msg: format!(
+                    "failed to cast timestamp column '{timestamp_column_name}' of datatype {:?} to millisecond: {e}",
+                    ts_column.data_type()
+                ),
+            }
+            .build()
+        })?,
+        _ => {
+            return error::InvalidPromRemoteReadQueryResultSnafu {
+                msg: format!(
+                    "Expect timestamp column of datatype Timestamp(Millisecond), actual {:?}",
+                    ts_column.data_type()
+                ),
+            }
+            .fail();
+        }
+    };
     let ts_column = ts_column
         .as_primitive_opt::<TimestampMillisecondType>()
         .with_context(|| error::InvalidPromRemoteReadQueryResultSnafu {

tests-integration/src/otlp.rs

@@ -19,6 +19,7 @@ mod test {
     use client::{DEFAULT_CATALOG_NAME, OutputData};
     use common_recordbatch::RecordBatches;
     use datatypes::arrow::array::AsArray;
+    use datatypes::arrow::datatypes::TimestampMicrosecondType;
     use frontend::instance::Instance;
     use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
     use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest;
@@ -37,6 +38,7 @@ mod test {
     use servers::query_handler::OpenTelemetryProtocolHandler;
     use servers::query_handler::sql::SqlQueryHandler;
     use session::context::QueryContext;
+    use session::protocol_ctx::{OtlpMetricCtx, ProtocolCtx};
 
     use crate::standalone::GreptimeDbStandaloneBuilder;
     use crate::tests;
@@ -888,6 +890,99 @@ WITH(
         );
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    pub async fn test_otlp_metrics_into_microsecond_physical_table_on_standalone() {
+        let standalone = GreptimeDbStandaloneBuilder::new("test_otlp_us_physical")
+            .build()
+            .await;
+
+        test_otlp_metrics_into_microsecond_physical_table(standalone.fe_instance()).await;
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    pub async fn test_otlp_metrics_into_microsecond_physical_table_on_distributed() {
+        let instance = tests::create_distributed_instance("test_otlp_us_physical_dist").await;
+
+        test_otlp_metrics_into_microsecond_physical_table(&instance.frontend()).await;
+    }
+
+    /// Regression test for <https://github.com/GreptimeTeam/greptimedb/issues/9231>:
+    /// a physical metric table pre-created with a TIMESTAMP(6) (microsecond)
+    /// time index accepts OTLP ingestion; nanosecond samples are converted to
+    /// the physical table's unit, keeping microsecond precision.
+    async fn test_otlp_metrics_into_microsecond_physical_table(instance: &Arc<Instance>) {
+        let db = "otlp_us_physical";
+        let mut ctx = QueryContext::with(DEFAULT_CATALOG_NAME, db);
+        // Route the request to the metric engine, like the OTLP HTTP handler
+        // does with the default `prom_store.with_metric_engine = true`.
+        ctx.set_protocol_ctx(ProtocolCtx::OtlpMetric(OtlpMetricCtx {
+            with_metric_engine: true,
+            ..Default::default()
+        }));
+        let ctx = Arc::new(ctx);
+        assert!(
+            SqlQueryHandler::do_query(
+                instance.as_ref(),
+                &format!("CREATE DATABASE IF NOT EXISTS {db}"),
+                ctx.clone(),
+            )
+            .await
+            .first()
+            .unwrap()
+            .is_ok()
+        );
+
+        let mut output = instance
+            .do_query(
+                "CREATE TABLE greptime_physical_table (\
+                 greptime_timestamp TIMESTAMP(6) NOT NULL, \
+                 greptime_value DOUBLE NULL, \
+                 TIME INDEX (greptime_timestamp)) \
+                 ENGINE = metric WITH ('physical_metric_table' = 'true')",
+                ctx.clone(),
+            )
+            .await;
+        assert!(output.remove(0).is_ok());
+
+        let request = ExportMetricsServiceRequest {
+            resource_metrics: vec![ResourceMetrics {
+                scope_metrics: vec![ScopeMetrics {
+                    metrics: vec![Metric {
+                        name: "my_gauge".to_string(),
+                        data: Some(metric::Data::Gauge(Gauge {
+                            data_points: vec![NumberDataPoint {
+                                attributes: vec![keyvalue("host", "h1")],
+                                time_unix_nano: 1_704_067_200_123_456_789,
+                                value: Some(Value::AsDouble(1.0)),
+                                ..Default::default()
+                            }],
+                        })),
+                        ..Default::default()
+                    }],
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }],
+        };
+        instance.metrics(request, ctx.clone()).await.unwrap();
+
+        let mut output = instance
+            .do_query("SELECT greptime_timestamp FROM my_gauge", ctx.clone())
+            .await;
+        let OutputData::Stream(stream) = output.remove(0).unwrap().data else {
+            unreachable!()
+        };
+        let batches = RecordBatches::try_collect(stream).await.unwrap().take();
+        assert_eq!(batches[0].num_rows(), 1);
+        // The logical table's time index keeps the physical table's
+        // microsecond unit and precision: 1704067200123456789ns ->
+        // 1704067200123456us.
+        let timestamps = batches[0]
+            .column(0)
+            .as_primitive::<TimestampMicrosecondType>();
+        assert_eq!(timestamps.value(0), 1_704_067_200_123_456);
+    }
+
     fn build_request() -> ExportMetricsServiceRequest {
         let data_points = vec![
             NumberDataPoint {

tests-integration/src/prom_store.rs

@@ -20,7 +20,10 @@ mod tests {
     use api::prom_store::remote::{
         Label, LabelMatcher, Query, ReadRequest, ReadResponse, Sample, WriteRequest,
     };
+    use client::OutputData;
     use common_catalog::consts::DEFAULT_CATALOG_NAME;
+    use datatypes::arrow::array::AsArray;
+    use datatypes::arrow::datatypes::TimestampMicrosecondType;
     use frontend::instance::Instance;
     use prost::Message;
     use servers::http::prom_store::PHYSICAL_TABLE_PARAM;
@@ -83,6 +86,121 @@ mod tests {
         .await;
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_standalone_prom_store_remote_rw_microsecond_physical_table() {
+        common_telemetry::init_default_ut_logging();
+        let standalone =
+            GreptimeDbStandaloneBuilder::new("test_prom_store_remote_rw_us_physical_table")
+                .build()
+                .await;
+        let instance = standalone.fe_instance();
+
+        test_prom_store_remote_rw_microsecond_physical_table(instance).await;
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_distributed_prom_store_remote_rw_microsecond_physical_table() {
+        common_telemetry::init_default_ut_logging();
+        let distributed =
+            tests::create_distributed_instance("test_prom_store_remote_rw_us_physical_table").await;
+        test_prom_store_remote_rw_microsecond_physical_table(&distributed.frontend()).await;
+    }
+
+    /// Regression test for <https://github.com/GreptimeTeam/greptimedb/issues/9231>:
+    /// prometheus remote write/read against a physical metric table
+    /// pre-created with a TIMESTAMP(6) (microsecond) time index. Millisecond
+    /// samples are widened to the physical table's unit on write and narrowed
+    /// back to milliseconds on remote read.
+    async fn test_prom_store_remote_rw_microsecond_physical_table(instance: &Arc<Instance>) {
+        let db = "prometheus_us";
+        let ctx = Arc::new(QueryContext::with(DEFAULT_CATALOG_NAME, db));
+        assert!(
+            SqlQueryHandler::do_query(
+                instance.as_ref(),
+                &format!("CREATE DATABASE IF NOT EXISTS {db}"),
+                ctx.clone(),
+            )
+            .await
+            .first()
+            .unwrap()
+            .is_ok()
+        );
+
+        let mut output = instance
+            .do_query(
+                "CREATE TABLE greptime_physical_table (\
+                 greptime_timestamp TIMESTAMP(6) NOT NULL, \
+                 greptime_value DOUBLE NULL, \
+                 TIME INDEX (greptime_timestamp)) \
+                 ENGINE = metric WITH ('physical_metric_table' = 'true')",
+                ctx.clone(),
+            )
+            .await;
+        assert!(output.remove(0).is_ok());
+
+        let write_request = WriteRequest {
+            timeseries: vec![prom_store::mock_timeseries()[0].clone()],
+            ..Default::default()
+        };
+        let (row_inserts, _) = to_grpc_row_insert_requests(&write_request).unwrap();
+        instance
+            .write(row_inserts, ctx.clone(), true)
+            .await
+            .unwrap();
+
+        let read_request = ReadRequest {
+            queries: vec![Query {
+                start_timestamp_ms: 1000,
+                end_timestamp_ms: 2000,
+                matchers: vec![LabelMatcher {
+                    name: prom_store::METRIC_NAME_LABEL.to_string(),
+                    value: "metric1".to_string(),
+                    r#type: 0,
+                }],
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        let resp = instance.read(read_request, ctx.clone()).await.unwrap();
+        let body = prom_store::snappy_decompress(&resp.body).unwrap();
+        let read_response = ReadResponse::decode(&body[..]).unwrap();
+        assert_eq!(1, read_response.results.len());
+        assert_eq!(1, read_response.results[0].timeseries.len());
+        let timeseries = &read_response.results[0].timeseries[0];
+        assert_eq!(
+            timeseries.samples,
+            vec![
+                Sample {
+                    value: 1.0,
+                    timestamp: 1000,
+                },
+                Sample {
+                    value: 2.0,
+                    timestamp: 2000,
+                },
+            ]
+        );
+
+        // The stored timestamps keep the physical table's microsecond unit.
+        let mut output = instance
+            .do_query("SELECT greptime_timestamp FROM metric1", ctx.clone())
+            .await;
+        let OutputData::Stream(stream) = output.remove(0).unwrap().data else {
+            unreachable!()
+        };
+        let batches = common_recordbatch::RecordBatches::try_collect(stream)
+            .await
+            .unwrap()
+            .take();
+        assert_eq!(batches[0].num_rows(), 2);
+        let timestamps = batches[0]
+            .column(0)
+            .as_primitive::<TimestampMicrosecondType>();
+        // Millisecond samples are widened to the microsecond time index.
+        assert_eq!(timestamps.value(0), 1_000_000);
+        assert_eq!(timestamps.value(1), 2_000_000);
+    }
+
     async fn test_prom_store_remote_rw(instance: &Arc<Instance>, physical_table: Option<String>) {
         let write_request = WriteRequest {
             timeseries: prom_store::mock_timeseries(),