Merge pull request #10561 from youngsofun/trans

feat: support copy into table with transform.

Author: mergify[bot]

src/query/ast/src/ast/statements/stage.rs

@@ -90,6 +90,15 @@ pub struct SelectStageOptions {
     pub connection: BTreeMap<String, String>,
 }
 
+impl SelectStageOptions {
+    pub fn is_empty(&self) -> bool {
+        self.files.is_none()
+            && self.pattern.is_none()
+            && self.file_format.is_none()
+            && self.connection.is_empty()
+    }
+}
+
 // SELECT <columns> FROM
 // {@<stage_name>[/<path>] | '<uri>'} [(
 // [ PARTTERN => '<regex_pattern>']

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

@@ -18,6 +18,7 @@ use common_arrow::arrow::datatypes::Schema as ArrowSchema;
 use common_expression::TableSchema;
 use common_meta_app::principal::StageInfo;
 use common_meta_app::schema::TableInfo;
+use common_storage::StageFileInfo;
 use common_storage::StageFilesInfo;
 
 use crate::plan::datasource::datasource_info::parquet_read_options::ParquetReadOptions;
@@ -30,6 +31,7 @@ pub struct ParquetTableInfo {
 
     pub table_info: TableInfo,
     pub arrow_schema: ArrowSchema,
+    pub files_to_read: Option<Vec<StageFileInfo>>,
 }
 
 impl ParquetTableInfo {

src/query/catalog/src/table_context.rs

@@ -30,8 +30,10 @@ use common_io::prelude::FormatSettings;
 use common_meta_app::principal::FileFormatOptions;
 use common_meta_app::principal::RoleInfo;
 use common_meta_app::principal::UserInfo;
+use common_meta_app::schema::TableCopiedFileInfo;
 use common_settings::Settings;
 use common_storage::DataOperator;
+use common_storage::StageFileInfo;
 use common_storage::StorageMetrics;
 
 use crate::catalog::Catalog;
@@ -133,4 +135,20 @@ pub trait TableContext: Send + Sync {
 
     async fn get_table(&self, catalog: &str, database: &str, table: &str)
     -> Result<Arc<dyn Table>>;
+
+    async fn color_copied_files(
+        &self,
+        catalog_name: &str,
+        database_name: &str,
+        table_name: &str,
+        files: Vec<StageFileInfo>,
+    ) -> Result<Vec<StageFileInfo>>;
+
+    async fn upsert_copied_files(
+        &self,
+        catalog_name: &str,
+        database_name: &str,
+        table_name: &str,
+        copy_stage_files: BTreeMap<String, TableCopiedFileInfo>,
+    ) -> Result<()>;
 }