From 227210fe6e4bb5e60a71dbfe80e806dc375d4e41 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Sat, 10 Oct 2026 15:02:38 +0300 Subject: [PATCH 1/7] Report file catalog table commit conflicts consistently --- catalogs/iceberg-file-catalog/src/lib.rs | 94 ++++++++++++++++++++++-- 1 file changed, 89 insertions(+), 5 deletions(-) diff --git a/catalogs/iceberg-file-catalog/src/lib.rs b/catalogs/iceberg-file-catalog/src/lib.rs index a2a24b51..9839695e 100644 --- a/catalogs/iceberg-file-catalog/src/lib.rs +++ b/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() From 489329d472c14a239437e401aa2f3d3805bab374 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Mon, 12 Oct 2026 00:24:30 +0300 Subject: [PATCH 2/7] Prune transformed partition null predicates safely --- datafusion_iceberg/src/pruning_statistics.rs | 260 +++++++++++++++++-- datafusion_iceberg/tests/position_delete.rs | 97 +++++++ 2 files changed, 341 insertions(+), 16 deletions(-) diff --git a/datafusion_iceberg/src/pruning_statistics.rs b/datafusion_iceberg/src/pruning_statistics.rs index 23cc2f76..04a9ad91 100644 --- a/datafusion_iceberg/src/pruning_statistics.rs +++ b/datafusion_iceberg/src/pruning_statistics.rs @@ -17,7 +17,7 @@ use std::{any::Any, collections::HashSet, sync::Arc}; use crate::error::Error as DatafusionIcebergError; use datafusion::{ arrow::{ - array::ArrayRef, + array::{ArrayRef, Int32Array, Int64Array}, datatypes::{DataType, Schema as ArrowSchema, TimeUnit}, }, common::DataFusionError, @@ -35,7 +35,7 @@ use iceberg_rust::{ spec::{ decimal::{decimal_mantissa, decimal_scale, Decimal}, manifest::{Content, ManifestEntry}, - manifest_list::ManifestListEntry, + manifest_list::{Content as ManifestContent, ManifestListEntry}, partition::{BoundPartitionField, Transform}, schema::Schema, types::{PrimitiveType, Type}, @@ -62,6 +62,11 @@ impl<'table, 'manifests> PruneManifests<'table, 'manifests> { files, } } + + fn compatible_data_manifest(&self, manifest: &ManifestListEntry) -> bool { + manifest.partition_spec_id == self.partition_spec_id + && manifest.content == ManifestContent::Data + } } impl PruningStatistics for PruneManifests<'_, '_> { @@ -115,15 +120,16 @@ impl PruningStatistics for PruneManifests<'_, '_> { .partition_fields .iter() .enumerate() - .find(|(_, field)| field.source_name() == column.name())?; + .find(|(_, field)| field.name() == column.name())?; let contains_null = self.files.iter().map(|manifest| { - (manifest.partition_spec_id == self.partition_spec_id) + self.compatible_data_manifest(manifest) .then_some(manifest) .and_then(|manifest| manifest.partitions.as_ref()) .and_then(|partitions| partitions.get(index)) .and_then(|partition| (!partition.contains_null).then_some(0)) }); - ScalarValue::iter_to_array(contains_null.map(ScalarValue::Int32)).ok() + let counts: ArrayRef = Arc::new(Int32Array::from_iter(contains_null)); + Some(counts) } fn contained( &self, @@ -135,12 +141,11 @@ impl PruningStatistics for PruneManifests<'_, '_> { fn row_counts(&self) -> Option { let row_counts = self.files.iter().map(|x| { - match ( - x.added_rows_count, - x.existing_rows_count, - x.deleted_rows_count, - ) { - (Some(a), Some(e), Some(d)) => Some(a + e - d), + if !self.compatible_data_manifest(x) { + return None; + } + match (x.added_rows_count, x.existing_rows_count) { + (Some(a), Some(e)) => a.checked_add(e), _ => None, } }); @@ -206,8 +211,25 @@ impl PruningStatistics for PrunePartitionedDataFiles<'_, '_> { self.files.len() } - fn null_counts(&self, _column: &Column) -> Option { - None + fn null_counts(&self, column: &Column) -> Option { + self.partition_fields + .iter() + .find(|field| field.name() == column.name())?; + let counts = self.files.iter().map(|(path, entry)| { + let file = entry.data_file(); + if file.content() != &Content::Data + || !self.compatible_manifests.contains(path.as_str()) + { + return None; + } + match file.partition().get(column.name()) { + Some(Some(_)) => Some(0), + Some(None) => Some(*file.record_count()), + None => None, + } + }); + let counts: ArrayRef = Arc::new(Int64Array::from_iter(counts)); + Some(counts) } fn contained( @@ -425,6 +447,34 @@ pub(crate) fn transform_predicate( return Some(expr); } + let null_check = match &expr { + Expr::IsNull(inner) => Some((inner.as_ref(), true)), + Expr::IsNotNull(inner) => Some((inner.as_ref(), false)), + _ => None, + }; + if let Some((Expr::Column(column), is_null)) = null_check { + let field = partition_fields + .iter() + .find(|field| field.source_name() == column.name())?; + if !matches!( + field.transform(), + Transform::Identity + | Transform::Year + | Transform::Month + | Transform::Day + | Transform::Hour + | Transform::Truncate(_) + ) { + return None; + } + let partition_column = Expr::Column(Column::new(column.relation.clone(), field.name())); + return Some(if is_null { + Expr::IsNull(Box::new(partition_column)) + } else { + Expr::IsNotNull(Box::new(partition_column)) + }); + } + let Expr::BinaryExpr(BinaryExpr { left, op, right }) = expr else { return None; }; @@ -810,9 +860,106 @@ mod tests { } #[test] - fn partitioned_file_pruning_preserves_unknown_specs_and_delete_files() { + fn manifest_null_counts_use_transformed_partition_name_and_spec() { use datafusion::arrow::array::Int32Array; + let source = StructField::new( + 1, + "ts", + false, + Type::Primitive(PrimitiveType::Timestamp), + None, + ); + let partition = PartitionField::new(1, 1000, "ts_year", Transform::Year); + let fields = [BoundPartitionField::new(&partition, &source)]; + let entry = |spec_id, contains_null| ManifestListEntry { + format_version: FormatVersion::V2, + manifest_path: format!("/{spec_id}.avro"), + manifest_length: 1, + partition_spec_id: spec_id, + content: ManifestContent::Data, + sequence_number: 1, + min_sequence_number: 1, + added_snapshot_id: 1, + added_files_count: Some(1), + existing_files_count: Some(0), + deleted_files_count: Some(0), + added_rows_count: Some(1), + existing_rows_count: Some(0), + deleted_rows_count: Some(0), + partitions: Some(vec![FieldSummary { + contains_null, + contains_nan: None, + lower_bound: None, + upper_bound: None, + }]), + key_metadata: None, + first_row_id: None, + }; + let mut delete_manifest = entry(1, false); + delete_manifest.content = ManifestContent::Deletes; + delete_manifest.added_rows_count = Some(0); + let mut mixed_manifest = entry(1, false); + mixed_manifest.deleted_rows_count = Some(1); + let manifests = vec![ + entry(1, false), + entry(1, true), + entry(0, false), + delete_manifest, + mixed_manifest, + ]; + let pruning = PruneManifests::new(&fields, 1, &manifests); + assert!(pruning.null_counts(&Column::from_name("ts")).is_none()); + let counts = pruning.null_counts(&Column::from_name("ts_year")).unwrap(); + let counts = counts.as_any().downcast_ref::().unwrap(); + assert_eq!(counts.value(0), 0); + assert!(counts.is_null(1)); + assert!(counts.is_null(2)); + assert!(counts.is_null(3)); + assert_eq!(counts.value(4), 0); + + let row_counts = pruning.row_counts().unwrap(); + let row_counts = row_counts.as_any().downcast_ref::().unwrap(); + assert!(row_counts.is_null(2)); + assert!(row_counts.is_null(3)); + assert_eq!(row_counts.value(4), 1); + + let schema = Arc::new(ArrowSchema::new(vec![Field::new( + "ts_year", + DataType::Int32, + true, + )])); + let session = SessionContext::new(); + let state = session.state(); + for (expr, expected) in [ + ( + Expr::IsNull(Box::new(Expr::Column(Column::from_name("ts_year")))), + vec![false, true, true, true, false], + ), + ( + Expr::IsNotNull(Box::new(Expr::Column(Column::from_name("ts_year")))), + vec![true, true, true, true, true], + ), + ] { + let physical = create_physical_expr( + &expr, + &schema.clone().try_into().unwrap(), + state.execution_props(), + &PhysicalPlanningContext::default(), + ) + .unwrap(); + let predicate = PruningPredicateBuilder::new() + .with_file_schema(schema.clone()) + .try_build(physical) + .unwrap(); + assert_eq!(predicate.prune(&pruning).unwrap(), expected); + } + } + + #[test] + fn partitioned_file_pruning_preserves_unknown_specs_and_delete_files() { + use datafusion::arrow::array::{Int32Array, Int64Array}; + let source = StructField::new( 1, "ts", @@ -867,6 +1014,13 @@ mod tests { assert!(values.is_null(3)); assert!(values.is_null(4)); assert_eq!(pruning.num_containers(), 5); + let null_counts = pruning.null_counts(&Column::from_name("ts_year")).unwrap(); + let null_counts = null_counts.as_any().downcast_ref::().unwrap(); + assert_eq!(null_counts.value(0), 0); + assert_eq!(null_counts.value(1), 0); + assert!(null_counts.is_null(2)); + assert_eq!(null_counts.value(3), 1); + assert!(null_counts.is_null(4)); let schema = Arc::new(ArrowSchema::new(vec![Field::new( "ts_year", @@ -887,13 +1041,87 @@ mod tests { ) .unwrap(); let predicate = PruningPredicateBuilder::new() - .with_file_schema(schema) + .with_file_schema(schema.clone()) .try_build(physical) .unwrap(); assert_eq!( predicate.prune(&pruning).unwrap(), - vec![false, true, true, true, true] + vec![false, true, true, false, true] ); + + for (expr, expected) in [ + ( + Expr::IsNull(Box::new(Expr::Column(Column::from_name("ts_year")))), + vec![false, false, true, true, true], + ), + ( + Expr::IsNotNull(Box::new(Expr::Column(Column::from_name("ts_year")))), + vec![true, true, true, false, true], + ), + ] { + let physical = create_physical_expr( + &expr, + &schema.clone().try_into().unwrap(), + state.execution_props(), + &PhysicalPlanningContext::default(), + ) + .unwrap(); + let predicate = PruningPredicateBuilder::new() + .with_file_schema(schema.clone()) + .try_build(physical) + .unwrap(); + assert_eq!(predicate.prune(&pruning).unwrap(), expected); + } + } + + #[test] + fn transformed_null_projection_uses_partition_field_name() { + for (transform, source_type) in [ + (Transform::Year, PrimitiveType::Timestamp), + (Transform::Month, PrimitiveType::Timestamp), + (Transform::Day, PrimitiveType::Timestamp), + (Transform::Hour, PrimitiveType::Timestamp), + (Transform::Truncate(10), PrimitiveType::Long), + ] { + let source = StructField::new(1, "value", false, Type::Primitive(source_type), None); + let partition = PartitionField::new(1, 1000, "part", transform); + let fields = [BoundPartitionField::new(&partition, &source)]; + for is_null in [true, false] { + let input = Expr::Column(Column::from_name("value")); + let predicate = if is_null { + Expr::IsNull(Box::new(input)) + } else { + Expr::IsNotNull(Box::new(input)) + }; + let projected = transform_predicate(predicate, &fields).unwrap(); + let expected = Expr::Column(Column::from_name("part")); + assert_eq!( + projected, + if is_null { + Expr::IsNull(Box::new(expected)) + } else { + Expr::IsNotNull(Box::new(expected)) + } + ); + } + } + + let source = StructField::new( + 1, + "value", + false, + Type::Primitive(PrimitiveType::Long), + None, + ); + for transform in [Transform::Bucket(8), Transform::Void] { + let partition = PartitionField::new(1, 1000, "part", transform); + let fields = [BoundPartitionField::new(&partition, &source)]; + assert!(transform_predicate( + Expr::IsNull(Box::new(Expr::Column(Column::from_name("value")))), + &fields, + ) + .is_none()); + } } #[test] diff --git a/datafusion_iceberg/tests/position_delete.rs b/datafusion_iceberg/tests/position_delete.rs index ad4453d1..7bcc007e 100644 --- a/datafusion_iceberg/tests/position_delete.rs +++ b/datafusion_iceberg/tests/position_delete.rs @@ -158,6 +158,103 @@ fn write_deletion_vector_file(path: &str, data_file_path: &str, positions: &[u64 .unwrap() } +#[tokio::test] +async fn transformed_is_not_null_keeps_position_delete_manifest() { + let temp_dir = TempDir::new().unwrap(); + let table_dir = format!("{}/test/null_prune", temp_dir.path().display()); + let object_store = ObjectStoreBuilder::Filesystem(Arc::new(LocalFileSystem::new())); + let catalog: Arc = Arc::new( + SqlCatalog::new("sqlite://", "warehouse", object_store) + .await + .unwrap(), + ); + catalog + .create_namespace(&Namespace::try_new(&["test".to_string()]).unwrap(), None) + .await + .unwrap(); + let schema = Schema::builder() + .with_struct_field(StructField::new( + 1, + "id", + false, + Type::Primitive(PrimitiveType::Long), + None, + )) + .build() + .unwrap(); + let partition_spec = PartitionSpec::builder() + .with_partition_field(PartitionField::new( + 1, + 1000, + "id_trunc", + Transform::Truncate(10), + )) + .build() + .unwrap(); + Table::builder() + .with_name("null_prune") + .with_location(&table_dir) + .with_schema(schema) + .with_partition_spec(partition_spec) + .build(&["test".to_owned()], catalog.clone()) + .await + .unwrap(); + + let ctx = SessionContext::new(); + ctx.register_catalog( + "warehouse", + Arc::new(IcebergCatalog::new(catalog.clone(), None).await.unwrap()), + ); + run_query( + "INSERT INTO warehouse.test.null_prune VALUES (1), (2)", + &ctx, + ) + .await; + let identifier = Identifier::new(&["test".to_string()], "null_prune"); + let Tabular::Table(mut table) = catalog.clone().load_tabular(&identifier).await.unwrap() else { + panic!("expected Iceberg table"); + }; + let manifests = table.manifests(None, None).await.unwrap(); + let files = table + .datafiles(&manifests, None, (None, None)) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + assert_eq!(files.len(), 1); + let data_file = files[0].1.data_file(); + let delete_dir = format!("{table_dir}/data"); + std::fs::create_dir_all(&delete_dir).unwrap(); + let delete_file = write_position_delete_file( + &format!("{delete_dir}/position-delete.parquet"), + data_file.file_path(), + data_file.partition().clone(), + &[0], + ); + table + .new_transaction(None) + .append_delete(vec![delete_file]) + .commit() + .await + .unwrap(); + let Tabular::Table(table) = catalog.load_tabular(&identifier).await.unwrap() else { + panic!("expected Iceberg table"); + }; + let manifests = table.manifests(None, None).await.unwrap(); + assert!(manifests.iter().any(|manifest| { + manifest.content == iceberg_rust::spec::manifest_list::Content::Deletes + && manifest.added_rows_count == Some(0) + })); + + let rows = run_query( + "SELECT id FROM warehouse.test.null_prune WHERE id IS NOT NULL ORDER BY id", + &ctx, + ) + .await; + assert_batches_eq!(["+----+", "| id |", "+----+", "| 2 |", "+----+"], &rows); +} + #[tokio::test] async fn applies_v2_position_deletes() { let temp_dir = TempDir::new().unwrap(); From 827399e117dd1edeed185d29e720160911aa67a1 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Mon, 12 Oct 2026 00:48:31 +0300 Subject: [PATCH 3/7] Prune null predicates on Iceberg bucket partitions --- datafusion_iceberg/src/pruning_statistics.rs | 20 +++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/datafusion_iceberg/src/pruning_statistics.rs b/datafusion_iceberg/src/pruning_statistics.rs index 04a9ad91..42f7759d 100644 --- a/datafusion_iceberg/src/pruning_statistics.rs +++ b/datafusion_iceberg/src/pruning_statistics.rs @@ -463,6 +463,7 @@ pub(crate) fn transform_predicate( | Transform::Month | Transform::Day | Transform::Hour + | Transform::Bucket(_) | Transform::Truncate(_) ) { return None; @@ -1081,6 +1082,9 @@ mod tests { (Transform::Month, PrimitiveType::Timestamp), (Transform::Day, PrimitiveType::Timestamp), (Transform::Hour, PrimitiveType::Timestamp), + (Transform::Bucket(8), PrimitiveType::Int), + (Transform::Bucket(8), PrimitiveType::Long), + (Transform::Bucket(8), PrimitiveType::String), (Transform::Truncate(10), PrimitiveType::Long), ] { let source = StructField::new(1, "value", false, Type::Primitive(source_type), None); @@ -1113,15 +1117,13 @@ mod tests { Type::Primitive(PrimitiveType::Long), None, ); - for transform in [Transform::Bucket(8), Transform::Void] { - let partition = PartitionField::new(1, 1000, "part", transform); - let fields = [BoundPartitionField::new(&partition, &source)]; - assert!(transform_predicate( - Expr::IsNull(Box::new(Expr::Column(Column::from_name("value")))), - &fields, - ) - .is_none()); - } + let partition = PartitionField::new(1, 1000, "part", Transform::Void); + let fields = [BoundPartitionField::new(&partition, &source)]; + assert!(transform_predicate( + Expr::IsNull(Box::new(Expr::Column(Column::from_name("value")))), + &fields, + ) + .is_none()); } #[test] From 10144962611489b60794ba63371abd03d8082fc8 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 14 Oct 2026 09:08:59 +0300 Subject: [PATCH 4/7] Accept no-content rename responses from REST catalogs --- .../src/apis/catalog_api_api.rs | 4 +- .../iceberg-rest-catalog/src/apis/fetch.rs | 74 +++++++++++++++++++ 2 files changed, 76 insertions(+), 2 deletions(-) diff --git a/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs b/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs index db9d6dd7..d76f0cca 100644 --- a/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs +++ b/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs @@ -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, diff --git a/catalogs/iceberg-rest-catalog/src/apis/fetch.rs b/catalogs/iceberg-rest-catalog/src/apis/fetch.rs index 2500070f..fa3b3e06 100644 --- a/catalogs/iceberg-rest-catalog/src/apis/fetch.rs +++ b/catalogs/iceberg-rest-catalog/src/apis/fetch.rs @@ -109,6 +109,49 @@ where } } +pub(crate) async fn fetch_unit( + configuration: &configuration::Configuration, + method: reqwest::Method, + prefix: Option<&str>, + uri_str: &str, + request: &R, + headers: Option>, + query_params: Option>, +) -> Result<(), Error> +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(status: reqwest::StatusCode, content: String) -> Result<(), Error> +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( 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::(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::(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"), + } + } +} From a4d5b2d2affce793a253c0913e47cf80d29fc1f0 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 14 Oct 2026 10:23:16 +0300 Subject: [PATCH 5/7] Handle bodyless namespace HEAD responses by status --- .../src/apis/catalog_api_api.rs | 2 +- catalogs/iceberg-rest-catalog/src/catalog.rs | 58 +++++++++++++++---- 2 files changed, 47 insertions(+), 13 deletions(-) diff --git a/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs b/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs index d76f0cca..30ee1f0c 100644 --- a/catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs +++ b/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. diff --git a/catalogs/iceberg-rest-catalog/src/catalog.rs b/catalogs/iceberg-rest-catalog/src/catalog.rs index 072345cb..2657ff09 100644 --- a/catalogs/iceberg-rest-catalog/src/catalog.rs +++ b/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,46 @@ 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(); + let address = listener.local_addr().unwrap(); + let server = std::thread::spawn(move || { + use std::io::{Read, Write}; + + let (mut stream, _) = listener.accept().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() { From 5d850ea400fe10dbe5ade79e77401f5204204bcc Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 14 Oct 2026 10:25:34 +0300 Subject: [PATCH 6/7] Bound namespace REST test listener wait --- catalogs/iceberg-rest-catalog/src/catalog.rs | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/catalogs/iceberg-rest-catalog/src/catalog.rs b/catalogs/iceberg-rest-catalog/src/catalog.rs index 2657ff09..c5e4f312 100644 --- a/catalogs/iceberg-rest-catalog/src/catalog.rs +++ b/catalogs/iceberg-rest-catalog/src/catalog.rs @@ -757,11 +757,27 @@ pub mod tests { ("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 (mut stream, _) = listener.accept().unwrap(); + 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]) From 418a535fa8080a45fe33389234dcfcadc47dfd1d Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 14 Oct 2026 10:41:40 +0300 Subject: [PATCH 7/7] Fix df55 pruning test Cast import --- datafusion_iceberg/src/pruning_statistics.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/datafusion_iceberg/src/pruning_statistics.rs b/datafusion_iceberg/src/pruning_statistics.rs index f5d2e974..e53b112d 100644 --- a/datafusion_iceberg/src/pruning_statistics.rs +++ b/datafusion_iceberg/src/pruning_statistics.rs @@ -863,7 +863,6 @@ mod tests { #[test] fn manifest_null_counts_use_transformed_partition_name_and_spec() { use datafusion::arrow::array::Int32Array; - use datafusion_expr::expr::Cast; let source = StructField::new( 1, @@ -1034,7 +1033,7 @@ mod tests { let source_predicate = partition_comparison( Expr::Column(Column::from_name("ts")), Operator::Eq, - Expr::Cast(Cast::new( + Expr::Cast(datafusion_expr::expr::Cast::new( Box::new(Expr::Literal( ScalarValue::Utf8(Some("2025-06-15".to_owned())), None,