feat(catalog): support column-level alter table

crates/integrations/datafusion/src/sql_context.rs

@@ -3147,7 +3147,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");
@@ -3171,10 +3171,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");
@@ -3196,7 +3196,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");
@@ -3219,8 +3219,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();

crates/paimon/src/catalog/filesystem.rs

@@ -439,11 +439,29 @@ impl Catalog for FileSystemCatalog {
                 full_name: identifier.full_name(),
             })?;
 
-        let new_schema = current.apply_changes(changes)?;
+        let new_schema = current
+            .apply_changes(changes)
+            .map_err(|e| fill_table_name(e, identifier))?;
         self.save_table_schema(&table_path, &new_schema).await
     }
 }
 
+/// `TableSchema::apply_changes` returns column errors without a table name;
+/// fill in the identifier's full name so the message identifies the table.
+fn fill_table_name(err: Error, identifier: &Identifier) -> Error {
+    match err {
+        Error::ColumnNotExist { column, .. } => Error::ColumnNotExist {
+            full_name: identifier.full_name(),
+            column,
+        },
+        Error::ColumnAlreadyExist { column, .. } => Error::ColumnAlreadyExist {
+            full_name: identifier.full_name(),
+            column,
+        },
+        other => other,
+    }
+}
+
 #[cfg(test)]
 #[cfg(not(windows))] // Skip on Windows due to path compatibility issues
 mod tests {
@@ -728,4 +746,201 @@ mod tests {
             );
         }
     }
+
+    use crate::spec::{ColumnMove, DataType, IntType, SchemaChange, VarCharType};
+
+    /// Two-column table (id INT, name VARCHAR) used by the alter-table tests.
+    fn two_column_schema() -> Schema {
+        Schema::builder()
+            .column("id", DataType::Int(IntType::new()))
+            .column("name", DataType::VarChar(VarCharType::string_type()))
+            .build()
+            .unwrap()
+    }
+
+    async fn setup_table(catalog: &FileSystemCatalog, schema: Schema) -> Identifier {
+        catalog
+            .create_database("db", false, HashMap::new())
+            .await
+            .unwrap();
+        let id = Identifier::new("db", "t");
+        catalog.create_table(&id, schema, false).await.unwrap();
+        id
+    }
+
+    #[tokio::test]
+    async fn test_alter_table_column_changes() {
+        let (_tmp, catalog) = create_test_catalog();
+        let id = setup_table(&catalog, two_column_schema()).await;
+
+        // Add a column at the end; it must take highest_field_id + 1.
+        catalog
+            .alter_table(
+                &id,
+                vec![SchemaChange::add_column(
+                    "age".to_string(),
+                    DataType::Int(IntType::new()),
+                )],
+                false,
+            )
+            .await
+            .unwrap();
+        let ts = catalog.get_table(&id).await.unwrap();
+        let ts = ts.schema();
+        let names: Vec<&str> = ts.fields().iter().map(|f| f.name()).collect();
+        assert_eq!(names, vec!["id", "name", "age"]);
+        let age = ts.fields().iter().find(|f| f.name() == "age").unwrap();
+        assert_eq!(age.id(), 2, "new column gets highest_field_id + 1");
+        assert_eq!(ts.id(), 1, "schema version bumped");
+
+        // Add a column moved to the front.
+        catalog
+            .alter_table(
+                &id,
+                vec![SchemaChange::add_column_with_description_and_column_move(
+                    "rowkey".to_string(),
+                    DataType::Int(IntType::new()),
+                    "primary".to_string(),
+                    ColumnMove::move_first("rowkey".to_string()),
+                )],
+                false,
+            )
+            .await
+            .unwrap();
+        let ts = catalog.get_table(&id).await.unwrap();
+        let ts = ts.schema();
+        assert_eq!(ts.fields()[0].name(), "rowkey");
+        assert_eq!(ts.fields()[0].description(), Some("primary"));
+
+        // Rename, update comment, update type, update nullability, drop.
+        catalog
+            .alter_table(
+                &id,
+                vec![
+                    SchemaChange::rename_column("name".to_string(), "full_name".to_string()),
+                    SchemaChange::update_column_comment("id".to_string(), "the id".to_string()),
+                    SchemaChange::update_column_type(
+                        "age".to_string(),
+                        DataType::BigInt(crate::spec::BigIntType::new()),
+                    ),
+                    SchemaChange::update_column_nullability("id".to_string(), false),
+                    SchemaChange::drop_column("rowkey".to_string()),
+                ],
+                false,
+            )
+            .await
+            .unwrap();
+        let ts = catalog.get_table(&id).await.unwrap();
+        let ts = ts.schema();
+        let names: Vec<&str> = ts.fields().iter().map(|f| f.name()).collect();
+        assert_eq!(names, vec!["id", "full_name", "age"]);
+        let id_field = ts.fields().iter().find(|f| f.name() == "id").unwrap();
+        assert_eq!(id_field.description(), Some("the id"));
+        assert!(!id_field.data_type().is_nullable());
+        let age_field = ts.fields().iter().find(|f| f.name() == "age").unwrap();
+        assert!(matches!(age_field.data_type(), DataType::BigInt(_)));
+    }
+
+    #[tokio::test]
+    async fn test_alter_table_reposition_column() {
+        let (_tmp, catalog) = create_test_catalog();
+        let id = setup_table(&catalog, two_column_schema()).await;
+
+        // Move `name` before `id`.
+        catalog
+            .alter_table(
+                &id,
+                vec![SchemaChange::update_column_position(
+                    ColumnMove::move_first("name".to_string()),
+                )],
+                false,
+            )
+            .await
+            .unwrap();
+        let ts = catalog.get_table(&id).await.unwrap();
+        let names: Vec<&str> = ts.schema().fields().iter().map(|f| f.name()).collect();
+        assert_eq!(names, vec!["name", "id"]);
+    }
+
+    #[tokio::test]
+    async fn test_alter_table_errors() {
+        let (_tmp, catalog) = create_test_catalog();
+        let id = setup_table(&catalog, two_column_schema()).await;
+
+        // Add a duplicate column -> ColumnAlreadyExist.
+        let err = catalog
+            .alter_table(
+                &id,
+                vec![SchemaChange::add_column(
+                    "name".to_string(),
+                    DataType::Int(IntType::new()),
+                )],
+                false,
+            )
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(err, Error::ColumnAlreadyExist { .. }),
+            "got {err:?}"
+        );
+
+        // Drop a missing column -> ColumnNotExist.
+        let err = catalog
+            .alter_table(
+                &id,
+                vec![SchemaChange::drop_column("ghost".to_string())],
+                false,
+            )
+            .await
+            .unwrap_err();
+        assert!(matches!(err, Error::ColumnNotExist { .. }), "got {err:?}");
+
+        // Altering a missing table: ignored vs error.
+        let missing = Identifier::new("db", "nope");
+        catalog
+            .alter_table(
+                &missing,
+                vec![SchemaChange::update_column_comment(
+                    "id".to_string(),
+                    "x".to_string(),
+                )],
+                true,
+            )
+            .await
+            .unwrap();
+        let err = catalog
+            .alter_table(
+                &missing,
+                vec![SchemaChange::update_column_comment(
+                    "id".to_string(),
+                    "x".to_string(),
+                )],
+                false,
+            )
+            .await
+            .unwrap_err();
+        assert!(matches!(err, Error::TableNotExist { .. }), "got {err:?}");
+    }
+
+    #[tokio::test]
+    async fn test_alter_table_drop_primary_key_column_rejected() {
+        let (_tmp, catalog) = create_test_catalog();
+        let schema = Schema::builder()
+            .column("id", DataType::Int(IntType::new()))
+            .column("name", DataType::VarChar(VarCharType::string_type()))
+            .primary_key(["id"])
+            .build()
+            .unwrap();
+        let id = setup_table(&catalog, schema).await;
+
+        let err = catalog
+            .alter_table(
+                &id,
+                vec![SchemaChange::drop_column("id".to_string())],
+                false,
+            )
+            .await
+            .unwrap_err();
+        assert!(matches!(err, Error::Unsupported { .. }), "got {err:?}");
+    }
 }