feat: support alter partition table

Cargo.toml

@@ -92,7 +92,7 @@ bytes = "1"
 bytes_ext = { path = "components/bytes_ext" }
 catalog = { path = "catalog" }
 catalog_impls = { path = "catalog_impls" }
-ceresdbproto = "1.0.16"
+ceresdbproto = { git = "https://github.com/chunshao90/ceresdbproto.git", rev = "a95b6fe9ab988869456eea7f9dbd3154a673b0c1" }
 codec = { path = "components/codec" }
 notifier = { path = "components/notifier" }
 chrono = "0.4"

analytic_engine/src/instance/alter.rs

@@ -72,7 +72,14 @@ impl<'a> Alterer<'a> {
         );
 
         // Validate alter schema request.
-        self.validate_before_alter(&request)?;
+        // if the alter schema request is idempotent, we can skip the alter operation.
+        if self.validate_before_alter(&request)? {
+            info!(
+                "Instance alter schema, idempotent, skip alter, table:{:?}",
+                self.table_data.name
+            );
+            return Ok(());
+        }
 
         // Now we can persist and update the schema, since this function is called by
         // write worker, so there is no other concurrent writer altering the
@@ -157,14 +164,21 @@ impl<'a> Alterer<'a> {
 
     // Most validation should be done by catalog module, so we don't do too much
     // duplicate check here, especially the schema compatibility.
-    fn validate_before_alter(&self, request: &AlterSchemaRequest) -> Result<()> {
+    // Boolean return value indicates whether the alter operation is idempotent.
+    // If it is idempotent, we can skip the alter operation.
+    // true: idempotent, false: not idempotent.
+    fn validate_before_alter(&self, request: &AlterSchemaRequest) -> Result<bool> {
         ensure!(
             !self.table_data.is_dropped(),
             AlterDroppedTable {
                 table: &self.table_data.name,
             }
         );
 
+        if self.table_data.schema().columns() == request.schema.columns() {
+            return Ok(true);
+        }
+
         let current_version = self.table_data.schema_version();
         ensure!(
             current_version < request.schema.version(),
@@ -184,7 +198,7 @@ impl<'a> Alterer<'a> {
             }
         );
 
-        Ok(())
+        Ok(false)
     }
 
     pub async fn alter_options_of_table(

partition_table_engine/src/partition.rs

@@ -24,6 +24,7 @@ use common_types::{
 };
 use futures::{stream::FuturesUnordered, StreamExt};
 use generic_error::BoxError;
+use log::error;
 use snafu::ResultExt;
 use table_engine::{
     partition::{
@@ -36,16 +37,16 @@ use table_engine::{
     },
     remote::{
         model::{
-            ReadRequest as RemoteReadRequest, TableIdentifier, WriteBatchResult,
-            WriteRequest as RemoteWriteRequest,
+            AlterTableOptionsRequest, AlterTableSchemaRequest, ReadRequest as RemoteReadRequest,
+            TableIdentifier, WriteBatchResult, WriteRequest as RemoteWriteRequest,
         },
         RemoteEngineRef,
     },
     stream::{PartitionedStreams, SendableRecordBatchStream},
     table::{
-        AlterSchemaRequest, CreatePartitionRule, FlushRequest, GetRequest, LocatePartitions,
-        ReadRequest, Result, Scan, Table, TableId, TableStats, UnexpectedWithMsg,
-        UnsupportedMethod, Write, WriteBatch, WriteRequest,
+        AlterOptions, AlterSchema, AlterSchemaRequest, CreatePartitionRule, FlushRequest,
+        GetRequest, LocatePartitions, ReadRequest, Result, Scan, Table, TableId, TableStats,
+        UnexpectedWithMsg, UnsupportedMethod, Write, WriteBatch, WriteRequest,
     },
 };
 
@@ -328,20 +329,98 @@ impl Table for PartitionTableImpl {
         Ok(PartitionedStreams { streams })
     }
 
-    async fn alter_schema(&self, _request: AlterSchemaRequest) -> Result<usize> {
-        UnsupportedMethod {
-            table: self.name(),
-            method: "alter_schema",
+    async fn alter_schema(&self, request: AlterSchemaRequest) -> Result<usize> {
+        let partition_num = match self.partition_info() {
+            None => UnexpectedWithMsg {
+                msg: "partition table partition info can't be empty",
+            }
+            .fail()?,
+            Some(partition_info) => partition_info.get_partition_num(),
+        };
+
+        // Alter schema of partitions except the first one.
+        // Because the schema of partition table is stored in the first partition.
+        let mut futures = FuturesUnordered::new();
+        for id in 1..partition_num {
+            let partition = self
+                .remote_engine
+                .alter_table_schema(AlterTableSchemaRequest {
+                    table_ident: self.get_sub_table_ident(id),
+                    table_schema: request.schema.clone(),
+                    pre_schema_version: request.pre_schema_version,
+                });
+            futures.push(partition);
         }
-        .fail()
+
+        let mut rets = Vec::with_capacity(futures.len());
+        while let Some(ret) = futures.next().await {
+            if ret.is_err() {
+                error!("alter schema failed, err:{:?}", ret);
+                rets.push(ret.box_err().context(AlterSchema { table: self.name() }));
+            }
+        }
+        if !rets.is_empty() {
+            rets.remove(0)?;
+        }
+
+        // Alter schema of the first partition.
+        self.remote_engine
+            .alter_table_schema(AlterTableSchemaRequest {
+                table_ident: self.get_sub_table_ident(0),
+                table_schema: request.schema.clone(),
+                pre_schema_version: request.pre_schema_version,
+            })
+            .await
+            .box_err()
+            .context(AlterSchema { table: self.name() })?;
+
+        Ok(0)
     }
 
-    async fn alter_options(&self, _options: HashMap<String, String>) -> Result<usize> {
-        UnsupportedMethod {
-            table: self.name(),
-            method: "alter_options",
+    async fn alter_options(&self, options: HashMap<String, String>) -> Result<usize> {
+        let partition_num = match self.partition_info() {
+            None => UnexpectedWithMsg {
+                msg: "partition table partition info can't be empty",
+            }
+            .fail()?,
+            Some(partition_info) => partition_info.get_partition_num(),
+        };
+
+        // Alter options of partitions except the first one.
+        // Because the schema of partition table is stored in the first partition.
+        let mut futures = FuturesUnordered::new();
+        for id in 1..partition_num {
+            let partition = self
+                .remote_engine
+                .alter_table_options(AlterTableOptionsRequest {
+                    table_ident: self.get_sub_table_ident(id),
+                    options: options.clone(),
+                });
+            futures.push(partition);
         }
-        .fail()
+
+        let mut rets = Vec::with_capacity(futures.len());
+        while let Some(ret) = futures.next().await {
+            if ret.is_err() {
+                error!("alter options failed, err:{:?}", ret);
+                rets.push(ret.box_err().context(AlterOptions { table: self.name() }));
+            }
+        }
+        if !rets.is_empty() {
+            rets.remove(0)?;
+        }
+
+        // Alter options of the first partition.
+        self.remote_engine
+            .alter_table_options(AlterTableOptionsRequest {
+                table_ident: self.get_sub_table_ident(0),
+                options: options.clone(),
+            })
+            .await
+            .box_err()
+            .context(AlterOptions { table: self.name() })?;
+
+        Ok(0)
     }
 
     // Partition table is a virtual table, so it don't need to flush.