Revert "fix: align backend_tls default value with example config (#7496)"

This reverts commit 83932c8c9e.

.github/actions/setup-greptimedb-cluster/action.yml

@@ -70,23 +70,19 @@ runs:
         --wait \
         --wait-for-jobs
   - name: Wait for GreptimeDB
-    uses: nick-fields/retry@v3
-    with:
-      timeout_minutes: 3
-      max_attempts: 1
-      shell: bash
-      command: |
-        while true; do
-          PHASE=$(kubectl -n my-greptimedb get gtc my-greptimedb -o jsonpath='{.status.clusterPhase}')
-          if [ "$PHASE" == "Running" ]; then
-            echo "Cluster is ready"
-            break
-          else
-            echo "Cluster is not ready yet: Current phase: $PHASE"
-            kubectl get pods -n my-greptimedb
-            sleep 5 # wait for 5 seconds before check again.
-          fi
-        done
+    shell: bash
+    run: |
+      while true; do
+        PHASE=$(kubectl -n my-greptimedb get gtc my-greptimedb -o jsonpath='{.status.clusterPhase}')
+        if [ "$PHASE" == "Running" ]; then
+          echo "Cluster is ready"
+          break
+        else
+          echo "Cluster is not ready yet: Current phase: $PHASE"
+          kubectl get pods -n my-greptimedb
+          sleep 5 # wait for 5 seconds before check again.
+        fi
+      done
   - name: Print GreptimeDB info
     if: always()
     shell: bash

Cargo.lock

@@ -4062,7 +4062,6 @@ dependencies = [
  "mito2",
  "num_cpus",
  "object-store",
- "partition",
  "prometheus",
  "prost 0.13.5",
  "query",
@@ -4100,7 +4099,6 @@ dependencies = [
  "num-traits",
  "ordered-float 4.6.0",
  "paste",
- "regex",
  "serde",
  "serde_json",
  "snafu 0.8.6",
@@ -9476,7 +9474,6 @@ dependencies = [
  "ahash 0.8.12",
  "api",
  "arrow",
- "arrow-schema",
  "async-trait",
  "catalog",
  "chrono",

src/api/src/helper.rs

@@ -895,7 +895,7 @@ pub fn is_column_type_value_eq(
         .unwrap_or(false)
 }
 
-pub fn encode_json_value(value: JsonValue) -> v1::JsonValue {
+fn encode_json_value(value: JsonValue) -> v1::JsonValue {
     fn helper(json: JsonVariant) -> v1::JsonValue {
         let value = match json {
             JsonVariant::Null => None,

src/api/src/v1/column_def.rs

@@ -17,8 +17,8 @@ use std::collections::HashMap;
 use arrow_schema::extension::{EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY};
 use datatypes::schema::{
     COMMENT_KEY, ColumnDefaultConstraint, ColumnSchema, FULLTEXT_KEY, FulltextAnalyzer,
-    FulltextBackend, FulltextOptions, INVERTED_INDEX_KEY, Metadata, SKIPPING_INDEX_KEY,
-    SkippingIndexOptions, SkippingIndexType,
+    FulltextBackend, FulltextOptions, INVERTED_INDEX_KEY, SKIPPING_INDEX_KEY, SkippingIndexOptions,
+    SkippingIndexType,
 };
 use greptime_proto::v1::{
     Analyzer, FulltextBackend as PbFulltextBackend, SkippingIndexType as PbSkippingIndexType,
@@ -36,14 +36,6 @@ const INVERTED_INDEX_GRPC_KEY: &str = "inverted_index";
 /// Key used to store skip index options in gRPC column options.
 const SKIPPING_INDEX_GRPC_KEY: &str = "skipping_index";
 
-const COLUMN_OPTION_MAPPINGS: [(&str, &str); 5] = [
-    (FULLTEXT_GRPC_KEY, FULLTEXT_KEY),
-    (INVERTED_INDEX_GRPC_KEY, INVERTED_INDEX_KEY),
-    (SKIPPING_INDEX_GRPC_KEY, SKIPPING_INDEX_KEY),
-    (EXTENSION_TYPE_NAME_KEY, EXTENSION_TYPE_NAME_KEY),
-    (EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_METADATA_KEY),
-];
-
 /// Tries to construct a `ColumnSchema` from the given  `ColumnDef`.
 pub fn try_as_column_schema(column_def: &ColumnDef) -> Result<ColumnSchema> {
     let data_type = ColumnDataTypeWrapper::try_new(
@@ -139,21 +131,6 @@ pub fn try_as_column_def(column_schema: &ColumnSchema, is_primary_key: bool) ->
     })
 }
 
-/// Collect the [ColumnOptions] into the [Metadata] that can be used in, for example, [ColumnSchema].
-pub fn collect_column_options(column_options: Option<&ColumnOptions>) -> Metadata {
-    let Some(ColumnOptions { options }) = column_options else {
-        return Metadata::default();
-    };
-
-    let mut metadata = Metadata::with_capacity(options.len());
-    for (x, y) in COLUMN_OPTION_MAPPINGS {
-        if let Some(v) = options.get(x) {
-            metadata.insert(y.to_string(), v.clone());
-        }
-    }
-    metadata
-}
-
 /// Constructs a `ColumnOptions` from the given `ColumnSchema`.
 pub fn options_from_column_schema(column_schema: &ColumnSchema) -> Option<ColumnOptions> {
     let mut options = ColumnOptions::default();

src/catalog/src/system_schema/information_schema/columns.rs

@@ -399,8 +399,8 @@ impl InformationSchemaColumnsBuilder {
             self.is_nullables.push(Some("No"));
         }
         self.column_types.push(Some(&data_type));
-        let column_comment = column_schema.column_comment().map(|x| x.as_ref());
-        self.column_comments.push(column_comment);
+        self.column_comments
+            .push(column_schema.column_comment().map(|x| x.as_ref()));
     }
 
     fn finish(&mut self) -> Result<RecordBatch> {

src/cli/src/common/store.rs

@@ -92,7 +92,7 @@ impl StoreConfig {
     pub fn tls_config(&self) -> Option<TlsOption> {
         if self.backend_tls_mode != TlsMode::Disable {
             Some(TlsOption {
-                mode: self.backend_tls_mode,
+                mode: self.backend_tls_mode.clone(),
                 cert_path: self.backend_tls_cert_path.clone(),
                 key_path: self.backend_tls_key_path.clone(),
                 ca_cert_path: self.backend_tls_ca_cert_path.clone(),

src/cmd/src/frontend.rs

@@ -236,7 +236,7 @@ impl StartCommand {
         };
 
         let tls_opts = TlsOption::new(
-            self.tls_mode,
+            self.tls_mode.clone(),
             self.tls_cert_path.clone(),
             self.tls_key_path.clone(),
             self.tls_watch,

src/cmd/src/standalone.rs

@@ -261,7 +261,7 @@ impl StartCommand {
         };
 
         let tls_opts = TlsOption::new(
-            self.tls_mode,
+            self.tls_mode.clone(),
             self.tls_cert_path.clone(),
             self.tls_key_path.clone(),
             self.tls_watch,

src/common/meta/src/instruction.rs

@@ -530,49 +530,6 @@ impl Display for EnterStagingRegion {
     }
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
-pub struct RemapManifest {
-    pub region_id: RegionId,
-    /// Regions to remap manifests from.
-    pub input_regions: Vec<RegionId>,
-    /// For each old region, which new regions should receive its files
-    pub region_mapping: HashMap<RegionId, Vec<RegionId>>,
-    /// New partition expressions for the new regions.
-    pub new_partition_exprs: HashMap<RegionId, String>,
-}
-
-impl Display for RemapManifest {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "RemapManifest(region_id={}, input_regions={:?}, region_mapping={:?}, new_partition_exprs={:?})",
-            self.region_id, self.input_regions, self.region_mapping, self.new_partition_exprs
-        )
-    }
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
-pub struct ApplyStagingManifest {
-    /// The region ID to apply the staging manifest to.
-    pub region_id: RegionId,
-    /// The partition expression of the staging region.
-    pub partition_expr: String,
-    /// The region that stores the staging manifests in its staging blob storage.
-    pub central_region_id: RegionId,
-    /// The relative path to the staging manifest within the central region's staging blob storage.
-    pub manifest_path: String,
-}
-
-impl Display for ApplyStagingManifest {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "ApplyStagingManifest(region_id={}, partition_expr={}, central_region_id={}, manifest_path={})",
-            self.region_id, self.partition_expr, self.central_region_id, self.manifest_path
-        )
-    }
-}
-
 #[derive(Debug, Clone, Serialize, Deserialize, Display, PartialEq)]
 pub enum Instruction {
     /// Opens regions.
@@ -602,10 +559,6 @@ pub enum Instruction {
     Suspend,
     /// Makes regions enter staging state.
     EnterStagingRegions(Vec<EnterStagingRegion>),
-    /// Remaps manifests for a region.
-    RemapManifest(RemapManifest),
-    /// Applies staging manifests for a region.
-    ApplyStagingManifests(Vec<ApplyStagingManifest>),
 }
 
 impl Instruction {
@@ -784,48 +737,6 @@ impl EnterStagingRegionsReply {
     }
 }
 
-#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
-pub struct RemapManifestReply {
-    /// Returns false if the region does not exist.
-    pub exists: bool,
-    /// A map from region IDs to their corresponding remapped manifest paths.
-    pub manifest_paths: HashMap<RegionId, String>,
-    /// Return error if any during the operation.
-    pub error: Option<String>,
-}
-
-impl Display for RemapManifestReply {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "RemapManifestReply(manifest_paths={:?}, error={:?})",
-            self.manifest_paths, self.error
-        )
-    }
-}
-
-#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
-pub struct ApplyStagingManifestsReply {
-    pub replies: Vec<ApplyStagingManifestReply>,
-}
-
-impl ApplyStagingManifestsReply {
-    pub fn new(replies: Vec<ApplyStagingManifestReply>) -> Self {
-        Self { replies }
-    }
-}
-
-#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
-pub struct ApplyStagingManifestReply {
-    pub region_id: RegionId,
-    /// Returns true if the region is ready to serve reads and writes.
-    pub ready: bool,
-    /// Indicates whether the region exists.
-    pub exists: bool,
-    /// Return error if any during the operation.
-    pub error: Option<String>,
-}
-
 #[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
 #[serde(tag = "type", rename_all = "snake_case")]
 pub enum InstructionReply {
@@ -847,8 +758,6 @@ pub enum InstructionReply {
     GetFileRefs(GetFileRefsReply),
     GcRegions(GcRegionsReply),
     EnterStagingRegions(EnterStagingRegionsReply),
-    RemapManifest(RemapManifestReply),
-    ApplyStagingManifests(ApplyStagingManifestsReply),
 }
 
 impl Display for InstructionReply {
@@ -872,12 +781,6 @@ impl Display for InstructionReply {
                     reply.replies
                 )
             }
-            Self::RemapManifest(reply) => write!(f, "InstructionReply::RemapManifest({})", reply),
-            Self::ApplyStagingManifests(reply) => write!(
-                f,
-                "InstructionReply::ApplyStagingManifests({:?})",
-                reply.replies
-            ),
         }
     }
 }
@@ -925,20 +828,6 @@ impl InstructionReply {
             _ => panic!("Expected EnterStagingRegion reply"),
         }
     }
-
-    pub fn expect_remap_manifest_reply(self) -> RemapManifestReply {
-        match self {
-            Self::RemapManifest(reply) => reply,
-            _ => panic!("Expected RemapManifest reply"),
-        }
-    }
-
-    pub fn expect_apply_staging_manifests_reply(self) -> Vec<ApplyStagingManifestReply> {
-        match self {
-            Self::ApplyStagingManifests(reply) => reply.replies,
-            _ => panic!("Expected ApplyStagingManifest reply"),
-        }
-    }
 }
 
 #[cfg(test)]

src/common/meta/src/kv_backend/rds/postgres.rs

@@ -868,8 +868,6 @@ impl PgStore {
         let client = match pool.get().await {
             Ok(client) => client,
             Err(e) => {
-                // We need to log the debug for the error to help diagnose the issue.
-                common_telemetry::error!(e; "Failed to get Postgres connection.");
                 return GetPostgresConnectionSnafu {
                     reason: e.to_string(),
                 }

src/datanode/Cargo.toml

@@ -77,5 +77,4 @@ common-query.workspace = true
 common-test-util.workspace = true
 datafusion-common.workspace = true
 mito2 = { workspace = true, features = ["test"] }
-partition.workspace = true
 session.workspace = true

src/datanode/src/error.rs

@@ -201,7 +201,6 @@ pub enum Error {
     ShutdownServer {
         #[snafu(implicit)]
         location: Location,
-        #[snafu(source)]
         source: servers::error::Error,
     },
 
@@ -209,7 +208,6 @@ pub enum Error {
     ShutdownInstance {
         #[snafu(implicit)]
         location: Location,
-        #[snafu(source)]
         source: BoxedError,
     },
 

src/datanode/src/heartbeat/handler.rs

@@ -22,7 +22,6 @@ use common_telemetry::error;
 use snafu::OptionExt;
 use store_api::storage::GcReport;
 
-mod apply_staging_manifest;
 mod close_region;
 mod downgrade_region;
 mod enter_staging;
@@ -30,10 +29,8 @@ mod file_ref;
 mod flush_region;
 mod gc_worker;
 mod open_region;
-mod remap_manifest;
 mod upgrade_region;
 
-use crate::heartbeat::handler::apply_staging_manifest::ApplyStagingManifestsHandler;
 use crate::heartbeat::handler::close_region::CloseRegionsHandler;
 use crate::heartbeat::handler::downgrade_region::DowngradeRegionsHandler;
 use crate::heartbeat::handler::enter_staging::EnterStagingRegionsHandler;
@@ -41,7 +38,6 @@ use crate::heartbeat::handler::file_ref::GetFileRefsHandler;
 use crate::heartbeat::handler::flush_region::FlushRegionsHandler;
 use crate::heartbeat::handler::gc_worker::GcRegionsHandler;
 use crate::heartbeat::handler::open_region::OpenRegionsHandler;
-use crate::heartbeat::handler::remap_manifest::RemapManifestHandler;
 use crate::heartbeat::handler::upgrade_region::UpgradeRegionsHandler;
 use crate::heartbeat::task_tracker::TaskTracker;
 use crate::region_server::RegionServer;
@@ -132,10 +128,6 @@ impl RegionHeartbeatResponseHandler {
             Instruction::EnterStagingRegions(_) => {
                 Ok(Some(Box::new(EnterStagingRegionsHandler.into())))
             }
-            Instruction::RemapManifest(_) => Ok(Some(Box::new(RemapManifestHandler.into()))),
-            Instruction::ApplyStagingManifests(_) => {
-                Ok(Some(Box::new(ApplyStagingManifestsHandler.into())))
-            }
         }
     }
 }
@@ -150,8 +142,6 @@ pub enum InstructionHandlers {
     GetFileRefs(GetFileRefsHandler),
     GcRegions(GcRegionsHandler),
     EnterStagingRegions(EnterStagingRegionsHandler),
-    RemapManifest(RemapManifestHandler),
-    ApplyStagingManifests(ApplyStagingManifestsHandler),
 }
 
 macro_rules! impl_from_handler {
@@ -174,9 +164,7 @@ impl_from_handler!(
     UpgradeRegionsHandler => UpgradeRegions,
     GetFileRefsHandler => GetFileRefs,
     GcRegionsHandler => GcRegions,
-    EnterStagingRegionsHandler => EnterStagingRegions,
-    RemapManifestHandler => RemapManifest,
-    ApplyStagingManifestsHandler => ApplyStagingManifests
+    EnterStagingRegionsHandler => EnterStagingRegions
 );
 
 macro_rules! dispatch_instr {
@@ -221,9 +209,7 @@ dispatch_instr!(
     UpgradeRegions => UpgradeRegions,
     GetFileRefs => GetFileRefs,
     GcRegions => GcRegions,
-    EnterStagingRegions => EnterStagingRegions,
-    RemapManifest => RemapManifest,
-    ApplyStagingManifests => ApplyStagingManifests,
+    EnterStagingRegions => EnterStagingRegions
 );
 
 #[async_trait]

src/datanode/src/heartbeat/handler/apply_staging_manifest.rs

@@ -1,287 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use common_meta::instruction::{
-    ApplyStagingManifest, ApplyStagingManifestReply, ApplyStagingManifestsReply, InstructionReply,
-};
-use common_telemetry::{error, warn};
-use futures::future::join_all;
-use store_api::region_request::{ApplyStagingManifestRequest, RegionRequest};
-
-use crate::heartbeat::handler::{HandlerContext, InstructionHandler};
-
-pub struct ApplyStagingManifestsHandler;
-
-#[async_trait::async_trait]
-impl InstructionHandler for ApplyStagingManifestsHandler {
-    type Instruction = Vec<ApplyStagingManifest>;
-    async fn handle(
-        &self,
-        ctx: &HandlerContext,
-        requests: Self::Instruction,
-    ) -> Option<InstructionReply> {
-        let results = join_all(
-            requests
-                .into_iter()
-                .map(|request| Self::handle_apply_staging_manifest(ctx, request)),
-        )
-        .await;
-        Some(InstructionReply::ApplyStagingManifests(
-            ApplyStagingManifestsReply::new(results),
-        ))
-    }
-}
-
-impl ApplyStagingManifestsHandler {
-    async fn handle_apply_staging_manifest(
-        ctx: &HandlerContext,
-        request: ApplyStagingManifest,
-    ) -> ApplyStagingManifestReply {
-        let Some(leader) = ctx.region_server.is_region_leader(request.region_id) else {
-            warn!("Region: {} is not found", request.region_id);
-            return ApplyStagingManifestReply {
-                region_id: request.region_id,
-                exists: false,
-                ready: false,
-                error: None,
-            };
-        };
-        if !leader {
-            warn!("Region: {} is not leader", request.region_id);
-            return ApplyStagingManifestReply {
-                region_id: request.region_id,
-                exists: true,
-                ready: false,
-                error: Some("Region is not leader".into()),
-            };
-        }
-
-        match ctx
-            .region_server
-            .handle_request(
-                request.region_id,
-                RegionRequest::ApplyStagingManifest(ApplyStagingManifestRequest {
-                    partition_expr: request.partition_expr,
-                    central_region_id: request.central_region_id,
-                    manifest_path: request.manifest_path,
-                }),
-            )
-            .await
-        {
-            Ok(_) => ApplyStagingManifestReply {
-                region_id: request.region_id,
-                exists: true,
-                ready: true,
-                error: None,
-            },
-            Err(err) => {
-                error!(err; "Failed to apply staging manifest");
-                ApplyStagingManifestReply {
-                    region_id: request.region_id,
-                    exists: true,
-                    ready: false,
-                    error: Some(format!("{err:?}")),
-                }
-            }
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::collections::HashMap;
-    use std::sync::Arc;
-
-    use common_meta::instruction::RemapManifest;
-    use datatypes::value::Value;
-    use mito2::config::MitoConfig;
-    use mito2::engine::MITO_ENGINE_NAME;
-    use mito2::test_util::{CreateRequestBuilder, TestEnv};
-    use partition::expr::{PartitionExpr, col};
-    use store_api::path_utils::table_dir;
-    use store_api::region_engine::RegionRole;
-    use store_api::region_request::EnterStagingRequest;
-    use store_api::storage::RegionId;
-
-    use super::*;
-    use crate::heartbeat::handler::remap_manifest::RemapManifestHandler;
-    use crate::region_server::RegionServer;
-    use crate::tests::{MockRegionEngine, mock_region_server};
-
-    #[tokio::test]
-    async fn test_region_not_exist() {
-        let mut mock_region_server = mock_region_server();
-        let (mock_engine, _) = MockRegionEngine::new(MITO_ENGINE_NAME);
-        mock_region_server.register_engine(mock_engine);
-        let handler_context = HandlerContext::new_for_test(mock_region_server);
-        let region_id = RegionId::new(1024, 1);
-        let reply = ApplyStagingManifestsHandler
-            .handle(
-                &handler_context,
-                vec![ApplyStagingManifest {
-                    region_id,
-                    partition_expr: "".to_string(),
-                    central_region_id: RegionId::new(1024, 9999), // use a dummy value
-                    manifest_path: "".to_string(),
-                }],
-            )
-            .await
-            .unwrap();
-        let replies = reply.expect_apply_staging_manifests_reply();
-        let reply = &replies[0];
-        assert!(!reply.exists);
-        assert!(!reply.ready);
-        assert!(reply.error.is_none());
-    }
-
-    #[tokio::test]
-    async fn test_region_not_leader() {
-        let mock_region_server = mock_region_server();
-        let region_id = RegionId::new(1024, 1);
-        let (mock_engine, _) =
-            MockRegionEngine::with_custom_apply_fn(MITO_ENGINE_NAME, |region_engine| {
-                region_engine.mock_role = Some(Some(RegionRole::Follower));
-                region_engine.handle_request_mock_fn = Some(Box::new(|_, _| Ok(0)));
-            });
-        mock_region_server.register_test_region(region_id, mock_engine);
-        let handler_context = HandlerContext::new_for_test(mock_region_server);
-        let region_id = RegionId::new(1024, 1);
-        let reply = ApplyStagingManifestsHandler
-            .handle(
-                &handler_context,
-                vec![ApplyStagingManifest {
-                    region_id,
-                    partition_expr: "".to_string(),
-                    central_region_id: RegionId::new(1024, 2),
-                    manifest_path: "".to_string(),
-                }],
-            )
-            .await
-            .unwrap();
-        let replies = reply.expect_apply_staging_manifests_reply();
-        let reply = &replies[0];
-        assert!(reply.exists);
-        assert!(!reply.ready);
-        assert!(reply.error.is_some());
-    }
-
-    fn range_expr(col_name: &str, start: i64, end: i64) -> PartitionExpr {
-        col(col_name)
-            .gt_eq(Value::Int64(start))
-            .and(col(col_name).lt(Value::Int64(end)))
-    }
-
-    async fn prepare_region(region_server: &RegionServer) {
-        let region_specs = [
-            (RegionId::new(1024, 1), range_expr("x", 0, 49)),
-            (RegionId::new(1024, 2), range_expr("x", 49, 100)),
-        ];
-
-        for (region_id, partition_expr) in region_specs {
-            let builder = CreateRequestBuilder::new();
-            let mut create_req = builder.build();
-            create_req.table_dir = table_dir("test", 1024);
-            region_server
-                .handle_request(region_id, RegionRequest::Create(create_req))
-                .await
-                .unwrap();
-            region_server
-                .handle_request(
-                    region_id,
-                    RegionRequest::EnterStaging(EnterStagingRequest {
-                        partition_expr: partition_expr.as_json_str().unwrap(),
-                    }),
-                )
-                .await
-                .unwrap();
-        }
-    }
-
-    #[tokio::test]
-    async fn test_apply_staging_manifest() {
-        common_telemetry::init_default_ut_logging();
-        let mut region_server = mock_region_server();
-        let region_id = RegionId::new(1024, 1);
-        let mut engine_env = TestEnv::new().await;
-        let engine = engine_env.create_engine(MitoConfig::default()).await;
-        region_server.register_engine(Arc::new(engine.clone()));
-        prepare_region(&region_server).await;
-
-        let handler_context = HandlerContext::new_for_test(region_server);
-        let region_id2 = RegionId::new(1024, 2);
-        let reply = RemapManifestHandler
-            .handle(
-                &handler_context,
-                RemapManifest {
-                    region_id,
-                    input_regions: vec![region_id, region_id2],
-                    region_mapping: HashMap::from([
-                        // [0,49) <- [0, 50)
-                        (region_id, vec![region_id]),
-                        // [49, 100) <- [0, 50), [50,100)
-                        (region_id2, vec![region_id, region_id2]),
-                    ]),
-                    new_partition_exprs: HashMap::from([
-                        (region_id, range_expr("x", 0, 49).as_json_str().unwrap()),
-                        (region_id2, range_expr("x", 49, 100).as_json_str().unwrap()),
-                    ]),
-                },
-            )
-            .await
-            .unwrap();
-        let reply = reply.expect_remap_manifest_reply();
-        assert!(reply.exists);
-        assert!(reply.error.is_none(), "{}", reply.error.unwrap());
-        assert_eq!(reply.manifest_paths.len(), 2);
-        let manifest_path_1 = reply.manifest_paths[&region_id].clone();
-        let manifest_path_2 = reply.manifest_paths[&region_id2].clone();
-
-        let reply = ApplyStagingManifestsHandler
-            .handle(
-                &handler_context,
-                vec![ApplyStagingManifest {
-                    region_id,
-                    partition_expr: range_expr("x", 0, 49).as_json_str().unwrap(),
-                    central_region_id: region_id,
-                    manifest_path: manifest_path_1,
-                }],
-            )
-            .await
-            .unwrap();
-        let replies = reply.expect_apply_staging_manifests_reply();
-        let reply = &replies[0];
-        assert!(reply.exists);
-        assert!(reply.ready);
-        assert!(reply.error.is_none());
-
-        // partition expr mismatch
-        let reply = ApplyStagingManifestsHandler
-            .handle(
-                &handler_context,
-                vec![ApplyStagingManifest {
-                    region_id: region_id2,
-                    partition_expr: range_expr("x", 50, 100).as_json_str().unwrap(),
-                    central_region_id: region_id,
-                    manifest_path: manifest_path_2,
-                }],
-            )
-            .await
-            .unwrap();
-        let replies = reply.expect_apply_staging_manifests_reply();
-        let reply = &replies[0];
-        assert!(reply.exists);
-        assert!(!reply.ready);
-        assert!(reply.error.is_some());
-    }
-}

src/datanode/src/heartbeat/handler/remap_manifest.rs

@@ -1,246 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use common_meta::instruction::{InstructionReply, RemapManifest, RemapManifestReply};
-use common_telemetry::warn;
-use store_api::region_engine::RemapManifestsRequest;
-
-use crate::heartbeat::handler::{HandlerContext, InstructionHandler};
-
-pub struct RemapManifestHandler;
-
-#[async_trait::async_trait]
-impl InstructionHandler for RemapManifestHandler {
-    type Instruction = RemapManifest;
-    async fn handle(
-        &self,
-        ctx: &HandlerContext,
-        request: Self::Instruction,
-    ) -> Option<InstructionReply> {
-        let RemapManifest {
-            region_id,
-            input_regions,
-            region_mapping,
-            new_partition_exprs,
-        } = request;
-        let Some(leader) = ctx.region_server.is_region_leader(region_id) else {
-            warn!("Region: {} is not found", region_id);
-            return Some(InstructionReply::RemapManifest(RemapManifestReply {
-                exists: false,
-                manifest_paths: Default::default(),
-                error: None,
-            }));
-        };
-
-        if !leader {
-            warn!("Region: {} is not leader", region_id);
-            return Some(InstructionReply::RemapManifest(RemapManifestReply {
-                exists: true,
-                manifest_paths: Default::default(),
-                error: Some("Region is not leader".into()),
-            }));
-        }
-
-        let reply = match ctx
-            .region_server
-            .remap_manifests(RemapManifestsRequest {
-                region_id,
-                input_regions,
-                region_mapping,
-                new_partition_exprs,
-            })
-            .await
-        {
-            Ok(result) => InstructionReply::RemapManifest(RemapManifestReply {
-                exists: true,
-                manifest_paths: result.manifest_paths,
-                error: None,
-            }),
-            Err(e) => InstructionReply::RemapManifest(RemapManifestReply {
-                exists: true,
-                manifest_paths: Default::default(),
-                error: Some(format!("{e:?}")),
-            }),
-        };
-
-        Some(reply)
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::collections::HashMap;
-    use std::sync::Arc;
-
-    use common_meta::instruction::RemapManifest;
-    use datatypes::value::Value;
-    use mito2::config::MitoConfig;
-    use mito2::engine::MITO_ENGINE_NAME;
-    use mito2::test_util::{CreateRequestBuilder, TestEnv};
-    use partition::expr::{PartitionExpr, col};
-    use store_api::path_utils::table_dir;
-    use store_api::region_engine::RegionRole;
-    use store_api::region_request::{EnterStagingRequest, RegionRequest};
-    use store_api::storage::RegionId;
-
-    use crate::heartbeat::handler::remap_manifest::RemapManifestHandler;
-    use crate::heartbeat::handler::{HandlerContext, InstructionHandler};
-    use crate::region_server::RegionServer;
-    use crate::tests::{MockRegionEngine, mock_region_server};
-
-    #[tokio::test]
-    async fn test_region_not_exist() {
-        let mut mock_region_server = mock_region_server();
-        let (mock_engine, _) = MockRegionEngine::new(MITO_ENGINE_NAME);
-        mock_region_server.register_engine(mock_engine);
-        let handler_context = HandlerContext::new_for_test(mock_region_server);
-        let region_id = RegionId::new(1024, 1);
-        let reply = RemapManifestHandler
-            .handle(
-                &handler_context,
-                RemapManifest {
-                    region_id,
-                    input_regions: vec![],
-                    region_mapping: HashMap::new(),
-                    new_partition_exprs: HashMap::new(),
-                },
-            )
-            .await
-            .unwrap();
-        let reply = &reply.expect_remap_manifest_reply();
-        assert!(!reply.exists);
-        assert!(reply.error.is_none());
-        assert!(reply.manifest_paths.is_empty());
-    }
-
-    #[tokio::test]
-    async fn test_region_not_leader() {
-        let mock_region_server = mock_region_server();
-        let region_id = RegionId::new(1024, 1);
-        let (mock_engine, _) =
-            MockRegionEngine::with_custom_apply_fn(MITO_ENGINE_NAME, |region_engine| {
-                region_engine.mock_role = Some(Some(RegionRole::Follower));
-                region_engine.handle_request_mock_fn = Some(Box::new(|_, _| Ok(0)));
-            });
-        mock_region_server.register_test_region(region_id, mock_engine);
-        let handler_context = HandlerContext::new_for_test(mock_region_server);
-        let reply = RemapManifestHandler
-            .handle(
-                &handler_context,
-                RemapManifest {
-                    region_id,
-                    input_regions: vec![],
-                    region_mapping: HashMap::new(),
-                    new_partition_exprs: HashMap::new(),
-                },
-            )
-            .await
-            .unwrap();
-        let reply = reply.expect_remap_manifest_reply();
-        assert!(reply.exists);
-        assert!(reply.error.is_some());
-    }
-
-    fn range_expr(col_name: &str, start: i64, end: i64) -> PartitionExpr {
-        col(col_name)
-            .gt_eq(Value::Int64(start))
-            .and(col(col_name).lt(Value::Int64(end)))
-    }
-
-    async fn prepare_region(region_server: &RegionServer) {
-        let region_specs = [
-            (RegionId::new(1024, 1), range_expr("x", 0, 50)),
-            (RegionId::new(1024, 2), range_expr("x", 50, 100)),
-        ];
-
-        for (region_id, partition_expr) in region_specs {
-            let builder = CreateRequestBuilder::new();
-            let mut create_req = builder.build();
-            create_req.table_dir = table_dir("test", 1024);
-            region_server
-                .handle_request(region_id, RegionRequest::Create(create_req))
-                .await
-                .unwrap();
-            region_server
-                .handle_request(
-                    region_id,
-                    RegionRequest::EnterStaging(EnterStagingRequest {
-                        partition_expr: partition_expr.as_json_str().unwrap(),
-                    }),
-                )
-                .await
-                .unwrap();
-        }
-    }
-
-    #[tokio::test]
-    async fn test_remap_manifest() {
-        common_telemetry::init_default_ut_logging();
-        let mut region_server = mock_region_server();
-        let region_id = RegionId::new(1024, 1);
-        let mut engine_env = TestEnv::new().await;
-        let engine = engine_env.create_engine(MitoConfig::default()).await;
-        region_server.register_engine(Arc::new(engine.clone()));
-        prepare_region(&region_server).await;
-
-        let handler_context = HandlerContext::new_for_test(region_server);
-        let region_id2 = RegionId::new(1024, 2);
-        let reply = RemapManifestHandler
-            .handle(
-                &handler_context,
-                RemapManifest {
-                    region_id,
-                    input_regions: vec![region_id, region_id2],
-                    region_mapping: HashMap::from([
-                        (region_id, vec![region_id]),
-                        (region_id2, vec![region_id]),
-                    ]),
-                    new_partition_exprs: HashMap::from([(
-                        region_id,
-                        range_expr("x", 0, 100).as_json_str().unwrap(),
-                    )]),
-                },
-            )
-            .await
-            .unwrap();
-        let reply = reply.expect_remap_manifest_reply();
-        assert!(reply.exists);
-        assert!(reply.error.is_none(), "{}", reply.error.unwrap());
-        assert_eq!(reply.manifest_paths.len(), 1);
-
-        // Remap failed
-        let reply = RemapManifestHandler
-            .handle(
-                &handler_context,
-                RemapManifest {
-                    region_id,
-                    input_regions: vec![region_id],
-                    region_mapping: HashMap::from([
-                        (region_id, vec![region_id]),
-                        (region_id2, vec![region_id]),
-                    ]),
-                    new_partition_exprs: HashMap::from([(
-                        region_id,
-                        range_expr("x", 0, 100).as_json_str().unwrap(),
-                    )]),
-                },
-            )
-            .await
-            .unwrap();
-        let reply = reply.expect_remap_manifest_reply();
-        assert!(reply.exists);
-        assert!(reply.error.is_some());
-        assert!(reply.manifest_paths.is_empty());
-    }
-}

src/datanode/src/region_server.rs

@@ -65,9 +65,8 @@ use store_api::metric_engine_consts::{
     FILE_ENGINE_NAME, LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME,
 };
 use store_api::region_engine::{
-    RegionEngineRef, RegionManifestInfo, RegionRole, RegionStatistic, RemapManifestsRequest,
-    RemapManifestsResponse, SetRegionRoleStateResponse, SettableRegionRoleState,
-    SyncRegionFromRequest,
+    RegionEngineRef, RegionManifestInfo, RegionRole, RegionStatistic, SetRegionRoleStateResponse,
+    SettableRegionRoleState, SyncRegionFromRequest,
 };
 use store_api::region_request::{
     AffectedRows, BatchRegionDdlRequest, RegionCatchupRequest, RegionCloseRequest,
@@ -605,25 +604,6 @@ impl RegionServer {
             .await
     }
 
-    /// Remaps manifests from old regions to new regions.
-    pub async fn remap_manifests(
-        &self,
-        request: RemapManifestsRequest,
-    ) -> Result<RemapManifestsResponse> {
-        let region_id = request.region_id;
-        let engine_with_status = self
-            .inner
-            .region_map
-            .get(&region_id)
-            .with_context(|| RegionNotFoundSnafu { region_id })?;
-
-        engine_with_status
-            .engine()
-            .remap_manifests(request)
-            .await
-            .with_context(|_| HandleRegionRequestSnafu { region_id })
-    }
-
     fn is_suspended(&self) -> bool {
         self.suspend.load(Ordering::Relaxed)
     }

src/datatypes/Cargo.toml

@@ -28,7 +28,6 @@ num = "0.4"
 num-traits = "0.2"
 ordered-float.workspace = true
 paste.workspace = true
-regex.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 snafu.workspace = true

src/datatypes/src/json.rs

@@ -816,7 +816,7 @@ mod tests {
         let result = encode_by_struct(&json_struct, json);
         assert_eq!(
             result.unwrap_err().to_string(),
-            r#"Cannot cast value bar to "<Number>""#
+            "Cannot cast value bar to Number(I64)"
         );
 
         let json = json!({

src/datatypes/src/types/json_type.rs

@@ -13,13 +13,12 @@
 // limitations under the License.
 
 use std::collections::BTreeMap;
-use std::fmt::{Debug, Display, Formatter};
+use std::fmt::{Display, Formatter};
 use std::str::FromStr;
-use std::sync::{Arc, LazyLock};
+use std::sync::Arc;
 
 use arrow::datatypes::DataType as ArrowDataType;
 use common_base::bytes::Bytes;
-use regex::{Captures, Regex};
 use serde::{Deserialize, Serialize};
 use snafu::ResultExt;
 
@@ -135,24 +134,24 @@ impl From<&ConcreteDataType> for JsonNativeType {
 impl Display for JsonNativeType {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         match self {
-            JsonNativeType::Null => write!(f, r#""<Null>""#),
-            JsonNativeType::Bool => write!(f, r#""<Bool>""#),
-            JsonNativeType::Number(_) => {
-                write!(f, r#""<Number>""#)
+            JsonNativeType::Null => write!(f, "Null"),
+            JsonNativeType::Bool => write!(f, "Bool"),
+            JsonNativeType::Number(t) => {
+                write!(f, "Number({t:?})")
             }
-            JsonNativeType::String => write!(f, r#""<String>""#),
+            JsonNativeType::String => write!(f, "String"),
             JsonNativeType::Array(item_type) => {
-                write!(f, "[{}]", item_type)
+                write!(f, "Array[{}]", item_type)
             }
             JsonNativeType::Object(object) => {
                 write!(
                     f,
-                    "{{{}}}",
+                    "Object{{{}}}",
                     object
                         .iter()
-                        .map(|(k, v)| format!(r#""{k}":{v}"#))
+                        .map(|(k, v)| format!(r#""{k}": {v}"#))
                         .collect::<Vec<_>>()
-                        .join(",")
+                        .join(", ")
                 )
             }
         }
@@ -184,11 +183,7 @@ impl JsonType {
         }
     }
 
-    pub fn is_native_type(&self) -> bool {
-        matches!(self.format, JsonFormat::Native(_))
-    }
-
-    pub fn native_type(&self) -> &JsonNativeType {
+    pub(crate) fn native_type(&self) -> &JsonNativeType {
         match &self.format {
             JsonFormat::Jsonb => &JsonNativeType::String,
             JsonFormat::Native(x) => x.as_ref(),
@@ -402,81 +397,8 @@ pub fn jsonb_to_string(val: &[u8]) -> Result<String> {
 /// Converts a json type value to serde_json::Value
 pub fn jsonb_to_serde_json(val: &[u8]) -> Result<serde_json::Value> {
     let json_string = jsonb_to_string(val)?;
-    jsonb_string_to_serde_value(&json_string)
-}
-
-/// Attempts to deserialize a JSON text into `serde_json::Value`, with a best-effort
-/// fallback for Rust-style Unicode escape sequences.
-///
-/// This function is intended to be used on JSON strings produced from the internal
-/// JSONB representation (e.g. via [`jsonb_to_string`]). It first calls
-/// `serde_json::Value::from_str` directly. If that succeeds, the parsed value is
-/// returned as-is.
-///
-/// If the initial parse fails, the input is scanned for Rust-style Unicode code
-/// point escapes of the form `\\u{H...}` (a backslash, `u`, an opening brace,
-/// followed by 1–6 hexadecimal digits, and a closing brace). Each such escape is
-/// converted into JSON-compatible UTF‑16 escape sequences:
-///
-/// - For code points in the Basic Multilingual Plane (≤ `0xFFFF`), the escape is
-///   converted to a single JSON `\\uXXXX` sequence with four uppercase hex digits.
-/// - For code points above `0xFFFF` and less than Unicode max code point `0x10FFFF`,
-///   the code point is encoded as a UTF‑16 surrogate pair and emitted as two consecutive
-///   `\\uXXXX` sequences (as JSON format required).
-///
-/// After this normalization, the function retries parsing the resulting string as
-/// JSON and returns the deserialized value or a `DeserializeSnafu` error if it
-/// still cannot be parsed.
-fn jsonb_string_to_serde_value(json: &str) -> Result<serde_json::Value> {
-    match serde_json::Value::from_str(json) {
-        Ok(v) => Ok(v),
-        Err(e) => {
-            // If above deserialization is failed, the JSON string might contain some Rust chars
-            // that are somehow incorrectly represented as Unicode code point literal. For example,
-            // "\u{fe0f}". We have to convert them to JSON compatible format, like "\uFE0F", then
-            // try to deserialize the JSON string again.
-            if !e.is_syntax() || !e.to_string().contains("invalid escape") {
-                return Err(e).context(DeserializeSnafu { json });
-            }
-
-            static UNICODE_CODE_POINT_PATTERN: LazyLock<Regex> = LazyLock::new(|| {
-                // Match literal "\u{...}" sequences, capturing 1–6 (code point range) hex digits
-                // inside braces.
-                Regex::new(r"\\u\{([0-9a-fA-F]{1,6})}").unwrap_or_else(|e| panic!("{}", e))
-            });
-
-            let v = UNICODE_CODE_POINT_PATTERN.replace_all(json, |caps: &Captures| {
-                // Extract the hex payload (without braces) and parse to a code point.
-                let hex = &caps[1];
-                let Ok(code) = u32::from_str_radix(hex, 16) else {
-                    // On parse failure, leave the original escape sequence unchanged.
-                    return caps[0].to_string();
-                };
-
-                if code <= 0xFFFF {
-                    // Basic Multilingual Plane: JSON can represent this directly as \uXXXX.
-                    format!("\\u{:04X}", code)
-                } else if code > 0x10FFFF {
-                    // Beyond max Unicode code point
-                    caps[0].to_string()
-                } else {
-                    // Supplementary planes: JSON needs UTF-16 surrogate pairs.
-                    // Convert the code point to a 20-bit value.
-                    let code = code - 0x10000;
-
-                    // High surrogate: top 10 bits, offset by 0xD800.
-                    let high = 0xD800 + ((code >> 10) & 0x3FF);
-
-                    // Low surrogate: bottom 10 bits, offset by 0xDC00.
-                    let low = 0xDC00 + (code & 0x3FF);
-
-                    // Emit two \uXXXX escapes in sequence.
-                    format!("\\u{:04X}\\u{:04X}", high, low)
-                }
-            });
-            serde_json::Value::from_str(&v).context(DeserializeSnafu { json })
-        }
-    }
+    serde_json::Value::from_str(json_string.as_str())
+        .context(DeserializeSnafu { json: json_string })
 }
 
 /// Parses a string to a json type value
@@ -491,49 +413,6 @@ mod tests {
     use super::*;
     use crate::json::JsonStructureSettings;
 
-    #[test]
-    fn test_jsonb_string_to_serde_value() -> Result<()> {
-        let valid_cases = vec![
-            (r#"{"data": "simple ascii"}"#, r#"{"data":"simple ascii"}"#),
-            (
-                r#"{"data": "Greek sigma: \u{03a3}"}"#,
-                r#"{"data":"Greek sigma: Σ"}"#,
-            ),
-            (
-                r#"{"data": "Joker card: \u{1f0df}"}"#,
-                r#"{"data":"Joker card: 🃟"}"#,
-            ),
-            (
-                r#"{"data": "BMP boundary: \u{ffff}"}"#,
-                r#"{"data":"BMP boundary: ￿"}"#,
-            ),
-            (
-                r#"{"data": "Supplementary min: \u{10000}"}"#,
-                r#"{"data":"Supplementary min: 𐀀"}"#,
-            ),
-            (
-                r#"{"data": "Supplementary max: \u{10ffff}"}"#,
-                r#"{"data":"Supplementary max: 􏿿"}"#,
-            ),
-        ];
-        for (input, expect) in valid_cases {
-            let v = jsonb_string_to_serde_value(input)?;
-            assert_eq!(v.to_string(), expect);
-        }
-
-        let invalid_cases = vec![
-            r#"{"data": "Invalid hex: \u{gggg}"}"#,
-            r#"{"data": "Beyond max Unicode code point: \u{110000}"}"#,
-            r#"{"data": "Out of range: \u{1100000}"}"#, // 7 digit
-            r#"{"data": "Empty braces: \u{}"}"#,
-        ];
-        for input in invalid_cases {
-            let result = jsonb_string_to_serde_value(input);
-            assert!(result.is_err());
-        }
-        Ok(())
-    }
-
     #[test]
     fn test_json_type_include() {
         fn test(this: &JsonNativeType, that: &JsonNativeType, expected: bool) {
@@ -771,16 +650,15 @@ mod tests {
             "list": [1, 2, 3],
             "object": {"a": 1}
         }"#;
-        let expected =
-            r#"Json<{"hello":"<String>","list":["<Number>"],"object":{"a":"<Number>"}}>"#;
+        let expected = r#"Json<Object{"hello": String, "list": Array[Number(I64)], "object": Object{"a": Number(I64)}}>"#;
         test(json, json_type, Ok(expected))?;
 
         // cannot merge with other non-object json values:
         let jsons = [r#""s""#, "1", "[1]"];
         let expects = [
-            r#"Failed to merge JSON datatype: datatypes have conflict, this: {"hello":"<String>","list":["<Number>"],"object":{"a":"<Number>"}}, that: "<String>""#,
-            r#"Failed to merge JSON datatype: datatypes have conflict, this: {"hello":"<String>","list":["<Number>"],"object":{"a":"<Number>"}}, that: "<Number>""#,
-            r#"Failed to merge JSON datatype: datatypes have conflict, this: {"hello":"<String>","list":["<Number>"],"object":{"a":"<Number>"}}, that: ["<Number>"]"#,
+            r#"Failed to merge JSON datatype: datatypes have conflict, this: Object{"hello": String, "list": Array[Number(I64)], "object": Object{"a": Number(I64)}}, that: String"#,
+            r#"Failed to merge JSON datatype: datatypes have conflict, this: Object{"hello": String, "list": Array[Number(I64)], "object": Object{"a": Number(I64)}}, that: Number(I64)"#,
+            r#"Failed to merge JSON datatype: datatypes have conflict, this: Object{"hello": String, "list": Array[Number(I64)], "object": Object{"a": Number(I64)}}, that: Array[Number(I64)]"#,
         ];
         for (json, expect) in jsons.into_iter().zip(expects.into_iter()) {
             test(json, json_type, Err(expect))?;
@@ -792,7 +670,7 @@ mod tests {
             "float": 0.123,
             "no": 42
         }"#;
-        let expected = r#"Failed to merge JSON datatype: datatypes have conflict, this: "<String>", that: "<Number>""#;
+        let expected = r#"Failed to merge JSON datatype: datatypes have conflict, this: String, that: Number(I64)"#;
         test(json, json_type, Err(expected))?;
 
         // can merge with another json object:
@@ -801,7 +679,7 @@ mod tests {
             "float": 0.123,
             "int": 42
         }"#;
-        let expected = r#"Json<{"float":"<Number>","hello":"<String>","int":"<Number>","list":["<Number>"],"object":{"a":"<Number>"}}>"#;
+        let expected = r#"Json<Object{"float": Number(F64), "hello": String, "int": Number(I64), "list": Array[Number(I64)], "object": Object{"a": Number(I64)}}>"#;
         test(json, json_type, Ok(expected))?;
 
         // can merge with some complex nested json object:
@@ -811,7 +689,7 @@ mod tests {
             "float": 0.456,
             "int": 0
         }"#;
-        let expected = r#"Json<{"float":"<Number>","hello":"<String>","int":"<Number>","list":["<Number>"],"object":{"a":"<Number>","foo":"<String>","l":["<String>"],"o":{"key":"<String>"}}}>"#;
+        let expected = r#"Json<Object{"float": Number(F64), "hello": String, "int": Number(I64), "list": Array[Number(I64)], "object": Object{"a": Number(I64), "foo": String, "l": Array[String], "o": Object{"key": String}}}>"#;
         test(json, json_type, Ok(expected))?;
 
         Ok(())

src/datatypes/src/vectors/json/builder.rs

@@ -321,10 +321,10 @@ mod tests {
             Ok(()),
             Ok(()),
             Err(
-                r#"Failed to merge JSON datatype: datatypes have conflict, this: "<Number>", that: "<String>""#,
+                "Failed to merge JSON datatype: datatypes have conflict, this: Number(I64), that: String",
             ),
             Err(
-                r#"Failed to merge JSON datatype: datatypes have conflict, this: "<Number>", that: ["<Bool>"]"#,
+                "Failed to merge JSON datatype: datatypes have conflict, this: Number(I64), that: Array[Bool]",
             ),
         ];
         let mut builder = JsonVectorBuilder::new(JsonNativeType::Null, 1);
@@ -396,12 +396,12 @@ mod tests {
         // test children builders:
         assert_eq!(builder.builders.len(), 6);
         let expect_types = [
-            r#"Json<{"list":["<Number>"],"s":"<String>"}>"#,
-            r#"Json<{"float":"<Number>","s":"<String>"}>"#,
-            r#"Json<{"float":"<Number>","int":"<Number>"}>"#,
-            r#"Json<{"int":"<Number>","object":{"hello":"<String>","timestamp":"<Number>"}}>"#,
-            r#"Json<{"nested":{"a":{"b":{"b":{"a":"<String>"}}}},"object":{"timestamp":"<Number>"}}>"#,
-            r#"Json<{"nested":{"a":{"b":{"a":{"b":"<String>"}}}},"object":{"timestamp":"<Number>"}}>"#,
+            r#"Json<Object{"list": Array[Number(I64)], "s": String}>"#,
+            r#"Json<Object{"float": Number(F64), "s": String}>"#,
+            r#"Json<Object{"float": Number(F64), "int": Number(I64)}>"#,
+            r#"Json<Object{"int": Number(I64), "object": Object{"hello": String, "timestamp": Number(I64)}}>"#,
+            r#"Json<Object{"nested": Object{"a": Object{"b": Object{"b": Object{"a": String}}}}, "object": Object{"timestamp": Number(I64)}}>"#,
+            r#"Json<Object{"nested": Object{"a": Object{"b": Object{"a": Object{"b": String}}}}, "object": Object{"timestamp": Number(I64)}}>"#,
         ];
         let expect_vectors = [
             r#"
@@ -456,7 +456,7 @@ mod tests {
         }
 
         // test final merged json type:
-        let expected = r#"Json<{"float":"<Number>","int":"<Number>","list":["<Number>"],"nested":{"a":{"b":{"a":{"b":"<String>"},"b":{"a":"<String>"}}}},"object":{"hello":"<String>","timestamp":"<Number>"},"s":"<String>"}>"#;
+        let expected = r#"Json<Object{"float": Number(F64), "int": Number(I64), "list": Array[Number(I64)], "nested": Object{"a": Object{"b": Object{"a": Object{"b": String}, "b": Object{"a": String}}}}, "object": Object{"hello": String, "timestamp": Number(I64)}, "s": String}>"#;
         assert_eq!(builder.data_type().to_string(), expected);
 
         // test final produced vector:

src/meta-srv/src/metasrv.rs

@@ -299,7 +299,7 @@ impl Default for MetasrvOptions {
             #[allow(deprecated)]
             server_addr: String::new(),
             store_addrs: vec!["127.0.0.1:2379".to_string()],
-            backend_tls: Some(TlsOption::prefer()),
+            backend_tls: None,
             selector: SelectorType::default(),
             enable_region_failover: false,
             heartbeat_interval: distributed_time_constants::BASE_HEARTBEAT_INTERVAL,

src/meta-srv/src/procedure/repartition/dispatch.rs

@@ -13,41 +13,18 @@
 // limitations under the License.
 
 use std::any::Any;
-use std::collections::HashMap;
 
 use common_procedure::{Context as ProcedureContext, ProcedureWithId, Status};
 use serde::{Deserialize, Serialize};
-use store_api::storage::RegionId;
 
 use crate::error::Result;
 use crate::procedure::repartition::collect::{Collect, ProcedureMeta};
 use crate::procedure::repartition::group::RepartitionGroupProcedure;
-use crate::procedure::repartition::plan::RegionDescriptor;
 use crate::procedure::repartition::{self, Context, State};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Dispatch;
 
-#[allow(dead_code)]
-fn build_region_mapping(
-    source_regions: &[RegionDescriptor],
-    target_regions: &[RegionDescriptor],
-    transition_map: &[Vec<usize>],
-) -> HashMap<RegionId, Vec<RegionId>> {
-    transition_map
-        .iter()
-        .enumerate()
-        .map(|(source_idx, indices)| {
-            let source_region = source_regions[source_idx].region_id;
-            let target_regions = indices
-                .iter()
-                .map(|&target_idx| target_regions[target_idx].region_id)
-                .collect::<Vec<_>>();
-            (source_region, target_regions)
-        })
-        .collect::<HashMap<RegionId, _>>()
-}
-
 #[async_trait::async_trait]
 #[typetag::serde]
 impl State for Dispatch {
@@ -60,19 +37,11 @@ impl State for Dispatch {
         let mut procedures = Vec::with_capacity(ctx.persistent_ctx.plans.len());
         let mut procedure_metas = Vec::with_capacity(ctx.persistent_ctx.plans.len());
         for (plan_index, plan) in ctx.persistent_ctx.plans.iter().enumerate() {
-            let region_mapping = build_region_mapping(
-                &plan.source_regions,
-                &plan.target_regions,
-                &plan.transition_map,
-            );
             let persistent_ctx = repartition::group::PersistentContext::new(
                 plan.group_id,
                 table_id,
-                ctx.persistent_ctx.catalog_name.clone(),
-                ctx.persistent_ctx.schema_name.clone(),
                 plan.source_regions.clone(),
                 plan.target_regions.clone(),
-                region_mapping,
             );
 
             let group_procedure = RepartitionGroupProcedure::new(persistent_ctx, ctx);

src/meta-srv/src/procedure/repartition/group.rs

@@ -12,34 +12,27 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub(crate) mod apply_staging_manifest;
 pub(crate) mod enter_staging_region;
-pub(crate) mod remap_manifest;
-pub(crate) mod repartition_end;
 pub(crate) mod repartition_start;
 pub(crate) mod update_metadata;
 pub(crate) mod utils;
 
 use std::any::Any;
-use std::collections::HashMap;
 use std::fmt::Debug;
 use std::time::Duration;
 
 use common_error::ext::BoxedError;
+use common_meta::DatanodeId;
 use common_meta::cache_invalidator::CacheInvalidatorRef;
 use common_meta::instruction::CacheIdent;
 use common_meta::key::datanode_table::{DatanodeTableKey, DatanodeTableValue, RegionInfo};
 use common_meta::key::table_route::TableRouteValue;
 use common_meta::key::{DeserializedValueWithBytes, TableMetadataManagerRef};
-use common_meta::lock_key::{CatalogLock, RegionLock, SchemaLock};
-use common_meta::peer::Peer;
 use common_meta::rpc::router::RegionRoute;
-use common_procedure::error::ToJsonSnafu;
 use common_procedure::{
-    Context as ProcedureContext, Error as ProcedureError, LockKey, Procedure,
-    Result as ProcedureResult, Status, StringKey, UserMetadata,
+    Context as ProcedureContext, LockKey, Procedure, Result as ProcedureResult, Status,
+    UserMetadata,
 };
-use common_telemetry::error;
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::{RegionId, TableId};
@@ -78,12 +71,6 @@ impl RepartitionGroupProcedure {
     }
 }
 
-#[derive(Debug, Serialize)]
-pub struct RepartitionGroupData<'a> {
-    persistent_ctx: &'a PersistentContext,
-    state: &'a dyn State,
-}
-
 #[async_trait::async_trait]
 impl Procedure for RepartitionGroupProcedure {
     fn type_name(&self) -> &str {
@@ -91,48 +78,27 @@ impl Procedure for RepartitionGroupProcedure {
     }
 
     async fn execute(&mut self, _ctx: &ProcedureContext) -> ProcedureResult<Status> {
-        let state = &mut self.state;
+        todo!()
+    }
 
-        match state.next(&mut self.context, _ctx).await {
-            Ok((next, status)) => {
-                *state = next;
-                Ok(status)
-            }
-            Err(e) => {
-                if e.is_retryable() {
-                    Err(ProcedureError::retry_later(e))
-                } else {
-                    error!(
-                        e;
-                        "Repartition group procedure failed, group id: {}, table id: {}",
-                        self.context.persistent_ctx.group_id,
-                        self.context.persistent_ctx.table_id,
-                    );
-                    Err(ProcedureError::external(e))
-                }
-            }
-        }
+    async fn rollback(&mut self, _: &ProcedureContext) -> ProcedureResult<()> {
+        todo!()
     }
 
     fn rollback_supported(&self) -> bool {
-        false
+        true
     }
 
     fn dump(&self) -> ProcedureResult<String> {
-        let data = RepartitionGroupData {
-            persistent_ctx: &self.context.persistent_ctx,
-            state: self.state.as_ref(),
-        };
-        serde_json::to_string(&data).context(ToJsonSnafu)
+        todo!()
     }
 
     fn lock_key(&self) -> LockKey {
-        LockKey::new(self.context.persistent_ctx.lock_key())
+        todo!()
     }
 
     fn user_metadata(&self) -> Option<UserMetadata> {
-        // TODO(weny): support user metadata.
-        None
+        todo!()
     }
 }
 
@@ -157,8 +123,8 @@ pub struct GroupPrepareResult {
     pub target_routes: Vec<RegionRoute>,
     /// The primary source region id (first source region), used for retrieving region options.
     pub central_region: RegionId,
-    /// The peer where the primary source region is located.
-    pub central_region_datanode: Peer,
+    /// The datanode id where the primary source region is located.
+    pub central_region_datanode_id: DatanodeId,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
@@ -166,59 +132,30 @@ pub struct PersistentContext {
     pub group_id: GroupId,
     /// The table id of the repartition group.
     pub table_id: TableId,
-    /// The catalog name of the repartition group.
-    pub catalog_name: String,
-    /// The schema name of the repartition group.
-    pub schema_name: String,
     /// The source regions of the repartition group.
     pub sources: Vec<RegionDescriptor>,
     /// The target regions of the repartition group.
     pub targets: Vec<RegionDescriptor>,
-    /// For each `source region`, the corresponding
-    /// `target regions` that overlap with it.
-    pub region_mapping: HashMap<RegionId, Vec<RegionId>>,
     /// The result of group prepare.
     /// The value will be set in [RepartitionStart](crate::procedure::repartition::group::repartition_start::RepartitionStart) state.
     pub group_prepare_result: Option<GroupPrepareResult>,
-    /// The staging manifest paths of the repartition group.
-    /// The value will be set in [RemapManifest](crate::procedure::repartition::group::remap_manifest::RemapManifest) state.
-    pub staging_manifest_paths: HashMap<RegionId, String>,
 }
 
 impl PersistentContext {
     pub fn new(
         group_id: GroupId,
         table_id: TableId,
-        catalog_name: String,
-        schema_name: String,
         sources: Vec<RegionDescriptor>,
         targets: Vec<RegionDescriptor>,
-        region_mapping: HashMap<RegionId, Vec<RegionId>>,
     ) -> Self {
         Self {
             group_id,
             table_id,
-            catalog_name,
-            schema_name,
             sources,
             targets,
-            region_mapping,
             group_prepare_result: None,
-            staging_manifest_paths: HashMap::new(),
         }
     }
-
-    pub fn lock_key(&self) -> Vec<StringKey> {
-        let mut lock_keys = Vec::with_capacity(2 + self.sources.len());
-        lock_keys.extend([
-            CatalogLock::Read(&self.catalog_name).into(),
-            SchemaLock::read(&self.catalog_name, &self.schema_name).into(),
-        ]);
-        for source in &self.sources {
-            lock_keys.push(RegionLock::Write(source.region_id).into());
-        }
-        lock_keys
-    }
 }
 
 impl Context {
@@ -316,7 +253,7 @@ impl Context {
         // Safety: prepare result is set in [RepartitionStart] state.
         let prepare_result = self.persistent_ctx.group_prepare_result.as_ref().unwrap();
         let central_region_datanode_table_value = self
-            .get_datanode_table_value(table_id, prepare_result.central_region_datanode.id)
+            .get_datanode_table_value(table_id, prepare_result.central_region_datanode_id)
             .await?;
         let RegionInfo {
             region_options,

src/meta-srv/src/procedure/repartition/group/apply_staging_manifest.rs

@@ -1,333 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::any::Any;
-use std::collections::HashMap;
-use std::time::{Duration, Instant};
-
-use api::v1::meta::MailboxMessage;
-use common_meta::instruction::{
-    ApplyStagingManifestReply, ApplyStagingManifestsReply, Instruction, InstructionReply,
-};
-use common_meta::peer::Peer;
-use common_meta::rpc::router::RegionRoute;
-use common_procedure::{Context as ProcedureContext, Status};
-use common_telemetry::info;
-use futures::future::join_all;
-use serde::{Deserialize, Serialize};
-use snafu::{OptionExt, ResultExt, ensure};
-use store_api::storage::RegionId;
-
-use crate::error::{self, Error, Result};
-use crate::handler::HeartbeatMailbox;
-use crate::procedure::repartition::group::update_metadata::UpdateMetadata;
-use crate::procedure::repartition::group::utils::{
-    HandleMultipleResult, group_region_routes_by_peer, handle_multiple_results,
-};
-use crate::procedure::repartition::group::{Context, State};
-use crate::procedure::repartition::plan::RegionDescriptor;
-use crate::service::mailbox::{Channel, MailboxRef};
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct ApplyStagingManifest;
-
-#[async_trait::async_trait]
-#[typetag::serde]
-impl State for ApplyStagingManifest {
-    async fn next(
-        &mut self,
-        ctx: &mut Context,
-        _procedure_ctx: &ProcedureContext,
-    ) -> Result<(Box<dyn State>, Status)> {
-        self.apply_staging_manifests(ctx).await?;
-
-        Ok((
-            Box::new(UpdateMetadata::ApplyStaging),
-            Status::executing(true),
-        ))
-    }
-
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-}
-
-impl ApplyStagingManifest {
-    fn build_apply_staging_manifest_instructions(
-        staging_manifest_paths: &HashMap<RegionId, String>,
-        target_routes: &[RegionRoute],
-        targets: &[RegionDescriptor],
-        central_region_id: RegionId,
-    ) -> Result<HashMap<Peer, Vec<common_meta::instruction::ApplyStagingManifest>>> {
-        let target_partition_expr_by_region = targets
-            .iter()
-            .map(|target| {
-                Ok((
-                    target.region_id,
-                    target
-                        .partition_expr
-                        .as_json_str()
-                        .context(error::SerializePartitionExprSnafu)?,
-                ))
-            })
-            .collect::<Result<HashMap<_, _>>>()?;
-        // Safety: `leader_peer` is set for all region routes, checked in `repartition_start`.
-        let target_region_routes_by_peer = group_region_routes_by_peer(target_routes);
-        let mut instructions = HashMap::with_capacity(target_region_routes_by_peer.len());
-
-        for (peer, region_ids) in target_region_routes_by_peer {
-            let apply_staging_manifests = region_ids
-                .into_iter()
-                .map(|region_id| common_meta::instruction::ApplyStagingManifest {
-                    region_id,
-                    partition_expr: target_partition_expr_by_region[&region_id].clone(),
-                    central_region_id,
-                    manifest_path: staging_manifest_paths[&region_id].clone(),
-                })
-                .collect();
-            instructions.insert(peer.clone(), apply_staging_manifests);
-        }
-
-        Ok(instructions)
-    }
-
-    #[allow(dead_code)]
-    async fn apply_staging_manifests(&self, ctx: &mut Context) -> Result<()> {
-        let table_id = ctx.persistent_ctx.table_id;
-        let group_id = ctx.persistent_ctx.group_id;
-        let staging_manifest_paths = &ctx.persistent_ctx.staging_manifest_paths;
-        // Safety: the group prepare result is set in the RepartitionStart state.
-        let prepare_result = ctx.persistent_ctx.group_prepare_result.as_ref().unwrap();
-        let targets = &ctx.persistent_ctx.targets;
-        let target_routes = &prepare_result.target_routes;
-        let central_region_id = prepare_result.central_region;
-        let instructions = Self::build_apply_staging_manifest_instructions(
-            staging_manifest_paths,
-            target_routes,
-            targets,
-            central_region_id,
-        )?;
-        let operation_timeout =
-            ctx.next_operation_timeout()
-                .context(error::ExceededDeadlineSnafu {
-                    operation: "Apply staging manifests",
-                })?;
-
-        let (peers, tasks): (Vec<_>, Vec<_>) = instructions
-            .iter()
-            .map(|(peer, apply_staging_manifests)| {
-                (
-                    peer,
-                    Self::apply_staging_manifest(
-                        &ctx.mailbox,
-                        &ctx.server_addr,
-                        peer,
-                        apply_staging_manifests,
-                        operation_timeout,
-                    ),
-                )
-            })
-            .unzip();
-        info!(
-            "Sent apply staging manifests instructions to peers: {:?} for repartition table {}, group id {}",
-            peers, table_id, group_id
-        );
-
-        let format_err_msg = |idx: usize, error: &Error| {
-            let peer = peers[idx];
-            format!(
-                "Failed to apply staging manifests on datanode {:?}, error: {:?}",
-                peer, error
-            )
-        };
-        // Waits for all tasks to complete.
-        let results = join_all(tasks).await;
-        let result = handle_multiple_results(&results);
-        match result {
-            HandleMultipleResult::AllSuccessful => Ok(()),
-            HandleMultipleResult::AllRetryable(retryable_errors) => error::RetryLaterSnafu {
-                reason: format!(
-                    "All retryable errors during applying staging manifests for repartition table {}, group id {}: {:?}",
-                    table_id, group_id,
-                    retryable_errors
-                        .iter()
-                        .map(|(idx, error)| format_err_msg(*idx, error))
-                        .collect::<Vec<_>>()
-                        .join(",")
-                ),
-            }
-            .fail(),
-            HandleMultipleResult::AllNonRetryable(non_retryable_errors) => error::UnexpectedSnafu {
-                violated: format!(
-                    "All non retryable errors during applying staging manifests for repartition table {}, group id {}: {:?}",
-                    table_id, group_id,
-                    non_retryable_errors
-                        .iter()
-                        .map(|(idx, error)| format_err_msg(*idx, error))
-                        .collect::<Vec<_>>()
-                        .join(",")
-                ),
-            }
-            .fail(),
-            HandleMultipleResult::PartialRetryable {
-                retryable_errors,
-                non_retryable_errors,
-            } => error::UnexpectedSnafu {
-                violated: format!(
-                    "Partial retryable errors during applying staging manifests for repartition table {}, group id {}: {:?}, non retryable errors: {:?}",
-                    table_id, group_id,
-                    retryable_errors
-                        .iter()
-                        .map(|(idx, error)| format_err_msg(*idx, error))
-                        .collect::<Vec<_>>()
-                        .join(","),
-                    non_retryable_errors
-                        .iter()
-                        .map(|(idx, error)| format_err_msg(*idx, error))
-                        .collect::<Vec<_>>()
-                        .join(","),
-                ),
-            }
-            .fail(),
-        }
-    }
-
-    async fn apply_staging_manifest(
-        mailbox: &MailboxRef,
-        server_addr: &str,
-        peer: &Peer,
-        apply_staging_manifests: &[common_meta::instruction::ApplyStagingManifest],
-        timeout: Duration,
-    ) -> Result<()> {
-        let ch = Channel::Datanode(peer.id);
-        let instruction = Instruction::ApplyStagingManifests(apply_staging_manifests.to_vec());
-        let message = MailboxMessage::json_message(
-            &format!(
-                "Apply staging manifests for regions: {:?}",
-                apply_staging_manifests
-                    .iter()
-                    .map(|r| r.region_id)
-                    .collect::<Vec<_>>()
-            ),
-            &format!("Metasrv@{}", server_addr),
-            &format!("Datanode-{}@{}", peer.id, peer.addr),
-            common_time::util::current_time_millis(),
-            &instruction,
-        )
-        .with_context(|_| error::SerializeToJsonSnafu {
-            input: instruction.to_string(),
-        })?;
-
-        let now = Instant::now();
-        let receiver = mailbox.send(&ch, message, timeout).await;
-
-        let receiver = match receiver {
-            Ok(receiver) => receiver,
-            Err(error::Error::PusherNotFound { .. }) => error::RetryLaterSnafu {
-                reason: format!(
-                    "Pusher not found for apply staging manifests on datanode {:?}, elapsed: {:?}",
-                    peer,
-                    now.elapsed()
-                ),
-            }
-            .fail()?,
-            Err(err) => {
-                return Err(err);
-            }
-        };
-
-        match receiver.await {
-            Ok(msg) => {
-                let reply = HeartbeatMailbox::json_reply(&msg)?;
-                info!(
-                    "Received apply staging manifests reply: {:?}, elapsed: {:?}",
-                    reply,
-                    now.elapsed()
-                );
-                let InstructionReply::ApplyStagingManifests(ApplyStagingManifestsReply { replies }) =
-                    reply
-                else {
-                    return error::UnexpectedInstructionReplySnafu {
-                        mailbox_message: msg.to_string(),
-                        reason: "expect apply staging manifests reply",
-                    }
-                    .fail();
-                };
-                for reply in replies {
-                    Self::handle_apply_staging_manifest_reply(&reply, &now, peer)?;
-                }
-
-                Ok(())
-            }
-            Err(error::Error::MailboxTimeout { .. }) => {
-                let reason = format!(
-                    "Mailbox received timeout for apply staging manifests on datanode {:?}, elapsed: {:?}",
-                    peer,
-                    now.elapsed()
-                );
-                error::RetryLaterSnafu { reason }.fail()
-            }
-            Err(err) => Err(err),
-        }
-    }
-
-    fn handle_apply_staging_manifest_reply(
-        ApplyStagingManifestReply {
-            region_id,
-            ready,
-            exists,
-            error,
-        }: &ApplyStagingManifestReply,
-        now: &Instant,
-        peer: &Peer,
-    ) -> Result<()> {
-        ensure!(
-            exists,
-            error::UnexpectedSnafu {
-                violated: format!(
-                    "Region {} doesn't exist on datanode {:?}, elapsed: {:?}",
-                    region_id,
-                    peer,
-                    now.elapsed()
-                )
-            }
-        );
-
-        if error.is_some() {
-            return error::RetryLaterSnafu {
-                reason: format!(
-                    "Failed to apply staging manifest on datanode {:?}, error: {:?}, elapsed: {:?}",
-                    peer,
-                    error,
-                    now.elapsed()
-                ),
-            }
-            .fail();
-        }
-
-        ensure!(
-            ready,
-            error::RetryLaterSnafu {
-                reason: format!(
-                    "Region {} is still applying staging manifest on datanode {:?}, elapsed: {:?}",
-                    region_id,
-                    peer,
-                    now.elapsed()
-                ),
-            }
-        );
-
-        Ok(())
-    }
-}

src/meta-srv/src/procedure/repartition/group/enter_staging_region.rs

@@ -29,7 +29,6 @@ use snafu::{OptionExt, ResultExt, ensure};
 
 use crate::error::{self, Error, Result};
 use crate::handler::HeartbeatMailbox;
-use crate::procedure::repartition::group::remap_manifest::RemapManifest;
 use crate::procedure::repartition::group::utils::{
     HandleMultipleResult, group_region_routes_by_peer, handle_multiple_results,
 };
@@ -50,7 +49,7 @@ impl State for EnterStagingRegion {
     ) -> Result<(Box<dyn State>, Status)> {
         self.enter_staging_regions(ctx).await?;
 
-        Ok((Box::new(RemapManifest), Status::executing(true)))
+        Ok(Self::next_state())
     }
 
     fn as_any(&self) -> &dyn Any {
@@ -59,10 +58,16 @@ impl State for EnterStagingRegion {
 }
 
 impl EnterStagingRegion {
+    #[allow(dead_code)]
+    fn next_state() -> (Box<dyn State>, Status) {
+        // TODO(weny): change it later.
+        (Box::new(EnterStagingRegion), Status::executing(true))
+    }
+
     fn build_enter_staging_instructions(
         prepare_result: &GroupPrepareResult,
         targets: &[RegionDescriptor],
-    ) -> Result<HashMap<Peer, Vec<common_meta::instruction::EnterStagingRegion>>> {
+    ) -> Result<HashMap<Peer, Instruction>> {
         let target_partition_expr_by_region = targets
             .iter()
             .map(|target| {
@@ -88,7 +93,10 @@ impl EnterStagingRegion {
                     partition_expr: target_partition_expr_by_region[&region_id].clone(),
                 })
                 .collect();
-            instructions.insert(peer.clone(), enter_staging_regions);
+            instructions.insert(
+                peer.clone(),
+                Instruction::EnterStagingRegions(enter_staging_regions),
+            );
         }
 
         Ok(instructions)
@@ -109,14 +117,14 @@ impl EnterStagingRegion {
                 })?;
         let (peers, tasks): (Vec<_>, Vec<_>) = instructions
             .iter()
-            .map(|(peer, enter_staging_regions)| {
+            .map(|(peer, instruction)| {
                 (
                     peer,
                     Self::enter_staging_region(
                         &ctx.mailbox,
                         &ctx.server_addr,
                         peer,
-                        enter_staging_regions,
+                        instruction,
                         operation_timeout,
                     ),
                 )
@@ -200,19 +208,12 @@ impl EnterStagingRegion {
         mailbox: &MailboxRef,
         server_addr: &str,
         peer: &Peer,
-        enter_staging_regions: &[common_meta::instruction::EnterStagingRegion],
+        instruction: &Instruction,
         timeout: Duration,
     ) -> Result<()> {
         let ch = Channel::Datanode(peer.id);
-        let instruction = Instruction::EnterStagingRegions(enter_staging_regions.to_vec());
         let message = MailboxMessage::json_message(
-            &format!(
-                "Enter staging regions: {:?}",
-                enter_staging_regions
-                    .iter()
-                    .map(|r| r.region_id)
-                    .collect::<Vec<_>>()
-            ),
+            &format!("Enter staging regions: {:?}", instruction),
             &format!("Metasrv@{}", server_addr),
             &format!("Datanode-{}@{}", peer.id, peer.addr),
             common_time::util::current_time_millis(),
@@ -327,6 +328,7 @@ mod tests {
     use std::assert_matches::assert_matches;
     use std::time::Duration;
 
+    use common_meta::instruction::Instruction;
     use common_meta::peer::Peer;
     use common_meta::rpc::router::{Region, RegionRoute};
     use store_api::storage::RegionId;
@@ -374,7 +376,7 @@ mod tests {
                 },
             ],
             central_region: RegionId::new(table_id, 1),
-            central_region_datanode: Peer::empty(1),
+            central_region_datanode_id: 1,
         };
         let targets = test_targets();
         let instructions =
@@ -382,7 +384,12 @@ mod tests {
                 .unwrap();
 
         assert_eq!(instructions.len(), 2);
-        let instruction_1 = instructions.get(&Peer::empty(1)).unwrap().clone();
+        let instruction_1 = instructions
+            .get(&Peer::empty(1))
+            .unwrap()
+            .clone()
+            .into_enter_staging_regions()
+            .unwrap();
         assert_eq!(
             instruction_1,
             vec![common_meta::instruction::EnterStagingRegion {
@@ -390,7 +397,12 @@ mod tests {
                 partition_expr: range_expr("x", 0, 10).as_json_str().unwrap(),
             }]
         );
-        let instruction_2 = instructions.get(&Peer::empty(2)).unwrap().clone();
+        let instruction_2 = instructions
+            .get(&Peer::empty(2))
+            .unwrap()
+            .clone()
+            .into_enter_staging_regions()
+            .unwrap();
         assert_eq!(
             instruction_2,
             vec![common_meta::instruction::EnterStagingRegion {
@@ -405,17 +417,18 @@ mod tests {
         let env = TestingEnv::new();
         let server_addr = "localhost";
         let peer = Peer::empty(1);
-        let enter_staging_regions = vec![common_meta::instruction::EnterStagingRegion {
-            region_id: RegionId::new(1024, 1),
-            partition_expr: range_expr("x", 0, 10).as_json_str().unwrap(),
-        }];
+        let instruction =
+            Instruction::EnterStagingRegions(vec![common_meta::instruction::EnterStagingRegion {
+                region_id: RegionId::new(1024, 1),
+                partition_expr: range_expr("x", 0, 10).as_json_str().unwrap(),
+            }]);
         let timeout = Duration::from_secs(10);
 
         let err = EnterStagingRegion::enter_staging_region(
             env.mailbox_ctx.mailbox(),
             server_addr,
             &peer,
-            &enter_staging_regions,
+            &instruction,
             timeout,
         )
         .await
@@ -434,10 +447,11 @@ mod tests {
             .await;
         let server_addr = "localhost";
         let peer = Peer::empty(1);
-        let enter_staging_regions = vec![common_meta::instruction::EnterStagingRegion {
-            region_id: RegionId::new(1024, 1),
-            partition_expr: range_expr("x", 0, 10).as_json_str().unwrap(),
-        }];
+        let instruction =
+            Instruction::EnterStagingRegions(vec![common_meta::instruction::EnterStagingRegion {
+                region_id: RegionId::new(1024, 1),
+                partition_expr: range_expr("x", 0, 10).as_json_str().unwrap(),
+            }]);
         let timeout = Duration::from_secs(10);
 
         // Sends a timeout error.
@@ -449,7 +463,7 @@ mod tests {
             env.mailbox_ctx.mailbox(),
             server_addr,
             &peer,
-            &enter_staging_regions,
+            &instruction,
             timeout,
         )
         .await
@@ -465,10 +479,11 @@ mod tests {
 
         let server_addr = "localhost";
         let peer = Peer::empty(1);
-        let enter_staging_regions = vec![common_meta::instruction::EnterStagingRegion {
-            region_id: RegionId::new(1024, 1),
-            partition_expr: range_expr("x", 0, 10).as_json_str().unwrap(),
-        }];
+        let instruction =
+            Instruction::EnterStagingRegions(vec![common_meta::instruction::EnterStagingRegion {
+                region_id: RegionId::new(1024, 1),
+                partition_expr: range_expr("x", 0, 10).as_json_str().unwrap(),
+            }]);
         let timeout = Duration::from_secs(10);
 
         env.mailbox_ctx
@@ -483,7 +498,7 @@ mod tests {
             env.mailbox_ctx.mailbox(),
             server_addr,
             &peer,
-            &enter_staging_regions,
+            &instruction,
             timeout,
         )
         .await
@@ -501,10 +516,11 @@ mod tests {
             .await;
         let server_addr = "localhost";
         let peer = Peer::empty(1);
-        let enter_staging_regions = vec![common_meta::instruction::EnterStagingRegion {
-            region_id: RegionId::new(1024, 1),
-            partition_expr: range_expr("x", 0, 10).as_json_str().unwrap(),
-        }];
+        let instruction =
+            Instruction::EnterStagingRegions(vec![common_meta::instruction::EnterStagingRegion {
+                region_id: RegionId::new(1024, 1),
+                partition_expr: range_expr("x", 0, 10).as_json_str().unwrap(),
+            }]);
         let timeout = Duration::from_secs(10);
 
         // Sends a failed reply.
@@ -522,7 +538,7 @@ mod tests {
             env.mailbox_ctx.mailbox(),
             server_addr,
             &peer,
-            &enter_staging_regions,
+            &instruction,
             timeout,
         )
         .await
@@ -549,7 +565,7 @@ mod tests {
             env.mailbox_ctx.mailbox(),
             server_addr,
             &peer,
-            &enter_staging_regions,
+            &instruction,
             timeout,
         )
         .await
@@ -580,7 +596,7 @@ mod tests {
                 },
             ],
             central_region: RegionId::new(table_id, 1),
-            central_region_datanode: Peer::empty(1),
+            central_region_datanode_id: 1,
         }
     }
 

src/meta-srv/src/procedure/repartition/group/remap_manifest.rs

@@ -1,222 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::any::Any;
-use std::collections::HashMap;
-use std::time::{Duration, Instant};
-
-use api::v1::meta::MailboxMessage;
-use common_meta::instruction::{Instruction, InstructionReply, RemapManifestReply};
-use common_meta::peer::Peer;
-use common_procedure::{Context as ProcedureContext, Status};
-use common_telemetry::{info, warn};
-use serde::{Deserialize, Serialize};
-use snafu::{OptionExt, ResultExt, ensure};
-use store_api::storage::RegionId;
-
-use crate::error::{self, Result};
-use crate::handler::HeartbeatMailbox;
-use crate::procedure::repartition::group::apply_staging_manifest::ApplyStagingManifest;
-use crate::procedure::repartition::group::{Context, State};
-use crate::procedure::repartition::plan::RegionDescriptor;
-use crate::service::mailbox::{Channel, MailboxRef};
-
-#[derive(Debug, Serialize, Deserialize)]
-pub(crate) struct RemapManifest;
-
-#[async_trait::async_trait]
-#[typetag::serde]
-impl State for RemapManifest {
-    async fn next(
-        &mut self,
-        ctx: &mut Context,
-        _procedure_ctx: &ProcedureContext,
-    ) -> Result<(Box<dyn State>, Status)> {
-        let prepare_result = ctx.persistent_ctx.group_prepare_result.as_ref().unwrap();
-        let remap = Self::build_remap_manifest_instructions(
-            &ctx.persistent_ctx.sources,
-            &ctx.persistent_ctx.targets,
-            &ctx.persistent_ctx.region_mapping,
-            prepare_result.central_region,
-        )?;
-        let operation_timeout =
-            ctx.next_operation_timeout()
-                .context(error::ExceededDeadlineSnafu {
-                    operation: "Remap manifests",
-                })?;
-        let manifest_paths = Self::remap_manifests(
-            &ctx.mailbox,
-            &ctx.server_addr,
-            &prepare_result.central_region_datanode,
-            &remap,
-            operation_timeout,
-        )
-        .await?;
-        let table_id = ctx.persistent_ctx.table_id;
-        let group_id = ctx.persistent_ctx.group_id;
-
-        if manifest_paths.len() != ctx.persistent_ctx.targets.len() {
-            warn!(
-                "Mismatch in manifest paths count: expected {}, got {}. This occurred during remapping manifests for group {} and table {}.",
-                ctx.persistent_ctx.targets.len(),
-                manifest_paths.len(),
-                group_id,
-                table_id
-            );
-        }
-
-        ctx.persistent_ctx.staging_manifest_paths = manifest_paths;
-
-        Ok((Box::new(ApplyStagingManifest), Status::executing(true)))
-    }
-
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-}
-
-impl RemapManifest {
-    fn build_remap_manifest_instructions(
-        source_regions: &[RegionDescriptor],
-        target_regions: &[RegionDescriptor],
-        region_mapping: &HashMap<RegionId, Vec<RegionId>>,
-        central_region_id: RegionId,
-    ) -> Result<common_meta::instruction::RemapManifest> {
-        let new_partition_exprs = target_regions
-            .iter()
-            .map(|r| {
-                Ok((
-                    r.region_id,
-                    r.partition_expr
-                        .as_json_str()
-                        .context(error::SerializePartitionExprSnafu)?,
-                ))
-            })
-            .collect::<Result<HashMap<RegionId, String>>>()?;
-
-        Ok(common_meta::instruction::RemapManifest {
-            region_id: central_region_id,
-            input_regions: source_regions.iter().map(|r| r.region_id).collect(),
-            region_mapping: region_mapping.clone(),
-            new_partition_exprs,
-        })
-    }
-
-    async fn remap_manifests(
-        mailbox: &MailboxRef,
-        server_addr: &str,
-        peer: &Peer,
-        remap: &common_meta::instruction::RemapManifest,
-        timeout: Duration,
-    ) -> Result<HashMap<RegionId, String>> {
-        let ch = Channel::Datanode(peer.id);
-        let instruction = Instruction::RemapManifest(remap.clone());
-        let message = MailboxMessage::json_message(
-            &format!(
-                "Remap manifests, central region: {}, input regions: {:?}",
-                remap.region_id, remap.input_regions
-            ),
-            &format!("Metasrv@{}", server_addr),
-            &format!("Datanode-{}@{}", peer.id, peer.addr),
-            common_time::util::current_time_millis(),
-            &instruction,
-        )
-        .with_context(|_| error::SerializeToJsonSnafu {
-            input: instruction.to_string(),
-        })?;
-        let now = Instant::now();
-        let receiver = mailbox.send(&ch, message, timeout).await;
-
-        let receiver = match receiver {
-            Ok(receiver) => receiver,
-            Err(error::Error::PusherNotFound { .. }) => error::RetryLaterSnafu {
-                reason: format!(
-                    "Pusher not found for remap manifests on datanode {:?}, elapsed: {:?}",
-                    peer,
-                    now.elapsed()
-                ),
-            }
-            .fail()?,
-            Err(err) => {
-                return Err(err);
-            }
-        };
-
-        match receiver.await {
-            Ok(msg) => {
-                let reply = HeartbeatMailbox::json_reply(&msg)?;
-                info!(
-                    "Received remap manifest reply: {:?}, elapsed: {:?}",
-                    reply,
-                    now.elapsed()
-                );
-                let InstructionReply::RemapManifest(reply) = reply else {
-                    return error::UnexpectedInstructionReplySnafu {
-                        mailbox_message: msg.to_string(),
-                        reason: "expect remap manifest reply",
-                    }
-                    .fail();
-                };
-
-                Self::handle_remap_manifest_reply(remap.region_id, reply, &now, peer)
-            }
-            Err(error::Error::MailboxTimeout { .. }) => {
-                let reason = format!(
-                    "Mailbox received timeout for remap manifests on datanode {:?}, elapsed: {:?}",
-                    peer,
-                    now.elapsed()
-                );
-                error::RetryLaterSnafu { reason }.fail()
-            }
-            Err(err) => Err(err),
-        }
-    }
-
-    fn handle_remap_manifest_reply(
-        region_id: RegionId,
-        RemapManifestReply {
-            exists,
-            manifest_paths,
-            error,
-        }: RemapManifestReply,
-        now: &Instant,
-        peer: &Peer,
-    ) -> Result<HashMap<RegionId, String>> {
-        ensure!(
-            exists,
-            error::UnexpectedSnafu {
-                violated: format!(
-                    "Region {} doesn't exist on datanode {:?}, elapsed: {:?}",
-                    region_id,
-                    peer,
-                    now.elapsed()
-                )
-            }
-        );
-
-        if error.is_some() {
-            return error::RetryLaterSnafu {
-                reason: format!(
-                    "Failed to remap manifest on datanode {:?}, error: {:?}, elapsed: {:?}",
-                    peer,
-                    error,
-                    now.elapsed()
-                ),
-            }
-            .fail();
-        }
-
-        Ok(manifest_paths)
-    }
-}

src/meta-srv/src/procedure/repartition/group/repartition_end.rs

@@ -1,40 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::any::Any;
-
-use common_procedure::{Context as ProcedureContext, Status};
-use serde::{Deserialize, Serialize};
-
-use crate::error::Result;
-use crate::procedure::repartition::group::{Context, State};
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct RepartitionEnd;
-
-#[async_trait::async_trait]
-#[typetag::serde]
-impl State for RepartitionEnd {
-    async fn next(
-        &mut self,
-        _ctx: &mut Context,
-        _procedure_ctx: &ProcedureContext,
-    ) -> Result<(Box<dyn State>, Status)> {
-        Ok((Box::new(RepartitionEnd), Status::done()))
-    }
-
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-}

src/meta-srv/src/procedure/repartition/group/repartition_start.rs

@@ -22,7 +22,6 @@ use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt, ensure};
 
 use crate::error::{self, Result};
-use crate::procedure::repartition::group::update_metadata::UpdateMetadata;
 use crate::procedure::repartition::group::{
     Context, GroupId, GroupPrepareResult, State, region_routes,
 };
@@ -110,7 +109,7 @@ impl RepartitionStart {
             );
         }
         let central_region = sources[0].region_id;
-        let central_region_datanode = source_region_routes[0]
+        let central_region_datanode_id = source_region_routes[0]
             .leader_peer
             .as_ref()
             .context(error::UnexpectedSnafu {
@@ -119,22 +118,20 @@ impl RepartitionStart {
                     central_region
                 ),
             })?
-            .clone();
+            .id;
 
         Ok(GroupPrepareResult {
             source_routes: source_region_routes,
             target_routes: target_region_routes,
             central_region,
-            central_region_datanode,
+            central_region_datanode_id,
         })
     }
 
     #[allow(dead_code)]
     fn next_state() -> (Box<dyn State>, Status) {
-        (
-            Box::new(UpdateMetadata::ApplyStaging),
-            Status::executing(true),
-        )
+        // TODO(weny): change it later.
+        (Box::new(RepartitionStart), Status::executing(true))
     }
 }
 

src/meta-srv/src/procedure/repartition/group/update_metadata.rs

@@ -17,14 +17,12 @@ pub(crate) mod rollback_staging_region;
 
 use std::any::Any;
 
-use common_meta::lock_key::TableLock;
 use common_procedure::{Context as ProcedureContext, Status};
 use common_telemetry::warn;
 use serde::{Deserialize, Serialize};
 
 use crate::error::Result;
-use crate::procedure::repartition::group::enter_staging_region::EnterStagingRegion;
-use crate::procedure::repartition::group::repartition_end::RepartitionEnd;
+use crate::procedure::repartition::group::repartition_start::RepartitionStart;
 use crate::procedure::repartition::group::{Context, State};
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -35,16 +33,22 @@ pub enum UpdateMetadata {
     RollbackStaging,
 }
 
+impl UpdateMetadata {
+    #[allow(dead_code)]
+    fn next_state() -> (Box<dyn State>, Status) {
+        // TODO(weny): change it later.
+        (Box::new(RepartitionStart), Status::executing(true))
+    }
+}
+
 #[async_trait::async_trait]
 #[typetag::serde]
 impl State for UpdateMetadata {
     async fn next(
         &mut self,
         ctx: &mut Context,
-        procedure_ctx: &ProcedureContext,
+        _procedure_ctx: &ProcedureContext,
     ) -> Result<(Box<dyn State>, Status)> {
-        let table_lock = TableLock::Write(ctx.persistent_ctx.table_id).into();
-        let _guard = procedure_ctx.provider.acquire_lock(&table_lock).await;
         match self {
             UpdateMetadata::ApplyStaging => {
                 // TODO(weny): If all metadata have already been updated, skip applying staging regions.
@@ -55,7 +59,7 @@ impl State for UpdateMetadata {
                         "Failed to broadcast the invalidate table cache message during the apply staging regions, error: {err:?}"
                     );
                 };
-                Ok((Box::new(EnterStagingRegion), Status::executing(false)))
+                Ok(Self::next_state())
             }
             UpdateMetadata::RollbackStaging => {
                 self.rollback_staging_regions(ctx).await?;
@@ -65,7 +69,7 @@ impl State for UpdateMetadata {
                         "Failed to broadcast the invalidate table cache message during the rollback staging regions, error: {err:?}"
                     );
                 };
-                Ok((Box::new(RepartitionEnd), Status::executing(false)))
+                Ok(Self::next_state())
             }
         }
     }

src/meta-srv/src/procedure/repartition/plan.rs

@@ -41,9 +41,6 @@ pub struct AllocationPlanEntry {
     pub regions_to_allocate: usize,
     /// The number of regions that need to be deallocated (source count - target count, if positive).
     pub regions_to_deallocate: usize,
-    /// For each `source_regions[k]`, the corresponding vector contains global
-    /// `target_partition_exprs` that overlap with it.
-    pub transition_map: Vec<Vec<usize>>,
 }
 
 /// A plan entry for the dispatch phase after region allocation,
@@ -60,9 +57,6 @@ pub struct RepartitionPlanEntry {
     pub allocated_region_ids: Vec<RegionId>,
     /// The region ids of the regions that are pending deallocation.
     pub pending_deallocate_region_ids: Vec<RegionId>,
-    /// For each `source_regions[k]`, the corresponding vector contains global
-    /// `target_regions` that overlap with it.
-    pub transition_map: Vec<Vec<usize>>,
 }
 
 impl RepartitionPlanEntry {
@@ -77,7 +71,6 @@ impl RepartitionPlanEntry {
             target_partition_exprs,
             regions_to_allocate,
             regions_to_deallocate,
-            transition_map,
         }: &AllocationPlanEntry,
     ) -> Self {
         debug_assert!(*regions_to_allocate == 0 && *regions_to_deallocate == 0);
@@ -96,7 +89,6 @@ impl RepartitionPlanEntry {
             target_regions,
             allocated_region_ids: vec![],
             pending_deallocate_region_ids: vec![],
-            transition_map: transition_map.clone(),
         }
     }
 }

src/meta-srv/src/procedure/repartition/repartition_start.rs

@@ -129,7 +129,6 @@ impl RepartitionStart {
                     target_partition_exprs,
                     regions_to_allocate,
                     regions_to_deallocate,
-                    transition_map: subtask.transition_map,
                 }
             })
             .collect::<Vec<_>>()

src/meta-srv/src/procedure/repartition/test_util.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashMap;
 use std::sync::Arc;
 
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
@@ -88,13 +87,9 @@ pub fn new_persistent_context(
 ) -> PersistentContext {
     PersistentContext {
         group_id: Uuid::new_v4(),
-        catalog_name: "test_catalog".to_string(),
-        schema_name: "test_schema".to_string(),
         table_id,
         sources,
         targets,
-        region_mapping: HashMap::new(),
         group_prepare_result: None,
-        staging_manifest_paths: HashMap::new(),
     }
 }

src/meta-srv/src/utils/etcd.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 use common_meta::kv_backend::etcd::create_etcd_tls_options;
-use common_telemetry::warn;
 use etcd_client::{Client, ConnectOptions};
 use servers::tls::{TlsMode, TlsOption};
 use snafu::ResultExt;
@@ -39,12 +38,9 @@ pub async fn create_etcd_client_with_tls(
             client_options.keep_alive_interval,
             client_options.keep_alive_timeout,
         );
-
-    let all_endpoints_use_https = etcd_endpoints.iter().all(|e| e.starts_with("https"));
     if let Some(tls_config) = tls_config
-        && let Some(tls_options) =
-            create_etcd_tls_options(&convert_tls_option(all_endpoints_use_https, tls_config))
-                .context(BuildTlsOptionsSnafu)?
+        && let Some(tls_options) = create_etcd_tls_options(&convert_tls_option(tls_config))
+            .context(BuildTlsOptionsSnafu)?
     {
         connect_options = connect_options.with_tls(tls_options);
     }
@@ -54,22 +50,9 @@ pub async fn create_etcd_client_with_tls(
         .context(error::ConnectEtcdSnafu)
 }
 
-fn convert_tls_option(
-    all_endpoints_use_https: bool,
-    tls_option: &TlsOption,
-) -> common_meta::kv_backend::etcd::TlsOption {
+fn convert_tls_option(tls_option: &TlsOption) -> common_meta::kv_backend::etcd::TlsOption {
     let mode = match tls_option.mode {
         TlsMode::Disable => common_meta::kv_backend::etcd::TlsMode::Disable,
-        TlsMode::Prefer => {
-            if all_endpoints_use_https {
-                common_meta::kv_backend::etcd::TlsMode::Require
-            } else {
-                warn!(
-                    "All endpoints use HTTP, TLS prefer mode is not supported, using disable mode"
-                );
-                common_meta::kv_backend::etcd::TlsMode::Disable
-            }
-        }
         _ => common_meta::kv_backend::etcd::TlsMode::Require,
     };
     common_meta::kv_backend::etcd::TlsOption {

src/pipeline/Cargo.toml

@@ -13,7 +13,6 @@ workspace = true
 ahash.workspace = true
 api.workspace = true
 arrow.workspace = true
-arrow-schema.workspace = true
 async-trait.workspace = true
 catalog.workspace = true
 chrono.workspace = true

src/pipeline/src/error.rs

@@ -609,14 +609,10 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Multiple pipelines with different schemas found, but none under current schema. Please replicate one of them or delete until only one schema left. name: {}, current_schema: {}, schemas: {}",
-        name,
-        current_schema,
-        schemas,
+        "Multiple pipelines with different schemas found, but none under current schema. Please replicate one of them or delete until only one schema left. schemas: {}",
+        schemas
     ))]
     MultiPipelineWithDiffSchema {
-        name: String,
-        current_schema: String,
         schemas: String,
         #[snafu(implicit)]
         location: Location,
@@ -804,20 +800,6 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
-
-    #[snafu(transparent)]
-    GreptimeProto {
-        source: api::error::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
-    #[snafu(transparent)]
-    Datatypes {
-        source: datatypes::error::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -938,9 +920,6 @@ impl ErrorExt for Error {
             FloatIsNan { .. }
             | InvalidEpochForResolution { .. }
             | UnsupportedTypeInPipeline { .. } => StatusCode::InvalidArguments,
-
-            GreptimeProto { source, .. } => source.status_code(),
-            Datatypes { source, .. } => source.status_code(),
         }
     }
 

src/pipeline/src/etl/transform/transformer/greptime.rs

@@ -19,17 +19,13 @@ use std::collections::{BTreeMap, HashSet};
 use std::sync::Arc;
 
 use ahash::{HashMap, HashMapExt};
-use api::helper::{ColumnDataTypeWrapper, encode_json_value};
-use api::v1::column_def::{collect_column_options, options_from_column_schema};
+use api::helper::proto_value_type;
+use api::v1::column_data_type_extension::TypeExt;
 use api::v1::value::ValueData;
-use api::v1::{ColumnDataType, SemanticType};
-use arrow_schema::extension::ExtensionType;
+use api::v1::{ColumnDataType, ColumnDataTypeExtension, JsonTypeExtension, SemanticType};
 use coerce::{coerce_columns, coerce_value};
 use common_query::prelude::{greptime_timestamp, greptime_value};
 use common_telemetry::warn;
-use datatypes::data_type::ConcreteDataType;
-use datatypes::extension::json::JsonExtensionType;
-use datatypes::value::Value;
 use greptime_proto::v1::{ColumnSchema, Row, Rows, Value as GreptimeValue};
 use itertools::Itertools;
 use jsonb::Number;
@@ -37,15 +33,12 @@ use once_cell::sync::OnceCell;
 use serde_json as serde_json_crate;
 use session::context::Channel;
 use snafu::OptionExt;
-use table::Table;
 use vrl::prelude::{Bytes, VrlValueConvert};
-use vrl::value::value::StdError;
 use vrl::value::{KeyString, Value as VrlValue};
 
 use crate::error::{
-    ArrayElementMustBeObjectSnafu, CoerceIncompatibleTypesSnafu,
-    IdentifyPipelineColumnTypeMismatchSnafu, InvalidTimestampSnafu, Result,
-    TimeIndexMustBeNonNullSnafu, TransformColumnNameMustBeUniqueSnafu,
+    ArrayElementMustBeObjectSnafu, IdentifyPipelineColumnTypeMismatchSnafu, InvalidTimestampSnafu,
+    Result, TimeIndexMustBeNonNullSnafu, TransformColumnNameMustBeUniqueSnafu,
     TransformMultipleTimestampIndexSnafu, TransformTimestampIndexCountSnafu, ValueMustBeMapSnafu,
 };
 use crate::etl::PipelineDocVersion;
@@ -279,75 +272,15 @@ impl GreptimeTransformer {
     }
 }
 
-#[derive(Clone)]
-pub struct ColumnMetadata {
-    column_schema: datatypes::schema::ColumnSchema,
-    semantic_type: SemanticType,
-}
-
-impl From<ColumnSchema> for ColumnMetadata {
-    fn from(value: ColumnSchema) -> Self {
-        let datatype = value.datatype();
-        let semantic_type = value.semantic_type();
-        let ColumnSchema {
-            column_name,
-            datatype: _,
-            semantic_type: _,
-            datatype_extension,
-            options,
-        } = value;
-
-        let column_schema = datatypes::schema::ColumnSchema::new(
-            column_name,
-            ColumnDataTypeWrapper::new(datatype, datatype_extension).into(),
-            semantic_type != SemanticType::Timestamp,
-        );
-
-        let metadata = collect_column_options(options.as_ref());
-        let column_schema = column_schema.with_metadata(metadata);
-
-        Self {
-            column_schema,
-            semantic_type,
-        }
-    }
-}
-
-impl TryFrom<ColumnMetadata> for ColumnSchema {
-    type Error = api::error::Error;
-
-    fn try_from(value: ColumnMetadata) -> std::result::Result<Self, Self::Error> {
-        let ColumnMetadata {
-            column_schema,
-            semantic_type,
-        } = value;
-
-        let options = options_from_column_schema(&column_schema);
-
-        let (datatype, datatype_extension) =
-            ColumnDataTypeWrapper::try_from(column_schema.data_type).map(|x| x.into_parts())?;
-
-        Ok(ColumnSchema {
-            column_name: column_schema.name,
-            datatype: datatype as _,
-            semantic_type: semantic_type as _,
-            datatype_extension,
-            options,
-        })
-    }
-}
-
 /// This is used to record the current state schema information and a sequential cache of field names.
 /// As you traverse the user input JSON, this will change.
 /// It will record a superset of all user input schemas.
-#[derive(Default)]
+#[derive(Debug, Default)]
 pub struct SchemaInfo {
     /// schema info
-    pub schema: Vec<ColumnMetadata>,
+    pub schema: Vec<ColumnSchema>,
     /// index of the column name
     pub index: HashMap<String, usize>,
-    /// The pipeline's corresponding table (if already created). Useful to retrieve column schemas.
-    table: Option<Arc<Table>>,
 }
 
 impl SchemaInfo {
@@ -355,7 +288,6 @@ impl SchemaInfo {
         Self {
             schema: Vec::with_capacity(capacity),
             index: HashMap::with_capacity(capacity),
-            table: None,
         }
     }
 
@@ -365,88 +297,46 @@ impl SchemaInfo {
             index.insert(schema.column_name.clone(), i);
         }
         Self {
-            schema: schema_list.into_iter().map(Into::into).collect(),
+            schema: schema_list,
             index,
-            table: None,
         }
     }
-
-    pub fn set_table(&mut self, table: Option<Arc<Table>>) {
-        self.table = table;
-    }
-
-    fn find_column_schema_in_table(&self, column_name: &str) -> Option<ColumnMetadata> {
-        if let Some(table) = &self.table
-            && let Some(i) = table.schema_ref().column_index_by_name(column_name)
-        {
-            let column_schema = table.schema_ref().column_schemas()[i].clone();
-
-            let semantic_type = if column_schema.is_time_index() {
-                SemanticType::Timestamp
-            } else if table.table_info().meta.primary_key_indices.contains(&i) {
-                SemanticType::Tag
-            } else {
-                SemanticType::Field
-            };
-
-            Some(ColumnMetadata {
-                column_schema,
-                semantic_type,
-            })
-        } else {
-            None
-        }
-    }
-
-    pub fn column_schemas(&self) -> api::error::Result<Vec<ColumnSchema>> {
-        self.schema
-            .iter()
-            .map(|x| x.clone().try_into())
-            .collect::<api::error::Result<Vec<_>>>()
-    }
 }
 
 fn resolve_schema(
     index: Option<usize>,
-    pipeline_context: &PipelineContext,
-    column: &str,
-    value_type: &ConcreteDataType,
+    value_data: ValueData,
+    column_schema: ColumnSchema,
+    row: &mut Vec<GreptimeValue>,
     schema_info: &mut SchemaInfo,
 ) -> Result<()> {
     if let Some(index) = index {
-        let column_type = &mut schema_info.schema[index].column_schema.data_type;
-        match (column_type, value_type) {
-            (column_type, value_type) if column_type == value_type => Ok(()),
-            (ConcreteDataType::Json(column_type), ConcreteDataType::Json(value_type))
-                if column_type.is_include(value_type) =>
-            {
-                Ok(())
+        let api_value = GreptimeValue {
+            value_data: Some(value_data),
+        };
+        // Safety unwrap is fine here because api_value is always valid
+        let value_column_data_type = proto_value_type(&api_value).unwrap();
+        // Safety unwrap is fine here because index is always valid
+        let schema_column_data_type = schema_info.schema.get(index).unwrap().datatype();
+        if value_column_data_type != schema_column_data_type {
+            IdentifyPipelineColumnTypeMismatchSnafu {
+                column: column_schema.column_name,
+                expected: schema_column_data_type.as_str_name(),
+                actual: value_column_data_type.as_str_name(),
             }
-            (column_type, value_type) => IdentifyPipelineColumnTypeMismatchSnafu {
-                column,
-                expected: column_type.to_string(),
-                actual: value_type.to_string(),
-            }
-            .fail(),
+            .fail()
+        } else {
+            row[index] = api_value;
+            Ok(())
         }
     } else {
-        let column_schema = schema_info
-            .find_column_schema_in_table(column)
-            .unwrap_or_else(|| {
-                let semantic_type = decide_semantic(pipeline_context, column);
-                let column_schema = datatypes::schema::ColumnSchema::new(
-                    column,
-                    value_type.clone(),
-                    semantic_type != SemanticType::Timestamp,
-                );
-                ColumnMetadata {
-                    column_schema,
-                    semantic_type,
-                }
-            });
-        let key = column.to_string();
+        let key = column_schema.column_name.clone();
         schema_info.schema.push(column_schema);
         schema_info.index.insert(key, schema_info.schema.len() - 1);
+        let api_value = GreptimeValue {
+            value_data: Some(value_data),
+        };
+        row.push(api_value);
         Ok(())
     }
 }
@@ -591,11 +481,11 @@ pub(crate) fn values_to_row(
     Ok(Row { values: row })
 }
 
-fn decide_semantic(p_ctx: &PipelineContext, column_name: &str) -> SemanticType {
+fn decide_semantic(p_ctx: &PipelineContext, column_name: &str) -> i32 {
     if p_ctx.channel == Channel::Prometheus && column_name != greptime_value() {
-        SemanticType::Tag
+        SemanticType::Tag as i32
     } else {
-        SemanticType::Field
+        SemanticType::Field as i32
     }
 }
 
@@ -607,56 +497,55 @@ fn resolve_value(
     p_ctx: &PipelineContext,
 ) -> Result<()> {
     let index = schema_info.index.get(&column_name).copied();
+    let mut resolve_simple_type =
+        |value_data: ValueData, column_name: String, data_type: ColumnDataType| {
+            let semantic_type = decide_semantic(p_ctx, &column_name);
+            resolve_schema(
+                index,
+                value_data,
+                ColumnSchema {
+                    column_name,
+                    datatype: data_type as i32,
+                    semantic_type,
+                    datatype_extension: None,
+                    options: None,
+                },
+                row,
+                schema_info,
+            )
+        };
 
-    let value_data = match value {
-        VrlValue::Null => None,
+    match value {
+        VrlValue::Null => {}
 
         VrlValue::Integer(v) => {
             // safe unwrap after type matched
-            resolve_schema(
-                index,
-                p_ctx,
-                &column_name,
-                &ConcreteDataType::int64_datatype(),
-                schema_info,
-            )?;
-            Some(ValueData::I64Value(v))
+            resolve_simple_type(ValueData::I64Value(v), column_name, ColumnDataType::Int64)?;
         }
 
         VrlValue::Float(v) => {
             // safe unwrap after type matched
-            resolve_schema(
-                index,
-                p_ctx,
-                &column_name,
-                &ConcreteDataType::float64_datatype(),
-                schema_info,
+            resolve_simple_type(
+                ValueData::F64Value(v.into()),
+                column_name,
+                ColumnDataType::Float64,
             )?;
-            Some(ValueData::F64Value(v.into()))
         }
 
         VrlValue::Boolean(v) => {
-            resolve_schema(
-                index,
-                p_ctx,
-                &column_name,
-                &ConcreteDataType::boolean_datatype(),
-                schema_info,
+            resolve_simple_type(
+                ValueData::BoolValue(v),
+                column_name,
+                ColumnDataType::Boolean,
             )?;
-            Some(ValueData::BoolValue(v))
         }
 
         VrlValue::Bytes(v) => {
-            resolve_schema(
-                index,
-                p_ctx,
-                &column_name,
-                &ConcreteDataType::string_datatype(),
-                schema_info,
+            resolve_simple_type(
+                ValueData::StringValue(String::from_utf8_lossy_owned(v.to_vec())),
+                column_name,
+                ColumnDataType::String,
             )?;
-            Some(ValueData::StringValue(String::from_utf8_lossy_owned(
-                v.to_vec(),
-            )))
         }
 
         VrlValue::Regex(v) => {
@@ -664,83 +553,42 @@ fn resolve_value(
                 "Persisting regex value in the table, this should not happen, column_name: {}",
                 column_name
             );
-            resolve_schema(
-                index,
-                p_ctx,
-                &column_name,
-                &ConcreteDataType::string_datatype(),
-                schema_info,
+            resolve_simple_type(
+                ValueData::StringValue(v.to_string()),
+                column_name,
+                ColumnDataType::String,
             )?;
-            Some(ValueData::StringValue(v.to_string()))
         }
 
         VrlValue::Timestamp(ts) => {
             let ns = ts.timestamp_nanos_opt().context(InvalidTimestampSnafu {
                 input: ts.to_rfc3339(),
             })?;
-            resolve_schema(
-                index,
-                p_ctx,
-                &column_name,
-                &ConcreteDataType::timestamp_nanosecond_datatype(),
-                schema_info,
+            resolve_simple_type(
+                ValueData::TimestampNanosecondValue(ns),
+                column_name,
+                ColumnDataType::TimestampNanosecond,
             )?;
-            Some(ValueData::TimestampNanosecondValue(ns))
         }
 
         VrlValue::Array(_) | VrlValue::Object(_) => {
-            let is_json_native_type = schema_info
-                .find_column_schema_in_table(&column_name)
-                .is_some_and(|x| {
-                    if let ConcreteDataType::Json(column_type) = &x.column_schema.data_type {
-                        column_type.is_native_type()
-                    } else {
-                        false
-                    }
-                });
-
-            let value = if is_json_native_type {
-                let json_extension_type: Option<JsonExtensionType> =
-                    if let Some(x) = schema_info.find_column_schema_in_table(&column_name) {
-                        x.column_schema.extension_type()?
-                    } else {
-                        None
-                    };
-                let settings = json_extension_type
-                    .and_then(|x| x.metadata().json_structure_settings.clone())
-                    .unwrap_or_default();
-                let value: serde_json::Value = value.try_into().map_err(|e: StdError| {
-                    CoerceIncompatibleTypesSnafu { msg: e.to_string() }.build()
-                })?;
-                let value = settings.encode(value)?;
-
-                resolve_schema(index, p_ctx, &column_name, &value.data_type(), schema_info)?;
-
-                let Value::Json(value) = value else {
-                    unreachable!()
-                };
-                ValueData::JsonValue(encode_json_value(*value))
-            } else {
-                resolve_schema(
-                    index,
-                    p_ctx,
-                    &column_name,
-                    &ConcreteDataType::binary_datatype(),
-                    schema_info,
-                )?;
-
-                let value = vrl_value_to_jsonb_value(&value);
-                ValueData::BinaryValue(value.to_vec())
-            };
-            Some(value)
+            let data = vrl_value_to_jsonb_value(&value);
+            resolve_schema(
+                index,
+                ValueData::BinaryValue(data.to_vec()),
+                ColumnSchema {
+                    column_name,
+                    datatype: ColumnDataType::Binary as i32,
+                    semantic_type: SemanticType::Field as i32,
+                    datatype_extension: Some(ColumnDataTypeExtension {
+                        type_ext: Some(TypeExt::JsonType(JsonTypeExtension::JsonBinary.into())),
+                    }),
+                    options: None,
+                },
+                row,
+                schema_info,
+            )?;
         }
-    };
-
-    let value = GreptimeValue { value_data };
-    if let Some(index) = index {
-        row[index] = value;
-    } else {
-        row.push(value);
     }
     Ok(())
 }
@@ -778,24 +626,20 @@ fn identity_pipeline_inner(
     let custom_ts = pipeline_ctx.pipeline_definition.get_custom_ts();
 
     // set time index column schema first
-    let column_schema = datatypes::schema::ColumnSchema::new(
-        custom_ts
+    schema_info.schema.push(ColumnSchema {
+        column_name: custom_ts
             .map(|ts| ts.get_column_name().to_string())
             .unwrap_or_else(|| greptime_timestamp().to_string()),
-        custom_ts
-            .map(|c| ConcreteDataType::from(ColumnDataTypeWrapper::new(c.get_datatype(), None)))
-            .unwrap_or_else(|| {
-                if pipeline_ctx.channel == Channel::Prometheus {
-                    ConcreteDataType::timestamp_millisecond_datatype()
-                } else {
-                    ConcreteDataType::timestamp_nanosecond_datatype()
-                }
-            }),
-        false,
-    );
-    schema_info.schema.push(ColumnMetadata {
-        column_schema,
-        semantic_type: SemanticType::Timestamp,
+        datatype: custom_ts.map(|c| c.get_datatype()).unwrap_or_else(|| {
+            if pipeline_ctx.channel == Channel::Prometheus {
+                ColumnDataType::TimestampMillisecond
+            } else {
+                ColumnDataType::TimestampNanosecond
+            }
+        }) as i32,
+        semantic_type: SemanticType::Timestamp as i32,
+        datatype_extension: None,
+        options: None,
     });
 
     let mut opt_map = HashMap::new();
@@ -853,29 +697,28 @@ pub fn identity_pipeline(
         input.push(result);
     }
 
-    identity_pipeline_inner(input, pipeline_ctx).and_then(|(mut schema, opt_map)| {
+    identity_pipeline_inner(input, pipeline_ctx).map(|(mut schema, opt_map)| {
         if let Some(table) = table {
             let table_info = table.table_info();
             for tag_name in table_info.meta.row_key_column_names() {
                 if let Some(index) = schema.index.get(tag_name) {
-                    schema.schema[*index].semantic_type = SemanticType::Tag;
+                    schema.schema[*index].semantic_type = SemanticType::Tag as i32;
                 }
             }
         }
 
-        let column_schemas = schema.column_schemas()?;
-        Ok(opt_map
+        opt_map
             .into_iter()
             .map(|(opt, rows)| {
                 (
                     opt,
                     Rows {
-                        schema: column_schemas.clone(),
+                        schema: schema.schema.clone(),
                         rows,
                     },
                 )
             })
-            .collect::<HashMap<ContextOpt, Rows>>())
+            .collect::<HashMap<ContextOpt, Rows>>()
     })
 }
 
@@ -1007,7 +850,7 @@ mod tests {
             assert!(rows.is_err());
             assert_eq!(
                 rows.err().unwrap().to_string(),
-                "Column datatype mismatch. For column: score, expected datatype: Float64, actual datatype: String".to_string(),
+                "Column datatype mismatch. For column: score, expected datatype: FLOAT64, actual datatype: STRING".to_string(),
             );
         }
         {
@@ -1036,7 +879,7 @@ mod tests {
             assert!(rows.is_err());
             assert_eq!(
                 rows.err().unwrap().to_string(),
-                "Column datatype mismatch. For column: score, expected datatype: Float64, actual datatype: Int64".to_string(),
+                "Column datatype mismatch. For column: score, expected datatype: FLOAT64, actual datatype: INT64".to_string(),
             );
         }
         {
@@ -1099,7 +942,7 @@ mod tests {
                     .map(|(mut schema, mut rows)| {
                         for name in tag_column_names {
                             if let Some(index) = schema.index.get(&name) {
-                                schema.schema[*index].semantic_type = SemanticType::Tag;
+                                schema.schema[*index].semantic_type = SemanticType::Tag as i32;
                             }
                         }
 
@@ -1107,7 +950,7 @@ mod tests {
                         let rows = rows.remove(&ContextOpt::default()).unwrap();
 
                         Rows {
-                            schema: schema.column_schemas().unwrap(),
+                            schema: schema.schema,
                             rows,
                         }
                     });

src/pipeline/src/manager/pipeline_cache.rs

@@ -15,7 +15,6 @@
 use std::sync::Arc;
 use std::time::Duration;
 
-use common_telemetry::debug;
 use datatypes::timestamp::TimestampNanosecond;
 use moka::sync::Cache;
 
@@ -34,18 +33,10 @@ const PIPELINES_CACHE_TTL: Duration = Duration::from_secs(10);
 /// to encapsulate inner cache. Only public methods are exposed.
 pub(crate) struct PipelineCache {
     pipelines: Cache<String, Arc<Pipeline>>,
-    original_pipelines: Cache<String, PipelineContent>,
+    original_pipelines: Cache<String, (String, TimestampNanosecond)>,
     /// If the pipeline table is invalid, we can use this cache to prevent failures when writing logs through the pipeline
     /// The failover cache never expires, but it will be updated when the pipelines cache is updated.
-    failover_cache: Cache<String, PipelineContent>,
-}
-
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub struct PipelineContent {
-    pub name: String,
-    pub content: String,
-    pub version: TimestampNanosecond,
-    pub schema: String,
+    failover_cache: Cache<String, (String, TimestampNanosecond)>,
 }
 
 impl PipelineCache {
@@ -54,17 +45,12 @@ impl PipelineCache {
             pipelines: Cache::builder()
                 .max_capacity(PIPELINES_CACHE_SIZE)
                 .time_to_live(PIPELINES_CACHE_TTL)
-                .name("pipelines")
                 .build(),
             original_pipelines: Cache::builder()
                 .max_capacity(PIPELINES_CACHE_SIZE)
                 .time_to_live(PIPELINES_CACHE_TTL)
-                .name("original_pipelines")
-                .build(),
-            failover_cache: Cache::builder()
-                .max_capacity(PIPELINES_CACHE_SIZE)
-                .name("failover_cache")
                 .build(),
+            failover_cache: Cache::builder().max_capacity(PIPELINES_CACHE_SIZE).build(),
         }
     }
 
@@ -86,15 +72,19 @@ impl PipelineCache {
         );
     }
 
-    pub(crate) fn insert_pipeline_str_cache(&self, pipeline: &PipelineContent, with_latest: bool) {
-        let schema = pipeline.schema.as_str();
-        let name = pipeline.name.as_str();
-        let version = pipeline.version;
+    pub(crate) fn insert_pipeline_str_cache(
+        &self,
+        schema: &str,
+        name: &str,
+        version: PipelineVersion,
+        pipeline: (String, TimestampNanosecond),
+        with_latest: bool,
+    ) {
         insert_cache_generic(
             &self.original_pipelines,
             schema,
             name,
-            Some(version),
+            version,
             pipeline.clone(),
             with_latest,
         );
@@ -102,8 +92,8 @@ impl PipelineCache {
             &self.failover_cache,
             schema,
             name,
-            Some(version),
-            pipeline.clone(),
+            version,
+            pipeline,
             with_latest,
         );
     }
@@ -122,7 +112,7 @@ impl PipelineCache {
         schema: &str,
         name: &str,
         version: PipelineVersion,
-    ) -> Result<Option<PipelineContent>> {
+    ) -> Result<Option<(String, TimestampNanosecond)>> {
         get_cache_generic(&self.failover_cache, schema, name, version)
     }
 
@@ -131,7 +121,7 @@ impl PipelineCache {
         schema: &str,
         name: &str,
         version: PipelineVersion,
-    ) -> Result<Option<PipelineContent>> {
+    ) -> Result<Option<(String, TimestampNanosecond)>> {
         get_cache_generic(&self.original_pipelines, schema, name, version)
     }
 
@@ -184,13 +174,13 @@ fn get_cache_generic<T: Clone + Send + Sync + 'static>(
     version: PipelineVersion,
 ) -> Result<Option<T>> {
     // lets try empty schema first
-    let emp_key = generate_pipeline_cache_key(EMPTY_SCHEMA_NAME, name, version);
-    if let Some(value) = cache.get(&emp_key) {
+    let k = generate_pipeline_cache_key(EMPTY_SCHEMA_NAME, name, version);
+    if let Some(value) = cache.get(&k) {
         return Ok(Some(value));
     }
     // use input schema
-    let schema_k = generate_pipeline_cache_key(schema, name, version);
-    if let Some(value) = cache.get(&schema_k) {
+    let k = generate_pipeline_cache_key(schema, name, version);
+    if let Some(value) = cache.get(&k) {
         return Ok(Some(value));
     }
 
@@ -203,28 +193,14 @@ fn get_cache_generic<T: Clone + Send + Sync + 'static>(
 
     match ks.len() {
         0 => Ok(None),
-        1 => {
-            let (_, value) = ks.remove(0);
-            Ok(Some(value))
-        }
-        _ => {
-            debug!(
-                "caches keys: {:?}, emp key: {:?}, schema key: {:?}, suffix key: {:?}",
-                cache.iter().map(|e| e.0).collect::<Vec<_>>(),
-                emp_key,
-                schema_k,
-                suffix_key
-            );
-            MultiPipelineWithDiffSchemaSnafu {
-                name: name.to_string(),
-                current_schema: schema.to_string(),
-                schemas: ks
-                    .iter()
-                    .filter_map(|(k, _)| k.split_once('/').map(|k| k.0))
-                    .collect::<Vec<_>>()
-                    .join(","),
-            }
-            .fail()?
+        1 => Ok(Some(ks.remove(0).1)),
+        _ => MultiPipelineWithDiffSchemaSnafu {
+            schemas: ks
+                .iter()
+                .filter_map(|(k, _)| k.split_once('/').map(|k| k.0))
+                .collect::<Vec<_>>()
+                .join(","),
         }
+        .fail()?,
     }
 }

src/pipeline/src/manager/pipeline_operator.rs

@@ -220,7 +220,6 @@ impl PipelineOperator {
                     .observe(timer.elapsed().as_secs_f64())
             })
             .await
-            .map(|p| (p.content, p.version))
     }
 
     /// Insert a pipeline into the pipeline table.

src/pipeline/src/manager/table.rs

@@ -46,7 +46,7 @@ use crate::error::{
     MultiPipelineWithDiffSchemaSnafu, PipelineNotFoundSnafu, RecordBatchLenNotMatchSnafu, Result,
 };
 use crate::etl::{Content, Pipeline, parse};
-use crate::manager::pipeline_cache::{PipelineCache, PipelineContent};
+use crate::manager::pipeline_cache::PipelineCache;
 use crate::manager::{PipelineInfo, PipelineVersion};
 use crate::metrics::METRIC_PIPELINE_TABLE_FIND_COUNT;
 use crate::util::prepare_dataframe_conditions;
@@ -260,22 +260,17 @@ impl PipelineTable {
         &self,
         schema: &str,
         name: &str,
-        input_version: PipelineVersion,
+        version: PipelineVersion,
     ) -> Result<Arc<Pipeline>> {
-        if let Some(pipeline) = self.cache.get_pipeline_cache(schema, name, input_version)? {
+        if let Some(pipeline) = self.cache.get_pipeline_cache(schema, name, version)? {
             return Ok(pipeline);
         }
 
-        let pipeline_content = self.get_pipeline_str(schema, name, input_version).await?;
-        let compiled_pipeline = Arc::new(Self::compile_pipeline(&pipeline_content.content)?);
+        let pipeline = self.get_pipeline_str(schema, name, version).await?;
+        let compiled_pipeline = Arc::new(Self::compile_pipeline(&pipeline.0)?);
 
-        self.cache.insert_pipeline_cache(
-            &pipeline_content.schema,
-            name,
-            Some(pipeline_content.version),
-            compiled_pipeline.clone(),
-            input_version.is_none(),
-        );
+        self.cache
+            .insert_pipeline_cache(schema, name, version, compiled_pipeline.clone(), false);
         Ok(compiled_pipeline)
     }
 
@@ -285,17 +280,14 @@ impl PipelineTable {
         &self,
         schema: &str,
         name: &str,
-        input_version: PipelineVersion,
-    ) -> Result<PipelineContent> {
-        if let Some(pipeline) = self
-            .cache
-            .get_pipeline_str_cache(schema, name, input_version)?
-        {
+        version: PipelineVersion,
+    ) -> Result<(String, TimestampNanosecond)> {
+        if let Some(pipeline) = self.cache.get_pipeline_str_cache(schema, name, version)? {
             return Ok(pipeline);
         }
 
         let mut pipeline_vec;
-        match self.find_pipeline(name, input_version).await {
+        match self.find_pipeline(name, version).await {
             Ok(p) => {
                 METRIC_PIPELINE_TABLE_FIND_COUNT
                     .with_label_values(&["true"])
@@ -312,11 +304,8 @@ impl PipelineTable {
                             .inc();
                         return self
                             .cache
-                            .get_failover_cache(schema, name, input_version)?
-                            .context(PipelineNotFoundSnafu {
-                                name,
-                                version: input_version,
-                            });
+                            .get_failover_cache(schema, name, version)?
+                            .ok_or(PipelineNotFoundSnafu { name, version }.build());
                     }
                     _ => {
                         // if other error, we should return it
@@ -327,40 +316,42 @@ impl PipelineTable {
         };
         ensure!(
             !pipeline_vec.is_empty(),
-            PipelineNotFoundSnafu {
-                name,
-                version: input_version
-            }
+            PipelineNotFoundSnafu { name, version }
         );
 
         // if the result is exact one, use it
         if pipeline_vec.len() == 1 {
-            let pipeline_content = pipeline_vec.remove(0);
-
-            self.cache
-                .insert_pipeline_str_cache(&pipeline_content, input_version.is_none());
-            return Ok(pipeline_content);
+            let (pipeline_content, found_schema, version) = pipeline_vec.remove(0);
+            let p = (pipeline_content, version);
+            self.cache.insert_pipeline_str_cache(
+                &found_schema,
+                name,
+                Some(version),
+                p.clone(),
+                false,
+            );
+            return Ok(p);
         }
 
         // check if there's empty schema pipeline
         // if there isn't, check current schema
         let pipeline = pipeline_vec
             .iter()
-            .position(|v| v.schema == EMPTY_SCHEMA_NAME)
-            .or_else(|| pipeline_vec.iter().position(|v| v.schema == schema))
-            .map(|idx| pipeline_vec.remove(idx));
+            .find(|v| v.1 == EMPTY_SCHEMA_NAME)
+            .or_else(|| pipeline_vec.iter().find(|v| v.1 == schema));
 
         // multiple pipeline with no empty or current schema
         // throw an error
-        let pipeline_content = pipeline.with_context(|| MultiPipelineWithDiffSchemaSnafu {
-            name: name.to_string(),
-            current_schema: schema.to_string(),
-            schemas: pipeline_vec.iter().map(|v| v.schema.clone()).join(","),
-        })?;
+        let (pipeline_content, found_schema, version) =
+            pipeline.context(MultiPipelineWithDiffSchemaSnafu {
+                schemas: pipeline_vec.iter().map(|v| v.1.clone()).join(","),
+            })?;
 
+        let v = *version;
+        let p = (pipeline_content.clone(), v);
         self.cache
-            .insert_pipeline_str_cache(&pipeline_content, input_version.is_none());
-        Ok(pipeline_content)
+            .insert_pipeline_str_cache(found_schema, name, Some(v), p.clone(), false);
+        Ok(p)
     }
 
     /// Insert a pipeline into the pipeline table and compile it.
@@ -387,15 +378,13 @@ impl PipelineTable {
                 true,
             );
 
-            let pipeline_content = PipelineContent {
-                name: name.to_string(),
-                content: pipeline.to_string(),
-                version: TimestampNanosecond(version),
-                schema: EMPTY_SCHEMA_NAME.to_string(),
-            };
-
-            self.cache
-                .insert_pipeline_str_cache(&pipeline_content, true);
+            self.cache.insert_pipeline_str_cache(
+                EMPTY_SCHEMA_NAME,
+                name,
+                Some(TimestampNanosecond(version)),
+                (pipeline.to_owned(), TimestampNanosecond(version)),
+                true,
+            );
         }
 
         Ok((version, compiled_pipeline))
@@ -477,7 +466,7 @@ impl PipelineTable {
         &self,
         name: &str,
         version: PipelineVersion,
-    ) -> Result<Vec<PipelineContent>> {
+    ) -> Result<Vec<(String, String, TimestampNanosecond)>> {
         // 1. prepare dataframe
         let dataframe = self
             .query_engine
@@ -577,12 +566,11 @@ impl PipelineTable {
 
             let len = pipeline_content.len();
             for i in 0..len {
-                re.push(PipelineContent {
-                    name: name.to_string(),
-                    content: pipeline_content.value(i).to_string(),
-                    version: TimestampNanosecond::new(pipeline_created_at.value(i)),
-                    schema: pipeline_schema.value(i).to_string(),
-                });
+                re.push((
+                    pipeline_content.value(i).to_string(),
+                    pipeline_schema.value(i).to_string(),
+                    TimestampNanosecond::new(pipeline_created_at.value(i)),
+                ));
             }
         }
 

src/pipeline/tests/common.rs

@@ -61,7 +61,7 @@ pub fn parse_and_exec(input_str: &str, pipeline_yaml: &str) -> Rows {
     }
 
     Rows {
-        schema: schema_info.column_schemas().unwrap(),
+        schema: schema_info.schema.clone(),
         rows,
     }
 }

src/pipeline/tests/json_parse.rs

@@ -52,7 +52,7 @@ transform:
 
     // check schema
     assert_eq!(output.schema[0].column_name, "commit");
-    let type_id: i32 = ColumnDataType::Json.into();
+    let type_id: i32 = ColumnDataType::Binary.into();
     assert_eq!(output.schema[0].datatype, type_id);
 
     // check value
@@ -91,7 +91,7 @@ transform:
 
     // check schema
     assert_eq!(output.schema[0].column_name, "commit_json");
-    let type_id: i32 = ColumnDataType::Json.into();
+    let type_id: i32 = ColumnDataType::Binary.into();
     assert_eq!(output.schema[0].datatype, type_id);
 
     // check value
@@ -160,7 +160,7 @@ transform:
 
     // check schema
     assert_eq!(output.schema[0].column_name, "commit");
-    let type_id: i32 = ColumnDataType::Json.into();
+    let type_id: i32 = ColumnDataType::Binary.into();
     assert_eq!(output.schema[0].datatype, type_id);
 
     // check value

src/query/src/dist_plan/analyzer.rs

@@ -281,18 +281,18 @@ struct PlanRewriter {
     /// 2: Sort: t.pk1+t.pk2
     /// 3. Projection: t.number, t.pk1, t.pk2
     /// ```
-    /// `Sort` will make a column requirement for `t.pk1+t.pk2` at level 2.
+    /// `Sort` will make a column requirement for `t.pk1` at level 2.
     /// Which making `Projection` at level 1 need to add a ref to `t.pk1` as well.
     /// So that the expanded plan will be
     /// ```ignore
     /// Projection: t.number
-    ///   MergeSort: t.pk1+t.pk2
+    ///   MergeSort: t.pk1
     ///     MergeScan: remote_input=
     /// Projection: t.number, "t.pk1+t.pk2" <--- the original `Projection` at level 1 get added with `t.pk1+t.pk2`
     ///  Sort: t.pk1+t.pk2
     ///    Projection: t.number, t.pk1, t.pk2
     /// ```
-    /// Making `MergeSort` can have `t.pk1+t.pk2` as input.
+    /// Making `MergeSort` can have `t.pk1` as input.
     /// Meanwhile `Projection` at level 3 doesn't need to add any new column because 3 > 2
     /// and col requirements at level 2 is not applicable for level 3.
     ///
@@ -392,11 +392,10 @@ impl PlanRewriter {
                     && ext_b.node.name() == MergeSortLogicalPlan::name()
                 {
                     // revert last `ConditionalCommutative` result for Sort plan in this case.
-                    // also need to remove any column requirements made by the Sort Plan
-                    // as it may refer to columns later no longer exist(rightfully) like by aggregate or projection
+                    // `update_column_requirements` left unchanged because Sort won't generate
+                    // new columns or remove existing columns.
                     self.stage.pop();
                     self.expand_on_next_part_cond_trans_commutative = false;
-                    self.column_requirements.clear();
                 }
             }
             Commutativity::PartialCommutative => {
@@ -681,10 +680,6 @@ struct EnforceDistRequirementRewriter {
 
 impl EnforceDistRequirementRewriter {
     fn new(column_requirements: Vec<(HashSet<Column>, usize)>, cur_level: usize) -> Self {
-        debug!(
-            "Create EnforceDistRequirementRewriter with column_requirements: {:?} at cur_level: {}",
-            column_requirements, cur_level
-        );
         Self {
             column_requirements,
             cur_level,
@@ -738,7 +733,7 @@ impl EnforceDistRequirementRewriter {
                             .filter(|a| !a.is_empty())
                         else {
                             return Err(datafusion_common::DataFusionError::Internal(format!(
-                                "EnforceDistRequirementRewriter: no alias found for required column {original_col} at level {level} in current node's child plan: \n{child} from original plan: \n{original}",
+                                "EnforceDistRequirementRewriter: no alias found for required column {original_col} in child plan {child} from original plan {original}",
                             )));
                         };
 

src/query/src/dist_plan/analyzer/test.rs

@@ -777,67 +777,6 @@ fn expand_step_aggr_proj() {
     assert_eq!(expected, result.to_string());
 }
 
-/// Make sure that `SeriesDivide` special handling correctly clean up column requirements from it's previous sort
-#[test]
-fn expand_complex_col_req_sort_pql() {
-    // use logging for better debugging
-    init_default_ut_logging();
-    let test_table = TestTable::table_with_name(0, "t".to_string());
-    let table_source = Arc::new(DefaultTableSource::new(Arc::new(
-        DfTableProviderAdapter::new(test_table),
-    )));
-
-    let plan = LogicalPlanBuilder::scan_with_filters("t", table_source.clone(), None, vec![])
-        .unwrap()
-        .sort(vec![
-            col("pk1").sort(true, false),
-            col("pk2").sort(true, false),
-            col("pk3").sort(true, false), // make some col req here
-        ])
-        .unwrap()
-        .build()
-        .unwrap();
-    let plan = SeriesDivide::new(
-        vec!["pk1".to_string(), "pk2".to_string(), "pk3".to_string()],
-        "ts".to_string(),
-        plan,
-    );
-    let plan = LogicalPlan::Extension(datafusion_expr::Extension {
-        node: Arc::new(plan),
-    });
-
-    let plan = LogicalPlanBuilder::from(plan)
-        .aggregate(vec![col("pk1"), col("pk2")], vec![min(col("number"))])
-        .unwrap()
-        .sort(vec![
-            col("pk1").sort(true, false),
-            col("pk2").sort(true, false),
-        ])
-        .unwrap()
-        .project(vec![col("pk1"), col("pk2")])
-        .unwrap()
-        .build()
-        .unwrap();
-
-    let config = ConfigOptions::default();
-    let result = DistPlannerAnalyzer {}.analyze(plan, &config).unwrap();
-
-    let expected = [
-        "Projection: t.pk1, t.pk2",
-        "  MergeSort: t.pk1 ASC NULLS LAST, t.pk2 ASC NULLS LAST",
-        "    MergeScan [is_placeholder=false, remote_input=[",
-        "Projection: t.pk1, t.pk2",
-        "  Sort: t.pk1 ASC NULLS LAST, t.pk2 ASC NULLS LAST",
-        "    Aggregate: groupBy=[[t.pk1, t.pk2]], aggr=[[min(t.number)]]",
-        r#"      PromSeriesDivide: tags=["pk1", "pk2", "pk3"]"#,
-        "        Sort: t.pk1 ASC NULLS LAST, t.pk2 ASC NULLS LAST, t.pk3 ASC NULLS LAST",
-        "          TableScan: t",
-        "]]",
-    ]
-    .join("\n");
-    assert_eq!(expected, result.to_string());
-}
-
 /// should only expand `Sort`, notice `Sort` before `Aggregate` usually can and
 /// will be optimized out, and dist planner shouldn't handle that case, but
 /// for now, still handle that be expanding the `Sort` node

src/query/src/dist_plan/commutativity.rs

@@ -144,7 +144,7 @@ impl Categorizer {
                     }
                 }
                 // all group by expressions are partition columns can push down, unless
-                // another push down(including `Limit` or `Sort`) is already in progress(which will then prevent next cond commutative node from being push down).
+                // another push down(including `Limit` or `Sort`) is already in progress(which will then prvent next cond commutative node from being push down).
                 // TODO(discord9): This is a temporary solution(that works), a better description of
                 // commutativity is needed under this situation.
                 Commutativity::ConditionalCommutative(None)

src/query/src/query_engine/state.rs

@@ -234,7 +234,7 @@ impl QueryEngineState {
         rules.retain(|rule| rule.name() != name);
     }
 
-    /// Optimize the logical plan by the extension analyzer rules.
+    /// Optimize the logical plan by the extension anayzer rules.
     pub fn optimize_by_extension_rules(
         &self,
         plan: DfLogicalPlan,

src/servers/src/error.rs

@@ -664,13 +664,6 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
-
-    #[snafu(transparent)]
-    GreptimeProto {
-        source: api::error::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
 }
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;
@@ -801,8 +794,6 @@ impl ErrorExt for Error {
             Suspended { .. } => StatusCode::Suspended,
 
             MemoryLimitExceeded { .. } => StatusCode::RateLimited,
-
-            GreptimeProto { source, .. } => source.status_code(),
         }
     }
 

src/servers/src/grpc.rs

@@ -23,7 +23,6 @@ pub mod memory_limit;
 pub mod prom_query_gateway;
 pub mod region_server;
 
-use std::any::Any;
 use std::net::SocketAddr;
 use std::time::Duration;
 
@@ -400,8 +399,4 @@ impl Server for GrpcServer {
     fn bind_addr(&self) -> Option<SocketAddr> {
         self.bind_addr
     }
-
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
 }

src/servers/src/http.rs

@@ -1285,10 +1285,6 @@ impl Server for HttpServer {
     fn bind_addr(&self) -> Option<SocketAddr> {
         self.bind_addr
     }
-
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
 }
 
 #[cfg(test)]

src/servers/src/http/event.rs

@@ -31,7 +31,7 @@ use axum_extra::TypedHeader;
 use common_catalog::consts::default_engine;
 use common_error::ext::{BoxedError, ErrorExt};
 use common_query::{Output, OutputData};
-use common_telemetry::{error, warn};
+use common_telemetry::{debug, error, warn};
 use headers::ContentType;
 use lazy_static::lazy_static;
 use mime_guess::mime;
@@ -738,6 +738,11 @@ pub async fn log_ingester(
 
     let value = extract_pipeline_value_by_content_type(content_type, payload, ignore_errors)?;
 
+    debug!(
+        "receiving logs: {:?}",
+        serde_json::to_string(&value).unwrap()
+    );
+
     query_ctx.set_channel(Channel::Log);
     let query_ctx = Arc::new(query_ctx);
 

src/servers/src/http/loki.rs

@@ -152,7 +152,7 @@ pub async fn loki_ingest(
             rows.push(row);
         }
 
-        let schemas = schema_info.column_schemas()?;
+        let schemas = schema_info.schema;
         // fill Null for missing values
         for row in rows.iter_mut() {
             row.resize(schemas.len(), GreptimeValue::default());
@@ -746,16 +746,13 @@ fn process_labels(
         } else {
             // not exist
             // add schema and append to values
-            schemas.push(
-                ColumnSchema {
-                    column_name: k.clone(),
-                    datatype: ColumnDataType::String.into(),
-                    semantic_type: SemanticType::Tag.into(),
-                    datatype_extension: None,
-                    options: None,
-                }
-                .into(),
-            );
+            schemas.push(ColumnSchema {
+                column_name: k.clone(),
+                datatype: ColumnDataType::String.into(),
+                semantic_type: SemanticType::Tag.into(),
+                datatype_extension: None,
+                options: None,
+            });
             column_indexer.insert(k, schemas.len() - 1);
 
             row.push(GreptimeValue {

src/servers/src/mysql/server.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::any::Any;
 use std::future::Future;
 use std::net::SocketAddr;
 use std::sync::Arc;
@@ -266,8 +265,4 @@ impl Server for MysqlServer {
     fn bind_addr(&self) -> Option<SocketAddr> {
         self.bind_addr
     }
-
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
 }

src/servers/src/otlp/logs.rs

@@ -381,7 +381,6 @@ fn extract_field_from_attr_and_combine_schema(
 
         if let Some(index) = select_schema.index.get(key) {
             let column_schema = &select_schema.schema[*index];
-            let column_schema: ColumnSchema = column_schema.clone().try_into()?;
             // datatype of the same column name should be the same
             ensure!(
                 column_schema.datatype == schema.datatype,
@@ -394,7 +393,7 @@ fn extract_field_from_attr_and_combine_schema(
             );
             extracted_values[*index] = value;
         } else {
-            select_schema.schema.push(schema.into());
+            select_schema.schema.push(schema);
             select_schema
                 .index
                 .insert(key.clone(), select_schema.schema.len() - 1);
@@ -481,7 +480,7 @@ fn parse_export_logs_service_request_to_rows(
     let mut parse_ctx = ParseContext::new(select_info);
     let mut rows = parse_resource(&mut parse_ctx, request.resource_logs)?;
 
-    schemas.extend(parse_ctx.select_schema.column_schemas()?);
+    schemas.extend(parse_ctx.select_schema.schema);
 
     rows.iter_mut().for_each(|row| {
         row.values.resize(schemas.len(), GreptimeValue::default());

src/servers/src/pipeline.rs

@@ -135,18 +135,12 @@ async fn run_custom_pipeline(
             let mut schema_info = SchemaInfo::default();
             schema_info
                 .schema
-                .push(time_index_column_schema(ts_name, timeunit).into());
+                .push(time_index_column_schema(ts_name, timeunit));
 
             schema_info
         }
     };
 
-    let table = handler
-        .get_table(&table_name, query_ctx)
-        .await
-        .context(CatalogSnafu)?;
-    schema_info.set_table(table);
-
     for pipeline_map in pipeline_maps {
         let result = pipeline
             .exec_mut(pipeline_map, pipeline_ctx, &mut schema_info)
@@ -200,7 +194,7 @@ async fn run_custom_pipeline(
             RowInsertRequest {
                 rows: Some(Rows {
                     rows,
-                    schema: schema_info.column_schemas()?,
+                    schema: schema_info.schema.clone(),
                 }),
                 table_name: table_name.clone(),
             },

src/servers/src/postgres/server.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::any::Any;
 use std::future::Future;
 use std::net::SocketAddr;
 use std::sync::Arc;
@@ -145,8 +144,4 @@ impl Server for PostgresServer {
     fn bind_addr(&self) -> Option<SocketAddr> {
         self.bind_addr
     }
-
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
 }

src/servers/src/server.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::any::Any;
 use std::collections::HashMap;
 use std::fmt::{Debug, Formatter};
 use std::net::SocketAddr;
@@ -148,8 +147,6 @@ pub trait Server: Send + Sync {
     fn bind_addr(&self) -> Option<SocketAddr> {
         None
     }
-
-    fn as_any(&self) -> &dyn Any;
 }
 
 struct AcceptTask {

src/servers/src/tls.rs

@@ -29,7 +29,7 @@ use strum::EnumString;
 use crate::error::{InternalIoSnafu, Result};
 
 /// TlsMode is used for Mysql and Postgres server start up.
-#[derive(Debug, Default, Serialize, Deserialize, Clone, Copy, PartialEq, Eq, EnumString)]
+#[derive(Debug, Default, Serialize, Deserialize, Clone, PartialEq, Eq, EnumString)]
 #[serde(rename_all = "snake_case")]
 pub enum TlsMode {
     #[default]
@@ -91,17 +91,6 @@ impl TlsOption {
         tls_option
     }
 
-    /// Creates a new TLS option with the prefer mode.
-    pub fn prefer() -> Self {
-        Self {
-            mode: TlsMode::Prefer,
-            cert_path: String::new(),
-            key_path: String::new(),
-            ca_cert_path: String::new(),
-            watch: false,
-        }
-    }
-
     /// Validates the TLS configuration.
     ///
     /// Returns an error if:

tests-integration/resources/jsonbench-select-all.txt

@@ -1,14 +1,14 @@
-+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+----------------------------+
-| data                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                       | time_us                    |
-+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+----------------------------+
-| {_raw: {"commit":{"cid":"bafyreidblutgvj75o4q4akzyyejedjj6l3it6hgqwee6jpwv2wqph5fsgm","record":{"$type":"app.bsky.feed.post","createdAt":"2024-11-21T16:09:27.095Z","langs":["en"],"reply":{"parent":{"cid":"bafyreibfglofvqou2yiqvwzk4rcgkhhxrbunyemshdjledgwymimqkg24e","uri":"at://did:plc:6tr6tuzlx2db3rduzr2d6r24/app.bsky.feed.post/3lbhqo2rtys2z"},"root":{"cid":"bafyreibfglofvqou2yiqvwzk4rcgkhhxrbunyemshdjledgwymimqkg24e","uri":"at://did:plc:6tr6tuzlx2db3rduzr2d6r24/app.bsky.feed.post/3lbhqo2rtys2z"}},"text":"aaaaah.  LIght shines in a corner of WTF...."},"rev":"3lbhtytnn2k2f","rkey":"3lbhtyteurk2y"}}, commit.collection: app.bsky.feed.post, commit.operation: create, did: did:plc:yj3sjq3blzpynh27cumnp5ks, kind: commit, time_us: 1732206349000167}                                                                                                                             | 2024-11-21T16:25:49.000167 |
-| {_raw: {"commit":{"cid":"bafyreia5l5vrkh5oj4cjyhcqby2dprhyvcyofo2q5562tijlae2pzih23m","record":{"$type":"app.bsky.feed.like","createdAt":"2024-11-21T16:25:46.221Z","subject":{"cid":"bafyreidjvrcmckkm765mct5fph36x7kupkfo35rjklbf2k76xkzwyiauge","uri":"at://did:plc:azrv4rcbws6kmcga4fsbphg2/app.bsky.feed.post/3lbgjdpbiec2l"}},"rev":"3lbhuvzds6d2a","rkey":"3lbhuvzdked2a"}}, commit.collection: app.bsky.feed.like, commit.operation: create, did: did:plc:3i4xf2v4wcnyktgv6satke64, kind: commit, time_us: 1732206349000644}                                                                                                                                                                                                                                                                                                                                                                       | 2024-11-21T16:25:49.000644 |
-| {_raw: {"commit":{"cid":"bafyreiew2p6cgirfaj45qoenm4fgumib7xoloclrap3jgkz5es7g7kby3i","record":{"$type":"app.bsky.graph.follow","createdAt":"2024-11-21T16:27:40.923Z","subject":"did:plc:r7cdh4sgzqbfdc6wcdxxti7c"},"rev":"3lbhuvze3gi2u","rkey":"3lbhuvzdtmi2u"}}, commit.collection: app.bsky.graph.follow, commit.operation: create, did: did:plc:gccfnqqizz4urhchsaie6jft, kind: commit, time_us: 1732206349001108}                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   | 2024-11-21T16:25:49.001108 |
-| {_raw: {"commit":{"cid":"bafyreidjix4dauj2afjlbzmhj3a7gwftcevvmmy6edww6vrjdbst26rkby","record":{"$type":"app.bsky.feed.like","createdAt":"2024-11-21T16:15:58.232Z","subject":{"cid":"bafyreiavpshyqzrlo5m7fqodjhs6jevweqnif4phasiwimv4a7mnsqi2fe","uri":"at://did:plc:fusulxqc52zbrc75fi6xrcof/app.bsky.feed.post/3lbhskq5zn22f"}},"rev":"3lbhueija5p22","rkey":"3lbhueiizcx22"}}, commit.collection: app.bsky.feed.like, commit.operation: create, did: did:plc:msxqf3twq7abtdw7dbfskphk, kind: commit, time_us: 1732206349001372}                                                                                                                                                                                                                                                                                                                                                                       | 2024-11-21T16:25:49.001372 |
-| {_raw: {"commit":{"cid":"bafyreigzdsdne3z2xxcakgisieyj7y47hj6eg7lj6v4q25ah5q2qotu5ku","record":{"$type":"app.bsky.feed.post","createdAt":"2024-11-21T16:09:27.254Z","langs":["en"],"reply":{"parent":{"cid":"bafyreih35fe2jj3gchmgk4amold4l6sfxd2sby5wrg3jrws5fkdypxrbg4","uri":"at://did:plc:6wx2gg5yqgvmlu35r6y3bk6d/app.bsky.feed.post/3lbhtj2eb4s2o"},"root":{"cid":"bafyreifipyt3vctd4ptuoicvio7rbr5xvjv4afwuggnd2prnmn55mu6luu","uri":"at://did:plc:474ldquxwzrlcvjhhbbk2wte/app.bsky.feed.post/3lbhdzrynik27"}},"text":"okay i take mine back because I hadn’t heard this one yet^^"},"rev":"3lbhtytohxc2o","rkey":"3lbhtytjqzk2q"}}, commit.collection: app.bsky.feed.post, commit.operation: create, did: did:plc:l5o3qjrmfztir54cpwlv2eme, kind: commit, time_us: 1732206349001905}                                                                                                              | 2024-11-21T16:25:49.001905 |
-| {_raw: {"commit":{"cid":"bafyreiaa2vsdr4ckwjg4jq47zfd7mewidywfz3qh3dmglcd6ozi4xwdega","record":{"$type":"app.bsky.graph.follow","createdAt":"2024-11-21T16:15:21.495Z","subject":"did:plc:amsdn2tbjxo3xrwqneqhh4cm"},"rev":"3lbhudfo3yi2w","rkey":"3lbhudfnw4y2w"}}, commit.collection: app.bsky.graph.follow, commit.operation: create, did: did:plc:jkaaf5j2yb2pvpx3ualm3vbh, kind: commit, time_us: 1732206349002758}                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   | 2024-11-21T16:25:49.002758 |
-| {_raw: {"commit":{"cid":"bafyreihaatlpar3abtx6ck3kde2ksic6zzflk4ppduhf6dxurytqrv33ni","record":{"$type":"app.bsky.graph.follow","createdAt":"2024-11-21T16:18:39.913Z","subject":"did:plc:gf3vum7insztt5rxrpxdz2id"},"rev":"3lbhujcp4ix2n","rkey":"3lbhujcoxmp2n"}}, commit.collection: app.bsky.graph.follow, commit.operation: create, did: did:plc:tdwz2h4id5dxezvohftsmffu, kind: commit, time_us: 1732206349003106}                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   | 2024-11-21T16:25:49.003106 |
-| {_raw: {"commit":{"cid":"bafyreid5ycocp5zq2g7fcx2xxzxrbafuh7b5qhtwuwiomzo6vqila2cbpu","record":{"$type":"app.bsky.feed.repost","createdAt":"2024-11-21T16:23:36.714Z","subject":{"cid":"bafyreieaacfiobnuqvjhhsndyi5s3fd6krbzdduxsyrzfv43kczpcmkl6y","uri":"at://did:plc:o5q6dynpme4ndolc3heztasm/app.bsky.feed.post/3lbfli3qsoc2o"}},"rev":"3lbhus5vior2t","rkey":"3lbhus5vbtz2t"}}, commit.collection: app.bsky.feed.repost, commit.operation: create, did: did:plc:cdsd346mwow7aj3tgfkwsct3, kind: commit, time_us: 1732206349003461}                                                                                                                                                                                                                                                                                                                                                                   | 2024-11-21T16:25:49.003461 |
-| {_raw: {"commit":{"cid":"bafyreibugobcike72y4zxvdyz2oopyt6ywwqfielcwojkb27p7s6rlomgm","record":{"$type":"app.bsky.feed.post","createdAt":"2024-11-21T16:25:44.376Z","langs":["en"],"reply":{"parent":{"cid":"bafyreiaev27cfcxxvn2pdhrwwquzwgclujnulzbcfnn4p4fwgb6migjhw4","uri":"at://did:plc:zec6cslvgc3hhdatrhk6pq5p/app.bsky.feed.post/3lbhujvds4c2b"},"root":{"cid":"bafyreif7qjxhvecwnhlynijj6pf47jwvtkahsz3zh2kaipwu2bw2dxwaqq","uri":"at://did:plc:s4bwqchfzm6gjqfeb6mexgbu/app.bsky.feed.post/3lbhug53kkk2m"}},"text":"\n⌜  Blinking. She hadn't realized she spoke out loud.  ⌟\n\n‘ It was nothing like that  —  .  I was only thinking  . . .  ’\n\n⌜  Trailing off, her mind occupied.  ⌟\n"},"rev":"3lbhuvzeccx2w","rkey":"3lbhuvxf4qs2m"}}, commit.collection: app.bsky.feed.post, commit.operation: create, did: did:plc:s4bwqchfzm6gjqfeb6mexgbu, kind: commit, time_us: 1732206349003907} | 2024-11-21T16:25:49.003907 |
-| {_raw: {"commit":{"cid":"bafyreidjk2svg2fdjiiwohmfmvp3hdxhpb33ycnixzbkyib5m6cocindxq","record":{"$type":"app.bsky.feed.like","createdAt":"2024-11-21T16:25:46.167Z","subject":{"cid":"bafyreiaumopip75nzx2xjbugtwemdppsyx54bd2odf6q45f3o7xkocgari","uri":"at://did:plc:ig2jv6gqup4t7gdq2pmanknw/app.bsky.feed.post/3lbhuvtlaec2c"}},"rev":"3lbhuvzedg52j","rkey":"3lbhuvzdyof2j"}}, commit.collection: app.bsky.feed.like, commit.operation: create, did: did:plc:hbc74dlsxhq53kp5oxges6d7, kind: commit, time_us: 1732206349004769}                                                                                                                                                                                                                                                                                                                                                                       | 2024-11-21T16:25:49.004769 |
-+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+----------------------------+
++------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-------------------------+
+| data                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                       | ts                      |
++------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-------------------------+
+| {_raw: {"commit":{"cid":"bafyreidblutgvj75o4q4akzyyejedjj6l3it6hgqwee6jpwv2wqph5fsgm","record":{"$type":"app.bsky.feed.post","createdAt":"2024-11-21T16:09:27.095Z","langs":["en"],"reply":{"parent":{"cid":"bafyreibfglofvqou2yiqvwzk4rcgkhhxrbunyemshdjledgwymimqkg24e","uri":"at://did:plc:6tr6tuzlx2db3rduzr2d6r24/app.bsky.feed.post/3lbhqo2rtys2z"},"root":{"cid":"bafyreibfglofvqou2yiqvwzk4rcgkhhxrbunyemshdjledgwymimqkg24e","uri":"at://did:plc:6tr6tuzlx2db3rduzr2d6r24/app.bsky.feed.post/3lbhqo2rtys2z"}},"text":"aaaaah.  LIght shines in a corner of WTF...."},"rev":"3lbhtytnn2k2f","rkey":"3lbhtyteurk2y"}}, commit.collection: app.bsky.feed.post, commit.operation: create, did: did:plc:yj3sjq3blzpynh27cumnp5ks, kind: commit, time_us: 1732206349000167}                                                                                                                             | 1970-01-01T00:00:00.001 |
+| {_raw: {"commit":{"cid":"bafyreia5l5vrkh5oj4cjyhcqby2dprhyvcyofo2q5562tijlae2pzih23m","record":{"$type":"app.bsky.feed.like","createdAt":"2024-11-21T16:25:46.221Z","subject":{"cid":"bafyreidjvrcmckkm765mct5fph36x7kupkfo35rjklbf2k76xkzwyiauge","uri":"at://did:plc:azrv4rcbws6kmcga4fsbphg2/app.bsky.feed.post/3lbgjdpbiec2l"}},"rev":"3lbhuvzds6d2a","rkey":"3lbhuvzdked2a"}}, commit.collection: app.bsky.feed.like, commit.operation: create, did: did:plc:3i4xf2v4wcnyktgv6satke64, kind: commit, time_us: 1732206349000644}                                                                                                                                                                                                                                                                                                                                                                       | 1970-01-01T00:00:00.002 |
+| {_raw: {"commit":{"cid":"bafyreiew2p6cgirfaj45qoenm4fgumib7xoloclrap3jgkz5es7g7kby3i","record":{"$type":"app.bsky.graph.follow","createdAt":"2024-11-21T16:27:40.923Z","subject":"did:plc:r7cdh4sgzqbfdc6wcdxxti7c"},"rev":"3lbhuvze3gi2u","rkey":"3lbhuvzdtmi2u"}}, commit.collection: app.bsky.graph.follow, commit.operation: create, did: did:plc:gccfnqqizz4urhchsaie6jft, kind: commit, time_us: 1732206349001108}                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   | 1970-01-01T00:00:00.003 |
+| {_raw: {"commit":{"cid":"bafyreidjix4dauj2afjlbzmhj3a7gwftcevvmmy6edww6vrjdbst26rkby","record":{"$type":"app.bsky.feed.like","createdAt":"2024-11-21T16:15:58.232Z","subject":{"cid":"bafyreiavpshyqzrlo5m7fqodjhs6jevweqnif4phasiwimv4a7mnsqi2fe","uri":"at://did:plc:fusulxqc52zbrc75fi6xrcof/app.bsky.feed.post/3lbhskq5zn22f"}},"rev":"3lbhueija5p22","rkey":"3lbhueiizcx22"}}, commit.collection: app.bsky.feed.like, commit.operation: create, did: did:plc:msxqf3twq7abtdw7dbfskphk, kind: commit, time_us: 1732206349001372}                                                                                                                                                                                                                                                                                                                                                                       | 1970-01-01T00:00:00.004 |
+| {_raw: {"commit":{"cid":"bafyreigzdsdne3z2xxcakgisieyj7y47hj6eg7lj6v4q25ah5q2qotu5ku","record":{"$type":"app.bsky.feed.post","createdAt":"2024-11-21T16:09:27.254Z","langs":["en"],"reply":{"parent":{"cid":"bafyreih35fe2jj3gchmgk4amold4l6sfxd2sby5wrg3jrws5fkdypxrbg4","uri":"at://did:plc:6wx2gg5yqgvmlu35r6y3bk6d/app.bsky.feed.post/3lbhtj2eb4s2o"},"root":{"cid":"bafyreifipyt3vctd4ptuoicvio7rbr5xvjv4afwuggnd2prnmn55mu6luu","uri":"at://did:plc:474ldquxwzrlcvjhhbbk2wte/app.bsky.feed.post/3lbhdzrynik27"}},"text":"okay i take mine back because I hadn’t heard this one yet^^"},"rev":"3lbhtytohxc2o","rkey":"3lbhtytjqzk2q"}}, commit.collection: app.bsky.feed.post, commit.operation: create, did: did:plc:l5o3qjrmfztir54cpwlv2eme, kind: commit, time_us: 1732206349001905}                                                                                                              | 1970-01-01T00:00:00.005 |
+| {_raw: {"commit":{"cid":"bafyreiaa2vsdr4ckwjg4jq47zfd7mewidywfz3qh3dmglcd6ozi4xwdega","record":{"$type":"app.bsky.graph.follow","createdAt":"2024-11-21T16:15:21.495Z","subject":"did:plc:amsdn2tbjxo3xrwqneqhh4cm"},"rev":"3lbhudfo3yi2w","rkey":"3lbhudfnw4y2w"}}, commit.collection: app.bsky.graph.follow, commit.operation: create, did: did:plc:jkaaf5j2yb2pvpx3ualm3vbh, kind: commit, time_us: 1732206349002758}                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   | 1970-01-01T00:00:00.006 |
+| {_raw: {"commit":{"cid":"bafyreihaatlpar3abtx6ck3kde2ksic6zzflk4ppduhf6dxurytqrv33ni","record":{"$type":"app.bsky.graph.follow","createdAt":"2024-11-21T16:18:39.913Z","subject":"did:plc:gf3vum7insztt5rxrpxdz2id"},"rev":"3lbhujcp4ix2n","rkey":"3lbhujcoxmp2n"}}, commit.collection: app.bsky.graph.follow, commit.operation: create, did: did:plc:tdwz2h4id5dxezvohftsmffu, kind: commit, time_us: 1732206349003106}                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   | 1970-01-01T00:00:00.007 |
+| {_raw: {"commit":{"cid":"bafyreid5ycocp5zq2g7fcx2xxzxrbafuh7b5qhtwuwiomzo6vqila2cbpu","record":{"$type":"app.bsky.feed.repost","createdAt":"2024-11-21T16:23:36.714Z","subject":{"cid":"bafyreieaacfiobnuqvjhhsndyi5s3fd6krbzdduxsyrzfv43kczpcmkl6y","uri":"at://did:plc:o5q6dynpme4ndolc3heztasm/app.bsky.feed.post/3lbfli3qsoc2o"}},"rev":"3lbhus5vior2t","rkey":"3lbhus5vbtz2t"}}, commit.collection: app.bsky.feed.repost, commit.operation: create, did: did:plc:cdsd346mwow7aj3tgfkwsct3, kind: commit, time_us: 1732206349003461}                                                                                                                                                                                                                                                                                                                                                                   | 1970-01-01T00:00:00.008 |
+| {_raw: {"commit":{"cid":"bafyreibugobcike72y4zxvdyz2oopyt6ywwqfielcwojkb27p7s6rlomgm","record":{"$type":"app.bsky.feed.post","createdAt":"2024-11-21T16:25:44.376Z","langs":["en"],"reply":{"parent":{"cid":"bafyreiaev27cfcxxvn2pdhrwwquzwgclujnulzbcfnn4p4fwgb6migjhw4","uri":"at://did:plc:zec6cslvgc3hhdatrhk6pq5p/app.bsky.feed.post/3lbhujvds4c2b"},"root":{"cid":"bafyreif7qjxhvecwnhlynijj6pf47jwvtkahsz3zh2kaipwu2bw2dxwaqq","uri":"at://did:plc:s4bwqchfzm6gjqfeb6mexgbu/app.bsky.feed.post/3lbhug53kkk2m"}},"text":"\n⌜  Blinking. She hadn't realized she spoke out loud.  ⌟\n\n‘ It was nothing like that  —  .  I was only thinking  . . .  ’\n\n⌜  Trailing off, her mind occupied.  ⌟\n"},"rev":"3lbhuvzeccx2w","rkey":"3lbhuvxf4qs2m"}}, commit.collection: app.bsky.feed.post, commit.operation: create, did: did:plc:s4bwqchfzm6gjqfeb6mexgbu, kind: commit, time_us: 1732206349003907} | 1970-01-01T00:00:00.009 |
+| {_raw: {"commit":{"cid":"bafyreidjk2svg2fdjiiwohmfmvp3hdxhpb33ycnixzbkyib5m6cocindxq","record":{"$type":"app.bsky.feed.like","createdAt":"2024-11-21T16:25:46.167Z","subject":{"cid":"bafyreiaumopip75nzx2xjbugtwemdppsyx54bd2odf6q45f3o7xkocgari","uri":"at://did:plc:ig2jv6gqup4t7gdq2pmanknw/app.bsky.feed.post/3lbhuvtlaec2c"}},"rev":"3lbhuvzedg52j","rkey":"3lbhuvzdyof2j"}}, commit.collection: app.bsky.feed.like, commit.operation: create, did: did:plc:hbc74dlsxhq53kp5oxges6d7, kind: commit, time_us: 1732206349004769}                                                                                                                                                                                                                                                                                                                                                                       | 1970-01-01T00:00:00.010 |
++------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-------------------------+

tests-integration/src/standalone.rs

@@ -48,9 +48,9 @@ use flow::{FlownodeBuilder, FrontendClient, GrpcQueryHandlerWithBoxedError};
 use frontend::frontend::Frontend;
 use frontend::instance::builder::FrontendBuilder;
 use frontend::instance::{Instance, StandaloneDatanodeManager};
-use frontend::server::Services;
 use meta_srv::metasrv::{FLOW_ID_SEQ, TABLE_ID_SEQ};
 use servers::grpc::GrpcOptions;
+use servers::server::ServerHandlers;
 use snafu::ResultExt;
 use standalone::options::StandaloneOptions;
 
@@ -249,7 +249,7 @@ impl GreptimeDbStandaloneBuilder {
             procedure_executor.clone(),
             Arc::new(ProcessManager::new(server_addr, None)),
         )
-        .with_plugin(plugins.clone())
+        .with_plugin(plugins)
         .try_build()
         .await
         .unwrap();
@@ -282,15 +282,14 @@ impl GreptimeDbStandaloneBuilder {
 
         test_util::prepare_another_catalog_and_schema(&instance).await;
 
-        let servers = Services::new(opts.clone(), instance.clone(), plugins)
-            .build()
-            .unwrap();
-        let frontend = Frontend {
+        let mut frontend = Frontend {
             instance,
-            servers,
+            servers: ServerHandlers::default(),
             heartbeat_task: None,
         };
 
+        frontend.start().await.unwrap();
+
         GreptimeDbStandalone {
             frontend: Arc::new(frontend),
             opts,

tests-integration/tests/jsonbench.rs

@@ -18,114 +18,14 @@ use std::{fs, io};
 
 use common_test_util::find_workspace_path;
 use frontend::instance::Instance;
-use http::StatusCode;
-use servers::http::test_helpers::TestClient;
-use servers::http::{HTTP_SERVER, HttpServer};
-use servers::server::ServerHandlers;
 use tests_integration::standalone::GreptimeDbStandaloneBuilder;
 use tests_integration::test_util::execute_sql_and_expect;
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn test_load_jsonbench_data_by_pipeline() -> io::Result<()> {
+#[tokio::test]
+async fn test_load_jsonbench_data() {
     common_telemetry::init_default_ut_logging();
 
-    let instance = GreptimeDbStandaloneBuilder::new("test_load_jsonbench_data_by_pipeline")
-        .build()
-        .await;
-    let frontend = instance.fe_instance();
-
-    let ServerHandlers::Init(handlers) = instance.frontend.server_handlers() else {
-        unreachable!()
-    };
-    let router = {
-        let handlers = handlers.lock().unwrap();
-        let server = handlers
-            .get(HTTP_SERVER)
-            .and_then(|x| x.0.as_any().downcast_ref::<HttpServer>())
-            .unwrap();
-        server.build(server.make_app()).unwrap()
-    };
-    let client = TestClient::new(router).await;
-
-    create_table(frontend).await;
-
-    desc_table(frontend).await;
-
-    create_pipeline(&client).await;
-
-    insert_data_by_pipeline(&client).await?;
-
-    query_data(frontend).await
-}
-
-async fn insert_data_by_pipeline(client: &TestClient) -> io::Result<()> {
-    let file = fs::read(find_workspace_path(
-        "tests-integration/resources/jsonbench-head-10.ndjson",
-    ))?;
-
-    let response = client
-        .post("/v1/ingest?table=bluesky&pipeline_name=jsonbench")
-        .header("Content-Type", "text/plain")
-        .body(file)
-        .send()
-        .await;
-    assert_eq!(response.status(), StatusCode::OK);
-
-    let response = response.text().await;
-    // Note that this pattern also matches the inserted rows: "10".
-    let pattern = r#"{"output":[{"affectedrows":10}]"#;
-    assert!(response.starts_with(pattern));
-    Ok(())
-}
-
-async fn create_pipeline(client: &TestClient) {
-    let pipeline = r#"
-version: 2
-
-processors:
-  - json_parse:
-      fields:
-        - message, data
-      ignore_missing: true
-  - simple_extract:
-      fields:
-        - data, time_us
-      key: "time_us"
-      ignore_missing: false
-  - epoch:
-      fields:
-        - time_us
-      resolution: microsecond
-  - select:
-      fields:
-        - time_us
-        - data
-
-transform:
-  - fields:
-      - time_us
-    type: epoch, us
-    index: timestamp
-"#;
-
-    let response = client
-        .post("/v1/pipelines/jsonbench")
-        .header("Content-Type", "application/x-yaml")
-        .body(pipeline)
-        .send()
-        .await;
-    assert_eq!(response.status(), StatusCode::OK);
-
-    let response = response.text().await;
-    let pattern = r#"{"pipelines":[{"name":"jsonbench""#;
-    assert!(response.starts_with(pattern));
-}
-
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn test_load_jsonbench_data_by_sql() -> io::Result<()> {
-    common_telemetry::init_default_ut_logging();
-
-    let instance = GreptimeDbStandaloneBuilder::new("test_load_jsonbench_data_by_sql")
+    let instance = GreptimeDbStandaloneBuilder::new("test_load_jsonbench_data")
         .build()
         .await;
     let frontend = instance.fe_instance();
@@ -134,9 +34,9 @@ async fn test_load_jsonbench_data_by_sql() -> io::Result<()> {
 
     desc_table(frontend).await;
 
-    insert_data_by_sql(frontend).await?;
+    insert_data(frontend).await.unwrap();
 
-    query_data(frontend).await
+    query_data(frontend).await.unwrap();
 }
 
 async fn query_data(frontend: &Arc<Instance>) -> io::Result<()> {
@@ -146,21 +46,22 @@ async fn query_data(frontend: &Arc<Instance>) -> io::Result<()> {
 | count(*) |
 +----------+
 | 10       |
-+----------+"#;
++----------+
+"#;
     execute_sql_and_expect(frontend, sql, expected).await;
 
-    let sql = "SELECT * FROM bluesky ORDER BY time_us";
+    let sql = "SELECT * FROM bluesky ORDER BY ts";
     let expected = fs::read_to_string(find_workspace_path(
         "tests-integration/resources/jsonbench-select-all.txt",
     ))?;
     execute_sql_and_expect(frontend, sql, &expected).await;
 
     // query 1:
-    let sql = "
-SELECT
-    json_get_string(data, '$.commit.collection') AS event, count() AS count
-FROM bluesky
-GROUP BY event
+    let sql = "\
+SELECT \
+    json_get_string(data, '$.commit.collection') AS event, count() AS count \
+FROM bluesky \
+GROUP BY event \
 ORDER BY count DESC, event ASC";
     let expected = r#"
 +-----------------------+-------+
@@ -174,16 +75,16 @@ ORDER BY count DESC, event ASC";
     execute_sql_and_expect(frontend, sql, expected).await;
 
     // query 2:
-    let sql = "
-SELECT
-    json_get_string(data, '$.commit.collection') AS event,
-    count() AS count,
-    count(DISTINCT json_get_string(data, '$.did')) AS users
-FROM bluesky
-WHERE
-    (json_get_string(data, '$.kind') = 'commit') AND
-    (json_get_string(data, '$.commit.operation') = 'create')
-GROUP BY event
+    let sql = "\
+SELECT \
+    json_get_string(data, '$.commit.collection') AS event, \
+    count() AS count, \
+    count(DISTINCT json_get_string(data, '$.did')) AS users \
+FROM bluesky \
+WHERE \
+    (json_get_string(data, '$.kind') = 'commit') AND \
+    (json_get_string(data, '$.commit.operation') = 'create') \
+GROUP BY event \
 ORDER BY count DESC, event ASC";
     let expected = r#"
 +-----------------------+-------+-------+
@@ -197,18 +98,18 @@ ORDER BY count DESC, event ASC";
     execute_sql_and_expect(frontend, sql, expected).await;
 
     // query 3:
-    let sql = "
-SELECT
-    json_get_string(data, '$.commit.collection') AS event,
-    date_part('hour', to_timestamp_micros(json_get_int(data, '$.time_us'))) as hour_of_day,
-    count() AS count
-FROM bluesky
-WHERE
-    (json_get_string(data, '$.kind') = 'commit') AND
-    (json_get_string(data, '$.commit.operation') = 'create') AND
-    json_get_string(data, '$.commit.collection') IN
-        ('app.bsky.feed.post', 'app.bsky.feed.repost', 'app.bsky.feed.like')
-GROUP BY event, hour_of_day
+    let sql = "\
+SELECT \
+    json_get_string(data, '$.commit.collection') AS event, \
+    date_part('hour', to_timestamp_micros(json_get_int(data, '$.time_us'))) as hour_of_day, \
+    count() AS count \
+FROM bluesky \
+WHERE \
+    (json_get_string(data, '$.kind') = 'commit') AND \
+    (json_get_string(data, '$.commit.operation') = 'create') AND \
+    json_get_string(data, '$.commit.collection') IN \
+        ('app.bsky.feed.post', 'app.bsky.feed.repost', 'app.bsky.feed.like') \
+GROUP BY event, hour_of_day \
 ORDER BY hour_of_day, event";
     let expected = r#"
 +----------------------+-------------+-------+
@@ -221,7 +122,7 @@ ORDER BY hour_of_day, event";
     execute_sql_and_expect(frontend, sql, expected).await;
 
     // query 4:
-    let sql = "
+    let sql = "\
 SELECT
     json_get_string(data, '$.did') as user_id,
     min(to_timestamp_micros(json_get_int(data, '$.time_us'))) AS first_post_ts
@@ -273,23 +174,19 @@ LIMIT 3";
     Ok(())
 }
 
-async fn insert_data_by_sql(frontend: &Arc<Instance>) -> io::Result<()> {
+async fn insert_data(frontend: &Arc<Instance>) -> io::Result<()> {
     let file = fs::File::open(find_workspace_path(
         "tests-integration/resources/jsonbench-head-10.ndjson",
     ))?;
     let reader = io::BufReader::new(file);
-    for line in reader.lines() {
+    for (i, line) in reader.lines().enumerate() {
         let line = line?;
         if line.is_empty() {
             continue;
         }
-
-        let json: serde_json::Value = serde_json::from_str(&line)?;
-        let time_us = json.pointer("/time_us").and_then(|x| x.as_u64()).unwrap();
-
         let sql = format!(
-            "INSERT INTO bluesky (time_us, data) VALUES ({}, '{}')",
-            time_us,
+            "INSERT INTO bluesky (ts, data) VALUES ({}, '{}')",
+            i + 1,
             line.replace("'", "''"), // standard method to escape the single quote
         );
         execute_sql_and_expect(frontend, &sql, "Affected Rows: 1").await;
@@ -300,12 +197,12 @@ async fn insert_data_by_sql(frontend: &Arc<Instance>) -> io::Result<()> {
 async fn desc_table(frontend: &Arc<Instance>) {
     let sql = "DESC TABLE bluesky";
     let expected = r#"
-+---------+------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
-| Column  | Type                                                                                                                                           | Key | Null | Default | Semantic Type |
-+---------+------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
-| data    | Json<{"_raw":"<String>","commit.collection":"<String>","commit.operation":"<String>","did":"<String>","kind":"<String>","time_us":"<Number>"}> |     | YES  |         | FIELD         |
-| time_us | TimestampMicrosecond                                                                                                                           | PRI | NO   |         | TIMESTAMP     |
-+---------+------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+"#;
++--------+----------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
+| Column | Type                                                                                                                                         | Key | Null | Default | Semantic Type |
++--------+----------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
+| data   | Json<Object{"_raw": String, "commit.collection": String, "commit.operation": String, "did": String, "kind": String, "time_us": Number(I64)}> |     | YES  |         | FIELD         |
+| ts     | TimestampMillisecond                                                                                                                         | PRI | NO   |         | TIMESTAMP     |
++--------+----------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+"#;
     execute_sql_and_expect(frontend, sql, expected).await;
 }
 
@@ -322,7 +219,7 @@ CREATE TABLE bluesky (
       time_us Bigint
     >,
   ),
-  time_us TimestampMicrosecond TIME INDEX,
+  ts Timestamp TIME INDEX,
 )
 "#;
     execute_sql_and_expect(frontend, sql, "Affected Rows: 0").await;

tests/cases/standalone/common/comment.result

@@ -32,17 +32,6 @@ SHOW CREATE TABLE comment_table_test;
 |                    | )                                                 |
 +--------------------+---------------------------------------------------+
 
-SELECT table_comment
-FROM information_schema.tables
-WHERE table_schema = 'public'
-  AND table_name = 'comment_table_test';
-
-+-------------------------+
-| table_comment           |
-+-------------------------+
-| table level description |
-+-------------------------+
-
 -- Remove table comment
 COMMENT ON TABLE comment_table_test IS NULL;
 
@@ -65,17 +54,6 @@ SHOW CREATE TABLE comment_table_test;
 |                    |                                                   |
 +--------------------+---------------------------------------------------+
 
-SELECT table_comment
-FROM information_schema.tables
-WHERE table_schema = 'public'
-  AND table_name = 'comment_table_test';
-
-+---------------+
-| table_comment |
-+---------------+
-|               |
-+---------------+
-
 DROP TABLE comment_table_test;
 
 Affected Rows: 0
@@ -112,18 +90,6 @@ SHOW CREATE TABLE comment_column_test;
 |                     |                                                         |
 +---------------------+---------------------------------------------------------+
 
-SELECT column_comment
-FROM information_schema.columns
-WHERE table_schema = 'public'
-  AND table_name = 'comment_column_test'
-  AND column_name = 'val';
-
-+--------------------------+
-| column_comment           |
-+--------------------------+
-| value column description |
-+--------------------------+
-
 -- Remove column comment
 COMMENT ON COLUMN comment_column_test.val IS NULL;
 
@@ -146,18 +112,6 @@ SHOW CREATE TABLE comment_column_test;
 |                     |                                                    |
 +---------------------+----------------------------------------------------+
 
-SELECT column_comment
-FROM information_schema.columns
-WHERE table_schema = 'public'
-  AND table_name = 'comment_column_test'
-  AND column_name = 'val';
-
-+----------------+
-| column_comment |
-+----------------+
-|                |
-+----------------+
-
 DROP TABLE comment_column_test;
 
 Affected Rows: 0
@@ -201,16 +155,6 @@ SHOW CREATE FLOW flow_comment_test;
 |                   | AS SELECT desc_str, ts FROM flow_source_comment_test |
 +-------------------+------------------------------------------------------+
 
-SELECT comment
-FROM information_schema.flows
-WHERE flow_name = 'flow_comment_test';
-
-+------------------------+
-| comment                |
-+------------------------+
-| flow level description |
-+------------------------+
-
 -- Remove flow comment
 COMMENT ON FLOW flow_comment_test IS NULL;
 
@@ -226,16 +170,6 @@ SHOW CREATE FLOW flow_comment_test;
 |                   | AS SELECT desc_str, ts FROM flow_source_comment_test |
 +-------------------+------------------------------------------------------+
 
-SELECT comment
-FROM information_schema.flows
-WHERE flow_name = 'flow_comment_test';
-
-+---------+
-| comment |
-+---------+
-|         |
-+---------+
-
 DROP FLOW flow_comment_test;
 
 Affected Rows: 0

tests/cases/standalone/common/comment.sql

@@ -9,18 +9,10 @@ CREATE TABLE comment_table_test (
 -- Add table comment
 COMMENT ON TABLE comment_table_test IS 'table level description';
 SHOW CREATE TABLE comment_table_test;
-SELECT table_comment
-FROM information_schema.tables
-WHERE table_schema = 'public'
-  AND table_name = 'comment_table_test';
 
 -- Remove table comment
 COMMENT ON TABLE comment_table_test IS NULL;
 SHOW CREATE TABLE comment_table_test;
-SELECT table_comment
-FROM information_schema.tables
-WHERE table_schema = 'public'
-  AND table_name = 'comment_table_test';
 
 DROP TABLE comment_table_test;
 
@@ -35,20 +27,10 @@ CREATE TABLE comment_column_test (
 -- Add column comment
 COMMENT ON COLUMN comment_column_test.val IS 'value column description';
 SHOW CREATE TABLE comment_column_test;
-SELECT column_comment
-FROM information_schema.columns
-WHERE table_schema = 'public'
-  AND table_name = 'comment_column_test'
-  AND column_name = 'val';
 
 -- Remove column comment
 COMMENT ON COLUMN comment_column_test.val IS NULL;
 SHOW CREATE TABLE comment_column_test;
-SELECT column_comment
-FROM information_schema.columns
-WHERE table_schema = 'public'
-  AND table_name = 'comment_column_test'
-  AND column_name = 'val';
 
 DROP TABLE comment_column_test;
 
@@ -72,17 +54,12 @@ SELECT desc_str, ts FROM flow_source_comment_test;
 -- Add flow comment
 COMMENT ON FLOW flow_comment_test IS 'flow level description';
 SHOW CREATE FLOW flow_comment_test;
-SELECT comment
-FROM information_schema.flows
-WHERE flow_name = 'flow_comment_test';
 
 -- Remove flow comment
 COMMENT ON FLOW flow_comment_test IS NULL;
 SHOW CREATE FLOW flow_comment_test;
-SELECT comment
-FROM information_schema.flows
-WHERE flow_name = 'flow_comment_test';
 
 DROP FLOW flow_comment_test;
 DROP TABLE flow_source_comment_test;
 DROP TABLE flow_sink_comment_test;
+

tests/cases/standalone/common/types/json/json-structured.result

@@ -12,7 +12,7 @@ DESC TABLE t;
 | Column | Type                 | Key | Null | Default | Semantic Type |
 +--------+----------------------+-----+------+---------+---------------+
 | ts     | TimestampMillisecond | PRI | NO   |         | TIMESTAMP     |
-| j      | Json<"<Null>">       |     | YES  |         | FIELD         |
+| j      | Json<Null>           |     | YES  |         | FIELD         |
 +--------+----------------------+-----+------+---------+---------------+
 
 INSERT INTO t VALUES
@@ -24,12 +24,12 @@ Affected Rows: 3
 
 DESC TABLE t;
 
-+--------+---------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
-| Column | Type                                                                                              | Key | Null | Default | Semantic Type |
-+--------+---------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
-| ts     | TimestampMillisecond                                                                              | PRI | NO   |         | TIMESTAMP     |
-| j      | Json<{"int":"<Number>","list":["<Number>"],"nested":{"a":{"x":"<String>"},"b":{"y":"<Number>"}}}> |     | YES  |         | FIELD         |
-+--------+---------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
++--------+-----------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
+| Column | Type                                                                                                                                    | Key | Null | Default | Semantic Type |
++--------+-----------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
+| ts     | TimestampMillisecond                                                                                                                    | PRI | NO   |         | TIMESTAMP     |
+| j      | Json<Object{"int": Number(I64), "list": Array[Number(F64)], "nested": Object{"a": Object{"x": String}, "b": Object{"y": Number(I64)}}}> |     | YES  |         | FIELD         |
++--------+-----------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
 
 INSERT INTO t VALUES
 (1762128004000, '{"int": 4, "bool": true, "nested": {"a": {"y": 1}}}'),
@@ -39,12 +39,12 @@ Affected Rows: 2
 
 DESC TABLE t;
 
-+--------+-------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
-| Column | Type                                                                                                                                            | Key | Null | Default | Semantic Type |
-+--------+-------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
-| ts     | TimestampMillisecond                                                                                                                            | PRI | NO   |         | TIMESTAMP     |
-| j      | Json<{"bool":"<Bool>","int":"<Number>","list":["<Number>"],"nested":{"a":{"x":"<String>","y":"<Number>"},"b":{"x":"<String>","y":"<Number>"}}}> |     | YES  |         | FIELD         |
-+--------+-------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
++--------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
+| Column | Type                                                                                                                                                                                 | Key | Null | Default | Semantic Type |
++--------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
+| ts     | TimestampMillisecond                                                                                                                                                                 | PRI | NO   |         | TIMESTAMP     |
+| j      | Json<Object{"bool": Bool, "int": Number(I64), "list": Array[Number(F64)], "nested": Object{"a": Object{"x": String, "y": Number(I64)}, "b": Object{"x": String, "y": Number(I64)}}}> |     | YES  |         | FIELD         |
++--------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
 
 INSERT INTO t VALUES (1762128006000, '{"int": 6, "list": [-6.0], "bool": true, "nested": {"a": {"x": "ax", "y": 66}, "b": {"y": -66, "x": "bx"}}}');
 
@@ -52,12 +52,12 @@ Affected Rows: 1
 
 DESC TABLE t;
 
-+--------+-------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
-| Column | Type                                                                                                                                            | Key | Null | Default | Semantic Type |
-+--------+-------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
-| ts     | TimestampMillisecond                                                                                                                            | PRI | NO   |         | TIMESTAMP     |
-| j      | Json<{"bool":"<Bool>","int":"<Number>","list":["<Number>"],"nested":{"a":{"x":"<String>","y":"<Number>"},"b":{"x":"<String>","y":"<Number>"}}}> |     | YES  |         | FIELD         |
-+--------+-------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
++--------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
+| Column | Type                                                                                                                                                                                 | Key | Null | Default | Semantic Type |
++--------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
+| ts     | TimestampMillisecond                                                                                                                                                                 | PRI | NO   |         | TIMESTAMP     |
+| j      | Json<Object{"bool": Bool, "int": Number(I64), "list": Array[Number(F64)], "nested": Object{"a": Object{"x": String, "y": Number(I64)}, "b": Object{"x": String, "y": Number(I64)}}}> |     | YES  |         | FIELD         |
++--------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----+------+---------+---------------+
 
 INSERT INTO t VALUES (1762128011000, '{}');