feat(query): supporting changing endpoint in IPFS copy

1. remove over designed tests
2. enable changing endpoints in IPFS
  ```sql
  COPY INTO ontime FROM 'ipfs://<QMHASH>' CONNECTION = ( ENDPOINT_URL \
  = 'https://ipfs.filebase.io' );
  ```

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

Author: ClSlaid

src/common/storage/src/location.rs

@@ -98,7 +98,11 @@ pub fn parse_uri_location(l: &UriLocation) -> Result<(StorageParams, String)> {
             root: root.to_string(),
         }),
         Scheme::Ipfs => StorageParams::Ipfs(StorageIpfsConfig {
-            endpoint_url: STORAGE_IPFS_DEFAULT_ENDPOINT.to_string(),
+            endpoint_url: l
+                .connection
+                .get("endpoint_url")
+                .cloned()
+                .unwrap_or_else(|| STORAGE_IPFS_DEFAULT_ENDPOINT.to_string()),
             root: "/ipfs/".to_string() + l.name.as_str(),
         }),
         Scheme::S3 => StorageParams::S3(StorageS3Config {

src/common/storage/tests/location.rs

@@ -30,35 +30,38 @@ use common_storage::STORAGE_S3_DEFAULT_ENDPOINT;
 fn test_parse_uri_location() -> Result<()> {
     let cases = vec![
         (
-            "ipfs-directory",
+            "ipfs-default-endpoint",
             UriLocation {
                 protocol: "ipfs".to_string(),
-                name: "too-simple/".to_string(),
+                name: "too-simple".to_string(),
                 path: "/".to_string(),
                 connection: BTreeMap::new(),
             },
             (
                 StorageParams::Ipfs(StorageIpfsConfig {
                     endpoint_url: STORAGE_IPFS_DEFAULT_ENDPOINT.to_string(),
-                    root: "/ipfs/too-simple/".to_string(),
+                    root: "/ipfs/too-simple".to_string(),
                 }),
                 "/".to_string(),
             ),
         ),
         (
-            "ipfs-file",
+            "ipfs-change-endpoint",
             UriLocation {
                 protocol: "ipfs".to_string(),
                 name: "too-naive".to_string(),
                 path: "/".to_string(),
-                connection: Default::default(),
+                connection: vec![("endpoint_url", "https://ipfs.filebase.io")]
+                    .into_iter()
+                    .map(|(k, v)| (k.to_string(), v.to_string()))
+                    .collect(),
             },
             (
                 StorageParams::Ipfs(StorageIpfsConfig {
-                    endpoint_url: STORAGE_IPFS_DEFAULT_ENDPOINT.to_string(),
-                    root: "/ipfs//".to_string(),
+                    endpoint_url: "https://ipfs.filebase.io".to_string(),
+                    root: "/ipfs/too-naive".to_string(),
                 }),
-                "too-naive".to_string(),
+                "/".to_string(),
             ),
         ),
         (

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

@@ -2,39 +2,31 @@
 
 CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
 . "$CURDIR"/../../../shell_env.sh
-TABLE=ontime200
-FILE=ontime_200
 QMHASH=QmPpCt1aYGb9JWJRmXRUnmJtVgeFFTJGzWFYEEX7bo9zGJ
 
 echo "drop table if exists ${TABLE};" | $MYSQL_CLIENT_CONNECT
 
 ## Create table
-cat $CURDIR/../ddl/ontime.sql | sed "s/ontime/$TABLE/g" | $MYSQL_CLIENT_CONNECT
+cat $CURDIR/../ddl/ontime.sql | sed "s/ontime/ontime_199/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 into ontime_199 from 'ipfs://$QMHASH/ontime.csv' CONNECTION = (ENDPOINT_URL='https://ipfs.filebase.io') 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 into ontime_199 from 'ipfs://$QMHASH/ontime.csv.gz' CONNECTION = (ENDPOINT_URL='https://ipfs.filebase.io') 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 into ontime_199 from 'ipfs://$QMHASH/ontime.csv.zst' CONNECTION = (ENDPOINT_URL='https://ipfs.filebase.io') 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 into ontime_199 from 'ipfs://$QMHASH/ontime.csv.bz2' CONNECTION = (ENDPOINT_URL='https://ipfs.filebase.io') 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')"
+  "copy into ontime_199 from 'ipfs://$QMHASH/ontime.csv.xz' CONNECTION = (ENDPOINT_URL='https://ipfs.filebase.io') FILE_FORMAT = (type = 'CSV' field_delimiter = ',' compression = 'xz'  record_delimiter = '\n' skip_header = 1)"
 )
 
 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
+  echo "select count(1), avg(Year), sum(DayOfWeek)  from ontime_199" | $MYSQL_CLIENT_CONNECT
+  echo "truncate table ontime_199" | $MYSQL_CLIENT_CONNECT
 done
 
 ## Drop table
-echo "drop table if exists $TABLE;" | $MYSQL_CLIENT_CONNECT
+echo "drop table if exists ontime_199;" | $MYSQL_CLIENT_CONNECT