only use one optional Partition definition on creating table

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

.cargo/config.toml

@@ -12,3 +12,6 @@ fetch = true
 checkout = true
 list_files = true
 internal_use_git2 = false
+
+[env]
+CARGO_WORKSPACE_DIR = { value = "", relative = true }

Cargo.lock

@@ -1602,6 +1602,17 @@ version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "acbc26382d871df4b7442e3df10a9402bf3cf5e55cbd66f12be38861425f0564"
 
+[[package]]
+name = "cargo-manifest"
+version = "0.19.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a1d8af896b707212cd0e99c112a78c9497dd32994192a463ed2f7419d29bd8c6"
+dependencies = [
+ "serde",
+ "thiserror 2.0.12",
+ "toml 0.8.19",
+]
+
 [[package]]
 name = "cast"
 version = "0.3.0"
@@ -2520,6 +2531,7 @@ dependencies = [
  "tokio",
  "tokio-postgres",
  "tonic 0.12.3",
+ "tracing",
  "typetag",
  "uuid",
 ]
@@ -2668,12 +2680,31 @@ dependencies = [
  "strum 0.27.1",
 ]
 
+[[package]]
+name = "common-sql"
+version = "0.16.0"
+dependencies = [
+ "common-base",
+ "common-datasource",
+ "common-decimal",
+ "common-error",
+ "common-macro",
+ "common-time",
+ "datafusion-sql",
+ "datatypes",
+ "hex",
+ "jsonb",
+ "snafu 0.8.5",
+ "sqlparser 0.54.0 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=0cf6c04490d59435ee965edd2078e8855bd8471e)",
+]
+
 [[package]]
 name = "common-telemetry"
 version = "0.16.0"
 dependencies = [
  "backtrace",
  "common-error",
+ "common-version",
  "console-subscriber",
  "greptime-proto",
  "humantime-serde",
@@ -2731,6 +2762,7 @@ name = "common-version"
 version = "0.16.0"
 dependencies = [
  "build-data",
+ "cargo-manifest",
  "const_format",
  "serde",
  "shadow-rs",
@@ -2964,9 +2996,9 @@ dependencies = [
 
 [[package]]
 name = "crc"
-version = "3.2.1"
+version = "3.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "69e6e4d7b33a94f0991c26729976b10ebde1d34c3ee82408fb536164fa10d636"
+checksum = "9710d3b3739c2e349eb44fe848ad0b7c8cb1e42bd87ee49371df2f7acaf3e675"
 dependencies = [
  "crc-catalog",
 ]
@@ -3775,6 +3807,7 @@ dependencies = [
  "tokio",
  "toml 0.8.19",
  "tonic 0.12.3",
+ "tracing",
 ]
 
 [[package]]
@@ -3797,7 +3830,7 @@ dependencies = [
  "jsonb",
  "num",
  "num-traits",
- "ordered-float 3.9.2",
+ "ordered-float 4.3.0",
  "paste",
  "serde",
  "serde_json",
@@ -4118,12 +4151,16 @@ dependencies = [
 
 [[package]]
 name = "domain"
-version = "0.10.4"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4c84070523f8ba0f9127ff156920f27eb27b302b425efe60bf5f41ec244d1c60"
+checksum = "a11dd7f04a6a6d2aea0153c6e31f5ea7af8b2efdf52cdaeea7a9a592c7fefef9"
 dependencies = [
+ "bumpalo",
  "bytes",
+ "domain-macros",
  "futures-util",
+ "hashbrown 0.14.5",
+ "log",
  "moka",
  "octseq",
  "rand 0.8.5",
@@ -4134,6 +4171,17 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "domain-macros"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0e197fdfd2cdb5fdeb7f8ddcf3aed5d5d04ecde2890d448b14ffb716f7376b70"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.100",
+]
+
 [[package]]
 name = "dotenv"
 version = "0.15.0"
@@ -4763,6 +4811,7 @@ dependencies = [
  "toml 0.8.19",
  "tonic 0.12.3",
  "tower 0.5.2",
+ "tracing",
  "uuid",
 ]
 
@@ -7266,6 +7315,7 @@ dependencies = [
  "snafu 0.8.5",
  "store-api",
  "tokio",
+ "tracing",
 ]
 
 [[package]]
@@ -7379,6 +7429,7 @@ dependencies = [
  "datafusion-expr",
  "datatypes",
  "dotenv",
+ "either",
  "futures",
  "humantime-serde",
  "index",
@@ -7414,6 +7465,7 @@ dependencies = [
  "tokio-stream",
  "tokio-util",
  "toml 0.8.19",
+ "tracing",
  "uuid",
 ]
 
@@ -8457,6 +8509,7 @@ dependencies = [
  "common-query",
  "common-recordbatch",
  "common-runtime",
+ "common-sql",
  "common-telemetry",
  "common-test-util",
  "common-time",
@@ -8492,6 +8545,7 @@ dependencies = [
  "tokio",
  "tokio-util",
  "tonic 0.12.3",
+ "tracing",
 ]
 
 [[package]]
@@ -8528,17 +8582,6 @@ dependencies = [
  "num-traits",
 ]
 
-[[package]]
-name = "ordered-float"
-version = "3.9.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1e1c390732d15f1d48471625cd92d154e66db2c56645e29a9cd26f4699f72dc"
-dependencies = [
- "num-traits",
- "rand 0.8.5",
- "serde",
-]
-
 [[package]]
 name = "ordered-float"
 version = "4.3.0"
@@ -8546,6 +8589,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "44d501f1a72f71d3c063a6bbc8f7271fa73aa09fe5d6283b6571e2ed176a2537"
 dependencies = [
  "num-traits",
+ "rand 0.8.5",
+ "serde",
 ]
 
 [[package]]
@@ -9082,6 +9127,7 @@ dependencies = [
  "moka",
  "once_cell",
  "operator",
+ "ordered-float 4.3.0",
  "paste",
  "prometheus",
  "query",
@@ -9522,8 +9568,9 @@ dependencies = [
 
 [[package]]
 name = "promql-parser"
-version = "0.5.1"
-source = "git+https://github.com/GreptimeTeam/promql-parser.git?rev=0410e8b459dda7cb222ce9596f8bf3971bd07bd2#0410e8b459dda7cb222ce9596f8bf3971bd07bd2"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "328fe69c2443ec4f8e6c33ea925dde04a1026e6c95928e89ed02343944cac9bf"
 dependencies = [
  "cfgrammar",
  "chrono",
@@ -9892,6 +9939,7 @@ dependencies = [
  "table",
  "tokio",
  "tokio-stream",
+ "tracing",
  "unescaper",
  "uuid",
 ]
@@ -11243,6 +11291,7 @@ dependencies = [
  "common-recordbatch",
  "common-runtime",
  "common-session",
+ "common-sql",
  "common-telemetry",
  "common-test-util",
  "common-time",
@@ -11324,8 +11373,10 @@ dependencies = [
  "tonic-reflection",
  "tower 0.5.2",
  "tower-http 0.6.2",
+ "tracing",
  "urlencoding",
  "uuid",
+ "vrl",
  "zstd 0.13.2",
 ]
 
@@ -11681,6 +11732,7 @@ dependencies = [
  "common-error",
  "common-macro",
  "common-query",
+ "common-sql",
  "common-time",
  "datafusion",
  "datafusion-common",
@@ -12987,9 +13039,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
 
 [[package]]
 name = "tokio"
-version = "1.44.2"
+version = "1.45.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e6b88822cbe49de4185e3a4cbf8321dd487cf5fe0c5c65695fef6346371e9c48"
+checksum = "75ef51a33ef1da925cea3e4eb122833cb377c61439ca401b770f54902b806779"
 dependencies = [
  "backtrace",
  "bytes",
@@ -13155,6 +13207,7 @@ version = "0.8.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a1ed1f98e3fdc28d6d910e6737ae6ab1a93bf1985935a1193e68f93eeb68d24e"
 dependencies = [
+ "indexmap 2.9.0",
  "serde",
  "serde_spanned",
  "toml_datetime",
@@ -13944,9 +13997,9 @@ dependencies = [
 
 [[package]]
 name = "vrl"
-version = "0.24.0"
+version = "0.25.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f9ceadaa40aef567a26079ff014ca7a567ba85344f1b81090b5ec7d7bb16a219"
+checksum = "4f49394b948406ea1564aa00152e011d87a38ad35d277ebddda257a9ee39c419"
 dependencies = [
  "aes",
  "aes-siv",

Cargo.toml

@@ -30,6 +30,7 @@ members = [
     "src/common/recordbatch",
     "src/common/runtime",
     "src/common/session",
+    "src/common/sql",
     "src/common/stat",
     "src/common/substrait",
     "src/common/telemetry",
@@ -166,14 +167,13 @@ opentelemetry-proto = { version = "0.27", features = [
     "with-serde",
     "logs",
 ] }
+ordered-float = { version = "4.3", features = ["serde"] }
 parking_lot = "0.12"
 parquet = { version = "54.2", default-features = false, features = ["arrow", "async", "object_store"] }
 paste = "1.0"
 pin-project = "1.0"
 prometheus = { version = "0.13.3", features = ["process"] }
-promql-parser = { git = "https://github.com/GreptimeTeam/promql-parser.git", rev = "0410e8b459dda7cb222ce9596f8bf3971bd07bd2", features = [
-    "ser",
-] }
+promql-parser = { version = "0.6", features = ["ser"] }
 prost = { version = "0.13", features = ["no-recursion-limit"] }
 raft-engine = { version = "0.4.1", default-features = false }
 rand = "0.9"
@@ -224,10 +224,12 @@ tokio-util = { version = "0.7", features = ["io-util", "compat"] }
 toml = "0.8.8"
 tonic = { version = "0.12", features = ["tls", "gzip", "zstd"] }
 tower = "0.5"
+tracing = "0.1"
 tracing-appender = "0.2"
 tracing-subscriber = { version = "0.3", features = ["env-filter", "json", "fmt"] }
 typetag = "0.2"
 uuid = { version = "1.7", features = ["serde", "v4", "fast-rng"] }
+vrl = "0.25"
 zstd = "0.13"
 # DO_NOT_REMOVE_THIS: END_OF_EXTERNAL_DEPENDENCIES
 
@@ -262,6 +264,7 @@ common-query = { path = "src/common/query" }
 common-recordbatch = { path = "src/common/recordbatch" }
 common-runtime = { path = "src/common/runtime" }
 common-session = { path = "src/common/session" }
+common-sql = { path = "src/common/sql" }
 common-telemetry = { path = "src/common/telemetry" }
 common-test-util = { path = "src/common/test-util" }
 common-time = { path = "src/common/time" }

scripts/install.sh

@@ -53,6 +53,54 @@ get_arch_type() {
   esac
 }
 
+# Verify SHA256 checksum
+verify_sha256() {
+  file="$1"
+  expected_sha256="$2"
+
+  if command -v sha256sum >/dev/null 2>&1; then
+    actual_sha256=$(sha256sum "$file" | cut -d' ' -f1)
+  elif command -v shasum >/dev/null 2>&1; then
+    actual_sha256=$(shasum -a 256 "$file" | cut -d' ' -f1)
+  else
+    echo "Warning: No SHA256 verification tool found (sha256sum or shasum). Skipping checksum verification."
+    return 0
+  fi
+
+  if [ "$actual_sha256" = "$expected_sha256" ]; then
+    echo "SHA256 checksum verified successfully."
+    return 0
+  else
+    echo "Error: SHA256 checksum verification failed!"
+    echo "Expected: $expected_sha256"
+    echo "Actual: $actual_sha256"
+    return 1
+  fi
+}
+
+# Prompt for user confirmation (compatible with different shells)
+prompt_confirmation() {
+  message="$1"
+  printf "%s (y/N): " "$message"
+
+  # Try to read user input, fallback if read fails
+  answer=""
+  if read answer </dev/tty 2>/dev/null; then
+    case "$answer" in
+      [Yy]|[Yy][Ee][Ss])
+        return 0
+        ;;
+      *)
+        return 1
+        ;;
+    esac
+  else
+    echo ""
+    echo "Cannot read user input. Defaulting to No."
+    return 1
+  fi
+}
+
 download_artifact() {
   if [ -n "${OS_TYPE}" ] && [ -n "${ARCH_TYPE}" ]; then
     # Use the latest stable released version.
@@ -71,17 +119,104 @@ download_artifact() {
     fi
 
     echo "Downloading ${BIN}, OS: ${OS_TYPE}, Arch: ${ARCH_TYPE}, Version: ${VERSION}"
-    PACKAGE_NAME="${BIN}-${OS_TYPE}-${ARCH_TYPE}-${VERSION}.tar.gz"
+    PKG_NAME="${BIN}-${OS_TYPE}-${ARCH_TYPE}-${VERSION}"
+    PACKAGE_NAME="${PKG_NAME}.tar.gz"
+    SHA256_FILE="${PKG_NAME}.sha256sum"
 
     if [ -n "${PACKAGE_NAME}" ]; then
-      wget "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${PACKAGE_NAME}"
+      # Check if files already exist and prompt for override
+      if [ -f "${PACKAGE_NAME}" ]; then
+        echo "File ${PACKAGE_NAME} already exists."
+        if prompt_confirmation "Do you want to override it?"; then
+          echo "Overriding existing file..."
+          rm -f "${PACKAGE_NAME}"
+        else
+          echo "Skipping download. Using existing file."
+        fi
+      fi
+
+      if [ -f "${BIN}" ]; then
+        echo "Binary ${BIN} already exists."
+        if prompt_confirmation "Do you want to override it?"; then
+          echo "Will override existing binary..."
+          rm -f "${BIN}"
+        else
+          echo "Installation cancelled."
+          exit 0
+        fi
+      fi
+
+      # Download package if not exists
+      if [ ! -f "${PACKAGE_NAME}" ]; then
+        echo "Downloading ${PACKAGE_NAME}..."
+        # Use curl instead of wget for better compatibility
+        if command -v curl >/dev/null 2>&1; then
+          if ! curl -L -o "${PACKAGE_NAME}" "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${PACKAGE_NAME}"; then
+            echo "Error: Failed to download ${PACKAGE_NAME}"
+            exit 1
+          fi
+        elif command -v wget >/dev/null 2>&1; then
+          if ! wget -O "${PACKAGE_NAME}" "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${PACKAGE_NAME}"; then
+            echo "Error: Failed to download ${PACKAGE_NAME}"
+            exit 1
+          fi
+        else
+          echo "Error: Neither curl nor wget is available for downloading."
+          exit 1
+        fi
+      fi
+
+      # Download and verify SHA256 checksum
+      echo "Downloading SHA256 checksum..."
+      sha256_download_success=0
+      if command -v curl >/dev/null 2>&1; then
+        if curl -L -s -o "${SHA256_FILE}" "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${SHA256_FILE}" 2>/dev/null; then
+          sha256_download_success=1
+        fi
+      elif command -v wget >/dev/null 2>&1; then
+        if wget -q -O "${SHA256_FILE}" "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${SHA256_FILE}" 2>/dev/null; then
+          sha256_download_success=1
+        fi
+      fi
+
+      if [ $sha256_download_success -eq 1 ] && [ -f "${SHA256_FILE}" ]; then
+        expected_sha256=$(cat "${SHA256_FILE}" | cut -d' ' -f1)
+        if [ -n "$expected_sha256" ]; then
+          if ! verify_sha256 "${PACKAGE_NAME}" "${expected_sha256}"; then
+            echo "SHA256 verification failed. Removing downloaded file."
+            rm -f "${PACKAGE_NAME}" "${SHA256_FILE}"
+            exit 1
+          fi
+        else
+          echo "Warning: Could not parse SHA256 checksum from file."
+        fi
+        rm -f "${SHA256_FILE}"
+      else
+        echo "Warning: Could not download SHA256 checksum file. Skipping verification."
+      fi
 
       # Extract the binary and clean the rest.
-      tar xvf "${PACKAGE_NAME}" && \
-      mv "${PACKAGE_NAME%.tar.gz}/${BIN}" "${PWD}" && \
-      rm -r "${PACKAGE_NAME}" && \
-      rm -r "${PACKAGE_NAME%.tar.gz}" && \
-      echo "Run './${BIN} --help' to get started"
+      echo "Extracting ${PACKAGE_NAME}..."
+      if ! tar xf "${PACKAGE_NAME}"; then
+        echo "Error: Failed to extract ${PACKAGE_NAME}"
+        exit 1
+      fi
+
+      # Find the binary in the extracted directory
+      extracted_dir="${PACKAGE_NAME%.tar.gz}"
+      if [ -f "${extracted_dir}/${BIN}" ]; then
+        mv "${extracted_dir}/${BIN}" "${PWD}/"
+        rm -f "${PACKAGE_NAME}"
+        rm -rf "${extracted_dir}"
+        chmod +x "${BIN}"
+        echo "Installation completed successfully!"
+        echo "Run './${BIN} --help' to get started"
+      else
+        echo "Error: Binary ${BIN} not found in extracted archive"
+        rm -f "${PACKAGE_NAME}"
+        rm -rf "${extracted_dir}"
+        exit 1
+      fi
     fi
   fi
 }

src/catalog/src/system_schema/pg_catalog/pg_class.rs

@@ -169,7 +169,7 @@ impl DfPartitionStream for PGClass {
 }
 
 /// Builds the `pg_catalog.pg_class` table row by row
-/// TODO(J0HN50N133): `relowner` is always the [`DUMMY_OWNER_ID`] cuz we don't have user.
+/// TODO(J0HN50N133): `relowner` is always the [`DUMMY_OWNER_ID`] because we don't have users.
 /// Once we have user system, make it the actual owner of the table.
 struct PGClassBuilder {
     schema: SchemaRef,

src/cli/src/metadata/repair.rs

@@ -241,7 +241,6 @@ impl RepairTool {
             let alter_table_request = alter_table::make_alter_region_request_for_peer(
                 logical_table_id,
                 &alter_table_expr,
-                full_table_metadata.table_info.ident.version,
                 peer,
                 physical_region_routes,
             )?;

src/cli/src/metadata/repair/alter_table.rs

@@ -66,7 +66,6 @@ pub fn generate_alter_table_expr_for_all_columns(
 pub fn make_alter_region_request_for_peer(
     logical_table_id: TableId,
     alter_table_expr: &AlterTableExpr,
-    schema_version: u64,
     peer: &Peer,
     region_routes: &[RegionRoute],
 ) -> Result<RegionRequest> {
@@ -74,7 +73,7 @@ pub fn make_alter_region_request_for_peer(
     let mut requests = Vec::with_capacity(regions_on_this_peer.len());
     for region_number in &regions_on_this_peer {
         let region_id = RegionId::new(logical_table_id, *region_number);
-        let request = make_alter_region_request(region_id, alter_table_expr, schema_version);
+        let request = make_alter_region_request(region_id, alter_table_expr);
         requests.push(request);
     }
 

src/client/src/database.rs

@@ -23,7 +23,7 @@ use api::v1::greptime_request::Request;
 use api::v1::query_request::Query;
 use api::v1::{
     AlterTableExpr, AuthHeader, Basic, CreateTableExpr, DdlRequest, GreptimeRequest,
-    InsertRequests, QueryRequest, RequestHeader,
+    InsertRequests, QueryRequest, RequestHeader, RowInsertRequests,
 };
 use arrow_flight::{FlightData, Ticket};
 use async_stream::stream;
@@ -118,6 +118,7 @@ impl Database {
         }
     }
 
+    /// Set the catalog for the database client.
     pub fn set_catalog(&mut self, catalog: impl Into<String>) {
         self.catalog = catalog.into();
     }
@@ -130,6 +131,7 @@ impl Database {
         }
     }
 
+    /// Set the schema for the database client.
     pub fn set_schema(&mut self, schema: impl Into<String>) {
         self.schema = schema.into();
     }
@@ -142,20 +144,24 @@ impl Database {
         }
     }
 
+    /// Set the timezone for the database client.
     pub fn set_timezone(&mut self, timezone: impl Into<String>) {
         self.timezone = timezone.into();
     }
 
+    /// Set the auth scheme for the database client.
     pub fn set_auth(&mut self, auth: AuthScheme) {
         self.ctx.auth_header = Some(AuthHeader {
             auth_scheme: Some(auth),
         });
     }
 
+    /// Make an InsertRequests request to the database.
     pub async fn insert(&self, requests: InsertRequests) -> Result<u32> {
         self.handle(Request::Inserts(requests)).await
     }
 
+    /// Make an InsertRequests request to the database with hints.
     pub async fn insert_with_hints(
         &self,
         requests: InsertRequests,
@@ -172,6 +178,28 @@ impl Database {
         from_grpc_response(response)
     }
 
+    /// Make a RowInsertRequests request to the database.
+    pub async fn row_inserts(&self, requests: RowInsertRequests) -> Result<u32> {
+        self.handle(Request::RowInserts(requests)).await
+    }
+
+    /// Make a RowInsertRequests request to the database with hints.
+    pub async fn row_inserts_with_hints(
+        &self,
+        requests: RowInsertRequests,
+        hints: &[(&str, &str)],
+    ) -> Result<u32> {
+        let mut client = make_database_client(&self.client)?.inner;
+        let request = self.to_rpc_request(Request::RowInserts(requests));
+
+        let mut request = tonic::Request::new(request);
+        let metadata = request.metadata_mut();
+        Self::put_hints(metadata, hints)?;
+
+        let response = client.handle(request).await?.into_inner();
+        from_grpc_response(response)
+    }
+
     fn put_hints(metadata: &mut MetadataMap, hints: &[(&str, &str)]) -> Result<()> {
         let Some(value) = hints
             .iter()
@@ -187,6 +215,7 @@ impl Database {
         Ok(())
     }
 
+    /// Make a request to the database.
     pub async fn handle(&self, request: Request) -> Result<u32> {
         let mut client = make_database_client(&self.client)?.inner;
         let request = self.to_rpc_request(request);
@@ -250,6 +279,7 @@ impl Database {
         }
     }
 
+    /// Executes a SQL query without any hints.
     pub async fn sql<S>(&self, sql: S) -> Result<Output>
     where
         S: AsRef<str>,
@@ -257,6 +287,7 @@ impl Database {
         self.sql_with_hint(sql, &[]).await
     }
 
+    /// Executes a SQL query with optional hints for query optimization.
     pub async fn sql_with_hint<S>(&self, sql: S, hints: &[(&str, &str)]) -> Result<Output>
     where
         S: AsRef<str>,
@@ -267,6 +298,7 @@ impl Database {
         self.do_get(request, hints).await
     }
 
+    /// Executes a logical plan directly without SQL parsing.
     pub async fn logical_plan(&self, logical_plan: Vec<u8>) -> Result<Output> {
         let request = Request::Query(QueryRequest {
             query: Some(Query::LogicalPlan(logical_plan)),
@@ -274,6 +306,7 @@ impl Database {
         self.do_get(request, &[]).await
     }
 
+    /// Creates a new table using the provided table expression.
     pub async fn create(&self, expr: CreateTableExpr) -> Result<Output> {
         let request = Request::Ddl(DdlRequest {
             expr: Some(DdlExpr::CreateTable(expr)),
@@ -281,6 +314,7 @@ impl Database {
         self.do_get(request, &[]).await
     }
 
+    /// Alters an existing table using the provided alter expression.
     pub async fn alter(&self, expr: AlterTableExpr) -> Result<Output> {
         let request = Request::Ddl(DdlRequest {
             expr: Some(DdlExpr::AlterTable(expr)),

src/cmd/src/bin/greptime.rs

@@ -20,11 +20,11 @@ use cmd::error::{InitTlsProviderSnafu, Result};
 use cmd::options::GlobalOptions;
 use cmd::{cli, datanode, flownode, frontend, metasrv, standalone, App};
 use common_base::Plugins;
-use common_version::version;
+use common_version::{verbose_version, version};
 use servers::install_ring_crypto_provider;
 
 #[derive(Parser)]
-#[command(name = "greptime", author, version, long_version = version(), about)]
+#[command(name = "greptime", author, version, long_version = verbose_version(), about)]
 #[command(propagate_version = true)]
 pub(crate) struct Command {
     #[clap(subcommand)]
@@ -143,10 +143,8 @@ async fn start(cli: Command) -> Result<()> {
 }
 
 fn setup_human_panic() {
-    human_panic::setup_panic!(
-        human_panic::Metadata::new("GreptimeDB", env!("CARGO_PKG_VERSION"))
-            .homepage("https://github.com/GreptimeTeam/greptimedb/discussions")
-    );
+    human_panic::setup_panic!(human_panic::Metadata::new("GreptimeDB", version())
+        .homepage("https://github.com/GreptimeTeam/greptimedb/discussions"));
 
     common_telemetry::set_panic_hook();
 }

src/cmd/src/datanode/builder.rs

@@ -19,7 +19,7 @@ use catalog::kvbackend::MetaKvBackend;
 use common_base::Plugins;
 use common_meta::cache::LayeredCacheRegistryBuilder;
 use common_telemetry::info;
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use datanode::datanode::DatanodeBuilder;
 use datanode::service::DatanodeServiceBuilder;
 use meta_client::MetaClientType;
@@ -67,7 +67,7 @@ impl InstanceBuilder {
             None,
         );
 
-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
         create_resource_limit_metrics(APP_NAME);
 
         plugins::setup_datanode_plugins(plugins, &opts.plugins, dn_opts)

src/cmd/src/flownode.rs

@@ -32,7 +32,7 @@ use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::TableMetadataManager;
 use common_telemetry::info;
 use common_telemetry::logging::{TracingOptions, DEFAULT_LOGGING_DIR};
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use flow::{
     get_flow_auth_options, FlownodeBuilder, FlownodeInstance, FlownodeServiceBuilder,
     FrontendClient, FrontendInvoker,
@@ -279,7 +279,7 @@ impl StartCommand {
             None,
         );
 
-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Flownode start command: {:#?}", self);

src/cmd/src/frontend.rs

@@ -33,7 +33,7 @@ use common_meta::heartbeat::handler::HandlerGroupExecutor;
 use common_telemetry::info;
 use common_telemetry::logging::{TracingOptions, DEFAULT_LOGGING_DIR};
 use common_time::timezone::set_default_timezone;
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use frontend::frontend::Frontend;
 use frontend::heartbeat::HeartbeatTask;
 use frontend::instance::builder::FrontendBuilder;
@@ -102,7 +102,7 @@ impl App for Instance {
 #[derive(Parser)]
 pub struct Command {
     #[clap(subcommand)]
-    subcmd: SubCommand,
+    pub subcmd: SubCommand,
 }
 
 impl Command {
@@ -116,7 +116,7 @@ impl Command {
 }
 
 #[derive(Parser)]
-enum SubCommand {
+pub enum SubCommand {
     Start(StartCommand),
 }
 
@@ -153,7 +153,7 @@ pub struct StartCommand {
     #[clap(long)]
     postgres_addr: Option<String>,
     #[clap(short, long)]
-    config_file: Option<String>,
+    pub config_file: Option<String>,
     #[clap(short, long)]
     influxdb_enable: Option<bool>,
     #[clap(long, value_delimiter = ',', num_args = 1..)]
@@ -169,7 +169,7 @@ pub struct StartCommand {
     #[clap(long)]
     disable_dashboard: Option<bool>,
     #[clap(long, default_value = "GREPTIMEDB_FRONTEND")]
-    env_prefix: String,
+    pub env_prefix: String,
 }
 
 impl StartCommand {
@@ -282,7 +282,7 @@ impl StartCommand {
             opts.component.slow_query.as_ref(),
         );
 
-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Frontend start command: {:#?}", self);

src/cmd/src/lib.rs

@@ -112,7 +112,7 @@ pub trait App: Send {
 pub fn log_versions(version: &str, short_version: &str, app: &str) {
     // Report app version as gauge.
     APP_VERSION
-        .with_label_values(&[env!("CARGO_PKG_VERSION"), short_version, app])
+        .with_label_values(&[common_version::version(), short_version, app])
         .inc();
 
     // Log version and argument flags.

src/cmd/src/metasrv.rs

@@ -22,7 +22,7 @@ use common_base::Plugins;
 use common_config::Configurable;
 use common_telemetry::info;
 use common_telemetry::logging::{TracingOptions, DEFAULT_LOGGING_DIR};
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use meta_srv::bootstrap::MetasrvInstance;
 use meta_srv::metasrv::BackendImpl;
 use snafu::ResultExt;
@@ -324,7 +324,7 @@ impl StartCommand {
             None,
         );
 
-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Metasrv start command: {:#?}", self);
@@ -340,12 +340,12 @@ impl StartCommand {
             .await
             .context(StartMetaServerSnafu)?;
 
-        let builder = meta_srv::bootstrap::metasrv_builder(&opts, plugins.clone(), None)
+        let builder = meta_srv::bootstrap::metasrv_builder(&opts, plugins, None)
             .await
             .context(error::BuildMetaServerSnafu)?;
         let metasrv = builder.build().await.context(error::BuildMetaServerSnafu)?;
 
-        let instance = MetasrvInstance::new(opts, plugins, metasrv)
+        let instance = MetasrvInstance::new(metasrv)
             .await
             .context(error::BuildMetaServerSnafu)?;
 

src/cmd/src/standalone.rs

@@ -51,7 +51,7 @@ use common_telemetry::logging::{
     LoggingOptions, SlowQueryOptions, TracingOptions, DEFAULT_LOGGING_DIR,
 };
 use common_time::timezone::set_default_timezone;
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use common_wal::config::DatanodeWalConfig;
 use datanode::config::{DatanodeOptions, ProcedureConfig, RegionEngineConfig, StorageConfig};
 use datanode::datanode::{Datanode, DatanodeBuilder};
@@ -485,7 +485,7 @@ impl StartCommand {
             opts.component.slow_query.as_ref(),
         );
 
-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Standalone start command: {:#?}", self);

src/common/function/src/system/pg_catalog/version.rs

@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt;
 use std::sync::Arc;
-use std::{env, fmt};
 
 use common_query::error::Result;
 use common_query::prelude::{Signature, Volatility};
@@ -47,7 +47,7 @@ impl Function for PGVersionFunction {
     fn eval(&self, _func_ctx: &FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
         let result = StringVector::from(vec![format!(
             "PostgreSQL 16.3 GreptimeDB {}",
-            env!("CARGO_PKG_VERSION")
+            common_version::version()
         )]);
         Ok(Arc::new(result))
     }

src/common/function/src/system/version.rs

@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt;
 use std::sync::Arc;
-use std::{env, fmt};
 
 use common_query::error::Result;
 use common_query::prelude::{Signature, Volatility};
@@ -52,13 +52,13 @@ impl Function for VersionFunction {
                     "{}-greptimedb-{}",
                     std::env::var("GREPTIMEDB_MYSQL_SERVER_VERSION")
                         .unwrap_or_else(|_| "8.4.2".to_string()),
-                    env!("CARGO_PKG_VERSION")
+                    common_version::version()
                 )
             }
             Channel::Postgres => {
-                format!("16.3-greptimedb-{}", env!("CARGO_PKG_VERSION"))
+                format!("16.3-greptimedb-{}", common_version::version())
             }
-            _ => env!("CARGO_PKG_VERSION").to_string(),
+            _ => common_version::version().to_string(),
         };
         let result = StringVector::from(vec![version]);
         Ok(Arc::new(result))

src/common/meta/Cargo.toml

@@ -69,6 +69,7 @@ table = { workspace = true, features = ["testing"] }
 tokio.workspace = true
 tokio-postgres = { workspace = true, optional = true }
 tonic.workspace = true
+tracing.workspace = true
 typetag.workspace = true
 
 [dev-dependencies]

src/common/meta/src/ddl.rs

@@ -50,7 +50,6 @@ pub mod drop_flow;
 pub mod drop_table;
 pub mod drop_view;
 pub mod flow_meta;
-mod physical_table_metadata;
 pub mod table_meta;
 #[cfg(any(test, feature = "testing"))]
 pub mod test_util;

src/common/meta/src/ddl/alter_logical_tables.rs

@@ -12,20 +12,17 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod check;
-mod metadata;
-mod region_request;
-mod table_cache_keys;
+mod executor;
 mod update_metadata;
+mod validator;
 
 use api::region::RegionResponse;
 use async_trait::async_trait;
 use common_catalog::format_full_table_name;
 use common_procedure::error::{FromJsonSnafu, Result as ProcedureResult, ToJsonSnafu};
 use common_procedure::{Context, LockKey, Procedure, Status};
-use common_telemetry::{error, info, warn};
-use futures_util::future;
-pub use region_request::make_alter_region_request;
+use common_telemetry::{debug, error, info, warn};
+pub use executor::make_alter_region_request;
 use serde::{Deserialize, Serialize};
 use snafu::ResultExt;
 use store_api::metadata::ColumnMetadata;
@@ -33,10 +30,12 @@ use store_api::metric_engine_consts::ALTER_PHYSICAL_EXTENSION_KEY;
 use strum::AsRefStr;
 use table::metadata::TableId;
 
-use crate::ddl::utils::{
-    add_peer_context_if_needed, extract_column_metadatas, map_to_procedure_error,
-    sync_follower_regions,
+use crate::cache_invalidator::Context as CacheContext;
+use crate::ddl::alter_logical_tables::executor::AlterLogicalTablesExecutor;
+use crate::ddl::alter_logical_tables::validator::{
+    retain_unskipped, AlterLogicalTableValidator, ValidatorResult,
 };
+use crate::ddl::utils::{extract_column_metadatas, map_to_procedure_error, sync_follower_regions};
 use crate::ddl::DdlContext;
 use crate::error::Result;
 use crate::instruction::CacheIdent;
@@ -46,13 +45,38 @@ use crate::key::DeserializedValueWithBytes;
 use crate::lock_key::{CatalogLock, SchemaLock, TableLock};
 use crate::metrics;
 use crate::rpc::ddl::AlterTableTask;
-use crate::rpc::router::{find_leaders, RegionRoute};
+use crate::rpc::router::RegionRoute;
 
 pub struct AlterLogicalTablesProcedure {
     pub context: DdlContext,
     pub data: AlterTablesData,
 }
 
+/// Builds the validator from the [`AlterTablesData`].
+fn build_validator_from_alter_table_data<'a>(
+    data: &'a AlterTablesData,
+) -> AlterLogicalTableValidator<'a> {
+    let phsycial_table_id = data.physical_table_id;
+    let alters = data
+        .tasks
+        .iter()
+        .map(|task| &task.alter_table)
+        .collect::<Vec<_>>();
+    AlterLogicalTableValidator::new(phsycial_table_id, alters)
+}
+
+/// Builds the executor from the [`AlterTablesData`].
+fn build_executor_from_alter_expr<'a>(data: &'a AlterTablesData) -> AlterLogicalTablesExecutor<'a> {
+    debug_assert_eq!(data.tasks.len(), data.table_info_values.len());
+    let alters = data
+        .tasks
+        .iter()
+        .zip(data.table_info_values.iter())
+        .map(|(task, table_info)| (table_info.table_info.ident.table_id, &task.alter_table))
+        .collect::<Vec<_>>();
+    AlterLogicalTablesExecutor::new(alters)
+}
+
 impl AlterLogicalTablesProcedure {
     pub const TYPE_NAME: &'static str = "metasrv-procedure::AlterLogicalTables";
 
@@ -82,35 +106,44 @@ impl AlterLogicalTablesProcedure {
     }
 
     pub(crate) async fn on_prepare(&mut self) -> Result<Status> {
-        // Checks all the tasks
-        self.check_input_tasks()?;
-        // Fills the table info values
-        self.fill_table_info_values().await?;
-        // Checks the physical table, must after [fill_table_info_values]
-        self.check_physical_table().await?;
-        // Fills the physical table info
-        self.fill_physical_table_info().await?;
-        // Filter the finished tasks
-        let finished_tasks = self.check_finished_tasks()?;
-        let already_finished_count = finished_tasks
-            .iter()
-            .map(|x| if *x { 1 } else { 0 })
-            .sum::<usize>();
-        let apply_tasks_count = self.data.tasks.len();
-        if already_finished_count == apply_tasks_count {
+        let validator = build_validator_from_alter_table_data(&self.data);
+        let ValidatorResult {
+            num_skipped,
+            skip_alter,
+            table_info_values,
+            physical_table_info,
+            physical_table_route,
+        } = validator
+            .validate(&self.context.table_metadata_manager)
+            .await?;
+
+        let num_tasks = self.data.tasks.len();
+        if num_skipped == num_tasks {
             info!("All the alter tasks are finished, will skip the procedure.");
+            let cache_ident_keys = AlterLogicalTablesExecutor::build_cache_ident_keys(
+                &physical_table_info,
+                &table_info_values
+                    .iter()
+                    .map(|v| v.get_inner_ref())
+                    .collect::<Vec<_>>(),
+            );
+            self.data.table_cache_keys_to_invalidate = cache_ident_keys;
             // Re-invalidate the table cache
             self.data.state = AlterTablesState::InvalidateTableCache;
             return Ok(Status::executing(true));
-        } else if already_finished_count > 0 {
+        } else if num_skipped > 0 {
             info!(
                 "There are {} alter tasks, {} of them were already finished.",
-                apply_tasks_count, already_finished_count
+                num_tasks, num_skipped
             );
         }
-        self.filter_task(&finished_tasks)?;
 
-        // Next state
+        // Updates the procedure state.
+        retain_unskipped(&mut self.data.tasks, &skip_alter);
+        self.data.physical_table_info = Some(physical_table_info);
+        self.data.physical_table_route = Some(physical_table_route);
+        self.data.table_info_values = table_info_values;
+        debug_assert_eq!(self.data.tasks.len(), self.data.table_info_values.len());
         self.data.state = AlterTablesState::SubmitAlterRegionRequests;
         Ok(Status::executing(true))
     }
@@ -118,25 +151,13 @@ impl AlterLogicalTablesProcedure {
     pub(crate) async fn on_submit_alter_region_requests(&mut self) -> Result<Status> {
         // Safety: we have checked the state in on_prepare
         let physical_table_route = &self.data.physical_table_route.as_ref().unwrap();
-        let leaders = find_leaders(&physical_table_route.region_routes);
-        let mut alter_region_tasks = Vec::with_capacity(leaders.len());
-
-        for peer in leaders {
-            let requester = self.context.node_manager.datanode(&peer).await;
-            let request = self.make_request(&peer, &physical_table_route.region_routes)?;
-
-            alter_region_tasks.push(async move {
-                requester
-                    .handle(request)
-                    .await
-                    .map_err(add_peer_context_if_needed(peer))
-            });
-        }
-
-        let mut results = future::join_all(alter_region_tasks)
-            .await
-            .into_iter()
-            .collect::<Result<Vec<_>>>()?;
+        let executor = build_executor_from_alter_expr(&self.data);
+        let mut results = executor
+            .on_alter_regions(
+                &self.context.node_manager,
+                &physical_table_route.region_routes,
+            )
+            .await?;
 
         if let Some(column_metadatas) =
             extract_column_metadatas(&mut results, ALTER_PHYSICAL_EXTENSION_KEY)?
@@ -177,7 +198,18 @@ impl AlterLogicalTablesProcedure {
         self.update_physical_table_metadata().await?;
         self.update_logical_tables_metadata().await?;
 
-        self.data.build_cache_keys_to_invalidate();
+        let logical_table_info_values = self
+            .data
+            .table_info_values
+            .iter()
+            .map(|v| v.get_inner_ref())
+            .collect::<Vec<_>>();
+
+        let cache_ident_keys = AlterLogicalTablesExecutor::build_cache_ident_keys(
+            self.data.physical_table_info.as_ref().unwrap(),
+            &logical_table_info_values,
+        );
+        self.data.table_cache_keys_to_invalidate = cache_ident_keys;
         self.data.clear_metadata_fields();
 
         self.data.state = AlterTablesState::InvalidateTableCache;
@@ -187,9 +219,16 @@ impl AlterLogicalTablesProcedure {
     pub(crate) async fn on_invalidate_table_cache(&mut self) -> Result<Status> {
         let to_invalidate = &self.data.table_cache_keys_to_invalidate;
 
+        let ctx = CacheContext {
+            subject: Some(format!(
+                "Invalidate table cache by altering logical tables, physical_table_id: {}",
+                self.data.physical_table_id,
+            )),
+        };
+
         self.context
             .cache_invalidator
-            .invalidate(&Default::default(), to_invalidate)
+            .invalidate(&ctx, to_invalidate)
             .await?;
         Ok(Status::done())
     }
@@ -209,6 +248,10 @@ impl Procedure for AlterLogicalTablesProcedure {
         let _timer = metrics::METRIC_META_PROCEDURE_ALTER_TABLE
             .with_label_values(&[step])
             .start_timer();
+        debug!(
+            "Executing alter logical tables procedure, state: {:?}",
+            state
+        );
 
         match state {
             AlterTablesState::Prepare => self.on_prepare().await,

src/common/meta/src/ddl/alter_logical_tables/check.rs

@@ -1,136 +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::collections::HashSet;
-
-use api::v1::alter_table_expr::Kind;
-use snafu::{ensure, OptionExt};
-
-use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
-use crate::error::{AlterLogicalTablesInvalidArgumentsSnafu, Result};
-use crate::key::table_info::TableInfoValue;
-use crate::key::table_route::TableRouteValue;
-use crate::rpc::ddl::AlterTableTask;
-
-impl AlterLogicalTablesProcedure {
-    pub(crate) fn check_input_tasks(&self) -> Result<()> {
-        self.check_schema()?;
-        self.check_alter_kind()?;
-        Ok(())
-    }
-
-    pub(crate) async fn check_physical_table(&self) -> Result<()> {
-        let table_route_manager = self.context.table_metadata_manager.table_route_manager();
-        let table_ids = self
-            .data
-            .table_info_values
-            .iter()
-            .map(|v| v.table_info.ident.table_id)
-            .collect::<Vec<_>>();
-        let table_routes = table_route_manager
-            .table_route_storage()
-            .batch_get(&table_ids)
-            .await?;
-        let physical_table_id = self.data.physical_table_id;
-        let is_same_physical_table = table_routes.iter().all(|r| {
-            if let Some(TableRouteValue::Logical(r)) = r {
-                r.physical_table_id() == physical_table_id
-            } else {
-                false
-            }
-        });
-
-        ensure!(
-            is_same_physical_table,
-            AlterLogicalTablesInvalidArgumentsSnafu {
-                err_msg: "All the tasks should have the same physical table id"
-            }
-        );
-
-        Ok(())
-    }
-
-    pub(crate) fn check_finished_tasks(&self) -> Result<Vec<bool>> {
-        let task = &self.data.tasks;
-        let table_info_values = &self.data.table_info_values;
-
-        Ok(task
-            .iter()
-            .zip(table_info_values.iter())
-            .map(|(task, table)| Self::check_finished_task(task, table))
-            .collect())
-    }
-
-    // Checks if the schemas of the tasks are the same
-    fn check_schema(&self) -> Result<()> {
-        let is_same_schema = self.data.tasks.windows(2).all(|pair| {
-            pair[0].alter_table.catalog_name == pair[1].alter_table.catalog_name
-                && pair[0].alter_table.schema_name == pair[1].alter_table.schema_name
-        });
-
-        ensure!(
-            is_same_schema,
-            AlterLogicalTablesInvalidArgumentsSnafu {
-                err_msg: "Schemas of the tasks are not the same"
-            }
-        );
-
-        Ok(())
-    }
-
-    fn check_alter_kind(&self) -> Result<()> {
-        for task in &self.data.tasks {
-            let kind = task.alter_table.kind.as_ref().context(
-                AlterLogicalTablesInvalidArgumentsSnafu {
-                    err_msg: "Alter kind is missing",
-                },
-            )?;
-            let Kind::AddColumns(_) = kind else {
-                return AlterLogicalTablesInvalidArgumentsSnafu {
-                    err_msg: "Only support add columns operation",
-                }
-                .fail();
-            };
-        }
-
-        Ok(())
-    }
-
-    fn check_finished_task(task: &AlterTableTask, table: &TableInfoValue) -> bool {
-        let columns = table
-            .table_info
-            .meta
-            .schema
-            .column_schemas
-            .iter()
-            .map(|c| &c.name)
-            .collect::<HashSet<_>>();
-
-        let Some(kind) = task.alter_table.kind.as_ref() else {
-            return true; // Never get here since we have checked it in `check_alter_kind`
-        };
-        let Kind::AddColumns(add_columns) = kind else {
-            return true; // Never get here since we have checked it in `check_alter_kind`
-        };
-
-        // We only check that all columns have been finished. That is to say,
-        // if one part is finished but another part is not, it will be considered
-        // unfinished.
-        add_columns
-            .add_columns
-            .iter()
-            .map(|add_column| add_column.column_def.as_ref().map(|c| &c.name))
-            .all(|column| column.map(|c| columns.contains(c)).unwrap_or(false))
-    }
-}

src/common/meta/src/ddl/alter_logical_tables/executor.rs

@@ -0,0 +1,216 @@
+// 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::collections::HashMap;
+
+use api::region::RegionResponse;
+use api::v1::alter_table_expr::Kind;
+use api::v1::region::{
+    alter_request, region_request, AddColumn, AddColumns, AlterRequest, AlterRequests,
+    RegionColumnDef, RegionRequest, RegionRequestHeader,
+};
+use api::v1::{self, AlterTableExpr};
+use common_telemetry::tracing_context::TracingContext;
+use common_telemetry::{debug, warn};
+use futures::future;
+use store_api::metadata::ColumnMetadata;
+use store_api::storage::{RegionId, RegionNumber, TableId};
+
+use crate::ddl::utils::{add_peer_context_if_needed, raw_table_info};
+use crate::error::Result;
+use crate::instruction::CacheIdent;
+use crate::key::table_info::TableInfoValue;
+use crate::key::{DeserializedValueWithBytes, RegionDistribution, TableMetadataManagerRef};
+use crate::node_manager::NodeManagerRef;
+use crate::rpc::router::{find_leaders, region_distribution, RegionRoute};
+
+/// [AlterLogicalTablesExecutor] performs:
+/// - Alters logical regions on the datanodes.
+/// - Updates table metadata for alter table operation.
+pub struct AlterLogicalTablesExecutor<'a> {
+    /// The alter table expressions.
+    ///
+    /// The first element is the logical table id, the second element is the alter table expression.
+    alters: Vec<(TableId, &'a AlterTableExpr)>,
+}
+
+impl<'a> AlterLogicalTablesExecutor<'a> {
+    pub fn new(alters: Vec<(TableId, &'a AlterTableExpr)>) -> Self {
+        Self { alters }
+    }
+
+    /// Alters logical regions on the datanodes.
+    pub(crate) async fn on_alter_regions(
+        &self,
+        node_manager: &NodeManagerRef,
+        region_routes: &[RegionRoute],
+    ) -> Result<Vec<RegionResponse>> {
+        let region_distribution = region_distribution(region_routes);
+        let leaders = find_leaders(region_routes)
+            .into_iter()
+            .map(|p| (p.id, p))
+            .collect::<HashMap<_, _>>();
+        let mut alter_region_tasks = Vec::with_capacity(leaders.len());
+        for (datanode_id, region_role_set) in region_distribution {
+            if region_role_set.leader_regions.is_empty() {
+                continue;
+            }
+            // Safety: must exists.
+            let peer = leaders.get(&datanode_id).unwrap();
+            let requester = node_manager.datanode(peer).await;
+            let requests = self.make_alter_region_request(&region_role_set.leader_regions);
+            let requester = requester.clone();
+            let peer = peer.clone();
+
+            debug!("Sending alter region requests to datanode {}", peer);
+            alter_region_tasks.push(async move {
+                requester
+                    .handle(make_request(requests))
+                    .await
+                    .map_err(add_peer_context_if_needed(peer))
+            });
+        }
+
+        future::join_all(alter_region_tasks)
+            .await
+            .into_iter()
+            .collect::<Result<Vec<_>>>()
+    }
+
+    fn make_alter_region_request(&self, region_numbers: &[RegionNumber]) -> AlterRequests {
+        let mut requests = Vec::with_capacity(region_numbers.len() * self.alters.len());
+        for (table_id, alter) in self.alters.iter() {
+            for region_number in region_numbers {
+                let region_id = RegionId::new(*table_id, *region_number);
+                let request = make_alter_region_request(region_id, alter);
+                requests.push(request);
+            }
+        }
+
+        AlterRequests { requests }
+    }
+
+    /// Updates table metadata for alter table operation.
+    ///
+    /// ## Panic:
+    /// - If the region distribution is not set when updating table metadata.
+    pub(crate) async fn on_alter_metadata(
+        physical_table_id: TableId,
+        table_metadata_manager: &TableMetadataManagerRef,
+        current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
+        region_distribution: RegionDistribution,
+        physical_columns: &[ColumnMetadata],
+    ) -> Result<()> {
+        if physical_columns.is_empty() {
+            warn!("No physical columns found, leaving the physical table's schema unchanged when altering logical tables");
+            return Ok(());
+        }
+
+        let table_ref = current_table_info_value.table_ref();
+        let table_id = physical_table_id;
+
+        // Generates new table info
+        let old_raw_table_info = current_table_info_value.table_info.clone();
+        let new_raw_table_info =
+            raw_table_info::build_new_physical_table_info(old_raw_table_info, physical_columns);
+
+        debug!(
+            "Starting update table: {} metadata, table_id: {}, new table info: {:?}",
+            table_ref, table_id, new_raw_table_info
+        );
+
+        table_metadata_manager
+            .update_table_info(
+                current_table_info_value,
+                Some(region_distribution),
+                new_raw_table_info,
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    /// Builds the cache ident keys for the alter logical tables.
+    ///
+    /// The cache ident keys are:
+    /// - The table id of the logical tables.
+    /// - The table name of the logical tables.
+    /// - The table id of the physical table.
+    pub(crate) fn build_cache_ident_keys(
+        physical_table_info: &TableInfoValue,
+        logical_table_info_values: &[&TableInfoValue],
+    ) -> Vec<CacheIdent> {
+        let mut cache_keys = Vec::with_capacity(logical_table_info_values.len() * 2 + 2);
+        cache_keys.extend(logical_table_info_values.iter().flat_map(|table| {
+            vec![
+                CacheIdent::TableId(table.table_info.ident.table_id),
+                CacheIdent::TableName(table.table_name()),
+            ]
+        }));
+        cache_keys.push(CacheIdent::TableId(
+            physical_table_info.table_info.ident.table_id,
+        ));
+        cache_keys.push(CacheIdent::TableName(physical_table_info.table_name()));
+
+        cache_keys
+    }
+}
+
+fn make_request(alter_requests: AlterRequests) -> RegionRequest {
+    RegionRequest {
+        header: Some(RegionRequestHeader {
+            tracing_context: TracingContext::from_current_span().to_w3c(),
+            ..Default::default()
+        }),
+        body: Some(region_request::Body::Alters(alter_requests)),
+    }
+}
+
+/// Makes an alter region request.
+pub fn make_alter_region_request(
+    region_id: RegionId,
+    alter_table_expr: &AlterTableExpr,
+) -> AlterRequest {
+    let region_id = region_id.as_u64();
+    let kind = match &alter_table_expr.kind {
+        Some(Kind::AddColumns(add_columns)) => Some(alter_request::Kind::AddColumns(
+            to_region_add_columns(add_columns),
+        )),
+        _ => unreachable!(), // Safety: we have checked the kind in check_input_tasks
+    };
+
+    AlterRequest {
+        region_id,
+        schema_version: 0,
+        kind,
+    }
+}
+
+fn to_region_add_columns(add_columns: &v1::AddColumns) -> AddColumns {
+    let add_columns = add_columns
+        .add_columns
+        .iter()
+        .map(|add_column| {
+            let region_column_def = RegionColumnDef {
+                column_def: add_column.column_def.clone(),
+                ..Default::default() // other fields are not used in alter logical table
+            };
+            AddColumn {
+                column_def: Some(region_column_def),
+                ..Default::default() // other fields are not used in alter logical table
+            }
+        })
+        .collect();
+    AddColumns { add_columns }
+}

src/common/meta/src/ddl/alter_logical_tables/metadata.rs

@@ -1,158 +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_catalog::format_full_table_name;
-use snafu::OptionExt;
-use table::metadata::TableId;
-
-use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
-use crate::error::{
-    AlterLogicalTablesInvalidArgumentsSnafu, Result, TableInfoNotFoundSnafu, TableNotFoundSnafu,
-    TableRouteNotFoundSnafu,
-};
-use crate::key::table_info::TableInfoValue;
-use crate::key::table_name::TableNameKey;
-use crate::key::table_route::TableRouteValue;
-use crate::key::DeserializedValueWithBytes;
-use crate::rpc::ddl::AlterTableTask;
-
-impl AlterLogicalTablesProcedure {
-    pub(crate) fn filter_task(&mut self, finished_tasks: &[bool]) -> Result<()> {
-        debug_assert_eq!(finished_tasks.len(), self.data.tasks.len());
-        debug_assert_eq!(finished_tasks.len(), self.data.table_info_values.len());
-        self.data.tasks = self
-            .data
-            .tasks
-            .drain(..)
-            .zip(finished_tasks.iter())
-            .filter_map(|(task, finished)| if *finished { None } else { Some(task) })
-            .collect();
-        self.data.table_info_values = self
-            .data
-            .table_info_values
-            .drain(..)
-            .zip(finished_tasks.iter())
-            .filter_map(|(table_info_value, finished)| {
-                if *finished {
-                    None
-                } else {
-                    Some(table_info_value)
-                }
-            })
-            .collect();
-
-        Ok(())
-    }
-
-    pub(crate) async fn fill_physical_table_info(&mut self) -> Result<()> {
-        let (physical_table_info, physical_table_route) = self
-            .context
-            .table_metadata_manager
-            .get_full_table_info(self.data.physical_table_id)
-            .await?;
-
-        let physical_table_info = physical_table_info.with_context(|| TableInfoNotFoundSnafu {
-            table: format!("table id - {}", self.data.physical_table_id),
-        })?;
-        let physical_table_route = physical_table_route
-            .context(TableRouteNotFoundSnafu {
-                table_id: self.data.physical_table_id,
-            })?
-            .into_inner();
-
-        self.data.physical_table_info = Some(physical_table_info);
-        let TableRouteValue::Physical(physical_table_route) = physical_table_route else {
-            return AlterLogicalTablesInvalidArgumentsSnafu {
-                err_msg: format!(
-                    "expected a physical table but got a logical table: {:?}",
-                    self.data.physical_table_id
-                ),
-            }
-            .fail();
-        };
-        self.data.physical_table_route = Some(physical_table_route);
-
-        Ok(())
-    }
-
-    pub(crate) async fn fill_table_info_values(&mut self) -> Result<()> {
-        let table_ids = self.get_all_table_ids().await?;
-        let table_info_values = self.get_all_table_info_values(&table_ids).await?;
-        debug_assert_eq!(table_info_values.len(), self.data.tasks.len());
-        self.data.table_info_values = table_info_values;
-
-        Ok(())
-    }
-
-    async fn get_all_table_info_values(
-        &self,
-        table_ids: &[TableId],
-    ) -> Result<Vec<DeserializedValueWithBytes<TableInfoValue>>> {
-        let table_info_manager = self.context.table_metadata_manager.table_info_manager();
-        let mut table_info_map = table_info_manager.batch_get_raw(table_ids).await?;
-        let mut table_info_values = Vec::with_capacity(table_ids.len());
-        for (table_id, task) in table_ids.iter().zip(self.data.tasks.iter()) {
-            let table_info_value =
-                table_info_map
-                    .remove(table_id)
-                    .with_context(|| TableInfoNotFoundSnafu {
-                        table: extract_table_name(task),
-                    })?;
-            table_info_values.push(table_info_value);
-        }
-
-        Ok(table_info_values)
-    }
-
-    async fn get_all_table_ids(&self) -> Result<Vec<TableId>> {
-        let table_name_manager = self.context.table_metadata_manager.table_name_manager();
-        let table_name_keys = self
-            .data
-            .tasks
-            .iter()
-            .map(|task| extract_table_name_key(task))
-            .collect();
-
-        let table_name_values = table_name_manager.batch_get(table_name_keys).await?;
-        let mut table_ids = Vec::with_capacity(table_name_values.len());
-        for (value, task) in table_name_values.into_iter().zip(self.data.tasks.iter()) {
-            let table_id = value
-                .with_context(|| TableNotFoundSnafu {
-                    table_name: extract_table_name(task),
-                })?
-                .table_id();
-            table_ids.push(table_id);
-        }
-
-        Ok(table_ids)
-    }
-}
-
-#[inline]
-fn extract_table_name(task: &AlterTableTask) -> String {
-    format_full_table_name(
-        &task.alter_table.catalog_name,
-        &task.alter_table.schema_name,
-        &task.alter_table.table_name,
-    )
-}
-
-#[inline]
-fn extract_table_name_key(task: &AlterTableTask) -> TableNameKey {
-    TableNameKey::new(
-        &task.alter_table.catalog_name,
-        &task.alter_table.schema_name,
-        &task.alter_table.table_name,
-    )
-}

src/common/meta/src/ddl/alter_logical_tables/region_request.rs

@@ -1,113 +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 api::v1::alter_table_expr::Kind;
-use api::v1::region::{
-    alter_request, region_request, AddColumn, AddColumns, AlterRequest, AlterRequests,
-    RegionColumnDef, RegionRequest, RegionRequestHeader,
-};
-use api::v1::{self, AlterTableExpr};
-use common_telemetry::tracing_context::TracingContext;
-use store_api::storage::RegionId;
-
-use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
-use crate::error::Result;
-use crate::peer::Peer;
-use crate::rpc::router::{find_leader_regions, RegionRoute};
-
-impl AlterLogicalTablesProcedure {
-    pub(crate) fn make_request(
-        &self,
-        peer: &Peer,
-        region_routes: &[RegionRoute],
-    ) -> Result<RegionRequest> {
-        let alter_requests = self.make_alter_region_requests(peer, region_routes)?;
-        let request = RegionRequest {
-            header: Some(RegionRequestHeader {
-                tracing_context: TracingContext::from_current_span().to_w3c(),
-                ..Default::default()
-            }),
-            body: Some(region_request::Body::Alters(alter_requests)),
-        };
-
-        Ok(request)
-    }
-
-    fn make_alter_region_requests(
-        &self,
-        peer: &Peer,
-        region_routes: &[RegionRoute],
-    ) -> Result<AlterRequests> {
-        let tasks = &self.data.tasks;
-        let regions_on_this_peer = find_leader_regions(region_routes, peer);
-        let mut requests = Vec::with_capacity(tasks.len() * regions_on_this_peer.len());
-        for (task, table) in self
-            .data
-            .tasks
-            .iter()
-            .zip(self.data.table_info_values.iter())
-        {
-            for region_number in &regions_on_this_peer {
-                let region_id = RegionId::new(table.table_info.ident.table_id, *region_number);
-                let request = make_alter_region_request(
-                    region_id,
-                    &task.alter_table,
-                    table.table_info.ident.version,
-                );
-                requests.push(request);
-            }
-        }
-
-        Ok(AlterRequests { requests })
-    }
-}
-
-/// Makes an alter region request.
-pub fn make_alter_region_request(
-    region_id: RegionId,
-    alter_table_expr: &AlterTableExpr,
-    schema_version: u64,
-) -> AlterRequest {
-    let region_id = region_id.as_u64();
-    let kind = match &alter_table_expr.kind {
-        Some(Kind::AddColumns(add_columns)) => Some(alter_request::Kind::AddColumns(
-            to_region_add_columns(add_columns),
-        )),
-        _ => unreachable!(), // Safety: we have checked the kind in check_input_tasks
-    };
-
-    AlterRequest {
-        region_id,
-        schema_version,
-        kind,
-    }
-}
-
-fn to_region_add_columns(add_columns: &v1::AddColumns) -> AddColumns {
-    let add_columns = add_columns
-        .add_columns
-        .iter()
-        .map(|add_column| {
-            let region_column_def = RegionColumnDef {
-                column_def: add_column.column_def.clone(),
-                ..Default::default() // other fields are not used in alter logical table
-            };
-            AddColumn {
-                column_def: Some(region_column_def),
-                ..Default::default() // other fields are not used in alter logical table
-            }
-        })
-        .collect();
-    AddColumns { add_columns }
-}

src/common/meta/src/ddl/alter_logical_tables/table_cache_keys.rs

@@ -1,50 +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 table::metadata::RawTableInfo;
-use table::table_name::TableName;
-
-use crate::ddl::alter_logical_tables::AlterTablesData;
-use crate::instruction::CacheIdent;
-
-impl AlterTablesData {
-    pub(crate) fn build_cache_keys_to_invalidate(&mut self) {
-        let mut cache_keys = self
-            .table_info_values
-            .iter()
-            .flat_map(|table| {
-                vec![
-                    CacheIdent::TableId(table.table_info.ident.table_id),
-                    CacheIdent::TableName(extract_table_name(&table.table_info)),
-                ]
-            })
-            .collect::<Vec<_>>();
-        cache_keys.push(CacheIdent::TableId(self.physical_table_id));
-        // Safety: physical_table_info already filled in previous steps
-        let physical_table_info = &self.physical_table_info.as_ref().unwrap().table_info;
-        cache_keys.push(CacheIdent::TableName(extract_table_name(
-            physical_table_info,
-        )));
-
-        self.table_cache_keys_to_invalidate = cache_keys;
-    }
-}
-
-fn extract_table_name(table_info: &RawTableInfo) -> TableName {
-    TableName::new(
-        &table_info.catalog_name,
-        &table_info.schema_name,
-        &table_info.name,
-    )
-}

src/common/meta/src/ddl/alter_logical_tables/update_metadata.rs

@@ -13,41 +13,35 @@
 // limitations under the License.
 
 use common_grpc_expr::alter_expr_to_request;
-use common_telemetry::warn;
 use itertools::Itertools;
 use snafu::ResultExt;
 use table::metadata::{RawTableInfo, TableInfo};
 
+use crate::ddl::alter_logical_tables::executor::AlterLogicalTablesExecutor;
 use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
-use crate::ddl::physical_table_metadata;
 use crate::error;
 use crate::error::{ConvertAlterTableRequestSnafu, Result};
 use crate::key::table_info::TableInfoValue;
 use crate::key::DeserializedValueWithBytes;
 use crate::rpc::ddl::AlterTableTask;
+use crate::rpc::router::region_distribution;
 
 impl AlterLogicalTablesProcedure {
     pub(crate) async fn update_physical_table_metadata(&mut self) -> Result<()> {
-        if self.data.physical_columns.is_empty() {
-            warn!("No physical columns found, leaving the physical table's schema unchanged when altering logical tables");
-            return Ok(());
-        }
-
         // Safety: must exist.
         let physical_table_info = self.data.physical_table_info.as_ref().unwrap();
+        let physical_table_route = self.data.physical_table_route.as_ref().unwrap();
+        let region_distribution = region_distribution(&physical_table_route.region_routes);
 
-        // Generates new table info
-        let old_raw_table_info = physical_table_info.table_info.clone();
-        let new_raw_table_info = physical_table_metadata::build_new_physical_table_info(
-            old_raw_table_info,
+        // Updates physical table's metadata.
+        AlterLogicalTablesExecutor::on_alter_metadata(
+            self.data.physical_table_id,
+            &self.context.table_metadata_manager,
+            physical_table_info,
+            region_distribution,
             &self.data.physical_columns,
-        );
-
-        // Updates physical table's metadata, and we don't need to touch per-region settings.
-        self.context
-            .table_metadata_manager
-            .update_table_info(physical_table_info, None, new_raw_table_info)
-            .await?;
+        )
+        .await?;
 
         Ok(())
     }

src/common/meta/src/ddl/alter_logical_tables/validator.rs

@@ -0,0 +1,284 @@
+// 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::collections::HashSet;
+
+use api::v1::alter_table_expr::Kind;
+use api::v1::AlterTableExpr;
+use snafu::{ensure, OptionExt};
+use store_api::storage::TableId;
+use table::table_reference::TableReference;
+
+use crate::ddl::utils::table_id::get_all_table_ids_by_names;
+use crate::ddl::utils::table_info::get_all_table_info_values_by_table_ids;
+use crate::error::{
+    AlterLogicalTablesInvalidArgumentsSnafu, Result, TableInfoNotFoundSnafu,
+    TableRouteNotFoundSnafu,
+};
+use crate::key::table_info::TableInfoValue;
+use crate::key::table_route::{PhysicalTableRouteValue, TableRouteManager, TableRouteValue};
+use crate::key::{DeserializedValueWithBytes, TableMetadataManagerRef};
+
+/// [AlterLogicalTableValidator] validates the alter logical expressions.
+pub struct AlterLogicalTableValidator<'a> {
+    physical_table_id: TableId,
+    alters: Vec<&'a AlterTableExpr>,
+}
+
+impl<'a> AlterLogicalTableValidator<'a> {
+    pub fn new(physical_table_id: TableId, alters: Vec<&'a AlterTableExpr>) -> Self {
+        Self {
+            physical_table_id,
+            alters,
+        }
+    }
+
+    /// Validates all alter table expressions have the same schema and catalog.
+    fn validate_schema(&self) -> Result<()> {
+        let is_same_schema = self.alters.windows(2).all(|pair| {
+            pair[0].catalog_name == pair[1].catalog_name
+                && pair[0].schema_name == pair[1].schema_name
+        });
+
+        ensure!(
+            is_same_schema,
+            AlterLogicalTablesInvalidArgumentsSnafu {
+                err_msg: "Schemas of the alter table expressions are not the same"
+            }
+        );
+
+        Ok(())
+    }
+
+    /// Validates that all alter table expressions are of the supported kind.
+    /// Currently only supports `AddColumns` operations.
+    fn validate_alter_kind(&self) -> Result<()> {
+        for alter in &self.alters {
+            let kind = alter
+                .kind
+                .as_ref()
+                .context(AlterLogicalTablesInvalidArgumentsSnafu {
+                    err_msg: "Alter kind is missing",
+                })?;
+
+            let Kind::AddColumns(_) = kind else {
+                return AlterLogicalTablesInvalidArgumentsSnafu {
+                    err_msg: "Only support add columns operation",
+                }
+                .fail();
+            };
+        }
+
+        Ok(())
+    }
+
+    fn table_names(&self) -> Vec<TableReference> {
+        self.alters
+            .iter()
+            .map(|alter| {
+                TableReference::full(&alter.catalog_name, &alter.schema_name, &alter.table_name)
+            })
+            .collect()
+    }
+
+    /// Validates that the physical table info and route exist.
+    ///
+    /// This method performs the following validations:
+    /// 1. Retrieves the full table info and route for the given physical table id
+    /// 2. Ensures the table info and table route exists
+    /// 3. Verifies that the table route is actually a physical table route, not a logical one
+    ///
+    /// Returns a tuple containing the validated table info and physical table route.
+    async fn validate_physical_table(
+        &self,
+        table_metadata_manager: &TableMetadataManagerRef,
+    ) -> Result<(
+        DeserializedValueWithBytes<TableInfoValue>,
+        PhysicalTableRouteValue,
+    )> {
+        let (table_info, table_route) = table_metadata_manager
+            .get_full_table_info(self.physical_table_id)
+            .await?;
+
+        let table_info = table_info.with_context(|| TableInfoNotFoundSnafu {
+            table: format!("table id - {}", self.physical_table_id),
+        })?;
+
+        let physical_table_route = table_route
+            .context(TableRouteNotFoundSnafu {
+                table_id: self.physical_table_id,
+            })?
+            .into_inner();
+
+        let TableRouteValue::Physical(table_route) = physical_table_route else {
+            return AlterLogicalTablesInvalidArgumentsSnafu {
+                err_msg: format!(
+                    "expected a physical table but got a logical table: {:?}",
+                    self.physical_table_id
+                ),
+            }
+            .fail();
+        };
+
+        Ok((table_info, table_route))
+    }
+
+    /// Validates that all logical table routes have the same physical table id.
+    ///
+    /// This method performs the following validations:
+    /// 1. Retrieves table routes for all the given table ids.
+    /// 2. Ensures that all retrieved routes are logical table routes (not physical)
+    /// 3. Verifies that all logical table routes reference the same physical table id.
+    /// 4. Returns an error if any route is not logical or references a different physical table.
+    async fn validate_logical_table_routes(
+        &self,
+        table_route_manager: &TableRouteManager,
+        table_ids: &[TableId],
+    ) -> Result<()> {
+        let table_routes = table_route_manager
+            .table_route_storage()
+            .batch_get(table_ids)
+            .await?;
+
+        let physical_table_id = self.physical_table_id;
+
+        let is_same_physical_table = table_routes.iter().all(|r| {
+            if let Some(TableRouteValue::Logical(r)) = r {
+                r.physical_table_id() == physical_table_id
+            } else {
+                false
+            }
+        });
+
+        ensure!(
+            is_same_physical_table,
+            AlterLogicalTablesInvalidArgumentsSnafu {
+                err_msg: "All the tasks should have the same physical table id"
+            }
+        );
+
+        Ok(())
+    }
+
+    /// Validates the alter logical expressions.
+    ///
+    /// This method performs the following validations:
+    /// 1. Validates that all alter table expressions have the same schema and catalog.
+    /// 2. Validates that all alter table expressions are of the supported kind.
+    /// 3. Validates that the physical table info and route exist.
+    /// 4. Validates that all logical table routes have the same physical table id.
+    ///
+    /// Returns a [ValidatorResult] containing the validation results.
+    pub async fn validate(
+        &self,
+        table_metadata_manager: &TableMetadataManagerRef,
+    ) -> Result<ValidatorResult> {
+        self.validate_schema()?;
+        self.validate_alter_kind()?;
+        let (physical_table_info, physical_table_route) =
+            self.validate_physical_table(table_metadata_manager).await?;
+        let table_names = self.table_names();
+        let table_ids =
+            get_all_table_ids_by_names(table_metadata_manager.table_name_manager(), &table_names)
+                .await?;
+        let mut table_info_values = get_all_table_info_values_by_table_ids(
+            table_metadata_manager.table_info_manager(),
+            &table_ids,
+            &table_names,
+        )
+        .await?;
+        self.validate_logical_table_routes(
+            table_metadata_manager.table_route_manager(),
+            &table_ids,
+        )
+        .await?;
+        let skip_alter = self
+            .alters
+            .iter()
+            .zip(table_info_values.iter())
+            .map(|(task, table)| skip_alter_logical_region(task, table))
+            .collect::<Vec<_>>();
+        retain_unskipped(&mut table_info_values, &skip_alter);
+        let num_skipped = skip_alter.iter().filter(|&&x| x).count();
+
+        Ok(ValidatorResult {
+            num_skipped,
+            skip_alter,
+            table_info_values,
+            physical_table_info,
+            physical_table_route,
+        })
+    }
+}
+
+/// The result of the validator.
+pub(crate) struct ValidatorResult {
+    pub(crate) num_skipped: usize,
+    pub(crate) skip_alter: Vec<bool>,
+    pub(crate) table_info_values: Vec<DeserializedValueWithBytes<TableInfoValue>>,
+    pub(crate) physical_table_info: DeserializedValueWithBytes<TableInfoValue>,
+    pub(crate) physical_table_route: PhysicalTableRouteValue,
+}
+
+/// Retains the elements that are not skipped.
+pub(crate) fn retain_unskipped<T>(target: &mut Vec<T>, skipped: &[bool]) {
+    debug_assert_eq!(target.len(), skipped.len());
+    let mut iter = skipped.iter();
+    target.retain(|_| !iter.next().unwrap());
+}
+
+/// Returns true if does not required to alter the logical region.
+fn skip_alter_logical_region(alter: &AlterTableExpr, table: &TableInfoValue) -> bool {
+    let existing_columns = table
+        .table_info
+        .meta
+        .schema
+        .column_schemas
+        .iter()
+        .map(|c| &c.name)
+        .collect::<HashSet<_>>();
+
+    let Some(kind) = alter.kind.as_ref() else {
+        return true; // Never get here since we have checked it in `validate_alter_kind`
+    };
+    let Kind::AddColumns(add_columns) = kind else {
+        return true; // Never get here since we have checked it in `validate_alter_kind`
+    };
+
+    // We only check that all columns have been finished. That is to say,
+    // if one part is finished but another part is not, it will be considered
+    // unfinished.
+    add_columns
+        .add_columns
+        .iter()
+        .map(|add_column| add_column.column_def.as_ref().map(|c| &c.name))
+        .all(|column| {
+            column
+                .map(|c| existing_columns.contains(c))
+                .unwrap_or(false)
+        })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_retain_unskipped() {
+        let mut target = vec![1, 2, 3, 4, 5];
+        let skipped = vec![false, true, false, true, false];
+        retain_unskipped(&mut target, &skipped);
+        assert_eq!(target, vec![1, 3, 5]);
+    }
+}

src/common/meta/src/ddl/alter_table.rs

@@ -12,10 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod check;
+mod executor;
 mod metadata;
 mod region_request;
-mod update_metadata;
 
 use std::vec;
 
@@ -29,33 +28,29 @@ use common_procedure::{
     Context as ProcedureContext, ContextProvider, Error as ProcedureError, LockKey, PoisonKey,
     PoisonKeys, Procedure, ProcedureId, Status, StringKey,
 };
-use common_telemetry::{debug, error, info, warn};
-use futures::future::{self};
+use common_telemetry::{error, info, warn};
 use serde::{Deserialize, Serialize};
 use snafu::{ensure, ResultExt};
 use store_api::metadata::ColumnMetadata;
 use store_api::metric_engine_consts::TABLE_COLUMN_METADATA_EXTENSION_KEY;
-use store_api::storage::RegionId;
 use strum::AsRefStr;
 use table::metadata::{RawTableInfo, TableId, TableInfo};
 use table::table_reference::TableReference;
 
-use crate::cache_invalidator::Context;
-use crate::ddl::physical_table_metadata::update_table_info_column_ids;
+use crate::ddl::alter_table::executor::AlterTableExecutor;
 use crate::ddl::utils::{
-    add_peer_context_if_needed, extract_column_metadatas, handle_multiple_results,
-    map_to_procedure_error, sync_follower_regions, MultipleResults,
+    extract_column_metadatas, handle_multiple_results, map_to_procedure_error,
+    sync_follower_regions, MultipleResults,
 };
 use crate::ddl::DdlContext;
 use crate::error::{AbortProcedureSnafu, NoLeaderSnafu, PutPoisonSnafu, Result, RetryLaterSnafu};
-use crate::instruction::CacheIdent;
 use crate::key::table_info::TableInfoValue;
 use crate::key::{DeserializedValueWithBytes, RegionDistribution};
 use crate::lock_key::{CatalogLock, SchemaLock, TableLock, TableNameLock};
 use crate::metrics;
 use crate::poison_key::table_poison_key;
 use crate::rpc::ddl::AlterTableTask;
-use crate::rpc::router::{find_leader_regions, find_leaders, region_distribution, RegionRoute};
+use crate::rpc::router::{find_leaders, region_distribution, RegionRoute};
 
 /// The alter table procedure
 pub struct AlterTableProcedure {
@@ -67,6 +62,24 @@ pub struct AlterTableProcedure {
     /// If we recover the procedure from json, then the table info value is not cached.
     /// But we already validated it in the prepare step.
     new_table_info: Option<TableInfo>,
+    /// The alter table executor.
+    executor: AlterTableExecutor,
+}
+
+/// Builds the executor from the [`AlterTableData`].
+///
+/// # Panics
+/// - If the alter kind is not set.
+fn build_executor_from_alter_expr(alter_data: &AlterTableData) -> AlterTableExecutor {
+    let table_name = alter_data.table_ref().into();
+    let table_id = alter_data.table_id;
+    let alter_kind = alter_data.task.alter_table.kind.as_ref().unwrap();
+    let new_table_name = if let Kind::RenameTable(RenameTable { new_table_name }) = alter_kind {
+        Some(new_table_name.to_string())
+    } else {
+        None
+    };
+    AlterTableExecutor::new(table_name, table_id, new_table_name)
 }
 
 impl AlterTableProcedure {
@@ -74,33 +87,42 @@ impl AlterTableProcedure {
 
     pub fn new(table_id: TableId, task: AlterTableTask, context: DdlContext) -> Result<Self> {
         task.validate()?;
+        let data = AlterTableData::new(task, table_id);
+        let executor = build_executor_from_alter_expr(&data);
         Ok(Self {
             context,
-            data: AlterTableData::new(task, table_id),
+            data,
             new_table_info: None,
+            executor,
         })
     }
 
     pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult<Self> {
         let data: AlterTableData = serde_json::from_str(json).context(FromJsonSnafu)?;
+        let executor = build_executor_from_alter_expr(&data);
+
         Ok(AlterTableProcedure {
             context,
             data,
             new_table_info: None,
+            executor,
         })
     }
 
     // Checks whether the table exists.
     pub(crate) async fn on_prepare(&mut self) -> Result<Status> {
-        self.check_alter().await?;
+        self.executor
+            .on_prepare(&self.context.table_metadata_manager)
+            .await?;
         self.fill_table_info().await?;
 
-        // Validates the request and builds the new table info.
-        // We need to build the new table info here because we should ensure the alteration
-        // is valid in `UpdateMeta` state as we already altered the region.
-        // Safety: `fill_table_info()` already set it.
+        // Safety: filled in `fill_table_info`.
         let table_info_value = self.data.table_info_value.as_ref().unwrap();
-        self.new_table_info = Some(self.build_new_table_info(&table_info_value.table_info)?);
+        let new_table_info = AlterTableExecutor::validate_alter_table_expr(
+            &table_info_value.table_info,
+            self.data.task.alter_table.clone(),
+        )?;
+        self.new_table_info = Some(new_table_info);
 
         // Safety: Checked in `AlterTableProcedure::new`.
         let alter_kind = self.data.task.alter_table.kind.as_ref().unwrap();
@@ -143,9 +165,7 @@ impl AlterTableProcedure {
 
         self.data.region_distribution =
             Some(region_distribution(&physical_table_route.region_routes));
-
         let leaders = find_leaders(&physical_table_route.region_routes);
-        let mut alter_region_tasks = Vec::with_capacity(leaders.len());
         let alter_kind = self.make_region_alter_kind()?;
 
         info!(
@@ -158,31 +178,14 @@ impl AlterTableProcedure {
         ensure!(!leaders.is_empty(), NoLeaderSnafu { table_id });
         // Puts the poison before submitting alter region requests to datanodes.
         self.put_poison(ctx_provider, procedure_id).await?;
-        for datanode in leaders {
-            let requester = self.context.node_manager.datanode(&datanode).await;
-            let regions = find_leader_regions(&physical_table_route.region_routes, &datanode);
-
-            for region in regions {
-                let region_id = RegionId::new(table_id, region);
-                let request = self.make_alter_region_request(region_id, alter_kind.clone())?;
-                debug!("Submitting {request:?} to {datanode}");
-
-                let datanode = datanode.clone();
-                let requester = requester.clone();
-
-                alter_region_tasks.push(async move {
-                    requester
-                        .handle(request)
-                        .await
-                        .map_err(add_peer_context_if_needed(datanode))
-                });
-            }
-        }
-
-        let results = future::join_all(alter_region_tasks)
-            .await
-            .into_iter()
-            .collect::<Vec<_>>();
+        let results = self
+            .executor
+            .on_alter_regions(
+                &self.context.node_manager,
+                &physical_table_route.region_routes,
+                alter_kind,
+            )
+            .await;
 
         match handle_multiple_results(results) {
             MultipleResults::PartialRetryable(error) => {
@@ -260,43 +263,34 @@ impl AlterTableProcedure {
     pub(crate) async fn on_update_metadata(&mut self) -> Result<Status> {
         let table_id = self.data.table_id();
         let table_ref = self.data.table_ref();
-        // Safety: checked before.
+        // Safety: filled in `fill_table_info`.
         let table_info_value = self.data.table_info_value.as_ref().unwrap();
+        // Safety: Checked in `AlterTableProcedure::new`.
+        let alter_kind = self.data.task.alter_table.kind.as_ref().unwrap();
+
         // Gets the table info from the cache or builds it.
-        let new_info = match &self.new_table_info {
+        let  new_info = match &self.new_table_info {
             Some(cached) => cached.clone(),
-            None => self.build_new_table_info(&table_info_value.table_info)
+            None => AlterTableExecutor::validate_alter_table_expr(
+                &table_info_value.table_info,
+                self.data.task.alter_table.clone(),
+               )
                 .inspect_err(|e| {
                     // We already check the table info in the prepare step so this should not happen.
                     error!(e; "Unable to build info for table {} in update metadata step, table_id: {}", table_ref, table_id);
                 })?,
         };
 
-        debug!(
-            "Starting update table: {} metadata, new table info {:?}",
-            table_ref.to_string(),
-            new_info
-        );
-
-        // Safety: Checked in `AlterTableProcedure::new`.
-        let alter_kind = self.data.task.alter_table.kind.as_ref().unwrap();
-        if let Kind::RenameTable(RenameTable { new_table_name }) = alter_kind {
-            self.on_update_metadata_for_rename(new_table_name.to_string(), table_info_value)
-                .await?;
-        } else {
-            let mut raw_table_info = new_info.into();
-            if !self.data.column_metadatas.is_empty() {
-                update_table_info_column_ids(&mut raw_table_info, &self.data.column_metadatas);
-            }
-            // region distribution is set in submit_alter_region_requests
-            let region_distribution = self.data.region_distribution.as_ref().unwrap().clone();
-            self.on_update_metadata_for_alter(
-                raw_table_info,
-                region_distribution,
+        // Safety: region distribution is set in `submit_alter_region_requests`.
+        self.executor
+            .on_alter_metadata(
+                &self.context.table_metadata_manager,
                 table_info_value,
+                self.data.region_distribution.as_ref(),
+                new_info.into(),
+                &self.data.column_metadatas,
             )
             .await?;
-        }
 
         info!("Updated table metadata for table {table_ref}, table_id: {table_id}, kind: {alter_kind:?}");
         self.data.state = AlterTableState::InvalidateTableCache;
@@ -305,18 +299,9 @@ impl AlterTableProcedure {
 
     /// Broadcasts the invalidating table cache instructions.
     async fn on_broadcast(&mut self) -> Result<Status> {
-        let cache_invalidator = &self.context.cache_invalidator;
-
-        cache_invalidator
-            .invalidate(
-                &Context::default(),
-                &[
-                    CacheIdent::TableId(self.data.table_id()),
-                    CacheIdent::TableName(self.data.table_ref().into()),
-                ],
-            )
+        self.executor
+            .invalidate_table_cache(&self.context.cache_invalidator)
             .await?;
-
         Ok(Status::done())
     }
 

src/common/meta/src/ddl/alter_table/check.rs

@@ -1,62 +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 api::v1::alter_table_expr::Kind;
-use api::v1::RenameTable;
-use common_catalog::format_full_table_name;
-use snafu::ensure;
-
-use crate::ddl::alter_table::AlterTableProcedure;
-use crate::error::{self, Result};
-use crate::key::table_name::TableNameKey;
-
-impl AlterTableProcedure {
-    /// Checks:
-    /// - The new table name doesn't exist (rename).
-    /// - Table exists.
-    pub(crate) async fn check_alter(&self) -> Result<()> {
-        let alter_expr = &self.data.task.alter_table;
-        let catalog = &alter_expr.catalog_name;
-        let schema = &alter_expr.schema_name;
-        let table_name = &alter_expr.table_name;
-        // Safety: Checked in `AlterTableProcedure::new`.
-        let alter_kind = self.data.task.alter_table.kind.as_ref().unwrap();
-
-        let manager = &self.context.table_metadata_manager;
-        if let Kind::RenameTable(RenameTable { new_table_name }) = alter_kind {
-            let new_table_name_key = TableNameKey::new(catalog, schema, new_table_name);
-            let exists = manager
-                .table_name_manager()
-                .exists(new_table_name_key)
-                .await?;
-            ensure!(
-                !exists,
-                error::TableAlreadyExistsSnafu {
-                    table_name: format_full_table_name(catalog, schema, new_table_name),
-                }
-            )
-        }
-
-        let table_name_key = TableNameKey::new(catalog, schema, table_name);
-        let exists = manager.table_name_manager().exists(table_name_key).await?;
-        ensure!(
-            exists,
-            error::TableNotFoundSnafu {
-                table_name: format_full_table_name(catalog, schema, &alter_expr.table_name),
-            }
-        );
-
-        Ok(())
-    }
-}

src/common/meta/src/ddl/alter_table/executor.rs

@@ -0,0 +1,313 @@
+// 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::collections::HashMap;
+
+use api::region::RegionResponse;
+use api::v1::region::region_request::Body;
+use api::v1::region::{alter_request, AlterRequest, RegionRequest, RegionRequestHeader};
+use api::v1::AlterTableExpr;
+use common_catalog::format_full_table_name;
+use common_grpc_expr::alter_expr_to_request;
+use common_telemetry::tracing_context::TracingContext;
+use common_telemetry::{debug, info};
+use futures::future;
+use snafu::{ensure, ResultExt};
+use store_api::metadata::ColumnMetadata;
+use store_api::storage::{RegionId, TableId};
+use table::metadata::{RawTableInfo, TableInfo};
+use table::requests::AlterKind;
+use table::table_name::TableName;
+
+use crate::cache_invalidator::{CacheInvalidatorRef, Context};
+use crate::ddl::utils::{add_peer_context_if_needed, raw_table_info};
+use crate::error::{self, Result, UnexpectedSnafu};
+use crate::instruction::CacheIdent;
+use crate::key::table_info::TableInfoValue;
+use crate::key::table_name::TableNameKey;
+use crate::key::{DeserializedValueWithBytes, RegionDistribution, TableMetadataManagerRef};
+use crate::node_manager::NodeManagerRef;
+use crate::rpc::router::{find_leaders, region_distribution, RegionRoute};
+
+/// [AlterTableExecutor] performs:
+/// - Alters the metadata of the table.
+/// - Alters regions on the datanode nodes.
+pub struct AlterTableExecutor {
+    table: TableName,
+    table_id: TableId,
+    /// The new table name if the alter kind is rename table.
+    new_table_name: Option<String>,
+}
+
+impl AlterTableExecutor {
+    /// Creates a new [`AlterTableExecutor`].
+    pub fn new(table: TableName, table_id: TableId, new_table_name: Option<String>) -> Self {
+        Self {
+            table,
+            table_id,
+            new_table_name,
+        }
+    }
+
+    /// Prepares to alter the table.
+    ///
+    /// ## Checks:
+    /// - The new table name doesn't exist (rename).
+    /// - Table exists.
+    pub(crate) async fn on_prepare(
+        &self,
+        table_metadata_manager: &TableMetadataManagerRef,
+    ) -> Result<()> {
+        let catalog = &self.table.catalog_name;
+        let schema = &self.table.schema_name;
+        let table_name = &self.table.table_name;
+
+        let manager = table_metadata_manager;
+        if let Some(new_table_name) = &self.new_table_name {
+            let new_table_name_key = TableNameKey::new(catalog, schema, new_table_name);
+            let exists = manager
+                .table_name_manager()
+                .exists(new_table_name_key)
+                .await?;
+            ensure!(
+                !exists,
+                error::TableAlreadyExistsSnafu {
+                    table_name: format_full_table_name(catalog, schema, new_table_name),
+                }
+            )
+        }
+
+        let table_name_key = TableNameKey::new(catalog, schema, table_name);
+        let exists = manager.table_name_manager().exists(table_name_key).await?;
+        ensure!(
+            exists,
+            error::TableNotFoundSnafu {
+                table_name: format_full_table_name(catalog, schema, table_name),
+            }
+        );
+
+        Ok(())
+    }
+
+    /// Validates the alter table expression and builds the new table info.
+    ///
+    /// This validation is performed early to ensure the alteration is valid before
+    /// proceeding to the `on_alter_metadata` state, where regions have already been altered.
+    /// Building the new table info here allows us to catch any issues with the
+    /// alteration before committing metadata changes.
+    pub(crate) fn validate_alter_table_expr(
+        table_info: &RawTableInfo,
+        alter_table_expr: AlterTableExpr,
+    ) -> Result<TableInfo> {
+        build_new_table_info(table_info, alter_table_expr)
+    }
+
+    /// Updates table metadata for alter table operation.
+    pub(crate) async fn on_alter_metadata(
+        &self,
+        table_metadata_manager: &TableMetadataManagerRef,
+        current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
+        region_distribution: Option<&RegionDistribution>,
+        mut raw_table_info: RawTableInfo,
+        column_metadatas: &[ColumnMetadata],
+    ) -> Result<()> {
+        let table_ref = self.table.table_ref();
+        let table_id = self.table_id;
+
+        if let Some(new_table_name) = &self.new_table_name {
+            debug!(
+                "Starting update table: {} metadata, table_id: {}, new table info: {:?}, new table name: {}",
+                table_ref, table_id, raw_table_info, new_table_name
+            );
+
+            table_metadata_manager
+                .rename_table(current_table_info_value, new_table_name.to_string())
+                .await?;
+        } else {
+            debug!(
+                "Starting update table: {} metadata, table_id: {}, new table info: {:?}",
+                table_ref, table_id, raw_table_info
+            );
+
+            ensure!(
+                region_distribution.is_some(),
+                UnexpectedSnafu {
+                    err_msg: "region distribution is not set when updating table metadata",
+                }
+            );
+
+            if !column_metadatas.is_empty() {
+                raw_table_info::update_table_info_column_ids(&mut raw_table_info, column_metadatas);
+            }
+            table_metadata_manager
+                .update_table_info(
+                    current_table_info_value,
+                    region_distribution.cloned(),
+                    raw_table_info,
+                )
+                .await?;
+        }
+
+        Ok(())
+    }
+
+    /// Alters regions on the datanode nodes.
+    pub(crate) async fn on_alter_regions(
+        &self,
+        node_manager: &NodeManagerRef,
+        region_routes: &[RegionRoute],
+        kind: Option<alter_request::Kind>,
+    ) -> Vec<Result<RegionResponse>> {
+        let region_distribution = region_distribution(region_routes);
+        let leaders = find_leaders(region_routes)
+            .into_iter()
+            .map(|p| (p.id, p))
+            .collect::<HashMap<_, _>>();
+        let total_num_region = region_distribution
+            .values()
+            .map(|r| r.leader_regions.len())
+            .sum::<usize>();
+        let mut alter_region_tasks = Vec::with_capacity(total_num_region);
+        for (datanode_id, region_role_set) in region_distribution {
+            if region_role_set.leader_regions.is_empty() {
+                continue;
+            }
+            // Safety: must exists.
+            let peer = leaders.get(&datanode_id).unwrap();
+            let requester = node_manager.datanode(peer).await;
+
+            for region_id in region_role_set.leader_regions {
+                let region_id = RegionId::new(self.table_id, region_id);
+                let request = make_alter_region_request(region_id, kind.clone());
+
+                let requester = requester.clone();
+                let peer = peer.clone();
+
+                alter_region_tasks.push(async move {
+                    requester
+                        .handle(request)
+                        .await
+                        .map_err(add_peer_context_if_needed(peer))
+                });
+            }
+        }
+
+        future::join_all(alter_region_tasks)
+            .await
+            .into_iter()
+            .collect::<Vec<_>>()
+    }
+
+    /// Invalidates cache for the table.
+    pub(crate) async fn invalidate_table_cache(
+        &self,
+        cache_invalidator: &CacheInvalidatorRef,
+    ) -> Result<()> {
+        let ctx = Context {
+            subject: Some(format!(
+                "Invalidate table cache by altering table {}, table_id: {}",
+                self.table.table_ref(),
+                self.table_id,
+            )),
+        };
+
+        cache_invalidator
+            .invalidate(
+                &ctx,
+                &[
+                    CacheIdent::TableName(self.table.clone()),
+                    CacheIdent::TableId(self.table_id),
+                ],
+            )
+            .await?;
+
+        Ok(())
+    }
+}
+
+/// Makes alter region request.
+pub(crate) fn make_alter_region_request(
+    region_id: RegionId,
+    kind: Option<alter_request::Kind>,
+) -> RegionRequest {
+    RegionRequest {
+        header: Some(RegionRequestHeader {
+            tracing_context: TracingContext::from_current_span().to_w3c(),
+            ..Default::default()
+        }),
+        body: Some(Body::Alter(AlterRequest {
+            region_id: region_id.as_u64(),
+            kind,
+            ..Default::default()
+        })),
+    }
+}
+
+/// Builds new table info after alteration.
+///
+/// This function creates a new table info by applying the alter table expression
+/// to the existing table info. For add column operations, it increments the
+/// `next_column_id` by the number of columns being added, which may result in gaps
+/// in the column id sequence.
+fn build_new_table_info(
+    table_info: &RawTableInfo,
+    alter_table_expr: AlterTableExpr,
+) -> Result<TableInfo> {
+    let table_info =
+        TableInfo::try_from(table_info.clone()).context(error::ConvertRawTableInfoSnafu)?;
+    let schema_name = &table_info.schema_name;
+    let catalog_name = &table_info.catalog_name;
+    let table_name = &table_info.name;
+    let table_id = table_info.ident.table_id;
+    let request = alter_expr_to_request(table_id, alter_table_expr)
+        .context(error::ConvertAlterTableRequestSnafu)?;
+
+    let new_meta = table_info
+        .meta
+        .builder_with_alter_kind(table_name, &request.alter_kind)
+        .context(error::TableSnafu)?
+        .build()
+        .with_context(|_| error::BuildTableMetaSnafu {
+            table_name: format_full_table_name(catalog_name, schema_name, table_name),
+        })?;
+
+    let mut new_info = table_info.clone();
+    new_info.meta = new_meta;
+    new_info.ident.version = table_info.ident.version + 1;
+    match request.alter_kind {
+        AlterKind::AddColumns { columns } => {
+            // Bumps the column id for the new columns.
+            // It may bump more than the actual number of columns added if there are
+            // existing columns, but it's fine.
+            new_info.meta.next_column_id += columns.len() as u32;
+        }
+        AlterKind::RenameTable { new_table_name } => {
+            new_info.name = new_table_name.to_string();
+        }
+        AlterKind::DropColumns { .. }
+        | AlterKind::ModifyColumnTypes { .. }
+        | AlterKind::SetTableOptions { .. }
+        | AlterKind::UnsetTableOptions { .. }
+        | AlterKind::SetIndex { .. }
+        | AlterKind::UnsetIndex { .. }
+        | AlterKind::DropDefaults { .. } => {}
+    }
+
+    info!(
+        "Built new table info: {:?} for table {}, table_id: {}",
+        new_info.meta, table_name, table_id
+    );
+
+    Ok(new_info)
+}

src/common/meta/src/ddl/alter_table/region_request.rs

@@ -15,43 +15,16 @@
 use std::collections::HashSet;
 
 use api::v1::alter_table_expr::Kind;
-use api::v1::region::region_request::Body;
 use api::v1::region::{
-    alter_request, AddColumn, AddColumns, AlterRequest, DropColumn, DropColumns, RegionColumnDef,
-    RegionRequest, RegionRequestHeader,
+    alter_request, AddColumn, AddColumns, DropColumn, DropColumns, RegionColumnDef,
 };
-use common_telemetry::tracing_context::TracingContext;
 use snafu::OptionExt;
-use store_api::storage::RegionId;
 use table::metadata::RawTableInfo;
 
 use crate::ddl::alter_table::AlterTableProcedure;
 use crate::error::{InvalidProtoMsgSnafu, Result};
 
 impl AlterTableProcedure {
-    /// Makes alter region request from existing an alter kind.
-    /// Region alter request always add columns if not exist.
-    pub(crate) fn make_alter_region_request(
-        &self,
-        region_id: RegionId,
-        kind: Option<alter_request::Kind>,
-    ) -> Result<RegionRequest> {
-        // Safety: checked
-        let table_info = self.data.table_info().unwrap();
-
-        Ok(RegionRequest {
-            header: Some(RegionRequestHeader {
-                tracing_context: TracingContext::from_current_span().to_w3c(),
-                ..Default::default()
-            }),
-            body: Some(Body::Alter(AlterRequest {
-                region_id: region_id.as_u64(),
-                schema_version: table_info.ident.version,
-                kind,
-            })),
-        })
-    }
-
     /// Makes alter kind proto that all regions can reuse.
     /// Region alter request always add columns if not exist.
     pub(crate) fn make_region_alter_kind(&self) -> Result<Option<alter_request::Kind>> {
@@ -155,6 +128,7 @@ mod tests {
     use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
     use store_api::storage::{RegionId, TableId};
 
+    use crate::ddl::alter_table::executor::make_alter_region_request;
     use crate::ddl::alter_table::AlterTableProcedure;
     use crate::ddl::test_util::columns::TestColumnDefBuilder;
     use crate::ddl::test_util::create_table::{
@@ -261,15 +235,13 @@ mod tests {
         let mut procedure = AlterTableProcedure::new(table_id, task, ddl_context).unwrap();
         procedure.on_prepare().await.unwrap();
         let alter_kind = procedure.make_region_alter_kind().unwrap();
-        let Some(Body::Alter(alter_region_request)) = procedure
-            .make_alter_region_request(region_id, alter_kind)
-            .unwrap()
-            .body
+        let Some(Body::Alter(alter_region_request)) =
+            make_alter_region_request(region_id, alter_kind).body
         else {
             unreachable!()
         };
         assert_eq!(alter_region_request.region_id, region_id.as_u64());
-        assert_eq!(alter_region_request.schema_version, 1);
+        assert_eq!(alter_region_request.schema_version, 0);
         assert_eq!(
             alter_region_request.kind,
             Some(region::alter_request::Kind::AddColumns(
@@ -319,15 +291,13 @@ mod tests {
         let mut procedure = AlterTableProcedure::new(table_id, task, ddl_context).unwrap();
         procedure.on_prepare().await.unwrap();
         let alter_kind = procedure.make_region_alter_kind().unwrap();
-        let Some(Body::Alter(alter_region_request)) = procedure
-            .make_alter_region_request(region_id, alter_kind)
-            .unwrap()
-            .body
+        let Some(Body::Alter(alter_region_request)) =
+            make_alter_region_request(region_id, alter_kind).body
         else {
             unreachable!()
         };
         assert_eq!(alter_region_request.region_id, region_id.as_u64());
-        assert_eq!(alter_region_request.schema_version, 1);
+        assert_eq!(alter_region_request.schema_version, 0);
         assert_eq!(
             alter_region_request.kind,
             Some(region::alter_request::Kind::ModifyColumnTypes(

src/common/meta/src/ddl/alter_table/update_metadata.rs

@@ -1,103 +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_grpc_expr::alter_expr_to_request;
-use snafu::ResultExt;
-use table::metadata::{RawTableInfo, TableInfo};
-use table::requests::AlterKind;
-
-use crate::ddl::alter_table::AlterTableProcedure;
-use crate::error::{self, Result};
-use crate::key::table_info::TableInfoValue;
-use crate::key::{DeserializedValueWithBytes, RegionDistribution};
-
-impl AlterTableProcedure {
-    /// Builds new table info after alteration.
-    /// It bumps the column id of the table by the number of the add column requests.
-    /// So there may be holes in the column id sequence.
-    pub(crate) fn build_new_table_info(&self, table_info: &RawTableInfo) -> Result<TableInfo> {
-        let table_info =
-            TableInfo::try_from(table_info.clone()).context(error::ConvertRawTableInfoSnafu)?;
-        let table_ref = self.data.table_ref();
-        let alter_expr = self.data.task.alter_table.clone();
-        let request = alter_expr_to_request(self.data.table_id(), alter_expr)
-            .context(error::ConvertAlterTableRequestSnafu)?;
-
-        let new_meta = table_info
-            .meta
-            .builder_with_alter_kind(table_ref.table, &request.alter_kind)
-            .context(error::TableSnafu)?
-            .build()
-            .with_context(|_| error::BuildTableMetaSnafu {
-                table_name: table_ref.table,
-            })?;
-
-        let mut new_info = table_info.clone();
-        new_info.meta = new_meta;
-        new_info.ident.version = table_info.ident.version + 1;
-        match request.alter_kind {
-            AlterKind::AddColumns { columns } => {
-                // Bumps the column id for the new columns.
-                // It may bump more than the actual number of columns added if there are
-                // existing columns, but it's fine.
-                new_info.meta.next_column_id += columns.len() as u32;
-            }
-            AlterKind::RenameTable { new_table_name } => {
-                new_info.name = new_table_name.to_string();
-            }
-            AlterKind::DropColumns { .. }
-            | AlterKind::ModifyColumnTypes { .. }
-            | AlterKind::SetTableOptions { .. }
-            | AlterKind::UnsetTableOptions { .. }
-            | AlterKind::SetIndex { .. }
-            | AlterKind::UnsetIndex { .. }
-            | AlterKind::DropDefaults { .. } => {}
-        }
-
-        Ok(new_info)
-    }
-
-    /// Updates table metadata for rename table operation.
-    pub(crate) async fn on_update_metadata_for_rename(
-        &self,
-        new_table_name: String,
-        current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
-    ) -> Result<()> {
-        let table_metadata_manager = &self.context.table_metadata_manager;
-        table_metadata_manager
-            .rename_table(current_table_info_value, new_table_name)
-            .await?;
-
-        Ok(())
-    }
-
-    /// Updates table metadata for alter table operation.
-    pub(crate) async fn on_update_metadata_for_alter(
-        &self,
-        new_table_info: RawTableInfo,
-        region_distribution: RegionDistribution,
-        current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
-    ) -> Result<()> {
-        let table_metadata_manager = &self.context.table_metadata_manager;
-        table_metadata_manager
-            .update_table_info(
-                current_table_info_value,
-                Some(region_distribution),
-                new_table_info,
-            )
-            .await?;
-
-        Ok(())
-    }
-}

src/common/meta/src/ddl/create_flow/metadata.rs

@@ -21,7 +21,7 @@ use crate::key::table_name::TableNameKey;
 impl CreateFlowProcedure {
     /// Allocates the [FlowId].
     pub(crate) async fn allocate_flow_id(&mut self) -> Result<()> {
-        //TODO(weny, ruihang): We doesn't support the partitions. It's always be 1, now.
+        // TODO(weny, ruihang): We don't support the partitions. It's always be 1, now.
         let partitions = 1;
         let (flow_id, peers) = self
             .context

src/common/meta/src/ddl/create_logical_tables/update_metadata.rs

@@ -22,7 +22,7 @@ use table::table_name::TableName;
 
 use crate::cache_invalidator::Context;
 use crate::ddl::create_logical_tables::CreateLogicalTablesProcedure;
-use crate::ddl::physical_table_metadata;
+use crate::ddl::utils::raw_table_info;
 use crate::error::{Result, TableInfoNotFoundSnafu};
 use crate::instruction::CacheIdent;
 
@@ -47,7 +47,7 @@ impl CreateLogicalTablesProcedure {
         // Generates new table info
         let raw_table_info = physical_table_info.deref().table_info.clone();
 
-        let new_table_info = physical_table_metadata::build_new_physical_table_info(
+        let new_table_info = raw_table_info::build_new_physical_table_info(
             raw_table_info,
             &self.data.physical_columns,
         );

src/common/meta/src/ddl/create_table.rs

@@ -35,7 +35,7 @@ use table::metadata::{RawTableInfo, TableId};
 use table::table_reference::TableReference;
 
 use crate::ddl::create_table_template::{build_template, CreateRequestBuilder};
-use crate::ddl::physical_table_metadata::update_table_info_column_ids;
+use crate::ddl::utils::raw_table_info::update_table_info_column_ids;
 use crate::ddl::utils::{
     add_peer_context_if_needed, convert_region_routes_to_detecting_regions,
     extract_column_metadatas, map_to_procedure_error, region_storage_path,

src/common/meta/src/ddl/drop_table/executor.rs

@@ -185,11 +185,15 @@ impl DropTableExecutor {
             .await
     }
 
-    /// Invalidates frontend caches
+    /// Invalidates caches for the table.
     pub async fn invalidate_table_cache(&self, ctx: &DdlContext) -> Result<()> {
         let cache_invalidator = &ctx.cache_invalidator;
         let ctx = Context {
-            subject: Some("Invalidate table cache by dropping table".to_string()),
+            subject: Some(format!(
+                "Invalidate table cache by dropping table {}, table_id: {}",
+                self.table.table_ref(),
+                self.table_id,
+            )),
         };
 
         cache_invalidator

src/common/meta/src/ddl/table_meta.rs

@@ -113,15 +113,19 @@ impl TableMetadataAllocator {
         table_id: TableId,
         task: &CreateTableTask,
     ) -> Result<PhysicalTableRouteValue> {
-        let regions = task.partitions.len();
+        let num_regions = task
+            .partitions
+            .as_ref()
+            .map(|p| p.value_list.len())
+            .unwrap_or(1);
         ensure!(
-            regions > 0,
+            num_regions > 0,
             error::UnexpectedSnafu {
                 err_msg: "The number of partitions must be greater than 0"
             }
         );
 
-        let peers = self.peer_allocator.alloc(regions).await?;
+        let peers = self.peer_allocator.alloc(num_regions).await?;
         debug!("Allocated peers {:?} for table {}", peers, table_id);
         let region_routes = task
             .partitions

src/common/meta/src/ddl/test_util.rs

@@ -21,7 +21,6 @@ pub mod flownode_handler;
 use std::assert_matches::assert_matches;
 use std::collections::HashMap;
 
-use api::v1::meta::Partition;
 use api::v1::{ColumnDataType, SemanticType};
 use common_procedure::Status;
 use datatypes::prelude::ConcreteDataType;
@@ -145,10 +144,7 @@ pub fn test_create_logical_table_task(name: &str) -> CreateTableTask {
     CreateTableTask {
         create_table,
         // Single region
-        partitions: vec![Partition {
-            column_list: vec![],
-            value_list: vec![],
-        }],
+        partitions: None,
         table_info,
     }
 }
@@ -183,10 +179,7 @@ pub fn test_create_physical_table_task(name: &str) -> CreateTableTask {
     CreateTableTask {
         create_table,
         // Single region
-        partitions: vec![Partition {
-            column_list: vec![],
-            value_list: vec![],
-        }],
+        partitions: None,
         table_info,
     }
 }

src/common/meta/src/ddl/test_util/create_table.rs

@@ -15,7 +15,6 @@
 use std::collections::HashMap;
 
 use api::v1::column_def::try_as_column_schema;
-use api::v1::meta::Partition;
 use api::v1::{ColumnDataType, ColumnDef, CreateTableExpr, SemanticType};
 use chrono::DateTime;
 use common_catalog::consts::{
@@ -175,10 +174,7 @@ pub fn test_create_table_task(name: &str, table_id: TableId) -> CreateTableTask
     CreateTableTask {
         create_table,
         // Single region
-        partitions: vec![Partition {
-            column_list: vec![],
-            value_list: vec![],
-        }],
+        partitions: None,
         table_info,
     }
 }

src/common/meta/src/ddl/tests/create_table.rs

@@ -17,7 +17,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use api::region::RegionResponse;
-use api::v1::meta::{Partition, Peer};
+use api::v1::meta::Peer;
 use api::v1::region::{region_request, RegionRequest};
 use api::v1::{ColumnDataType, SemanticType};
 use common_error::ext::ErrorExt;
@@ -141,10 +141,7 @@ pub(crate) fn test_create_table_task(name: &str) -> CreateTableTask {
     CreateTableTask {
         create_table,
         // Single region
-        partitions: vec![Partition {
-            column_list: vec![],
-            value_list: vec![],
-        }],
+        partitions: None,
         table_info,
     }
 }
@@ -218,7 +215,7 @@ async fn test_on_prepare_with_no_partition_err() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
     let mut task = test_create_table_task("foo");
-    task.partitions = vec![];
+    task.partitions = None;
     task.create_table.create_if_not_exists = true;
     let mut procedure = CreateTableProcedure::new(task, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();

src/common/meta/src/ddl/utils.rs

@@ -12,6 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub(crate) mod raw_table_info;
+pub(crate) mod table_id;
+pub(crate) mod table_info;
+
 use std::collections::HashMap;
 use std::fmt::Debug;
 

src/common/meta/src/ddl/utils/table_id.rs

@@ -0,0 +1,46 @@
+// 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 snafu::OptionExt;
+use store_api::storage::TableId;
+use table::table_reference::TableReference;
+
+use crate::error::{Result, TableNotFoundSnafu};
+use crate::key::table_name::{TableNameKey, TableNameManager};
+
+/// Get all the table ids from the table names.
+///
+/// Returns an error if any table does not exist.
+pub(crate) async fn get_all_table_ids_by_names<'a>(
+    table_name_manager: &TableNameManager,
+    table_names: &[TableReference<'a>],
+) -> Result<Vec<TableId>> {
+    let table_name_keys = table_names
+        .iter()
+        .map(TableNameKey::from)
+        .collect::<Vec<_>>();
+    let table_name_values = table_name_manager.batch_get(table_name_keys).await?;
+    let mut table_ids = Vec::with_capacity(table_name_values.len());
+    for (value, table_name) in table_name_values.into_iter().zip(table_names) {
+        let value = value
+            .with_context(|| TableNotFoundSnafu {
+                table_name: table_name.to_string(),
+            })?
+            .table_id();
+
+        table_ids.push(value);
+    }
+
+    Ok(table_ids)
+}

src/common/meta/src/ddl/utils/table_info.rs

@@ -0,0 +1,44 @@
+// 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 snafu::OptionExt;
+use store_api::storage::TableId;
+use table::table_reference::TableReference;
+
+use crate::error::{Result, TableInfoNotFoundSnafu};
+use crate::key::table_info::{TableInfoManager, TableInfoValue};
+use crate::key::DeserializedValueWithBytes;
+
+/// Get all table info values by table ids.
+///
+/// Returns an error if any table does not exist.
+pub(crate) async fn get_all_table_info_values_by_table_ids<'a>(
+    table_info_manager: &TableInfoManager,
+    table_ids: &[TableId],
+    table_names: &[TableReference<'a>],
+) -> Result<Vec<DeserializedValueWithBytes<TableInfoValue>>> {
+    let mut table_info_map = table_info_manager.batch_get_raw(table_ids).await?;
+    let mut table_info_values = Vec::with_capacity(table_ids.len());
+    for (table_id, table_name) in table_ids.iter().zip(table_names) {
+        let table_info_value =
+            table_info_map
+                .remove(table_id)
+                .with_context(|| TableInfoNotFoundSnafu {
+                    table: table_name.to_string(),
+                })?;
+        table_info_values.push(table_info_value);
+    }
+
+    Ok(table_info_values)
+}

src/common/meta/src/key/table_name.rs

@@ -103,6 +103,26 @@ pub fn table_decoder(kv: KeyValue) -> Result<(String, TableNameValue)> {
     Ok((table_name_key.table.to_string(), table_name_value))
 }
 
+impl<'a> From<&TableReference<'a>> for TableNameKey<'a> {
+    fn from(value: &TableReference<'a>) -> Self {
+        Self {
+            catalog: value.catalog,
+            schema: value.schema,
+            table: value.table,
+        }
+    }
+}
+
+impl<'a> From<TableReference<'a>> for TableNameKey<'a> {
+    fn from(value: TableReference<'a>) -> Self {
+        Self {
+            catalog: value.catalog,
+            schema: value.schema,
+            table: value.table,
+        }
+    }
+}
+
 impl<'a> From<&'a TableName> for TableNameKey<'a> {
     fn from(value: &'a TableName) -> Self {
         Self {

src/common/meta/src/peer.rs

@@ -19,11 +19,17 @@ pub use api::v1::meta::Peer;
 use crate::error::Error;
 use crate::{DatanodeId, FlownodeId};
 
-/// can query peer given a node id
+/// PeerLookupService is a service that can lookup peers.
 #[async_trait::async_trait]
 pub trait PeerLookupService {
+    /// Returns the datanode with the given id. It may return inactive peers.
     async fn datanode(&self, id: DatanodeId) -> Result<Option<Peer>, Error>;
+
+    /// Returns the flownode with the given id. It may return inactive peers.
     async fn flownode(&self, id: FlownodeId) -> Result<Option<Peer>, Error>;
+
+    /// Returns all currently active frontend nodes that have reported a heartbeat within the most recent heartbeat interval from the in-memory backend.
+    async fn active_frontends(&self) -> Result<Vec<Peer>, Error>;
 }
 
 pub type PeerLookupServiceRef = Arc<dyn PeerLookupService + Send + Sync>;

src/common/meta/src/rpc/ddl.rs

@@ -96,7 +96,7 @@ impl DdlTask {
     /// Creates a [`DdlTask`] to create a table.
     pub fn new_create_table(
         expr: CreateTableExpr,
-        partitions: Vec<Partition>,
+        partitions: Option<Partition>,
         table_info: RawTableInfo,
     ) -> Self {
         DdlTask::CreateTable(CreateTableTask::new(expr, partitions, table_info))
@@ -107,7 +107,7 @@ impl DdlTask {
         DdlTask::CreateLogicalTables(
             table_data
                 .into_iter()
-                .map(|(expr, table_info)| CreateTableTask::new(expr, Vec::new(), table_info))
+                .map(|(expr, table_info)| CreateTableTask::new(expr, None, table_info))
                 .collect(),
         )
     }
@@ -606,7 +606,10 @@ impl From<DropTableTask> for PbDropTableTask {
 #[derive(Debug, PartialEq, Clone)]
 pub struct CreateTableTask {
     pub create_table: CreateTableExpr,
-    pub partitions: Vec<Partition>,
+    /// The partitions of the table.
+    ///
+    /// If the table is created with a single region (not partitioned), this field is `None`.
+    pub partitions: Option<Partition>,
     pub table_info: RawTableInfo,
 }
 
@@ -620,7 +623,7 @@ impl TryFrom<PbCreateTableTask> for CreateTableTask {
             pb.create_table.context(error::InvalidProtoMsgSnafu {
                 err_msg: "expected create table",
             })?,
-            pb.partitions,
+            pb.partitions.first().cloned(),
             table_info,
         ))
     }
@@ -633,7 +636,10 @@ impl TryFrom<CreateTableTask> for PbCreateTableTask {
         Ok(PbCreateTableTask {
             table_info: serde_json::to_vec(&task.table_info).context(error::SerdeJsonSnafu)?,
             create_table: Some(task.create_table),
-            partitions: task.partitions,
+            partitions: match task.partitions {
+                Some(p) => vec![p],
+                None => vec![],
+            },
         })
     }
 }
@@ -641,7 +647,7 @@ impl TryFrom<CreateTableTask> for PbCreateTableTask {
 impl CreateTableTask {
     pub fn new(
         expr: CreateTableExpr,
-        partitions: Vec<Partition>,
+        partitions: Option<Partition>,
         table_info: RawTableInfo,
     ) -> CreateTableTask {
         CreateTableTask {
@@ -701,7 +707,10 @@ impl Serialize for CreateTableTask {
 
         let pb = PbCreateTableTask {
             create_table: Some(self.create_table.clone()),
-            partitions: self.partitions.clone(),
+            partitions: match &self.partitions {
+                Some(p) => vec![p.clone()],
+                None => vec![],
+            },
             table_info,
         };
         let buf = pb.encode_to_vec();
@@ -1315,7 +1324,7 @@ mod tests {
         let table_info = test_table_info(1025, "foo", "bar", "baz", Arc::new(schema));
         let task = CreateTableTask::new(
             CreateTableExpr::default(),
-            Vec::new(),
+            None,
             RawTableInfo::from(table_info),
         );
 
@@ -1411,8 +1420,7 @@ mod tests {
             ..Default::default()
         };
 
-        let mut create_table_task =
-            CreateTableTask::new(create_table_expr, Vec::new(), raw_table_info);
+        let mut create_table_task = CreateTableTask::new(create_table_expr, None, raw_table_info);
 
         // Call the sort_columns method
         create_table_task.sort_columns();

src/common/meta/src/rpc/router.rs

@@ -391,6 +391,9 @@ impl From<Region> for PbRegion {
     }
 }
 
+/// Serialized version of `PartitionDef`.
+///
+/// Represent the entire partition part of one table
 #[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
 pub struct Partition {
     #[serde(serialize_with = "as_utf8_vec", deserialize_with = "from_utf8_vec")]

src/common/meta/src/test_util.rs

@@ -213,6 +213,10 @@ impl PeerLookupService for NoopPeerLookupService {
     async fn flownode(&self, id: FlownodeId) -> Result<Option<Peer>> {
         Ok(Some(Peer::empty(id)))
     }
+
+    async fn active_frontends(&self) -> Result<Vec<Peer>> {
+        Ok(vec![])
+    }
 }
 
 /// Create a kafka topic pool for testing.

src/common/sql/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "common-sql"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[dependencies]
+common-base.workspace = true
+common-datasource.workspace = true
+common-decimal.workspace = true
+common-error.workspace = true
+common-macro.workspace = true
+common-time.workspace = true
+datafusion-sql.workspace = true
+datatypes.workspace = true
+hex = "0.4"
+jsonb.workspace = true
+snafu.workspace = true
+sqlparser.workspace = true
+
+[lints]
+workspace = true

src/common/sql/src/default_constraint.rs

@@ -0,0 +1,182 @@
+// 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_time::timezone::Timezone;
+use datatypes::prelude::ConcreteDataType;
+use datatypes::schema::constraint::{CURRENT_TIMESTAMP, CURRENT_TIMESTAMP_FN};
+use datatypes::schema::ColumnDefaultConstraint;
+pub use sqlparser::ast::{
+    visit_expressions_mut, visit_statements_mut, BinaryOperator, ColumnDef, ColumnOption,
+    ColumnOptionDef, DataType, Expr, Function, FunctionArg, FunctionArgExpr, FunctionArguments,
+    Ident, ObjectName, SqlOption, TableConstraint, TimezoneInfo, UnaryOperator, Value as SqlValue,
+    Visit, VisitMut, Visitor, VisitorMut,
+};
+
+use crate::convert::{sql_number_to_value, sql_value_to_value};
+use crate::error::{Result, UnsupportedDefaultValueSnafu};
+
+pub fn parse_column_default_constraint(
+    column_name: &str,
+    data_type: &ConcreteDataType,
+    opts: &[ColumnOptionDef],
+    timezone: Option<&Timezone>,
+) -> Result<Option<ColumnDefaultConstraint>> {
+    if let Some(opt) = opts
+        .iter()
+        .find(|o| matches!(o.option, ColumnOption::Default(_)))
+    {
+        let default_constraint = match &opt.option {
+            ColumnOption::Default(Expr::Value(v)) => ColumnDefaultConstraint::Value(
+                sql_value_to_value(column_name, data_type, v, timezone, None, false)?,
+            ),
+            ColumnOption::Default(Expr::Function(func)) => {
+                let mut func = format!("{func}").to_lowercase();
+                // normalize CURRENT_TIMESTAMP to CURRENT_TIMESTAMP()
+                if func == CURRENT_TIMESTAMP {
+                    func = CURRENT_TIMESTAMP_FN.to_string();
+                }
+                // Always use lowercase for function expression
+                ColumnDefaultConstraint::Function(func.to_lowercase())
+            }
+
+            ColumnOption::Default(Expr::UnaryOp { op, expr }) => {
+                // Specialized process for handling numerical inputs to prevent
+                // overflow errors during the parsing of negative numbers,
+                // See https://github.com/GreptimeTeam/greptimedb/issues/4351
+                if let (UnaryOperator::Minus, Expr::Value(SqlValue::Number(n, _))) =
+                    (op, expr.as_ref())
+                {
+                    return Ok(Some(ColumnDefaultConstraint::Value(sql_number_to_value(
+                        data_type,
+                        &format!("-{n}"),
+                    )?)));
+                }
+
+                if let Expr::Value(v) = &**expr {
+                    let value =
+                        sql_value_to_value(column_name, data_type, v, timezone, Some(*op), false)?;
+                    ColumnDefaultConstraint::Value(value)
+                } else {
+                    return UnsupportedDefaultValueSnafu {
+                        column_name,
+                        expr: *expr.clone(),
+                    }
+                    .fail();
+                }
+            }
+            ColumnOption::Default(others) => {
+                return UnsupportedDefaultValueSnafu {
+                    column_name,
+                    expr: others.clone(),
+                }
+                .fail();
+            }
+            _ => {
+                return UnsupportedDefaultValueSnafu {
+                    column_name,
+                    expr: Expr::Value(SqlValue::Null),
+                }
+                .fail();
+            }
+        };
+
+        Ok(Some(default_constraint))
+    } else {
+        Ok(None)
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use std::assert_matches::assert_matches;
+
+    use datatypes::prelude::{ConcreteDataType, Value};
+    use datatypes::types::BooleanType;
+
+    use super::*;
+
+    #[test]
+    pub fn test_parse_column_default_constraint() {
+        let bool_value = sqlparser::ast::Value::Boolean(true);
+
+        let opts = vec![
+            ColumnOptionDef {
+                name: None,
+                option: ColumnOption::Default(Expr::Value(bool_value)),
+            },
+            ColumnOptionDef {
+                name: None,
+                option: ColumnOption::NotNull,
+            },
+        ];
+
+        let constraint = parse_column_default_constraint(
+            "coll",
+            &ConcreteDataType::Boolean(BooleanType),
+            &opts,
+            None,
+        )
+        .unwrap();
+
+        assert_matches!(
+            constraint,
+            Some(ColumnDefaultConstraint::Value(Value::Boolean(true)))
+        );
+
+        // Test negative number
+        let opts = vec![ColumnOptionDef {
+            name: None,
+            option: ColumnOption::Default(Expr::UnaryOp {
+                op: UnaryOperator::Minus,
+                expr: Box::new(Expr::Value(SqlValue::Number("32768".to_string(), false))),
+            }),
+        }];
+
+        let constraint = parse_column_default_constraint(
+            "coll",
+            &ConcreteDataType::int16_datatype(),
+            &opts,
+            None,
+        )
+        .unwrap();
+
+        assert_matches!(
+            constraint,
+            Some(ColumnDefaultConstraint::Value(Value::Int16(-32768)))
+        );
+    }
+
+    #[test]
+    fn test_incorrect_default_value_issue_3479() {
+        let opts = vec![ColumnOptionDef {
+            name: None,
+            option: ColumnOption::Default(Expr::Value(SqlValue::Number(
+                "0.047318541668048164".into(),
+                false,
+            ))),
+        }];
+        let constraint = parse_column_default_constraint(
+            "coll",
+            &ConcreteDataType::float64_datatype(),
+            &opts,
+            None,
+        )
+        .unwrap()
+        .unwrap();
+        assert_eq!("0.047318541668048164", constraint.to_string());
+        let encoded: Vec<u8> = constraint.clone().try_into().unwrap();
+        let decoded = ColumnDefaultConstraint::try_from(encoded.as_ref()).unwrap();
+        assert_eq!(decoded, constraint);
+    }
+}

src/common/sql/src/error.rs

@@ -0,0 +1,158 @@
+// 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_error::ext::ErrorExt;
+use common_error::status_code::StatusCode;
+use common_macro::stack_trace_debug;
+use common_time::timestamp::TimeUnit;
+use common_time::Timestamp;
+use datafusion_sql::sqlparser::ast::UnaryOperator;
+use datatypes::prelude::{ConcreteDataType, Value};
+use snafu::{Location, Snafu};
+pub use sqlparser::ast::{Expr, Value as SqlValue};
+
+pub type Result<T> = std::result::Result<T, Error>;
+
+/// SQL parser errors.
+// Now the error in parser does not contain backtrace to avoid generating backtrace
+// every time the parser parses an invalid SQL.
+#[derive(Snafu)]
+#[snafu(visibility(pub))]
+#[stack_trace_debug]
+pub enum Error {
+    #[snafu(display(
+        "Column {} expect type: {:?}, actual: {:?}",
+        column_name,
+        expect,
+        actual,
+    ))]
+    ColumnTypeMismatch {
+        column_name: String,
+        expect: ConcreteDataType,
+        actual: ConcreteDataType,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Failed to parse value: {}", msg))]
+    ParseSqlValue {
+        msg: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display(
+        "Unsupported expr in default constraint: {:?} for column: {}",
+        expr,
+        column_name
+    ))]
+    UnsupportedDefaultValue {
+        column_name: String,
+        expr: Expr,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Unable to convert sql value {} to datatype {:?}", value, datatype))]
+    ConvertSqlValue {
+        value: SqlValue,
+        datatype: ConcreteDataType,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Invalid sql value: {}", value))]
+    InvalidSqlValue {
+        value: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Unsupported unary operator {}", unary_op))]
+    UnsupportedUnaryOp {
+        unary_op: UnaryOperator,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Invalid unary operator {} for value {}", unary_op, value))]
+    InvalidUnaryOp {
+        unary_op: UnaryOperator,
+        value: Value,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Failed to cast SQL value {} to datatype {}", sql_value, datatype))]
+    InvalidCast {
+        sql_value: sqlparser::ast::Value,
+        datatype: ConcreteDataType,
+        #[snafu(implicit)]
+        location: Location,
+        source: datatypes::error::Error,
+    },
+
+    #[snafu(display("Unable to convert {} to datatype {:?}", value, datatype))]
+    ConvertStr {
+        value: String,
+        datatype: ConcreteDataType,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display(
+        "Converting timestamp {:?} to unit {:?} overflow",
+        timestamp,
+        target_unit
+    ))]
+    TimestampOverflow {
+        timestamp: Timestamp,
+        target_unit: TimeUnit,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Datatype error: {}", source))]
+    Datatype {
+        source: datatypes::error::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
+}
+
+impl ErrorExt for Error {
+    fn status_code(&self) -> StatusCode {
+        use Error::*;
+
+        match self {
+            UnsupportedDefaultValue { .. } => StatusCode::Unsupported,
+            ParseSqlValue { .. } => StatusCode::InvalidSyntax,
+            ColumnTypeMismatch { .. }
+            | InvalidSqlValue { .. }
+            | UnsupportedUnaryOp { .. }
+            | InvalidUnaryOp { .. }
+            | InvalidCast { .. }
+            | ConvertStr { .. }
+            | TimestampOverflow { .. } => StatusCode::InvalidArguments,
+            Datatype { source, .. } => source.status_code(),
+            ConvertSqlValue { .. } => StatusCode::Unsupported,
+        }
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+}

src/common/sql/src/lib.rs

@@ -0,0 +1,19 @@
+// 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.
+
+#![feature(assert_matches)]
+
+pub mod convert;
+pub mod default_constraint;
+pub mod error;

src/common/telemetry/Cargo.toml

@@ -14,6 +14,7 @@ workspace = true
 [dependencies]
 backtrace = "0.3"
 common-error.workspace = true
+common-version.workspace = true
 console-subscriber = { version = "0.1", optional = true }
 greptime-proto.workspace = true
 humantime-serde.workspace = true

src/common/telemetry/src/logging.rs

@@ -403,7 +403,7 @@ pub fn init_global_logging(
                         resource::SERVICE_INSTANCE_ID,
                         node_id.unwrap_or("none".to_string()),
                     ),
-                    KeyValue::new(resource::SERVICE_VERSION, env!("CARGO_PKG_VERSION")),
+                    KeyValue::new(resource::SERVICE_VERSION, common_version::version()),
                     KeyValue::new(resource::PROCESS_PID, std::process::id().to_string()),
                 ]));
 

src/common/time/src/range.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::fmt::{Debug, Display, Formatter};
+use std::ops::{Bound, RangeBounds};
 
 use serde::{Deserialize, Serialize};
 
@@ -303,6 +304,41 @@ impl TimestampRange {
     }
 }
 
+/// Create [TimestampRange] from a timestamp tuple.
+/// The tuple's two elements form the "start" and "end" (both inclusive) of the timestamp range.
+impl From<(Timestamp, Timestamp)> for TimestampRange {
+    fn from((start, end): (Timestamp, Timestamp)) -> Self {
+        if start > end {
+            Self::empty()
+        } else {
+            Self::new_inclusive(Some(start), Some(end))
+        }
+    }
+}
+
+/// Create [TimestampRange] from Rust's "range".
+impl<R: RangeBounds<Timestamp>> From<R> for TimestampRange {
+    fn from(r: R) -> Self {
+        let start = match r.start_bound() {
+            Bound::Included(x) => Some(*x),
+            Bound::Excluded(x) => x
+                .value()
+                .checked_sub(1)
+                .map(|v| Timestamp::new(v, x.unit())),
+            Bound::Unbounded => None,
+        };
+        let end = match r.end_bound() {
+            Bound::Included(x) => x
+                .value()
+                .checked_add(1)
+                .map(|v| Timestamp::new(v, x.unit())),
+            Bound::Excluded(x) => Some(*x),
+            Bound::Unbounded => None,
+        };
+        Self::from_optional(start, end)
+    }
+}
+
 /// Time range in milliseconds.
 pub type RangeMillis = GenericRange<TimestampMillis>;
 
@@ -545,4 +581,75 @@ mod tests {
             TimeUnit::Nanosecond
         );
     }
+
+    #[test]
+    fn test_from_timestamp_tuple() {
+        let timestamp_range: TimestampRange =
+            (Timestamp::new_millisecond(1), Timestamp::new_millisecond(3)).into();
+        assert_eq!(
+            timestamp_range,
+            TimestampRange::from_optional(
+                Some(Timestamp::new_millisecond(1)),
+                Some(Timestamp::new_millisecond(4))
+            )
+        );
+
+        let timestamp_range: TimestampRange =
+            (Timestamp::new_millisecond(1), Timestamp::new_millisecond(1)).into();
+        assert_eq!(
+            timestamp_range,
+            TimestampRange::from_optional(
+                Some(Timestamp::new_millisecond(1)),
+                Some(Timestamp::new_millisecond(2))
+            )
+        );
+
+        let timestamp_range: TimestampRange =
+            (Timestamp::new_second(1), Timestamp::new_millisecond(3)).into();
+        assert_eq!(timestamp_range, TimestampRange::empty());
+    }
+
+    #[test]
+    fn test_from_timestamp_range() {
+        let timestamp_range: TimestampRange =
+            (Timestamp::new_millisecond(1)..Timestamp::new_millisecond(3)).into();
+        assert_eq!(
+            timestamp_range,
+            TimestampRange::from_optional(
+                Some(Timestamp::new_millisecond(1)),
+                Some(Timestamp::new_millisecond(3))
+            )
+        );
+
+        let timestamp_range: TimestampRange =
+            (Timestamp::new_millisecond(1)..=Timestamp::new_millisecond(3)).into();
+        assert_eq!(
+            timestamp_range,
+            TimestampRange::from_optional(
+                Some(Timestamp::new_millisecond(1)),
+                Some(Timestamp::new_millisecond(4))
+            )
+        );
+
+        let timestamp_range: TimestampRange = (Timestamp::new_millisecond(1)..).into();
+        assert_eq!(
+            timestamp_range,
+            TimestampRange::from_optional(Some(Timestamp::new_millisecond(1)), None)
+        );
+
+        let timestamp_range: TimestampRange = (..Timestamp::new_millisecond(3)).into();
+        assert_eq!(
+            timestamp_range,
+            TimestampRange::from_optional(None, Some(Timestamp::new_millisecond(3)))
+        );
+
+        let timestamp_range: TimestampRange = (..=Timestamp::new_millisecond(3)).into();
+        assert_eq!(
+            timestamp_range,
+            TimestampRange::from_optional(None, Some(Timestamp::new_millisecond(4)))
+        );
+
+        let timestamp_range: TimestampRange = (..).into();
+        assert_eq!(timestamp_range, TimestampRange::min_to_max(),);
+    }
 }

src/common/time/src/timestamp.rs

@@ -498,6 +498,17 @@ impl Timestamp {
 
     pub const MIN_NANOSECOND: Self = Self::new_nanosecond(i64::MIN);
     pub const MAX_NANOSECOND: Self = Self::new_nanosecond(i64::MAX);
+
+    /// Checks if a value would overflow for the given time unit.
+    pub fn is_overflow(value: i64, unit: TimeUnit) -> bool {
+        let (min_val, max_val) = match unit {
+            TimeUnit::Second => (Self::MIN_SECOND.value(), Self::MAX_SECOND.value()),
+            TimeUnit::Millisecond => (Self::MIN_MILLISECOND.value(), Self::MAX_MILLISECOND.value()),
+            TimeUnit::Microsecond => (Self::MIN_MICROSECOND.value(), Self::MAX_MICROSECOND.value()),
+            TimeUnit::Nanosecond => (Self::MIN_NANOSECOND.value(), Self::MAX_NANOSECOND.value()),
+        };
+        value < min_val || value > max_val
+    }
 }
 
 /// Converts the naive datetime (which has no specific timezone) to a

src/common/version/Cargo.toml

@@ -17,4 +17,5 @@ shadow-rs.workspace = true
 
 [build-dependencies]
 build-data = "0.2"
+cargo-manifest = "0.19"
 shadow-rs.workspace = true

src/common/version/build.rs

@@ -14,8 +14,10 @@
 
 use std::collections::BTreeSet;
 use std::env;
+use std::path::PathBuf;
 
 use build_data::{format_timestamp, get_source_time};
+use cargo_manifest::Manifest;
 use shadow_rs::{BuildPattern, ShadowBuilder, CARGO_METADATA, CARGO_TREE};
 
 fn main() -> shadow_rs::SdResult<()> {
@@ -33,6 +35,24 @@ fn main() -> shadow_rs::SdResult<()> {
     // solve the problem where the "CARGO_MANIFEST_DIR" is not what we want when this repo is
     // made as a submodule in another repo.
     let src_path = env::var("CARGO_WORKSPACE_DIR").or_else(|_| env::var("CARGO_MANIFEST_DIR"))?;
+
+    let manifest = Manifest::from_path(PathBuf::from(&src_path).join("Cargo.toml"))
+        .expect("Failed to parse Cargo.toml");
+    if let Some(product_version) = manifest.workspace.as_ref().and_then(|w| {
+        w.metadata.as_ref().and_then(|m| {
+            m.get("greptime")
+                .and_then(|g| g.get("product_version").and_then(|v| v.as_str()))
+        })
+    }) {
+        println!(
+            "cargo:rustc-env=GREPTIME_PRODUCT_VERSION={}",
+            product_version
+        );
+    } else {
+        let version = env::var("CARGO_PKG_VERSION").unwrap();
+        println!("cargo:rustc-env=GREPTIME_PRODUCT_VERSION={}", version,);
+    }
+
     let out_path = env::var("OUT_DIR")?;
 
     let _ = ShadowBuilder::builder()

src/common/version/src/lib.rs

@@ -105,13 +105,17 @@ pub const fn build_info() -> BuildInfo {
         build_time: env!("BUILD_TIMESTAMP"),
         rustc: build::RUST_VERSION,
         target: build::BUILD_TARGET,
-        version: build::PKG_VERSION,
+        version: env!("GREPTIME_PRODUCT_VERSION"),
     }
 }
 
 const BUILD_INFO: BuildInfo = build_info();
 
 pub const fn version() -> &'static str {
+    BUILD_INFO.version
+}
+
+pub const fn verbose_version() -> &'static str {
     const_format::formatcp!(
         "\nbranch: {}\ncommit: {}\nclean: {}\nversion: {}",
         BUILD_INFO.branch,

src/datanode/Cargo.toml

@@ -66,6 +66,7 @@ table.workspace = true
 tokio.workspace = true
 toml.workspace = true
 tonic.workspace = true
+tracing.workspace = true
 
 [dev-dependencies]
 cache.workspace = true

src/datanode/src/alive_keeper.rs

@@ -86,26 +86,33 @@ impl RegionAliveKeeper {
     /// Add the countdown task for a specific region.
     /// It will be ignored if the task exists.
     pub async fn register_region(&self, region_id: RegionId) {
-        if self.find_handle(region_id).await.is_some() {
-            return;
-        }
-
         let handle = Arc::new(CountdownTaskHandle::new(
             self.region_server.clone(),
             self.countdown_task_handler_ext.clone(),
             region_id,
         ));
 
-        let mut handles = self.tasks.lock().await;
-        let _ = handles.insert(region_id, handle.clone());
+        let should_start = {
+            let mut handles = self.tasks.lock().await;
 
-        if self.started.load(Ordering::Relaxed) {
+            // Check if already exists, return early if so
+            if handles.contains_key(&region_id) {
+                return;
+            }
+
+            // Insert new handle
+            handles.insert(region_id, handle.clone());
+
+            // Return whether we should start (check state inside lock)
+            self.started.load(Ordering::Relaxed)
+        };
+
+        if should_start {
             handle.start(self.heartbeat_interval_millis).await;
-
-            info!("Region alive countdown for region {region_id} is started!",);
+            info!("Region alive countdown for region {region_id} is started!");
         } else {
             info!(
-                "Region alive countdown for region {region_id} is registered but not started yet!",
+                "Region alive countdown for region {region_id} is registered but not started yet!"
             );
         }
     }
@@ -417,7 +424,15 @@ impl CountdownTask {
                         },
                         Some(CountdownCommand::Reset((role, deadline, extension_info))) => {
                             if let Err(err) = self.region_server.set_region_role(self.region_id, role) {
-                                error!(err; "Failed to set region role to {role} for region {region_id}");
+                                if err.status_code() == StatusCode::RegionNotFound {
+                                    // Table metadata in metasrv is deleted after its regions are dropped.
+                                    // The datanode may still receive lease renewal responses that depend on the metadata
+                                    // during the short period before it is removed.
+                                    warn!(err; "Failed to set region role to {role} for region {region_id}");
+                                }else{
+                                    error!(err; "Failed to set region role to {role} for region {region_id}");
+                                }
+
                             }
                             if let Some(ext_handler) = self.handler_ext.as_ref() {
                                 ext_handler.reset_deadline(

src/datanode/src/error.rs

@@ -402,6 +402,9 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display("Not yet implemented: {what}"))]
+    NotYetImplemented { what: String },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -456,7 +459,7 @@ impl ErrorExt for Error {
 
             OpenLogStore { source, .. } => source.status_code(),
             MetaClientInit { source, .. } => source.status_code(),
-            UnsupportedOutput { .. } => StatusCode::Unsupported,
+            UnsupportedOutput { .. } | NotYetImplemented { .. } => StatusCode::Unsupported,
             HandleRegionRequest { source, .. }
             | GetRegionMetadata { source, .. }
             | HandleBatchOpenRequest { source, .. }

src/datanode/src/metrics.rs

@@ -27,14 +27,14 @@ lazy_static! {
     pub static ref HANDLE_REGION_REQUEST_ELAPSED: HistogramVec = register_histogram_vec!(
         "greptime_datanode_handle_region_request_elapsed",
         "datanode handle region request elapsed",
-        &[REGION_ID, REGION_REQUEST_TYPE]
+        &[REGION_REQUEST_TYPE]
     )
     .unwrap();
     /// The number of rows in region request received by region server, labeled with request type.
     pub static ref REGION_CHANGED_ROW_COUNT: IntCounterVec = register_int_counter_vec!(
         "greptime_datanode_region_changed_row_count",
         "datanode region changed row count",
-        &[REGION_ID, REGION_REQUEST_TYPE]
+        &[REGION_REQUEST_TYPE]
     )
     .unwrap();
     /// The elapsed time since the last received heartbeat.

src/datanode/src/region_server.rs

@@ -968,9 +968,8 @@ impl RegionServerInner {
         request: RegionRequest,
     ) -> Result<RegionResponse> {
         let request_type = request.request_type();
-        let region_id_str = region_id.to_string();
         let _timer = crate::metrics::HANDLE_REGION_REQUEST_ELAPSED
-            .with_label_values(&[&region_id_str, request_type])
+            .with_label_values(&[request_type])
             .start_timer();
 
         let region_change = match &request {
@@ -1010,7 +1009,7 @@ impl RegionServerInner {
                 // Update metrics
                 if matches!(region_change, RegionChange::Ingest) {
                     crate::metrics::REGION_CHANGED_ROW_COUNT
-                        .with_label_values(&[&region_id_str, request_type])
+                        .with_label_values(&[request_type])
                         .inc_by(result.affected_rows as u64);
                 }
                 // Sets corresponding region status to ready.

src/datanode/src/tests.rs

@@ -40,7 +40,7 @@ use store_api::storage::{RegionId, ScanRequest, SequenceNumber};
 use table::TableRef;
 use tokio::sync::mpsc::{Receiver, Sender};
 
-use crate::error::Error;
+use crate::error::{Error, NotYetImplementedSnafu};
 use crate::event_listener::NoopRegionServerEventListener;
 use crate::region_server::RegionServer;
 
@@ -232,7 +232,9 @@ impl RegionEngine for MockRegionEngine {
         _region_id: RegionId,
         _request: ScanRequest,
     ) -> Result<RegionScannerRef, BoxedError> {
-        unimplemented!()
+        Err(BoxedError::new(
+            NotYetImplementedSnafu { what: "blah" }.build(),
+        ))
     }
 
     async fn get_metadata(&self, region_id: RegionId) -> Result<RegionMetadataRef, BoxedError> {

src/datatypes/Cargo.toml

@@ -28,7 +28,7 @@ greptime-proto.workspace = true
 jsonb.workspace = true
 num = "0.4"
 num-traits = "0.2"
-ordered-float = { version = "3.0", features = ["serde"] }
+ordered-float.workspace = true
 paste.workspace = true
 serde.workspace = true
 serde_json.workspace = true

src/flow/src/adapter.rs

@@ -497,7 +497,7 @@ impl StreamingEngine {
         &self,
         schema: &RelationDesc,
     ) -> Result<(Vec<String>, Vec<ColumnSchema>, bool), Error> {
-        // TODO(discord9): condiser remove buggy auto create by schema
+        // TODO(discord9): consider remove buggy auto create by schema
 
         // TODO(discord9): use default key from schema
         let primary_keys = schema

src/flow/src/batching_mode/engine.rs

@@ -14,7 +14,7 @@
 
 //! Batching mode engine
 
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::sync::Arc;
 
 use api::v1::flow::{DirtyWindowRequests, FlowResponse};
@@ -142,7 +142,7 @@ impl BatchingEngine {
 
             let handle: JoinHandle<Result<(), Error>> = tokio::spawn(async move {
                 let src_table_names = &task.config.source_table_names;
-                let mut all_dirty_windows = vec![];
+                let mut all_dirty_windows = HashSet::new();
                 for src_table_name in src_table_names {
                     if let Some((timestamps, unit)) = group_by_table_name.get(src_table_name) {
                         let Some(expr) = &task.config.time_window_expr else {
@@ -155,7 +155,7 @@ impl BatchingEngine {
                                 .context(UnexpectedSnafu {
                                     reason: "Failed to eval start value",
                                 })?;
-                            all_dirty_windows.push(align_start);
+                            all_dirty_windows.insert(align_start);
                         }
                     }
                 }

src/flow/src/batching_mode/time_window.rs

@@ -50,7 +50,8 @@ use snafu::{ensure, OptionExt, ResultExt};
 
 use crate::adapter::util::from_proto_to_data_type;
 use crate::error::{
-    ArrowSnafu, DatafusionSnafu, DatatypesSnafu, ExternalSnafu, PlanSnafu, UnexpectedSnafu,
+    ArrowSnafu, DatafusionSnafu, DatatypesSnafu, ExternalSnafu, PlanSnafu, TimeSnafu,
+    UnexpectedSnafu,
 };
 use crate::expr::error::DataTypeSnafu;
 use crate::Error;
@@ -74,6 +75,7 @@ pub struct TimeWindowExpr {
     logical_expr: Expr,
     df_schema: DFSchema,
     eval_time_window_size: Option<std::time::Duration>,
+    eval_time_original: Option<Timestamp>,
 }
 
 impl std::fmt::Display for TimeWindowExpr {
@@ -106,10 +108,11 @@ impl TimeWindowExpr {
             logical_expr: expr.clone(),
             df_schema: df_schema.clone(),
             eval_time_window_size: None,
+            eval_time_original: None,
         };
         let test_ts = DEFAULT_TEST_TIMESTAMP;
-        let (l, u) = zelf.eval(test_ts)?;
-        let time_window_size = match (l, u) {
+        let (lower, upper) = zelf.eval(test_ts)?;
+        let time_window_size = match (lower, upper) {
             (Some(l), Some(u)) => u.sub(&l).map(|r| r.to_std()).transpose().map_err(|_| {
                 UnexpectedSnafu {
                     reason: format!(
@@ -121,13 +124,59 @@ impl TimeWindowExpr {
             _ => None,
         };
         zelf.eval_time_window_size = time_window_size;
+        zelf.eval_time_original = lower;
+
         Ok(zelf)
     }
 
+    /// TODO(discord9): add `eval_batch` too
     pub fn eval(
         &self,
         current: Timestamp,
     ) -> Result<(Option<Timestamp>, Option<Timestamp>), Error> {
+        fn compute_distance(time_diff_ns: i64, stride_ns: i64) -> i64 {
+            if stride_ns == 0 {
+                return time_diff_ns;
+            }
+            // a - (a % n) impl ceil to nearest n * stride
+            let time_delta = time_diff_ns - (time_diff_ns % stride_ns);
+
+            if time_diff_ns < 0 && time_delta != time_diff_ns {
+                // The origin is later than the source timestamp, round down to the previous bin
+
+                time_delta - stride_ns
+            } else {
+                time_delta
+            }
+        }
+
+        // FAST PATH: if we have eval_time_original and eval_time_window_size,
+        // we can compute the bounds directly
+        if let (Some(original), Some(window_size)) =
+            (self.eval_time_original, self.eval_time_window_size)
+        {
+            // date_bin align current to lower bound
+            let time_diff_ns = current.sub(&original).and_then(|s|s.num_nanoseconds()).with_context(||UnexpectedSnafu {
+                reason: format!(
+                    "Failed to compute time difference between current {current:?} and original {original:?}"
+                ),
+            })?;
+
+            let window_size_ns = window_size.as_nanos() as i64;
+
+            let distance_ns = compute_distance(time_diff_ns, window_size_ns);
+
+            let lower_bound = if distance_ns >= 0 {
+                original.add_duration(std::time::Duration::from_nanos(distance_ns as u64))
+            } else {
+                original.sub_duration(std::time::Duration::from_nanos((-distance_ns) as u64))
+            }
+            .context(TimeSnafu)?;
+            let upper_bound = lower_bound.add_duration(window_size).context(TimeSnafu)?;
+
+            return Ok((Some(lower_bound), Some(upper_bound)));
+        }
+
         let lower_bound =
             calc_expr_time_window_lower_bound(&self.phy_expr, &self.df_schema, current)?;
         let upper_bound =

src/frontend/Cargo.toml

@@ -74,6 +74,7 @@ tokio.workspace = true
 tokio-util.workspace = true
 toml.workspace = true
 tonic.workspace = true
+tracing.workspace = true
 
 [dev-dependencies]
 catalog = { workspace = true, features = ["testing"] }

src/frontend/src/instance.rs

@@ -383,6 +383,13 @@ impl SqlQueryHandler for Instance {
             .and_then(|stmts| query_interceptor.post_parsing(stmts, query_ctx.clone()))
         {
             Ok(stmts) => {
+                if stmts.is_empty() {
+                    return vec![InvalidSqlSnafu {
+                        err_msg: "empty statements",
+                    }
+                    .fail()];
+                }
+
                 let mut results = Vec::with_capacity(stmts.len());
                 for stmt in stmts {
                     if let Err(e) = checker

src/meta-srv/src/bootstrap.rs

@@ -95,11 +95,10 @@ pub struct MetasrvInstance {
 }
 
 impl MetasrvInstance {
-    pub async fn new(
-        opts: MetasrvOptions,
-        plugins: Plugins,
-        metasrv: Metasrv,
-    ) -> Result<MetasrvInstance> {
+    pub async fn new(metasrv: Metasrv) -> Result<MetasrvInstance> {
+        let opts = metasrv.options().clone();
+        let plugins = metasrv.plugins().clone();
+
         let builder = HttpServerBuilder::new(opts.http.clone())
             .with_metrics_handler(MetricsHandler)
             .with_greptime_config_options(opts.to_toml().context(error::TomlFormatSnafu)?);

src/meta-srv/src/lease.rs

@@ -20,8 +20,11 @@ use std::task::{Context, Poll};
 
 use api::v1::meta::heartbeat_request::NodeWorkloads;
 use common_error::ext::BoxedError;
+use common_meta::cluster::{NodeInfo, NodeInfoKey, Role as ClusterRole};
+use common_meta::distributed_time_constants::FRONTEND_HEARTBEAT_INTERVAL_MILLIS;
 use common_meta::kv_backend::{KvBackend, ResettableKvBackendRef};
 use common_meta::peer::{Peer, PeerLookupService};
+use common_meta::rpc::store::RangeRequest;
 use common_meta::{util, DatanodeId, FlownodeId};
 use common_time::util as time_util;
 use common_workload::DatanodeWorkloadType;
@@ -31,10 +34,19 @@ use crate::cluster::MetaPeerClientRef;
 use crate::error::{Error, KvBackendSnafu, Result};
 use crate::key::{DatanodeLeaseKey, FlownodeLeaseKey, LeaseValue};
 
-fn build_lease_filter(lease_secs: u64) -> impl Fn(&LeaseValue) -> bool {
-    move |v: &LeaseValue| {
-        ((time_util::current_time_millis() - v.timestamp_millis) as u64)
-            < lease_secs.saturating_mul(1000)
+enum Value<'a> {
+    LeaseValue(&'a LeaseValue),
+    NodeInfo(&'a NodeInfo),
+}
+
+fn build_lease_filter(lease_secs: u64) -> impl Fn(Value) -> bool {
+    move |value: Value| {
+        let active_time = match value {
+            Value::LeaseValue(lease_value) => lease_value.timestamp_millis,
+            Value::NodeInfo(node_info) => node_info.last_activity_ts,
+        };
+
+        ((time_util::current_time_millis() - active_time) as u64) < lease_secs.saturating_mul(1000)
     }
 }
 
@@ -91,7 +103,7 @@ pub async fn lookup_datanode_peer(
         return Ok(None);
     };
     let lease_value: LeaseValue = kv.value.try_into()?;
-    let is_alive = lease_filter(&lease_value);
+    let is_alive = lease_filter(Value::LeaseValue(&lease_value));
     if is_alive {
         Ok(Some(Peer {
             id: lease_key.node_id,
@@ -155,7 +167,7 @@ where
             let condition = this.condition;
             let key_prefix = std::mem::take(&mut this.key_prefix);
             let fut = filter(key_prefix, this.meta_peer_client, move |v| {
-                lease_filter(v) && condition.unwrap_or(|_| true)(v)
+                lease_filter(Value::LeaseValue(v)) && condition.unwrap_or(|_| true)(v)
             });
 
             this.inner_future = Some(Box::pin(fut));
@@ -192,7 +204,7 @@ pub async fn lookup_flownode_peer(
     };
     let lease_value: LeaseValue = kv.value.try_into()?;
 
-    let is_alive = lease_filter(&lease_value);
+    let is_alive = lease_filter(Value::LeaseValue(&lease_value));
     if is_alive {
         Ok(Some(Peer {
             id: lease_key.node_id,
@@ -203,6 +215,29 @@ pub async fn lookup_flownode_peer(
     }
 }
 
+/// Lookup all alive frontends from the memory backend, only return if it's alive under given `lease_secs`.
+pub async fn lookup_frontends(
+    meta_peer_client: &MetaPeerClientRef,
+    lease_secs: u64,
+) -> Result<Vec<Peer>> {
+    let range_request =
+        RangeRequest::new().with_prefix(NodeInfoKey::key_prefix_with_role(ClusterRole::Frontend));
+
+    let response = meta_peer_client.range(range_request).await?;
+    let lease_filter = build_lease_filter(lease_secs);
+
+    let mut peers = Vec::with_capacity(response.kvs.len());
+    for kv in response.kvs {
+        let node_info = NodeInfo::try_from(kv.value).context(KvBackendSnafu)?;
+        let is_alive = lease_filter(Value::NodeInfo(&node_info));
+        if is_alive {
+            peers.push(node_info.peer);
+        }
+    }
+
+    Ok(peers)
+}
+
 /// Find all alive flownodes
 pub fn alive_flownodes(
     meta_peer_client: &MetaPeerClientRef,
@@ -264,25 +299,42 @@ impl PeerLookupService for MetaPeerLookupService {
             .map_err(BoxedError::new)
             .context(common_meta::error::ExternalSnafu)
     }
+
     async fn flownode(&self, id: FlownodeId) -> common_meta::error::Result<Option<Peer>> {
         lookup_flownode_peer(id, &self.meta_peer_client, u64::MAX)
             .await
             .map_err(BoxedError::new)
             .context(common_meta::error::ExternalSnafu)
     }
+
+    async fn active_frontends(&self) -> common_meta::error::Result<Vec<Peer>> {
+        // Get the active frontends within the last heartbeat interval.
+        lookup_frontends(
+            &self.meta_peer_client,
+            // TODO(zyy17): How to get the heartbeat interval of the frontend if it uses a custom heartbeat interval?
+            FRONTEND_HEARTBEAT_INTERVAL_MILLIS,
+        )
+        .await
+        .map_err(BoxedError::new)
+        .context(common_meta::error::ExternalSnafu)
+    }
 }
 
 #[cfg(test)]
 mod tests {
     use api::v1::meta::heartbeat_request::NodeWorkloads;
     use api::v1::meta::DatanodeWorkloads;
+    use common_meta::cluster::{FrontendStatus, NodeInfo, NodeInfoKey, NodeStatus};
     use common_meta::kv_backend::ResettableKvBackendRef;
+    use common_meta::peer::Peer;
     use common_meta::rpc::store::PutRequest;
     use common_time::util::current_time_millis;
     use common_workload::DatanodeWorkloadType;
 
     use crate::key::{DatanodeLeaseKey, LeaseValue};
-    use crate::lease::{alive_datanodes, is_datanode_accept_ingest_workload};
+    use crate::lease::{
+        alive_datanodes, is_datanode_accept_ingest_workload, lookup_frontends, ClusterRole,
+    };
     use crate::test_util::create_meta_peer_client;
 
     async fn put_lease_value(
@@ -391,4 +443,60 @@ mod tests {
         assert_eq!(leases.len(), 1);
         assert!(leases.contains_key(&DatanodeLeaseKey { node_id: 2 }));
     }
+
+    #[tokio::test]
+    async fn test_lookup_frontends() {
+        let client = create_meta_peer_client();
+        let in_memory = client.memory_backend();
+        let lease_secs = 10;
+
+        let active_frontend_node = NodeInfo {
+            peer: Peer {
+                id: 0,
+                addr: "127.0.0.1:20201".to_string(),
+            },
+            last_activity_ts: current_time_millis(),
+            status: NodeStatus::Frontend(FrontendStatus {}),
+            version: "1.0.0".to_string(),
+            git_commit: "1234567890".to_string(),
+            start_time_ms: current_time_millis() as u64,
+        };
+
+        let key_prefix = NodeInfoKey::key_prefix_with_role(ClusterRole::Frontend);
+
+        in_memory
+            .put(PutRequest {
+                key: format!("{}{}", key_prefix, "0").into(),
+                value: active_frontend_node.try_into().unwrap(),
+                prev_kv: false,
+            })
+            .await
+            .unwrap();
+
+        let inactive_frontend_node = NodeInfo {
+            peer: Peer {
+                id: 1,
+                addr: "127.0.0.1:20201".to_string(),
+            },
+            last_activity_ts: current_time_millis() - 20 * 1000,
+            status: NodeStatus::Frontend(FrontendStatus {}),
+            version: "1.0.0".to_string(),
+            git_commit: "1234567890".to_string(),
+            start_time_ms: current_time_millis() as u64,
+        };
+
+        in_memory
+            .put(PutRequest {
+                key: format!("{}{}", key_prefix, "1").into(),
+                value: inactive_frontend_node.try_into().unwrap(),
+                prev_kv: false,
+            })
+            .await
+            .unwrap();
+
+        let peers = lookup_frontends(&client, lease_secs as u64).await.unwrap();
+
+        assert_eq!(peers.len(), 1);
+        assert_eq!(peers[0].id, 0);
+    }
 }

src/meta-srv/src/procedure/tests.rs

@@ -15,7 +15,6 @@
 use std::collections::{HashMap, HashSet};
 use std::sync::{Arc, Mutex};
 
-use api::v1::meta::Partition;
 use api::v1::region::region_request::Body as PbRegionRequest;
 use api::v1::region::{CreateRequest as PbCreateRegionRequest, RegionColumnDef};
 use api::v1::{ColumnDataType, ColumnDef as PbColumnDef, SemanticType};
@@ -84,14 +83,7 @@ fn create_table_task(table_name: Option<&str>) -> CreateTableTask {
         .into();
 
     let table_info = build_raw_table_info_from_expr(&expr);
-    CreateTableTask::new(
-        expr,
-        vec![Partition {
-            column_list: vec![],
-            value_list: vec![],
-        }],
-        table_info,
-    )
+    CreateTableTask::new(expr, None, table_info)
 }
 
 #[test]

src/metric-engine/Cargo.toml

@@ -38,6 +38,7 @@ smallvec.workspace = true
 snafu.workspace = true
 store-api.workspace = true
 tokio.workspace = true
+tracing.workspace = true
 
 [dev-dependencies]
 common-meta = { workspace = true, features = ["testing"] }

src/mito2/Cargo.toml

@@ -42,6 +42,7 @@ datafusion-common.workspace = true
 datafusion-expr.workspace = true
 datatypes.workspace = true
 dotenv.workspace = true
+either.workspace = true
 futures.workspace = true
 humantime-serde.workspace = true
 index.workspace = true
@@ -75,6 +76,7 @@ table.workspace = true
 tokio.workspace = true
 tokio-stream.workspace = true
 tokio-util.workspace = true
+tracing.workspace = true
 uuid.workspace = true
 
 [dev-dependencies]

src/mito2/src/compaction.rs

@@ -368,6 +368,7 @@ impl CompactionScheduler {
                     picker_output: picker_output.clone(),
                     start_time,
                     waiters,
+                    ttl,
                 };
 
                 let result = remote_job_scheduler

src/mito2/src/compaction/compactor.rs

@@ -20,6 +20,7 @@ use api::v1::region::compact_request;
 use common_meta::key::SchemaMetadataManagerRef;
 use common_telemetry::{info, warn};
 use common_time::TimeToLive;
+use either::Either;
 use itertools::Itertools;
 use object_store::manager::ObjectStoreManagerRef;
 use serde::{Deserialize, Serialize};
@@ -116,7 +117,7 @@ pub async fn open_compaction_region(
     req: &OpenCompactionRegionRequest,
     mito_config: &MitoConfig,
     object_store_manager: ObjectStoreManagerRef,
-    schema_metadata_manager: SchemaMetadataManagerRef,
+    ttl_provider: Either<TimeToLive, SchemaMetadataManagerRef>,
 ) -> Result<CompactionRegion> {
     let object_store = {
         let name = &req.region_options.storage;
@@ -197,16 +198,22 @@ pub async fn open_compaction_region(
         }
     };
 
-    let ttl = find_ttl(
-        req.region_id.table_id(),
-        current_version.options.ttl,
-        &schema_metadata_manager,
-    )
-    .await
-    .unwrap_or_else(|e| {
-        warn!(e; "Failed to get ttl for region: {}", region_metadata.region_id);
-        TimeToLive::default()
-    });
+    let ttl = match ttl_provider {
+        // Use the specified ttl.
+        Either::Left(ttl) => ttl,
+        // Get the ttl from the schema metadata manager.
+        Either::Right(schema_metadata_manager) => find_ttl(
+            req.region_id.table_id(),
+            current_version.options.ttl,
+            &schema_metadata_manager,
+        )
+        .await
+        .unwrap_or_else(|e| {
+            warn!(e; "Failed to get ttl for region: {}", region_metadata.region_id);
+            TimeToLive::default()
+        }),
+    };
+
     Ok(CompactionRegion {
         region_id: req.region_id,
         region_options: req.region_options.clone(),

src/mito2/src/engine/open_test.rs

@@ -19,6 +19,7 @@ use api::v1::Rows;
 use common_error::ext::ErrorExt;
 use common_error::status_code::StatusCode;
 use common_recordbatch::RecordBatches;
+use either::Either;
 use store_api::region_engine::{RegionEngine, RegionRole};
 use store_api::region_request::{
     RegionCloseRequest, RegionOpenRequest, RegionPutRequest, RegionRequest,
@@ -474,7 +475,7 @@ async fn test_open_compaction_region() {
         &req,
         &mito_config,
         object_store_manager.clone(),
-        schema_metadata_manager,
+        Either::Right(schema_metadata_manager),
     )
     .await
     .unwrap();

src/mito2/src/extension.rs

@@ -1,3 +1,4 @@
+use std::fmt::Display;
 use std::sync::Arc;
 
 use async_trait::async_trait;
@@ -19,7 +20,7 @@ pub type InclusiveTimeRange = (Timestamp, Timestamp);
 /// memtable range and sst file range, but resides on the outside.
 /// It can be scanned side by side as other ranges to produce the final result, so it's very useful
 /// to extend the source of data in GreptimeDB.
-pub trait ExtensionRange: Send + Sync {
+pub trait ExtensionRange: Display + Send + Sync {
     /// The number of rows in this range.
     fn num_rows(&self) -> u64;
 

src/mito2/src/memtable.rs

@@ -98,6 +98,12 @@ impl MemtableStats {
         self
     }
 
+    #[cfg(feature = "test")]
+    pub fn with_max_sequence(mut self, max_sequence: SequenceNumber) -> Self {
+        self.max_sequence = max_sequence;
+        self
+    }
+
     /// Returns the estimated bytes allocated by this memtable.
     pub fn bytes_allocated(&self) -> usize {
         self.estimated_bytes

src/mito2/src/memtable/bulk/part.rs

@@ -694,7 +694,7 @@ mod tests {
         let read_format = ReadFormat::new_with_all_columns(metadata.clone());
         let mut batches = VecDeque::new();
         read_format
-            .convert_record_batch(&batch, &mut batches)
+            .convert_record_batch(&batch, None, &mut batches)
             .unwrap();
         if !dedup {
             assert_eq!(

src/mito2/src/read/scan_region.rs

@@ -1040,13 +1040,15 @@ impl StreamContext {
 
     /// Format the context for explain.
     pub(crate) fn format_for_explain(&self, verbose: bool, f: &mut fmt::Formatter) -> fmt::Result {
-        let (mut num_mem_ranges, mut num_file_ranges) = (0, 0);
+        let (mut num_mem_ranges, mut num_file_ranges, mut num_other_ranges) = (0, 0, 0);
         for range_meta in &self.ranges {
             for idx in &range_meta.row_group_indices {
                 if self.is_mem_range_index(*idx) {
                     num_mem_ranges += 1;
-                } else {
+                } else if self.is_file_range_index(*idx) {
                     num_file_ranges += 1;
+                } else {
+                    num_other_ranges += 1;
                 }
             }
         }
@@ -1055,12 +1057,17 @@ impl StreamContext {
         }
         write!(
             f,
-            "\"partition_count\":{{\"count\":{}, \"mem_ranges\":{}, \"files\":{}, \"file_ranges\":{}}}",
+            r#""partition_count":{{"count":{}, "mem_ranges":{}, "files":{}, "file_ranges":{}"#,
             self.ranges.len(),
             num_mem_ranges,
             self.input.num_files(),
             num_file_ranges,
         )?;
+        if num_other_ranges > 0 {
+            write!(f, r#"", other_ranges":{}"#, num_other_ranges)?;
+        }
+        write!(f, "}}")?;
+
         if let Some(selector) = &self.input.series_row_selector {
             write!(f, ", \"selector\":\"{}\"", selector)?;
         }
@@ -1102,6 +1109,24 @@ impl StreamContext {
             input: &'a ScanInput,
         }
 
+        #[cfg(feature = "enterprise")]
+        impl InputWrapper<'_> {
+            fn format_extension_ranges(&self, f: &mut fmt::Formatter) -> fmt::Result {
+                if self.input.extension_ranges.is_empty() {
+                    return Ok(());
+                }
+
+                let mut delimiter = "";
+                write!(f, ", extension_ranges: [")?;
+                for range in self.input.extension_ranges() {
+                    write!(f, "{}{}", delimiter, range)?;
+                    delimiter = ", ";
+                }
+                write!(f, "]")?;
+                Ok(())
+            }
+        }
+
         impl fmt::Debug for InputWrapper<'_> {
             fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
                 let output_schema = self.input.mapper.output_schema();
@@ -1127,6 +1152,9 @@ impl StreamContext {
                         .finish()?;
                 }
 
+                #[cfg(feature = "enterprise")]
+                self.format_extension_ranges(f)?;
+
                 Ok(())
             }
         }

src/mito2/src/read/scan_util.rs

@@ -598,16 +598,17 @@ pub fn build_file_range_scan_stream(
 pub(crate) async fn scan_extension_range(
     context: Arc<StreamContext>,
     index: RowGroupIndex,
-    metrics: PartitionMetrics,
+    partition_metrics: PartitionMetrics,
 ) -> Result<BoxedBatchStream> {
     use snafu::ResultExt;
 
     let range = context.input.extension_range(index.index);
     let reader = range.reader(context.as_ref());
-    reader
-        .read(context, metrics, index)
+    let stream = reader
+        .read(context, partition_metrics, index)
         .await
-        .context(crate::error::ScanExternalRangeSnafu)
+        .context(crate::error::ScanExternalRangeSnafu)?;
+    Ok(stream)
 }
 
 pub(crate) async fn maybe_scan_other_ranges(

src/mito2/src/schedule/remote_job_scheduler.rs

@@ -17,6 +17,7 @@ use std::sync::Arc;
 use std::time::Instant;
 
 use common_telemetry::error;
+use common_time::TimeToLive;
 use serde::{Deserialize, Serialize};
 use snafu::{Location, ResultExt, Snafu};
 use store_api::storage::RegionId;
@@ -108,6 +109,7 @@ pub struct CompactionJob {
     pub compaction_region: CompactionRegion,
     pub picker_output: PickerOutput,
     pub start_time: Instant,
+    pub ttl: TimeToLive,
     /// Send the result of the compaction job to these waiters.
     pub waiters: Vec<OutputTx>,
 }

src/mito2/src/sst/parquet.rs

@@ -95,7 +95,7 @@ mod tests {
     use datafusion_common::{Column, ScalarValue};
     use datafusion_expr::{col, lit, BinaryExpr, Expr, Operator};
     use datatypes::arrow;
-    use datatypes::arrow::array::RecordBatch;
+    use datatypes::arrow::array::{RecordBatch, UInt64Array};
     use datatypes::arrow::datatypes::{DataType, Field, Schema};
     use parquet::arrow::AsyncArrowWriter;
     use parquet::basic::{Compression, Encoding, ZstdLevel};
@@ -107,7 +107,7 @@ mod tests {
     use super::*;
     use crate::access_layer::{FilePathProvider, OperationType, RegionFilePathFactory};
     use crate::cache::{CacheManager, CacheStrategy, PageKey};
-    use crate::read::BatchReader;
+    use crate::read::{BatchBuilder, BatchReader};
     use crate::region::options::{IndexOptions, InvertedIndexOptions};
     use crate::sst::file::{FileHandle, FileMeta};
     use crate::sst::file_purger::NoopFilePurger;
@@ -120,8 +120,8 @@ mod tests {
     use crate::sst::{location, DEFAULT_WRITE_CONCURRENCY};
     use crate::test_util::sst_util::{
         assert_parquet_metadata_eq, build_test_binary_test_region_metadata, new_batch_by_range,
-        new_batch_with_binary, new_source, sst_file_handle, sst_file_handle_with_file_id,
-        sst_region_metadata,
+        new_batch_with_binary, new_batch_with_custom_sequence, new_source, sst_file_handle,
+        sst_file_handle_with_file_id, sst_region_metadata,
     };
     use crate::test_util::{check_reader_result, TestEnv};
 
@@ -895,4 +895,84 @@ mod tests {
         assert!(cached.contains_row_group(2));
         assert!(cached.contains_row_group(3));
     }
+
+    #[tokio::test]
+    async fn test_read_with_override_sequence() {
+        let mut env = TestEnv::new().await;
+        let object_store = env.init_object_store_manager();
+        let handle = sst_file_handle(0, 1000);
+        let file_path = FixedPathProvider {
+            file_id: handle.file_id(),
+        };
+        let metadata = Arc::new(sst_region_metadata());
+
+        // Create batches with sequence 0 to trigger override functionality
+        let batch1 = new_batch_with_custom_sequence(&["a", "d"], 0, 60, 0);
+        let batch2 = new_batch_with_custom_sequence(&["b", "f"], 0, 40, 0);
+        let source = new_source(&[batch1, batch2]);
+
+        let write_opts = WriteOptions {
+            row_group_size: 50,
+            ..Default::default()
+        };
+
+        let mut writer = ParquetWriter::new_with_object_store(
+            object_store.clone(),
+            metadata.clone(),
+            NoopIndexBuilder,
+            file_path,
+        )
+        .await;
+
+        writer
+            .write_all(source, None, &write_opts)
+            .await
+            .unwrap()
+            .remove(0);
+
+        // Read without override sequence (should read sequence 0)
+        let builder =
+            ParquetReaderBuilder::new(FILE_DIR.to_string(), handle.clone(), object_store.clone());
+        let mut reader = builder.build().await.unwrap();
+        let mut normal_batches = Vec::new();
+        while let Some(batch) = reader.next_batch().await.unwrap() {
+            normal_batches.push(batch);
+        }
+
+        // Read with override sequence using FileMeta.sequence
+        let custom_sequence = 12345u64;
+        let file_meta = handle.meta_ref();
+        let mut override_file_meta = file_meta.clone();
+        override_file_meta.sequence = Some(std::num::NonZero::new(custom_sequence).unwrap());
+        let override_handle = FileHandle::new(
+            override_file_meta,
+            Arc::new(crate::sst::file_purger::NoopFilePurger),
+        );
+
+        let builder =
+            ParquetReaderBuilder::new(FILE_DIR.to_string(), override_handle, object_store.clone());
+        let mut reader = builder.build().await.unwrap();
+        let mut override_batches = Vec::new();
+        while let Some(batch) = reader.next_batch().await.unwrap() {
+            override_batches.push(batch);
+        }
+
+        // Compare the results
+        assert_eq!(normal_batches.len(), override_batches.len());
+        for (normal, override_batch) in normal_batches.into_iter().zip(override_batches.iter()) {
+            // Create expected batch with override sequence
+            let expected_batch = {
+                let num_rows = normal.num_rows();
+                let mut builder = BatchBuilder::from(normal);
+                builder
+                    .sequences_array(Arc::new(UInt64Array::from_value(custom_sequence, num_rows)))
+                    .unwrap();
+
+                builder.build().unwrap()
+            };
+
+            // Override batch should match expected batch
+            assert_eq!(*override_batch, expected_batch);
+        }
+    }
 }

src/mito2/src/sst/parquet/format.rs

@@ -146,6 +146,8 @@ pub struct ReadFormat {
     /// Field column id to their index in the projected schema (
     /// the schema of [Batch]).
     field_id_to_projected_index: HashMap<ColumnId, usize>,
+    /// Sequence number to override the sequence read from the SST.
+    override_sequence: Option<SequenceNumber>,
 }
 
 impl ReadFormat {
@@ -197,9 +199,15 @@ impl ReadFormat {
             field_id_to_index,
             projection_indices,
             field_id_to_projected_index,
+            override_sequence: None,
         }
     }
 
+    /// Sets the sequence number to override.
+    pub(crate) fn set_override_sequence(&mut self, sequence: Option<SequenceNumber>) {
+        self.override_sequence = sequence;
+    }
+
     /// Gets the arrow schema of the SST file.
     ///
     /// This schema is computed from the region metadata but should be the same
@@ -218,12 +226,20 @@ impl ReadFormat {
         &self.projection_indices
     }
 
+    /// Creates a sequence array to override.
+    pub(crate) fn new_override_sequence_array(&self, length: usize) -> Option<ArrayRef> {
+        self.override_sequence
+            .map(|seq| Arc::new(UInt64Array::from_value(seq, length)) as ArrayRef)
+    }
+
     /// Convert a arrow record batch into `batches`.
     ///
+    /// The length of `override_sequence_array` must be larger than the length of the record batch.
     /// Note that the `record_batch` may only contains a subset of columns if it is projected.
     pub fn convert_record_batch(
         &self,
         record_batch: &RecordBatch,
+        override_sequence_array: Option<&ArrayRef>,
         batches: &mut VecDeque<Batch>,
     ) -> Result<()> {
         debug_assert!(batches.is_empty());
@@ -246,11 +262,23 @@ impl ReadFormat {
             .take(FIXED_POS_COLUMN_NUM);
         // Safety: We have checked the column number.
         let op_type_array = fixed_pos_columns.next().unwrap();
-        let sequence_array = fixed_pos_columns.next().unwrap();
+        let mut sequence_array = fixed_pos_columns.next().unwrap().clone();
         let pk_array = fixed_pos_columns.next().unwrap();
         let ts_array = fixed_pos_columns.next().unwrap();
         let field_batch_columns = self.get_field_batch_columns(record_batch)?;
 
+        // Override sequence array if provided.
+        if let Some(override_array) = override_sequence_array {
+            assert!(override_array.len() >= sequence_array.len());
+            // It's fine to assign the override array directly, but we slice it to make
+            // sure it matches the length of the original sequence array.
+            sequence_array = if override_array.len() > sequence_array.len() {
+                override_array.slice(0, sequence_array.len())
+            } else {
+                override_array.clone()
+            };
+        }
+
         // Compute primary key offsets.
         let pk_dict_array = pk_array
             .as_any()
@@ -691,6 +719,39 @@ pub(crate) fn parquet_row_group_time_range(
     Some((Timestamp::new(min, unit), Timestamp::new(max, unit)))
 }
 
+/// Checks if sequence override is needed based on all row groups' statistics.
+/// Returns true if ALL row groups have sequence min-max values of 0.
+pub(crate) fn need_override_sequence(parquet_meta: &ParquetMetaData) -> bool {
+    let num_columns = parquet_meta.file_metadata().schema_descr().num_columns();
+    if num_columns < FIXED_POS_COLUMN_NUM {
+        return false;
+    }
+
+    // The sequence column is the second-to-last column (before op_type)
+    let sequence_pos = num_columns - 2;
+
+    // Check all row groups - all must have sequence min-max of 0
+    for row_group in parquet_meta.row_groups() {
+        if let Some(Statistics::Int64(value_stats)) = row_group.column(sequence_pos).statistics() {
+            if let (Some(min_val), Some(max_val)) = (value_stats.min_opt(), value_stats.max_opt()) {
+                // If any row group doesn't have min=0 and max=0, return false
+                if *min_val != 0 || *max_val != 0 {
+                    return false;
+                }
+            } else {
+                // If any row group doesn't have statistics, return false
+                return false;
+            }
+        } else {
+            // If any row group doesn't have Int64 statistics, return false
+            return false;
+        }
+    }
+
+    // All row groups have sequence min-max of 0, or there are no row groups
+    !parquet_meta.row_groups().is_empty()
+}
+
 #[cfg(test)]
 mod tests {
     use api::v1::OpType;
@@ -775,9 +836,19 @@ mod tests {
     }
 
     fn new_batch(primary_key: &[u8], start_ts: i64, start_field: i64, num_rows: usize) -> Batch {
+        new_batch_with_sequence(primary_key, start_ts, start_field, num_rows, TEST_SEQUENCE)
+    }
+
+    fn new_batch_with_sequence(
+        primary_key: &[u8],
+        start_ts: i64,
+        start_field: i64,
+        num_rows: usize,
+        sequence: u64,
+    ) -> Batch {
         let ts_values = (0..num_rows).map(|i| start_ts + i as i64);
         let timestamps = Arc::new(TimestampMillisecondVector::from_values(ts_values));
-        let sequences = Arc::new(UInt64Vector::from_vec(vec![TEST_SEQUENCE; num_rows]));
+        let sequences = Arc::new(UInt64Vector::from_vec(vec![sequence; num_rows]));
         let op_types = Arc::new(UInt8Vector::from_vec(vec![TEST_OP_TYPE; num_rows]));
         let fields = vec![
             BatchColumn {
@@ -930,7 +1001,7 @@ mod tests {
         let record_batch = RecordBatch::new_empty(arrow_schema);
         let mut batches = VecDeque::new();
         read_format
-            .convert_record_batch(&record_batch, &mut batches)
+            .convert_record_batch(&record_batch, None, &mut batches)
             .unwrap();
         assert!(batches.is_empty());
     }
@@ -957,7 +1028,7 @@ mod tests {
         let record_batch = RecordBatch::try_new(arrow_schema, columns).unwrap();
         let mut batches = VecDeque::new();
         read_format
-            .convert_record_batch(&record_batch, &mut batches)
+            .convert_record_batch(&record_batch, None, &mut batches)
             .unwrap();
 
         assert_eq!(
@@ -965,4 +1036,45 @@ mod tests {
             batches.into_iter().collect::<Vec<_>>(),
         );
     }
+
+    #[test]
+    fn test_convert_record_batch_with_override_sequence() {
+        let metadata = build_test_region_metadata();
+        let column_ids: Vec<_> = metadata
+            .column_metadatas
+            .iter()
+            .map(|col| col.column_id)
+            .collect();
+        let read_format = ReadFormat::new(metadata, column_ids.iter().copied());
+
+        let columns: Vec<ArrayRef> = vec![
+            Arc::new(Int64Array::from(vec![1, 1, 10, 10])), // field1
+            Arc::new(Int64Array::from(vec![2, 2, 11, 11])), // field0
+            Arc::new(TimestampMillisecondArray::from(vec![1, 2, 11, 12])), // ts
+            build_test_pk_array(&[(b"one".to_vec(), 2), (b"two".to_vec(), 2)]), // primary key
+            Arc::new(UInt64Array::from(vec![TEST_SEQUENCE; 4])), // sequence
+            Arc::new(UInt8Array::from(vec![TEST_OP_TYPE; 4])), // op type
+        ];
+        let arrow_schema = build_test_arrow_schema();
+        let record_batch = RecordBatch::try_new(arrow_schema, columns).unwrap();
+
+        // Create override sequence array with custom values
+        let override_sequence: u64 = 12345;
+        let override_sequence_array: ArrayRef =
+            Arc::new(UInt64Array::from_value(override_sequence, 4));
+
+        let mut batches = VecDeque::new();
+        read_format
+            .convert_record_batch(&record_batch, Some(&override_sequence_array), &mut batches)
+            .unwrap();
+
+        // Create expected batches with override sequence
+        let expected_batch1 = new_batch_with_sequence(b"one", 1, 1, 2, override_sequence);
+        let expected_batch2 = new_batch_with_sequence(b"two", 11, 10, 2, override_sequence);
+
+        assert_eq!(
+            vec![expected_batch1, expected_batch2],
+            batches.into_iter().collect::<Vec<_>>(),
+        );
+    }
 }

src/mito2/src/sst/parquet/reader.rs

@@ -23,6 +23,7 @@ use async_trait::async_trait;
 use common_recordbatch::filter::SimpleFilterEvaluator;
 use common_telemetry::{debug, warn};
 use datafusion_expr::Expr;
+use datatypes::arrow::array::ArrayRef;
 use datatypes::arrow::error::ArrowError;
 use datatypes::arrow::record_batch::RecordBatch;
 use datatypes::data_type::ConcreteDataType;
@@ -54,7 +55,7 @@ use crate::sst::index::bloom_filter::applier::BloomFilterIndexApplierRef;
 use crate::sst::index::fulltext_index::applier::FulltextIndexApplierRef;
 use crate::sst::index::inverted_index::applier::InvertedIndexApplierRef;
 use crate::sst::parquet::file_range::{FileRangeContext, FileRangeContextRef};
-use crate::sst::parquet::format::ReadFormat;
+use crate::sst::parquet::format::{need_override_sequence, ReadFormat};
 use crate::sst::parquet::metadata::MetadataLoader;
 use crate::sst::parquet::row_group::InMemoryRowGroup;
 use crate::sst::parquet::row_selection::RowGroupSelection;
@@ -220,7 +221,7 @@ impl ParquetReaderBuilder {
         let key_value_meta = parquet_meta.file_metadata().key_value_metadata();
         // Gets the metadata stored in the SST.
         let region_meta = Arc::new(Self::get_region_metadata(&file_path, key_value_meta)?);
-        let read_format = if let Some(column_ids) = &self.projection {
+        let mut read_format = if let Some(column_ids) = &self.projection {
             ReadFormat::new(region_meta.clone(), column_ids.iter().copied())
         } else {
             // Lists all column ids to read, we always use the expected metadata if possible.
@@ -233,6 +234,10 @@ impl ParquetReaderBuilder {
                     .map(|col| col.column_id),
             )
         };
+        if need_override_sequence(&parquet_meta) {
+            read_format
+                .set_override_sequence(self.file_handle.meta_ref().sequence.map(|x| x.get()));
+        }
 
         // Computes the projection mask.
         let parquet_schema_desc = parquet_meta.file_metadata().schema_descr();
@@ -1230,12 +1235,7 @@ pub(crate) type RowGroupReader = RowGroupReaderBase<FileRangeContextRef>;
 impl RowGroupReader {
     /// Creates a new reader from file range.
     pub(crate) fn new(context: FileRangeContextRef, reader: ParquetRecordBatchReader) -> Self {
-        Self {
-            context,
-            reader,
-            batches: VecDeque::new(),
-            metrics: ReaderMetrics::default(),
-        }
+        Self::create(context, reader)
     }
 }
 
@@ -1249,6 +1249,8 @@ pub(crate) struct RowGroupReaderBase<T> {
     batches: VecDeque<Batch>,
     /// Local scan metrics.
     metrics: ReaderMetrics,
+    /// Cached sequence array to override sequences.
+    override_sequence: Option<ArrayRef>,
 }
 
 impl<T> RowGroupReaderBase<T>
@@ -1257,11 +1259,16 @@ where
 {
     /// Creates a new reader.
     pub(crate) fn create(context: T, reader: ParquetRecordBatchReader) -> Self {
+        // The batch length from the reader should be less than or equal to DEFAULT_READ_BATCH_SIZE.
+        let override_sequence = context
+            .read_format()
+            .new_override_sequence_array(DEFAULT_READ_BATCH_SIZE);
         Self {
             context,
             reader,
             batches: VecDeque::new(),
             metrics: ReaderMetrics::default(),
+            override_sequence,
         }
     }
 
@@ -1297,9 +1304,11 @@ where
             };
             self.metrics.num_record_batches += 1;
 
-            self.context
-                .read_format()
-                .convert_record_batch(&record_batch, &mut self.batches)?;
+            self.context.read_format().convert_record_batch(
+                &record_batch,
+                self.override_sequence.as_ref(),
+                &mut self.batches,
+            )?;
             self.metrics.num_batches += self.batches.len();
         }
         let batch = self.batches.pop_front();

src/mito2/src/test_util.rs

@@ -72,7 +72,7 @@ use crate::error::Result;
 use crate::flush::{WriteBufferManager, WriteBufferManagerRef};
 use crate::manifest::manager::{RegionManifestManager, RegionManifestOptions};
 use crate::read::{Batch, BatchBuilder, BatchReader};
-use crate::sst::file_purger::{FilePurger, FilePurgerRef, NoopFilePurger, PurgeRequest};
+use crate::sst::file_purger::{FilePurgerRef, NoopFilePurger};
 use crate::sst::index::intermediate::IntermediateManager;
 use crate::sst::index::puffin_manager::PuffinManagerFactory;
 use crate::time_provider::{StdTimeProvider, TimeProviderRef};

src/mito2/src/test_util/sst_util.rs

@@ -138,11 +138,17 @@ pub fn sst_file_handle(start_ms: i64, end_ms: i64) -> FileHandle {
     sst_file_handle_with_file_id(FileId::random(), start_ms, end_ms)
 }
 
-pub fn new_batch_by_range(tags: &[&str], start: usize, end: usize) -> Batch {
+/// Creates a new batch with custom sequence for testing.
+pub fn new_batch_with_custom_sequence(
+    tags: &[&str],
+    start: usize,
+    end: usize,
+    sequence: u64,
+) -> Batch {
     assert!(end >= start);
     let pk = new_primary_key(tags);
     let timestamps: Vec<_> = (start..end).map(|v| v as i64).collect();
-    let sequences = vec![1000; end - start];
+    let sequences = vec![sequence; end - start];
     let op_types = vec![OpType::Put; end - start];
     let field: Vec<_> = (start..end).map(|v| v as u64).collect();
     new_batch_builder(&pk, &timestamps, &sequences, &op_types, 2, &field)
@@ -150,6 +156,10 @@ pub fn new_batch_by_range(tags: &[&str], start: usize, end: usize) -> Batch {
         .unwrap()
 }
 
+pub fn new_batch_by_range(tags: &[&str], start: usize, end: usize) -> Batch {
+    new_batch_with_custom_sequence(tags, start, end, 1000)
+}
+
 pub fn new_batch_with_binary(tags: &[&str], start: usize, end: usize) -> Batch {
     assert!(end >= start);
     let pk = new_primary_key(tags);

src/operator/Cargo.toml

@@ -35,6 +35,7 @@ common-meta.workspace = true
 common-query.workspace = true
 common-recordbatch.workspace = true
 common-runtime.workspace = true
+common-sql.workspace = true
 common-telemetry.workspace = true
 common-time.workspace = true
 datafusion.workspace = true
@@ -68,6 +69,7 @@ table.workspace = true
 tokio.workspace = true
 tokio-util.workspace = true
 tonic.workspace = true
+tracing.workspace = true
 
 [dev-dependencies]
 common-meta = { workspace = true, features = ["testing"] }

src/operator/src/error.rs

@@ -844,6 +844,13 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display("Sql common error"))]
+    SqlCommon {
+        source: common_sql::error::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -972,6 +979,7 @@ impl ErrorExt for Error {
             Error::InvalidProcessId { .. } => StatusCode::InvalidArguments,
             Error::ProcessManagerMissing { .. } => StatusCode::Unexpected,
             Error::PathNotFound { .. } => StatusCode::InvalidArguments,
+            Error::SqlCommon { source, .. } => source.status_code(),
         }
     }
 

src/operator/src/req_convert/insert/stmt_to_region.rs

@@ -22,7 +22,6 @@ use datatypes::schema::{ColumnSchema, SchemaRef};
 use partition::manager::PartitionRuleManager;
 use session::context::{QueryContext, QueryContextRef};
 use snafu::{ensure, OptionExt, ResultExt};
-use sql::statements;
 use sql::statements::insert::Insert;
 use sqlparser::ast::{ObjectName, Value as SqlValue};
 use table::metadata::TableInfoRef;
@@ -227,7 +226,7 @@ fn sql_value_to_grpc_value(
             column: column.clone(),
         })?
     } else {
-        statements::sql_value_to_value(
+        common_sql::convert::sql_value_to_value(
             column,
             &column_schema.data_type,
             sql_val,
@@ -235,7 +234,7 @@ fn sql_value_to_grpc_value(
             None,
             auto_string_to_numeric,
         )
-        .context(ParseSqlSnafu)?
+        .context(crate::error::SqlCommonSnafu)?
     };
 
     let grpc_value = value_to_grpc_value(value);

src/operator/src/statement/admin.rs

@@ -19,6 +19,7 @@ use common_function::function_registry::FUNCTION_REGISTRY;
 use common_query::prelude::TypeSignature;
 use common_query::Output;
 use common_recordbatch::{RecordBatch, RecordBatches};
+use common_sql::convert::sql_value_to_value;
 use common_telemetry::tracing;
 use common_time::Timezone;
 use datatypes::data_type::DataType;
@@ -30,7 +31,6 @@ use session::context::QueryContextRef;
 use snafu::{ensure, OptionExt, ResultExt};
 use sql::ast::{Expr, FunctionArg, FunctionArgExpr, FunctionArguments, Value as SqlValue};
 use sql::statements::admin::Admin;
-use sql::statements::sql_value_to_value;
 
 use crate::error::{self, Result};
 use crate::statement::StatementExecutor;
@@ -186,7 +186,7 @@ fn values_to_vectors_by_exact_types(
         .zip(exact_types.iter())
         .map(|(value, data_type)| {
             let value = sql_value_to_value(DUMMY_COLUMN, data_type, value, tz, None, false)
-                .context(error::ParseSqlValueSnafu)?;
+                .context(error::SqlCommonSnafu)?;
 
             Ok(value_to_vector(value))
         })

src/operator/src/statement/ddl.rs

@@ -45,6 +45,7 @@ use common_meta::rpc::ddl::{
 };
 use common_meta::rpc::router::{Partition, Partition as MetaPartition};
 use common_query::Output;
+use common_sql::convert::sql_value_to_value;
 use common_telemetry::{debug, info, tracing, warn};
 use common_time::Timezone;
 use datafusion_common::tree_node::TreeNodeVisitor;
@@ -71,7 +72,6 @@ use sql::statements::create::trigger::CreateTrigger;
 use sql::statements::create::{
     CreateExternalTable, CreateFlow, CreateTable, CreateTableLike, CreateView, Partitions,
 };
-use sql::statements::sql_value_to_value;
 use sql::statements::statement::Statement;
 use sqlparser::ast::{Expr, Ident, UnaryOperator, Value as ParserValue};
 use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME};
@@ -87,10 +87,10 @@ use crate::error::{
     ColumnNotFoundSnafu, ConvertSchemaSnafu, CreateLogicalTablesSnafu, CreateTableInfoSnafu,
     DeserializePartitionSnafu, EmptyDdlExprSnafu, ExternalSnafu, ExtractTableNamesSnafu,
     FlowNotFoundSnafu, InvalidPartitionRuleSnafu, InvalidPartitionSnafu, InvalidSqlSnafu,
-    InvalidTableNameSnafu, InvalidViewNameSnafu, InvalidViewStmtSnafu, ParseSqlValueSnafu, Result,
-    SchemaInUseSnafu, SchemaNotFoundSnafu, SchemaReadOnlySnafu, SubstraitCodecSnafu,
-    TableAlreadyExistsSnafu, TableMetadataManagerSnafu, TableNotFoundSnafu,
-    UnrecognizedTableOptionSnafu, ViewAlreadyExistsSnafu,
+    InvalidTableNameSnafu, InvalidViewNameSnafu, InvalidViewStmtSnafu, Result, SchemaInUseSnafu,
+    SchemaNotFoundSnafu, SchemaReadOnlySnafu, SubstraitCodecSnafu, TableAlreadyExistsSnafu,
+    TableMetadataManagerSnafu, TableNotFoundSnafu, UnrecognizedTableOptionSnafu,
+    ViewAlreadyExistsSnafu,
 };
 use crate::expr_helper;
 use crate::statement::show::create_partitions_stmt;
@@ -654,7 +654,7 @@ impl StatementExecutor {
             ctx.clone(),
         )?;
 
-        //TODO(dennis): validate the logical plan
+        // TODO(dennis): validate the logical plan
         self.create_view_by_expr(expr, ctx).await
     }
 
@@ -1389,12 +1389,11 @@ impl StatementExecutor {
     async fn create_table_procedure(
         &self,
         create_table: CreateTableExpr,
-        partitions: Vec<Partition>,
+        partitions: Option<Partition>,
         table_info: RawTableInfo,
         query_context: QueryContextRef,
     ) -> Result<SubmitDdlTaskResponse> {
-        let partitions = partitions.into_iter().map(Into::into).collect();
-
+        let partitions = partitions.map(|p| p.into()); // to PbPartition
         let request = SubmitDdlTaskRequest {
             query_context,
             task: DdlTask::new_create_table(create_table, partitions, table_info),
@@ -1590,7 +1589,7 @@ fn parse_partitions(
     create_table: &CreateTableExpr,
     partitions: Option<Partitions>,
     query_ctx: &QueryContextRef,
-) -> Result<(Vec<MetaPartition>, Vec<String>)> {
+) -> Result<(Option<MetaPartition>, Vec<String>)> {
     // If partitions are not defined by user, use the timestamp column (which has to be existed) as
     // the partition column, and create only one partition.
     let partition_columns = find_partition_columns(&partitions)?;
@@ -1600,23 +1599,26 @@ fn parse_partitions(
     // Validates partition
     let mut exprs = vec![];
     for partition in &partition_entries {
-        for bound in partition {
-            if let PartitionBound::Expr(expr) = bound {
-                exprs.push(expr.clone());
-            }
+        if let PartitionBound::Expr(expr) = partition {
+            exprs.push(expr.clone());
         }
     }
     MultiDimPartitionRule::try_new(partition_columns.clone(), vec![], exprs, true)
         .context(InvalidPartitionSnafu)?;
 
-    Ok((
-        partition_entries
-            .into_iter()
-            .map(|x| MetaPartition::try_from(PartitionDef::new(partition_columns.clone(), x)))
-            .collect::<std::result::Result<_, _>>()
+    let meta_partition = if partition_entries.is_empty() {
+        None
+    } else {
+        Some(
+            MetaPartition::try_from(PartitionDef::new(
+                partition_columns.clone(),
+                partition_entries,
+            ))
             .context(DeserializePartitionSnafu)?,
-        partition_columns,
-    ))
+        )
+    };
+
+    Ok((meta_partition, partition_columns))
 }
 
 fn create_table_info(
@@ -1727,7 +1729,7 @@ fn find_partition_entries(
     partitions: &Option<Partitions>,
     partition_columns: &[String],
     query_ctx: &QueryContextRef,
-) -> Result<Vec<Vec<PartitionBound>>> {
+) -> Result<Vec<PartitionBound>> {
     let entries = if let Some(partitions) = partitions {
         // extract concrete data type of partition columns
         let column_defs = partition_columns
@@ -1756,17 +1758,17 @@ fn find_partition_entries(
         for partition in &partitions.exprs {
             let partition_expr =
                 convert_one_expr(partition, &column_name_and_type, &query_ctx.timezone())?;
-            partition_exprs.push(vec![PartitionBound::Expr(partition_expr)]);
+            partition_exprs.push(PartitionBound::Expr(partition_expr));
         }
 
         // fallback for no expr
         if partition_exprs.is_empty() {
-            partition_exprs.push(vec![PartitionBound::MaxValue]);
+            partition_exprs.push(PartitionBound::MaxValue);
         }
 
         partition_exprs
     } else {
-        vec![vec![PartitionBound::MaxValue]]
+        vec![PartitionBound::MaxValue]
     };
     Ok(entries)
 }
@@ -1859,7 +1861,7 @@ fn convert_value(
         unary_op,
         false,
     )
-    .context(ParseSqlValueSnafu)
+    .context(error::SqlCommonSnafu)
 }
 
 #[cfg(test)]

src/partition/src/partition.rs

@@ -48,17 +48,21 @@ pub trait PartitionRule: Sync + Send {
     ) -> Result<HashMap<RegionNumber, RegionMask>>;
 }
 
-/// The right bound(exclusive) of partition range.
+/// The bound of one partition.
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
 pub enum PartitionBound {
+    /// Deprecated since 0.9.0.
     Value(Value),
+    /// Deprecated since 0.15.0.
     MaxValue,
     Expr(crate::expr::PartitionExpr),
 }
 
+/// The partition definition of one table.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct PartitionDef {
     partition_columns: Vec<String>,
+    /// Each element represents one partition.
     partition_bounds: Vec<PartitionBound>,
 }
 

src/pipeline/Cargo.toml

@@ -47,6 +47,7 @@ lazy_static.workspace = true
 moka = { workspace = true, features = ["sync"] }
 once_cell.workspace = true
 operator.workspace = true
+ordered-float.workspace = true
 paste.workspace = true
 prometheus.workspace = true
 query.workspace = true
@@ -59,7 +60,7 @@ sql.workspace = true
 table.workspace = true
 tokio.workspace = true
 urlencoding = "2.1"
-vrl = "0.24"
+vrl.workspace = true
 yaml-rust = "0.4"
 
 [dev-dependencies]

src/pipeline/benches/processor.rs

@@ -16,23 +16,21 @@ use std::sync::Arc;
 
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
 use pipeline::error::Result;
-use pipeline::{
-    json_to_map, parse, setup_pipeline, Content, Pipeline, PipelineContext, SchemaInfo,
-};
-use serde_json::{Deserializer, Value};
+use pipeline::{parse, setup_pipeline, Content, Pipeline, PipelineContext, SchemaInfo};
+use serde_json::Deserializer;
+use vrl::value::Value as VrlValue;
 
 fn processor_mut(
     pipeline: Arc<Pipeline>,
     pipeline_ctx: &PipelineContext<'_>,
     schema_info: &mut SchemaInfo,
-    input_values: Vec<Value>,
+    input_values: Vec<VrlValue>,
 ) -> Result<Vec<greptime_proto::v1::Row>> {
     let mut result = Vec::with_capacity(input_values.len());
 
     for v in input_values {
-        let payload = json_to_map(v).unwrap();
         let r = pipeline
-            .exec_mut(payload, pipeline_ctx, schema_info)?
+            .exec_mut(v, pipeline_ctx, schema_info)?
             .into_transformed()
             .expect("expect transformed result ");
         result.push(r.0);
@@ -237,7 +235,7 @@ transform:
 fn criterion_benchmark(c: &mut Criterion) {
     let input_value_str = include_str!("./data.log");
     let input_value = Deserializer::from_str(input_value_str)
-        .into_iter::<serde_json::Value>()
+        .into_iter::<VrlValue>()
         .collect::<std::result::Result<Vec<_>, _>>()
         .unwrap();
     let pipeline = prepare_pipeline();