fix: deserialize parquet error when stream's base table modify column type (#18828)

zhyass · web-flow · commit fa8ef002fcd6 · 2025-10-16T11:17:17.000+08:00
* fix

* add test

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix test

* fix

* fix
diff --git a/src/query/expression/src/schema.rs b/src/query/expression/src/schema.rs
@@ -647,6 +647,13 @@ impl TableSchema {
         Ok(i)
     }
 
+    pub fn drop_column_unchecked(&mut self, column: &str) -> Result<FieldIndex> {
+        let i = self.index_of(column)?;
+        self.fields.remove(i);
+
+        Ok(i)
+    }
+
     pub fn to_leaf_column_id_set(&self) -> HashSet<ColumnId> {
         HashSet::from_iter(self.to_leaf_column_ids().iter().cloned())
     }
diff --git a/src/query/service/src/interpreters/interpreter_table_drop.rs b/src/query/service/src/interpreters/interpreter_table_drop.rs
@@ -93,7 +93,7 @@ impl Interpreter for DropTableInterpreter {
         let engine = tbl.get_table_info().engine();
         if matches!(engine, VIEW_ENGINE | STREAM_ENGINE) {
             return Err(ErrorCode::TableEngineNotSupported(format!(
-                "{}.{} engine is {} that doesn't support drop, use `DROP {} {}.{}` instead",
+                "{}.{} engine is {}, use `DROP {} {}.{}` instead",
                 &self.plan.database,
                 &self.plan.table,
                 engine,
diff --git a/src/query/service/src/interpreters/interpreter_table_modify_column.rs b/src/query/service/src/interpreters/interpreter_table_modify_column.rs
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashSet;
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use databend_common_catalog::catalog::Catalog;
@@ -151,28 +151,28 @@ impl ModifyTableColumnInterpreter {
         let schema = table.schema().as_ref().clone();
         let table_info = table.get_table_info();
         let mut new_schema = schema.clone();
-        let mut default_expr_binder = DefaultExprBinder::try_new(self.ctx.clone())?;
         // first check default expr before lock table
         for (field, _comment) in field_and_comments {
             if let Some((i, old_field)) = schema.column_with_name(&field.name) {
                 // if the field has different leaf column numbers, we need drop the old column
                 // and add a new one to generate new column id. otherwise, leaf column ids will conflict.
                 if old_field.data_type.num_leaf_columns() != field.data_type.num_leaf_columns() {
-                    let _ = new_schema.drop_column(&field.name);
-                    let _ = new_schema.add_column(field, i);
+                    let _ = new_schema.drop_column_unchecked(&field.name)?;
+                    new_schema.add_column(field, i)?;
                 } else {
                     // new field don't have `column_id`, assign field directly will cause `column_id` lost.
                     new_schema.fields[i].data_type = field.data_type.clone();
                     // TODO: support set computed field.
                     new_schema.fields[i].computed_expr = field.computed_expr.clone();
                 }
+
                 if let Some(default_expr) = &field.default_expr {
                     let default_expr = default_expr.to_string();
                     new_schema.fields[i].default_expr = Some(default_expr);
-                    let _ = default_expr_binder.get_scalar(&new_schema.fields[i])?;
                 } else {
                     new_schema.fields[i].default_expr = None;
                 }
+
                 if old_field.data_type != field.data_type {
                     // Check if this column is referenced by computed columns.
                     let data_schema = DataSchema::from(&new_schema);
@@ -256,11 +256,12 @@ impl ModifyTableColumnInterpreter {
             return Ok(PipelineBuildResult::create());
         }
 
-        let mut modified_field_indices = HashSet::new();
+        let mut modified_default_scalars = HashMap::new();
+        let mut default_expr_binder = DefaultExprBinder::try_new(self.ctx.clone())?;
         let new_schema_without_computed_fields = new_schema.remove_computed_fields();
+        let format_as_parquet = fuse_table.storage_format_as_parquet();
         if schema != new_schema {
             for (field, _) in field_and_comments {
-                let field_index = new_schema_without_computed_fields.index_of(&field.name)?;
                 let old_field = schema.field_with_name(&field.name)?;
                 let is_alter_column_string_to_binary =
                     is_string_to_binary(&old_field.data_type, &field.data_type);
@@ -269,20 +270,25 @@ impl ModifyTableColumnInterpreter {
                 // 1. alter column from string to binary in parquet or data type not changed.
                 // 2. default expr and computed expr not changed. Otherwise, we need fill value for
                 //    new added column.
-                if ((table.storage_format_as_parquet() && is_alter_column_string_to_binary)
+                if ((format_as_parquet && is_alter_column_string_to_binary)
                     || old_field.data_type == field.data_type)
                     && old_field.default_expr == field.default_expr
                     && old_field.computed_expr == field.computed_expr
                 {
                     continue;
                 }
-                modified_field_indices.insert(field_index);
+                let field_index = new_schema_without_computed_fields.index_of(&field.name)?;
+                let default_scalar = default_expr_binder
+                    .get_scalar(&new_schema_without_computed_fields.fields[field_index])?;
+                modified_default_scalars.insert(field_index, default_scalar);
             }
             table_info.meta.schema = new_schema.clone().into();
         }
 
         // if don't need rebuild table, only update table meta.
-        if modified_field_indices.is_empty() {
+        if modified_default_scalars.is_empty()
+            || base_snapshot.is_none_or(|v| v.summary.row_count == 0)
+        {
             commit_table_meta(
                 &self.ctx,
                 table.as_ref(),
@@ -295,14 +301,24 @@ impl ModifyTableColumnInterpreter {
             return Ok(PipelineBuildResult::create());
         }
 
+        if fuse_table.change_tracking_enabled() {
+            // Modifying columns while change tracking is active may break
+            // the consistency between tracked changes and the current table schema,
+            // leading to incorrect or incomplete change records.
+            return Err(ErrorCode::AlterTableError(format!(
+                "table {} has change tracking enabled, modifying columns should be avoided",
+                table_info.desc
+            )));
+        }
+
         // construct sql for selecting data from old table.
         // computed columns are ignored, as it is build from other columns.
         let query_fields = new_schema_without_computed_fields
             .fields()
             .iter()
             .enumerate()
             .map(|(index, field)| {
-                if modified_field_indices.contains(&index) {
+                if let Some(default_scalar) = modified_default_scalars.get(&index) {
                     let old_field = schema.field_with_name(&field.name).unwrap();
                     let need_remove_nullable =
                         old_field.data_type.is_nullable() && !field.data_type.is_nullable();
@@ -427,7 +443,13 @@ impl ModifyTableColumnInterpreter {
                         }
                         (_, _) => {
                             if need_remove_nullable {
-                                format!("remove_nullable(`{}`)", field.name)
+                                // If the column is being changed from NULLABLE to NOT NULL,
+                                // wrap it with `coalesce()` to replace NULL values with the default,
+                                // and `remove_nullable()` to mark the resulting expression as non-nullable.
+                                format!(
+                                    "remove_nullable(coalesce(`{}`, {}))",
+                                    field.name, default_scalar
+                                )
                             } else {
                                 format!("`{}`", field.name)
                             }
diff --git a/tests/sqllogictests/suites/ee/06_ee_stream/06_0009_stream_issue_18827.test b/tests/sqllogictests/suites/ee/06_ee_stream/06_0009_stream_issue_18827.test
@@ -0,0 +1,56 @@
+## Copyright 2023 Databend Cloud
+##
+## Licensed under the Elastic 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
+##
+##     https://www.elastic.co/licensing/elastic-license
+##
+## 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.
+
+statement ok
+create or replace table t_18827(a string, b int not null default 1);
+
+query I
+insert into t_18827 values('a', 1),('b', 2);
+----
+2
+
+statement ok
+create or replace stream s_18827 on table t_18827 append_only=false;
+
+statement ok
+create or replace stream s1_18827 on table t_18827 append_only=true;
+
+statement error 1132
+alter table t_18827 modify column b float64;
+
+statement ok
+alter table t_18827 modify column a binary;
+
+query T
+select a, b from t_18827;
+----
+61 1
+62 2
+
+query T
+select a, b, change$action, change$is_update from s1_18827;
+----
+
+query T
+select a, b, change$action, change$is_update from s_18827 order by change$action, a;
+----
+
+statement ok
+drop stream s_18827;
+
+statement ok
+drop stream s1_18827;
+
+statement ok
+drop table t_18827 all;
diff --git a/tests/suites/0_stateless/17_altertable/17_0005_alter_table_modify_column_type.sh b/tests/suites/0_stateless/17_altertable/17_0005_alter_table_modify_column_type.sh
@@ -61,7 +61,7 @@ echo "SELECT a,b from test_modify_column_type.f order by b"  | $BENDSQL_CLIENT_C
 echo "DESC test_modify_column_type.f"  | $BENDSQL_CLIENT_CONNECT
 
 echo "begin test modify column NULL to not NULL"
-echo "CREATE TABLE test_modify_column_type.g(a STRING NULL, b INT NULL) STORAGE_FORMAT='native'"  | $BENDSQL_CLIENT_CONNECT
+echo "CREATE TABLE test_modify_column_type.g(a STRING NULL, b INT NULL)"  | $BENDSQL_CLIENT_CONNECT
 echo "INSERT INTO test_modify_column_type.g VALUES('a',1),('b',NULL),(NULL,3),('d',4)"  | $BENDSQL_CLIENT_CONNECT
 echo "SELECT a,b from test_modify_column_type.g" | $BENDSQL_CLIENT_CONNECT
 echo "ALTER TABLE test_modify_column_type.g MODIFY COLUMN a STRING NOT NULL"  | $BENDSQL_CLIENT_CONNECT

Original file line number	Diff line number	Diff line change
`@@ -647,6 +647,13 @@ impl TableSchema {`
`647`	`647`	`Ok(i)`
`648`	`648`	`}`
`649`	`649`
	`650`	`+ pub fn drop_column_unchecked(&mut self, column: &str) -> Result<FieldIndex> {`
	`651`	`+ let i = self.index_of(column)?;`
	`652`	`+ self.fields.remove(i);`
	`653`	`+`
	`654`	`+ Ok(i)`
	`655`	`+ }`
	`656`	`+`
`650`	`657`	`pub fn to_leaf_column_id_set(&self) -> HashSet<ColumnId> {`
`651`	`658`	`HashSet::from_iter(self.to_leaf_column_ids().iter().cloned())`
`652`	`659`	`}`