refactor the by comment

Author: lichuang

src/query/service/src/interpreters/interpreter_common.rs

@@ -160,6 +160,26 @@ pub async fn validate_grant_object_exists(
     Ok(())
 }
 
+pub async fn stat_file(
+    ctx: &Arc<QueryContext>,
+    stage: &UserStageInfo,
+    path: &str,
+) -> Result<StageFile> {
+    let table_ctx: Arc<dyn TableContext> = ctx.clone();
+    let op = StageSourceHelper::get_op(&table_ctx, stage).await?;
+    let meta = op.object(path).metadata().await?;
+    Ok(StageFile {
+        path: path.to_owned(),
+        size: meta.content_length(),
+        md5: meta.content_md5().map(str::to_string),
+        last_modified: meta
+            .last_modified()
+            .map_or(Utc::now(), |t| Utc.timestamp(t.unix_timestamp(), 0)),
+        creator: None,
+        etag: meta.etag().map(str::to_string),
+    })
+}
+
 pub async fn list_files(
     ctx: &Arc<QueryContext>,
     stage: &UserStageInfo,

src/query/service/src/interpreters/interpreter_copy_v2.rs

@@ -32,7 +32,7 @@ use futures::TryStreamExt;
 use regex::Regex;
 
 use super::append2table;
-use crate::interpreters::interpreter_common::list_files;
+use crate::interpreters::interpreter_common::stat_file;
 use crate::interpreters::Interpreter;
 use crate::interpreters::SelectInterpreterV2;
 use crate::pipelines::PipelineBuildResult;
@@ -79,12 +79,7 @@ impl CopyInterpreterV2 {
             // if force is false, copy only the files that unmatch to the meta copied files info.
             let resp = catalog.get_table_copied_file_info(req).await?;
             for file in files.iter() {
-                let stage_file = list_files(&self.ctx, &table_info.stage_info, file, "").await?;
-                if stage_file.is_empty() {
-                    continue;
-                }
-
-                let stage_file = &stage_file[0];
+                let stage_file = stat_file(&self.ctx, &table_info.stage_info, file).await?;
 
                 if let Some(file_info) = resp.file_info.get(file) {
                     match &file_info.etag {
@@ -120,12 +115,8 @@ impl CopyInterpreterV2 {
         } else {
             // if force is true, copy all the file.
             for file in files.iter() {
-                let stage_file = list_files(&self.ctx, &table_info.stage_info, file, "").await?;
-                if stage_file.is_empty() {
-                    continue;
-                }
+                let stage_file = stat_file(&self.ctx, &table_info.stage_info, file).await?;
 
-                let stage_file = &stage_file[0];
                 file_map.insert(file.clone(), TableCopiedFileInfo {
                     etag: stage_file.etag.clone(),
                     content_length: stage_file.size,