save

Author: ZhiHanZ

Cargo.lock

@@ -30,6 +30,7 @@ members = [
     "src/query/catalog",
     "src/query/datablocks",
     "src/query/sharing",
+    "src/query/sharing-endpoint",
     "src/query/datavalues",
     "src/query/expression",
     "src/query/formats",
@@ -75,6 +76,7 @@ members = [
     "src/meta/protos",
     # databend-meta
     "src/meta/service",
+
 ]
 
 [workspace.dependencies]

Cargo.toml

@@ -44,6 +44,7 @@ common-metrics = { path = "../common/metrics" }
 common-tracing = { path = "../common/tracing" }
 databend-meta = { path = "../meta/service" }
 databend-query = { path = "../query/service" }
+sharing-endpoint = { path = "../query/sharing-endpoint" }
 
 # Crates.io dependencies
 anyhow = { workspace = true }
@@ -57,6 +58,7 @@ tokio-stream = "0.1.10"
 tonic = "0.8.1"
 tracing = "0.1.36"
 url = "2.3.1"
+poem = { version = "1", features = ["rustls", "multipart", "compression"] }
 
 [[bin]]
 name = "databend-meta"
@@ -81,3 +83,9 @@ name = "databend-query"
 path = "query/main.rs"
 doctest = false
 test = false
+
+[[bin]]
+name = "open-sharing"
+path = "opensharing/main.rs"
+doctest = false
+test = false

src/binaries/Cargo.toml

@@ -0,0 +1,33 @@
+// Copyright 2022 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use poem::{EndpointExt, Route, Server};
+use common_base::base::tokio;
+use poem::listener::TcpListener;
+use sharing_endpoint::handlers::presign_files;
+use sharing_endpoint::middlewares::SharingAuth;
+use sharing_endpoint::configs::Config;
+use sharing_endpoint::services::SharingServices;
+#[tokio::main]
+async fn main() ->Result<(), std::io::Error> {
+    let config = Config::load().expect("cfgs");
+    println!("config: {:?}", config);
+    SharingServices::init(config).await.expect("failed to init sharing service");
+    let app = Route::new().at("/tenant/:tenant_id/:share_name/table/:table_name/presign", poem::post(presign_files)).with(SharingAuth);
+
+    // TODO(zhihanz): remove the hard coded port into a config
+    Server::new(TcpListener::bind("127.0.0.1:33003"))
+        .run(app)
+        .await
+}

src/binaries/opensharing/main.rs

@@ -32,5 +32,8 @@ pub use inner::CatalogHiveConfig;
 pub use inner::Config;
 pub use inner::QueryConfig;
 pub use inner::ThriftProtocol;
+
+pub use outer_v0::StorageConfig;
+
 pub use version::DATABEND_COMMIT_VERSION;
 pub use version::QUERY_SEMVER;

src/query/config/src/lib.rs

@@ -0,0 +1,31 @@
+[package]
+name = "sharing-endpoint"
+version = { workspace = true }
+authors = { workspace = true }
+license = { workspace = true }
+publish = { workspace = true }
+edition = { workspace = true }
+
+[lib]
+doctest = false
+
+[dependencies]
+common-exception = { path = "../../common/exception" }
+common-base = { path = "../../common/base" }
+common-storage = { path = "../../common/storage" }
+common-config = { path = "../../query/config" }
+uuid = { version = "1.1.2", features = ["serde", "v4"] }
+once_cell = "1.15.0"
+
+time = { version = "0.3", features = ["serde"] }
+
+anyhow = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+thiserror = { version = "1" }
+opendal = "0.22.2"
+poem = { version = "1", features = ["rustls", "multipart", "compression"] }
+clap = { workspace = true }
+serfig = "0.0.3"
+base64 = "0.13.0"
+

src/query/sharing-endpoint/Cargo.toml

@@ -0,0 +1,124 @@
+use std::sync::Arc;
+use time::Duration;
+use opendal::Operator;
+use crate::configs::Config;
+use once_cell::sync::OnceCell;
+use common_base::base::Singleton;
+use common_exception::Result;
+use common_storage::{init_operator, StorageParams};
+use crate::models;
+use crate::models::{PresignFileResponse, SharedTableResponse};
+
+static SHARING_ACCESSOR: OnceCell<Singleton<Arc<SharingAccessor>>> = OnceCell::new();
+
+#[derive(Clone)]
+pub struct SharingAccessor {
+    op : Arc<Operator>,
+    config: Config,
+}
+
+// file would have two kind of path:
+// 1. with root. e.g. /root1/root2/root3/db1/tb1/file1
+// 2. without root e.g. db1/tb1/file1
+// after it would be converted to file1
+// and then it would use the location in table spec to form the final path
+// {localtion}/file1
+pub fn truncate_root(root: String, loc: String) -> String {
+    let root = root.trim_matches('/');
+    let loc = loc.trim_matches('/');
+    return if loc.starts_with(root) {
+        let o1 = loc.strip_prefix(root).unwrap();
+
+        let updated = o1.trim_matches('/');
+        let arr = updated.split('/').collect::<Vec<&str>>();
+        if arr.len() > 2 {
+            return arr[2..].join("/");
+        }
+        updated.to_string()
+    } else {
+        let arr = loc.split('/').collect::<Vec<&str>>();
+        if arr.len() > 2 {
+            return arr[2..].join("/");
+        }
+        loc.to_string()
+    }
+}
+
+impl SharingAccessor {
+    pub async fn init(cfg: &Config, v: Singleton<Arc<SharingAccessor>>) -> Result<()> {
+        let op = init_operator(&cfg.storage.params)?;
+        v.init(Arc::new(SharingAccessor {
+            op:    Arc::new(op),
+            config: cfg.clone(),
+        }))?;
+
+        SHARING_ACCESSOR.set(v).ok();
+        Ok(())
+    }
+    pub fn instance() -> Arc<SharingAccessor> {
+        match SHARING_ACCESSOR.get() {
+            None => panic!("Sharing Accessor is not init"),
+            Some(sharing_accessor) => sharing_accessor.get(),
+        }
+    }
+
+    fn get_root(&self) -> String {
+        match self.config.storage.params {
+            StorageParams::S3(ref s3) => s3.root.trim_matches('/').to_string(),
+            _ => "".to_string(),
+        }
+    }
+    fn get_path(&self) -> String {
+        format!("{}/{}", self.get_root(), self.config.tenant)
+    }
+
+    fn get_share_location(&self) -> String {
+        format!("{}/_share_config/share_specs.json", self.config.tenant)
+    }
+
+    pub async fn get_shared_table(&self, input: &models::LambdaInput) -> Result<Option<SharedTableResponse>> {
+        let sharing_accessor = Self::instance();
+        let path = sharing_accessor.get_share_location();
+        println!("path: {}", path);
+        // let path = "t1/_share_config/share_specs.json";
+        let data = sharing_accessor.op.object(&*path).read().await?;
+        let share_specs: models::SharingConfig = serde_json::from_slice(data.as_slice())?;
+        return share_specs.get_tables(input);
+    }
+
+    pub async fn presign_file(&self, table: &SharedTableResponse, input: &models::RequestFile) -> Result<PresignFileResponse> {
+        let loc_prefix = table.location.trim_matches('/');
+        println!("loc_prefix: {}", loc_prefix.clone());
+        println!("input: {}", input.file_name);
+        let file_path = truncate_root(self.get_root(), input.file_name.clone());
+        let loc_prefix = loc_prefix.strip_prefix(self.get_root().as_str()).unwrap();
+        let obj_path = format!("{}/{}", loc_prefix, file_path);
+        let op =  self.op.clone();
+        if input.method == "HEAD" {
+            let s = op.object(obj_path.as_str()).presign_stat(Duration::hours(1))?;
+            return Ok( PresignFileResponse::new(&s, input.file_name.clone()));
+        }
+
+        let s = op.object(obj_path.as_str()).presign_read(Duration::hours(1))?;
+        return Ok( PresignFileResponse::new(&s, input.file_name.clone()));
+    }
+
+    pub async fn get_presigned_files(input: &models::LambdaInput) -> Result<Vec<PresignFileResponse>> {
+
+        let accessor = Self::instance();
+        let table = accessor.get_shared_table(input).await?;
+        return match table {
+            Some(t) => {
+                let mut presigned_files = vec![];
+                for f in input.request_files.iter() {
+                    let presigned_file = accessor.presign_file(&t, f).await?;
+                    presigned_files.push(presigned_file);
+                }
+                Ok(presigned_files)
+            }
+            None => {
+                Ok(vec![])
+            }
+        }
+    }
+}

src/query/sharing-endpoint/src/accessor.rs

@@ -0,0 +1,64 @@
+// Copyright 2022 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use serde::Serialize;
+use serde::Deserialize;
+use common_storage::StorageConfig;
+use common_exception::Result;
+use super::outer_v0::Config as OuterV0Config;
+
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Config {
+    pub tenant : String,
+    pub storage: StorageConfig,
+}
+
+impl Config {
+
+    /// As requires by [RFC: Config Backward Compatibility](https://github.com/datafuselabs/databend/pull/5324), we will load user's config via wrapper [`ConfigV0`] and then convert from [`ConfigV0`] to [`Config`].
+    ///
+    /// In the future, we could have `ConfigV1` and `ConfigV2`.
+    pub fn load() -> Result<Self> {
+        let cfg: Self = OuterV0Config::load(true)?.try_into()?;
+        Ok(cfg)
+    }
+
+    /// # NOTE
+    ///
+    /// This function is served for tests only.
+    pub fn load_for_test() -> Result<Self> {
+        let cfg: Self = OuterV0Config::load(false)?.try_into()?;
+        Ok(cfg)
+    }
+
+
+    /// Transform config into the outer style.
+    ///
+    /// This function should only be used for end-users.
+    pub fn into_outer(self) -> OuterV0Config {
+        OuterV0Config::from(self)
+    }
+}
+
+
+impl Default for Config {
+    fn default() -> Self {
+        Self {
+            tenant: "".to_string(),
+            storage: StorageConfig::default(),
+        }
+    }
+}
+

src/query/sharing-endpoint/src/configs/inner.rs

@@ -0,0 +1,18 @@
+// Copyright 2021 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+mod inner;
+mod outer_v0;
+
+pub use inner::Config;