chore: Update to arrow/parquet 60.0.0 and object_store 0.14.2

Cargo.toml

@@ -90,31 +90,31 @@ version = "55.1.0"
 #
 # See for more details: https://github.com/rust-lang/cargo/issues/11329
 apache-avro = { version = "0.21", default-features = false }
-arrow = { version = "59.2.0", features = [
+arrow = { version = "60.0.0", features = [
     "prettyprint",
     "chrono-tz",
 ] }
-arrow-avro = { version = "59.2.0", default-features = false, features = [
+arrow-avro = { version = "60.0.0", default-features = false, features = [
     "deflate",
     "snappy",
     "zstd",
     "bzip2",
     "xz",
 ] }
-arrow-buffer = { version = "59.2.0", default-features = false }
-arrow-data = { version = "59.2.0", default-features = false }
-arrow-flight = { version = "59.2.0", features = [
-    "flight-sql-experimental",
+arrow-buffer = { version = "60.0.0", default-features = false }
+arrow-data = { version = "60.0.0", default-features = false }
+arrow-flight = { version = "60.0.0", features = [
+    "flight-sql",
 ] }
 # Both codecs are required here to make sure that code paths like
 # file-spilling have access to all compression codecs.
-arrow-ipc = { version = "59.2.0", default-features = false, features = [
+arrow-ipc = { version = "60.0.0", default-features = false, features = [
     "lz4",
     "zstd",
 ] }
-arrow-ord = { version = "59.2.0", default-features = false }
-arrow-schema = { version = "59.2.0", default-features = false }
-arrow-select = { version = "59.2.0", default-features = false }
+arrow-ord = { version = "60.0.0", default-features = false }
+arrow-schema = { version = "60.0.0", default-features = false }
+arrow-select = { version = "60.0.0", default-features = false }
 async-trait = "0.1.89"
 bigdecimal = "0.4.8"
 bytes = "1.11"
@@ -179,9 +179,9 @@ log = "^0.4"
 loom = "0.7"
 memchr = "2.8.1"
 num-traits = { version = "0.2" }
-object_store = { version = "0.13.2", default-features = false }
+object_store = { version = "0.14.2", default-features = false }
 parking_lot = "0.12"
-parquet = { version = "59.2.0", default-features = false, features = [
+parquet = { version = "60.0.0", default-features = false, features = [
     "arrow",
     "async",
     "object_store",

datafusion-cli/src/main.rs

@@ -662,9 +662,9 @@ mod tests {
         +-----------------------------------+-----------------+---------------------+------+------------------+
         | filename                          | file_size_bytes | metadata_size_bytes | hits | extra            |
         +-----------------------------------+-----------------+---------------------+------+------------------+
-        | alltypes_plain.parquet            | 1851            | 8794                | 1    | page_index=false |
-        | alltypes_tiny_pages.parquet       | 454233          | 268970              | 2    | page_index=true  |
-        | lz4_raw_compressed_larger.parquet | 380836          | 1331                | 1    | page_index=false |
+        | alltypes_plain.parquet            | 1851            | 8938                | 1    | page_index=false |
+        | alltypes_tiny_pages.parquet       | 454233          | 268826              | 2    | page_index=true  |
+        | lz4_raw_compressed_larger.parquet | 380836          | 1315                | 1    | page_index=false |
         +-----------------------------------+-----------------+---------------------+------+------------------+
         ");
 
@@ -693,9 +693,9 @@ mod tests {
         +-----------------------------------+-----------------+---------------------+------+------------------+
         | filename                          | file_size_bytes | metadata_size_bytes | hits | extra            |
         +-----------------------------------+-----------------+---------------------+------+------------------+
-        | alltypes_plain.parquet            | 1851            | 8794                | 4    | page_index=false |
-        | alltypes_tiny_pages.parquet       | 454233          | 268970              | 2    | page_index=true  |
-        | lz4_raw_compressed_larger.parquet | 380836          | 1331                | 2    | page_index=false |
+        | alltypes_plain.parquet            | 1851            | 8938                | 4    | page_index=false |
+        | alltypes_tiny_pages.parquet       | 454233          | 268826              | 2    | page_index=true  |
+        | lz4_raw_compressed_larger.parquet | 380836          | 1315                | 2    | page_index=false |
         +-----------------------------------+-----------------+---------------------+------+------------------+
         ");
 
@@ -862,14 +862,14 @@ mod tests {
             ])?
             .sort(vec![col("filename").sort(true, false)])?;
         let rbs = df.collect().await?;
-        assert_snapshot!(batches_to_string(&rbs),@r"
+        assert_snapshot!(batches_to_string(&rbs),@r#"
         +---------------------+-----------+-----------------+------+------+
         | metadata_size_bytes | filename  | file_size_bytes | etag | hits |
         +---------------------+-----------+-----------------+------+------+
-        | 212                 | 0.parquet | 3642            | 0    | 2    |
-        | 212                 | 1.parquet | 3642            | 1    | 2    |
+        | 216                 | 0.parquet | 3620            | "0"  | 2    |
+        | 216                 | 1.parquet | 3620            | "1"  | 2    |
         +---------------------+-----------+-----------------+------+------+
-        ");
+        "#);
 
         Ok(())
     }

datafusion/catalog-listing/src/helpers.rs

@@ -30,7 +30,7 @@ use datafusion_expr::{BinaryExpr, Operator, lit, utils};
 
 use arrow::{
     array::AsArray,
-    datatypes::{DataType, Field},
+    datatypes::{DataType, Field, Metadata},
     record_batch::RecordBatch,
 };
 use datafusion_expr::execution_props::ExecutionProps;
@@ -423,7 +423,7 @@ pub async fn pruned_partition_list<'a>(
                 .iter()
                 .map(|(n, d)| Field::new(n, d.clone(), true))
                 .collect(),
-            Default::default(),
+            Metadata::new(),
         )?;
 
         Ok(objects

datafusion/catalog-listing/src/table.rs

@@ -20,7 +20,7 @@ use crate::helpers::{
     expr_applicable_for_cols, filter_partitioned_file, pruned_partition_list,
 };
 use crate::{ListingOptions, ListingTableConfig};
-use arrow::datatypes::{Field, Schema, SchemaBuilder, SchemaRef};
+use arrow::datatypes::{Field, Metadata, Schema, SchemaBuilder, SchemaRef};
 use async_trait::async_trait;
 use datafusion_catalog::{ScanArgs, ScanResult, Session, TableProvider};
 use datafusion_common::stats::{Precision, is_known_empty};
@@ -1044,7 +1044,7 @@ impl ListingTable {
                 .iter()
                 .map(|(name, data_type)| Field::new(name, data_type.clone(), true))
                 .collect(),
-            Default::default(),
+            Metadata::new(),
         )?;
 
         file_groups

datafusion/common/src/dfschema.rs

@@ -18,7 +18,7 @@
 //! DFSchema is an extended schema struct that DataFusion uses to provide support for
 //! fields with optional relation names.
 
-use std::collections::{BTreeSet, HashMap, HashSet};
+use std::collections::{BTreeSet, HashSet};
 use std::fmt::Write as _;
 use std::fmt::{Display, Formatter};
 use std::hash::Hash;
@@ -32,7 +32,7 @@ use crate::{
 
 use arrow::compute::can_cast_types;
 use arrow::datatypes::{
-    DataType, Field, FieldRef, Fields, Schema, SchemaBuilder, SchemaRef,
+    DataType, Field, FieldRef, Fields, Metadata, Schema, SchemaBuilder, SchemaRef,
 };
 
 /// A reference-counted reference to a [DFSchema].
@@ -154,7 +154,7 @@ impl DFSchema {
     /// Create a `DFSchema` from an Arrow schema where all the fields have a given qualifier
     pub fn new_with_metadata(
         qualified_fields: Vec<(Option<TableReference>, Arc<Field>)>,
-        metadata: HashMap<String, String>,
+        metadata: impl Into<Metadata>,
     ) -> Result<Self> {
         let (qualifiers, fields): (Vec<Option<TableReference>>, Vec<Arc<Field>>) =
             qualified_fields.into_iter().unzip();
@@ -173,7 +173,7 @@ impl DFSchema {
     /// Create a new `DFSchema` from a list of Arrow [Field]s
     pub fn from_unqualified_fields(
         fields: Fields,
-        metadata: HashMap<String, String>,
+        metadata: impl Into<Metadata>,
     ) -> Result<Self> {
         let field_count = fields.len();
         let schema = Arc::new(Schema::new_with_metadata(fields, metadata));
@@ -865,7 +865,7 @@ impl DFSchema {
     }
 
     /// Get metadata of this schema
-    pub fn metadata(&self) -> &HashMap<String, String> {
+    pub fn metadata(&self) -> &Metadata {
         &self.inner.metadata
     }
 
@@ -1192,7 +1192,7 @@ impl ToDFSchema for Vec<Field> {
         let field_count = self.len();
         let schema = Schema {
             fields: self.into(),
-            metadata: HashMap::new(),
+            metadata: Metadata::new(),
         };
         let dfschema = DFSchema {
             inner: schema.into(),
@@ -1234,7 +1234,7 @@ pub trait ExprSchema: std::fmt::Debug {
     }
 
     /// Returns the column's optional metadata.
-    fn metadata(&self, col: &Column) -> Result<&HashMap<String, String>> {
+    fn metadata(&self, col: &Column) -> Result<&Metadata> {
         Ok(self.field_from_column(col)?.metadata())
     }
 
@@ -1258,7 +1258,7 @@ impl<P: AsRef<DFSchema> + std::fmt::Debug> ExprSchema for P {
         self.as_ref().data_type(col)
     }
 
-    fn metadata(&self, col: &Column) -> Result<&HashMap<String, String>> {
+    fn metadata(&self, col: &Column) -> Result<&Metadata> {
         ExprSchema::metadata(self.as_ref(), col)
     }
 
@@ -1391,6 +1391,7 @@ pub fn qualified_name(qualifier: Option<&TableReference>, name: &str) -> String
 #[cfg(test)]
 mod tests {
     use crate::assert_contains;
+    use std::collections::HashMap;
 
     use super::*;
 

datafusion/common/src/hash_utils.rs

@@ -689,8 +689,8 @@ fn hash_map_array(
     let offsets = array.offsets();
 
     // Create hashes for each entry in each row
-    let first_offset = offsets.first().copied().unwrap_or_default() as usize;
-    let last_offset = offsets.last().copied().unwrap_or_default() as usize;
+    let first_offset = offsets.first() as usize;
+    let last_offset = offsets.last() as usize;
     let entries_len = last_offset - first_offset;
 
     // Only hash the entries that are actually referenced

datafusion/common/src/heap_size.rs

@@ -47,12 +47,12 @@ use arrow::array::{
 };
 use arrow::datatypes::{
     DataType, Field, Fields, IntervalDayTime, IntervalMonthDayNano, IntervalUnit,
-    TimeUnit, UnionFields, UnionMode, i256,
+    Metadata, TimeUnit, UnionFields, UnionMode, i256,
 };
 use chrono::{DateTime, Utc};
 use half::f16;
 use hashbrown::HashSet;
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 use std::fmt::Debug;
 use std::sync::Arc;
 
@@ -396,6 +396,24 @@ impl DFHeapSize for UnionFields {
     }
 }
 
+impl<K: DFHeapSize, V: DFHeapSize> DFHeapSize for BTreeMap<K, V> {
+    fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize {
+        // BTreeMap does not provide a way to get its heap size, so this
+        // approximates it as the entries' sizes, ignoring node overhead.
+        self.iter()
+            .map(|(k, v)| size_of::<(K, V)>() + k.heap_size(ctx) + v.heap_size(ctx))
+            .sum()
+    }
+}
+
+impl DFHeapSize for Metadata {
+    fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize {
+        // `as_arc` returns `None` when the metadata is empty; the `Arc` impl
+        // dedupes instances that share the same allocation.
+        self.as_arc().map_or(0, |map| map.heap_size(ctx))
+    }
+}
+
 impl DFHeapSize for Field {
     fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize {
         self.name().heap_size(ctx)