This is an automated email from the ASF dual-hosted git repository.
liurenjie1024 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new e196725 feat: rest client respect prefix prop (#297)
e196725 is described below
commit e196725f0f871141e6d6e8539069b3151d404b70
Author: TennyZhuang <[email protected]>
AuthorDate: Tue Mar 26 17:00:50 2024 +0800
feat: rest client respect prefix prop (#297)
* feat: rest client respect prefix prop
Signed-off-by: TennyZhuang <[email protected]>
* add test
Signed-off-by: TennyZhuang <[email protected]>
* fix tests without prefix
Signed-off-by: TennyZhuang <[email protected]>
* fix clippy
Signed-off-by: TennyZhuang <[email protected]>
---------
Signed-off-by: TennyZhuang <[email protected]>
---
crates/catalog/rest/Cargo.toml | 1 +
crates/catalog/rest/src/catalog.rs | 78 ++++++++++++++++++++++++++++----------
2 files changed, 59 insertions(+), 20 deletions(-)
diff --git a/crates/catalog/rest/Cargo.toml b/crates/catalog/rest/Cargo.toml
index 7cb5fb9..7abe9c8 100644
--- a/crates/catalog/rest/Cargo.toml
+++ b/crates/catalog/rest/Cargo.toml
@@ -33,6 +33,7 @@ keywords = ["iceberg", "rest", "catalog"]
async-trait = { workspace = true }
chrono = { workspace = true }
iceberg = { workspace = true }
+itertools = { workspace = true }
log = "0.4.20"
reqwest = { workspace = true }
serde = { workspace = true }
diff --git a/crates/catalog/rest/src/catalog.rs
b/crates/catalog/rest/src/catalog.rs
index 33a2ea2..52731c8 100644
--- a/crates/catalog/rest/src/catalog.rs
+++ b/crates/catalog/rest/src/catalog.rs
@@ -20,6 +20,7 @@
use std::collections::HashMap;
use async_trait::async_trait;
+use itertools::Itertools;
use reqwest::header::{self, HeaderMap, HeaderName, HeaderValue};
use reqwest::{Client, Request, Response, StatusCode};
use serde::de::DeserializeOwned;
@@ -57,47 +58,45 @@ pub struct RestCatalogConfig {
}
impl RestCatalogConfig {
+ fn url_prefixed(&self, parts: &[&str]) -> String {
+ [&self.uri, PATH_V1]
+ .into_iter()
+ .chain(self.props.get("prefix").map(|s| &**s))
+ .chain(parts.iter().cloned())
+ .join("/")
+ }
+
fn config_endpoint(&self) -> String {
[&self.uri, PATH_V1, "config"].join("/")
}
+ fn get_token_endpoint(&self) -> String {
+ [&self.uri, PATH_V1, "oauth", "tokens"].join("/")
+ }
+
fn namespaces_endpoint(&self) -> String {
- [&self.uri, PATH_V1, "namespaces"].join("/")
+ self.url_prefixed(&["namespaces"])
}
fn namespace_endpoint(&self, ns: &NamespaceIdent) -> String {
- [&self.uri, PATH_V1, "namespaces", &ns.encode_in_url()].join("/")
+ self.url_prefixed(&["namespaces", &ns.encode_in_url()])
}
fn tables_endpoint(&self, ns: &NamespaceIdent) -> String {
- [
- &self.uri,
- PATH_V1,
- "namespaces",
- &ns.encode_in_url(),
- "tables",
- ]
- .join("/")
+ self.url_prefixed(&["namespaces", &ns.encode_in_url(), "tables"])
}
fn rename_table_endpoint(&self) -> String {
- [&self.uri, PATH_V1, "tables", "rename"].join("/")
+ self.url_prefixed(&["tables", "rename"])
}
fn table_endpoint(&self, table: &TableIdent) -> String {
- [
- &self.uri,
- PATH_V1,
+ self.url_prefixed(&[
"namespaces",
&table.namespace.encode_in_url(),
"tables",
encode(&table.name).as_ref(),
- ]
- .join("/")
- }
-
- fn get_token_endpoint(&self) -> String {
- [&self.uri, PATH_V1, "oauth", "tokens"].join("/")
+ ])
}
fn try_create_rest_client(&self) -> Result<HttpClient> {
@@ -956,6 +955,45 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_config_override_prefix() {
+ let mut server = Server::new_async().await;
+
+ let config_mock = server
+ .mock("GET", "/v1/config")
+ .with_status(200)
+ .with_body(
+ r#"{
+ "overrides": {
+ "warehouse": "s3://iceberg-catalog",
+ "prefix": "ice/warehouses/my"
+ },
+ "defaults": {}
+ }"#,
+ )
+ .create_async()
+ .await;
+
+ let list_ns_mock = server
+ .mock("GET", "/v1/ice/warehouses/my/namespaces")
+ .with_body(
+ r#"{
+ "namespaces": []
+ }"#,
+ )
+ .create_async()
+ .await;
+
+ let catalog =
RestCatalog::new(RestCatalogConfig::builder().uri(server.url()).build())
+ .await
+ .unwrap();
+
+ let _namespaces = catalog.list_namespaces(None).await.unwrap();
+
+ config_mock.assert_async().await;
+ list_ns_mock.assert_async().await;
+ }
+
#[tokio::test]
async fn test_list_namespace() {
let mut server = Server::new_async().await;
