Merge branch 'main' into feature/implement-project-node-for-insert-into-datafusion

Author: fvaleye

crates/catalog/glue/tests/glue_catalog_test.rs

@@ -73,6 +73,11 @@ async fn get_catalog() -> GlueCatalog {
         sleep(std::time::Duration::from_millis(1000)).await;
     }
 
+    while !scan_port_addr(minio_socket_addr) {
+        info!("Waiting for 1s minio to ready...");
+        sleep(std::time::Duration::from_millis(1000)).await;
+    }
+
     let props = HashMap::from([
         (AWS_ACCESS_KEY_ID.to_string(), "my_access_id".to_string()),
         (
@@ -89,6 +94,24 @@ async fn get_catalog() -> GlueCatalog {
         (S3_REGION.to_string(), "us-east-1".to_string()),
     ]);
 
+    // Wait for bucket to actually exist
+    let file_io = iceberg::io::FileIO::from_path("s3a://")
+        .unwrap()
+        .with_props(props.clone())
+        .build()
+        .unwrap();
+
+    let mut retries = 0;
+    while retries < 30 {
+        if file_io.exists("s3a://warehouse/").await.unwrap_or(false) {
+            info!("S3 bucket 'warehouse' is ready");
+            break;
+        }
+        info!("Waiting for bucket creation... (attempt {})", retries + 1);
+        sleep(std::time::Duration::from_millis(1000)).await;
+        retries += 1;
+    }
+
     let config = GlueCatalogConfig::builder()
         .uri(format!("http://{}", glue_socket_addr))
         .warehouse("s3a://warehouse/hive".to_string())

crates/catalog/hms/tests/hms_catalog_test.rs

@@ -73,6 +73,11 @@ async fn get_catalog() -> HmsCatalog {
         sleep(std::time::Duration::from_millis(1000)).await;
     }
 
+    while !scan_port_addr(minio_socket_addr) {
+        info!("Waiting for 1s minio to ready...");
+        sleep(std::time::Duration::from_millis(1000)).await;
+    }
+
     let props = HashMap::from([
         (
             S3_ENDPOINT.to_string(),
@@ -83,6 +88,24 @@ async fn get_catalog() -> HmsCatalog {
         (S3_REGION.to_string(), "us-east-1".to_string()),
     ]);
 
+    // Wait for bucket to actually exist
+    let file_io = iceberg::io::FileIO::from_path("s3a://")
+        .unwrap()
+        .with_props(props.clone())
+        .build()
+        .unwrap();
+
+    let mut retries = 0;
+    while retries < 30 {
+        if file_io.exists("s3a://warehouse/").await.unwrap_or(false) {
+            info!("S3 bucket 'warehouse' is ready");
+            break;
+        }
+        info!("Waiting for bucket creation... (attempt {})", retries + 1);
+        sleep(std::time::Duration::from_millis(1000)).await;
+        retries += 1;
+    }
+
     let config = HmsCatalogConfig::builder()
         .address(hms_socket_addr.to_string())
         .thrift_transport(HmsThriftTransport::Buffered)

crates/iceberg/src/arrow/nan_val_cnt_visitor.rs

@@ -25,7 +25,7 @@ use arrow_array::{ArrayRef, Float32Array, Float64Array, RecordBatch, StructArray
 use arrow_schema::DataType;
 
 use crate::Result;
-use crate::arrow::ArrowArrayAccessor;
+use crate::arrow::{ArrowArrayAccessor, FieldMatchMode};
 use crate::spec::{
     ListType, MapType, NestedFieldRef, PrimitiveType, Schema, SchemaRef, SchemaWithPartnerVisitor,
     StructType, visit_struct_with_partner,
@@ -71,6 +71,7 @@ macro_rules! count_float_nans {
 pub struct NanValueCountVisitor {
     /// Stores field ID to NaN value count mapping
     pub nan_value_counts: HashMap<i32, u64>,
+    match_mode: FieldMatchMode,
 }
 
 impl SchemaWithPartnerVisitor<ArrayRef> for NanValueCountVisitor {
@@ -149,14 +150,20 @@ impl SchemaWithPartnerVisitor<ArrayRef> for NanValueCountVisitor {
 impl NanValueCountVisitor {
     /// Creates new instance of NanValueCountVisitor
     pub fn new() -> Self {
+        Self::new_with_match_mode(FieldMatchMode::Id)
+    }
+
+    /// Creates new instance of NanValueCountVisitor with explicit match mode
+    pub fn new_with_match_mode(match_mode: FieldMatchMode) -> Self {
         Self {
             nan_value_counts: HashMap::new(),
+            match_mode,
         }
     }
 
     /// Compute nan value counts in given schema and record batch
     pub fn compute(&mut self, schema: SchemaRef, batch: RecordBatch) -> Result<()> {
-        let arrow_arr_partner_accessor = ArrowArrayAccessor {};
+        let arrow_arr_partner_accessor = ArrowArrayAccessor::new_with_match_mode(self.match_mode);
 
         let struct_arr = Arc::new(StructArray::from(batch)) as ArrayRef;
         visit_struct_with_partner(