You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "liurenjie1024 (via GitHub)" <gi...@apache.org> on 2023/10/13 03:29:31 UTC
[PR] feat: First version of rest catalog. [iceberg-rust]
liurenjie1024 opened a new pull request, #78:
URL: https://github.com/apache/iceberg-rust/pull/78
In this pr we add initial support for rest, which finished simple rest apis.
Complex apis such as create table, update table, commits which be added in following pr so that we can make each pr's size reasonable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1360330437
##########
crates/iceberg-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-rest"
Review Comment:
How about calling it `iceberg-catalog-rest`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1360370023
##########
crates/iceberg-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-rest"
Review Comment:
Make sense. The rest api is mostly about catalog operations, so adding `catalog` prefix is not confusing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1375302993
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -0,0 +1,845 @@
+// 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 = "0.14.1";
+const PATH_V1: &str = "v1";
+
+/// Rest catalog configuration.
+#[derive(Debug, TypedBuilder)]
+pub struct RestCatalogConfig {
+ uri: String,
+ #[builder(default, setter(strip_option))]
+ warehouse: Option<String>,
+
+ #[builder(default)]
+ props: HashMap<String, String>,
+}
+
+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<HttpClient> {
+ //TODO: We will add oauth, ssl config, sigv4 later
+ let mut headers = HeaderMap::new();
+ headers.insert(
+ header::CONTENT_TYPE,
+ HeaderValue::from_static("application/json"),
+ );
+ headers.insert(
+ HeaderName::from_static("x-client-version"),
+ HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
+ );
+ headers.insert(
+ header::USER_AGENT,
+ HeaderValue::from_str(&format!("iceberg-rs/{}", env!("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<Error>,
+ const SUCCESS_CODE: u16,
+ >(
+ &self,
+ request: Request,
+ ) -> Result<R> {
+ let resp = self.0.execute(request).await?;
+
+ if resp.status().as_u16() == SUCCESS_CODE {
+ let text = resp.bytes().await?;
+ Ok(serde_json::from_slice::<R>(&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::<E>(&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<E: DeserializeOwned + Into<Error>, 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::<E>(&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<Vec<NamespaceIdent>> {
+ 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::<ListNamespaceResponse, ErrorModel, OK>(request.build()?)
+ .await?;
+
+ resp.namespaces
+ .into_iter()
+ .map(NamespaceIdent::from_vec)
+ .collect::<Result<Vec<NamespaceIdent>>>()
+ }
+
+ /// Create a new namespace inside the catalog.
+ async fn create_namespace(
+ &self,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
+ ) -> Result<Namespace> {
+ 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::<NamespaceSerde, ErrorModel, OK>(request)
+ .await?;
+
+ Namespace::try_from(resp)
+ }
+
+ /// Get a namespace information from the catalog.
+ async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
+ let request = self
+ .client
+ .0
+ .get(self.config.namespace_endpoint(namespace))
+ .build()?;
+
+ let resp = self
+ .client
+ .query::<NamespaceSerde, ErrorModel, OK>(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<String, String>,
+ ) -> Result<()> {
+ Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Updating namespace not supported yet!",
+ ))
+ }
+
+ /// 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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// List tables from namespace.
+ async fn list_tables(&self, namespace: &NamespaceIdent) -> Result<Vec<TableIdent>> {
+ let request = self
+ .client
+ .0
+ .get(self.config.tables_endpoint(namespace))
+ .build()?;
+
+ let resp = self
+ .client
+ .query::<ListTableResponse, ErrorModel, OK>(request)
+ .await?;
+
+ Ok(resp.identifiers)
+ }
+
+ /// Create a new table inside the namespace.
+ async fn create_table(
+ &self,
+ _namespace: &NamespaceIdent,
+ _creation: TableCreation,
+ ) -> Result<Table> {
+ Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Creating table not supported yet!",
+ ))
+ }
+
+ /// Load table from the catalog.
+ async fn load_table(&self, _table: &TableIdent) -> Result<Table> {
+ 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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// Check if a table exists in the catalog.
+ async fn stat_table(&self, table: &TableIdent) -> Result<bool> {
+ let request = self
+ .client
+ .0
+ .head(self.config.table_endpoint(table))
+ .build()?;
+
+ self.client
+ .execute::<ErrorModel, NO_CONTENT>(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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// Update a table to the catalog.
+ async fn update_table(&self, _table: &TableIdent, _commit: TableCommit) -> Result<Table> {
+ todo!()
+ }
+
+ /// Update multiple tables to the catalog as an atomic operation.
+ async fn update_tables(&self, _tables: &[(TableIdent, TableCommit)]) -> Result<()> {
+ todo!()
+ }
Review Comment:
Thank you! Just some context around the concern: Within Iceberg, when creating a new table, the IDs are being reassigned ([python](https://github.com/apache/iceberg-python/blob/0a077fcc0dceb95ca4316505bba1f41c60f8f74f/pyiceberg/schema.py#L1221-L1273), [java](https://github.com/apache/iceberg/blob/385a6bd296e551bdfdab48d2a291d3516f999f5a/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java#L27)). This can lead to confusion and potentially also errors. Therefore we refrain users from assigning IDs as much as possible (or reassigning them).
```python
tbl = table.create_table(
Schema(
Field(1, "id", IntegerType()),
Field(2, "str", StringType())
)
)
tbl.set_schema(
Schema(
Field(1, "id", IntegerType()),
Field(3, "dt", DateType()),
Field(2, "str", StringType())
)
)
```
Because the IDs are re-assigned, it will evolve from:
```
Schema {
1: id
2: str
}
```
to:
```
Schema {
1: id
2: dt
3: str
}
```
And this will corrupt the table since there is no valid promotion from string to date. The general consensus is that users shouldn't care about the IDs themselves, and this should not be exposed to users through APIs. The API should handle it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1357818399
##########
crates/iceberg/src/catalog/rest.rs:
##########
@@ -0,0 +1,900 @@
+// 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 urlencoding::encode;
+
+use crate::error::Result;
+use crate::table::Table;
+use crate::{
+ 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 = "0.14.1";
+const PATH_V1: &str = "v1";
+
+#[derive(Debug, Builder)]
+pub struct RestCatalogConfig {
+ uri: String,
+ #[builder(default)]
+ warehouse: Option<String>,
+
+ #[builder(default)]
+ props: HashMap<String, String>,
+}
+
+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<HttpClient> {
+ //TODO: We will add oauth, ssl config, sigv4 later
+ let mut headers = HeaderMap::new();
+ headers.insert(
+ header::CONTENT_TYPE,
+ HeaderValue::from_static("application/json"),
+ );
+ headers.insert(
+ HeaderName::from_static("x-client-version"),
+ HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
+ );
+ headers.insert(
+ header::USER_AGENT,
+ HeaderValue::from_str(&format!("iceberg-rs/{}", env!("CARGO_PKG_VERSION"))).unwrap(),
+ );
+
+ Ok(HttpClient(
+ Client::builder().default_headers(headers).build()?,
+ ))
+ }
+}
+
+impl NamespaceIdent {
+ /// Returns url encoded format.
+ pub fn encode_in_url(&self) -> String {
+ encode(&self.0.join("\u{1F}")).to_string()
+ }
+}
+
+struct HttpClient(Client);
+
+impl HttpClient {
+ async fn execute<
+ R: DeserializeOwned,
+ E: DeserializeOwned + Into<Error>,
+ const SUCCESS_CODE: u16,
+ >(
+ &self,
+ request: Request,
+ ) -> Result<R> {
+ let resp = self.0.execute(request).await?;
+
+ if resp.status().as_u16() == SUCCESS_CODE {
+ let text = resp.bytes().await?;
+ Ok(serde_json::from_slice::<R>(&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::<E>(&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 execute2<E: DeserializeOwned + Into<Error>, const SUCCESS_CODE: u16>(
Review Comment:
Is the difference between execute2 and execute is response? Can we give a more clear name to distinguish them🤔, such as:
- query - means that there is response
- execute - means that there is no response
##########
crates/iceberg/src/catalog/rest.rs:
##########
@@ -0,0 +1,900 @@
+// 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 urlencoding::encode;
+
+use crate::error::Result;
+use crate::table::Table;
+use crate::{
+ 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 = "0.14.1";
+const PATH_V1: &str = "v1";
+
+#[derive(Debug, Builder)]
+pub struct RestCatalogConfig {
+ uri: String,
+ #[builder(default)]
+ warehouse: Option<String>,
+
+ #[builder(default)]
+ props: HashMap<String, String>,
+}
+
+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<HttpClient> {
+ //TODO: We will add oauth, ssl config, sigv4 later
+ let mut headers = HeaderMap::new();
+ headers.insert(
+ header::CONTENT_TYPE,
+ HeaderValue::from_static("application/json"),
+ );
+ headers.insert(
+ HeaderName::from_static("x-client-version"),
+ HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
+ );
+ headers.insert(
+ header::USER_AGENT,
+ HeaderValue::from_str(&format!("iceberg-rs/{}", env!("CARGO_PKG_VERSION"))).unwrap(),
+ );
+
+ Ok(HttpClient(
+ Client::builder().default_headers(headers).build()?,
+ ))
+ }
+}
+
+impl NamespaceIdent {
+ /// Returns url encoded format.
+ pub fn encode_in_url(&self) -> String {
+ encode(&self.0.join("\u{1F}")).to_string()
+ }
+}
+
+struct HttpClient(Client);
+
+impl HttpClient {
+ async fn execute<
+ R: DeserializeOwned,
+ E: DeserializeOwned + Into<Error>,
+ const SUCCESS_CODE: u16,
+ >(
+ &self,
+ request: Request,
+ ) -> Result<R> {
+ let resp = self.0.execute(request).await?;
+
+ if resp.status().as_u16() == SUCCESS_CODE {
+ let text = resp.bytes().await?;
+ Ok(serde_json::from_slice::<R>(&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::<E>(&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 execute2<E: DeserializeOwned + Into<Error>, const SUCCESS_CODE: u16>(
Review Comment:
Is the difference between execute2 and execute is response? Can we give a more clear name to distinguish them🤔, such as:
- query - means that there is response
- execute - means that there is no response
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#issuecomment-1760707832
cc @JanKaul @Xuanwo @Fokko @ZENOTME PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1360337146
##########
crates/iceberg-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-rest"
Review Comment:
I leave it as `iceberg-rest` on purpose. Maybe in future we can build rest catalog server in rust.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1357738084
##########
crates/iceberg/Cargo.toml:
##########
@@ -41,20 +41,24 @@ 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"] }
Review Comment:
The problem with the features is that they are add-only, making it difficult for users to disable them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1360184684
##########
crates/iceberg/Cargo.toml:
##########
@@ -41,20 +41,24 @@ 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"] }
Review Comment:
Cool, I'll move it a separate crate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#issuecomment-1772894697
cc @Fokko Any other comments?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1357791341
##########
crates/iceberg/Cargo.toml:
##########
@@ -41,20 +41,24 @@ 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"] }
Review Comment:
Sorry, I don't get your point, would you give a concrete example?
One concern with separate crate approach is that it makes loading catalog dynamically like Python difficult.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1357791341
##########
crates/iceberg/Cargo.toml:
##########
@@ -41,20 +41,24 @@ 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"] }
Review Comment:
> The problem with the features is that they are add-only, making it difficult for users to disable them.
Sorry, I don't get your point, would you give a concrete example?
One concern with separate crate approach is that it makes loading catalog dynamically like Python difficult.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1364823837
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -0,0 +1,845 @@
+// 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 = "0.14.1";
+const PATH_V1: &str = "v1";
+
+/// Rest catalog configuration.
+#[derive(Debug, TypedBuilder)]
+pub struct RestCatalogConfig {
+ uri: String,
+ #[builder(default, setter(strip_option))]
+ warehouse: Option<String>,
+
+ #[builder(default)]
+ props: HashMap<String, String>,
+}
+
+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<HttpClient> {
+ //TODO: We will add oauth, ssl config, sigv4 later
+ let mut headers = HeaderMap::new();
+ headers.insert(
+ header::CONTENT_TYPE,
+ HeaderValue::from_static("application/json"),
+ );
+ headers.insert(
+ HeaderName::from_static("x-client-version"),
+ HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
+ );
+ headers.insert(
+ header::USER_AGENT,
+ HeaderValue::from_str(&format!("iceberg-rs/{}", env!("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<Error>,
+ const SUCCESS_CODE: u16,
+ >(
+ &self,
+ request: Request,
+ ) -> Result<R> {
+ let resp = self.0.execute(request).await?;
+
+ if resp.status().as_u16() == SUCCESS_CODE {
+ let text = resp.bytes().await?;
+ Ok(serde_json::from_slice::<R>(&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::<E>(&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<E: DeserializeOwned + Into<Error>, 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::<E>(&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<Vec<NamespaceIdent>> {
+ 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::<ListNamespaceResponse, ErrorModel, OK>(request.build()?)
+ .await?;
+
+ resp.namespaces
+ .into_iter()
+ .map(NamespaceIdent::from_vec)
+ .collect::<Result<Vec<NamespaceIdent>>>()
+ }
+
+ /// Create a new namespace inside the catalog.
+ async fn create_namespace(
+ &self,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
+ ) -> Result<Namespace> {
+ 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::<NamespaceSerde, ErrorModel, OK>(request)
+ .await?;
+
+ Namespace::try_from(resp)
+ }
+
+ /// Get a namespace information from the catalog.
+ async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
+ let request = self
+ .client
+ .0
+ .get(self.config.namespace_endpoint(namespace))
+ .build()?;
+
+ let resp = self
+ .client
+ .query::<NamespaceSerde, ErrorModel, OK>(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<String, String>,
+ ) -> Result<()> {
+ Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Updating namespace not supported yet!",
+ ))
+ }
+
+ /// 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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// List tables from namespace.
+ async fn list_tables(&self, namespace: &NamespaceIdent) -> Result<Vec<TableIdent>> {
+ let request = self
+ .client
+ .0
+ .get(self.config.tables_endpoint(namespace))
+ .build()?;
+
+ let resp = self
+ .client
+ .query::<ListTableResponse, ErrorModel, OK>(request)
+ .await?;
+
+ Ok(resp.identifiers)
+ }
+
+ /// Create a new table inside the namespace.
+ async fn create_table(
+ &self,
+ _namespace: &NamespaceIdent,
+ _creation: TableCreation,
+ ) -> Result<Table> {
+ Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Creating table not supported yet!",
+ ))
+ }
+
+ /// Load table from the catalog.
+ async fn load_table(&self, _table: &TableIdent) -> Result<Table> {
+ 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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// Check if a table exists in the catalog.
+ async fn stat_table(&self, table: &TableIdent) -> Result<bool> {
+ let request = self
+ .client
+ .0
+ .head(self.config.table_endpoint(table))
+ .build()?;
+
+ self.client
+ .execute::<ErrorModel, NO_CONTENT>(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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// Update a table to the catalog.
+ async fn update_table(&self, _table: &TableIdent, _commit: TableCommit) -> Result<Table> {
+ todo!()
+ }
+
+ /// Update multiple tables to the catalog as an atomic operation.
+ async fn update_tables(&self, _tables: &[(TableIdent, TableCommit)]) -> Result<()> {
+ todo!()
+ }
Review Comment:
Yes, I agree. It should be called by transaction api rather by user directly. The main challenge is that we should limit its visibility rather than exposing directly to user. I will figure out a way when actually implementing it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1364807441
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -0,0 +1,845 @@
+// 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 = "0.14.1";
+const PATH_V1: &str = "v1";
+
+/// Rest catalog configuration.
+#[derive(Debug, TypedBuilder)]
+pub struct RestCatalogConfig {
+ uri: String,
+ #[builder(default, setter(strip_option))]
+ warehouse: Option<String>,
+
+ #[builder(default)]
+ props: HashMap<String, String>,
+}
+
+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<HttpClient> {
+ //TODO: We will add oauth, ssl config, sigv4 later
+ let mut headers = HeaderMap::new();
Review Comment:
Good catch!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1360351583
##########
crates/iceberg-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-rest"
Review Comment:
I am considering `iceberg-catalog-hive` and `iceberg-catalog-jdbc`. If we follow the same naming style as `iceberg-rest`, it might be a bit confusing. For instance, `iceberg-hive` and `iceberg-jdbc`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1357732820
##########
crates/iceberg/Cargo.toml:
##########
@@ -41,20 +41,24 @@ 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"] }
Review Comment:
Maybe we should make it a feature? I'm not sure if it deserves another crate. cc @JanKaul @ZENOTME How do you guys think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1364818198
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -0,0 +1,845 @@
+// 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 = "0.14.1";
Review Comment:
Updated to `1.14`, and add `NamespaceExists` method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko merged PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1357725856
##########
crates/iceberg/src/catalog/rest.rs:
##########
@@ -0,0 +1,912 @@
+// 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 urlencoding::encode;
+
+use crate::error::Result;
+use crate::table::Table;
+use crate::{
+ 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 = "0.14.1";
+const PATH_V1: &str = "v1";
+
+#[derive(Debug, Builder)]
+pub struct RestCatalogConfig {
+ uri: String,
+ #[builder(default)]
+ warehouse: Option<String>,
+
+ #[builder(default)]
+ props: HashMap<String, String>,
+}
+
+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) -> Result<String> {
+ Ok([&self.uri, PATH_V1, "namespaces", &ns.encode_in_url()?].join("/"))
+ }
+
+ fn tables_endpoint(&self, ns: &NamespaceIdent) -> Result<String> {
+ Ok([
+ &self.uri,
+ PATH_V1,
+ "namespaces",
+ &ns.encode_in_url()?,
+ "tables",
+ ]
+ .join("/"))
+ }
+
+ fn rename_table_endpoint(&self) -> Result<String> {
+ Ok([&self.uri, PATH_V1, "tables", "rename"].join("/"))
+ }
+
+ fn table_endpoint(&self, table: &TableIdent) -> Result<String> {
+ Ok([
+ &self.uri,
+ PATH_V1,
+ "namespaces",
+ &table.namespace.encode_in_url()?,
+ "tables",
+ encode(&table.name).as_ref(),
+ ]
+ .join("/"))
+ }
+
+ fn try_create_rest_client(&self) -> Result<HttpClient> {
+ //TODO: We will add oauth, ssl config, sigv4 later
+ let mut headers = HeaderMap::new();
+ headers.insert(
+ header::CONTENT_TYPE,
+ HeaderValue::from_static("application/json"),
+ );
+ headers.insert(
+ HeaderName::from_static("x-client-version"),
+ HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
+ );
+ headers.insert(
+ header::USER_AGENT,
+ HeaderValue::from_str(&format!("iceberg-rs/{}", env!("CARGO_PKG_VERSION"))).unwrap(),
+ );
+
+ Ok(HttpClient(
+ Client::builder().default_headers(headers).build()?,
+ ))
+ }
+}
+
+impl NamespaceIdent {
+ /// Returns url encoded format.
+ pub fn encode_in_url(&self) -> Result<String> {
+ if self.0.is_empty() {
Review Comment:
It's better to ensure that `NamespaceIdent` is valid so that we don't have to check it when using it. This change can removes a lot of `Result<T>` in related APIs.
##########
crates/iceberg/src/catalog/rest.rs:
##########
@@ -0,0 +1,912 @@
+// 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 urlencoding::encode;
+
+use crate::error::Result;
+use crate::table::Table;
+use crate::{
+ 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 = "0.14.1";
+const PATH_V1: &str = "v1";
+
+#[derive(Debug, Builder)]
+pub struct RestCatalogConfig {
+ uri: String,
+ #[builder(default)]
+ warehouse: Option<String>,
+
+ #[builder(default)]
+ props: HashMap<String, String>,
+}
+
+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) -> Result<String> {
+ Ok([&self.uri, PATH_V1, "namespaces", &ns.encode_in_url()?].join("/"))
+ }
+
+ fn tables_endpoint(&self, ns: &NamespaceIdent) -> Result<String> {
+ Ok([
+ &self.uri,
+ PATH_V1,
+ "namespaces",
+ &ns.encode_in_url()?,
+ "tables",
+ ]
+ .join("/"))
+ }
+
+ fn rename_table_endpoint(&self) -> Result<String> {
+ Ok([&self.uri, PATH_V1, "tables", "rename"].join("/"))
+ }
+
+ fn table_endpoint(&self, table: &TableIdent) -> Result<String> {
+ Ok([
+ &self.uri,
+ PATH_V1,
+ "namespaces",
+ &table.namespace.encode_in_url()?,
+ "tables",
+ encode(&table.name).as_ref(),
+ ]
+ .join("/"))
+ }
+
+ fn try_create_rest_client(&self) -> Result<HttpClient> {
+ //TODO: We will add oauth, ssl config, sigv4 later
+ let mut headers = HeaderMap::new();
+ headers.insert(
+ header::CONTENT_TYPE,
+ HeaderValue::from_static("application/json"),
+ );
+ headers.insert(
+ HeaderName::from_static("x-client-version"),
+ HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
+ );
+ headers.insert(
+ header::USER_AGENT,
+ HeaderValue::from_str(&format!("iceberg-rs/{}", env!("CARGO_PKG_VERSION"))).unwrap(),
+ );
+
+ Ok(HttpClient(
+ Client::builder().default_headers(headers).build()?,
+ ))
+ }
+}
+
+impl NamespaceIdent {
+ /// Returns url encoded format.
+ pub fn encode_in_url(&self) -> Result<String> {
+ if self.0.is_empty() {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ "Can't encode empty namespace in url!",
+ ));
+ }
+
+ Ok(encode(&self.0.join("\u{1F}")).to_string())
+ }
+}
+
+struct HttpClient(Client);
+
+impl HttpClient {
+ async fn execute<
+ R: DeserializeOwned,
+ E: DeserializeOwned + Into<Error>,
+ const SUCCESS_CODE: u16,
+ >(
+ &self,
+ request: Request,
+ ) -> Result<R> {
+ let resp = self.0.execute(request).await?;
+
+ if resp.status().as_u16() == SUCCESS_CODE {
+ let text = resp.text().await?;
+ log::debug!("Response text is: {text}");
+ Ok(serde_json::from_slice::<R>(text.as_bytes()).map_err(|e| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "Failed to parse response from rest catalog server!",
+ )
+ .with_context("json", text)
+ .with_source(e)
+ })?)
+ } else {
+ let text = resp.text().await?;
+ log::debug!("Response text is: {text}");
+ let e = serde_json::from_slice::<E>(text.as_bytes()).map_err(|e| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "Failed to parse response from rest catalog server!",
+ )
+ .with_context("json", text)
+ .with_source(e)
+ })?;
+ Err(e.into())
+ }
+ }
+
+ async fn execute2<E: DeserializeOwned + Into<Error>, 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.text().await?;
+ log::debug!("Response text is: {text}");
+ let e = serde_json::from_slice::<E>(text.as_bytes()).map_err(|e| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "Failed to parse response from rest catalog server!",
+ )
+ .with_context("json", text)
+ .with_source(e)
+ })?;
+ Err(e.into())
+ }
+ }
+}
+
+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<Vec<NamespaceIdent>> {
+ 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
+ .execute::<ListNamespaceResponse, ErrorModel, OK>(request.build()?)
+ .await?;
+
+ Ok(resp
+ .namespaces
+ .into_iter()
+ .map(NamespaceIdent::from_vec)
Review Comment:
from_vec should returns `Result<T>` to make sure `NamespaceIdent` is valid.
##########
crates/iceberg/Cargo.toml:
##########
@@ -41,20 +41,24 @@ 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"] }
Review Comment:
Do we need to put rest catalog in a new crate? I'm guessing not all iceberg users need this.
##########
crates/iceberg/src/catalog/rest.rs:
##########
@@ -0,0 +1,912 @@
+// 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 urlencoding::encode;
+
+use crate::error::Result;
+use crate::table::Table;
+use crate::{
+ 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 = "0.14.1";
+const PATH_V1: &str = "v1";
+
+#[derive(Debug, Builder)]
+pub struct RestCatalogConfig {
+ uri: String,
+ #[builder(default)]
+ warehouse: Option<String>,
+
+ #[builder(default)]
+ props: HashMap<String, String>,
+}
+
+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) -> Result<String> {
+ Ok([&self.uri, PATH_V1, "namespaces", &ns.encode_in_url()?].join("/"))
+ }
+
+ fn tables_endpoint(&self, ns: &NamespaceIdent) -> Result<String> {
+ Ok([
+ &self.uri,
+ PATH_V1,
+ "namespaces",
+ &ns.encode_in_url()?,
+ "tables",
+ ]
+ .join("/"))
+ }
+
+ fn rename_table_endpoint(&self) -> Result<String> {
+ Ok([&self.uri, PATH_V1, "tables", "rename"].join("/"))
+ }
+
+ fn table_endpoint(&self, table: &TableIdent) -> Result<String> {
+ Ok([
+ &self.uri,
+ PATH_V1,
+ "namespaces",
+ &table.namespace.encode_in_url()?,
+ "tables",
+ encode(&table.name).as_ref(),
+ ]
+ .join("/"))
+ }
+
+ fn try_create_rest_client(&self) -> Result<HttpClient> {
+ //TODO: We will add oauth, ssl config, sigv4 later
+ let mut headers = HeaderMap::new();
+ headers.insert(
+ header::CONTENT_TYPE,
+ HeaderValue::from_static("application/json"),
+ );
+ headers.insert(
+ HeaderName::from_static("x-client-version"),
+ HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
+ );
+ headers.insert(
+ header::USER_AGENT,
+ HeaderValue::from_str(&format!("iceberg-rs/{}", env!("CARGO_PKG_VERSION"))).unwrap(),
+ );
+
+ Ok(HttpClient(
+ Client::builder().default_headers(headers).build()?,
+ ))
+ }
+}
+
+impl NamespaceIdent {
+ /// Returns url encoded format.
+ pub fn encode_in_url(&self) -> Result<String> {
+ if self.0.is_empty() {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ "Can't encode empty namespace in url!",
+ ));
+ }
+
+ Ok(encode(&self.0.join("\u{1F}")).to_string())
+ }
+}
+
+struct HttpClient(Client);
+
+impl HttpClient {
+ async fn execute<
+ R: DeserializeOwned,
+ E: DeserializeOwned + Into<Error>,
+ const SUCCESS_CODE: u16,
+ >(
+ &self,
+ request: Request,
+ ) -> Result<R> {
+ let resp = self.0.execute(request).await?;
+
+ if resp.status().as_u16() == SUCCESS_CODE {
+ let text = resp.text().await?;
Review Comment:
Using `resp.bytes()` here to avoid not need UTC-8 checks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "JanKaul (via GitHub)" <gi...@apache.org>.
JanKaul commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1360172363
##########
crates/iceberg/Cargo.toml:
##########
@@ -41,20 +41,24 @@ 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"] }
Review Comment:
I'm also in favor of using a separate crate for the rest catalog. This way we avoid bringing in dependencies for catalogs that don't need them and that might prevent the crate from compiling, like wasm/wasi for example.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1365327942
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -0,0 +1,845 @@
+// 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 = "0.14.1";
+const PATH_V1: &str = "v1";
+
+/// Rest catalog configuration.
+#[derive(Debug, TypedBuilder)]
+pub struct RestCatalogConfig {
+ uri: String,
+ #[builder(default, setter(strip_option))]
+ warehouse: Option<String>,
+
+ #[builder(default)]
+ props: HashMap<String, String>,
+}
+
+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<HttpClient> {
+ //TODO: We will add oauth, ssl config, sigv4 later
+ let mut headers = HeaderMap::new();
+ headers.insert(
+ header::CONTENT_TYPE,
+ HeaderValue::from_static("application/json"),
+ );
+ headers.insert(
+ HeaderName::from_static("x-client-version"),
+ HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
+ );
+ headers.insert(
+ header::USER_AGENT,
+ HeaderValue::from_str(&format!("iceberg-rs/{}", env!("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<Error>,
+ const SUCCESS_CODE: u16,
+ >(
+ &self,
+ request: Request,
+ ) -> Result<R> {
+ let resp = self.0.execute(request).await?;
+
+ if resp.status().as_u16() == SUCCESS_CODE {
+ let text = resp.bytes().await?;
+ Ok(serde_json::from_slice::<R>(&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::<E>(&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<E: DeserializeOwned + Into<Error>, 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::<E>(&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<Vec<NamespaceIdent>> {
+ 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::<ListNamespaceResponse, ErrorModel, OK>(request.build()?)
+ .await?;
+
+ resp.namespaces
+ .into_iter()
+ .map(NamespaceIdent::from_vec)
+ .collect::<Result<Vec<NamespaceIdent>>>()
+ }
+
+ /// Create a new namespace inside the catalog.
+ async fn create_namespace(
+ &self,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
+ ) -> Result<Namespace> {
+ 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::<NamespaceSerde, ErrorModel, OK>(request)
+ .await?;
+
+ Namespace::try_from(resp)
+ }
+
+ /// Get a namespace information from the catalog.
+ async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
+ let request = self
+ .client
+ .0
+ .get(self.config.namespace_endpoint(namespace))
+ .build()?;
+
+ let resp = self
+ .client
+ .query::<NamespaceSerde, ErrorModel, OK>(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<String, String>,
+ ) -> Result<()> {
+ Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Updating namespace not supported yet!",
+ ))
+ }
+
+ /// 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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// List tables from namespace.
+ async fn list_tables(&self, namespace: &NamespaceIdent) -> Result<Vec<TableIdent>> {
+ let request = self
+ .client
+ .0
+ .get(self.config.tables_endpoint(namespace))
+ .build()?;
+
+ let resp = self
+ .client
+ .query::<ListTableResponse, ErrorModel, OK>(request)
+ .await?;
+
+ Ok(resp.identifiers)
+ }
+
+ /// Create a new table inside the namespace.
+ async fn create_table(
+ &self,
+ _namespace: &NamespaceIdent,
+ _creation: TableCreation,
+ ) -> Result<Table> {
+ Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Creating table not supported yet!",
+ ))
+ }
+
+ /// Load table from the catalog.
+ async fn load_table(&self, _table: &TableIdent) -> Result<Table> {
+ 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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// Check if a table exists in the catalog.
+ async fn stat_table(&self, table: &TableIdent) -> Result<bool> {
+ let request = self
+ .client
+ .0
+ .head(self.config.table_endpoint(table))
+ .build()?;
+
+ self.client
+ .execute::<ErrorModel, NO_CONTENT>(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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// Update a table to the catalog.
+ async fn update_table(&self, _table: &TableIdent, _commit: TableCommit) -> Result<Table> {
+ todo!()
+ }
+
+ /// Update multiple tables to the catalog as an atomic operation.
+ async fn update_tables(&self, _tables: &[(TableIdent, TableCommit)]) -> Result<()> {
+ todo!()
+ }
Review Comment:
As I have said in pr description, the goal of this pr is to implement simple apis. Others will be left in following pr.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: First version of rest catalog. [iceberg-rust]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #78:
URL: https://github.com/apache/iceberg-rust/pull/78#discussion_r1363673663
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -0,0 +1,845 @@
+// 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 = "0.14.1";
Review Comment:
I think you've copied this from the PyIceberg repository. I think we need to bump this. Along the way, quite a few fixes have been made to the spec.
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -0,0 +1,845 @@
+// 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 = "0.14.1";
+const PATH_V1: &str = "v1";
+
+/// Rest catalog configuration.
+#[derive(Debug, TypedBuilder)]
+pub struct RestCatalogConfig {
+ uri: String,
+ #[builder(default, setter(strip_option))]
+ warehouse: Option<String>,
+
+ #[builder(default)]
+ props: HashMap<String, String>,
+}
+
+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<HttpClient> {
+ //TODO: We will add oauth, ssl config, sigv4 later
+ let mut headers = HeaderMap::new();
Review Comment:
I'm learning from @Xuanwo: https://github.com/apache/iceberg-rust/pull/76/files#r1363173256
It looks like we can use `from_iter` on the `HeaderMap` as well: https://docs.rs/http/latest/http/header/struct.HeaderMap.html
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -0,0 +1,845 @@
+// 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 = "0.14.1";
+const PATH_V1: &str = "v1";
+
+/// Rest catalog configuration.
+#[derive(Debug, TypedBuilder)]
+pub struct RestCatalogConfig {
+ uri: String,
+ #[builder(default, setter(strip_option))]
+ warehouse: Option<String>,
+
+ #[builder(default)]
+ props: HashMap<String, String>,
+}
+
+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<HttpClient> {
+ //TODO: We will add oauth, ssl config, sigv4 later
+ let mut headers = HeaderMap::new();
+ headers.insert(
+ header::CONTENT_TYPE,
+ HeaderValue::from_static("application/json"),
+ );
+ headers.insert(
+ HeaderName::from_static("x-client-version"),
+ HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
+ );
+ headers.insert(
+ header::USER_AGENT,
+ HeaderValue::from_str(&format!("iceberg-rs/{}", env!("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<Error>,
+ const SUCCESS_CODE: u16,
+ >(
+ &self,
+ request: Request,
+ ) -> Result<R> {
+ let resp = self.0.execute(request).await?;
+
+ if resp.status().as_u16() == SUCCESS_CODE {
+ let text = resp.bytes().await?;
+ Ok(serde_json::from_slice::<R>(&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::<E>(&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<E: DeserializeOwned + Into<Error>, 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::<E>(&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<Vec<NamespaceIdent>> {
+ 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::<ListNamespaceResponse, ErrorModel, OK>(request.build()?)
+ .await?;
+
+ resp.namespaces
+ .into_iter()
+ .map(NamespaceIdent::from_vec)
+ .collect::<Result<Vec<NamespaceIdent>>>()
+ }
+
+ /// Create a new namespace inside the catalog.
+ async fn create_namespace(
+ &self,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
+ ) -> Result<Namespace> {
+ 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::<NamespaceSerde, ErrorModel, OK>(request)
+ .await?;
+
+ Namespace::try_from(resp)
+ }
+
+ /// Get a namespace information from the catalog.
+ async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
+ let request = self
+ .client
+ .0
+ .get(self.config.namespace_endpoint(namespace))
+ .build()?;
+
+ let resp = self
+ .client
+ .query::<NamespaceSerde, ErrorModel, OK>(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<String, String>,
+ ) -> Result<()> {
+ Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Updating namespace not supported yet!",
+ ))
+ }
+
+ /// 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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// List tables from namespace.
+ async fn list_tables(&self, namespace: &NamespaceIdent) -> Result<Vec<TableIdent>> {
+ let request = self
+ .client
+ .0
+ .get(self.config.tables_endpoint(namespace))
+ .build()?;
+
+ let resp = self
+ .client
+ .query::<ListTableResponse, ErrorModel, OK>(request)
+ .await?;
+
+ Ok(resp.identifiers)
+ }
+
+ /// Create a new table inside the namespace.
+ async fn create_table(
+ &self,
+ _namespace: &NamespaceIdent,
+ _creation: TableCreation,
+ ) -> Result<Table> {
+ Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Creating table not supported yet!",
+ ))
+ }
+
+ /// Load table from the catalog.
+ async fn load_table(&self, _table: &TableIdent) -> Result<Table> {
+ 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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// Check if a table exists in the catalog.
+ async fn stat_table(&self, table: &TableIdent) -> Result<bool> {
+ let request = self
+ .client
+ .0
+ .head(self.config.table_endpoint(table))
+ .build()?;
+
+ self.client
+ .execute::<ErrorModel, NO_CONTENT>(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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// Update a table to the catalog.
+ async fn update_table(&self, _table: &TableIdent, _commit: TableCommit) -> Result<Table> {
+ todo!()
+ }
+
+ /// Update multiple tables to the catalog as an atomic operation.
+ async fn update_tables(&self, _tables: &[(TableIdent, TableCommit)]) -> Result<()> {
+ todo!()
+ }
Review Comment:
I'm not sure if this is the way that we want to expose update tables. It is important that the right updates and requirements are set, otherwise, race conditions might occur. Also, there is little validation here, for example, can you do two distinct schema changes (so two updates), in a single commit? I would maybe leave this out for now so we can decide later on.
We might want to introduce a similar API as PyIceberg (which is inspired on Java): https://py.iceberg.apache.org/api/#schema-evolution
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -0,0 +1,845 @@
+// 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 = "0.14.1";
+const PATH_V1: &str = "v1";
+
+/// Rest catalog configuration.
+#[derive(Debug, TypedBuilder)]
+pub struct RestCatalogConfig {
+ uri: String,
+ #[builder(default, setter(strip_option))]
+ warehouse: Option<String>,
+
+ #[builder(default)]
+ props: HashMap<String, String>,
+}
+
+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<HttpClient> {
+ //TODO: We will add oauth, ssl config, sigv4 later
+ let mut headers = HeaderMap::new();
+ headers.insert(
+ header::CONTENT_TYPE,
+ HeaderValue::from_static("application/json"),
+ );
+ headers.insert(
+ HeaderName::from_static("x-client-version"),
+ HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
+ );
+ headers.insert(
+ header::USER_AGENT,
+ HeaderValue::from_str(&format!("iceberg-rs/{}", env!("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<Error>,
+ const SUCCESS_CODE: u16,
+ >(
+ &self,
+ request: Request,
+ ) -> Result<R> {
+ let resp = self.0.execute(request).await?;
+
+ if resp.status().as_u16() == SUCCESS_CODE {
+ let text = resp.bytes().await?;
+ Ok(serde_json::from_slice::<R>(&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::<E>(&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<E: DeserializeOwned + Into<Error>, 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::<E>(&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<Vec<NamespaceIdent>> {
+ 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::<ListNamespaceResponse, ErrorModel, OK>(request.build()?)
+ .await?;
+
+ resp.namespaces
+ .into_iter()
+ .map(NamespaceIdent::from_vec)
+ .collect::<Result<Vec<NamespaceIdent>>>()
+ }
+
+ /// Create a new namespace inside the catalog.
+ async fn create_namespace(
+ &self,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
+ ) -> Result<Namespace> {
+ 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::<NamespaceSerde, ErrorModel, OK>(request)
+ .await?;
+
+ Namespace::try_from(resp)
+ }
+
+ /// Get a namespace information from the catalog.
+ async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
+ let request = self
+ .client
+ .0
+ .get(self.config.namespace_endpoint(namespace))
+ .build()?;
+
+ let resp = self
+ .client
+ .query::<NamespaceSerde, ErrorModel, OK>(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<String, String>,
+ ) -> Result<()> {
+ Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Updating namespace not supported yet!",
+ ))
+ }
+
+ /// 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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// List tables from namespace.
+ async fn list_tables(&self, namespace: &NamespaceIdent) -> Result<Vec<TableIdent>> {
+ let request = self
+ .client
+ .0
+ .get(self.config.tables_endpoint(namespace))
+ .build()?;
+
+ let resp = self
+ .client
+ .query::<ListTableResponse, ErrorModel, OK>(request)
+ .await?;
+
+ Ok(resp.identifiers)
+ }
+
+ /// Create a new table inside the namespace.
+ async fn create_table(
+ &self,
+ _namespace: &NamespaceIdent,
+ _creation: TableCreation,
+ ) -> Result<Table> {
+ Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Creating table not supported yet!",
+ ))
+ }
+
+ /// Load table from the catalog.
+ async fn load_table(&self, _table: &TableIdent) -> Result<Table> {
+ 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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// Check if a table exists in the catalog.
+ async fn stat_table(&self, table: &TableIdent) -> Result<bool> {
+ let request = self
+ .client
+ .0
+ .head(self.config.table_endpoint(table))
+ .build()?;
+
+ self.client
+ .execute::<ErrorModel, NO_CONTENT>(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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// Update a table to the catalog.
+ async fn update_table(&self, _table: &TableIdent, _commit: TableCommit) -> Result<Table> {
+ 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<Self> {
+ 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 mut config = self
+ .client
+ .query::<CatalogConfig, ErrorResponse, OK>(request.build()?)
+ .await?;
+
+ config.defaults.extend(self.config.props.clone());
+ config.defaults.extend(config.overrides);
+
+ self.config.props = config.defaults;
+
+ 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<String, String>,
+ pub(super) defaults: HashMap<String, String>,
+ }
+
+ #[derive(Debug, Serialize, Deserialize)]
+ pub(super) struct ErrorResponse {
+ error: ErrorModel,
+ }
+
+ impl From<ErrorResponse> 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: i32,
Review Comment:
Nit: We use u16 for the error code in other places.
```suggestion
pub(super) code: u16,
```
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -0,0 +1,845 @@
+// 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 = "0.14.1";
Review Comment:
The intent of this header is mostly debugging
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -0,0 +1,845 @@
+// 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 = "0.14.1";
Review Comment:
For example, the NamespaceExists had been added recently: https://github.com/apache/iceberg/pull/8569
##########
crates/catalog/rest/src/catalog.rs:
##########
@@ -0,0 +1,845 @@
+// 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 = "0.14.1";
+const PATH_V1: &str = "v1";
+
+/// Rest catalog configuration.
+#[derive(Debug, TypedBuilder)]
+pub struct RestCatalogConfig {
+ uri: String,
+ #[builder(default, setter(strip_option))]
+ warehouse: Option<String>,
+
+ #[builder(default)]
+ props: HashMap<String, String>,
+}
+
+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<HttpClient> {
+ //TODO: We will add oauth, ssl config, sigv4 later
+ let mut headers = HeaderMap::new();
+ headers.insert(
+ header::CONTENT_TYPE,
+ HeaderValue::from_static("application/json"),
+ );
+ headers.insert(
+ HeaderName::from_static("x-client-version"),
+ HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION),
+ );
+ headers.insert(
+ header::USER_AGENT,
+ HeaderValue::from_str(&format!("iceberg-rs/{}", env!("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<Error>,
+ const SUCCESS_CODE: u16,
+ >(
+ &self,
+ request: Request,
+ ) -> Result<R> {
+ let resp = self.0.execute(request).await?;
+
+ if resp.status().as_u16() == SUCCESS_CODE {
+ let text = resp.bytes().await?;
+ Ok(serde_json::from_slice::<R>(&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::<E>(&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<E: DeserializeOwned + Into<Error>, 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::<E>(&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<Vec<NamespaceIdent>> {
+ 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::<ListNamespaceResponse, ErrorModel, OK>(request.build()?)
+ .await?;
+
+ resp.namespaces
+ .into_iter()
+ .map(NamespaceIdent::from_vec)
+ .collect::<Result<Vec<NamespaceIdent>>>()
+ }
+
+ /// Create a new namespace inside the catalog.
+ async fn create_namespace(
+ &self,
+ namespace: &NamespaceIdent,
+ properties: HashMap<String, String>,
+ ) -> Result<Namespace> {
+ 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::<NamespaceSerde, ErrorModel, OK>(request)
+ .await?;
+
+ Namespace::try_from(resp)
+ }
+
+ /// Get a namespace information from the catalog.
+ async fn get_namespace(&self, namespace: &NamespaceIdent) -> Result<Namespace> {
+ let request = self
+ .client
+ .0
+ .get(self.config.namespace_endpoint(namespace))
+ .build()?;
+
+ let resp = self
+ .client
+ .query::<NamespaceSerde, ErrorModel, OK>(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<String, String>,
+ ) -> Result<()> {
+ Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Updating namespace not supported yet!",
+ ))
+ }
+
+ /// 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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// List tables from namespace.
+ async fn list_tables(&self, namespace: &NamespaceIdent) -> Result<Vec<TableIdent>> {
+ let request = self
+ .client
+ .0
+ .get(self.config.tables_endpoint(namespace))
+ .build()?;
+
+ let resp = self
+ .client
+ .query::<ListTableResponse, ErrorModel, OK>(request)
+ .await?;
+
+ Ok(resp.identifiers)
+ }
+
+ /// Create a new table inside the namespace.
+ async fn create_table(
+ &self,
+ _namespace: &NamespaceIdent,
+ _creation: TableCreation,
+ ) -> Result<Table> {
+ Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "Creating table not supported yet!",
+ ))
+ }
+
+ /// Load table from the catalog.
+ async fn load_table(&self, _table: &TableIdent) -> Result<Table> {
+ 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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// Check if a table exists in the catalog.
+ async fn stat_table(&self, table: &TableIdent) -> Result<bool> {
+ let request = self
+ .client
+ .0
+ .head(self.config.table_endpoint(table))
+ .build()?;
+
+ self.client
+ .execute::<ErrorModel, NO_CONTENT>(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::<ErrorModel, NO_CONTENT>(request).await
+ }
+
+ /// Update a table to the catalog.
+ async fn update_table(&self, _table: &TableIdent, _commit: TableCommit) -> Result<Table> {
+ 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<Self> {
+ 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 mut config = self
+ .client
+ .query::<CatalogConfig, ErrorResponse, OK>(request.build()?)
+ .await?;
+
+ config.defaults.extend(self.config.props.clone());
Review Comment:
This is correct, but it looks a bit odd because you overwrite the default (which is fine)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org