Merge pull request #7729 from ClSlaid/copy-from-ipfs

feat(query): implement copy from ipfs

Author: mergify[bot]

Cargo.lock

@@ -22,6 +22,7 @@ opendal = { version = "0.17.1", features = [
     "layers-retry",
     "layers-tracing",
     "layers-metrics",
+    "services-ipfs",
     "compress",
 ] }
 percent-encoding = "2.2.0"

src/common/storage/Cargo.toml

@@ -39,6 +39,7 @@ pub enum StorageParams {
     #[cfg(feature = "storage-hdfs")]
     Hdfs(StorageHdfsConfig),
     Http(StorageHttpConfig),
+    Ipfs(StorageIpfsConfig),
     Memory,
     Obs(StorageObsConfig),
     S3(StorageS3Config),
@@ -72,6 +73,9 @@ impl Display for StorageParams {
             StorageParams::Http(v) => {
                 write!(f, "http://endpoint={},paths={:?}", v.endpoint_url, v.paths)
             }
+            StorageParams::Ipfs(c) => {
+                write!(f, "ipfs://endpoint={},root={}", c.endpoint_url, c.root)
+            }
             StorageParams::Memory => write!(f, "memory://"),
             StorageParams::Obs(v) => write!(
                 f,
@@ -100,6 +104,7 @@ impl StorageParams {
             #[cfg(feature = "storage-hdfs")]
             StorageParams::Hdfs(_) => false,
             StorageParams::Http(v) => v.endpoint_url.starts_with("https://"),
+            StorageParams::Ipfs(c) => c.endpoint_url.starts_with("https://"),
             StorageParams::Memory => false,
             StorageParams::Obs(v) => v.endpoint_url.starts_with("https://"),
             StorageParams::S3(v) => v.endpoint_url.starts_with("https://"),
@@ -263,6 +268,14 @@ pub struct StorageHttpConfig {
     pub paths: Vec<String>,
 }
 
+pub const STORAGE_IPFS_DEFAULT_ENDPOINT: &str = "https://ipfs.io";
+/// Config for IPFS storage backend
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct StorageIpfsConfig {
+    pub endpoint_url: String,
+    pub root: String,
+}
+
 /// Config for storage backend obs.
 #[derive(Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
 pub struct StorageObsConfig {

src/common/storage/src/config.rs

@@ -23,10 +23,12 @@ pub use config::StorageFsConfig;
 pub use config::StorageGcsConfig;
 pub use config::StorageHdfsConfig;
 pub use config::StorageHttpConfig;
+pub use config::StorageIpfsConfig;
 pub use config::StorageObsConfig;
 pub use config::StorageParams;
 pub use config::StorageS3Config;
 pub use config::STORAGE_GCS_DEFAULT_ENDPOINT;
+pub use config::STORAGE_IPFS_DEFAULT_ENDPOINT;
 pub use config::STORAGE_S3_DEFAULT_ENDPOINT;
 
 mod operator;

src/common/storage/src/lib.rs

@@ -22,6 +22,8 @@ use opendal::Scheme;
 use percent_encoding::percent_decode_str;
 
 use crate::config::StorageHttpConfig;
+use crate::config::StorageIpfsConfig;
+use crate::config::STORAGE_IPFS_DEFAULT_ENDPOINT;
 use crate::config::STORAGE_S3_DEFAULT_ENDPOINT;
 use crate::StorageAzblobConfig;
 use crate::StorageParams;
@@ -95,6 +97,14 @@ pub fn parse_uri_location(l: &UriLocation) -> Result<(StorageParams, String)> {
                 .to_string(),
             root: root.to_string(),
         }),
+        Scheme::Ipfs => StorageParams::Ipfs(StorageIpfsConfig {
+            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 {
             endpoint_url: l
                 .connection

src/common/storage/src/location.rs

@@ -53,6 +53,7 @@ pub fn init_operator(cfg: &StorageParams) -> Result<Operator> {
         #[cfg(feature = "storage-hdfs")]
         StorageParams::Hdfs(cfg) => init_hdfs_operator(cfg)?,
         StorageParams::Http(cfg) => init_http_operator(cfg)?,
+        StorageParams::Ipfs(cfg) => init_ipfs_operator(cfg)?,
         StorageParams::Memory => init_memory_operator()?,
         StorageParams::Obs(cfg) => init_obs_operator(cfg)?,
         StorageParams::S3(cfg) => init_s3_operator(cfg)?,
@@ -122,6 +123,17 @@ pub fn init_hdfs_operator(cfg: &super::StorageHdfsConfig) -> Result<Operator> {
     Ok(Operator::new(builder.build()?).layer(LoggingLayer))
 }
 
+pub fn init_ipfs_operator(cfg: &super::StorageIpfsConfig) -> Result<Operator> {
+    use opendal::services::ipfs;
+
+    let mut builder = ipfs::Builder::default();
+
+    builder.root(&cfg.root);
+    builder.endpoint(&cfg.endpoint_url);
+
+    Ok(Operator::new(builder.build()?).layer(LoggingLayer))
+}
+
 pub fn init_http_operator(cfg: &StorageHttpConfig) -> Result<Operator> {
     let mut builder = http::Builder::default();
 

src/common/storage/src/operator.rs

@@ -18,15 +18,52 @@ use std::io::Result;
 use common_storage::parse_uri_location;
 use common_storage::StorageGcsConfig;
 use common_storage::StorageHttpConfig;
+use common_storage::StorageIpfsConfig;
 use common_storage::StorageParams;
 use common_storage::StorageS3Config;
 use common_storage::UriLocation;
 use common_storage::STORAGE_GCS_DEFAULT_ENDPOINT;
+use common_storage::STORAGE_IPFS_DEFAULT_ENDPOINT;
 use common_storage::STORAGE_S3_DEFAULT_ENDPOINT;
 
 #[test]
 fn test_parse_uri_location() -> Result<()> {
     let cases = vec![
+        (
+            "ipfs-default-endpoint",
+            UriLocation {
+                protocol: "ipfs".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(),
+                }),
+                "/".to_string(),
+            ),
+        ),
+        (
+            "ipfs-change-endpoint",
+            UriLocation {
+                protocol: "ipfs".to_string(),
+                name: "too-naive".to_string(),
+                path: "/".to_string(),
+                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: "https://ipfs.filebase.io".to_string(),
+                    root: "/ipfs/too-naive".to_string(),
+                }),
+                "/".to_string(),
+            ),
+        ),
         (
             "s3_with_access_key_id",
             UriLocation {

src/common/storage/tests/location.rs

@@ -334,7 +334,7 @@ impl CopyInterpreterV2 {
                 bind_context,
                 ..
             } => (s_expr, metadata, bind_context),
-            v => unreachable!("Input plan must be Query, but it's {v}"),
+            v => unreachable!("Input plan must be Query, but it's {}", v),
         };
 
         let select_interpreter = SelectInterpreterV2::try_create(

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

@@ -0,0 +1,4 @@
+199	2020.0	769
+199	2020.0	769
+199	2020.0	769
+199	2020.0	769

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

@@ -0,0 +1,30 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. "$CURDIR"/../../../shell_env.sh
+QMHASH=QmPpCt1aYGb9JWJRmXRUnmJtVgeFFTJGzWFYEEX7bo9zGJ
+
+echo "drop table if exists ontime_200;" | $MYSQL_CLIENT_CONNECT
+
+## Create table
+cat $CURDIR/../ddl/ontime.sql | sed "s/ontime/ontime_200/g" | $MYSQL_CLIENT_CONNECT
+
+copy_from_location_cases=(
+  # copy csv
+  "copy into ontime_200 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 ontime_200 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 ontime_200 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 ontime_200 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)"
+)
+
+for i in "${copy_from_location_cases[@]}"; do
+  echo "$i" | $MYSQL_CLIENT_CONNECT
+  echo "select count(1), avg(Year), sum(DayOfWeek)  from ontime_200" | $MYSQL_CLIENT_CONNECT
+  echo "truncate table ontime_200" | $MYSQL_CLIENT_CONNECT
+done
+
+## Drop table
+echo "drop table if exists ontime_200;" | $MYSQL_CLIENT_CONNECT