Upgrade hashbrown to 0.16 (#19554)

## Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->

- Closes #13433

## Rationale for this change

Keep up to date with hashbrown and possible (performance) improvements,
remove the `rawtable` functionality.


## What changes are included in this PR?

Upgrading, fixing case of nondeterminism in `HashSet` usage.

## Are these changes tested?

Yes existiing tests.

## Are there any user-facing changes?

No.

Author: Dandandan

Cargo.lock

@@ -156,7 +156,7 @@ flate2 = "1.1.5"
 futures = "0.3"
 glob = "0.3.0"
 half = { version = "2.7.0", default-features = false }
-hashbrown = { version = "0.14.5", features = ["raw"] }
+hashbrown = { version = "0.16.1" }
 hex = { version = "0.4.3" }
 indexmap = "2.12.1"
 insta = { version = "1.45.0", features = ["glob", "filters"] }

Cargo.toml

@@ -85,7 +85,7 @@ pub use functional_dependencies::{
     aggregate_functional_dependencies, get_required_group_by_exprs_indices,
     get_target_functional_dependencies,
 };
-use hashbrown::hash_map::DefaultHashBuilder;
+use hashbrown::DefaultHashBuilder;
 pub use join_type::{JoinConstraint, JoinSide, JoinType};
 pub use nested_struct::cast_column;
 pub use null_equality::NullEquality;

datafusion/common/src/lib.rs

@@ -1111,7 +1111,7 @@ mod tests {
         ])])?;
 
         // without compaction, the size is 17112
-        assert_eq!(acc.size(), 2184);
+        assert_eq!(acc.size(), 2224);
 
         Ok(())
     }

datafusion/functions-aggregate/src/array_agg.rs

@@ -47,5 +47,6 @@ chrono = { workspace = true }
 datafusion-common = { workspace = true }
 datafusion-expr-common = { workspace = true }
 hashbrown = { workspace = true }
+indexmap = { workspace = true }
 itertools = { workspace = true }
 parking_lot = { workspace = true }

datafusion/physical-expr-common/Cargo.toml

@@ -31,7 +31,7 @@ use arrow::datatypes::Schema;
 use arrow::record_batch::RecordBatch;
 use datafusion_common::{HashSet, Result};
 use datafusion_expr_common::columnar_value::ColumnarValue;
-
+use indexmap::IndexSet;
 /// Represents Sort operation for a column in a RecordBatch
 ///
 /// Example:
@@ -353,14 +353,14 @@ impl From<PhysicalSortRequirement> for PhysicalSortExpr {
 /// 1. It is non-degenerate, meaning it contains at least one element.
 /// 2. It is duplicate-free, meaning it does not contain multiple entries for
 ///    the same column.
-#[derive(Debug, Clone)]
+#[derive(Clone, Debug)]
 pub struct LexOrdering {
     /// Vector of sort expressions representing the lexicographical ordering.
     exprs: Vec<PhysicalSortExpr>,
     /// Set of expressions in the lexicographical ordering, used to ensure
     /// that the ordering is duplicate-free. Note that the elements in this
     /// set are the same underlying physical expressions as in `exprs`.
-    set: HashSet<Arc<dyn PhysicalExpr>>,
+    set: IndexSet<Arc<dyn PhysicalExpr>>,
 }
 
 impl LexOrdering {
@@ -371,7 +371,7 @@ impl LexOrdering {
         let mut candidate = Self {
             // not valid yet; valid publicly-returned instance must be non-empty
             exprs: Vec::new(),
-            set: HashSet::new(),
+            set: IndexSet::new(),
         };
         for expr in exprs {
             candidate.push(expr);
@@ -421,7 +421,7 @@ impl LexOrdering {
             return false;
         }
         for PhysicalSortExpr { expr, .. } in self.exprs[len..].iter() {
-            self.set.remove(expr);
+            self.set.swap_remove(expr);
         }
         self.exprs.truncate(len);
         true

datafusion/physical-expr-common/src/sort_expr.rs

@@ -27,7 +27,7 @@ use crate::projection::ProjectionTargets;
 use crate::{PhysicalExpr, PhysicalExprRef, PhysicalSortExpr, PhysicalSortRequirement};
 
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
-use datafusion_common::{HashMap, JoinType, Result, ScalarValue};
+use datafusion_common::{JoinType, Result, ScalarValue};
 use datafusion_physical_expr_common::physical_expr::format_physical_expr_list;
 
 use indexmap::{IndexMap, IndexSet};
@@ -303,7 +303,7 @@ type AugmentedMapping<'a> = IndexMap<
 #[derive(Clone, Debug, Default)]
 pub struct EquivalenceGroup {
     /// A mapping from expressions to their equivalence class key.
-    map: HashMap<Arc<dyn PhysicalExpr>, usize>,
+    map: IndexMap<Arc<dyn PhysicalExpr>, usize>,
     /// The equivalence classes in this group.
     classes: Vec<EquivalenceClass>,
 }
@@ -436,7 +436,7 @@ impl EquivalenceGroup {
         let cls = self.classes.swap_remove(idx);
         // Remove its entries from the lookup table:
         for expr in cls.iter() {
-            self.map.remove(expr);
+            self.map.swap_remove(expr);
         }
         // Update the lookup table for the moved class:
         if idx < self.classes.len() {
@@ -448,7 +448,7 @@ impl EquivalenceGroup {
     /// Updates the entry in lookup table for the given equivalence class with
     /// the given index.
     fn update_lookup_table(
-        map: &mut HashMap<Arc<dyn PhysicalExpr>, usize>,
+        map: &mut IndexMap<Arc<dyn PhysicalExpr>, usize>,
         cls: &EquivalenceClass,
         idx: usize,
     ) {

datafusion/physical-expr/src/equivalence/class.rs

@@ -29,10 +29,11 @@ use arrow::datatypes::{DataType, Schema, UInt32Type, UnionMode};
 use arrow::error::ArrowError;
 use datafusion_common::cast::as_boolean_array;
 use datafusion_common::{
-    DataFusionError, HashMap, HashSet, Result, ScalarValue, assert_or_internal_err,
-    exec_err, internal_datafusion_err, internal_err,
+    DataFusionError, Result, ScalarValue, assert_or_internal_err, exec_err,
+    internal_datafusion_err, internal_err,
 };
 use datafusion_expr::ColumnarValue;
+use indexmap::{IndexMap, IndexSet};
 use std::borrow::Cow;
 use std::hash::Hash;
 use std::{any::Any, sync::Arc};
@@ -122,7 +123,7 @@ impl CaseBody {
     /// Derives a [ProjectedCaseBody] from this [CaseBody].
     fn project(&self) -> Result<ProjectedCaseBody> {
         // Determine the set of columns that are used in all the expressions of the case body.
-        let mut used_column_indices = HashSet::<usize>::new();
+        let mut used_column_indices = IndexSet::<usize>::new();
         let mut collect_column_indices = |expr: &Arc<dyn PhysicalExpr>| {
             expr.apply(|expr| {
                 if let Some(column) = expr.as_any().downcast_ref::<Column>() {
@@ -149,7 +150,7 @@ impl CaseBody {
             .iter()
             .enumerate()
             .map(|(projected, original)| (*original, projected))
-            .collect::<HashMap<usize, usize>>();
+            .collect::<IndexMap<usize, usize>>();
 
         // Construct the projected body by rewriting each expression from the original body
         // using the column index mapping.

datafusion/physical-expr/src/expressions/case.rs

@@ -229,7 +229,7 @@ pub fn collect_columns(expr: &Arc<dyn PhysicalExpr>) -> HashSet<Column> {
     let mut columns = HashSet::<Column>::new();
     expr.apply(|expr| {
         if let Some(column) = expr.as_any().downcast_ref::<Column>() {
-            columns.get_or_insert_owned(column);
+            columns.get_or_insert_with(column, |c| c.clone());
         }
         Ok(TreeNodeRecursion::Continue)
     })