fix(Copy): max_files was set to 1000 when not provided. (#10813)

* fix(Copy): purge only copied files.

* chore(Copy): log max_files.

* ci(copy): polish tests.

Author: youngsofun

src/query/catalog/src/table_context.rs

@@ -136,12 +136,12 @@ 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(
+    async fn filter_out_copied_files(
         &self,
         catalog_name: &str,
         database_name: &str,
         table_name: &str,
-        files: Vec<StageFileInfo>,
+        files: &[StageFileInfo],
         max_files: Option<usize>,
     ) -> Result<Vec<StageFileInfo>>;
 }

src/query/service/src/interpreters/interpreter_copy.rs

@@ -35,7 +35,6 @@ use common_meta_types::MetaId;
 use common_pipeline_core::processors::processor::ProcessorPtr;
 use common_sql::executor::table_read_plan::ToReadDataSourcePlan;
 use common_storage::StageFileInfo;
-use common_storage::StageFileStatus;
 use common_storage::StageFilesInfo;
 use common_storages_fuse::io::Files;
 use common_storages_stage::StageTable;
@@ -167,7 +166,6 @@ impl CopyInterpreter {
         table_name: &str,
         query: &Plan,
         stage_info: StageInfo,
-        all_source_file_infos: Vec<StageFileInfo>,
         need_copy_file_infos: Vec<StageFileInfo>,
         force: bool,
     ) -> Result<PipelineBuildResult> {
@@ -208,7 +206,6 @@ impl CopyInterpreter {
                     ctx.clone(),
                     to_table,
                     stage_info,
-                    all_source_file_infos,
                     need_copy_file_infos,
                     force,
                 )?;
@@ -253,45 +250,43 @@ impl CopyInterpreter {
             Some(max_files)
         };
 
-        let mut all_source_file_infos = if force {
+        let all_source_file_infos = if force {
             StageTable::list_files(&stage_table_info, max_files).await?
         } else {
             StageTable::list_files(&stage_table_info, None).await?
         };
+        let num_all_files = all_source_file_infos.len();
 
-        info!("end to list files: {}", all_source_file_infos.len());
+        info!("end to list files: got {} files", num_all_files);
 
-        if !force {
+        let need_copy_file_infos = if force {
+            all_source_file_infos
+        } else {
             // Status.
             {
-                let status = "begin to color copied files";
+                let status = "begin to filter out copied files";
                 ctx.set_status_info(status);
                 info!(status);
             }
 
-            all_source_file_infos = table_ctx
-                .color_copied_files(
+            let files = table_ctx
+                .filter_out_copied_files(
                     catalog_name,
                     database_name,
                     table_name,
-                    all_source_file_infos,
+                    &all_source_file_infos,
                     max_files,
                 )
                 .await?;
 
-            info!("end to color copied files: {}", all_source_file_infos.len());
-        }
-
-        let mut need_copy_file_infos = vec![];
-        for file in &all_source_file_infos {
-            if file.status == StageFileStatus::NeedCopy {
-                need_copy_file_infos.push(file.clone());
-            }
-        }
+            info!("end filtering out copied files: {}", num_all_files);
+            files
+        };
 
         info!(
-            "copy: read all files finished, all:{}, need copy:{}, elapsed:{}",
-            all_source_file_infos.len(),
+            "copy: read files with max_files={:?} finished, all:{}, need copy:{}, elapsed:{}",
+            max_files,
+            num_all_files,
             need_copy_file_infos.len(),
             start.elapsed().as_secs()
         );
@@ -383,7 +378,6 @@ impl CopyInterpreter {
                     ctx.clone(),
                     to_table,
                     stage_table_info_clone.stage_info,
-                    all_source_file_infos,
                     need_copy_file_infos,
                     force,
                 )?;
@@ -410,18 +404,18 @@ impl CopyInterpreter {
         ctx: Arc<QueryContext>,
         to_table: Arc<dyn Table>,
         stage_info: StageInfo,
-        all_source_files: Vec<StageFileInfo>,
-        need_copy_files: Vec<StageFileInfo>,
+        copied_files: Vec<StageFileInfo>,
         force: bool,
     ) -> Result<()> {
-        let mut copied_files = BTreeMap::new();
-        for file in need_copy_files {
+        let num_copied_files = copied_files.len();
+        let mut copied_file_tree = BTreeMap::new();
+        for file in &copied_files {
             // Short the etag to 7 bytes for less space in metasrv.
             let short_etag = file.etag.clone().map(|mut v| {
                 v.truncate(7);
                 v
             });
-            copied_files.insert(file.path.clone(), TableCopiedFileInfo {
+            copied_file_tree.insert(file.path.clone(), TableCopiedFileInfo {
                 etag: short_etag,
                 content_length: file.size,
                 last_modified: Some(file.last_modified),
@@ -434,18 +428,17 @@ impl CopyInterpreter {
 
             let table_id = to_table.get_id();
             let expire_hours = ctx.get_settings().get_load_file_metadata_expire_hours()?;
-            let num_copied_files = copied_files.len();
 
             let fail_if_duplicated = !force;
             let upsert_copied_files_request = Self::upsert_copied_files_request(
                 table_id,
                 expire_hours,
-                copied_files,
+                copied_file_tree,
                 fail_if_duplicated,
             );
 
             {
-                let status = format!("begin commit, number of copied files:{}", num_copied_files,);
+                let status = format!("begin commit, number of copied files:{}", num_copied_files);
                 ctx.set_status_info(&status);
                 info!(status);
             }
@@ -482,17 +475,17 @@ impl CopyInterpreter {
 
                 // Status.
                 {
-                    let status = format!("begin to purge files:{}", all_source_files.len());
+                    let status = format!("begin to purge files:{}", num_copied_files);
                     ctx.set_status_info(&status);
                     info!(status);
                 }
 
-                CopyInterpreter::try_purge_files(ctx.clone(), &stage_info, &all_source_files).await;
+                CopyInterpreter::try_purge_files(ctx.clone(), &stage_info, &copied_files).await;
 
                 // Status.
                 info!(
                     "end to purge files:{}, elapsed:{}",
-                    all_source_files.len(),
+                    num_copied_files,
                     purge_start.elapsed().as_secs()
                 );
             }
@@ -560,7 +553,6 @@ impl Interpreter for CopyInterpreter {
                 table_name,
                 stage_info,
                 from,
-                all_source_file_infos,
                 need_copy_file_infos,
                 force,
                 ..
@@ -571,7 +563,6 @@ impl Interpreter for CopyInterpreter {
                     table_name,
                     from,
                     *stage_info.clone(),
-                    all_source_file_infos.clone(),
                     need_copy_file_infos.clone(),
                     *force,
                 )

src/query/service/src/sessions/query_ctx.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 use std::cmp::min;
-use std::collections::BTreeMap;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::collections::VecDeque;
@@ -55,7 +54,6 @@ use common_settings::ChangeValue;
 use common_settings::Settings;
 use common_storage::DataOperator;
 use common_storage::StageFileInfo;
-use common_storage::StageFileStatus;
 use common_storage::StorageMetrics;
 use common_storages_fuse::TableContext;
 use common_storages_parquet::ParquetTable;
@@ -494,46 +492,43 @@ impl TableContext for QueryContext {
         self.shared.get_table(catalog, database, table).await
     }
 
-    async fn color_copied_files(
+    async fn filter_out_copied_files(
         &self,
         catalog_name: &str,
         database_name: &str,
         table_name: &str,
-        files: Vec<StageFileInfo>,
+        files: &[StageFileInfo],
         max_files: Option<usize>,
     ) -> Result<Vec<StageFileInfo>> {
         let tenant = self.get_tenant();
-        let files = files.clone();
         let catalog = self.get_catalog(catalog_name)?;
         let table = catalog
             .get_table(&tenant, database_name, table_name)
             .await?;
         let table_id = table.get_id();
 
         let mut limit: usize = 0;
-        let max_files = max_files.unwrap_or(MAX_QUERY_COPIED_FILES_NUM);
-        let max_copied_files = min(MAX_QUERY_COPIED_FILES_NUM, max_files);
-        let mut copied_files = BTreeMap::new();
+        let max_files = max_files.unwrap_or(usize::MAX);
+        let batch_size = min(MAX_QUERY_COPIED_FILES_NUM, max_files);
 
         let mut results = Vec::with_capacity(files.len());
 
-        for chunk in files.chunks(max_copied_files) {
+        for chunk in files.chunks(batch_size) {
             let files = chunk.iter().map(|v| v.path.clone()).collect::<Vec<_>>();
             let req = GetTableCopiedFileReq { table_id, files };
-            let resp = catalog
+            let copied_files = catalog
                 .get_table_copied_file_info(&tenant, database_name, req)
-                .await?;
-            copied_files.extend(resp.file_info);
+                .await?
+                .file_info;
             // Colored
             for file in chunk {
-                let mut file = file.clone();
                 if let Some(copied_file) = copied_files.get(&file.path) {
                     match &copied_file.etag {
                         Some(copied_etag) => {
                             if let Some(file_etag) = &file.etag {
                                 // Check the 7 bytes etag prefix.
                                 if file_etag.starts_with(copied_etag) {
-                                    file.status = StageFileStatus::AlreadyCopied;
+                                    continue;
                                 }
                             }
                         }
@@ -542,17 +537,16 @@ impl TableContext for QueryContext {
                             if copied_file.content_length == file.size
                                 && copied_file.last_modified == Some(file.last_modified)
                             {
-                                file.status = StageFileStatus::AlreadyCopied;
+                                continue;
                             }
                         }
                     }
                 }
-                if file.status == StageFileStatus::NeedCopy {
-                    results.push(file);
-                    limit += 1;
-                    if limit == max_files {
-                        return Ok(results);
-                    }
+
+                results.push(file.clone());
+                limit += 1;
+                if limit == max_files {
+                    return Ok(results);
                 }
             }
         }

src/query/service/tests/it/storages/fuse/operations/commit.rs

@@ -509,12 +509,12 @@ impl TableContext for CtxDelegation {
         todo!()
     }
 
-    async fn color_copied_files(
+    async fn filter_out_copied_files(
         &self,
         _catalog_name: &str,
         _database_name: &str,
         _table_name: &str,
-        _files: Vec<StageFileInfo>,
+        _files: &[StageFileInfo],
         _max_files: Option<usize>,
     ) -> Result<Vec<StageFileInfo>> {
         todo!()

src/query/sql/src/planner/binder/copy.rs

@@ -557,11 +557,11 @@ impl<'a> Binder {
 
             files = self
                 .ctx
-                .color_copied_files(
+                .filter_out_copied_files(
                     dst_catalog_name,
                     dst_database_name,
                     dst_table_name,
-                    files,
+                    &files,
                     None,
                 )
                 .await?;
@@ -625,7 +625,6 @@ impl<'a> Binder {
             schema: dst_table.schema(),
             from: Box::new(query_plan),
             stage_info: Box::new(stage_info),
-            all_source_file_infos: files,
             need_copy_file_infos,
             validation_mode,
             force: stmt.force,

src/query/sql/src/planner/plans/copy.rs

@@ -75,7 +75,6 @@ pub enum CopyPlan {
         stage_info: Box<StageInfo>,
         validation_mode: ValidationMode,
         from: Box<Plan>,
-        all_source_file_infos: Vec<StageFileInfo>,
         need_copy_file_infos: Vec<StageFileInfo>,
         force: bool,
     },

tests/suites/1_stateful/00_copy/00_0004_copy_with_max_files.result

@@ -0,0 +1,28 @@
+--- force = false, purge = false
+4
+remain 3 files
+6
+remain 3 files
+6
+remain 3 files
+--- force = false, purge = true
+4
+remain 1 files
+6
+remain 0 files
+6
+remain 0 files
+--- force = true, purge = false
+4
+remain 3 files
+8
+remain 3 files
+12
+remain 3 files
+--- force = true, purge = true
+4
+remain 1 files
+6
+remain 0 files
+6
+remain 0 files