feat!: switch prom remote write to metric engine (#3198)

* feat: switch prom remote write to metric engine Signed-off-by: Ruihang Xia <waynestxia@gmail.com> * Apply suggestions from code review Co-authored-by: dennis zhuang <killme2008@gmail.com> * fix compile Signed-off-by: Ruihang Xia <waynestxia@gmail.com> * read physical table name from url Signed-off-by: Ruihang Xia <waynestxia@gmail.com> * remove physical table from header Signed-off-by: Ruihang Xia <waynestxia@gmail.com> * fix merge error Signed-off-by: Ruihang Xia <waynestxia@gmail.com> * fix format Signed-off-by: Ruihang Xia <waynestxia@gmail.com> * add with_metric_engine option to config remote write behavior Signed-off-by: Ruihang Xia <waynestxia@gmail.com> * check parameter Signed-off-by: Ruihang Xia <waynestxia@gmail.com> * add specific config param Signed-off-by: Ruihang Xia <waynestxia@gmail.com> * default with_metric_engine to true Signed-off-by: Ruihang Xia <waynestxia@gmail.com> * update UT Signed-off-by: Ruihang Xia <waynestxia@gmail.com> --------- Signed-off-by: Ruihang Xia <waynestxia@gmail.com> Co-authored-by: dennis zhuang <killme2008@gmail.com>
2026-01-07 05:42:57 +00:00 · 2024-01-22 22:48:28 +08:00
parent 6a12c27e78
commit 31787f4bfd
19 changed files with 380 additions and 53 deletions
--- a/config/frontend.example.toml
+++ b/config/frontend.example.toml
@@ -57,6 +57,9 @@ enable = true
 # Prometheus remote storage options, see `standalone.example.toml`.
 [prom_store]
 enable = true
+# Whether to store the data from Prometheus remote write in metric engine.
+# true by default
+with_metric_engine = true

 # Metasrv client options, see `datanode.example.toml`.
 [meta_client]
--- a/config/standalone.example.toml
+++ b/config/standalone.example.toml
@@ -81,6 +81,9 @@ enable = true
 [prom_store]
 # Whether to enable Prometheus remote write and read in HTTP API, true by default.
 enable = true
+# Whether to store the data from Prometheus remote write in metric engine.
+# true by default
+with_metric_engine = true

 [wal]
 # Available wal providers:
--- a/src/common/query/src/prelude.rs
+++ b/src/common/query/src/prelude.rs
@@ -25,3 +25,5 @@ pub const GREPTIME_TIMESTAMP: &str = "greptime_timestamp";
 pub const GREPTIME_VALUE: &str = "greptime_value";
 /// Default counter column name for OTLP metrics.
 pub const GREPTIME_COUNT: &str = "greptime_count";
+/// Default physical table name
+pub const GREPTIME_PHYSICAL_TABLE: &str = "greptime_physical_table";
--- a/src/frontend/src/instance/grpc.rs
+++ b/src/frontend/src/instance/grpc.rs
@@ -195,6 +195,18 @@ impl Instance {
            .context(TableOperationSnafu)
    }

+    pub async fn handle_metric_row_inserts(
+        &self,
+        requests: RowInsertRequests,
+        ctx: QueryContextRef,
+        physical_table: String,
+    ) -> Result<Output> {
+        self.inserter
+            .handle_metric_row_inserts(requests, ctx, &self.statement_executor, physical_table)
+            .await
+            .context(TableOperationSnafu)
+    }
+
    pub async fn handle_deletes(
        &self,
        requests: DeleteRequests,
--- a/src/frontend/src/instance/prom_store.rs
+++ b/src/frontend/src/instance/prom_store.rs
@@ -20,6 +20,7 @@ use async_trait::async_trait;
 use auth::{PermissionChecker, PermissionCheckerRef, PermissionReq};
 use common_catalog::format_full_table_name;
 use common_error::ext::BoxedError;
+use common_query::prelude::GREPTIME_PHYSICAL_TABLE;
 use common_query::Output;
 use common_recordbatch::RecordBatches;
 use common_telemetry::logging;
@@ -27,6 +28,7 @@ use operator::insert::InserterRef;
 use operator::statement::StatementExecutor;
 use prost::Message;
 use servers::error::{self, AuthSnafu, Result as ServerResult};
+use servers::http::prom_store::PHYSICAL_TABLE_PARAM;
 use servers::prom_store::{self, Metrics};
 use servers::query_handler::{
    PromStoreProtocolHandler, PromStoreProtocolHandlerRef, PromStoreResponse,
@@ -153,18 +155,36 @@ impl Instance {

 #[async_trait]
 impl PromStoreProtocolHandler for Instance {
-    async fn write(&self, request: WriteRequest, ctx: QueryContextRef) -> ServerResult<()> {
+    async fn write(
+        &self,
+        request: WriteRequest,
+        ctx: QueryContextRef,
+        with_metric_engine: bool,
+    ) -> ServerResult<()> {
        self.plugins
            .get::<PermissionCheckerRef>()
            .as_ref()
            .check_permission(ctx.current_user(), PermissionReq::PromStoreWrite)
            .context(AuthSnafu)?;
+
        let (requests, samples) = prom_store::to_grpc_row_insert_requests(request)?;
-        let _ = self
-            .handle_row_inserts(requests, ctx)
-            .await
-            .map_err(BoxedError::new)
-            .context(error::ExecuteGrpcQuerySnafu)?;
+        if with_metric_engine {
+            let physical_table = ctx
+                .extension(PHYSICAL_TABLE_PARAM)
+                .unwrap_or(GREPTIME_PHYSICAL_TABLE)
+                .to_string();
+            let _ = self
+                .handle_metric_row_inserts(requests, ctx.clone(), physical_table.to_string())
+                .await
+                .map_err(BoxedError::new)
+                .context(error::ExecuteGrpcQuerySnafu)?;
+        } else {
+            let _ = self
+                .handle_row_inserts(requests, ctx.clone())
+                .await
+                .map_err(BoxedError::new)
+                .context(error::ExecuteGrpcQuerySnafu)?;
+        }

        PROM_STORE_REMOTE_WRITE_SAMPLES.inc_by(samples as u64);
        Ok(())
@@ -239,10 +259,20 @@ impl ExportMetricHandler {

 #[async_trait]
 impl PromStoreProtocolHandler for ExportMetricHandler {
-    async fn write(&self, request: WriteRequest, ctx: QueryContextRef) -> ServerResult<()> {
+    async fn write(
+        &self,
+        request: WriteRequest,
+        ctx: QueryContextRef,
+        _: bool,
+    ) -> ServerResult<()> {
        let (requests, _) = prom_store::to_grpc_row_insert_requests(request)?;
        self.inserter
-            .handle_row_inserts(requests, ctx, self.statement_executor.as_ref())
+            .handle_metric_row_inserts(
+                requests,
+                ctx,
+                &self.statement_executor,
+                GREPTIME_PHYSICAL_TABLE.to_string(),
+            )
            .await
            .map_err(BoxedError::new)
            .context(error::ExecuteGrpcQuerySnafu)?;
--- a/src/frontend/src/server.rs
+++ b/src/frontend/src/server.rs
@@ -136,6 +136,8 @@ impl Services {
                let _ = http_server_builder
                    .with_prom_handler(instance.clone())
                    .with_prometheus_handler(instance.clone());
+                http_server_builder
+                    .set_prom_store_with_metric_engine(opts.prom_store.with_metric_engine);
            }

            if opts.otlp.enable {
--- a/src/frontend/src/service_config/prom_store.rs
+++ b/src/frontend/src/service_config/prom_store.rs
@@ -17,11 +17,15 @@ use serde::{Deserialize, Serialize};
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
 pub struct PromStoreOptions {
    pub enable: bool,
+    pub with_metric_engine: bool,
 }

 impl Default for PromStoreOptions {
    fn default() -> Self {
-        Self { enable: true }
+        Self {
+            enable: true,
+            with_metric_engine: true,
+        }
    }
 }

@@ -33,5 +37,6 @@ mod tests {
    fn test_prom_store_options() {
        let default = PromStoreOptions::default();
        assert!(default.enable);
+        assert!(default.with_metric_engine)
    }
 }
--- a/src/operator/src/insert.rs
+++ b/src/operator/src/insert.rs
@@ -18,13 +18,15 @@ use std::sync::Arc;
 use api::v1::alter_expr::Kind;
 use api::v1::region::{InsertRequests as RegionInsertRequests, RegionRequestHeader};
 use api::v1::{
-    AlterExpr, ColumnSchema, CreateTableExpr, InsertRequests, RowInsertRequest, RowInsertRequests,
+    AlterExpr, ColumnDataType, ColumnSchema, CreateTableExpr, InsertRequests, RowInsertRequest,
+    RowInsertRequests, SemanticType,
 };
 use catalog::CatalogManagerRef;
 use common_catalog::consts::default_engine;
 use common_grpc_expr::util::{extract_new_columns, ColumnExpr};
 use common_meta::datanode_manager::{AffectedRows, DatanodeManagerRef};
 use common_meta::peer::Peer;
+use common_query::prelude::{GREPTIME_TIMESTAMP, GREPTIME_VALUE};
 use common_query::Output;
 use common_telemetry::tracing_context::TracingContext;
 use common_telemetry::{error, info};
@@ -35,6 +37,9 @@ use partition::manager::PartitionRuleManagerRef;
 use session::context::QueryContextRef;
 use snafu::prelude::*;
 use sql::statements::insert::Insert;
+use store_api::metric_engine_consts::{
+    LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY,
+};
 use table::requests::InsertRequest as TableInsertRequest;
 use table::table_reference::TableReference;
 use table::TableRef;
@@ -95,7 +100,7 @@ impl Inserter {
        });
        validate_column_count_match(&requests)?;

-        self.create_or_alter_tables_on_demand(&requests, &ctx, statement_executor)
+        self.create_or_alter_tables_on_demand(&requests, &ctx, None, statement_executor)
            .await?;
        let inserts = RowToRegion::new(
            self.catalog_manager.as_ref(),
@@ -109,6 +114,44 @@ impl Inserter {
        Ok(Output::AffectedRows(affected_rows as _))
    }

+    /// Handle row inserts request with metric engine.
+    pub async fn handle_metric_row_inserts(
+        &self,
+        mut requests: RowInsertRequests,
+        ctx: QueryContextRef,
+        statement_executor: &StatementExecutor,
+        physical_table: String,
+    ) -> Result<Output> {
+        // remove empty requests
+        requests.inserts.retain(|req| {
+            req.rows
+                .as_ref()
+                .map(|r| !r.rows.is_empty())
+                .unwrap_or_default()
+        });
+        validate_column_count_match(&requests)?;
+
+        // check and create physical table
+        self.create_physical_table_on_demand(&ctx, physical_table.clone(), statement_executor)
+            .await?;
+
+        // check and create logical tables
+        self.create_or_alter_tables_on_demand(
+            &requests,
+            &ctx,
+            Some(physical_table.to_string()),
+            statement_executor,
+        )
+        .await?;
+        let inserts =
+            RowToRegion::new(self.catalog_manager.as_ref(), &self.partition_manager, &ctx)
+                .convert(requests)
+                .await?;
+
+        let affected_rows = self.do_request(inserts, &ctx).await?;
+        Ok(Output::AffectedRows(affected_rows as _))
+    }
+
    pub async fn handle_table_insert(
        &self,
        request: TableInsertRequest,
@@ -206,9 +249,10 @@ impl Inserter {
        &self,
        requests: &RowInsertRequests,
        ctx: &QueryContextRef,
+        on_physical_table: Option<String>,
        statement_executor: &StatementExecutor,
    ) -> Result<()> {
-        // TODO(jeremy): create and alter in batch?
+        // TODO(jeremy): create and alter in batch? (from `handle_metric_row_inserts`)
        for req in &requests.inserts {
            let catalog = ctx.current_catalog();
            let schema = ctx.current_schema();
@@ -219,13 +263,76 @@ impl Inserter {
                    self.alter_table_on_demand(req, table, ctx, statement_executor)
                        .await?
                }
-                None => self.create_table(req, ctx, statement_executor).await?,
+                None => {
+                    self.create_table(req, ctx, &on_physical_table, statement_executor)
+                        .await?
+                }
            }
        }

        Ok(())
    }

+    async fn create_physical_table_on_demand(
+        &self,
+        ctx: &QueryContextRef,
+        physical_table: String,
+        statement_executor: &StatementExecutor,
+    ) -> Result<()> {
+        let catalog_name = ctx.current_catalog();
+        let schema_name = ctx.current_schema();
+
+        // check if exist
+        if self
+            .get_table(catalog_name, schema_name, &physical_table)
+            .await?
+            .is_some()
+        {
+            return Ok(());
+        }
+
+        let table_reference = TableReference::full(catalog_name, schema_name, &physical_table);
+        info!("Physical metric table `{table_reference}` does not exist, try creating table");
+
+        // schema with timestamp and field column
+        let default_schema = vec![
+            ColumnSchema {
+                column_name: GREPTIME_TIMESTAMP.to_string(),
+                datatype: ColumnDataType::TimestampMillisecond as _,
+                semantic_type: SemanticType::Timestamp as _,
+                datatype_extension: None,
+            },
+            ColumnSchema {
+                column_name: GREPTIME_VALUE.to_string(),
+                datatype: ColumnDataType::Float64 as _,
+                semantic_type: SemanticType::Field as _,
+                datatype_extension: None,
+            },
+        ];
+        let create_table_expr = &mut build_create_table_expr(&table_reference, &default_schema)?;
+
+        create_table_expr.engine = METRIC_ENGINE_NAME.to_string();
+        create_table_expr
+            .table_options
+            .insert(PHYSICAL_TABLE_METADATA_KEY.to_string(), "true".to_string());
+
+        // create physical table
+        let res = statement_executor
+            .create_table_inner(create_table_expr, None)
+            .await;
+
+        match res {
+            Ok(_) => {
+                info!("Successfully created table {table_reference}",);
+                Ok(())
+            }
+            Err(err) => {
+                error!("Failed to create table {table_reference}: {err}",);
+                Err(err)
+            }
+        }
+    }
+
    async fn get_table(
        &self,
        catalog: &str,
@@ -289,10 +396,14 @@ impl Inserter {
        }
    }

+    /// Create a table with schema from insert request.
+    ///
+    /// To create a metric engine logical table, specify the `on_physical_table` parameter.
    async fn create_table(
        &self,
        req: &RowInsertRequest,
        ctx: &QueryContextRef,
+        on_physical_table: &Option<String>,
        statement_executor: &StatementExecutor,
    ) -> Result<()> {
        let table_ref =
@@ -301,10 +412,15 @@ impl Inserter {
        let request_schema = req.rows.as_ref().unwrap().schema.as_slice();
        let create_table_expr = &mut build_create_table_expr(&table_ref, request_schema)?;

-        info!(
-            "Table {}.{}.{} does not exist, try create table",
-            table_ref.catalog, table_ref.schema, table_ref.table,
-        );
+        if let Some(physical_table) = on_physical_table {
+            create_table_expr.engine = METRIC_ENGINE_NAME.to_string();
+            create_table_expr.table_options.insert(
+                LOGICAL_TABLE_METADATA_KEY.to_string(),
+                physical_table.clone(),
+            );
+        }
+
+        info!("Table `{table_ref}` does not exist, try creating table",);

        // TODO(weny): multiple regions table.
        let res = statement_executor
--- a/src/servers/src/error.rs
+++ b/src/servers/src/error.rs
@@ -430,6 +430,11 @@ pub enum Error {

    #[snafu(display("Missing query context"))]
    MissingQueryContext { location: Location },
+
+    #[snafu(display(
+        "Invalid parameter, physical_table is not expected when metric engine is disabled"
+    ))]
+    UnexpectedPhysicalTable { location: Location },
 }

 pub type Result<T> = std::result::Result<T, Error>;
@@ -488,7 +493,8 @@ impl ErrorExt for Error {
            | UrlDecode { .. }
            | IncompatibleSchema { .. }
            | MissingQueryContext { .. }
-            | MysqlValueConversion { .. } => StatusCode::InvalidArguments,
+            | MysqlValueConversion { .. }
+            | UnexpectedPhysicalTable { .. } => StatusCode::InvalidArguments,

            InfluxdbLinesWrite { source, .. }
            | PromSeriesWrite { source, .. }
--- a/src/servers/src/export_metrics.rs
+++ b/src/servers/src/export_metrics.rs
@@ -256,7 +256,7 @@ pub async fn write_system_metric_by_handler(
            filter.as_ref(),
            Timestamp::current_millis().value(),
        );
-        if let Err(e) = handler.write(request, ctx.clone()).await {
+        if let Err(e) = handler.write(request, ctx.clone(), false).await {
            error!("report export metrics by handler failed, error {}", e);
        }
    }
--- a/src/servers/src/http.rs
+++ b/src/servers/src/http.rs
@@ -99,9 +99,9 @@ pub static PUBLIC_APIS: [&str; 2] = ["/v1/influxdb/ping", "/v1/influxdb/health"]

 #[derive(Default)]
 pub struct HttpServer {
+    // server handlers
    sql_handler: Option<ServerSqlQueryHandlerRef>,
    grpc_handler: Option<ServerGrpcQueryHandlerRef>,
-    options: HttpOptions,
    influxdb_handler: Option<InfluxdbLineProtocolHandlerRef>,
    opentsdb_handler: Option<OpentsdbProtocolHandlerRef>,
    prom_handler: Option<PromStoreProtocolHandlerRef>,
@@ -111,8 +111,14 @@ pub struct HttpServer {
    shutdown_tx: Mutex<Option<Sender<()>>>,
    user_provider: Option<UserProviderRef>,
    metrics_handler: Option<MetricsHandler>,
-    greptime_config_options: Option<String>,
+
+    // plugins
    plugins: Plugins,
+
+    // server configs
+    options: HttpOptions,
+    greptime_config_options: Option<String>,
+    prom_store_with_metric_engine: bool,
 }

 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -421,6 +427,7 @@ impl HttpServerBuilder {
                shutdown_tx: Mutex::new(None),
                greptime_config_options: None,
                plugins: Default::default(),
+                prom_store_with_metric_engine: false,
            },
        }
    }
@@ -485,6 +492,11 @@ impl HttpServerBuilder {
        self
    }

+    pub fn set_prom_store_with_metric_engine(&mut self, with_metric_engine: bool) -> &mut Self {
+        self.inner.prom_store_with_metric_engine = with_metric_engine;
+        self
+    }
+
    pub fn build(&mut self) -> HttpServer {
        std::mem::take(self).inner
    }
@@ -674,10 +686,16 @@ impl HttpServer {
    }

    fn route_prom<S>(&self, prom_handler: PromStoreProtocolHandlerRef) -> Router<S> {
-        Router::new()
-            .route("/write", routing::post(prom_store::remote_write))
-            .route("/read", routing::post(prom_store::remote_read))
-            .with_state(prom_handler)
+        let mut router = Router::new().route("/read", routing::post(prom_store::remote_read));
+        if self.prom_store_with_metric_engine {
+            router = router.route("/write", routing::post(prom_store::remote_write));
+        } else {
+            router = router.route(
+                "/write",
+                routing::post(prom_store::route_write_without_metric_engine),
+            );
+        }
+        router.with_state(prom_handler)
    }

    fn route_influxdb<S>(&self, influxdb_handler: InfluxdbLineProtocolHandlerRef) -> Router<S> {
--- a/src/servers/src/http/authorize.rs
+++ b/src/servers/src/http/authorize.rs
@@ -60,12 +60,14 @@ pub async fn inner_auth<B>(
 ) -> std::result::Result<Request<B>, Response> {
    // 1. prepare
    let (catalog, schema) = extract_catalog_and_schema(&req);
-    let timezone = extract_timezone(&req);
-    let query_ctx = QueryContextBuilder::default()
+    // TODO(ruihang): move this out of auth module
+    let timezone = Arc::new(extract_timezone(&req));
+    let query_ctx_builder = QueryContextBuilder::default()
        .current_catalog(catalog.to_string())
        .current_schema(schema.to_string())
-        .timezone(Arc::new(timezone))
-        .build();
+        .timezone(timezone);
+
+    let query_ctx = query_ctx_builder.build();
    let need_auth = need_auth(&req);
    let is_influxdb = req.uri().path().contains("influxdb");

--- a/src/servers/src/http/header.rs
+++ b/src/servers/src/http/header.rs
@@ -17,7 +17,10 @@ use headers::{Header, HeaderName, HeaderValue};
 pub const GREPTIME_DB_HEADER_FORMAT: &str = "x-greptime-format";
 pub const GREPTIME_DB_HEADER_EXECUTION_TIME: &str = "x-greptime-execution-time";

+/// Header key of `db-name`. Example format of the header value is `greptime-public`.
 pub static GREPTIME_DB_HEADER_NAME: HeaderName = HeaderName::from_static("x-greptime-db-name");
+/// Header key of query specific timezone.
+/// Example format of the header value is `Asia/Shanghai` or `+08:00`.
 pub static GREPTIME_TIMEZONE_HEADER_NAME: HeaderName =
    HeaderName::from_static("x-greptime-timezone");

--- a/src/servers/src/http/prom_store.rs
+++ b/src/servers/src/http/prom_store.rs
@@ -12,12 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::sync::Arc;
+
 use api::prom_store::remote::{ReadRequest, WriteRequest};
 use axum::extract::{Query, RawBody, State};
 use axum::http::{header, StatusCode};
 use axum::response::IntoResponse;
 use axum::Extension;
 use common_catalog::consts::DEFAULT_SCHEMA_NAME;
+use common_query::prelude::GREPTIME_PHYSICAL_TABLE;
 use hyper::Body;
 use prost::Message;
 use schemars::JsonSchema;
@@ -25,25 +28,32 @@ use serde::{Deserialize, Serialize};
 use session::context::QueryContextRef;
 use snafu::prelude::*;

-use crate::error::{self, Result};
+use crate::error::{self, Result, UnexpectedPhysicalTableSnafu};
 use crate::prom_store::snappy_decompress;
 use crate::query_handler::{PromStoreProtocolHandlerRef, PromStoreResponse};

+pub const PHYSICAL_TABLE_PARAM: &str = "physical_table";
+
 #[derive(Debug, Serialize, Deserialize, JsonSchema)]
 pub struct DatabaseQuery {
    pub db: Option<String>,
+    /// Specify which physical table to use for storing metrics.
+    /// This only works on remote write requests.
+    pub physical_table: Option<String>,
 }

 impl Default for DatabaseQuery {
    fn default() -> DatabaseQuery {
        Self {
            db: Some(DEFAULT_SCHEMA_NAME.to_string()),
+            physical_table: Some(GREPTIME_PHYSICAL_TABLE.to_string()),
        }
    }
 }

+/// Same with [remote_write] but won't store data to metric engine.
 #[axum_macros::debug_handler]
-pub async fn remote_write(
+pub async fn route_write_without_metric_engine(
    State(handler): State<PromStoreProtocolHandlerRef>,
    Query(params): Query<DatabaseQuery>,
    Extension(query_ctx): Extension<QueryContextRef>,
@@ -51,12 +61,39 @@ pub async fn remote_write(
 ) -> Result<(StatusCode, ())> {
    let request = decode_remote_write_request(body).await?;
    let db = params.db.clone().unwrap_or_default();
+    let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_WRITE_ELAPSED
+        .with_label_values(&[db.as_str()])
+        .start_timer();
+
+    // reject if physical table is specified when metric engine is disabled
+    if params.physical_table.is_some() {
+        return UnexpectedPhysicalTableSnafu {}.fail();
+    }
+
+    handler.write(request, query_ctx, false).await?;
+    Ok((StatusCode::NO_CONTENT, ()))
+}
+
+#[axum_macros::debug_handler]
+pub async fn remote_write(
+    State(handler): State<PromStoreProtocolHandlerRef>,
+    Query(params): Query<DatabaseQuery>,
+    Extension(mut query_ctx): Extension<QueryContextRef>,
+    RawBody(body): RawBody,
+) -> Result<(StatusCode, ())> {
+    let request = decode_remote_write_request(body).await?;
+    let db = params.db.clone().unwrap_or_default();
+    if let Some(physical_table) = params.physical_table {
+        let mut new_query_ctx = query_ctx.as_ref().clone();
+        new_query_ctx.set_extension(PHYSICAL_TABLE_PARAM, physical_table);
+        query_ctx = Arc::new(new_query_ctx);
+    }

    let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_WRITE_ELAPSED
        .with_label_values(&[db.as_str()])
        .start_timer();

-    handler.write(request, query_ctx).await?;
+    handler.write(request, query_ctx, true).await?;
    Ok((StatusCode::NO_CONTENT, ()))
 }

--- a/src/servers/src/query_handler.rs
+++ b/src/servers/src/query_handler.rs
@@ -89,7 +89,12 @@ pub struct PromStoreResponse {
 #[async_trait]
 pub trait PromStoreProtocolHandler {
    /// Handling prometheus remote write requests
-    async fn write(&self, request: WriteRequest, ctx: QueryContextRef) -> Result<()>;
+    async fn write(
+        &self,
+        request: WriteRequest,
+        ctx: QueryContextRef,
+        with_metric_engine: bool,
+    ) -> Result<()>;
    /// Handling prometheus remote read requests
    async fn read(&self, request: ReadRequest, ctx: QueryContextRef) -> Result<PromStoreResponse>;
    /// Handling push gateway requests
--- a/src/servers/tests/http/prom_store_test.rs
+++ b/src/servers/tests/http/prom_store_test.rs
@@ -56,7 +56,7 @@ impl GrpcQueryHandler for DummyInstance {

 #[async_trait]
 impl PromStoreProtocolHandler for DummyInstance {
-    async fn write(&self, request: WriteRequest, ctx: QueryContextRef) -> Result<()> {
+    async fn write(&self, request: WriteRequest, ctx: QueryContextRef, _: bool) -> Result<()> {
        let _ = self
            .tx
            .send((ctx.current_schema().to_owned(), request.encode_to_vec()))
--- a/src/session/src/context.rs
+++ b/src/session/src/context.rs
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::collections::HashMap;
 use std::fmt::{Display, Formatter};
 use std::net::SocketAddr;
 use std::sync::Arc;
@@ -40,7 +41,9 @@ pub struct QueryContext {
    current_user: ArcSwap<Option<UserInfoRef>>,
    #[builder(setter(custom))]
    timezone: ArcSwap<Timezone>,
-    sql_dialect: Box<dyn Dialect + Send + Sync>,
+    sql_dialect: Arc<dyn Dialect + Send + Sync>,
+    #[builder(default)]
+    extension: HashMap<String, String>,
 }

 impl QueryContextBuilder {
@@ -61,6 +64,19 @@ impl Display for QueryContext {
    }
 }

+impl Clone for QueryContext {
+    fn clone(&self) -> Self {
+        Self {
+            current_catalog: self.current_catalog.clone(),
+            current_schema: self.current_schema.clone(),
+            current_user: self.current_user.load().clone().into(),
+            timezone: self.timezone.load().clone().into(),
+            sql_dialect: self.sql_dialect.clone(),
+            extension: self.extension.clone(),
+        }
+    }
+}
+
 impl From<&RegionRequestHeader> for QueryContext {
    fn from(value: &RegionRequestHeader) -> Self {
        let (catalog, schema) = parse_catalog_and_schema_from_db_string(&value.dbname);
@@ -70,7 +86,8 @@ impl From<&RegionRequestHeader> for QueryContext {
            current_user: Default::default(),
            // for request send to datanode, all timestamp have converted to UTC, so timezone is not important
            timezone: ArcSwap::new(Arc::new(get_timezone(None).clone())),
-            sql_dialect: Box::new(GreptimeDbDialect {}),
+            sql_dialect: Arc::new(GreptimeDbDialect {}),
+            extension: Default::default(),
        }
    }
 }
@@ -139,6 +156,14 @@ impl QueryContext {
        let _ = self.timezone.swap(Arc::new(timezone));
    }

+    pub fn set_extension<S1: Into<String>, S2: Into<String>>(&mut self, key: S1, value: S2) {
+        self.extension.insert(key.into(), value.into());
+    }
+
+    pub fn extension<S: AsRef<str>>(&self, key: S) -> Option<&str> {
+        self.extension.get(key.as_ref()).map(|v| v.as_str())
+    }
+
    /// SQL like `set variable` may change timezone or other info in `QueryContext`.
    /// We need persist these change in `Session`.
    pub fn update_session(&self, session: &SessionRef) {
@@ -166,9 +191,17 @@ impl QueryContextBuilder {
                .unwrap_or(ArcSwap::new(Arc::new(get_timezone(None).clone()))),
            sql_dialect: self
                .sql_dialect
-                .unwrap_or_else(|| Box::new(GreptimeDbDialect {})),
+                .unwrap_or_else(|| Arc::new(GreptimeDbDialect {})),
+            extension: self.extension.unwrap_or_default(),
        })
    }
+
+    pub fn set_extension(mut self, key: String, value: String) -> Self {
+        self.extension
+            .get_or_insert_with(HashMap::new)
+            .insert(key, value);
+        self
+    }
 }

 #[derive(Debug)]
@@ -207,10 +240,10 @@ pub enum Channel {
 }

 impl Channel {
-    pub fn dialect(&self) -> Box<dyn Dialect + Send + Sync> {
+    pub fn dialect(&self) -> Arc<dyn Dialect + Send + Sync> {
        match self {
-            Channel::Mysql => Box::new(MySqlDialect {}),
-            Channel::Postgres => Box::new(PostgreSqlDialect {}),
+            Channel::Mysql => Arc::new(MySqlDialect {}),
+            Channel::Postgres => Arc::new(PostgreSqlDialect {}),
        }
    }
 }
--- a/tests-integration/src/prom_store.rs
+++ b/tests-integration/src/prom_store.rs
@@ -23,6 +23,7 @@ mod tests {
    use common_catalog::consts::DEFAULT_CATALOG_NAME;
    use frontend::instance::Instance;
    use prost::Message;
+    use servers::http::prom_store::PHYSICAL_TABLE_PARAM;
    use servers::prom_store;
    use servers::query_handler::sql::SqlQueryHandler;
    use servers::query_handler::PromStoreProtocolHandler;
@@ -32,30 +33,69 @@ mod tests {
    use crate::tests;

    #[tokio::test(flavor = "multi_thread")]
-    async fn test_standalone_prom_store_remote_rw() {
-        let standalone = GreptimeDbStandaloneBuilder::new("test_standalone_prom_store_remote_rw")
-            .build()
-            .await;
+    async fn test_standalone_prom_store_remote_rw_default_physical_table() {
+        common_telemetry::init_default_ut_logging();
+        let standalone = GreptimeDbStandaloneBuilder::new(
+            "test_standalone_prom_store_remote_rw_default_physical_table",
+        )
+        .build()
+        .await;
        let instance = &standalone.instance;

-        test_prom_store_remote_rw(instance).await;
+        test_prom_store_remote_rw(instance, None).await;
    }

    #[tokio::test(flavor = "multi_thread")]
-    async fn test_distributed_prom_store_remote_rw() {
-        let distributed =
-            tests::create_distributed_instance("test_distributed_prom_store_remote_rw").await;
-        test_prom_store_remote_rw(&distributed.frontend()).await;
+    async fn test_distributed_prom_store_remote_rw_default_physical_table() {
+        common_telemetry::init_default_ut_logging();
+        let distributed = tests::create_distributed_instance(
+            "test_distributed_prom_store_remote_rw_default_physical_table",
+        )
+        .await;
+        test_prom_store_remote_rw(&distributed.frontend(), None).await;
    }

-    async fn test_prom_store_remote_rw(instance: &Arc<Instance>) {
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_standalone_prom_store_remote_rw_custom_physical_table() {
+        common_telemetry::init_default_ut_logging();
+        let standalone = GreptimeDbStandaloneBuilder::new(
+            "test_standalone_prom_store_remote_rw_custom_physical_table",
+        )
+        .build()
+        .await;
+        let instance = &standalone.instance;
+
+        test_prom_store_remote_rw(instance, Some("my_custom_physical_table".to_string())).await;
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_distributed_prom_store_remote_rw_custom_physical_table() {
+        common_telemetry::init_default_ut_logging();
+        let distributed = tests::create_distributed_instance(
+            "test_distributed_prom_store_remote_rw_custom_physical_table",
+        )
+        .await;
+        test_prom_store_remote_rw(
+            &distributed.frontend(),
+            Some("my_custom_physical_table".to_string()),
+        )
+        .await;
+    }
+
+    async fn test_prom_store_remote_rw(instance: &Arc<Instance>, physical_table: Option<String>) {
        let write_request = WriteRequest {
            timeseries: prom_store::mock_timeseries(),
            ..Default::default()
        };

        let db = "prometheus";
-        let ctx = QueryContext::with(DEFAULT_CATALOG_NAME, db);
+        let mut ctx = Arc::into_inner(QueryContext::with(DEFAULT_CATALOG_NAME, db)).unwrap();
+
+        // set physical table if provided
+        if let Some(physical_table) = &physical_table {
+            ctx.set_extension(PHYSICAL_TABLE_PARAM.to_string(), physical_table.clone());
+        }
+        let ctx = Arc::new(ctx);

        assert!(SqlQueryHandler::do_query(
            instance.as_ref(),
@@ -67,7 +107,10 @@ mod tests {
        .unwrap()
        .is_ok());

-        instance.write(write_request, ctx.clone()).await.unwrap();
+        instance
+            .write(write_request, ctx.clone(), true)
+            .await
+            .unwrap();

        let read_request = ReadRequest {
            queries: vec![
@@ -102,7 +145,7 @@ mod tests {
            ..Default::default()
        };

-        let resp = instance.read(read_request, ctx).await.unwrap();
+        let resp = instance.read(read_request, ctx.clone()).await.unwrap();
        assert_eq!(resp.content_type, "application/x-protobuf");
        assert_eq!(resp.content_encoding, "snappy");
        let body = prom_store::snappy_decompress(&resp.body).unwrap();
@@ -179,5 +222,11 @@ mod tests {
                }
            ]
        );
+
+        // check physical table if provided
+        if let Some(physical_table) = physical_table {
+            let sql = format!("DESC TABLE {physical_table};");
+            instance.do_query(&sql, ctx).await[0].as_ref().unwrap();
+        }
    }
 }
--- a/tests-integration/tests/http.rs
+++ b/tests-integration/tests/http.rs
@@ -706,6 +706,7 @@ enable = true

 [frontend.prom_store]
 enable = true
+with_metric_engine = true

 [frontend.otlp]
 enable = true