From 43a70fb4fa3462eeb760ba5b671c708e20a22aa4 Mon Sep 17 00:00:00 2001
From: Jiacai Liu <dev@liujiacai.net>
Date: Mon, 30 Jan 2023 16:38:44 +0800
Subject: [PATCH] feat: implement prom remote storage api (#578)

* feat: implement prom remote storage api

* translate req to sql

* remote read pass

* remote write pass

* add more docs

* sort labels before write

* add testcase for remote prom handler
---
 Cargo.lock                               | 117 +++--
 server/Cargo.toml                        |   1 +
 server/src/grpc/mod.rs                   |   2 +-
 server/src/grpc/storage_service/mod.rs   |   8 +-
 server/src/grpc/storage_service/write.rs | 248 +++++----
 server/src/handlers/mod.rs               |   1 +
 server/src/handlers/prom.rs              | 618 +++++++++++++++++++++++
 server/src/handlers/sql.rs               |  34 +-
 server/src/http.rs                       |  51 +-
 server/src/mysql/worker.rs               |   4 +-
 server/src/server.rs                     |   5 +
 11 files changed, 911 insertions(+), 178 deletions(-)
 create mode 100644 server/src/handlers/prom.rs

diff --git a/Cargo.lock b/Cargo.lock
index f1baf1bc66..2eb89531b8 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -95,7 +95,7 @@ dependencies = [
  "datafusion",
  "env_logger",
  "ethbloom",
- "futures 0.3.21",
+ "futures 0.3.25",
  "lazy_static",
  "log",
  "lru",
@@ -249,7 +249,7 @@ dependencies = [
  "chrono",
  "either",
  "fallible-streaming-iterator",
- "futures 0.3.21",
+ "futures 0.3.25",
  "hash_hasher",
  "num-traits",
  "parquet2",
@@ -298,9 +298,9 @@ dependencies = [
 
 [[package]]
 name = "async-trait"
-version = "0.1.57"
+version = "0.1.61"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "76464446b8bc32758d7e88ee1a804d9914cd9b1cb264c029899680b0be29826f"
+checksum = "705339e0e4a9690e2908d2b3d049d85682cf19fbd5782494498fbf7003a6a282"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -445,7 +445,7 @@ dependencies = [
  "common_util",
  "criterion",
  "env_logger",
- "futures 0.3.21",
+ "futures 0.3.25",
  "log",
  "object_store 1.0.0-alpha02",
  "parquet",
@@ -856,7 +856,7 @@ dependencies = [
  "ceresdbproto 0.1.0 (git+https://github.com/CeresDB/ceresdbproto.git?rev=12024e7f5c18cca7e40461f51b275c824dd37851)",
  "common_types 0.4.0",
  "dashmap 5.4.0",
- "futures 0.3.21",
+ "futures 0.3.25",
  "tokio 1.24.1",
  "tonic",
 ]
@@ -1527,7 +1527,7 @@ dependencies = [
  "datafusion-physical-expr",
  "datafusion-row",
  "datafusion-sql",
- "futures 0.3.21",
+ "futures 0.3.25",
  "glob",
  "hashbrown",
  "itertools",
@@ -2115,9 +2115,9 @@ checksum = "3a471a38ef8ed83cd6e40aa59c1ffe17db6855c18e3604d9c4ed8c08ebc28678"
 
 [[package]]
 name = "futures"
-version = "0.3.21"
+version = "0.3.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f73fe65f54d1e12b726f517d3e2135ca3125a437b6d998caf1962961f7172d9e"
+checksum = "38390104763dc37a5145a53c29c63c1290b5d316d6086ec32c293f6736051bb0"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -2130,9 +2130,9 @@ dependencies = [
 
 [[package]]
 name = "futures-channel"
-version = "0.3.24"
+version = "0.3.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "30bdd20c28fadd505d0fd6712cdfcb0d4b5648baf45faef7f852afb2399bb050"
+checksum = "52ba265a92256105f45b719605a571ffe2d1f0fea3807304b522c1d778f79eed"
 dependencies = [
  "futures-core",
  "futures-sink",
@@ -2140,9 +2140,9 @@ dependencies = [
 
 [[package]]
 name = "futures-core"
-version = "0.3.24"
+version = "0.3.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4e5aa3de05362c3fb88de6531e6296e85cde7739cccad4b9dfeeb7f6ebce56bf"
+checksum = "04909a7a7e4633ae6c4a9ab280aeb86da1236243a77b694a49eacd659a4bd3ac"
 
 [[package]]
 name = "futures-cpupool"
@@ -2156,9 +2156,9 @@ dependencies = [
 
 [[package]]
 name = "futures-executor"
-version = "0.3.21"
+version = "0.3.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9420b90cfa29e327d0429f19be13e7ddb68fa1cccb09d65e5706b8c7a749b8a6"
+checksum = "7acc85df6714c176ab5edf386123fafe217be88c0840ec11f199441134a074e2"
 dependencies = [
  "futures-core",
  "futures-task",
@@ -2167,15 +2167,15 @@ dependencies = [
 
 [[package]]
 name = "futures-io"
-version = "0.3.24"
+version = "0.3.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bbf4d2a7a308fd4578637c0b17c7e1c7ba127b8f6ba00b29f717e9655d85eb68"
+checksum = "00f5fb52a06bdcadeb54e8d3671f8888a39697dcb0b81b23b55174030427f4eb"
 
 [[package]]
 name = "futures-macro"
-version = "0.3.24"
+version = "0.3.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "42cd15d1c7456c04dbdf7e88bcd69760d74f3a798d6444e16974b505b0e62f17"
+checksum = "bdfb8ce053d86b91919aad980c220b1fb8401a9394410e1c289ed7e66b61835d"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2184,21 +2184,21 @@ dependencies = [
 
 [[package]]
 name = "futures-sink"
-version = "0.3.24"
+version = "0.3.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "21b20ba5a92e727ba30e72834706623d94ac93a725410b6a6b6fbc1b07f7ba56"
+checksum = "39c15cf1a4aa79df40f1bb462fb39676d0ad9e366c2a33b590d7c66f4f81fcf9"
 
 [[package]]
 name = "futures-task"
-version = "0.3.24"
+version = "0.3.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a6508c467c73851293f390476d4491cf4d227dbabcd4170f3bb6044959b294f1"
+checksum = "2ffb393ac5d9a6eaa9d3fdf37ae2776656b706e200c8e16b1bdb227f5198e6ea"
 
 [[package]]
 name = "futures-util"
-version = "0.3.24"
+version = "0.3.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "44fb6cb1be61cc1d2e43b262516aafcf63b241cffdb1d3fa115f91d9c7b09c90"
+checksum = "197676987abd2f9cadff84926f410af1c183608d36641465df73ae8211dc65d6"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -3180,7 +3180,7 @@ dependencies = [
  "async-trait",
  "chrono",
  "common_util",
- "futures 0.3.21",
+ "futures 0.3.25",
  "log",
  "rskafka",
  "serde",
@@ -3198,7 +3198,7 @@ dependencies = [
  "ceresdbproto 0.1.0 (git+https://github.com/CeresDB/ceresdbproto.git?rev=9430fe7d321955878ca06c431b115d73640f5d8f)",
  "common_types 1.0.0-alpha02",
  "common_util",
- "futures 0.3.21",
+ "futures 0.3.25",
  "log",
  "prost",
  "rand 0.7.3",
@@ -3673,7 +3673,7 @@ dependencies = [
  "async-trait",
  "bytes 1.2.1",
  "chrono",
- "futures 0.3.21",
+ "futures 0.3.25",
  "itertools",
  "parking_lot 0.12.1",
  "percent-encoding 2.1.0",
@@ -3694,7 +3694,7 @@ dependencies = [
  "clru",
  "common_util",
  "crc",
- "futures 0.3.21",
+ "futures 0.3.25",
  "lazy_static",
  "log",
  "lru",
@@ -3958,7 +3958,7 @@ dependencies = [
  "bytes 1.2.1",
  "chrono",
  "flate2",
- "futures 0.3.21",
+ "futures 0.3.25",
  "hashbrown",
  "lz4",
  "num",
@@ -3987,7 +3987,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e1a672c84c3e5b5eb6530286b2d22cc1ea8e1e3560e4c314218d6ab749c6db99"
 dependencies = [
  "async-trait",
- "futures 0.3.21",
+ "futures 0.3.25",
  "integer-encoding 3.0.4",
 ]
 
@@ -3999,7 +3999,7 @@ checksum = "73fd2690ad041f9296876daef1f2706f6347073bdbcc719090887f1691e4a09d"
 dependencies = [
  "async-stream",
  "bitpacking",
- "futures 0.3.21",
+ "futures 0.3.25",
  "parquet-format-async-temp",
  "streaming-decompression",
 ]
@@ -4263,6 +4263,20 @@ dependencies = [
  "tempfile",
 ]
 
+[[package]]
+name = "prom-remote-api"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbef7d8c450f6570e12606546fd941696561ed97a07a353e4131f4fc715da09f"
+dependencies = [
+ "async-trait",
+ "futures 0.3.25",
+ "prost",
+ "prost-build",
+ "snap",
+ "warp",
+]
+
 [[package]]
 name = "prometheus"
 version = "0.7.0"
@@ -4320,9 +4334,9 @@ dependencies = [
 
 [[package]]
 name = "prost"
-version = "0.11.0"
+version = "0.11.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "399c3c31cdec40583bb68f0b18403400d01ec4289c383aa047560439952c4dd7"
+checksum = "21dc42e00223fc37204bd4aa177e69420c604ca4a183209a8f9de30c6d934698"
 dependencies = [
  "bytes 1.2.1",
  "prost-derive",
@@ -4330,9 +4344,9 @@ dependencies = [
 
 [[package]]
 name = "prost-build"
-version = "0.11.1"
+version = "0.11.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7f835c582e6bd972ba8347313300219fed5bfa52caf175298d860b61ff6069bb"
+checksum = "a3f8ad728fb08fe212df3c05169e940fbb6d9d16a877ddde14644a983ba2012e"
 dependencies = [
  "bytes 1.2.1",
  "heck 0.4.0",
@@ -4341,18 +4355,20 @@ dependencies = [
  "log",
  "multimap",
  "petgraph",
+ "prettyplease",
  "prost",
  "prost-types",
  "regex",
+ "syn",
  "tempfile",
  "which",
 ]
 
 [[package]]
 name = "prost-derive"
-version = "0.11.0"
+version = "0.11.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7345d5f0e08c0536d7ac7229952590239e77abf0a0100a1b1d890add6ea96364"
+checksum = "8bda8c0881ea9f722eb9629376db3d0b903b462477c1aafcb0566610ac28ac5d"
 dependencies = [
  "anyhow",
  "itertools",
@@ -4363,9 +4379,9 @@ dependencies = [
 
 [[package]]
 name = "prost-types"
-version = "0.11.1"
+version = "0.11.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4dfaa718ad76a44b3415e6c4d53b17c8f99160dcb3a99b10470fce8ad43f6e3e"
+checksum = "a5e0526209433e96d83d750dd81a99118edbc55739e7e61a46764fd2ad537788"
 dependencies = [
  "bytes 1.2.1",
  "prost",
@@ -4468,7 +4484,7 @@ dependencies = [
  "datafusion",
  "datafusion-expr",
  "df_operator",
- "futures 0.3.21",
+ "futures 0.3.25",
  "log",
  "serde",
  "serde_derive",
@@ -4834,7 +4850,7 @@ dependencies = [
  "clru",
  "common_types 1.0.0-alpha02",
  "common_util",
- "futures 0.3.21",
+ "futures 0.3.25",
  "log",
  "proto 1.0.0-alpha02",
  "router",
@@ -4987,7 +5003,7 @@ dependencies = [
  "chrono",
  "crc32c",
  "flate2",
- "futures 0.3.21",
+ "futures 0.3.25",
  "integer-encoding 3.0.4",
  "lz4",
  "parking_lot 0.12.1",
@@ -5300,7 +5316,7 @@ dependencies = [
  "common_util",
  "datafusion",
  "df_operator",
- "futures 0.3.21",
+ "futures 0.3.25",
  "http 0.2.8",
  "interpreters",
  "lazy_static",
@@ -5310,6 +5326,7 @@ dependencies = [
  "opensrv-mysql",
  "paste 1.0.8",
  "profile",
+ "prom-remote-api",
  "prometheus 0.12.0",
  "prometheus-static-metric",
  "prost",
@@ -5591,9 +5608,9 @@ dependencies = [
 
 [[package]]
 name = "snap"
-version = "1.0.5"
+version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "45456094d1983e2ee2a18fdfebce3189fa451699d0502cb8e3b49dba5ba41451"
+checksum = "5e9f0ab6ef7eb7353d9119c170a436d1bf248eea575ac42d19d12f4e34130831"
 
 [[package]]
 name = "snappy-sys"
@@ -5874,7 +5891,7 @@ dependencies = [
  "catalog",
  "common_types 1.0.0-alpha02",
  "common_util",
- "futures 0.3.21",
+ "futures 0.3.25",
  "log",
  "prost",
  "proto 1.0.0-alpha02",
@@ -5897,7 +5914,7 @@ dependencies = [
  "datafusion-proto",
  "df_operator",
  "env_logger",
- "futures 0.3.21",
+ "futures 0.3.25",
  "itertools",
  "log",
  "parquet",
@@ -6445,7 +6462,7 @@ dependencies = [
  "common_types 1.0.0-alpha02",
  "common_util",
  "env_logger",
- "futures 0.3.21",
+ "futures 0.3.25",
  "object_store 1.0.0-alpha02",
  "parquet",
  "parquet_ext",
@@ -6851,7 +6868,7 @@ dependencies = [
  "common_types 1.0.0-alpha02",
  "common_util",
  "env_logger",
- "futures 0.3.21",
+ "futures 0.3.25",
  "log",
  "message_queue",
  "prost",
diff --git a/server/Cargo.toml b/server/Cargo.toml
index 9fad2b34c2..46f10689c0 100644
--- a/server/Cargo.toml
+++ b/server/Cargo.toml
@@ -33,6 +33,7 @@ meta_client = { workspace = true }
 opensrv-mysql = "0.1.0"
 paste = { workspace = true }
 profile = { workspace = true }
+prom-remote-api = { version = "0.2.1", features = ["warp"] }
 prometheus = { workspace = true }
 prometheus-static-metric = { workspace = true }
 prost = { workspace = true }
diff --git a/server/src/grpc/mod.rs b/server/src/grpc/mod.rs
index 2b0e250001..f94835f38a 100644
--- a/server/src/grpc/mod.rs
+++ b/server/src/grpc/mod.rs
@@ -47,7 +47,7 @@ pub mod forward;
 mod meta_event_service;
 mod metrics;
 mod remote_engine_service;
-mod storage_service;
+pub(crate) mod storage_service;
 
 #[derive(Debug, Snafu)]
 pub enum Error {
diff --git a/server/src/grpc/storage_service/mod.rs b/server/src/grpc/storage_service/mod.rs
index 336beeaadd..7c2175ffea 100644
--- a/server/src/grpc/storage_service/mod.rs
+++ b/server/src/grpc/storage_service/mod.rs
@@ -52,7 +52,7 @@ pub(crate) mod error;
 mod prom_query;
 mod query;
 mod route;
-mod write;
+pub(crate) mod write;
 
 const STREAM_QUERY_CHANNEL_LEN: usize = 20;
 
@@ -491,11 +491,11 @@ impl<Q: QueryExecutor + 'static> StorageService for StorageServiceImpl<Q> {
 
 /// Create CreateTablePlan from a write metric.
 // The caller must ENSURE that the HandlerContext's schema_config is not None.
-pub fn write_metric_to_create_table_plan<Q: QueryExecutor + 'static>(
-    ctx: &HandlerContext<Q>,
+pub fn write_metric_to_create_table_plan(
+    schema_config: Option<&SchemaConfig>,
     write_metric: &WriteMetric,
 ) -> Result<CreateTablePlan> {
-    let schema_config = ctx.schema_config.unwrap();
+    let schema_config = schema_config.unwrap();
     Ok(CreateTablePlan {
         engine: schema_config.default_engine_type.clone(),
         if_not_exists: true,
diff --git a/server/src/grpc/storage_service/write.rs b/server/src/grpc/storage_service/write.rs
index 976cfc30cc..fcc720a019 100644
--- a/server/src/grpc/storage_service/write.rs
+++ b/server/src/grpc/storage_service/write.rs
@@ -8,6 +8,7 @@ use std::{
 };
 
 use ceresdbproto::storage::{value, WriteEntry, WriteMetric, WriteRequest, WriteResponse};
+use cluster::config::SchemaConfig;
 use common_types::{
     bytes::Bytes,
     datum::{Datum, DatumKind},
@@ -24,10 +25,13 @@ use snafu::{ensure, OptionExt, ResultExt};
 use sql::plan::{InsertPlan, Plan};
 use table_engine::table::TableRef;
 
-use crate::grpc::storage_service::{
-    self,
-    error::{self, ErrNoCause, ErrWithCause, Result},
-    HandlerContext,
+use crate::{
+    grpc::storage_service::{
+        self,
+        error::{self, ErrNoCause, ErrWithCause, Result},
+        HandlerContext,
+    },
+    instance::InstanceRef,
 };
 
 pub(crate) async fn handle_write<Q: QueryExecutor + 'static>(
@@ -37,11 +41,12 @@ pub(crate) async fn handle_write<Q: QueryExecutor + 'static>(
     let request_id = RequestId::next_id();
     let begin_instant = Instant::now();
     let deadline = ctx.timeout.map(|t| begin_instant + t);
-
+    let catalog = ctx.catalog();
+    let schema = ctx.schema();
     debug!(
         "Grpc handle write begin, catalog:{}, schema:{}, request_id:{}, first_table:{:?}, num_tables:{}",
-        ctx.catalog(),
-        ctx.schema(),
+        catalog,
+        schema,
         request_id,
         req.metrics
             .first()
@@ -49,58 +54,28 @@ pub(crate) async fn handle_write<Q: QueryExecutor + 'static>(
         req.metrics.len(),
     );
 
-    let instance = &ctx.instance;
-    let plan_vec = write_request_to_insert_plan(ctx, req, request_id, deadline).await?;
+    let plan_vec = write_request_to_insert_plan(
+        request_id,
+        catalog,
+        schema,
+        ctx.instance.clone(),
+        req,
+        ctx.schema_config,
+        deadline,
+    )
+    .await?;
 
     let mut success = 0;
     for insert_plan in plan_vec {
-        debug!(
-            "Grpc handle write table begin, table:{}, row_num:{}",
-            insert_plan.table.name(),
-            insert_plan.rows.num_rows()
-        );
-        let plan = Plan::Insert(insert_plan);
-
-        ctx.instance
-            .limiter
-            .try_limit(&plan)
-            .map_err(|e| Box::new(e) as _)
-            .context(ErrWithCause {
-                code: StatusCode::FORBIDDEN,
-                msg: "Insert is blocked",
-            })?;
-
-        let interpreter_ctx = InterpreterContext::builder(request_id, deadline)
-            // Use current ctx's catalog and schema as default catalog and schema
-            .default_catalog_and_schema(ctx.catalog().to_string(), ctx.schema().to_string())
-            .build();
-        let interpreter_factory = Factory::new(
-            instance.query_executor.clone(),
-            instance.catalog_manager.clone(),
-            instance.table_engine.clone(),
-            instance.table_manipulator.clone(),
-        );
-        let interpreter = interpreter_factory
-            .create(interpreter_ctx, plan)
-            .map_err(|e| Box::new(e) as _)
-            .with_context(|| ErrWithCause {
-                code: StatusCode::INTERNAL_SERVER_ERROR,
-                msg: "Failed to create interpreter",
-            })?;
-
-        let row_num = match interpreter
-            .execute()
-            .await
-            .map_err(|e| Box::new(e) as _)
-            .context(ErrWithCause {
-                code: StatusCode::INTERNAL_SERVER_ERROR,
-                msg: "failed to execute interpreter",
-            })? {
-            Output::AffectedRows(n) => n,
-            _ => unreachable!(),
-        };
-
-        success += row_num;
+        success += execute_plan(
+            request_id,
+            catalog,
+            schema,
+            ctx.instance.clone(),
+            insert_plan,
+            deadline,
+        )
+        .await?;
     }
 
     let resp = WriteResponse {
@@ -111,32 +86,102 @@ pub(crate) async fn handle_write<Q: QueryExecutor + 'static>(
 
     debug!(
         "Grpc handle write finished, catalog:{}, schema:{}, resp:{:?}",
-        ctx.catalog(),
-        ctx.schema(),
-        resp
+        catalog, schema, resp
     );
 
     Ok(resp)
 }
 
-async fn write_request_to_insert_plan<Q: QueryExecutor + 'static>(
-    ctx: &HandlerContext<'_, Q>,
-    write_request: WriteRequest,
+pub async fn execute_plan<Q: QueryExecutor + 'static>(
+    request_id: RequestId,
+    catalog: &str,
+    schema: &str,
+    instance: InstanceRef<Q>,
+    insert_plan: InsertPlan,
+    deadline: Option<Instant>,
+) -> Result<usize> {
+    debug!(
+        "Grpc handle write table begin, table:{}, row_num:{}",
+        insert_plan.table.name(),
+        insert_plan.rows.num_rows()
+    );
+    let plan = Plan::Insert(insert_plan);
+
+    instance
+        .limiter
+        .try_limit(&plan)
+        .map_err(|e| Box::new(e) as _)
+        .context(ErrWithCause {
+            code: StatusCode::FORBIDDEN,
+            msg: "Insert is blocked",
+        })?;
+
+    let interpreter_ctx = InterpreterContext::builder(request_id, deadline)
+        // Use current ctx's catalog and schema as default catalog and schema
+        .default_catalog_and_schema(catalog.to_string(), schema.to_string())
+        .build();
+    let interpreter_factory = Factory::new(
+        instance.query_executor.clone(),
+        instance.catalog_manager.clone(),
+        instance.table_engine.clone(),
+        instance.table_manipulator.clone(),
+    );
+    let interpreter = interpreter_factory
+        .create(interpreter_ctx, plan)
+        .map_err(|e| Box::new(e) as _)
+        .with_context(|| ErrWithCause {
+            code: StatusCode::INTERNAL_SERVER_ERROR,
+            msg: "Failed to create interpreter",
+        })?;
+
+    interpreter
+        .execute()
+        .await
+        .map_err(|e| Box::new(e) as _)
+        .context(ErrWithCause {
+            code: StatusCode::INTERNAL_SERVER_ERROR,
+            msg: "failed to execute interpreter",
+        })
+        .and_then(|output| match output {
+            Output::AffectedRows(n) => Ok(n),
+            Output::Records(_) => ErrNoCause {
+                code: StatusCode::INTERNAL_SERVER_ERROR,
+                msg: "Invalid output type, expect AffectedRows, found Records",
+            }
+            .fail(),
+        })
+}
+
+pub async fn write_request_to_insert_plan<Q: QueryExecutor + 'static>(
     request_id: RequestId,
+    catalog: &str,
+    schema: &str,
+    instance: InstanceRef<Q>,
+    write_request: WriteRequest,
+    schema_config: Option<&SchemaConfig>,
     deadline: Option<Instant>,
 ) -> Result<Vec<InsertPlan>> {
     let mut plan_vec = Vec::with_capacity(write_request.metrics.len());
 
     for write_metric in write_request.metrics {
         let table_name = &write_metric.metric;
-        let mut table = try_get_table(ctx, table_name)?;
+        let mut table = try_get_table(catalog, schema, instance.clone(), table_name)?;
 
         if table.is_none() {
-            if let Some(config) = ctx.schema_config {
+            if let Some(config) = schema_config {
                 if config.auto_create_tables {
-                    create_table(ctx, &write_metric, request_id, deadline).await?;
+                    create_table(
+                        request_id,
+                        catalog,
+                        schema,
+                        instance.clone(),
+                        &write_metric,
+                        schema_config,
+                        deadline,
+                    )
+                    .await?;
                     // try to get table again
-                    table = try_get_table(ctx, table_name)?;
+                    table = try_get_table(catalog, schema, instance.clone(), table_name)?;
                 }
             }
         }
@@ -149,11 +194,7 @@ async fn write_request_to_insert_plan<Q: QueryExecutor + 'static>(
             None => {
                 return ErrNoCause {
                     code: StatusCode::BAD_REQUEST,
-                    msg: format!(
-                        "Table not found, schema:{}, table:{}",
-                        ctx.schema(),
-                        table_name
-                    ),
+                    msg: format!("Table not found, schema:{}, table:{}", schema, table_name),
                 }
                 .fail();
             }
@@ -164,30 +205,32 @@ async fn write_request_to_insert_plan<Q: QueryExecutor + 'static>(
 }
 
 fn try_get_table<Q: QueryExecutor + 'static>(
-    ctx: &HandlerContext<'_, Q>,
+    catalog: &str,
+    schema: &str,
+    instance: InstanceRef<Q>,
     table_name: &str,
 ) -> Result<Option<TableRef>> {
-    ctx.instance
+    instance
         .catalog_manager
-        .catalog_by_name(ctx.catalog())
+        .catalog_by_name(catalog)
         .map_err(|e| Box::new(e) as _)
         .with_context(|| ErrWithCause {
             code: StatusCode::INTERNAL_SERVER_ERROR,
-            msg: format!("Failed to find catalog, catalog_name:{}", ctx.catalog()),
+            msg: format!("Failed to find catalog, catalog_name:{}", catalog),
         })?
         .with_context(|| ErrNoCause {
             code: StatusCode::BAD_REQUEST,
-            msg: format!("Catalog not found, catalog_name:{}", ctx.catalog()),
+            msg: format!("Catalog not found, catalog_name:{}", catalog),
         })?
-        .schema_by_name(ctx.schema())
+        .schema_by_name(schema)
         .map_err(|e| Box::new(e) as _)
         .with_context(|| ErrWithCause {
             code: StatusCode::INTERNAL_SERVER_ERROR,
-            msg: format!("Failed to find schema, schema_name:{}", ctx.schema()),
+            msg: format!("Failed to find schema, schema_name:{}", schema),
         })?
         .with_context(|| ErrNoCause {
             code: StatusCode::BAD_REQUEST,
-            msg: format!("Schema not found, schema_name:{}", ctx.schema()),
+            msg: format!("Schema not found, schema_name:{}", schema),
         })?
         .table_by_name(table_name)
         .map_err(|e| Box::new(e) as _)
@@ -198,20 +241,24 @@ fn try_get_table<Q: QueryExecutor + 'static>(
 }
 
 async fn create_table<Q: QueryExecutor + 'static>(
-    ctx: &HandlerContext<'_, Q>,
-    write_metric: &WriteMetric,
     request_id: RequestId,
+    catalog: &str,
+    schema: &str,
+    instance: InstanceRef<Q>,
+    write_metric: &WriteMetric,
+    schema_config: Option<&SchemaConfig>,
     deadline: Option<Instant>,
 ) -> Result<()> {
-    let create_table_plan = storage_service::write_metric_to_create_table_plan(ctx, write_metric)
-        .map_err(|e| Box::new(e) as _)
-        .with_context(|| ErrWithCause {
-            code: StatusCode::INTERNAL_SERVER_ERROR,
-            msg: format!(
-                "Failed to build creating table plan from metric, table:{}",
-                write_metric.metric
-            ),
-        })?;
+    let create_table_plan =
+        storage_service::write_metric_to_create_table_plan(schema_config, write_metric)
+            .map_err(|e| Box::new(e) as _)
+            .with_context(|| ErrWithCause {
+                code: StatusCode::INTERNAL_SERVER_ERROR,
+                msg: format!(
+                    "Failed to build creating table plan from metric, table:{}",
+                    write_metric.metric
+                ),
+            })?;
 
     debug!(
         "Grpc handle create table begin, table:{}, schema:{:?}",
@@ -219,8 +266,6 @@ async fn create_table<Q: QueryExecutor + 'static>(
     );
     let plan = Plan::Create(create_table_plan);
 
-    let instance = &ctx.instance;
-
     instance
         .limiter
         .try_limit(&plan)
@@ -232,7 +277,7 @@ async fn create_table<Q: QueryExecutor + 'static>(
 
     let interpreter_ctx = InterpreterContext::builder(request_id, deadline)
         // Use current ctx's catalog and schema as default catalog and schema
-        .default_catalog_and_schema(ctx.catalog().to_string(), ctx.schema().to_string())
+        .default_catalog_and_schema(catalog.to_string(), schema.to_string())
         .build();
     let interpreter_factory = Factory::new(
         instance.query_executor.clone(),
@@ -248,19 +293,22 @@ async fn create_table<Q: QueryExecutor + 'static>(
             msg: "Failed to create interpreter",
         })?;
 
-    let _ = match interpreter
+    interpreter
         .execute()
         .await
         .map_err(|e| Box::new(e) as _)
         .context(ErrWithCause {
             code: StatusCode::INTERNAL_SERVER_ERROR,
             msg: "failed to execute interpreter",
-        })? {
-        Output::AffectedRows(n) => n,
-        _ => unreachable!(),
-    };
-
-    Ok(())
+        })
+        .and_then(|output| match output {
+            Output::AffectedRows(_) => Ok(()),
+            Output::Records(_) => ErrNoCause {
+                code: StatusCode::INTERNAL_SERVER_ERROR,
+                msg: "Invalid output type, expect AffectedRows, found Records",
+            }
+            .fail(),
+        })
 }
 
 fn write_metric_to_insert_plan(table: TableRef, write_metric: WriteMetric) -> Result<InsertPlan> {
diff --git a/server/src/handlers/mod.rs b/server/src/handlers/mod.rs
index e695b3b610..4735096514 100644
--- a/server/src/handlers/mod.rs
+++ b/server/src/handlers/mod.rs
@@ -4,6 +4,7 @@
 
 pub mod admin;
 pub mod error;
+pub mod prom;
 pub mod sql;
 
 mod prelude {
diff --git a/server/src/handlers/prom.rs b/server/src/handlers/prom.rs
new file mode 100644
index 0000000000..9d873c7c0a
--- /dev/null
+++ b/server/src/handlers/prom.rs
@@ -0,0 +1,618 @@
+// Copyright 2022 CeresDB Project Authors. Licensed under Apache-2.0.
+
+//! This module implements prometheus remote storage API.
+//! It converts write request to gRPC write request, and
+//! translates query request to SQL for execution.
+
+use std::{collections::HashMap, time::Instant};
+
+use async_trait::async_trait;
+use ceresdbproto::storage::{
+    value, Field, FieldGroup, Tag, Value, WriteEntry, WriteMetric, WriteRequest as WriteRequestPb,
+};
+use common_types::{
+    datum::DatumKind,
+    request_id::RequestId,
+    schema::{RecordSchema, TIMESTAMP_COLUMN, TSID_COLUMN},
+};
+use interpreters::interpreter::Output;
+use log::debug;
+use prom_remote_api::types::{
+    label_matcher, Label, LabelMatcher, Query, QueryResult, RemoteStorage, Sample, TimeSeries,
+    WriteRequest,
+};
+use query_engine::executor::{Executor as QueryExecutor, RecordBatchVec};
+use snafu::{ensure, Backtrace, OptionExt, ResultExt, Snafu};
+use warp::reject;
+
+use crate::{
+    context::RequestContext, handlers, instance::InstanceRef,
+    schema_config_provider::SchemaConfigProviderRef,
+};
+
+#[derive(Debug, Snafu)]
+pub enum Error {
+    #[snafu(display("Metric name is not found.\nBacktrace:\n{}", backtrace))]
+    MissingName { backtrace: Backtrace },
+
+    #[snafu(display("Invalid matcher type, value:{}.\nBacktrace:\n{}", value, backtrace))]
+    InvalidMatcherType { value: i32, backtrace: Backtrace },
+
+    #[snafu(display("Read response must be Rows.\nBacktrace:\n{}", backtrace))]
+    ResponseMustRows { backtrace: Backtrace },
+
+    #[snafu(display("TSID column is missing in query response.\nBacktrace:\n{}", backtrace))]
+    MissingTSID { backtrace: Backtrace },
+
+    #[snafu(display(
+        "Timestamp column is missing in query response.\nBacktrace:\n{}",
+        backtrace
+    ))]
+    MissingTimestamp { backtrace: Backtrace },
+
+    #[snafu(display(
+        "Value column is missing in query response.\nBacktrace:\n{}",
+        backtrace
+    ))]
+    MissingValue { backtrace: Backtrace },
+
+    #[snafu(display("Handle sql failed, err:{}.", source))]
+    SqlHandle {
+        source: Box<crate::handlers::error::Error>,
+    },
+
+    #[snafu(display("Tsid must be u64, current:{}.\nBacktrace:\n{}", kind, backtrace))]
+    TsidMustU64 {
+        kind: DatumKind,
+        backtrace: Backtrace,
+    },
+
+    #[snafu(display("Timestamp wrong type, current:{}.\nBacktrace:\n{}", kind, backtrace))]
+    MustTimestamp {
+        kind: DatumKind,
+        backtrace: Backtrace,
+    },
+
+    #[snafu(display(
+        "Value must be f64 compatible type, current:{}.\nBacktrace:\n{}",
+        kind,
+        backtrace
+    ))]
+    F64Castable {
+        kind: DatumKind,
+        backtrace: Backtrace,
+    },
+
+    #[snafu(display(
+        "Tag must be string type, current:{}.\nBacktrace:\n{}",
+        kind,
+        backtrace
+    ))]
+    TagMustString {
+        kind: DatumKind,
+        backtrace: Backtrace,
+    },
+
+    #[snafu(display("Failed to write via gRPC, source:{}.", source))]
+    GRPCWriteError {
+        source: crate::grpc::storage_service::error::Error,
+    },
+
+    #[snafu(display("Failed to get schema, source:{}.", source))]
+    SchemaError {
+        source: crate::schema_config_provider::Error,
+    },
+}
+
+define_result!(Error);
+
+impl reject::Reject for Error {}
+
+const NAME_LABEL: &str = "__name__";
+const VALUE_COLUMN: &str = "value";
+
+pub struct CeresDBStorage<Q> {
+    instance: InstanceRef<Q>,
+    schema_config_provider: SchemaConfigProviderRef,
+}
+
+impl<Q: QueryExecutor + 'static> CeresDBStorage<Q> {
+    pub fn new(instance: InstanceRef<Q>, schema_config_provider: SchemaConfigProviderRef) -> Self {
+        Self {
+            instance,
+            schema_config_provider,
+        }
+    }
+}
+
+impl<Q> CeresDBStorage<Q> {
+    /// Separate metric from labels, and sort labels by name.
+    fn normalize_labels(mut labels: Vec<Label>) -> Result<(String, Vec<Label>)> {
+        let metric_idx = labels
+            .iter()
+            .position(|label| label.name == NAME_LABEL)
+            .context(MissingName)?;
+        let metric = labels.swap_remove(metric_idx).value;
+        labels.sort_unstable_by(|a, b| a.name.cmp(&b.name));
+
+        Ok((metric, labels))
+    }
+
+    /// Separate metric from matchers, and convert remaining matchers to sql
+    /// filters.
+    fn normalize_matchers(mut matchers: Vec<LabelMatcher>) -> Result<(String, Vec<String>)> {
+        let metric_idx = matchers
+            .iter()
+            .position(|m| m.name == NAME_LABEL)
+            .context(MissingName)?;
+
+        let metric = matchers.swap_remove(metric_idx).value;
+        let filters = matchers
+            .iter()
+            .map(|m| match m.r#type() {
+                label_matcher::Type::Eq => format!("{} = '{}'", m.name, m.value),
+                label_matcher::Type::Neq => format!("{} != '{}'", m.name, m.value),
+                // https://github.com/prometheus/prometheus/blob/2ce94ac19673a3f7faf164e9e078a79d4d52b767/model/labels/regexp.go#L29
+                label_matcher::Type::Re => format!("{} ~ '^(?:{})'", m.name, m.value),
+                label_matcher::Type::Nre => format!("{} !~ '^(?:{})'", m.name, m.value),
+            })
+            .collect();
+
+        Ok((metric, filters))
+    }
+
+    fn convert_write_request(req: WriteRequest) -> Result<WriteRequestPb> {
+        let mut req_by_metric = HashMap::new();
+        for timeseries in req.timeseries {
+            let (metric, labels) = Self::normalize_labels(timeseries.labels)?;
+            let (tag_names, tag_values): (Vec<_>, Vec<_>) = labels
+                .into_iter()
+                .map(|label| (label.name, label.value))
+                .unzip();
+
+            req_by_metric
+                .entry(metric.to_string())
+                .or_insert_with(|| WriteMetric {
+                    metric,
+                    tag_names,
+                    field_names: vec![VALUE_COLUMN.to_string()],
+                    entries: Vec::new(),
+                })
+                .entries
+                .push(WriteEntry {
+                    tags: tag_values
+                        .into_iter()
+                        .enumerate()
+                        .map(|(idx, v)| Tag {
+                            name_index: idx as u32,
+                            value: Some(Value {
+                                value: Some(value::Value::StringValue(v)),
+                            }),
+                        })
+                        .collect(),
+                    field_groups: timeseries
+                        .samples
+                        .into_iter()
+                        .map(|sample| FieldGroup {
+                            timestamp: sample.timestamp,
+                            fields: vec![Field {
+                                name_index: 0,
+                                value: Some(Value {
+                                    value: Some(value::Value::Float64Value(sample.value)),
+                                }),
+                            }],
+                        })
+                        .collect(),
+                });
+        }
+
+        Ok(WriteRequestPb {
+            metrics: req_by_metric.into_values().collect(),
+        })
+    }
+
+    fn convert_query_result(metric: String, resp: Output) -> Result<QueryResult> {
+        let record_batches = match resp {
+            Output::AffectedRows(_) => return ResponseMustRows {}.fail(),
+            Output::Records(v) => v,
+        };
+
+        let converter = match record_batches.first() {
+            None => {
+                return Ok(QueryResult::default());
+            }
+            Some(batch) => Converter::try_new(batch.schema())?,
+        };
+
+        converter.convert(metric, record_batches)
+    }
+}
+
+#[async_trait]
+impl<Q: QueryExecutor + 'static> RemoteStorage for CeresDBStorage<Q> {
+    type Context = RequestContext;
+    type Err = Error;
+
+    /// Write samples to remote storage
+    async fn write(&self, ctx: Self::Context, req: WriteRequest) -> Result<()> {
+        let request_id = RequestId::next_id();
+        let deadline = ctx.timeout.map(|t| Instant::now() + t);
+        let catalog = &ctx.catalog;
+        self.instance.catalog_manager.default_catalog_name();
+        let schema = &ctx.schema;
+        let schema_config = self
+            .schema_config_provider
+            .schema_config(schema)
+            .context(SchemaError)?;
+        let plans = crate::grpc::storage_service::write::write_request_to_insert_plan(
+            request_id,
+            catalog,
+            schema,
+            self.instance.clone(),
+            Self::convert_write_request(req)?,
+            schema_config,
+            deadline,
+        )
+        .await
+        .context(GRPCWriteError)?;
+
+        let mut success = 0;
+        for insert_plan in plans {
+            success += crate::grpc::storage_service::write::execute_plan(
+                request_id,
+                catalog,
+                schema,
+                self.instance.clone(),
+                insert_plan,
+                deadline,
+            )
+            .await
+            .context(GRPCWriteError)?;
+        }
+        debug!(
+            "Remote write finished, catalog:{}, schema:{}, success:{}",
+            catalog, schema, success
+        );
+
+        Ok(())
+    }
+
+    /// Process one query within ReadRequest.
+    async fn process_query(&self, ctx: &Self::Context, q: Query) -> Result<QueryResult> {
+        let (metric, mut filters) = Self::normalize_matchers(q.matchers)?;
+        filters.push(format!(
+            "{} between {} AND {}",
+            TIMESTAMP_COLUMN, q.start_timestamp_ms, q.end_timestamp_ms
+        ));
+
+        let sql = format!(
+            "select * from {} where {} order by {}, {}",
+            metric,
+            filters.join(" and "),
+            TSID_COLUMN,
+            TIMESTAMP_COLUMN
+        );
+
+        let result = handlers::sql::handle_sql(ctx, self.instance.clone(), sql.into())
+            .await
+            .map_err(Box::new)
+            .context(SqlHandle)?;
+
+        Self::convert_query_result(metric, result)
+    }
+}
+
+/// Converter converts Arrow's RecordBatch into Prometheus's QueryResult
+struct Converter {
+    tsid_idx: usize,
+    timestamp_idx: usize,
+    value_idx: usize,
+    // (column_name, index)
+    tags: Vec<(String, usize)>,
+}
+
+impl Converter {
+    fn try_new(schema: &RecordSchema) -> Result<Self> {
+        let tsid_idx = schema.index_of(TSID_COLUMN).context(MissingTSID)?;
+        let timestamp_idx = schema
+            .index_of(TIMESTAMP_COLUMN)
+            .context(MissingTimestamp)?;
+        let value_idx = schema.index_of(VALUE_COLUMN).context(MissingValue)?;
+        let tags = schema
+            .columns()
+            .iter()
+            .enumerate()
+            .filter(|(_, col)| col.is_tag)
+            .map(|(i, col)| {
+                ensure!(
+                    matches!(col.data_type, DatumKind::String),
+                    TagMustString {
+                        kind: col.data_type
+                    }
+                );
+
+                Ok((col.name.to_string(), i))
+            })
+            .collect::<Result<Vec<_>>>()?;
+
+        ensure!(
+            matches!(schema.column(tsid_idx).data_type, DatumKind::UInt64),
+            TsidMustU64 {
+                kind: schema.column(tsid_idx).data_type
+            }
+        );
+        ensure!(
+            schema.column(timestamp_idx).data_type.is_timestamp(),
+            MustTimestamp {
+                kind: schema.column(timestamp_idx).data_type
+            }
+        );
+        ensure!(
+            schema.column(value_idx).data_type.is_f64_castable(),
+            F64Castable {
+                kind: schema.column(value_idx).data_type
+            }
+        );
+
+        Ok(Converter {
+            tsid_idx,
+            timestamp_idx,
+            value_idx,
+            tags,
+        })
+    }
+
+    fn convert(&self, metric: String, record_batches: RecordBatchVec) -> Result<QueryResult> {
+        let mut series_by_tsid = HashMap::new();
+        for batch in record_batches {
+            let tsid_col = batch.column(self.tsid_idx);
+            let timestamp_col = batch.column(self.timestamp_idx);
+            let value_col = batch.column(self.value_idx);
+            let tag_cols = self
+                .tags
+                .iter()
+                .map(|(_, idx)| batch.column(*idx))
+                .collect::<Vec<_>>();
+            for row_idx in 0..batch.num_rows() {
+                let tsid = tsid_col
+                    .datum(row_idx)
+                    .as_u64()
+                    .expect("checked in try_new");
+                series_by_tsid
+                    .entry(tsid)
+                    .or_insert_with(|| {
+                        let mut labels = self
+                            .tags
+                            .iter()
+                            .enumerate()
+                            .map(|(idx, (col_name, _))| {
+                                let col_value = tag_cols[idx].datum(row_idx);
+                                let col_value = col_value.as_str().expect("checked in try_new");
+                                Label {
+                                    name: col_name.to_string(),
+                                    value: col_value.to_string(),
+                                }
+                            })
+                            .collect::<Vec<_>>();
+                        labels.push(Label {
+                            name: NAME_LABEL.to_string(),
+                            value: metric.clone(),
+                        });
+
+                        TimeSeries {
+                            labels,
+                            ..Default::default()
+                        }
+                    })
+                    .samples
+                    .push(Sample {
+                        timestamp: timestamp_col
+                            .datum(row_idx)
+                            .as_timestamp()
+                            .expect("checked in try_new")
+                            .as_i64(),
+                        value: value_col
+                            .datum(row_idx)
+                            .as_f64()
+                            .expect("checked in try_new"),
+                    });
+            }
+        }
+
+        Ok(QueryResult {
+            timeseries: series_by_tsid.into_values().collect(),
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use arrow::{
+        array::{ArrayRef, Float64Array, StringArray, TimestampMillisecondArray, UInt64Array},
+        record_batch::RecordBatch as ArrowRecordBatch,
+    };
+    use common_types::{column_schema, record_batch::RecordBatch, schema};
+    use prom_remote_api::types::{label_matcher::Type, Label};
+
+    use super::*;
+
+    fn make_labels(tuples: Vec<(&str, &str)>) -> Vec<Label> {
+        tuples
+            .into_iter()
+            .map(|(name, value)| Label {
+                name: name.to_string(),
+                value: value.to_string(),
+            })
+            .collect()
+    }
+
+    fn make_matchers(tuples: Vec<(&str, &str, Type)>) -> Vec<LabelMatcher> {
+        tuples
+            .into_iter()
+            .map(|(name, value, matcher_type)| LabelMatcher {
+                name: name.to_string(),
+                value: value.to_string(),
+                r#type: matcher_type as i32,
+            })
+            .collect()
+    }
+
+    fn make_samples(tuples: Vec<(i64, f64)>) -> Vec<Sample> {
+        tuples
+            .into_iter()
+            .map(|(timestamp, value)| Sample { timestamp, value })
+            .collect()
+    }
+
+    #[test]
+    fn test_normailze_labels() {
+        let labels = make_labels(vec![
+            ("aa", "va"),
+            ("zz", "vz"),
+            (NAME_LABEL, "cpu"),
+            ("yy", "vy"),
+        ]);
+
+        let (metric, labels) = CeresDBStorage::<()>::normalize_labels(labels).unwrap();
+        assert_eq!("cpu", metric);
+        assert_eq!(
+            make_labels(vec![("aa", "va"), ("yy", "vy"), ("zz", "vz")]),
+            labels
+        );
+
+        assert!(CeresDBStorage::<()>::normalize_labels(vec![]).is_err());
+    }
+
+    #[test]
+    fn test_normailze_matchers() {
+        let matchers = make_matchers(vec![
+            ("a", "1", Type::Eq),
+            ("b", "2", Type::Neq),
+            ("c", "3", Type::Re),
+            ("d", "4", Type::Nre),
+            (NAME_LABEL, "cpu", Type::Eq),
+        ]);
+
+        let (metric, filters) = CeresDBStorage::<()>::normalize_matchers(matchers).unwrap();
+        assert_eq!("cpu", metric);
+        assert_eq!(
+            vec!["a = '1'", "b != '2'", "c ~ '^(?:3)'", "d !~ '^(?:4)'"],
+            filters
+        );
+
+        assert!(CeresDBStorage::<()>::normalize_matchers(vec![]).is_err());
+    }
+
+    // Build a schema with
+    // - 2 tags(tag1, tag2)
+    // - 1 field(value)
+    fn build_schema() -> schema::Schema {
+        schema::Builder::new()
+            .auto_increment_column_id(true)
+            .add_key_column(
+                column_schema::Builder::new(TSID_COLUMN.to_string(), DatumKind::UInt64)
+                    .build()
+                    .unwrap(),
+            )
+            .unwrap()
+            .add_key_column(
+                column_schema::Builder::new(TIMESTAMP_COLUMN.to_string(), DatumKind::Timestamp)
+                    .build()
+                    .unwrap(),
+            )
+            .unwrap()
+            .add_normal_column(
+                column_schema::Builder::new(VALUE_COLUMN.to_string(), DatumKind::Double)
+                    .build()
+                    .unwrap(),
+            )
+            .unwrap()
+            .add_normal_column(
+                column_schema::Builder::new("tag1".to_string(), DatumKind::String)
+                    .is_tag(true)
+                    .build()
+                    .unwrap(),
+            )
+            .unwrap()
+            .add_normal_column(
+                column_schema::Builder::new("tag2".to_string(), DatumKind::String)
+                    .is_tag(true)
+                    .build()
+                    .unwrap(),
+            )
+            .unwrap()
+            .build()
+            .unwrap()
+    }
+
+    fn build_record_batch(schema: &schema::Schema) -> RecordBatchVec {
+        let tsid: ArrayRef = Arc::new(UInt64Array::from(vec![1, 1, 2, 3, 3]));
+        let timestamp: ArrayRef = Arc::new(TimestampMillisecondArray::from(vec![
+            11111111, 11111112, 11111113, 11111111, 11111112,
+        ]));
+        let values: ArrayRef =
+            Arc::new(Float64Array::from(vec![100.0, 101.0, 200.0, 300.0, 301.0]));
+        let tag1: ArrayRef = Arc::new(StringArray::from(vec!["a", "a", "b", "c", "c"]));
+        let tag2: ArrayRef = Arc::new(StringArray::from(vec!["x", "x", "y", "z", "z"]));
+
+        let batch = ArrowRecordBatch::try_new(
+            schema.to_arrow_schema_ref(),
+            vec![tsid, timestamp, values, tag1, tag2],
+        )
+        .unwrap();
+
+        vec![RecordBatch::try_from(batch).unwrap()]
+    }
+
+    #[test]
+    fn test_convert_records_to_query_result() {
+        let metric = "cpu";
+        let schema = build_schema();
+        let batches = build_record_batch(&schema);
+        let record_schema = schema.to_record_schema();
+        let converter = Converter::try_new(&record_schema).unwrap();
+        let mut query_result = converter.convert(metric.to_string(), batches).unwrap();
+
+        query_result
+            .timeseries
+            // sort time series by first label's value(tag1 in this case)
+            .sort_unstable_by(|a, b| a.labels[0].value.cmp(&b.labels[0].value));
+
+        assert_eq!(
+            QueryResult {
+                timeseries: vec![
+                    TimeSeries {
+                        labels: make_labels(vec![
+                            ("tag1", "a"),
+                            ("tag2", "x"),
+                            (NAME_LABEL, metric)
+                        ]),
+                        samples: make_samples(vec![(11111111, 100.0), (11111112, 101.0),]),
+                        ..Default::default()
+                    },
+                    TimeSeries {
+                        labels: make_labels(vec![
+                            ("tag1", "b"),
+                            ("tag2", "y"),
+                            (NAME_LABEL, metric)
+                        ]),
+                        samples: make_samples(vec![(11111113, 200.0)]),
+                        ..Default::default()
+                    },
+                    TimeSeries {
+                        labels: make_labels(vec![
+                            ("tag1", "c"),
+                            ("tag2", "z"),
+                            (NAME_LABEL, metric)
+                        ]),
+                        samples: make_samples(vec![(11111111, 300.0), (11111112, 301.0),]),
+                        ..Default::default()
+                    },
+                ]
+            },
+            query_result
+        );
+    }
+}
diff --git a/server/src/handlers/sql.rs b/server/src/handlers/sql.rs
index a98e5183e0..57e740996e 100644
--- a/server/src/handlers/sql.rs
+++ b/server/src/handlers/sql.rs
@@ -4,7 +4,6 @@
 
 use std::time::Instant;
 
-use arrow::error::Result as ArrowResult;
 use common_types::{
     bytes::Bytes,
     datum::{Datum, DatumKind},
@@ -25,9 +24,7 @@ use sql::{
 };
 
 use crate::handlers::{
-    error::{
-        ArrowToString, CreatePlan, InterpreterExec, ParseSql, QueryBlock, QueryTimeout, TooMuchStmt,
-    },
+    error::{CreatePlan, InterpreterExec, ParseSql, QueryBlock, QueryTimeout, TooMuchStmt},
     prelude::*,
 };
 
@@ -108,10 +105,10 @@ impl From<Bytes> for Request {
 }
 
 pub async fn handle_sql<Q: QueryExecutor + 'static>(
-    ctx: RequestContext,
+    ctx: &RequestContext,
     instance: InstanceRef<Q>,
     request: Request,
-) -> Result<Response> {
+) -> Result<Output> {
     let request_id = RequestId::next_id();
     let begin_instant = Instant::now();
     let deadline = ctx.timeout.map(|t| begin_instant + t);
@@ -139,7 +136,7 @@ pub async fn handle_sql<Q: QueryExecutor + 'static>(
         .context(ParseSql)?;
 
     if stmts.is_empty() {
-        return Ok(Response::AffectedRows(0));
+        return Ok(Output::AffectedRows(0));
     }
 
     // TODO(yingwen): For simplicity, we only support executing one statement now
@@ -167,7 +164,7 @@ pub async fn handle_sql<Q: QueryExecutor + 'static>(
     // Execute in interpreter
     let interpreter_ctx = InterpreterContext::builder(request_id, deadline)
         // Use current ctx's catalog and tenant as default catalog and tenant
-        .default_catalog_and_schema(ctx.catalog, ctx.schema)
+        .default_catalog_and_schema(ctx.catalog.to_string(), ctx.schema.to_string())
         .enable_partition_table_access(ctx.enable_partition_table_access)
         .build();
     let interpreter_factory = Factory::new(
@@ -202,10 +199,6 @@ pub async fn handle_sql<Q: QueryExecutor + 'static>(
             query: &request.query,
         })?
     };
-    // Convert output to json
-    let resp = convert_output(output).context(ArrowToString {
-        query: &request.query,
-    })?;
 
     info!(
         "sql handler finished, request_id:{}, cost:{}ms, request:{:?}",
@@ -214,22 +207,23 @@ pub async fn handle_sql<Q: QueryExecutor + 'static>(
         request
     );
 
-    Ok(resp)
+    Ok(output)
 }
 
-fn convert_output(output: Output) -> ArrowResult<Response> {
+// Convert output to json
+pub fn convert_output(output: Output) -> Response {
     match output {
-        Output::AffectedRows(n) => Ok(Response::AffectedRows(n)),
+        Output::AffectedRows(n) => Response::AffectedRows(n),
         Output::Records(records) => convert_records(records),
     }
 }
 
-fn convert_records(records: RecordBatchVec) -> ArrowResult<Response> {
+fn convert_records(records: RecordBatchVec) -> Response {
     if records.is_empty() {
-        return Ok(Response::Rows(ResponseRows {
+        return Response::Rows(ResponseRows {
             column_names: Vec::new(),
             data: Vec::new(),
-        }));
+        });
     }
 
     let mut column_names = vec![];
@@ -261,8 +255,8 @@ fn convert_records(records: RecordBatchVec) -> ArrowResult<Response> {
         }
     }
 
-    Ok(Response::Rows(ResponseRows {
+    Response::Rows(ResponseRows {
         column_names,
         data: column_data,
-    }))
+    })
 }
diff --git a/server/src/http.rs b/server/src/http.rs
index 4f386bdcfd..b94e48714f 100644
--- a/server/src/http.rs
+++ b/server/src/http.rs
@@ -10,6 +10,7 @@ use std::{
 use log::error;
 use logger::RuntimeLevel;
 use profile::Profiler;
+use prom_remote_api::{types::RemoteStorageRef, web};
 use query_engine::executor::Executor as QueryExecutor;
 use router::endpoint::Endpoint;
 use serde_derive::Serialize;
@@ -28,9 +29,10 @@ use crate::{
     consts,
     context::RequestContext,
     error_util,
-    handlers::{self, sql::Request},
+    handlers::{self, prom::CeresDBStorage, sql::Request},
     instance::InstanceRef,
     metrics,
+    schema_config_provider::SchemaConfigProviderRef,
 };
 
 #[derive(Debug, Snafu)]
@@ -55,6 +57,9 @@ pub enum Error {
     #[snafu(display("Missing instance to build service.\nBacktrace:\n{}", backtrace))]
     MissingInstance { backtrace: Backtrace },
 
+    #[snafu(display("Missing schema config provider.\nBacktrace:\n{}", backtrace))]
+    MissingSchemaConfigProvider { backtrace: Backtrace },
+
     #[snafu(display(
         "Fail to do heap profiling, err:{}.\nBacktrace:\n{}",
         source,
@@ -100,6 +105,7 @@ pub struct Service<Q> {
     log_runtime: Arc<RuntimeLevel>,
     instance: InstanceRef<Q>,
     profiler: Arc<Profiler>,
+    prom_remote_storage: RemoteStorageRef<RequestContext, crate::handlers::prom::Error>,
     tx: Sender<()>,
     config: HttpConfig,
     enable_tenant_as_schema: bool,
@@ -121,6 +127,30 @@ impl<Q: QueryExecutor + 'static> Service<Q> {
             .or(self.admin_block())
             .or(self.flush_memtable())
             .or(self.update_log_level())
+            .or(self.prom_api())
+    }
+
+    /// Expose `/prom/v1/read` and `/prom/v1/write` to serve Prometheus remote
+    /// storage request
+    fn prom_api(&self) -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
+        let write_api = warp::path!("write")
+            .and(web::warp::with_remote_storage(
+                self.prom_remote_storage.clone(),
+            ))
+            .and(self.with_context())
+            .and(web::warp::protobuf_body())
+            .and_then(web::warp::write);
+        let query_api = warp::path!("read")
+            .and(web::warp::with_remote_storage(
+                self.prom_remote_storage.clone(),
+            ))
+            .and(self.with_context())
+            .and(web::warp::protobuf_body())
+            .and_then(web::warp::read);
+
+        warp::path!("prom" / "v1" / ..)
+            .and(warp::post())
+            .and(write_api.or(query_api))
     }
 
     fn home(&self) -> impl Filter<Extract = impl warp::Reply, Error = warp::Rejection> + Clone {
@@ -145,8 +175,9 @@ impl<Q: QueryExecutor + 'static> Service<Q> {
             .and(self.with_context())
             .and(self.with_instance())
             .and_then(|req, ctx, instance| async move {
-                let result = handlers::sql::handle_sql(ctx, instance, req)
+                let result = handlers::sql::handle_sql(&ctx, instance, req)
                     .await
+                    .map(handlers::sql::convert_output)
                     .map_err(|e| {
                         // TODO(yingwen): Maybe truncate and print the sql
                         error!("Http service Failed to handle sql, err:{}", e);
@@ -361,6 +392,7 @@ pub struct Builder<Q> {
     engine_runtimes: Option<Arc<EngineRuntimes>>,
     log_runtime: Option<Arc<RuntimeLevel>>,
     instance: Option<InstanceRef<Q>>,
+    schema_config_provider: Option<SchemaConfigProviderRef>,
 }
 
 impl<Q> Builder<Q> {
@@ -371,6 +403,7 @@ impl<Q> Builder<Q> {
             engine_runtimes: None,
             log_runtime: None,
             instance: None,
+            schema_config_provider: None,
         }
     }
 
@@ -393,6 +426,11 @@ impl<Q> Builder<Q> {
         self.enable_tenant_as_schema = enable_tenant_as_schema;
         self
     }
+
+    pub fn schema_config_provider(mut self, provider: SchemaConfigProviderRef) -> Self {
+        self.schema_config_provider = Some(provider);
+        self
+    }
 }
 
 impl<Q: QueryExecutor + 'static> Builder<Q> {
@@ -401,12 +439,20 @@ impl<Q: QueryExecutor + 'static> Builder<Q> {
         let engine_runtime = self.engine_runtimes.context(MissingEngineRuntimes)?;
         let log_runtime = self.log_runtime.context(MissingLogRuntime)?;
         let instance = self.instance.context(MissingInstance)?;
+        let schema_config_provider = self
+            .schema_config_provider
+            .context(MissingSchemaConfigProvider)?;
+        let prom_remote_storage = Arc::new(CeresDBStorage::new(
+            instance.clone(),
+            schema_config_provider,
+        ));
         let (tx, rx) = oneshot::channel();
 
         let service = Service {
             engine_runtimes: engine_runtime.clone(),
             log_runtime,
             instance,
+            prom_remote_storage,
             profiler: Arc::new(Profiler::default()),
             tx,
             config: self.config.clone(),
@@ -454,6 +500,7 @@ fn error_to_status_code(err: &Error) -> StatusCode {
         | Error::MissingEngineRuntimes { .. }
         | Error::MissingLogRuntime { .. }
         | Error::MissingInstance { .. }
+        | Error::MissingSchemaConfigProvider { .. }
         | Error::ParseIpAddr { .. }
         | Error::ProfileHeap { .. }
         | Error::Internal { .. }
diff --git a/server/src/mysql/worker.rs b/server/src/mysql/worker.rs
index 139e7d6d1d..0ffa173c5e 100644
--- a/server/src/mysql/worker.rs
+++ b/server/src/mysql/worker.rs
@@ -113,8 +113,10 @@ where
         let ctx = self.create_ctx()?;
 
         let req = Request::from(sql.to_string());
-        handlers::sql::handle_sql(ctx, self.instance.clone(), req)
+        handlers::sql::handle_sql(&ctx, self.instance.clone(), req)
             .await
+            // TODO(chenxiang): no need for this convert for MySQL, remove it
+            .map(handlers::sql::convert_output)
             .map_err(|e| {
                 error!("Mysql service Failed to handle sql, err: {}", e);
                 e
diff --git a/server/src/server.rs b/server/src/server.rs
index 25fa447fcc..b3cb14107e 100644
--- a/server/src/server.rs
+++ b/server/src/server.rs
@@ -296,11 +296,16 @@ impl<Q: QueryExecutor + 'static> Builder<Q> {
         // Start http service
         let engine_runtimes = self.engine_runtimes.context(MissingEngineRuntimes)?;
         let log_runtime = self.log_runtime.context(MissingLogRuntime)?;
+        let schema_config_provider = self
+            .schema_config_provider
+            .clone()
+            .context(MissingSchemaConfigProvider)?;
         let http_service = http::Builder::new(http_config)
             .engine_runtimes(engine_runtimes.clone())
             .log_runtime(log_runtime)
             .instance(instance.clone())
             .enable_tenant_as_schema(self.config.enable_tenant_as_schema)
+            .schema_config_provider(schema_config_provider)
             .build()
             .context(StartHttpService)?;