Merge branch 'main' into ct/remove-bound-partition-spec

Author: c-thiel

.asf.yaml

@@ -40,6 +40,7 @@ github:
 
       required_pull_request_reviews:
         required_approving_review_count: 1
+        dismiss_stale_reviews: true
 
       required_linear_history: true
   del_branch_on_merge: true

Cargo.toml

@@ -23,7 +23,6 @@ members = [
     "crates/iceberg",
     "crates/integration_tests",
     "crates/integrations/*",
-    "crates/puffin",
     "crates/test_utils",
 ]
 exclude = ["bindings/python"]
@@ -72,6 +71,7 @@ itertools = "0.13"
 log = "0.4"
 mockito = "1"
 murmur3 = "0.5.2"
+num-bigint = "0.4.6"
 once_cell = "1"
 opendal = "0.50.1"
 ordered-float = "4"

README.md

@@ -101,6 +101,7 @@ The Apache Iceberg community is built on the principles described in the [Apache
 
 - [Databend](https://github.com/datafuselabs/databend/): An open-source cloud data warehouse that serves as a cost-effective alternative to Snowflake.
 - [iceberg-catalog](https://github.com/hansetag/iceberg-catalog): A Rust implementation of the Iceberg REST Catalog specification.
+- [RisingWave](https://github.com/risingwavelabs/risingwave): A Postgres-compatible SQL database designed for real-time event streaming data processing, analysis, and management.
 
 ## License
 

crates/catalog/rest/src/catalog.rs

@@ -256,9 +256,10 @@ impl RestCatalog {
     async fn context(&self) -> Result<&RestContext> {
         self.ctx
             .get_or_try_init(|| async {
-                let catalog_config = RestCatalog::load_config(&self.user_config).await?;
+                let client = HttpClient::new(&self.user_config)?;
+                let catalog_config = RestCatalog::load_config(&client, &self.user_config).await?;
                 let config = self.user_config.clone().merge_with_config(catalog_config);
-                let client = HttpClient::new(&config)?;
+                let client = client.update_with(&config)?;
 
                 Ok(RestContext { config, client })
             })
@@ -268,9 +269,10 @@ impl RestCatalog {
     /// Load the runtime config from the server by user_config.
     ///
     /// It's required for a rest catalog to update it's config after creation.
-    async fn load_config(user_config: &RestCatalogConfig) -> Result<CatalogConfig> {
-        let client = HttpClient::new(user_config)?;
-
+    async fn load_config(
+        client: &HttpClient,
+        user_config: &RestCatalogConfig,
+    ) -> Result<CatalogConfig> {
         let mut request = client.request(Method::GET, user_config.config_endpoint());
 
         if let Some(warehouse_location) = &user_config.warehouse {
@@ -280,6 +282,7 @@ impl RestCatalog {
         let config = client
             .query::<CatalogConfig, ErrorResponse, OK>(request.build()?)
             .await?;
+
         Ok(config)
     }
 
@@ -777,7 +780,7 @@ mod tests {
                 "expires_in": 86400
                 }"#,
             )
-            .expect(2)
+            .expect(1)
             .create_async()
             .await
     }
@@ -831,7 +834,7 @@ mod tests {
                 "expires_in": 86400
                 }"#,
             )
-            .expect(2)
+            .expect(1)
             .create_async()
             .await;
 

crates/catalog/rest/src/client.rs

@@ -54,6 +54,7 @@ impl Debug for HttpClient {
 }
 
 impl HttpClient {
+    /// Create a new http client.
     pub fn new(cfg: &RestCatalogConfig) -> Result<Self> {
         Ok(HttpClient {
             client: Client::new(),
@@ -66,6 +67,32 @@ impl HttpClient {
         })
     }
 
+    /// Update the http client with new configuration.
+    ///
+    /// If cfg carries new value, we will use cfg instead.
+    /// Otherwise, we will keep the old value.
+    pub fn update_with(self, cfg: &RestCatalogConfig) -> Result<Self> {
+        Ok(HttpClient {
+            client: self.client,
+
+            token: Mutex::new(
+                cfg.token()
+                    .or_else(|| self.token.into_inner().ok().flatten()),
+            ),
+            token_endpoint: (!cfg.get_token_endpoint().is_empty())
+                .then(|| cfg.get_token_endpoint())
+                .unwrap_or(self.token_endpoint),
+            credential: cfg.credential().or(self.credential),
+            extra_headers: (!cfg.extra_headers()?.is_empty())
+                .then(|| cfg.extra_headers())
+                .transpose()?
+                .unwrap_or(self.extra_headers),
+            extra_oauth_params: (!cfg.extra_oauth_params().is_empty())
+                .then(|| cfg.extra_oauth_params())
+                .unwrap_or(self.extra_oauth_params),
+        })
+    }
+
     /// This API is testing only to assert the token.
     #[cfg(test)]
     pub(crate) async fn token(&self) -> Option<String> {
@@ -134,28 +161,39 @@ impl HttpClient {
             .request(Method::POST, &self.token_endpoint)
             .form(&params)
             .build()?;
+        let auth_url = auth_req.url().clone();
         let auth_resp = self.client.execute(auth_req).await?;
 
         let auth_res: TokenResponse = if auth_resp.status().as_u16() == OK {
-            let text = auth_resp.bytes().await?;
+            let text = auth_resp
+                .bytes()
+                .await
+                .map_err(|err| err.with_url(auth_url.clone()))?;
             Ok(serde_json::from_slice(&text).map_err(|e| {
                 Error::new(
                     ErrorKind::Unexpected,
                     "Failed to parse response from rest catalog server!",
                 )
+                .with_context("operation", "auth")
+                .with_context("url", auth_url.to_string())
                 .with_context("json", String::from_utf8_lossy(&text))
                 .with_source(e)
             })?)
         } else {
             let code = auth_resp.status();
-            let text = auth_resp.bytes().await?;
+            let text = auth_resp
+                .bytes()
+                .await
+                .map_err(|err| err.with_url(auth_url.clone()))?;
             let e: ErrorResponse = serde_json::from_slice(&text).map_err(|e| {
                 Error::new(
                     ErrorKind::Unexpected,
                     "Failed to parse response from rest catalog server!",
                 )
-                .with_context("json", String::from_utf8_lossy(&text))
                 .with_context("code", code.to_string())
+                .with_context("operation", "auth")
+                .with_context("url", auth_url.to_string())
+                .with_context("json", String::from_utf8_lossy(&text))
                 .with_source(e)
             })?;
             Err(Error::from(e))
@@ -193,28 +231,41 @@ impl HttpClient {
     ) -> Result<R> {
         self.authenticate(&mut request).await?;
 
+        let method = request.method().clone();
+        let url = request.url().clone();
+
         let resp = self.client.execute(request).await?;
 
         if resp.status().as_u16() == SUCCESS_CODE {
-            let text = resp.bytes().await?;
+            let text = resp
+                .bytes()
+                .await
+                .map_err(|err| err.with_url(url.clone()))?;
             Ok(serde_json::from_slice::<R>(&text).map_err(|e| {
                 Error::new(
                     ErrorKind::Unexpected,
                     "Failed to parse response from rest catalog server!",
                 )
+                .with_context("method", method.to_string())
+                .with_context("url", url.to_string())
                 .with_context("json", String::from_utf8_lossy(&text))
                 .with_source(e)
             })?)
         } else {
             let code = resp.status();
-            let text = resp.bytes().await?;
+            let text = resp
+                .bytes()
+                .await
+                .map_err(|err| err.with_url(url.clone()))?;
             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_context("code", code.to_string())
+                .with_context("method", method.to_string())
+                .with_context("url", url.to_string())
+                .with_context("json", String::from_utf8_lossy(&text))
                 .with_source(e)
             })?;
             Err(e.into())
@@ -227,20 +278,28 @@ impl HttpClient {
     ) -> Result<()> {
         self.authenticate(&mut request).await?;
 
+        let method = request.method().clone();
+        let url = request.url().clone();
+
         let resp = self.client.execute(request).await?;
 
         if resp.status().as_u16() == SUCCESS_CODE {
             Ok(())
         } else {
             let code = resp.status();
-            let text = resp.bytes().await?;
+            let text = resp
+                .bytes()
+                .await
+                .map_err(|err| err.with_url(url.clone()))?;
             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_context("code", code.to_string())
+                .with_context("method", method.to_string())
+                .with_context("url", url.to_string())
+                .with_context("json", String::from_utf8_lossy(&text))
                 .with_source(e)
             })?;
             Err(e.into())
@@ -255,19 +314,27 @@ impl HttpClient {
     ) -> Result<R> {
         self.authenticate(&mut request).await?;
 
+        let method = request.method().clone();
+        let url = request.url().clone();
+
         let resp = self.client.execute(request).await?;
 
         if let Some(ret) = handler(&resp) {
             Ok(ret)
         } else {
             let code = resp.status();
-            let text = resp.bytes().await?;
+            let text = resp
+                .bytes()
+                .await
+                .map_err(|err| err.with_url(url.clone()))?;
             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("code", code.to_string())
+                .with_context("method", method.to_string())
+                .with_context("url", url.to_string())
                 .with_context("json", String::from_utf8_lossy(&text))
                 .with_source(e)
             })?;

crates/iceberg/Cargo.toml

@@ -63,6 +63,7 @@ futures = { workspace = true }
 itertools = { workspace = true }
 moka = { version = "0.12.8", features = ["future"] }
 murmur3 = { workspace = true }
+num-bigint = { workspace = true }
 once_cell = { workspace = true }
 opendal = { workspace = true }
 ordered-float = { workspace = true }
@@ -81,6 +82,7 @@ tokio = { workspace = true, optional = true }
 typed-builder = { workspace = true }
 url = { workspace = true }
 uuid = { workspace = true }
+zstd = { workspace = true }
 
 [dev-dependencies]
 ctor = { workspace = true }