diff --git a/Cargo.toml b/Cargo.toml index 34eef8d27d..e8e0473c18 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,4 +17,4 @@ [workspace] resolver = "2" -members = ["crates/*"] +members = ["crates/catalog/*", "crates/iceberg"] diff --git a/crates/catalog/rest/Cargo.toml b/crates/catalog/rest/Cargo.toml new file mode 100644 index 0000000000..0531a513bd --- /dev/null +++ b/crates/catalog/rest/Cargo.toml @@ -0,0 +1,41 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you 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. + +[package] +name = "iceberg-catalog-rest" +version = "0.1.0" +edition = "2021" + +categories = ["database"] +description = "Apache Iceberg Rust REST API" +repository = "https://github.com/apache/iceberg-rust" +license = "Apache-2.0" +keywords = ["iceberg", "rest", "catalog"] + +[dependencies] +async-trait = "0.1" +iceberg = { path = "../../iceberg" } +reqwest = { version = "^0.11", features = ["json"] } +serde = { version = "^1.0", features = ["rc"] } +serde_derive = "^1.0" +serde_json = "^1.0" +typed-builder = "^0.17" +urlencoding = "2" + +[dev-dependencies] +mockito = "^1" +tokio = { version = "1", features = ["macros"] } diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs new file mode 100644 index 0000000000..f2674373c5 --- /dev/null +++ b/crates/catalog/rest/src/catalog.rs @@ -0,0 +1,887 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +//! This module contains rest catalog implementation. + +use std::collections::HashMap; + +use async_trait::async_trait; +use reqwest::header::{self, HeaderMap, HeaderName, HeaderValue}; +use reqwest::{Client, Request}; +use serde::de::DeserializeOwned; +use typed_builder::TypedBuilder; +use urlencoding::encode; + +use iceberg::table::Table; +use iceberg::Result; +use iceberg::{ + Catalog, Error, ErrorKind, Namespace, NamespaceIdent, TableCommit, TableCreation, TableIdent, +}; + +use self::_serde::{ + CatalogConfig, ErrorModel, ErrorResponse, ListNamespaceResponse, ListTableResponse, + NamespaceSerde, RenameTableRequest, NO_CONTENT, OK, +}; + +const ICEBERG_REST_SPEC_VERSION: &str = "1.14"; +const CARGO_PKG_VERSION: &str = env!("CARGO_PKG_VERSION"); +const PATH_V1: &str = "v1"; + +/// Rest catalog configuration. +#[derive(Debug, TypedBuilder)] +pub struct RestCatalogConfig { + uri: String, + #[builder(default, setter(strip_option))] + warehouse: Option, + + #[builder(default)] + props: HashMap, +} + +impl RestCatalogConfig { + fn config_endpoint(&self) -> String { + [&self.uri, PATH_V1, "config"].join("/") + } + + fn namespaces_endpoint(&self) -> String { + [&self.uri, PATH_V1, "namespaces"].join("/") + } + + fn namespace_endpoint(&self, ns: &NamespaceIdent) -> String { + [&self.uri, PATH_V1, "namespaces", &ns.encode_in_url()].join("/") + } + + fn tables_endpoint(&self, ns: &NamespaceIdent) -> String { + [ + &self.uri, + PATH_V1, + "namespaces", + &ns.encode_in_url(), + "tables", + ] + .join("/") + } + + fn rename_table_endpoint(&self) -> String { + [&self.uri, PATH_V1, "tables", "rename"].join("/") + } + + fn table_endpoint(&self, table: &TableIdent) -> String { + [ + &self.uri, + PATH_V1, + "namespaces", + &table.namespace.encode_in_url(), + "tables", + encode(&table.name).as_ref(), + ] + .join("/") + } + + fn try_create_rest_client(&self) -> Result { + //TODO: We will add oauth, ssl config, sigv4 later + let headers = HeaderMap::from_iter([ + ( + header::CONTENT_TYPE, + HeaderValue::from_static("application/json"), + ), + ( + HeaderName::from_static("x-client-version"), + HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION), + ), + ( + header::USER_AGENT, + HeaderValue::from_str(&format!("iceberg-rs/{}", CARGO_PKG_VERSION)).unwrap(), + ), + ]); + + Ok(HttpClient( + Client::builder().default_headers(headers).build()?, + )) + } +} + +struct HttpClient(Client); + +impl HttpClient { + async fn query< + R: DeserializeOwned, + E: DeserializeOwned + Into, + const SUCCESS_CODE: u16, + >( + &self, + request: Request, + ) -> Result { + let resp = self.0.execute(request).await?; + + if resp.status().as_u16() == SUCCESS_CODE { + let text = resp.bytes().await?; + Ok(serde_json::from_slice::(&text).map_err(|e| { + Error::new( + ErrorKind::Unexpected, + "Failed to parse response from rest catalog server!", + ) + .with_context("json", String::from_utf8_lossy(&text)) + .with_source(e) + })?) + } else { + let text = resp.bytes().await?; + let e = serde_json::from_slice::(&text).map_err(|e| { + Error::new( + ErrorKind::Unexpected, + "Failed to parse response from rest catalog server!", + ) + .with_context("json", String::from_utf8_lossy(&text)) + .with_source(e) + })?; + Err(e.into()) + } + } + + async fn execute, const SUCCESS_CODE: u16>( + &self, + request: Request, + ) -> Result<()> { + let resp = self.0.execute(request).await?; + + if resp.status().as_u16() == SUCCESS_CODE { + Ok(()) + } else { + let text = resp.bytes().await?; + let e = serde_json::from_slice::(&text).map_err(|e| { + Error::new( + ErrorKind::Unexpected, + "Failed to parse response from rest catalog server!", + ) + .with_context("json", String::from_utf8_lossy(&text)) + .with_source(e) + })?; + Err(e.into()) + } + } +} + +/// Rest catalog implementation. +pub struct RestCatalog { + config: RestCatalogConfig, + client: HttpClient, +} + +#[async_trait] +impl Catalog for RestCatalog { + /// List namespaces from table. + async fn list_namespaces( + &self, + parent: Option<&NamespaceIdent>, + ) -> Result> { + let mut request = self.client.0.get(self.config.namespaces_endpoint()); + if let Some(ns) = parent { + request = request.query(&[("parent", ns.encode_in_url())]); + } + + let resp = self + .client + .query::(request.build()?) + .await?; + + resp.namespaces + .into_iter() + .map(NamespaceIdent::from_vec) + .collect::>>() + } + + /// Create a new namespace inside the catalog. + async fn create_namespace( + &self, + namespace: &NamespaceIdent, + properties: HashMap, + ) -> Result { + let request = self + .client + .0 + .post(self.config.namespaces_endpoint()) + .json(&NamespaceSerde { + namespace: namespace.as_ref().clone(), + properties: Some(properties), + }) + .build()?; + + let resp = self + .client + .query::(request) + .await?; + + Namespace::try_from(resp) + } + + /// Get a namespace information from the catalog. + async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result { + let request = self + .client + .0 + .get(self.config.namespace_endpoint(namespace)) + .build()?; + + let resp = self + .client + .query::(request) + .await?; + Namespace::try_from(resp) + } + + /// Update a namespace inside the catalog. + /// + /// # Behavior + /// + /// The properties must be the full set of namespace. + async fn update_namespace( + &self, + _namespace: &NamespaceIdent, + _properties: HashMap, + ) -> Result<()> { + Err(Error::new( + ErrorKind::FeatureUnsupported, + "Updating namespace not supported yet!", + )) + } + + async fn namespace_exists(&self, ns: &NamespaceIdent) -> Result { + let request = self + .client + .0 + .head(self.config.namespace_endpoint(ns)) + .build()?; + + self.client + .execute::(request) + .await + .map(|_| true) + } + + /// Drop a namespace from the catalog. + async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> { + let request = self + .client + .0 + .delete(self.config.namespace_endpoint(namespace)) + .build()?; + + self.client.execute::(request).await + } + + /// List tables from namespace. + async fn list_tables(&self, namespace: &NamespaceIdent) -> Result> { + let request = self + .client + .0 + .get(self.config.tables_endpoint(namespace)) + .build()?; + + let resp = self + .client + .query::(request) + .await?; + + Ok(resp.identifiers) + } + + /// Create a new table inside the namespace. + async fn create_table( + &self, + _namespace: &NamespaceIdent, + _creation: TableCreation, + ) -> Result { + Err(Error::new( + ErrorKind::FeatureUnsupported, + "Creating table not supported yet!", + )) + } + + /// Load table from the catalog. + async fn load_table(&self, _table: &TableIdent) -> Result
{ + Err(Error::new( + ErrorKind::FeatureUnsupported, + "Creating table not supported yet!", + )) + } + + /// Drop a table from the catalog. + async fn drop_table(&self, table: &TableIdent) -> Result<()> { + let request = self + .client + .0 + .delete(self.config.table_endpoint(table)) + .build()?; + + self.client.execute::(request).await + } + + /// Check if a table exists in the catalog. + async fn stat_table(&self, table: &TableIdent) -> Result { + let request = self + .client + .0 + .head(self.config.table_endpoint(table)) + .build()?; + + self.client + .execute::(request) + .await + .map(|_| true) + } + + /// Rename a table in the catalog. + async fn rename_table(&self, src: &TableIdent, dest: &TableIdent) -> Result<()> { + let request = self + .client + .0 + .post(self.config.rename_table_endpoint()) + .json(&RenameTableRequest { + source: src.clone(), + destination: dest.clone(), + }) + .build()?; + + self.client.execute::(request).await + } + + /// Update a table to the catalog. + async fn update_table(&self, _table: &TableIdent, _commit: TableCommit) -> Result
{ + todo!() + } + + /// Update multiple tables to the catalog as an atomic operation. + async fn update_tables(&self, _tables: &[(TableIdent, TableCommit)]) -> Result<()> { + todo!() + } +} + +impl RestCatalog { + /// Creates a rest catalog from config. + pub async fn new(config: RestCatalogConfig) -> Result { + let mut catalog = Self { + client: config.try_create_rest_client()?, + config, + }; + + catalog.update_config().await?; + catalog.client = catalog.config.try_create_rest_client()?; + + Ok(catalog) + } + + async fn update_config(&mut self) -> Result<()> { + let mut request = self.client.0.get(self.config.config_endpoint()); + + if let Some(warehouse_location) = &self.config.warehouse { + request = request.query(&[("warehouse", warehouse_location)]); + } + + let config = self + .client + .query::(request.build()?) + .await?; + + let mut props = config.defaults; + props.extend(self.config.props.clone()); + props.extend(config.overrides); + + self.config.props = props; + + Ok(()) + } +} + +/// Requests and responses for rest api. +mod _serde { + use std::collections::HashMap; + + use serde_derive::{Deserialize, Serialize}; + + use iceberg::{Error, ErrorKind, Namespace, TableIdent}; + + pub(super) const OK: u16 = 200u16; + pub(super) const NO_CONTENT: u16 = 204u16; + #[derive(Clone, Debug, Serialize, Deserialize)] + pub(super) struct CatalogConfig { + pub(super) overrides: HashMap, + pub(super) defaults: HashMap, + } + + #[derive(Debug, Serialize, Deserialize)] + pub(super) struct ErrorResponse { + error: ErrorModel, + } + + impl From for Error { + fn from(resp: ErrorResponse) -> Error { + resp.error.into() + } + } + + #[derive(Debug, Serialize, Deserialize)] + pub(super) struct ErrorModel { + pub(super) message: String, + pub(super) r#type: String, + pub(super) code: u16, + pub(super) stack: Option>, + } + + impl From for Error { + fn from(value: ErrorModel) -> Self { + let mut error = Error::new(ErrorKind::DataInvalid, value.message) + .with_context("type", value.r#type) + .with_context("code", format!("{}", value.code)); + + if let Some(stack) = value.stack { + error = error.with_context("stack", stack.join("\n")); + } + + error + } + } + + #[derive(Debug, Serialize, Deserialize)] + pub(super) struct OAuthError { + pub(super) error: String, + pub(super) error_description: Option, + pub(super) error_uri: Option, + } + + impl From for Error { + fn from(value: OAuthError) -> Self { + let mut error = Error::new( + ErrorKind::DataInvalid, + format!("OAuthError: {}", value.error), + ); + + if let Some(desc) = value.error_description { + error = error.with_context("description", desc); + } + + if let Some(uri) = value.error_uri { + error = error.with_context("uri", uri); + } + + error + } + } + + #[derive(Debug, Serialize, Deserialize)] + pub(super) struct NamespaceSerde { + pub(super) namespace: Vec, + pub(super) properties: Option>, + } + + impl TryFrom for super::Namespace { + type Error = Error; + fn try_from(value: NamespaceSerde) -> std::result::Result { + Ok(super::Namespace::with_properties( + super::NamespaceIdent::from_vec(value.namespace)?, + value.properties.unwrap_or_default(), + )) + } + } + + impl From<&Namespace> for NamespaceSerde { + fn from(value: &Namespace) -> Self { + Self { + namespace: value.name().as_ref().clone(), + properties: Some(value.properties().clone()), + } + } + } + + #[derive(Debug, Serialize, Deserialize)] + pub(super) struct ListNamespaceResponse { + pub(super) namespaces: Vec>, + } + + #[derive(Debug, Serialize, Deserialize)] + pub(super) struct UpdateNamespacePropsRequest { + removals: Option>, + updates: Option>, + } + + #[derive(Debug, Serialize, Deserialize)] + pub(super) struct UpdateNamespacePropsResponse { + updated: Vec, + removed: Vec, + missing: Option>, + } + + #[derive(Debug, Serialize, Deserialize)] + pub(super) struct ListTableResponse { + pub(super) identifiers: Vec, + } + + #[derive(Debug, Serialize, Deserialize)] + pub(super) struct RenameTableRequest { + pub(super) source: TableIdent, + pub(super) destination: TableIdent, + } +} + +#[cfg(test)] +mod tests { + use mockito::{Mock, Server, ServerGuard}; + + use super::*; + + #[tokio::test] + async fn test_update_config() { + 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" + }, + "defaults": {} + }"#, + ) + .create_async() + .await; + + let catalog = RestCatalog::new(RestCatalogConfig::builder().uri(server.url()).build()) + .await + .unwrap(); + + assert_eq!( + catalog.config.props.get("warehouse"), + Some(&"s3://iceberg-catalog".to_string()) + ); + + config_mock.assert_async().await; + } + + async fn create_config_mock(server: &mut ServerGuard) -> Mock { + server + .mock("GET", "/v1/config") + .with_status(200) + .with_body( + r#"{ + "overrides": { + "warehouse": "s3://iceberg-catalog" + }, + "defaults": {} + }"#, + ) + .create_async() + .await + } + + #[tokio::test] + async fn test_list_namespace() { + let mut server = Server::new_async().await; + + let config_mock = create_config_mock(&mut server).await; + + let list_ns_mock = server + .mock("GET", "/v1/namespaces") + .with_body( + r#"{ + "namespaces": [ + ["ns1", "ns11"], + ["ns2"] + ] + }"#, + ) + .create_async() + .await; + + let catalog = RestCatalog::new(RestCatalogConfig::builder().uri(server.url()).build()) + .await + .unwrap(); + + let namespaces = catalog.list_namespaces(None).await.unwrap(); + + let expected_ns = vec![ + NamespaceIdent::from_vec(vec!["ns1".to_string(), "ns11".to_string()]).unwrap(), + NamespaceIdent::from_vec(vec!["ns2".to_string()]).unwrap(), + ]; + + assert_eq!(expected_ns, namespaces); + + config_mock.assert_async().await; + list_ns_mock.assert_async().await; + } + + #[tokio::test] + async fn test_create_namespace() { + let mut server = Server::new_async().await; + + let config_mock = create_config_mock(&mut server).await; + + let create_ns_mock = server + .mock("POST", "/v1/namespaces") + .with_body( + r#"{ + "namespace": [ "ns1", "ns11"], + "properties" : { + "key1": "value1" + } + }"#, + ) + .create_async() + .await; + + let catalog = RestCatalog::new(RestCatalogConfig::builder().uri(server.url()).build()) + .await + .unwrap(); + + let namespaces = catalog + .create_namespace( + &NamespaceIdent::from_vec(vec!["ns1".to_string(), "ns11".to_string()]).unwrap(), + HashMap::from([("key1".to_string(), "value1".to_string())]), + ) + .await + .unwrap(); + + let expected_ns = Namespace::with_properties( + NamespaceIdent::from_vec(vec!["ns1".to_string(), "ns11".to_string()]).unwrap(), + HashMap::from([("key1".to_string(), "value1".to_string())]), + ); + + assert_eq!(expected_ns, namespaces); + + config_mock.assert_async().await; + create_ns_mock.assert_async().await; + } + + #[tokio::test] + async fn test_get_namespace() { + let mut server = Server::new_async().await; + + let config_mock = create_config_mock(&mut server).await; + + let get_ns_mock = server + .mock("GET", "/v1/namespaces/ns1") + .with_body( + r#"{ + "namespace": [ "ns1"], + "properties" : { + "key1": "value1" + } + }"#, + ) + .create_async() + .await; + + let catalog = RestCatalog::new(RestCatalogConfig::builder().uri(server.url()).build()) + .await + .unwrap(); + + let namespaces = catalog + .get_namespace(&NamespaceIdent::new("ns1".to_string())) + .await + .unwrap(); + + let expected_ns = Namespace::with_properties( + NamespaceIdent::new("ns1".to_string()), + HashMap::from([("key1".to_string(), "value1".to_string())]), + ); + + assert_eq!(expected_ns, namespaces); + + config_mock.assert_async().await; + get_ns_mock.assert_async().await; + } + + #[tokio::test] + async fn check_namespace_exists() { + let mut server = Server::new_async().await; + + let config_mock = create_config_mock(&mut server).await; + + let get_ns_mock = server + .mock("HEAD", "/v1/namespaces/ns1") + .with_status(204) + .create_async() + .await; + + let catalog = RestCatalog::new(RestCatalogConfig::builder().uri(server.url()).build()) + .await + .unwrap(); + + assert!(catalog + .namespace_exists(&NamespaceIdent::new("ns1".to_string())) + .await + .unwrap()); + + config_mock.assert_async().await; + get_ns_mock.assert_async().await; + } + + #[tokio::test] + async fn test_drop_namespace() { + let mut server = Server::new_async().await; + + let config_mock = create_config_mock(&mut server).await; + + let drop_ns_mock = server + .mock("DELETE", "/v1/namespaces/ns1") + .with_status(204) + .create_async() + .await; + + let catalog = RestCatalog::new(RestCatalogConfig::builder().uri(server.url()).build()) + .await + .unwrap(); + + catalog + .drop_namespace(&NamespaceIdent::new("ns1".to_string())) + .await + .unwrap(); + + config_mock.assert_async().await; + drop_ns_mock.assert_async().await; + } + + #[tokio::test] + async fn test_list_tables() { + let mut server = Server::new_async().await; + + let config_mock = create_config_mock(&mut server).await; + + let list_tables_mock = server + .mock("GET", "/v1/namespaces/ns1/tables") + .with_status(200) + .with_body( + r#"{ + "identifiers": [ + { + "namespace": ["ns1"], + "name": "table1" + }, + { + "namespace": ["ns1"], + "name": "table2" + } + ] + }"#, + ) + .create_async() + .await; + + let catalog = RestCatalog::new(RestCatalogConfig::builder().uri(server.url()).build()) + .await + .unwrap(); + + let tables = catalog + .list_tables(&NamespaceIdent::new("ns1".to_string())) + .await + .unwrap(); + + let expected_tables = vec![ + TableIdent::new(NamespaceIdent::new("ns1".to_string()), "table1".to_string()), + TableIdent::new(NamespaceIdent::new("ns1".to_string()), "table2".to_string()), + ]; + + assert_eq!(tables, expected_tables); + + config_mock.assert_async().await; + list_tables_mock.assert_async().await; + } + + #[tokio::test] + async fn test_drop_tables() { + let mut server = Server::new_async().await; + + let config_mock = create_config_mock(&mut server).await; + + let delete_table_mock = server + .mock("DELETE", "/v1/namespaces/ns1/tables/table1") + .with_status(204) + .create_async() + .await; + + let catalog = RestCatalog::new(RestCatalogConfig::builder().uri(server.url()).build()) + .await + .unwrap(); + + catalog + .drop_table(&TableIdent::new( + NamespaceIdent::new("ns1".to_string()), + "table1".to_string(), + )) + .await + .unwrap(); + + config_mock.assert_async().await; + delete_table_mock.assert_async().await; + } + + #[tokio::test] + async fn test_check_table_exists() { + let mut server = Server::new_async().await; + + let config_mock = create_config_mock(&mut server).await; + + let check_table_exists_mock = server + .mock("HEAD", "/v1/namespaces/ns1/tables/table1") + .with_status(204) + .create_async() + .await; + + let catalog = RestCatalog::new(RestCatalogConfig::builder().uri(server.url()).build()) + .await + .unwrap(); + + assert!(catalog + .stat_table(&TableIdent::new( + NamespaceIdent::new("ns1".to_string()), + "table1".to_string(), + )) + .await + .unwrap()); + + config_mock.assert_async().await; + check_table_exists_mock.assert_async().await; + } + + #[tokio::test] + async fn test_rename_table() { + let mut server = Server::new_async().await; + + let config_mock = create_config_mock(&mut server).await; + + let rename_table_mock = server + .mock("POST", "/v1/tables/rename") + .with_status(204) + .create_async() + .await; + + let catalog = RestCatalog::new(RestCatalogConfig::builder().uri(server.url()).build()) + .await + .unwrap(); + + catalog + .rename_table( + &TableIdent::new(NamespaceIdent::new("ns1".to_string()), "table1".to_string()), + &TableIdent::new(NamespaceIdent::new("ns1".to_string()), "table2".to_string()), + ) + .await + .unwrap(); + + config_mock.assert_async().await; + rename_table_mock.assert_async().await; + } +} diff --git a/crates/catalog/rest/src/lib.rs b/crates/catalog/rest/src/lib.rs new file mode 100644 index 0000000000..023fe7ab29 --- /dev/null +++ b/crates/catalog/rest/src/lib.rs @@ -0,0 +1,23 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +//! Iceberg REST API implementation. + +#![deny(missing_docs)] + +mod catalog; +pub use catalog::*; diff --git a/crates/iceberg/Cargo.toml b/crates/iceberg/Cargo.toml index 195f1949c9..1b2ba29b5d 100644 --- a/crates/iceberg/Cargo.toml +++ b/crates/iceberg/Cargo.toml @@ -41,10 +41,12 @@ either = "1" futures = "0.3" itertools = "0.11" lazy_static = "1" +log = "^0.4" murmur3 = "0.5.2" once_cell = "1" opendal = "0.40" ordered-float = "4.0.0" +reqwest = { version = "^0.11", features = ["json"] } rust_decimal = "1.31.0" serde = { version = "^1.0", features = ["rc"] } serde_bytes = "0.11.8" @@ -52,6 +54,7 @@ serde_derive = "^1.0" serde_json = "^1.0" serde_repr = "0.1.16" url = "2" +urlencoding = "2" uuid = "1.4.1" [dev-dependencies] diff --git a/crates/iceberg/src/catalog/mod.rs b/crates/iceberg/src/catalog/mod.rs index f50d78c8a0..d58eaa2bdf 100644 --- a/crates/iceberg/src/catalog/mod.rs +++ b/crates/iceberg/src/catalog/mod.rs @@ -17,11 +17,15 @@ //! Catalog API for Apache Iceberg +use serde_derive::{Deserialize, Serialize}; +use urlencoding::encode; + use crate::spec::{PartitionSpec, Schema, SortOrder}; use crate::table::Table; -use crate::Result; +use crate::{Error, ErrorKind, Result}; use async_trait::async_trait; use std::collections::HashMap; +use std::ops::Deref; /// The catalog API for Iceberg Rust. #[async_trait] @@ -40,6 +44,9 @@ pub trait Catalog { /// Get a namespace information from the catalog. async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result; + /// Check if namespace exists in catalog. + async fn namespace_exists(&self, namesace: &NamespaceIdent) -> Result; + /// Update a namespace inside the catalog. /// /// # Behavior @@ -88,6 +95,8 @@ pub trait Catalog { /// The namespace identifier is a list of strings, where each string is a /// component of the namespace. It's catalog implementer's responsibility to /// handle the namespace identifier correctly. + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Hash)] pub struct NamespaceIdent(Vec); impl NamespaceIdent { @@ -97,8 +106,24 @@ impl NamespaceIdent { } /// Create a multi-level namespace identifier from vector. - pub fn from_vec(names: Vec) -> Self { - Self(names) + pub fn from_vec(names: Vec) -> Result { + if names.is_empty() { + return Err(Error::new( + ErrorKind::DataInvalid, + "Namespace identifier can't be empty!", + )); + } + Ok(Self(names)) + } + + /// Returns url encoded format. + pub fn encode_in_url(&self) -> String { + encode(&self.as_ref().join("\u{1F}")).to_string() + } + + /// Returns inner strings. + pub fn inner(self) -> Vec { + self.0 } } @@ -108,7 +133,16 @@ impl AsRef> for NamespaceIdent { } } +impl Deref for NamespaceIdent { + type Target = [String]; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + /// Namespace represents a namespace in the catalog. +#[derive(Debug, Clone, PartialEq, Eq)] pub struct Namespace { name: NamespaceIdent, properties: HashMap, @@ -137,9 +171,12 @@ impl Namespace { } /// TableIdent represents the identifier of a table in the catalog. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Hash)] pub struct TableIdent { - namespace: NamespaceIdent, - name: String, + /// Namespace of the table. + pub namespace: NamespaceIdent, + /// Table name. + pub name: String, } impl TableIdent { diff --git a/crates/iceberg/src/error.rs b/crates/iceberg/src/error.rs index e4ae576d82..8948c7d6b0 100644 --- a/crates/iceberg/src/error.rs +++ b/crates/iceberg/src/error.rs @@ -307,6 +307,18 @@ define_from_err!( "Failed to parse url" ); +define_from_err!( + reqwest::Error, + ErrorKind::Unexpected, + "Failed to execute http request" +); + +define_from_err!( + serde_json::Error, + ErrorKind::DataInvalid, + "Failed to parse json string" +); + /// Helper macro to check arguments. /// /// diff --git a/crates/iceberg/src/rest.rs b/crates/iceberg/src/rest.rs new file mode 100644 index 0000000000..f20e96175e --- /dev/null +++ b/crates/iceberg/src/rest.rs @@ -0,0 +1,22 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +//! This module provide rest catalog implementation. + +pub struct RestCatalog { + url: String, +}