feat(catalog): support column-level alter table (#370)

Author: TheR1sing3un

crates/integrations/datafusion/src/sql_context.rs

@@ -3182,7 +3182,7 @@ mod tests {
             assert_eq!(identifier.object(), "t1");
             assert_eq!(changes.len(), 1);
             assert!(
-                matches!(&changes[0], SchemaChange::AddColumn { field_name, .. } if field_name == "age")
+                matches!(&changes[0], SchemaChange::AddColumn { field_names, .. } if field_names.first().map(String::as_str) == Some("age"))
             );
         } else {
             panic!("expected AlterTable call");
@@ -3206,10 +3206,10 @@ mod tests {
             assert!(matches!(
                 &changes[0],
                 SchemaChange::AddColumn {
-                    field_name,
+                    field_names,
                     data_type,
                     ..
-                } if field_name == "payload" && matches!(data_type, PaimonDataType::Blob(_))
+                } if field_names.first().map(String::as_str) == Some("payload") && matches!(data_type, PaimonDataType::Blob(_))
             ));
         } else {
             panic!("expected AlterTable call");
@@ -3231,7 +3231,7 @@ mod tests {
         if let CatalogCall::AlterTable { changes, .. } = &calls[0] {
             assert_eq!(changes.len(), 1);
             assert!(
-                matches!(&changes[0], SchemaChange::DropColumn { field_name } if field_name == "age")
+                matches!(&changes[0], SchemaChange::DropColumn { field_names } if field_names.first().map(String::as_str) == Some("age"))
             );
         } else {
             panic!("expected AlterTable call");
@@ -3254,8 +3254,8 @@ mod tests {
             assert_eq!(changes.len(), 1);
             assert!(matches!(
                 &changes[0],
-                SchemaChange::RenameColumn { field_name, new_name }
-                    if field_name == "old_name" && new_name == "new_name"
+                SchemaChange::RenameColumn { field_names, new_name }
+                    if field_names.first().map(String::as_str) == Some("old_name") && new_name == "new_name"
             ));
         } else {
             panic!("expected AlterTable call");

crates/integrations/datafusion/tests/sql_context_tests.rs

@@ -480,23 +480,18 @@ async fn test_alter_table_add_column() {
         .await
         .unwrap();
 
-    // ALTER TABLE is not yet implemented in FileSystemCatalog, so we expect an error
-    let result = sql_context
+    sql_context
         .sql("ALTER TABLE paimon.mydb.alter_test ADD COLUMN age INT")
-        .await;
+        .await
+        .expect("ALTER TABLE ADD COLUMN should succeed");
 
-    // FileSystemCatalog does not support AddColumn schema change yet
-    assert!(
-        result.is_err(),
-        "ALTER TABLE ADD COLUMN should fail because AddColumn is not yet supported"
-    );
-    let err_msg = result.unwrap_err().to_string();
-    assert!(
-        err_msg.contains("not yet implemented")
-            || err_msg.contains("Unsupported")
-            || err_msg.contains("not yet supported"),
-        "Error should indicate alter_table is not implemented, got: {err_msg}"
-    );
+    // The new column is appended to the table schema.
+    let table = catalog
+        .get_table(&Identifier::new("mydb", "alter_test"))
+        .await
+        .unwrap();
+    let names: Vec<&str> = table.schema().fields().iter().map(|f| f.name()).collect();
+    assert_eq!(names, vec!["id", "name", "age"]);
 }
 
 #[tokio::test]

crates/paimon/src/api/api_request.rs

@@ -22,7 +22,10 @@
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 
-use crate::{catalog::Identifier, spec::Schema};
+use crate::{
+    catalog::Identifier,
+    spec::{Schema, SchemaChange},
+};
 
 /// Request to create a new database.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -95,6 +98,23 @@ impl CreateTableRequest {
     }
 }
 
+/// Request to alter a table's schema.
+///
+/// Wire-compatible with Java Paimon's `AlterTableRequest` (`{"changes": [...]}`).
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct AlterTableRequest {
+    /// The ordered list of schema changes to apply.
+    pub changes: Vec<SchemaChange>,
+}
+
+impl AlterTableRequest {
+    /// Create a new AlterTableRequest.
+    pub fn new(changes: Vec<SchemaChange>) -> Self {
+        Self { changes }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/paimon/src/api/mod.rs

@@ -31,7 +31,8 @@ mod api_response;
 
 // Re-export request types
 pub use api_request::{
-    AlterDatabaseRequest, CreateDatabaseRequest, CreateTableRequest, RenameTableRequest,
+    AlterDatabaseRequest, AlterTableRequest, CreateDatabaseRequest, CreateTableRequest,
+    RenameTableRequest,
 };
 
 // Re-export response types

crates/paimon/src/api/rest_api.rs

@@ -25,11 +25,12 @@ use std::collections::HashMap;
 use crate::api::rest_client::HttpClient;
 use crate::catalog::Identifier;
 use crate::common::{CatalogOptions, Options};
-use crate::spec::{Partition, PartitionStatistics, Schema, Snapshot};
+use crate::spec::{Partition, PartitionStatistics, Schema, SchemaChange, Snapshot};
 use crate::Result;
 
 use super::api_request::{
-    AlterDatabaseRequest, CreateDatabaseRequest, CreateTableRequest, RenameTableRequest,
+    AlterDatabaseRequest, AlterTableRequest, CreateDatabaseRequest, CreateTableRequest,
+    RenameTableRequest,
 };
 use super::api_response::{
     ConfigResponse, GetDatabaseResponse, GetTableResponse, ListDatabasesResponse,
@@ -343,6 +344,21 @@ impl RESTApi {
         Ok(())
     }
 
+    /// Alter a table's schema by applying a list of schema changes.
+    pub async fn alter_table(
+        &self,
+        identifier: &Identifier,
+        changes: Vec<SchemaChange>,
+    ) -> Result<()> {
+        let database = identifier.database();
+        let table = identifier.object();
+        validate_non_empty_multi(&[(database, "database name"), (table, "table name")])?;
+        let path = self.resource_paths.table(database, table);
+        let request = AlterTableRequest::new(changes);
+        let _resp: serde_json::Value = self.client.post(&path, &request).await?;
+        Ok(())
+    }
+
     /// Get table information.
     pub async fn get_table(&self, identifier: &Identifier) -> Result<GetTableResponse> {
         let database = identifier.database();