Add Map logical type and scalar support

encodings/sparse/src/canonical.rs

@@ -153,6 +153,7 @@ pub(super) fn execute_sparse(parts: SparseParts, ctx: &mut ExecutionCtx) -> Vort
         DType::FixedSizeList(.., nullability) => {
             execute_sparse_fixed_size_list(&patches, &fill_value, len, *nullability, ctx)?
         }
+        DType::Map(..) => vortex_bail!("Sparse canonicalization does not support Map arrays yet"),
         DType::Struct(struct_fields, ..) => execute_sparse_struct(
             struct_fields,
             fill_value.as_struct(),

fuzz/src/array/compare.rs

@@ -186,7 +186,11 @@ pub fn compare_canonical_array(
             }))
             .into_array()
         }
-        d @ (DType::Null | DType::Union(..) | DType::Variant(_) | DType::Extension(_)) => {
+        d @ (DType::Null
+        | DType::Map(..)
+        | DType::Union(..)
+        | DType::Variant(_)
+        | DType::Extension(_)) => {
             unreachable!("DType {d} not supported for fuzzing")
         }
     }

fuzz/src/array/filter.rs

@@ -121,7 +121,11 @@ pub fn filter_canonical_array(
             )
             .map(|a| a.into_array())
         }
-        d @ (DType::Null | DType::Union(..) | DType::Variant(_) | DType::Extension(_)) => {
+        d @ (DType::Null
+        | DType::Map(..)
+        | DType::Union(..)
+        | DType::Variant(_)
+        | DType::Extension(_)) => {
             unreachable!("DType {d} not supported for fuzzing")
         }
     }

fuzz/src/array/mod.rs

@@ -518,6 +518,7 @@ fn actions_for_dtype(dtype: &DType) -> HashSet<ActionType> {
                     acc.intersection(&actions).copied().collect()
                 })
         }
+        DType::Map(..) => HashSet::new(),
         DType::Union(..) => todo!("TODO(connor)[Union]: unimplemented"),
         // Currently, no support at all
         DType::Variant(_) => unreachable!("Variant dtype shouldn't be fuzzed"),

fuzz/src/array/search_sorted.rs

@@ -149,7 +149,11 @@ pub fn search_sorted_canonical_array(
                 .collect::<VortexResult<Vec<_>>>()?;
             scalar_vals.search_sorted(&scalar.cast(array.dtype())?, side)
         }
-        d @ (DType::Null | DType::Union(..) | DType::Variant(_) | DType::Extension(_)) => {
+        d @ (DType::Null
+        | DType::Map(..)
+        | DType::Union(..)
+        | DType::Variant(_)
+        | DType::Extension(_)) => {
             unreachable!("DType {d} not supported for fuzzing")
         }
     }

fuzz/src/array/slice.rs

@@ -125,7 +125,11 @@ pub fn slice_canonical_array(
             )
             .map(|a| a.into_array())
         }
-        d @ (DType::Null | DType::Union(..) | DType::Variant(_) | DType::Extension(_)) => {
+        d @ (DType::Null
+        | DType::Map(..)
+        | DType::Union(..)
+        | DType::Variant(_)
+        | DType::Extension(_)) => {
             unreachable!("DType {d} not supported for fuzzing")
         }
     }

fuzz/src/array/sort.rs

@@ -102,7 +102,11 @@ pub fn sort_canonical_array(array: &ArrayRef, ctx: &mut ExecutionCtx) -> VortexR
             });
             take_canonical_array_non_nullable_indices(array, &sort_indices, ctx)
         }
-        d @ (DType::Null | DType::Union(..) | DType::Variant(_) | DType::Extension(_)) => {
+        d @ (DType::Null
+        | DType::Map(..)
+        | DType::Union(..)
+        | DType::Variant(_)
+        | DType::Extension(_)) => {
             unreachable!("DType {d} not supported for fuzzing")
         }
     }

fuzz/src/array/take.rs

@@ -148,7 +148,11 @@ pub fn take_canonical_array(
             )
             .map(|a| a.into_array())
         }
-        d @ (DType::Null | DType::Union(..) | DType::Variant(_) | DType::Extension(_)) => {
+        d @ (DType::Null
+        | DType::Map(..)
+        | DType::Union(..)
+        | DType::Variant(_)
+        | DType::Extension(_)) => {
             unreachable!("DType {d} not supported for fuzzing")
         }
     }

vortex-array/src/aggregate_fn/fns/is_sorted/mod.rs

@@ -246,6 +246,7 @@ impl AggregateFnVTable for IsSorted {
             DType::Null
             | DType::List(..)
             | DType::FixedSizeList(..)
+            | DType::Map(..)
             | DType::Struct(..)
             | DType::Union(..)
             | DType::Variant(..)
@@ -263,6 +264,7 @@ impl AggregateFnVTable for IsSorted {
             DType::Null
             | DType::List(..)
             | DType::FixedSizeList(..)
+            | DType::Map(..)
             | DType::Struct(..)
             | DType::Union(..)
             | DType::Variant(..)

vortex-array/src/aggregate_fn/fns/uncompressed_size_in_bytes/mod.rs

@@ -236,6 +236,9 @@ pub(crate) fn constant_uncompressed_size_in_bytes(
             let canonical = array.array().clone().execute::<Canonical>(ctx)?;
             return canonical_uncompressed_size_in_bytes(&canonical, ctx);
         }
+        DType::Map(..) => {
+            vortex_bail!("UncompressedSizeInBytes is not supported for map arrays yet")
+        }
         DType::Union(..) => {
             todo!(
                 "TODO(connor)[Union]: support constant Union size accounting after constant Union \
@@ -294,6 +297,7 @@ fn supports_uncompressed_size_in_bytes(dtype: &DType) -> bool {
         DType::List(element_dtype, _) | DType::FixedSizeList(element_dtype, ..) => {
             supports_uncompressed_size_in_bytes(element_dtype)
         }
+        DType::Map(..) => false,
         DType::Struct(fields, _) => fields
             .fields()
             .all(|field| supports_uncompressed_size_in_bytes(&field)),

vortex-array/src/arrays/arbitrary.rs

@@ -158,6 +158,7 @@ fn random_array_chunk(
         DType::FixedSizeList(elem_dtype, list_size, null) => {
             random_fixed_size_list(u, elem_dtype, *list_size, *null, chunk_len)
         }
+        DType::Map(..) => Err(IncorrectFormat),
         DType::Struct(sdt, n) => {
             let first_array = sdt
                 .fields()

vortex-array/src/arrays/constant/vtable/canonical.rs

@@ -126,6 +126,7 @@ pub(crate) fn constant_canonicalize(
             ))
         }
         DType::List(..) => Canonical::List(constant_canonical_list_array(scalar, array.len())),
+        DType::Map(..) => vortex_error::vortex_bail!("canonical map arrays are not yet supported"),
         DType::FixedSizeList(element_dtype, list_size, _) => {
             let value = scalar.as_list();
 

vortex-array/src/builders/mod.rs

@@ -347,6 +347,9 @@ pub fn builder_with_capacity(dtype: &DType, capacity: usize) -> Box<dyn ArrayBui
             2 * capacity, // Arbitrarily choose 2 times the `offsets` capacity here.
             capacity,
         )),
+        DType::Map(..) => {
+            vortex_error::vortex_panic!(InvalidArgument: "map builders are not yet supported")
+        }
         DType::FixedSizeList(elem_dtype, list_size, null) => {
             Box::new(FixedSizeListBuilder::with_capacity(
                 Arc::clone(elem_dtype),

vortex-array/src/builders/tests.rs

@@ -631,6 +631,9 @@ fn create_test_scalars_for_dtype(dtype: &DType, count: usize) -> Vec<Scalar> {
                     .collect();
                 Scalar::fixed_size_list(Arc::clone(element_dtype), elements, *n)
             }
+            DType::Map(..) => {
+                panic!("map builders are not supported until MapArray exists")
+            }
             DType::Struct(fields, n) => {
                 // Create struct scalars with field values.
                 let field_values: Vec<Scalar> = fields

vortex-array/src/canonical.rs

@@ -215,6 +215,9 @@ impl Canonical {
                 // An empty list view is trivially copyable to a list.
                 .with_zero_copy_to_list(true)
             }),
+            DType::Map(..) => {
+                vortex_panic!(InvalidArgument: "canonical map arrays are not yet supported")
+            }
             DType::FixedSizeList(elem_dtype, list_size, null) => Canonical::FixedSizeList(unsafe {
                 FixedSizeListArray::new_unchecked(
                     Canonical::empty(elem_dtype).into_array(),

vortex-array/src/compute/conformance/consistency.rs

@@ -1227,6 +1227,7 @@ fn test_cast_slice_consistency(array: &ArrayRef, ctx: &mut ExecutionCtx) {
                 opposite,
             )]
         }
+        DType::Map(..) => vec![], /* Map arrays are not materializable until their layout is chosen. */
         DType::Struct(fields, nullability) => {
             let opposite = match nullability {
                 Nullability::NonNullable => Nullability::Nullable,

vortex-array/src/dtype/arbitrary/mod.rs

@@ -36,7 +36,7 @@ impl<'a> Arbitrary<'a> for FieldName {
 
 fn random_dtype(u: &mut Unstructured<'_>, depth: u8) -> Result<DType> {
     const BASE_TYPE_COUNT: i32 = 5;
-    const CONTAINER_TYPE_COUNT: i32 = 3;
+    const CONTAINER_TYPE_COUNT: i32 = 4;
     let max_dtype_kind = if depth == 0 {
         BASE_TYPE_COUNT
     } else {
@@ -59,6 +59,13 @@ fn random_dtype(u: &mut Unstructured<'_>, depth: u8) -> Result<DType> {
             u.choose_index(3)?.try_into().vortex_expect("impossible"),
             u.arbitrary()?,
         ),
+        9 => DType::map(
+            random_dtype(u, depth - 1)?.as_nonnullable(),
+            random_dtype(u, depth - 1)?,
+            u.arbitrary()?,
+            u.arbitrary()?,
+        )
+        .vortex_expect("non-nullable generated map keys are always valid"),
         // Null,
         // Extension(ExtDType, Nullability),
         _ => unreachable!("Number out of range"),

vortex-array/src/dtype/coercion.rs

@@ -100,6 +100,20 @@ impl DType {
             return Some(DType::List(Arc::new(elem), union_null));
         }
 
+        if let (DType::Map(lhs, _), DType::Map(rhs, _)) = (self, other) {
+            if lhs.key_dtype() != rhs.key_dtype() || lhs.value_dtype() != rhs.value_dtype() {
+                return None;
+            }
+
+            return DType::map(
+                lhs.key_dtype(),
+                lhs.value_dtype(),
+                lhs.keys_sorted() && rhs.keys_sorted(),
+                union_null,
+            )
+            .ok();
+        }
+
         // Identity (ignoring nullability): return self with union nullability
         if self.eq_ignore_nullability(other) {
             return Some(self.with_nullability(union_null));
@@ -187,6 +201,13 @@ impl DType {
                 && target_elem.can_coerce_from(source_elem);
         }
 
+        if let (DType::Map(target, _), DType::Map(source, _)) = (self, other) {
+            return (self.is_nullable() || !other.is_nullable())
+                && (!target.keys_sorted() || source.keys_sorted())
+                && target.key_dtype() == source.key_dtype()
+                && target.value_dtype() == source.value_dtype();
+        }
+
         // Same type (ignoring nullability): check nullability compatibility
         if self.eq_ignore_nullability(other) {
             return self.is_nullable() || !other.is_nullable();
@@ -804,4 +825,50 @@ mod tests {
             DType::Decimal(DecimalDType::new(15, 5), NonNullable)
         );
     }
+
+    #[test]
+    fn map_least_supertype_unions_outer_nullability_and_intersects_sortedness() {
+        let key = DType::Primitive(PType::I32, NonNullable);
+        let value = DType::Utf8(Nullable);
+        let sorted = DType::map(key.clone(), value.clone(), true, NonNullable).unwrap();
+        let unsorted = DType::map(key.clone(), value.clone(), false, Nullable).unwrap();
+
+        assert_eq!(
+            sorted.least_supertype(&unsorted),
+            Some(DType::map(key, value, false, Nullable).unwrap())
+        );
+    }
+
+    #[test]
+    fn map_least_supertype_requires_identical_key_and_value_dtypes() {
+        let i32_map = DType::map(
+            DType::Primitive(PType::I32, NonNullable),
+            DType::Utf8(Nullable),
+            false,
+            NonNullable,
+        )
+        .unwrap();
+        let i64_map = DType::map(
+            DType::Primitive(PType::I64, NonNullable),
+            DType::Utf8(Nullable),
+            false,
+            NonNullable,
+        )
+        .unwrap();
+
+        assert_eq!(i32_map.least_supertype(&i64_map), None);
+    }
+
+    #[test]
+    fn map_coercion_does_not_create_a_sortedness_assertion() {
+        let key = DType::Primitive(PType::I32, NonNullable);
+        let value = DType::Utf8(Nullable);
+        let sorted = DType::map(key.clone(), value.clone(), true, Nullable).unwrap();
+        let unsorted = DType::map(key.clone(), value, false, Nullable).unwrap();
+        let different_value = DType::map(key, DType::Utf8(NonNullable), false, Nullable).unwrap();
+
+        assert!(!sorted.can_coerce_from(&unsorted));
+        assert!(unsorted.can_coerce_from(&sorted));
+        assert!(!unsorted.can_coerce_from(&different_value));
+    }
 }