feat: basic support of schema evolution in copy for parquet (#19094)

* chore: add comments.

* chore: TableInfo add fn get_option

* add table option ENABLE_SCHEMA_EVOLUTION

* feat: basic support of schema evolution in copy for parquet

* feat: commit txn when table is altered.

Author: youngsofun

src/common/storage/src/stage.rs

@@ -98,7 +98,7 @@ pub fn init_stage_operator(stage_info: &StageInfo) -> Result<Operator> {
 }
 /// select * from @s1/<path> (FILES => <files> PATTERN => <pattern>)
 /// copy from @s1/<path> FILES = <files> PATTERN => <pattern>
-#[derive(serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq, Debug)]
+#[derive(serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq, Debug, Default)]
 pub struct StageFilesInfo {
     pub path: String,
     pub files: Option<Vec<String>>,

src/meta/app/src/principal/user_stage.rs

@@ -402,6 +402,7 @@ pub struct StageParams {
     pub storage: StorageParams,
 }
 
+// todo(yangxiufeng): not used, remove it later.
 #[derive(serde::Serialize, serde::Deserialize, Clone, Default, Debug, Eq, PartialEq)]
 #[serde(default)]
 pub struct CopyOptions {
@@ -428,6 +429,7 @@ pub struct StageInfo {
     // on `COPY INTO xx FROM 's3://xxx?ak=?&sk=?'`, the URL(ExternalLocation) will be treated as an temporary stage.
     pub is_temporary: bool,
     pub file_format_params: FileFormatParams,
+    // todo(yangxiufeng): not used, remove it later.
     pub copy_options: CopyOptions,
     pub comment: String,
     /// TODO(xuanwo): stage doesn't have this info anymore, remove it.

src/meta/app/src/schema/table/mod.rs

@@ -20,6 +20,7 @@ use std::fmt::Display;
 use std::fmt::Formatter;
 use std::ops::Deref;
 use std::ops::Range;
+use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -316,6 +317,13 @@ impl TableInfo {
             .clone()
             .map(|k| (self.meta.cluster_key_seq, k))
     }
+
+    pub fn get_option<T: FromStr>(&self, opt_key: &str, default: T) -> T {
+        self.options()
+            .get(opt_key)
+            .and_then(|s| s.parse::<T>().ok())
+            .unwrap_or(default)
+    }
 }
 
 impl Default for TablePartition {

src/query/catalog/src/plan/datasource/datasource_info/stage.rs

@@ -27,7 +27,9 @@ use databend_common_storage::init_stage_operator;
 use databend_common_storage::StageFileInfo;
 use databend_common_storage::StageFilesInfo;
 
-#[derive(serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+use crate::plan::FullParquetMeta;
+
+#[derive(serde::Serialize, serde::Deserialize, Clone, Default)]
 pub struct StageTableInfo {
     // common
     pub stage_root: String,
@@ -47,8 +49,21 @@ pub struct StageTableInfo {
     pub is_select: bool,
     pub copy_into_table_options: CopyIntoTableOptions,
     pub is_variant: bool,
+
+    // temp work round, when enable_schema_evolution, set it before read partition,
+    // then the StageTableInfo will be dropped, so no need to free it
+    #[serde(skip)]
+    pub parquet_metas: Option<Vec<Arc<FullParquetMeta>>>,
+}
+
+impl PartialEq for StageTableInfo {
+    fn eq(&self, other: &Self) -> bool {
+        self.stage_root == other.stage_root && self.stage_info == other.stage_info
+    }
 }
 
+impl Eq for StageTableInfo {}
+
 impl StageTableInfo {
     pub fn schema(&self) -> Arc<TableSchema> {
         self.schema.clone()

src/query/service/src/interpreters/common/table_option_validation.rs

@@ -47,6 +47,7 @@ use databend_storages_common_table_meta::table::OPT_KEY_COMMENT;
 use databend_storages_common_table_meta::table::OPT_KEY_CONNECTION_NAME;
 use databend_storages_common_table_meta::table::OPT_KEY_DATABASE_ID;
 use databend_storages_common_table_meta::table::OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH;
+use databend_storages_common_table_meta::table::OPT_KEY_ENABLE_SCHEMA_EVOLUTION;
 use databend_storages_common_table_meta::table::OPT_KEY_ENGINE;
 use databend_storages_common_table_meta::table::OPT_KEY_LOCATION;
 use databend_storages_common_table_meta::table::OPT_KEY_RANDOM_MAX_ARRAY_LEN;
@@ -90,6 +91,7 @@ pub static CREATE_FUSE_OPTIONS: LazyLock<HashSet<&'static str>> = LazyLock::new(
     r.insert(OPT_KEY_TEMP_PREFIX);
     r.insert(OPT_KEY_SEGMENT_FORMAT);
     r.insert(OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH);
+    r.insert(OPT_KEY_ENABLE_SCHEMA_EVOLUTION);
     r.insert(FUSE_OPT_KEY_ENABLE_PARQUET_DICTIONARY);
     r
 });

src/query/service/src/interpreters/interpreter_copy_into_table.rs

@@ -13,26 +13,46 @@
 // limitations under the License.
 
 use std::collections::BTreeMap;
+use std::collections::HashMap;
 use std::sync::Arc;
 
+use arrow_schema::Field;
+use databend_common_ast::ast::ColumnMatchMode;
+use databend_common_ast::Span;
 use databend_common_catalog::lock::LockTableOption;
+use databend_common_catalog::plan::StageTableInfo;
 use databend_common_catalog::table::TableExt;
+use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_expression::types::DataType;
 use databend_common_expression::types::Int32Type;
 use databend_common_expression::types::StringType;
 use databend_common_expression::DataBlock;
+use databend_common_expression::DataSchema;
 use databend_common_expression::FromData;
+use databend_common_expression::RemoteDefaultExpr;
+use databend_common_expression::RemoteExpr;
+use databend_common_expression::Scalar;
 use databend_common_expression::SendableDataBlockStream;
+use databend_common_expression::TableDataType;
+use databend_common_expression::TableField;
+use databend_common_expression::TableSchemaRef;
+use databend_common_meta_app::principal::FileFormatParams;
 use databend_common_meta_app::schema::TableInfo;
 use databend_common_meta_app::schema::UpdateStreamMetaReq;
 use databend_common_pipeline::core::Pipeline;
 use databend_common_sql::executor::physical_plans::FragmentKind;
 use databend_common_sql::executor::physical_plans::MutationKind;
 use databend_common_sql::executor::table_read_plan::ToReadDataSourcePlan;
+use databend_common_storage::init_stage_operator;
+use databend_common_storage::parquet::infer_schema_with_extension;
 use databend_common_storage::StageFileInfo;
 use databend_common_storages_fuse::FuseTable;
+use databend_common_storages_parquet::read_metas_in_parallel_for_copy;
 use databend_common_storages_stage::StageTable;
 use databend_storages_common_table_meta::meta::TableMetaTimestamps;
+use databend_storages_common_table_meta::readers::snapshot_reader::TableSnapshotAccessor;
+use itertools::Itertools;
 use log::debug;
 use log::info;
 
@@ -100,10 +120,15 @@ impl CopyIntoTableInterpreter {
     #[async_backtrace::framed]
     pub async fn build_physical_plan(
         &self,
-        table_info: TableInfo,
+        mut table_info: TableInfo,
         plan: &CopyIntoTablePlan,
         table_meta_timestamps: TableMetaTimestamps,
-    ) -> Result<(PhysicalPlan, Vec<UpdateStreamMetaReq>)> {
+    ) -> Result<(
+        PhysicalPlan,
+        Vec<UpdateStreamMetaReq>,
+        Option<TableSchemaRef>,
+    )> {
+        let mut new_schema = None;
         let mut update_stream_meta_reqs = vec![];
         let (source, project_columns) = if let Some(ref query) = plan.query {
             let query = if plan.enable_distributed {
@@ -122,7 +147,14 @@ impl CopyIntoTableInterpreter {
                 Some(result_columns),
             )
         } else {
-            let stage_table = StageTable::try_create(plan.stage_table_info.clone())?;
+            let mut stage_table_info = plan.stage_table_info.clone();
+            if plan.enable_schema_evolution {
+                new_schema = Self::infer_schema(&mut stage_table_info, self.ctx.clone())
+                    .await
+                    .map_err(|e| e.with_context("infer_schema"))?;
+            }
+
+            let stage_table = StageTable::try_create(stage_table_info)?;
 
             let data_source_plan = stage_table
                 .read_plan(self.ctx.clone(), None, None, false, false)
@@ -147,10 +179,19 @@ impl CopyIntoTableInterpreter {
             )
         };
 
+        let mut required_values_schema = plan.required_values_schema.clone();
+        let mut required_source_schema = plan.required_source_schema.clone();
+        if let Some(schema) = &new_schema {
+            table_info.meta.schema = schema.clone();
+            let data_schema: DataSchema = schema.into();
+            required_source_schema = Arc::new(data_schema);
+            required_values_schema = required_source_schema.clone();
+        }
+
         let mut root = PhysicalPlan::new(CopyIntoTable {
-            required_values_schema: plan.required_values_schema.clone(),
+            required_values_schema,
             values_consts: plan.values_consts.clone(),
-            required_source_schema: plan.required_source_schema.clone(),
+            required_source_schema,
             stage_table_info: plan.stage_table_info.clone(),
             table_info,
             write_mode: plan.write_mode,
@@ -176,7 +217,108 @@ impl CopyIntoTableInterpreter {
         let mut next_plan_id = 0;
         root.adjust_plan_id(&mut next_plan_id);
 
-        Ok((root, update_stream_meta_reqs))
+        Ok((root, update_stream_meta_reqs, new_schema))
+    }
+
+    async fn infer_schema(
+        stage_table_info: &mut StageTableInfo,
+        ctx: Arc<dyn TableContext>,
+    ) -> Result<Option<TableSchemaRef>> {
+        #[allow(clippy::single_match)]
+        match &stage_table_info.stage_info.file_format_params {
+            FileFormatParams::Parquet(_) => {
+                let settings = ctx.get_settings();
+                let max_threads = settings.get_max_threads()? as usize;
+                let max_memory_usage = settings.get_max_memory_usage()?;
+
+                let operator = init_stage_operator(&stage_table_info.stage_info)?;
+                // User set the files.
+                let files = stage_table_info.files_to_copy.as_ref().expect(
+                    "ParquetTableForCopy::do_read_partitions must be called with files_to_copy set",
+                );
+                let file_infos = files
+                    .iter()
+                    .filter(|f| f.size > 0)
+                    .map(|f| (f.path.clone(), f.size))
+                    .collect::<Vec<_>>();
+                ctx.set_status_info("[TABLE-SCAN] Infer Parquet Schemas");
+                let metas = read_metas_in_parallel_for_copy(
+                    &operator,
+                    &file_infos,
+                    max_threads,
+                    max_memory_usage,
+                )
+                .await?;
+
+                let case_sensitive = stage_table_info.copy_into_table_options.column_match_mode
+                    == Some(ColumnMatchMode::CaseSensitive);
+
+                let mut new_schema = stage_table_info.schema.as_ref().to_owned();
+                let old_fields: HashMap<String, TableDataType> = stage_table_info
+                    .schema
+                    .fields
+                    .iter()
+                    .map(|f| {
+                        (
+                            if case_sensitive {
+                                f.name.clone()
+                            } else {
+                                f.name.to_lowercase()
+                            },
+                            f.data_type.clone(),
+                        )
+                    })
+                    .collect::<_>();
+                let mut new_fields: HashMap<String, Field> = HashMap::new();
+                for meta in &metas {
+                    let arrow_schema = infer_schema_with_extension(meta.meta.file_metadata())?;
+                    for field in arrow_schema.fields().clone().into_iter() {
+                        let name = if case_sensitive {
+                            field.name().clone()
+                        } else {
+                            field.name().to_lowercase()
+                        };
+                        if !old_fields.contains_key(&name) {
+                            if let Some(f) = new_fields.get_mut(&name) {
+                                if f.data_type() != field.data_type() {
+                                    return Err(ErrorCode::BadBytes(format!(
+                                        "data type of {name} mismatch: {} and {}",
+                                        f.data_type(),
+                                        field.data_type()
+                                    )));
+                                }
+                            } else {
+                                new_fields.insert(name, field.as_ref().clone());
+                            }
+                        }
+                    }
+                }
+
+                stage_table_info.parquet_metas = Some(metas);
+                if new_fields.is_empty() {
+                    return Ok(None);
+                } else {
+                    let new_fields: Vec<_> = new_fields.into_iter().sorted().collect();
+                    for (_, f) in new_fields {
+                        let mut tf: TableField = (&f).try_into()?;
+                        tf.data_type = tf.data_type.wrap_nullable();
+                        if let Some(exprs) = &mut stage_table_info.default_exprs {
+                            exprs.push(RemoteDefaultExpr::RemoteExpr(RemoteExpr::Constant {
+                                scalar: Scalar::Null,
+                                data_type: DataType::Null,
+                                span: Span::default(),
+                            }))
+                        }
+                        new_schema.add_column(&tf, new_schema.num_fields())?;
+                    }
+                    let schema = Arc::new(new_schema);
+                    stage_table_info.schema = schema.clone();
+                    return Ok(Some(schema));
+                }
+            }
+            _ => {}
+        }
+        Ok(None)
     }
 
     fn get_copy_into_table_result(&self) -> Result<Vec<DataBlock>> {
@@ -234,18 +376,39 @@ impl CopyIntoTableInterpreter {
         deduplicated_label: Option<String>,
         path_prefix: Option<String>,
         table_meta_timestamps: TableMetaTimestamps,
+        new_schema: Option<TableSchemaRef>,
     ) -> Result<()> {
         let ctx = self.ctx.clone();
-        let to_table = ctx
+        let mut to_table = ctx
             .get_table(
                 plan.catalog_info.catalog_name(),
                 &plan.database_name,
                 &plan.table_name,
             )
             .await?;
 
+        let mut prev_snapshot_id = None;
+
         // Commit.
         {
+            let mut table_info = to_table.get_table_info().clone();
+            if let Some(new_schema) = new_schema {
+                let fuse_table = FuseTable::try_from_table(to_table.as_ref())?;
+                let base_snapshot = fuse_table.read_table_snapshot().await?;
+                prev_snapshot_id = base_snapshot.snapshot_id().map(|(id, _)| id);
+
+                table_info.meta.fill_field_comments();
+                while table_info.meta.field_comments.len() < new_schema.fields.len() {
+                    table_info.meta.field_comments.push("".to_string());
+                }
+                table_info.meta.schema = new_schema;
+                to_table = FuseTable::create_and_refresh_table_info(
+                    table_info,
+                    ctx.get_settings().get_s3_storage_class()?,
+                )?
+                .into();
+            }
+
             let copied_files_meta_req = PipelineBuilder::build_upsert_copied_files_to_meta_req(
                 ctx.clone(),
                 to_table.as_ref(),
@@ -260,7 +423,7 @@ impl CopyIntoTableInterpreter {
                 copied_files_meta_req,
                 update_stream_meta,
                 plan.write_mode.is_overwrite(),
-                None,
+                prev_snapshot_id,
                 deduplicated_label,
                 table_meta_timestamps,
             )?;
@@ -371,7 +534,7 @@ impl Interpreter for CopyIntoTableInterpreter {
             .ctx
             .get_table_meta_timestamps(to_table.as_ref(), snapshot)?;
 
-        let (physical_plan, update_stream_meta) = self
+        let (physical_plan, update_stream_meta, new_schema) = self
             .build_physical_plan(
                 to_table.get_table_info().clone(),
                 &self.plan,
@@ -403,6 +566,7 @@ impl Interpreter for CopyIntoTableInterpreter {
                 unsafe { self.ctx.get_settings().get_deduplicate_label()? },
                 self.plan.path_prefix.clone(),
                 table_meta_timestamps,
+                new_schema,
             )
             .await?;
         }

src/query/service/src/interpreters/interpreter_replace.rs

@@ -413,7 +413,7 @@ impl ReplaceInterpreter {
                 Plan::CopyIntoTable(copy_plan) => {
                     let interpreter =
                         CopyIntoTableInterpreter::try_create(ctx.clone(), *copy_plan.clone())?;
-                    let (physical_plan, _) = interpreter
+                    let (physical_plan, _, _) = interpreter
                         .build_physical_plan(table_info, &copy_plan, table_meta_timestamps)
                         .await?;
 

src/query/service/src/interpreters/interpreter_table_create.rs

@@ -409,7 +409,7 @@ impl CreateTableInterpreter {
             if settings.get_copy_dedup_full_path_by_default()? {
                 options.insert(
                     OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH.to_string(),
-                    "1".to_string(),
+                    "true".to_string(),
                 );
             };