This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 7810abe7 feat(datafusion): support alter column (#625)
7810abe7 is described below
commit 7810abe711e72255aeb20d005b457103f3dcb824
Author: QuakeWang <[email protected]>
AuthorDate: Tue Jul 28 23:12:26 2026 +0800
feat(datafusion): support alter column (#625)
---
crates/integrations/datafusion/src/sql_context.rs | 175 ++++++++++++++++++++-
.../datafusion/tests/sql_context_tests.rs | 47 ++++++
crates/paimon/src/spec/schema.rs | 132 ++++++++++++++++
3 files changed, 349 insertions(+), 5 deletions(-)
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index 3190487b..10532c57 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -28,6 +28,8 @@
//! - `ALTER TABLE db.t ADD COLUMN col TYPE`
//! - `ALTER TABLE db.t DROP COLUMN col`
//! - `ALTER TABLE db.t RENAME COLUMN old TO new`
+//! - `ALTER TABLE db.t ALTER COLUMN col TYPE new_type`
+//! - `ALTER TABLE db.t ALTER COLUMN col SET|DROP NOT NULL`
//! - `ALTER TABLE db.t RENAME TO new_name`
//! - `ALTER TABLE db.t DROP PARTITION (col = val, ...)`
//! - `CREATE VIEW [IF NOT EXISTS] view [(col, ...)] AS query`
@@ -55,11 +57,11 @@ use datafusion::logical_expr::{Expr as LogicalExpr,
LogicalPlan, Volatility};
use datafusion::prelude::{DataFrame, SessionContext};
use datafusion::sql::planner::IdentNormalizer;
use datafusion::sql::sqlparser::ast::{
- AlterTableOperation, BinaryLength, CharacterLength, ColumnDef,
ColumnOption, CreateFunction,
- CreateFunctionBody, CreateTable, CreateTableOptions, CreateView, Delete,
Expr as SqlExpr,
- FromTable, FunctionBehavior, FunctionReturnType, Insert, Merge,
ObjectName, ObjectType,
- RenameTableNameKind, Reset, ResetStatement, Set, ShowCreateObject,
SqlOption, Statement,
- TableFactor, TableObject, Truncate, Update, Value as SqlValue,
+ AlterColumnOperation, AlterTableOperation, BinaryLength, CharacterLength,
ColumnDef,
+ ColumnOption, CreateFunction, CreateFunctionBody, CreateTable,
CreateTableOptions, CreateView,
+ Delete, Expr as SqlExpr, FromTable, FunctionBehavior, FunctionReturnType,
Insert, Merge,
+ ObjectName, ObjectType, RenameTableNameKind, Reset, ResetStatement, Set,
ShowCreateObject,
+ SqlOption, Statement, TableFactor, TableObject, Truncate, Update, Value as
SqlValue,
};
use datafusion::sql::sqlparser::dialect::GenericDialect;
use datafusion::sql::sqlparser::keywords::Keyword;
@@ -1030,6 +1032,20 @@ impl SQLContext {
Self::ensure_main_branch_write_target(name, "ALTER TABLE")?;
let identifier = self.resolve_table_name(name)?;
+ if operations.len() > 1
+ && operations.iter().any(|operation| {
+ matches!(
+ operation,
+ AlterTableOperation::RenameTable { .. }
+ | AlterTableOperation::DropPartitions { .. }
+ )
+ })
+ {
+ return Err(DataFusionError::Plan(
+ "ALTER TABLE RENAME TO and DROP PARTITION must be used
alone".to_string(),
+ ));
+ }
+
let mut changes = Vec::new();
let mut rename_to: Option<Identifier> = None;
@@ -1052,6 +1068,9 @@ impl SQLContext {
new_column_name.value.clone(),
));
}
+ AlterTableOperation::AlterColumn { column_name, op } => {
+
changes.push(alter_column_to_schema_change(&column_name.value, op)?);
+ }
AlterTableOperation::RenameTable { table_name } => {
let new_name = match table_name {
RenameTableNameKind::To(name) |
RenameTableNameKind::As(name) => {
@@ -2380,6 +2399,41 @@ fn column_def_to_add_column(col: &ColumnDef) ->
DFResult<SchemaChange> {
})
}
+fn alter_column_to_schema_change(
+ column_name: &str,
+ operation: &AlterColumnOperation,
+) -> DFResult<SchemaChange> {
+ match operation {
+ AlterColumnOperation::SetNotNull =>
Ok(SchemaChange::update_column_nullability(
+ column_name.to_string(),
+ false,
+ )),
+ AlterColumnOperation::DropNotNull =>
Ok(SchemaChange::update_column_nullability(
+ column_name.to_string(),
+ true,
+ )),
+ AlterColumnOperation::SetDataType {
+ data_type, using, ..
+ } => {
+ if using.is_some() {
+ return Err(DataFusionError::Plan(
+ "ALTER COLUMN TYPE USING is not supported".to_string(),
+ ));
+ }
+ let new_data_type = sql_data_type_to_paimon_type(data_type, true)?;
+ Ok(SchemaChange::UpdateColumnType {
+ field_names: vec![column_name.to_string()],
+ new_data_type,
+ // A type-only SQL change must not change the column's
nullability.
+ keep_nullability: true,
+ })
+ }
+ other => Err(DataFusionError::Plan(format!(
+ "Unsupported ALTER COLUMN operation: {other}"
+ ))),
+ }
+}
+
fn column_def_to_paimon_type(col: &ColumnDef) -> DFResult<PaimonDataType> {
sql_data_type_to_paimon_type(&col.data_type, column_def_nullable(col))
}
@@ -3273,6 +3327,7 @@ mod tests {
use async_trait::async_trait;
use datafusion::arrow::array::StringViewArray;
+ use datafusion::sql::sqlparser::dialect::PostgreSqlDialect;
use paimon::catalog::Database;
use paimon::spec::{
DataField as PaimonDataField, DataType as PaimonDataType, IntType,
Schema as PaimonSchema,
@@ -6180,6 +6235,116 @@ mod tests {
}
}
+ #[tokio::test]
+ async fn test_alter_table_update_column_type_preserves_nullability() {
+ let catalog = Arc::new(MockCatalog::new());
+ let sql_context = make_sql_context(catalog.clone()).await;
+
+ for sql in [
+ "ALTER TABLE mydb.t1 ALTER COLUMN value TYPE BIGINT",
+ "ALTER TABLE mydb.t1 ALTER COLUMN value SET DATA TYPE BIGINT",
+ ] {
+ sql_context.sql(sql).await.unwrap();
+
+ let calls = catalog.take_calls();
+ assert_eq!(calls.len(), 1);
+ if let CatalogCall::AlterTable { changes, .. } = &calls[0] {
+ assert_eq!(changes.len(), 1);
+ assert!(matches!(
+ &changes[0],
+ SchemaChange::UpdateColumnType {
+ field_names,
+ new_data_type: PaimonDataType::BigInt(_),
+ keep_nullability: true,
+ } if field_names.first().map(String::as_str) ==
Some("value")
+ ));
+ } else {
+ panic!("expected AlterTable call");
+ }
+ }
+ }
+
+ #[tokio::test]
+ async fn test_alter_table_update_column_nullability() {
+ let catalog = Arc::new(MockCatalog::new());
+ let sql_context = make_sql_context(catalog.clone()).await;
+
+ for (sql, expected_nullability) in [
+ ("ALTER TABLE mydb.t1 ALTER COLUMN value SET NOT NULL", false),
+ ("ALTER TABLE mydb.t1 ALTER COLUMN value DROP NOT NULL", true),
+ ] {
+ sql_context.sql(sql).await.unwrap();
+
+ let calls = catalog.take_calls();
+ assert_eq!(calls.len(), 1);
+ if let CatalogCall::AlterTable { changes, .. } = &calls[0] {
+ assert_eq!(changes.len(), 1);
+ assert!(matches!(
+ &changes[0],
+ SchemaChange::UpdateColumnNullability {
+ field_names,
+ new_nullability,
+ } if field_names.first().map(String::as_str) ==
Some("value")
+ && *new_nullability == expected_nullability
+ ));
+ } else {
+ panic!("expected AlterTable call");
+ }
+ }
+ }
+
+ #[test]
+ fn test_alter_table_update_column_type_rejects_using() {
+ let statements = Parser::parse_sql(
+ &PostgreSqlDialect {},
+ "ALTER TABLE mydb.t1 ALTER COLUMN value \
+ TYPE BIGINT USING CAST(value AS BIGINT)",
+ )
+ .unwrap();
+ let Statement::AlterTable(alter_table) = &statements[0] else {
+ panic!("expected ALTER TABLE statement");
+ };
+ let AlterTableOperation::AlterColumn { column_name, op } =
&alter_table.operations[0]
+ else {
+ panic!("expected ALTER COLUMN operation");
+ };
+
+ let err = alter_column_to_schema_change(&column_name.value,
op).unwrap_err();
+
+ assert!(err.to_string().contains("USING is not supported"));
+ }
+
+ #[tokio::test]
+ async fn test_alter_table_update_column_default_is_unsupported() {
+ let catalog = Arc::new(MockCatalog::new());
+ let sql_context = make_sql_context(catalog.clone()).await;
+
+ let err = sql_context
+ .sql("ALTER TABLE mydb.t1 ALTER COLUMN value SET DEFAULT 1")
+ .await
+ .unwrap_err();
+
+ assert!(err
+ .to_string()
+ .contains("Unsupported ALTER COLUMN operation"));
+ assert!(catalog.take_calls().is_empty());
+ }
+
+ #[tokio::test]
+ async fn
test_alter_table_rejects_mixed_special_operations_before_catalog_calls() {
+ let catalog = Arc::new(MockCatalog::new());
+ let sql_context = make_sql_context(catalog.clone()).await;
+
+ for sql in [
+ "ALTER TABLE mydb.t1 RENAME TO t2, ADD COLUMN age INT",
+ "ALTER TABLE mydb.t1 DROP PARTITION (pt = 'a'), ADD COLUMN age
INT",
+ ] {
+ let err = sql_context.sql(sql).await.unwrap_err();
+ assert!(err.to_string().contains("must be used alone"));
+ assert!(catalog.take_calls().is_empty());
+ }
+ }
+
#[tokio::test]
async fn test_alter_table_rename_table() {
let catalog = Arc::new(MockCatalog::new());
diff --git a/crates/integrations/datafusion/tests/sql_context_tests.rs
b/crates/integrations/datafusion/tests/sql_context_tests.rs
index d0f1c244..f42acbda 100644
--- a/crates/integrations/datafusion/tests/sql_context_tests.rs
+++ b/crates/integrations/datafusion/tests/sql_context_tests.rs
@@ -1163,6 +1163,53 @@ async fn test_alter_table_add_column() {
assert_eq!(names, vec!["id", "name", "age"]);
}
+#[tokio::test]
+async fn test_alter_table_update_column_type_and_nullability() {
+ let (_tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+ let identifier = Identifier::new("mydb", "alter_column_test");
+
+ catalog
+ .create_database("mydb", false, Default::default())
+ .await
+ .unwrap();
+ let schema = paimon::spec::Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("alter-column-null-to-not-null.disabled", "false")
+ .build()
+ .unwrap();
+ catalog
+ .create_table(&identifier, schema, false)
+ .await
+ .unwrap();
+
+ sql_context
+ .sql("ALTER TABLE mydb.alter_column_test ALTER COLUMN value SET NOT
NULL")
+ .await
+ .expect("ALTER COLUMN SET NOT NULL should succeed");
+ let table = catalog.get_table(&identifier).await.unwrap();
+ assert!(!table.schema().fields()[1].data_type().is_nullable());
+
+ sql_context
+ .sql("ALTER TABLE mydb.alter_column_test ALTER COLUMN value TYPE
BIGINT")
+ .await
+ .expect("ALTER COLUMN TYPE should succeed");
+ let table = catalog.get_table(&identifier).await.unwrap();
+ assert!(matches!(
+ table.schema().fields()[1].data_type(),
+ DataType::BigInt(_)
+ ));
+ assert!(!table.schema().fields()[1].data_type().is_nullable());
+
+ sql_context
+ .sql("ALTER TABLE mydb.alter_column_test ALTER COLUMN value DROP NOT
NULL")
+ .await
+ .expect("ALTER COLUMN DROP NOT NULL should succeed");
+ let table = catalog.get_table(&identifier).await.unwrap();
+ assert!(table.schema().fields()[1].data_type().is_nullable());
+}
+
#[tokio::test]
async fn test_alter_table_rename() {
let (_tmp, catalog) = create_test_env();
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index f96bc0d7..90d74ded 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -255,6 +255,7 @@ impl TableSchema {
message: format!("Cannot rename partition column:
[{name}]"),
});
}
+
assert_not_updating_primary_key_index_column(&self.options, name, "rename")?;
let idx =
field_index(&fields, name).ok_or_else(||
crate::Error::ColumnNotExist {
full_name: full_name.to_string(),
@@ -302,6 +303,7 @@ impl TableSchema {
),
});
}
+
assert_not_updating_primary_key_index_column(&self.options, name, "drop")?;
// Dropping a column referenced by `bucket-key` /
`sequence.field`
// would silently break bucket assignment / sequence
ordering on
// existing data (e.g. `bucket_key_indices` becomes empty
and writes
@@ -355,6 +357,11 @@ impl TableSchema {
message: "Cannot update primary key".to_string(),
});
}
+ assert_not_updating_primary_key_index_column(
+ &self.options,
+ name,
+ "update type of",
+ )?;
let idx =
field_index(&fields, name).ok_or_else(||
crate::Error::ColumnNotExist {
full_name: full_name.to_string(),
@@ -568,6 +575,35 @@ fn assert_nullability_change(
Ok(())
}
+/// Reject destructive changes to columns referenced by a primary-key index.
+///
+/// The index metadata and existing index files are tied to the original column
+/// name and type. Mirrors Java
+/// `SchemaManager.assertNotUpdatingPrimaryKeyIndexColumn`.
+fn assert_not_updating_primary_key_index_column(
+ options: &HashMap<String, String>,
+ field_name: &str,
+ operation: &str,
+) -> crate::Result<()> {
+ let core_options = CoreOptions::new(options);
+ let is_vector_index_column =
core_options.primary_key_vector_index_enabled()
+ && core_options
+ .primary_key_vector_index_columns()?
+ .iter()
+ .any(|column| column == field_name);
+ let is_full_text_index_column = core_options
+ .primary_key_full_text_index_columns()
+ .iter()
+ .any(|column| column == field_name);
+
+ if is_vector_index_column || is_full_text_index_column {
+ return Err(crate::Error::Unsupported {
+ message: format!("Cannot {operation} primary-key index column:
[{field_name}]"),
+ });
+ }
+ Ok(())
+}
+
/// Rename a key in a partition/primary key list, if present.
fn rename_in_keys(keys: &mut [String], old: &str, new: &str) {
for key in keys.iter_mut() {
@@ -2981,6 +3017,102 @@ mod tests {
);
}
+ fn assert_primary_key_index_column_changes_rejected(
+ table_schema: &TableSchema,
+ column_name: &str,
+ new_data_type: DataType,
+ ) {
+ let changes = [
+ (
+ crate::spec::SchemaChange::rename_column(
+ column_name.to_string(),
+ format!("renamed_{column_name}"),
+ ),
+ format!("Cannot rename primary-key index column:
[{column_name}]"),
+ ),
+ (
+
crate::spec::SchemaChange::drop_column(column_name.to_string()),
+ format!("Cannot drop primary-key index column:
[{column_name}]"),
+ ),
+ (
+ crate::spec::SchemaChange::update_column_type(
+ column_name.to_string(),
+ new_data_type,
+ ),
+ format!("Cannot update type of primary-key index column:
[{column_name}]"),
+ ),
+ ];
+
+ for (change, expected_message) in changes {
+ let err = table_schema.apply_changes(vec![change]).unwrap_err();
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message == &expected_message),
+ "expected primary-key index guard, got {err:?}"
+ );
+ }
+ }
+
+ #[test]
+ fn test_rejects_destructive_primary_key_full_text_index_column_changes() {
+ let table_schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("content",
DataType::VarChar(VarCharType::string_type()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("deletion-vectors.enabled", "true")
+ .option("pk-full-text.index.columns", "content")
+ .build()
+ .unwrap(),
+ );
+
+ assert_primary_key_index_column_changes_rejected(
+ &table_schema,
+ "content",
+ DataType::Int(IntType::new()),
+ );
+
+ let err = table_schema
+ .apply_changes(vec![
+
crate::spec::SchemaChange::remove_option("pk-full-text.index.columns".to_string()),
+ crate::spec::SchemaChange::rename_column(
+ "content".to_string(),
+ "renamed_content".to_string(),
+ ),
+ ])
+ .unwrap_err();
+ assert!(matches!(
+ err,
+ crate::Error::Unsupported { ref message }
+ if message == "Cannot rename primary-key index column:
[content]"
+ ));
+ }
+
+ #[test]
+ fn test_rejects_destructive_primary_key_vector_index_column_changes() {
+ let vector_type = DataType::Vector(
+ VectorType::try_new(true, 3,
DataType::Float(FloatType::new())).unwrap(),
+ );
+ let table_schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("embedding", vector_type.clone())
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("deletion-vectors.enabled", "true")
+ .option("pk-vector.index.columns", "embedding")
+ .option("fields.embedding.pk-vector.index.type", "ivf_flat")
+ .option("fields.embedding.pk-vector.distance.metric", "l2")
+ .build()
+ .unwrap(),
+ );
+
+ assert_primary_key_index_column_changes_rejected(&table_schema,
"embedding", vector_type);
+ }
+
#[test]
fn test_create_schema_rejects_unknown_bucket_key() {
let err = Schema::builder()
