tests: try make integrated test

Signed-off-by: ClSlaid <cailue@bupt.edu.cn>

Author: ClSlaid

src/common/storage/src/location.rs

@@ -44,7 +44,7 @@ pub fn parse_uri_location(l: &UriLocation) -> Result<(StorageParams, String)> {
     // Path endswith `/` means it's a directory, otherwise it's a file.
     // If the path is a directory, we will use this path as root.
     // If the path is a file, we will use `/` as root (which is the default value)
-    let (mut root, mut path) = if l.path.ends_with('/') {
+    let (root, path) = if l.path.ends_with('/') {
         (l.path.as_str(), "/")
     } else {
         ("/", l.path.as_str())
@@ -98,14 +98,9 @@ pub fn parse_uri_location(l: &UriLocation) -> Result<(StorageParams, String)> {
             root: root.to_string(),
         }),
         Scheme::Ipfs => {
-            if l.name.ends_with('/') {
-                root = l.name.as_str();
-            } else {
-                path = l.name.as_str();
-            }
             StorageParams::Ipfs(StorageIpfsConfig {
                 endpoint_url: STORAGE_IPFS_DEFAULT_ENDPOINT.to_string(),
-                root: "/ipfs/".to_string() + root,
+                root: "/ipfs/".to_string() + l.name.as_str(),
             })
         }
         Scheme::S3 => StorageParams::S3(StorageS3Config {

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

@@ -163,15 +163,18 @@ impl CopyInterpreterV2 {
         match &from.source_info {
             SourceInfo::StageSource(table_info) => {
                 let path = &table_info.path;
+                tracing::debug!("list files under path: {}", path);
                 // Here we add the path to the file: /path/to/path/file1.
                 let files_with_path = if !files.is_empty() {
+                    tracing::trace!("got files offered in list");
                     let mut files_with_path = vec![];
                     for file in files {
                         let new_path = Path::new(path).join(file);
                         files_with_path.push(new_path.to_string_lossy().to_string());
                     }
                     files_with_path
                 } else if !path.ends_with('/') {
+                    tracing::trace!("path not ends with '/'");
                     let rename_me: Arc<dyn TableContext> = self.ctx.clone();
                     let op = StageSourceHelper::get_op(&rename_me, &table_info.stage_info).await?;
                     if op.object(path).is_exist().await? {
@@ -180,6 +183,7 @@ impl CopyInterpreterV2 {
                         vec![]
                     }
                 } else {
+                    tracing::trace!("have to list directories in stage");
                     let rename_me: Arc<dyn TableContext> = self.ctx.clone();
                     let op = StageSourceHelper::get_op(&rename_me, &table_info.stage_info).await?;
 
@@ -191,8 +195,10 @@ impl CopyInterpreterV2 {
                     let mut objects = op.batch().walk_top_down(path)?;
                     while let Some(de) = objects.try_next().await? {
                         if de.mode().is_dir() {
+                            tracing::trace!("listing in stage path:{} got directory: {}", path, de.path());
                             continue;
                         }
+                        tracing::trace!("listing in stage path:{} got file: {}",path, de.path());
                         list.insert(de.path().to_string());
                     }
 

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

@@ -0,0 +1,8 @@
+199	2020.0	769
+199	2020.0	769
+199	2020.0	769
+199	2020.0	769
+199	2020.0	769
+398	2020.0	1538
+398	2020.0	1538
+398	2020.0	1538

tests/suites/1_stateful/00_copy/00_0000_copy_from_ipfs_location.sh

@@ -0,0 +1,40 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. "$CURDIR"/../../../shell_env.sh
+TABLE=ontime200
+FILE=ontime_200
+QMHASH=Qmei4dyyPazUy24cfCAtmumC1Ff1VLLiqjzF1zCXXVtvSk
+
+echo "drop table if exists ${TABLE};" | $MYSQL_CLIENT_CONNECT
+
+## Create table
+cat $CURDIR/../ddl/ontime.sql | sed "s/ontime/$TABLE/g" | $MYSQL_CLIENT_CONNECT
+
+copy_from_location_cases=(
+  # copy csv
+  "copy into $TABLE from 'ipfs://$QMHASH/$FILE.csv' FILE_FORMAT = (type = 'CSV' field_delimiter = ','  record_delimiter = '\n' skip_header = 1)"
+  # copy gzip csv
+  "copy into $TABLE from 'ipfs://$QMHASH/$FILE.csv.gz' FILE_FORMAT = (type = 'CSV' field_delimiter = ',' compression = 'gzip'  record_delimiter = '\n' skip_header = 1)"
+  # copy zstd csv
+  "copy into $TABLE from 'ipfs://$QMHASH/$FILE.csv.zstd' FILE_FORMAT = (type = 'CSV' field_delimiter = ',' compression = 'zstd'  record_delimiter = '\n' skip_header = 1)"
+  # copy bz2 csv
+  "copy into $TABLE from 'ipfs://$QMHASH/$FILE.csv.bz2' FILE_FORMAT = (type = 'CSV' field_delimiter = ',' compression = 'bz2'  record_delimiter = '\n' skip_header = 1)"
+  # copy xz csv
+  "copy into $TABLE from 'ipfs://$QMHASH/$FILE.csv.xz' FILE_FORMAT = (type = 'CSV' field_delimiter = ',' compression = 'xz'  record_delimiter = '\n' skip_header = 1)"
+  # copy file
+  "copy into $TABLE from 'ipfs://$QMHASH' FILES = ('$FILE.csv', '${FILE}_v1.csv') FILE_FORMAT = (type = 'CSV' field_delimiter = ','  record_delimiter = '\n' skip_header = 1)"
+  # copy dir with pattern
+  "copy into $TABLE from 'ipfs://$QMHASH' PATTERN = 'ontime.*csv' FILE_FORMAT = (type = 'CSV' field_delimiter = ',' record_delimiter = '\n' skip_header = 1)"
+  # copy parquet
+  "copy into $TABLE from 'ipfs://$QMHASH'  PATTERN = 'ontime.*parquet' FILE_FORMAT = (type = 'PARQUET')"
+)
+
+for i in "${copy_from_location_cases[@]}"; do
+  echo "$i" | $MYSQL_CLIENT_CONNECT
+  echo "select count(1), avg(Year), sum(DayOfWeek)  from $TABLE" | $MYSQL_CLIENT_CONNECT
+  echo "truncate table $TABLE" | $MYSQL_CLIENT_CONNECT
+done
+
+## Drop table
+echo "drop table if exists $TABLE;" | $MYSQL_CLIENT_CONNECT