Sync df55 Iceberg fixes and handle Horizon no-content responses

catalogs/iceberg-file-catalog/src/lib.rs

@@ -465,9 +465,7 @@ impl Catalog for FileCatalog {
             ));
         };
         if !check_table_requirements(&commit.requirements, &metadata) {
-            return Err(IcebergError::InvalidFormat(
-                "Table requirements not valid".to_owned(),
-            ));
+            return Err(IcebergError::CommitConflict(identifier.to_string()));
         }
         apply_table_updates(&mut metadata, commit.updates)?;
         let temp_metadata_location = new_metadata_location(&metadata);
@@ -487,7 +485,13 @@ impl Catalog for FileCatalog {
                 &strip_prefix(&temp_metadata_location).into(),
                 &strip_prefix(&metadata_location).into(),
             )
-            .await?;
+            .await
+            .map_err(|error| match error {
+                object_store::Error::AlreadyExists { .. } => {
+                    IcebergError::CommitConflict(identifier.to_string())
+                }
+                error => error.into(),
+            })?;
 
         object_store.put_version_hint(&metadata_location).await.ok();
 
@@ -951,7 +955,14 @@ pub mod tests {
     };
     use futures::StreamExt;
     use iceberg_rust::{
-        catalog::{create::CreateTableBuilder, namespace::Namespace, Catalog},
+        catalog::{
+            commit::{CommitTable, TableRequirement},
+            create::CreateTableBuilder,
+            namespace::Namespace,
+            tabular::Tabular,
+            Catalog,
+        },
+        error::Error as IcebergError,
         object_store::{Bucket, ObjectStoreBuilder},
         spec::{
             schema::Schema,
@@ -1012,6 +1023,79 @@ pub mod tests {
         assert_eq!(tabular.identifier(), &destination);
     }
 
+    #[tokio::test]
+    async fn conflicting_table_commits_report_commit_conflict() {
+        let store = ObjectStoreBuilder::memory();
+        let first_catalog = Arc::new(
+            FileCatalog::new("/dev/embucket", store.clone())
+                .await
+                .unwrap(),
+        );
+        let second_catalog = Arc::new(FileCatalog::new("/dev/embucket", store).await.unwrap());
+        let namespace = Namespace::try_new(&["public".to_owned()]).unwrap();
+        let identifier =
+            iceberg_rust::catalog::identifier::Identifier::new(&namespace, "conflicting_table");
+        let schema = Schema::builder()
+            .with_struct_field(StructField {
+                id: 1,
+                name: "id".to_owned(),
+                required: true,
+                field_type: Type::Primitive(PrimitiveType::Long),
+                doc: None,
+                initial_default: None,
+                write_default: None,
+            })
+            .build()
+            .unwrap();
+        let mut builder = CreateTableBuilder::default();
+        let mut first = builder
+            .with_name(identifier.name())
+            .with_schema(schema)
+            .build(&namespace, first_catalog.clone())
+            .await
+            .unwrap();
+        let Tabular::Table(mut stale) = second_catalog
+            .clone()
+            .load_tabular(&identifier)
+            .await
+            .unwrap()
+        else {
+            panic!("expected table");
+        };
+
+        let rejected_requirement = first_catalog
+            .clone()
+            .update_table(CommitTable {
+                identifier: identifier.clone(),
+                requirements: vec![TableRequirement::AssertRefSnapshotId {
+                    r#ref: "main".to_owned(),
+                    snapshot_id: Some(123),
+                }],
+                updates: vec![],
+            })
+            .await;
+        assert!(matches!(
+            rejected_requirement,
+            Err(IcebergError::CommitConflict(_))
+        ));
+
+        first
+            .new_transaction(None)
+            .update_properties(vec![("writer".to_owned(), "first".to_owned())])
+            .commit()
+            .await
+            .unwrap();
+        let stale_version = stale
+            .new_transaction(None)
+            .update_properties(vec![("writer".to_owned(), "second".to_owned())])
+            .commit()
+            .await;
+        assert!(matches!(
+            stale_version,
+            Err(IcebergError::CommitConflict(_))
+        ));
+    }
+
     #[tokio::test]
     async fn test_create_update_drop_table() {
         let localstack = LocalStack::default()

catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs

@@ -881,7 +881,7 @@ pub async fn namespace_exists(
 
     let method = reqwest::Method::HEAD;
 
-    fetch::fetch(configuration, method, prefix, &uri_str, &(), None, None).await
+    fetch::fetch_unit(configuration, method, prefix, &uri_str, &(), None, None).await
 }
 
 /// Register a table using given metadata file location.
@@ -919,7 +919,7 @@ pub async fn rename_table(
     let uri_str = "tables/rename".to_string();
     let method = reqwest::Method::POST;
 
-    fetch::fetch(
+    fetch::fetch_unit(
         configuration,
         method,
         prefix,
@@ -940,7 +940,7 @@ pub async fn rename_view(
     let uri_str = "views/rename".to_string();
     let method = reqwest::Method::POST;
 
-    fetch::fetch(
+    fetch::fetch_unit(
         configuration,
         method,
         prefix,

catalogs/iceberg-rest-catalog/src/apis/fetch.rs

@@ -109,6 +109,49 @@ where
     }
 }
 
+pub(crate) async fn fetch_unit<R, E>(
+    configuration: &configuration::Configuration,
+    method: reqwest::Method,
+    prefix: Option<&str>,
+    uri_str: &str,
+    request: &R,
+    headers: Option<HashMap<String, String>>,
+    query_params: Option<HashMap<String, String>>,
+) -> Result<(), Error<E>>
+where
+    R: serde::Serialize + ?Sized,
+    E: serde::de::DeserializeOwned,
+{
+    let response = fetch_inner(
+        configuration,
+        method,
+        prefix,
+        uri_str,
+        request,
+        headers,
+        query_params,
+    )
+    .await?;
+    let status = response.status();
+    let content = response.text().await?;
+    unit_response(status, content)
+}
+
+fn unit_response<E>(status: reqwest::StatusCode, content: String) -> Result<(), Error<E>>
+where
+    E: serde::de::DeserializeOwned,
+{
+    if status.is_success() {
+        return Ok(());
+    }
+    let entity = serde_json::from_str(&content).ok();
+    Err(Error::ResponseError(ResponseContent {
+        status,
+        content,
+        entity,
+    }))
+}
+
 pub(crate) async fn fetch_conditional<R, T, E>(
     configuration: &configuration::Configuration,
     method: reqwest::Method,
@@ -176,3 +219,34 @@ fn build_uri_base(configuration: &configuration::Configuration, prefix: Option<&
         None => format!("{}/v1/", configuration.base_path),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn unit_response_accepts_empty_and_nonempty_success_bodies() {
+        for (status, body) in [
+            (reqwest::StatusCode::NO_CONTENT, ""),
+            (reqwest::StatusCode::OK, "null"),
+            (reqwest::StatusCode::OK, "\"NO_CONTENT\""),
+        ] {
+            assert!(unit_response::<serde_json::Value>(status, body.to_string()).is_ok());
+        }
+    }
+
+    #[test]
+    fn unit_response_keeps_error_status_and_body() {
+        let body = r#"{"error":"invalid rename"}"#;
+        let result =
+            unit_response::<serde_json::Value>(reqwest::StatusCode::BAD_REQUEST, body.to_string());
+        match result {
+            Err(Error::ResponseError(response)) => {
+                assert_eq!(response.status, reqwest::StatusCode::BAD_REQUEST);
+                assert_eq!(response.content, body);
+                assert_eq!(response.entity, serde_json::from_str(body).ok());
+            }
+            _ => panic!("expected REST response error"),
+        }
+    }
+}

catalogs/iceberg-rest-catalog/src/catalog.rs

@@ -1,7 +1,6 @@
 use crate::{
     apis::{
-        self,
-        catalog_api_api::{self, NamespaceExistsError},
+        self, catalog_api_api,
         configuration::{self, Configuration},
     },
     models::{self, StorageCredential},
@@ -178,17 +177,12 @@ impl Catalog for RestCatalog {
         .await
         {
             Ok(()) => Ok(true),
-            Err(err) => {
-                if let apis::Error::ResponseError(err) = err {
-                    if let Some(NamespaceExistsError::Status404(_)) = err.entity {
-                        Ok(false)
-                    } else {
-                        Err(apis::Error::ResponseError(err).into())
-                    }
-                } else {
-                    Err(err.into())
-                }
+            Err(apis::Error::ResponseError(response))
+                if response.status == reqwest::StatusCode::NOT_FOUND =>
+            {
+                Ok(false)
             }
+            Err(error) => Err(error.into()),
         }
     }
     /// Lists all tables in the given namespace.
@@ -754,6 +748,62 @@ pub mod tests {
             aws_v4_key: None,
         }
     }
+
+    #[tokio::test]
+    async fn namespace_exists_uses_status_for_bodyless_head_response() {
+        for (response_status, expected) in [
+            ("200 OK", Some(true)),
+            ("404 Not Found", Some(false)),
+            ("403 Forbidden", None),
+        ] {
+            let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+            listener.set_nonblocking(true).unwrap();
+            let address = listener.local_addr().unwrap();
+            let server = std::thread::spawn(move || {
+                use std::io::{Read, Write};
+
+                let deadline = std::time::Instant::now() + Duration::from_secs(5);
+                let (mut stream, _) = loop {
+                    match listener.accept() {
+                        Ok(connection) => break connection,
+                        Err(error)
+                            if error.kind() == std::io::ErrorKind::WouldBlock
+                                && std::time::Instant::now() < deadline =>
+                        {
+                            std::thread::sleep(Duration::from_millis(10));
+                        }
+                        Err(error) => panic!("local REST test server did not accept: {error}"),
+                    }
+                };
+                stream
+                    .set_read_timeout(Some(Duration::from_secs(5)))
+                    .unwrap();
+                let mut request = [0_u8; 1024];
+                let size = stream.read(&mut request).unwrap();
+                assert!(String::from_utf8_lossy(&request[..size])
+                    .starts_with("HEAD /v1/catalog/namespaces/PUBLIC "));
+                write!(
+                    stream,
+                    "HTTP/1.1 {response_status}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
+                )
+                .unwrap();
+            });
+            let catalog = RestCatalog::new(
+                Some("catalog"),
+                configuration(&format!("http://{address}")),
+                None,
+                false,
+            );
+            let namespace = Namespace::try_new(&["PUBLIC".to_string()]).unwrap();
+            let result = catalog.namespace_exists(&namespace).await;
+            server.join().unwrap();
+            match expected {
+                Some(exists) => assert_eq!(result.unwrap(), exists),
+                None => assert!(result.is_err()),
+            }
+        }
+    }
+
     #[tokio::test]
     async fn test_create_update_drop_table() {
         let container_host = if is_podman() {