Fix nested scalar hashing

vortex-array/src/dtype/dtype_impl.rs

@@ -3,6 +3,8 @@
 
 use std::fmt::Display;
 use std::fmt::Formatter;
+use std::hash::Hash;
+use std::hash::Hasher;
 use std::sync::Arc;
 
 use DType::*;
@@ -120,21 +122,9 @@ impl DType {
                 lhs_size == rhs_size && lhs_dtype.eq_ignore_nullability(rhs_dtype)
             }
             (Struct(lhs_dtype, _), Struct(rhs_dtype, _)) => {
-                (lhs_dtype.names() == rhs_dtype.names())
-                    && (lhs_dtype
-                        .fields()
-                        .zip_eq(rhs_dtype.fields())
-                        .all(|(l, r)| l.eq_ignore_nullability(&r)))
-            }
-            (Union(lhs, _), Union(rhs, _)) => {
-                // Equal `names` implies equal length by FieldNames equality.
-                lhs.names() == rhs.names()
-                    && lhs.type_ids() == rhs.type_ids()
-                    && lhs
-                        .variants()
-                        .zip_eq(rhs.variants())
-                        .all(|(l, r)| l.eq_ignore_nullability(&r))
+                lhs_dtype.eq_ignore_nullability(rhs_dtype)
             }
+            (Union(lhs, _), Union(rhs, _)) => lhs.eq_ignore_nullability(rhs),
             (Variant(_), Variant(_)) => true,
             (Extension(lhs_extdtype), Extension(rhs_extdtype)) => {
                 lhs_extdtype.eq_ignore_nullability(rhs_extdtype)
@@ -143,6 +133,25 @@ impl DType {
         }
     }
 
+    /// Hash this dtype using the same equivalence relation as [`Self::eq_ignore_nullability`].
+    pub(crate) fn hash_ignore_nullability<H: Hasher>(&self, state: &mut H) {
+        std::mem::discriminant(self).hash(state);
+
+        match self {
+            Null | Bool(_) | Utf8(_) | Binary(_) | Variant(_) => {}
+            Primitive(ptype, _) => ptype.hash(state),
+            Decimal(decimal, _) => decimal.hash(state),
+            List(element, _) => element.hash_ignore_nullability(state),
+            FixedSizeList(element, size, _) => {
+                element.hash_ignore_nullability(state);
+                size.hash(state);
+            }
+            Struct(fields, _) => fields.hash_ignore_nullability(state),
+            Union(variants, _) => variants.hash_ignore_nullability(state),
+            Extension(ext) => ext.hash_ignore_nullability(state),
+        }
+    }
+
     /// Returns `true` if `self` is a subset type of `other, otherwise `false`.
     ///
     /// If `self` is nullable, this means that the other `DType` must also be nullable (since a
@@ -514,18 +523,29 @@ impl Display for DType {
 
 #[cfg(test)]
 mod tests {
+    use std::collections::hash_map::DefaultHasher;
+    use std::hash::Hasher;
     use std::sync::Arc;
 
+    use vortex_error::VortexResult;
+
     use crate::dtype::DType;
     use crate::dtype::Nullability::NonNullable;
     use crate::dtype::Nullability::Nullable;
     use crate::dtype::PType;
+    use crate::dtype::UnionVariants;
     use crate::dtype::decimal::DecimalDType;
     use crate::extension::datetime::Date;
     use crate::extension::datetime::Time;
     use crate::extension::datetime::TimeUnit;
     use crate::extension::datetime::Timestamp;
 
+    fn hash_ignore_nullability(dtype: &DType) -> u64 {
+        let mut hasher = DefaultHasher::new();
+        dtype.hash_ignore_nullability(&mut hasher);
+        hasher.finish()
+    }
+
     #[test]
     fn test_ext_dtype_eq_ignore_nullability() {
         let d1 = DType::Extension(Time::new(TimeUnit::Seconds, Nullable).erased());
@@ -541,6 +561,37 @@ mod tests {
         assert!(!t1.eq_ignore_nullability(&t2));
     }
 
+    #[test]
+    fn test_union_dtype_hash_ignores_variant_nullability() -> VortexResult<()> {
+        let lhs = DType::Union(
+            UnionVariants::try_new(
+                ["int", "string"].into(),
+                vec![
+                    DType::Primitive(PType::I32, Nullable),
+                    DType::Utf8(NonNullable),
+                ],
+                vec![5, 9],
+            )?,
+            NonNullable,
+        );
+        let rhs = DType::Union(
+            UnionVariants::try_new(
+                ["int", "string"].into(),
+                vec![
+                    DType::Primitive(PType::I32, NonNullable),
+                    DType::Utf8(Nullable),
+                ],
+                vec![5, 9],
+            )?,
+            NonNullable,
+        );
+
+        assert!(lhs.eq_ignore_nullability(&rhs));
+        assert_eq!(hash_ignore_nullability(&lhs), hash_ignore_nullability(&rhs));
+
+        Ok(())
+    }
+
     #[test]
     fn element_size_null() {
         assert_eq!(DType::Null.element_size(), Some(0));

vortex-array/src/dtype/extension/erased.rs

@@ -75,6 +75,14 @@ impl ExtDTypeRef {
                 .eq_ignore_nullability(other.storage_dtype())
     }
 
+    /// Hash this extension dtype using the same equivalence relation as
+    /// [`Self::eq_ignore_nullability`].
+    pub(crate) fn hash_ignore_nullability<H: Hasher>(&self, state: &mut H) {
+        self.id().hash(state);
+        self.0.metadata_hash(state);
+        self.storage_dtype().hash_ignore_nullability(state);
+    }
+
     // TODO(connor): We should add a different type that returns something that can be serialized.
     /// Serialize the metadata into a byte vector.
     pub fn serialize_metadata(&self) -> VortexResult<Vec<u8>> {

vortex-array/src/dtype/struct_.rs

@@ -4,6 +4,7 @@
 use std::fmt::Display;
 use std::fmt::Formatter;
 use std::hash::Hash;
+use std::hash::Hasher;
 use std::sync::Arc;
 use std::sync::OnceLock;
 
@@ -83,7 +84,7 @@ impl PartialEq for FieldDTypeInner {
 impl Eq for FieldDTypeInner {}
 
 impl Hash for FieldDTypeInner {
-    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+    fn hash<H: Hasher>(&self, state: &mut H) {
         match self {
             FieldDTypeInner::Owned(owned) => {
                 owned.hash(state);
@@ -255,12 +256,33 @@ impl PartialEq for StructFieldsInner {
 impl Eq for StructFieldsInner {}
 
 impl Hash for StructFieldsInner {
-    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+    fn hash<H: Hasher>(&self, state: &mut H) {
         self.names.hash(state);
         self.dtypes.hash(state);
     }
 }
 
+impl StructFields {
+    /// Check if these struct fields are equal, ignoring field dtype nullability recursively.
+    pub fn eq_ignore_nullability(&self, other: &Self) -> bool {
+        Arc::ptr_eq(&self.0, &other.0)
+            || (self.0.names == other.0.names
+                && self
+                    .fields()
+                    .zip_eq(other.fields())
+                    .all(|(lhs, rhs)| lhs.eq_ignore_nullability(&rhs)))
+    }
+
+    /// Hash these struct fields using the same equivalence relation as
+    /// [`Self::eq_ignore_nullability`].
+    pub(crate) fn hash_ignore_nullability<H: Hasher>(&self, state: &mut H) {
+        self.0.names.hash(state);
+        for field in self.fields() {
+            field.hash_ignore_nullability(state);
+        }
+    }
+}
+
 impl Default for StructFields {
     fn default() -> Self {
         Self::empty()

vortex-array/src/dtype/union.rs

@@ -3,6 +3,7 @@
 
 use std::fmt;
 use std::hash::Hash;
+use std::hash::Hasher;
 use std::sync::Arc;
 
 use itertools::Itertools;
@@ -133,13 +134,36 @@ impl PartialEq for UnionVariantsInner {
 impl Eq for UnionVariantsInner {}
 
 impl Hash for UnionVariantsInner {
-    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+    fn hash<H: Hasher>(&self, state: &mut H) {
         self.names.hash(state);
         self.dtypes.hash(state);
         self.type_ids.hash(state);
     }
 }
 
+impl UnionVariants {
+    /// Check if these union variants are equal, ignoring variant dtype nullability recursively.
+    pub fn eq_ignore_nullability(&self, other: &Self) -> bool {
+        Arc::ptr_eq(&self.0, &other.0)
+            || (self.0.names == other.0.names
+                && self
+                    .variants()
+                    .zip_eq(other.variants())
+                    .all(|(lhs, rhs)| lhs.eq_ignore_nullability(&rhs))
+                && self.0.type_ids == other.0.type_ids)
+    }
+
+    /// Hash these union variants using the same equivalence relation as
+    /// [`Self::eq_ignore_nullability`].
+    pub(crate) fn hash_ignore_nullability<H: Hasher>(&self, state: &mut H) {
+        self.0.names.hash(state);
+        for variant in self.variants() {
+            variant.hash_ignore_nullability(state);
+        }
+        self.0.type_ids.hash(state);
+    }
+}
+
 impl Default for UnionVariants {
     fn default() -> Self {
         Self::empty()

vortex-array/src/scalar/scalar_impl.rs

@@ -282,7 +282,7 @@ impl Scalar {
 /// values have equal hashes.
 impl Hash for Scalar {
     fn hash<H: Hasher>(&self, state: &mut H) {
-        self.dtype.as_nonnullable().hash(state);
+        self.dtype.hash_ignore_nullability(state);
         self.value.hash(state);
     }
 }

vortex-array/src/scalar/tests/primitives.rs

@@ -5,6 +5,9 @@
 
 #[cfg(test)]
 mod tests {
+    use std::collections::hash_map::DefaultHasher;
+    use std::hash::Hash;
+    use std::hash::Hasher;
     use std::sync::Arc;
 
     use vortex_buffer::ByteBuffer;
@@ -24,6 +27,12 @@ mod tests {
     use crate::scalar::Scalar;
     use crate::scalar::ScalarValue;
 
+    fn scalar_hash(scalar: &Scalar) -> u64 {
+        let mut hasher = DefaultHasher::new();
+        scalar.hash(&mut hasher);
+        hasher.finish()
+    }
+
     #[test]
     fn default_value_for_complex_dtype() {
         let struct_dtype = DType::struct_(
@@ -412,6 +421,23 @@ mod tests {
         assert_eq!(set.len(), 5);
     }
 
+    #[test]
+    fn test_scalar_hash_ignores_nested_nullability() {
+        let nullable = Scalar::list(
+            DType::Primitive(PType::I32, Nullability::Nullable),
+            vec![Scalar::primitive(42_i32, Nullability::Nullable)],
+            Nullability::NonNullable,
+        );
+        let non_nullable = Scalar::list(
+            DType::Primitive(PType::I32, Nullability::NonNullable),
+            vec![Scalar::primitive(42_i32, Nullability::NonNullable)],
+            Nullability::NonNullable,
+        );
+
+        assert_eq!(nullable, non_nullable);
+        assert_eq!(scalar_hash(&nullable), scalar_hash(&non_nullable));
+    }
+
     #[test]
     fn test_scalar_partial_ord_incompatible_types() {
         let int_scalar = Scalar::primitive(42i32, Nullability::NonNullable);

vortex-array/src/scalar/typed_view/extension/mod.rs

@@ -139,7 +139,7 @@ impl PartialOrd for ExtScalar<'_> {
 
 impl Hash for ExtScalar<'_> {
     fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
-        self.ext_dtype.hash(state);
+        self.ext_dtype.hash_ignore_nullability(state);
         self.to_storage_scalar().hash(state);
     }
 }

vortex-array/src/scalar/typed_view/extension/tests.rs

@@ -1,6 +1,10 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+use std::collections::hash_map::DefaultHasher;
+use std::hash::Hash;
+use std::hash::Hasher;
+
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 
@@ -174,6 +178,28 @@ fn test_ext_scalar_hash() {
     assert_eq!(set.len(), 2);
 }
 
+#[test]
+fn test_ext_scalar_hash_ignores_storage_nullability() {
+    let nullable = Scalar::extension::<TestI32Ext>(
+        EmptyMetadata,
+        Scalar::primitive(42_i32, Nullability::Nullable),
+    );
+    let non_nullable = Scalar::extension::<TestI32Ext>(
+        EmptyMetadata,
+        Scalar::primitive(42_i32, Nullability::NonNullable),
+    );
+    let nullable = nullable.as_extension();
+    let non_nullable = non_nullable.as_extension();
+
+    assert_eq!(nullable, non_nullable);
+
+    let mut nullable_hasher = DefaultHasher::new();
+    nullable.hash(&mut nullable_hasher);
+    let mut non_nullable_hasher = DefaultHasher::new();
+    non_nullable.hash(&mut non_nullable_hasher);
+    assert_eq!(nullable_hasher.finish(), non_nullable_hasher.finish());
+}
+
 #[test]
 fn test_ext_scalar_storage() {
     let storage_scalar = Scalar::primitive(42i32, Nullability::NonNullable);