diff --git a/fuzz/src/array/fill_null.rs b/fuzz/src/array/fill_null.rs
index f3a49b25637..39b7885c248 100644
--- a/fuzz/src/array/fill_null.rs
+++ b/fuzz/src/array/fill_null.rs
@@ -49,6 +49,7 @@ pub fn fill_null_canonical_array(
         | Canonical::List(_)
         | Canonical::FixedSizeList(_)
         | Canonical::Extension(_) => canonical.into_array().fill_null(fill_value.clone())?,
+        Canonical::Union(_) => unreachable!("Union arrays are not fuzzed yet"),
         Canonical::Variant(_) => unreachable!("Variant arrays are not fuzzed"),
     })
 }
diff --git a/fuzz/src/array/mask.rs b/fuzz/src/array/mask.rs
index f3cecec2b10..5d6e411dd22 100644
--- a/fuzz/src/array/mask.rs
+++ b/fuzz/src/array/mask.rs
@@ -149,6 +149,7 @@ pub fn mask_canonical_array(
                 .with_nullability(masked_storage.dtype().nullability());
             ExtensionArray::new(ext_dtype, masked_storage).into_array()
         }
+        Canonical::Union(_) => unreachable!("Union arrays are not fuzzed yet"),
         Canonical::Variant(_) => unreachable!("Variant arrays are not fuzzed"),
     })
 }
diff --git a/fuzz/src/array/scalar_at.rs b/fuzz/src/array/scalar_at.rs
index e13ad75343d..7b809b5da93 100644
--- a/fuzz/src/array/scalar_at.rs
+++ b/fuzz/src/array/scalar_at.rs
@@ -104,6 +104,7 @@ pub fn scalar_at_canonical_array(
             let storage_scalar = scalar_at_canonical_array(storage_canonical, index, ctx)?;
             Scalar::extension_ref(array.ext_dtype().clone(), storage_scalar)
         }
+        Canonical::Union(_) => unreachable!("Union arrays are not fuzzed yet"),
         Canonical::Variant(_) => unreachable!("Variant arrays are not fuzzed"),
     })
 }
diff --git a/vortex-array/src/aggregate_fn/fns/is_constant/mod.rs b/vortex-array/src/aggregate_fn/fns/is_constant/mod.rs
index 490ede7f640..ad89513e7cd 100644
--- a/vortex-array/src/aggregate_fn/fns/is_constant/mod.rs
+++ b/vortex-array/src/aggregate_fn/fns/is_constant/mod.rs
@@ -404,6 +404,9 @@ impl AggregateFnVTable for IsConstant {
                     Canonical::List(l) => check_listview_constant(l, ctx)?,
                     Canonical::FixedSizeList(f) => check_fixed_size_list_constant(f, ctx)?,
                     Canonical::Null(_) => true,
+                    Canonical::Union(_) => {
+                        vortex_bail!("Union arrays don't support IsConstant yet")
+                    }
                     Canonical::Variant(_) => {
                         vortex_bail!("Variant arrays don't support IsConstant")
                     }
diff --git a/vortex-array/src/aggregate_fn/fns/min_max/mod.rs b/vortex-array/src/aggregate_fn/fns/min_max/mod.rs
index 51601ce76fd..b66b9fe792d 100644
--- a/vortex-array/src/aggregate_fn/fns/min_max/mod.rs
+++ b/vortex-array/src/aggregate_fn/fns/min_max/mod.rs
@@ -420,6 +420,7 @@ impl AggregateFnVTable for MinMax {
                 Canonical::Struct(_)
                 | Canonical::List(_)
                 | Canonical::FixedSizeList(_)
+                | Canonical::Union(_)
                 | Canonical::Variant(_) => {
                     vortex_bail!("Unsupported canonical type for min_max: {}", batch.dtype())
                 }
diff --git a/vortex-array/src/aggregate_fn/fns/uncompressed_size_in_bytes/mod.rs b/vortex-array/src/aggregate_fn/fns/uncompressed_size_in_bytes/mod.rs
index 6b2feb380a7..dd51e25edae 100644
--- a/vortex-array/src/aggregate_fn/fns/uncompressed_size_in_bytes/mod.rs
+++ b/vortex-array/src/aggregate_fn/fns/uncompressed_size_in_bytes/mod.rs
@@ -9,6 +9,7 @@ mod list_view;
 mod null;
 mod primitive;
 mod struct_;
+mod union;
 mod varbinview;
 
 use std::mem::size_of;
@@ -21,6 +22,7 @@ use list_view::list_view_uncompressed_size_in_bytes;
 use null::null_uncompressed_size_in_bytes;
 use primitive::primitive_uncompressed_size_in_bytes;
 use struct_::struct_uncompressed_size_in_bytes;
+use union::union_uncompressed_size_in_bytes;
 use varbinview::varbinview_uncompressed_size_in_bytes;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
@@ -199,6 +201,7 @@ pub(crate) fn canonical_uncompressed_size_in_bytes(
         Canonical::List(array) => list_view_uncompressed_size_in_bytes(array, ctx),
         Canonical::FixedSizeList(array) => fixed_size_list_uncompressed_size_in_bytes(array, ctx),
         Canonical::Struct(array) => struct_uncompressed_size_in_bytes(array, ctx),
+        Canonical::Union(array) => union_uncompressed_size_in_bytes(array, ctx),
         Canonical::Extension(array) => extension_uncompressed_size_in_bytes(array, ctx),
         Canonical::Variant(_) => {
             vortex_bail!("UncompressedSizeInBytes is not supported for Variant arrays")
@@ -229,11 +232,14 @@ pub(crate) fn constant_uncompressed_size_in_bytes(
             array.len(),
             array.scalar().as_binary().value().map(|value| value.len()),
         )?,
-        DType::List(..) | DType::FixedSizeList(..) | DType::Struct(..) | DType::Extension(_) => {
+        DType::List(..)
+        | DType::FixedSizeList(..)
+        | DType::Struct(..)
+        | DType::Union(..)
+        | DType::Extension(_) => {
             let canonical = array.array().clone().execute::<Canonical>(ctx)?;
             return canonical_uncompressed_size_in_bytes(&canonical, ctx);
         }
-        DType::Union(..) => todo!("TODO(connor)[Union]: unimplemented"),
         DType::Variant(_) => {
             vortex_bail!("UncompressedSizeInBytes is not supported for Variant arrays")
         }
diff --git a/vortex-array/src/aggregate_fn/fns/uncompressed_size_in_bytes/union.rs b/vortex-array/src/aggregate_fn/fns/uncompressed_size_in_bytes/union.rs
new file mode 100644
index 00000000000..2f13252de9d
--- /dev/null
+++ b/vortex-array/src/aggregate_fn/fns/uncompressed_size_in_bytes/union.rs
@@ -0,0 +1,25 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_error::VortexResult;
+use vortex_error::vortex_err;
+
+use super::uncompressed_size_in_bytes_u64;
+use crate::ExecutionCtx;
+use crate::arrays::UnionArray;
+use crate::arrays::union::UnionArrayExt;
+
+pub(super) fn union_uncompressed_size_in_bytes(
+    array: &UnionArray,
+    ctx: &mut ExecutionCtx,
+) -> VortexResult<u64> {
+    let mut size = uncompressed_size_in_bytes_u64(array.type_ids(), ctx)?;
+
+    for child in array.iter_children() {
+        size = size
+            .checked_add(uncompressed_size_in_bytes_u64(child, ctx)?)
+            .ok_or_else(|| vortex_err!("uncompressed size in bytes overflowed u64"))?;
+    }
+
+    Ok(size)
+}
diff --git a/vortex-array/src/arrays/chunked/vtable/canonical.rs b/vortex-array/src/arrays/chunked/vtable/canonical.rs
index 12c3bbcd9e4..dcf8af39ae5 100644
--- a/vortex-array/src/arrays/chunked/vtable/canonical.rs
+++ b/vortex-array/src/arrays/chunked/vtable/canonical.rs
@@ -19,11 +19,14 @@ use crate::arrays::FixedSizeListArray;
 use crate::arrays::ListViewArray;
 use crate::arrays::PrimitiveArray;
 use crate::arrays::StructArray;
+use crate::arrays::UnionArray;
 use crate::arrays::VariantArray;
 use crate::arrays::chunked::ChunkedArrayExt;
 use crate::arrays::fixed_size_list::FixedSizeListArrayExt;
 use crate::arrays::listview::ListViewArrayExt;
 use crate::arrays::listview::ListViewRebuildMode;
+use crate::arrays::union::TYPE_IDS_DTYPE;
+use crate::arrays::union::UnionArrayExt;
 use crate::arrays::variant::VariantArrayExt;
 use crate::builders::builder_with_capacity_in;
 use crate::builtins::ArrayBuiltins;
@@ -54,6 +57,7 @@ pub(super) fn _canonicalize(
             let struct_array = pack_struct_chunks(owned_chunks, ctx)?;
             Canonical::Struct(struct_array)
         }
+        DType::Union(..) => Canonical::Union(pack_union_chunks(owned_chunks, ctx)?),
         DType::List(elem_dtype, _) => Canonical::List(swizzle_list_chunks(
             &owned_chunks,
             array.array().validity()?,
@@ -89,6 +93,46 @@ fn pack_struct_chunks(chunks: Vec<ArrayRef>, ctx: &mut ExecutionCtx) -> VortexRe
         .process_results(|iter| StructArray::try_concat(iter))?
 }
 
+/// Packs sparse union chunks into one union with chunked type IDs and sparse children.
+fn pack_union_chunks(chunks: Vec<ArrayRef>, ctx: &mut ExecutionCtx) -> VortexResult<UnionArray> {
+    let union_chunks = chunks
+        .into_iter()
+        .map(|chunk| chunk.execute::<UnionArray>(ctx))
+        .collect::<VortexResult<Vec<_>>>()?;
+    let variants = union_chunks[0].variants().clone();
+    let type_ids = ChunkedArray::try_new(
+        union_chunks
+            .iter()
+            .map(|chunk| chunk.type_ids().clone())
+            .collect(),
+        TYPE_IDS_DTYPE,
+    )?
+    .into_array();
+    let children = variants
+        .variants()
+        .enumerate()
+        .map(|(index, dtype)| {
+            ChunkedArray::try_new(
+                union_chunks
+                    .iter()
+                    .map(|chunk| {
+                        chunk
+                            .child(index)
+                            .vortex_expect("variant index must have a sparse child")
+                            .clone()
+                    })
+                    .collect(),
+                dtype,
+            )
+            .map(IntoArray::into_array)
+        })
+        .collect::<VortexResult<Vec<_>>>()?;
+
+    // SAFETY: Every component was taken from validated UnionArrays with the same dtype and
+    // chunked along identical row boundaries.
+    Ok(unsafe { UnionArray::new_unchecked(type_ids, variants, children) })
+}
+
 /// Packs many [`VariantArray`]s into one [`VariantArray`] with chunked children.
 ///
 /// The caller guarantees there are at least 2 chunks.
diff --git a/vortex-array/src/arrays/chunked/vtable/mod.rs b/vortex-array/src/arrays/chunked/vtable/mod.rs
index 2eabd883742..8bb424c174f 100644
--- a/vortex-array/src/arrays/chunked/vtable/mod.rs
+++ b/vortex-array/src/arrays/chunked/vtable/mod.rs
@@ -251,9 +251,12 @@ impl VTable for Chunked {
 
     fn execute(array: Array<Self>, ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
         match array.dtype() {
-            // Struct, List, FixedSizeList, and Variant need child swizzling that the builder path
-            // cannot express.
-            DType::Struct(..) | DType::List(..) | DType::FixedSizeList(..) | DType::Variant(..) => {
+            // Nested canonical arrays need child swizzling that the builder path cannot express.
+            DType::Struct(..)
+            | DType::List(..)
+            | DType::FixedSizeList(..)
+            | DType::Union(..)
+            | DType::Variant(..) => {
                 // TODO(joe)[#7674]: iterative execution here too
                 Ok(ExecutionResult::done(_canonicalize(array.as_view(), ctx)?))
             }
diff --git a/vortex-array/src/arrays/constant/vtable/canonical.rs b/vortex-array/src/arrays/constant/vtable/canonical.rs
index e160bcc16a9..ff4c1784700 100644
--- a/vortex-array/src/arrays/constant/vtable/canonical.rs
+++ b/vortex-array/src/arrays/constant/vtable/canonical.rs
@@ -8,6 +8,7 @@ use vortex_buffer::Buffer;
 use vortex_buffer::buffer;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
 
 use crate::Canonical;
 use crate::ExecutionCtx;
@@ -23,6 +24,7 @@ use crate::arrays::ListViewArray;
 use crate::arrays::NullArray;
 use crate::arrays::PrimitiveArray;
 use crate::arrays::StructArray;
+use crate::arrays::UnionArray;
 use crate::arrays::VarBinViewArray;
 use crate::arrays::VariantArray;
 use crate::arrays::varbinview::BinaryView;
@@ -30,6 +32,7 @@ use crate::builders::builder_with_capacity;
 use crate::dtype::DType;
 use crate::dtype::DecimalType;
 use crate::dtype::Nullability;
+use crate::dtype::UnionVariants;
 use crate::match_each_decimal_value;
 use crate::match_each_decimal_value_type;
 use crate::match_each_native_ptype;
@@ -164,7 +167,9 @@ pub(crate) fn constant_canonicalize(
                 StructArray::new_unchecked(fields, struct_dtype.clone(), array.len(), validity)
             })
         }
-        DType::Union(..) => todo!("TODO(connor)[Union]: unimplemented"),
+        DType::Union(variants) => {
+            Canonical::Union(constant_canonical_union(scalar, variants, array.len())?)
+        }
         DType::Variant(_) => Canonical::Variant(VariantArray::try_new(
             array.array().clone().into_array(),
             None,
@@ -188,6 +193,51 @@ pub(crate) fn constant_canonicalize(
     })
 }
 
+fn constant_canonical_union(
+    scalar: &Scalar,
+    variants: &UnionVariants,
+    len: usize,
+) -> VortexResult<UnionArray> {
+    if scalar.is_null() {
+        vortex_bail!(
+            "Canonicalizing a null union scalar is not supported until null semantics are defined"
+        )
+    }
+    if variants.variants().any(|variant| variant.is_nullable()) {
+        vortex_bail!("Canonical UnionArray children must be non-nullable")
+    }
+
+    let union = scalar.as_union();
+    let type_id = union
+        .type_id()
+        .vortex_expect("non-null union scalar must have a type ID");
+    let selected_child = union
+        .child_index()
+        .vortex_expect("validated union scalar must select a child");
+    let selected_value = union
+        .value()
+        .vortex_expect("non-null union scalar must have a value");
+
+    let children = variants
+        .variants()
+        .enumerate()
+        .map(|(index, dtype)| {
+            let value = if index == selected_child {
+                selected_value.clone()
+            } else {
+                Scalar::zero_value(&dtype)
+            };
+            ConstantArray::new(value, len).into_array()
+        })
+        .collect::<Vec<_>>();
+
+    UnionArray::try_new(
+        ConstantArray::new(type_id, len).into_array(),
+        variants.clone(),
+        children,
+    )
+}
+
 fn constant_canonical_byte_view(
     scalar_bytes: Option<&[u8]>,
     dtype: &DType,
diff --git a/vortex-array/src/arrays/dict/execute.rs b/vortex-array/src/arrays/dict/execute.rs
index 47a9a051666..b4a6cc05017 100644
--- a/vortex-array/src/arrays/dict/execute.rs
+++ b/vortex-array/src/arrays/dict/execute.rs
@@ -30,6 +30,7 @@ use crate::arrays::VarBinViewArray;
 use crate::arrays::VariantArray;
 use crate::arrays::dict::TakeExecute;
 use crate::arrays::dict::TakeReduce;
+use crate::arrays::union::compute::take_union;
 use crate::arrays::variant::VariantArrayExt;
 
 /// Take from a canonical array using indices (codes), returning a new canonical array.
@@ -53,6 +54,10 @@ pub(crate) fn take_canonical(
             Canonical::FixedSizeList(take_fixed_size_list(&a, codes, ctx))
         }
         Canonical::Struct(a) => Canonical::Struct(take_struct(&a, codes)),
+        Canonical::Union(a) => {
+            let indices = codes.clone().into_array();
+            Canonical::Union(take_union(a.as_view(), &indices)?)
+        }
         Canonical::Extension(a) => Canonical::Extension(take_extension(&a, codes, ctx)),
         Canonical::Variant(a) => {
             let indices = codes.clone().into_array();
diff --git a/vortex-array/src/arrays/filter/execute/mod.rs b/vortex-array/src/arrays/filter/execute/mod.rs
index 52b0373ac49..b4653787a58 100644
--- a/vortex-array/src/arrays/filter/execute/mod.rs
+++ b/vortex-array/src/arrays/filter/execute/mod.rs
@@ -24,6 +24,7 @@ use crate::arrays::NullArray;
 use crate::arrays::VariantArray;
 use crate::arrays::extension::ExtensionArrayExt;
 use crate::arrays::filter::FilterArrayExt;
+use crate::arrays::union::compute::filter_union;
 use crate::arrays::variant::VariantArrayExt;
 use crate::scalar::Scalar;
 use crate::validity::Validity;
@@ -95,6 +96,13 @@ pub(super) fn execute_filter(canonical: Canonical, mask: &Arc<MaskValues>) -> Ca
             Canonical::FixedSizeList(fixed_size_list::filter_fixed_size_list(&a, mask))
         }
         Canonical::Struct(a) => Canonical::Struct(struct_::filter_struct(&a, mask)),
+        Canonical::Union(a) => {
+            let mask = Mask::Values(Arc::clone(mask));
+            Canonical::Union(
+                filter_union(a.as_view(), &mask)
+                    .vortex_expect("UnionArray children must support filter"),
+            )
+        }
         Canonical::Extension(a) => {
             let filtered_storage = a
                 .storage_array()
diff --git a/vortex-array/src/arrays/masked/execute.rs b/vortex-array/src/arrays/masked/execute.rs
index c4173dd0cf6..c5fc9ac90af 100644
--- a/vortex-array/src/arrays/masked/execute.rs
+++ b/vortex-array/src/arrays/masked/execute.rs
@@ -6,6 +6,7 @@
 use std::sync::Arc;
 
 use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
 
 use crate::Canonical;
 use crate::IntoArray;
@@ -50,6 +51,9 @@ pub fn mask_validity_canonical(
             Canonical::FixedSizeList(mask_validity_fixed_size_list(a, validity)?)
         }
         Canonical::Struct(a) => Canonical::Struct(mask_validity_struct(a, validity)?),
+        Canonical::Union(_) => {
+            vortex_bail!("Masking UnionArray is not supported until null semantics are defined")
+        }
         Canonical::Extension(a) => Canonical::Extension(mask_validity_extension(a, validity, ctx)?),
         Canonical::Variant(a) => Canonical::Variant(mask_validity_variant(a, validity, ctx)?),
     })
diff --git a/vortex-array/src/arrays/mod.rs b/vortex-array/src/arrays/mod.rs
index 15b6b5aa6be..8b8f5f28d75 100644
--- a/vortex-array/src/arrays/mod.rs
+++ b/vortex-array/src/arrays/mod.rs
@@ -5,8 +5,8 @@
 //!
 //! Canonical arrays are the default uncompressed representation for a logical dtype:
 //! [`NullArray`], [`BoolArray`], [`PrimitiveArray`], [`DecimalArray`], [`VarBinViewArray`],
-//! [`ListViewArray`], [`FixedSizeListArray`], [`StructArray`], [`ExtensionArray`], and
-//! [`VariantArray`].
+//! [`ListViewArray`], [`FixedSizeListArray`], [`StructArray`], [`UnionArray`],
+//! [`ExtensionArray`], and [`VariantArray`].
 //!
 //! Utility and lazy arrays represent common transformations without immediately materializing
 //! their result. Examples include [`ChunkedArray`] for concatenation, [`ConstantArray`] for repeated
@@ -108,6 +108,10 @@ pub mod struct_;
 pub use struct_::Struct;
 pub use struct_::StructArray;
 
+pub mod union;
+pub use union::Union;
+pub use union::UnionArray;
+
 pub mod varbin;
 pub use varbin::VarBin;
 pub use varbin::VarBinArray;
diff --git a/vortex-array/src/arrays/union/array.rs b/vortex-array/src/arrays/union/array.rs
new file mode 100644
index 00000000000..fef81d2672d
--- /dev/null
+++ b/vortex-array/src/arrays/union/array.rs
@@ -0,0 +1,215 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::iter::once;
+use std::sync::Arc;
+
+use vortex_error::VortexExpect;
+use vortex_error::VortexResult;
+use vortex_error::vortex_ensure;
+use vortex_error::vortex_err;
+
+use crate::ArrayRef;
+use crate::ArraySlots;
+use crate::IntoArray;
+use crate::VortexSessionExecute;
+use crate::array::Array;
+use crate::array::ArrayParts;
+use crate::array::EmptyArrayData;
+use crate::array::TypedArrayRef;
+use crate::arrays::PrimitiveArray;
+use crate::arrays::Union;
+use crate::arrays::union::TYPE_IDS_DTYPE;
+use crate::dtype::DType;
+use crate::dtype::UnionVariants;
+use crate::legacy_session;
+
+/// The row-aligned array of type IDs selecting a union child.
+pub(super) const TYPE_IDS_SLOT: usize = 0;
+/// The offset at which the sparse child arrays begin in the slots vector.
+pub(super) const CHILDREN_OFFSET: usize = 1;
+
+pub(super) fn make_union_parts(
+    type_ids: ArrayRef,
+    variants: UnionVariants,
+    children: &[ArrayRef],
+) -> ArrayParts<Union> {
+    let len = type_ids.len();
+    let slots: ArraySlots = once(Some(type_ids))
+        .chain(children.iter().cloned().map(Some))
+        .collect();
+
+    ArrayParts::new(Union, DType::Union(variants), len, EmptyArrayData).with_slots(slots)
+}
+
+/// Concrete parts of a [`UnionArray`](super::UnionArray).
+pub struct UnionDataParts {
+    /// The union variant schema.
+    pub variants: UnionVariants,
+    /// The row-aligned type IDs.
+    pub type_ids: ArrayRef,
+    /// The row-aligned sparse children in variant order.
+    pub children: Arc<[ArrayRef]>,
+}
+
+/// Accessors for a canonical sparse union array.
+pub trait UnionArrayExt: TypedArrayRef<Union> {
+    /// The union's variant schema.
+    fn variants(&self) -> &UnionVariants {
+        match self.as_ref().dtype() {
+            DType::Union(variants) => variants,
+            _ => unreachable!("UnionArrayExt requires a union dtype"),
+        }
+    }
+
+    /// The row-aligned non-nullable `i8` type IDs.
+    fn type_ids(&self) -> &ArrayRef {
+        self.as_ref().slots()[TYPE_IDS_SLOT]
+            .as_ref()
+            .vortex_expect("UnionArray type_ids slot")
+    }
+
+    /// Iterate over sparse children in variant order.
+    fn iter_children(&self) -> impl ExactSizeIterator<Item = &ArrayRef> + '_ {
+        self.as_ref().slots()[CHILDREN_OFFSET..]
+            .iter()
+            .map(|slot| slot.as_ref().vortex_expect("UnionArray child slot"))
+    }
+
+    /// Return the sparse children in variant order.
+    fn children(&self) -> Arc<[ArrayRef]> {
+        self.iter_children().cloned().collect()
+    }
+
+    /// Return a sparse child by its variant index.
+    fn child(&self, index: usize) -> Option<&ArrayRef> {
+        self.as_ref().slots().get(CHILDREN_OFFSET + index)?.as_ref()
+    }
+
+    /// Return a sparse child selected by a data-level type ID.
+    fn child_by_type_id(&self, type_id: i8) -> Option<&ArrayRef> {
+        self.child(self.variants().tag_to_child_index(type_id)?)
+    }
+
+    /// Return a sparse child selected by its variant name, if present.
+    fn child_by_name_opt(&self, name: impl AsRef<str>) -> Option<&ArrayRef> {
+        self.child(self.variants().find(name)?)
+    }
+
+    /// Return a sparse child selected by its variant name.
+    fn child_by_name(&self, name: impl AsRef<str>) -> VortexResult<&ArrayRef> {
+        let name = name.as_ref();
+        self.child_by_name_opt(name).ok_or_else(|| {
+            vortex_err!(
+                "Variant {name} not found in union array with names {:?}",
+                self.variants().names()
+            )
+        })
+    }
+}
+impl<T: TypedArrayRef<Union>> UnionArrayExt for T {}
+
+/// Validate host-resident type ID values eagerly.
+///
+/// Non-host arrays remain valid structurally and are checked when their values are accessed.
+#[allow(clippy::disallowed_methods)]
+fn validate_type_id_values(array: &Array<Union>) -> VortexResult<()> {
+    if !array.type_ids().is_host() {
+        return Ok(());
+    }
+
+    let type_ids = array
+        .type_ids()
+        .clone()
+        .execute::<PrimitiveArray>(&mut legacy_session().create_execution_ctx())?;
+    for type_id in type_ids.as_slice::<i8>() {
+        vortex_ensure!(
+            array.variants().tag_to_child_index(*type_id).is_some(),
+            "UnionArray type ID {type_id} is not present in {:?}",
+            array.variants().type_ids()
+        );
+    }
+
+    Ok(())
+}
+
+impl Array<Union> {
+    /// Construct a canonical sparse union array.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the components do not satisfy the invariants documented on
+    /// [`Self::new_unchecked`].
+    pub fn new(
+        type_ids: ArrayRef,
+        variants: UnionVariants,
+        children: impl Into<Arc<[ArrayRef]>>,
+    ) -> Self {
+        Self::try_new(type_ids, variants, children).vortex_expect("UnionArray construction failed")
+    }
+
+    /// Try to construct a canonical sparse union array.
+    ///
+    /// Until Union nullability semantics are settled, every child must be non-nullable.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the type IDs and sparse children do not match the variant schema or do
+    /// not all have the same length.
+    pub fn try_new(
+        type_ids: ArrayRef,
+        variants: UnionVariants,
+        children: impl Into<Arc<[ArrayRef]>>,
+    ) -> VortexResult<Self> {
+        vortex_ensure!(
+            type_ids.dtype() == &TYPE_IDS_DTYPE,
+            "UnionArray type_ids must be non-nullable i8, got {}",
+            type_ids.dtype()
+        );
+
+        let children = children.into();
+        let array = Array::try_from_parts(make_union_parts(type_ids, variants, &children))?;
+        validate_type_id_values(&array)?;
+
+        Ok(array)
+    }
+
+    /// Construct a canonical sparse union array without validation.
+    ///
+    /// # Safety
+    ///
+    /// The caller must ensure the type IDs are non-nullable `i8` values declared by `variants`,
+    /// every child has the corresponding variant dtype, and all arrays have the same length. The
+    /// children must currently be non-nullable.
+    pub unsafe fn new_unchecked(
+        type_ids: ArrayRef,
+        variants: UnionVariants,
+        children: impl Into<Arc<[ArrayRef]>>,
+    ) -> Self {
+        let children = children.into();
+        unsafe { Array::from_parts_unchecked(make_union_parts(type_ids, variants, &children)) }
+    }
+
+    /// Deconstruct this array into its type IDs, variant schema, and sparse children.
+    pub fn into_data_parts(self) -> UnionDataParts {
+        let variants = self.variants().clone();
+        let type_ids = self.type_ids().clone();
+        let children = self.children();
+        UnionDataParts {
+            variants,
+            type_ids,
+            children,
+        }
+    }
+
+    /// Create an empty array for a non-nullable union dtype.
+    pub(crate) fn empty(variants: UnionVariants) -> Self {
+        let type_ids = PrimitiveArray::from_iter(Vec::<i8>::new()).into_array();
+        let children: Vec<_> = variants
+            .variants()
+            .map(|dtype| crate::Canonical::empty(&dtype).into_array())
+            .collect();
+
+        Self::new(type_ids, variants, children)
+    }
+}
diff --git a/vortex-array/src/arrays/union/compute/filter.rs b/vortex-array/src/arrays/union/compute/filter.rs
new file mode 100644
index 00000000000..4b26916bdbc
--- /dev/null
+++ b/vortex-array/src/arrays/union/compute/filter.rs
@@ -0,0 +1,30 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_error::VortexResult;
+use vortex_mask::Mask;
+
+use crate::ArrayRef;
+use crate::IntoArray;
+use crate::array::ArrayView;
+use crate::arrays::Union;
+use crate::arrays::UnionArray;
+use crate::arrays::filter::FilterReduce;
+use crate::arrays::union::UnionArrayExt;
+
+pub(crate) fn filter_union(array: ArrayView<'_, Union>, mask: &Mask) -> VortexResult<UnionArray> {
+    let type_ids = array.type_ids().filter(mask.clone())?;
+    let children = array
+        .iter_children()
+        .map(|child| child.filter(mask.clone()))
+        .collect::<VortexResult<Vec<_>>>()?;
+
+    // SAFETY: Filtering every row-aligned component by the same mask preserves all invariants.
+    Ok(unsafe { UnionArray::new_unchecked(type_ids, array.variants().clone(), children) })
+}
+
+impl FilterReduce for Union {
+    fn filter(array: ArrayView<'_, Union>, mask: &Mask) -> VortexResult<Option<ArrayRef>> {
+        Ok(Some(filter_union(array, mask)?.into_array()))
+    }
+}
diff --git a/vortex-array/src/arrays/union/compute/mod.rs b/vortex-array/src/arrays/union/compute/mod.rs
new file mode 100644
index 00000000000..e9208bf2185
--- /dev/null
+++ b/vortex-array/src/arrays/union/compute/mod.rs
@@ -0,0 +1,12 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+mod filter;
+pub(crate) use filter::filter_union;
+
+pub(crate) mod rules;
+
+mod slice;
+
+mod take;
+pub(crate) use take::take_union;
diff --git a/vortex-array/src/arrays/union/compute/rules.rs b/vortex-array/src/arrays/union/compute/rules.rs
new file mode 100644
index 00000000000..4666a99b92f
--- /dev/null
+++ b/vortex-array/src/arrays/union/compute/rules.rs
@@ -0,0 +1,14 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use crate::arrays::Union;
+use crate::arrays::dict::TakeReduceAdaptor;
+use crate::arrays::filter::FilterReduceAdaptor;
+use crate::arrays::slice::SliceReduceAdaptor;
+use crate::optimizer::rules::ParentRuleSet;
+
+pub(crate) const PARENT_RULES: ParentRuleSet<Union> = ParentRuleSet::new(&[
+    ParentRuleSet::lift(&SliceReduceAdaptor(Union)),
+    ParentRuleSet::lift(&FilterReduceAdaptor(Union)),
+    ParentRuleSet::lift(&TakeReduceAdaptor(Union)),
+]);
diff --git a/vortex-array/src/arrays/union/compute/slice.rs b/vortex-array/src/arrays/union/compute/slice.rs
new file mode 100644
index 00000000000..3d912130e1b
--- /dev/null
+++ b/vortex-array/src/arrays/union/compute/slice.rs
@@ -0,0 +1,30 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::ops::Range;
+
+use vortex_error::VortexResult;
+
+use crate::ArrayRef;
+use crate::IntoArray;
+use crate::array::ArrayView;
+use crate::arrays::Union;
+use crate::arrays::UnionArray;
+use crate::arrays::slice::SliceReduce;
+use crate::arrays::union::UnionArrayExt;
+
+impl SliceReduce for Union {
+    fn slice(array: ArrayView<'_, Union>, range: Range<usize>) -> VortexResult<Option<ArrayRef>> {
+        let type_ids = array.type_ids().slice(range.clone())?;
+        let children = array
+            .iter_children()
+            .map(|child| child.slice(range.clone()))
+            .collect::<VortexResult<Vec<ArrayRef>>>()?;
+
+        // SAFETY: Slicing every row-aligned component by the same range preserves all invariants.
+        Ok(Some(
+            unsafe { UnionArray::new_unchecked(type_ids, array.variants().clone(), children) }
+                .into_array(),
+        ))
+    }
+}
diff --git a/vortex-array/src/arrays/union/compute/take.rs b/vortex-array/src/arrays/union/compute/take.rs
new file mode 100644
index 00000000000..9cca9a0794a
--- /dev/null
+++ b/vortex-array/src/arrays/union/compute/take.rs
@@ -0,0 +1,38 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
+
+use crate::ArrayRef;
+use crate::IntoArray;
+use crate::array::ArrayView;
+use crate::arrays::Union;
+use crate::arrays::UnionArray;
+use crate::arrays::dict::TakeReduce;
+use crate::arrays::union::UnionArrayExt;
+
+pub(crate) fn take_union(
+    array: ArrayView<'_, Union>,
+    indices: &ArrayRef,
+) -> VortexResult<UnionArray> {
+    if indices.dtype().is_nullable() {
+        vortex_bail!("Taking UnionArray with nullable indices is not supported yet")
+    }
+
+    let type_ids = array.type_ids().take(indices.clone())?;
+    let children = array
+        .iter_children()
+        .map(|child| child.take(indices.clone()))
+        .collect::<VortexResult<Vec<_>>>()?;
+
+    // SAFETY: Taking every row-aligned component with the same non-null indices preserves all
+    // invariants and cannot introduce nulls.
+    Ok(unsafe { UnionArray::new_unchecked(type_ids, array.variants().clone(), children) })
+}
+
+impl TakeReduce for Union {
+    fn take(array: ArrayView<'_, Union>, indices: &ArrayRef) -> VortexResult<Option<ArrayRef>> {
+        Ok(Some(take_union(array, indices)?.into_array()))
+    }
+}
diff --git a/vortex-array/src/arrays/union/mod.rs b/vortex-array/src/arrays/union/mod.rs
new file mode 100644
index 00000000000..b8671ccafd8
--- /dev/null
+++ b/vortex-array/src/arrays/union/mod.rs
@@ -0,0 +1,30 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Canonical sparse union arrays.
+//!
+//! A [`UnionArray`] stores one non-nullable `i8` type ID per row followed by one row-aligned child
+//! for each variant. The type ID selects which child's value is active for a row; values in all
+//! other children at that row are placeholders.
+//!
+//! Union nullability semantics are still being designed, so this encoding currently requires every
+//! variant child to be non-nullable.
+
+use crate::dtype::DType;
+use crate::dtype::Nullability;
+use crate::dtype::PType;
+
+mod array;
+pub use array::UnionArrayExt;
+pub use array::UnionDataParts;
+pub use vtable::UnionArray;
+
+pub(crate) mod compute;
+
+mod vtable;
+pub use vtable::Union;
+
+pub(crate) const TYPE_IDS_DTYPE: DType = DType::Primitive(PType::I8, Nullability::NonNullable);
+
+#[cfg(test)]
+mod tests;
diff --git a/vortex-array/src/arrays/union/tests.rs b/vortex-array/src/arrays/union/tests.rs
new file mode 100644
index 00000000000..e800647d56b
--- /dev/null
+++ b/vortex-array/src/arrays/union/tests.rs
@@ -0,0 +1,202 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_buffer::ByteBufferMut;
+use vortex_buffer::buffer;
+use vortex_error::VortexResult;
+use vortex_mask::Mask;
+use vortex_session::registry::ReadContext;
+
+use crate::ArrayContext;
+use crate::IntoArray;
+use crate::VortexSessionExecute;
+use crate::array_session;
+use crate::arrays::BoolArray;
+use crate::arrays::Chunked;
+use crate::arrays::ChunkedArray;
+use crate::arrays::ConstantArray;
+use crate::arrays::PrimitiveArray;
+use crate::arrays::Union;
+use crate::arrays::UnionArray;
+use crate::arrays::union::UnionArrayExt;
+use crate::dtype::DType;
+use crate::dtype::Nullability;
+use crate::dtype::PType;
+use crate::dtype::UnionVariants;
+use crate::scalar::Scalar;
+use crate::serde::SerializeOptions;
+use crate::serde::SerializedArray;
+use crate::validity::Validity;
+
+fn variants() -> VortexResult<UnionVariants> {
+    UnionVariants::try_new(
+        ["number", "flag"].into(),
+        vec![
+            DType::Primitive(PType::I32, Nullability::NonNullable),
+            DType::Bool(Nullability::NonNullable),
+        ],
+        vec![5, 9],
+    )
+}
+
+fn union_array() -> VortexResult<UnionArray> {
+    UnionArray::try_new(
+        PrimitiveArray::from_iter([5i8, 9, 5]).into_array(),
+        variants()?,
+        vec![
+            PrimitiveArray::from_iter([10i32, 0, 30]).into_array(),
+            BoolArray::from_iter([false, true, false]).into_array(),
+        ],
+    )
+}
+
+#[test]
+fn scalar_at_uses_type_id_indirection() -> VortexResult<()> {
+    let array = union_array()?;
+    let mut ctx = array_session().create_execution_ctx();
+
+    assert_eq!(
+        array.child_by_name("flag")?.dtype(),
+        &DType::Bool(Nullability::NonNullable)
+    );
+    assert!(array.child_by_name_opt("missing").is_none());
+    assert_eq!(
+        array.execute_scalar(0, &mut ctx)?,
+        Scalar::union(variants()?, 5, 10i32.into())?
+    );
+    assert_eq!(
+        array.execute_scalar(1, &mut ctx)?,
+        Scalar::union(variants()?, 9, true.into())?
+    );
+    assert!(matches!(array.validity()?, Validity::NonNullable));
+
+    Ok(())
+}
+
+#[test]
+fn validates_sparse_components() -> VortexResult<()> {
+    let children = vec![
+        PrimitiveArray::from_iter([10i32, 0, 30]).into_array(),
+        BoolArray::from_iter([false, true, false]).into_array(),
+    ];
+
+    let unknown_type_id = UnionArray::try_new(
+        PrimitiveArray::from_iter([5i8, 7, 5]).into_array(),
+        variants()?,
+        children.clone(),
+    );
+    assert!(unknown_type_id.is_err());
+
+    let mismatched_lengths = UnionArray::try_new(
+        PrimitiveArray::from_iter([5i8, 9]).into_array(),
+        variants()?,
+        children,
+    );
+    assert!(mismatched_lengths.is_err());
+
+    let nullable_child = UnionArray::try_new(
+        PrimitiveArray::from_iter([5i8, 9, 5]).into_array(),
+        variants()?,
+        vec![
+            PrimitiveArray::new(buffer![10i32, 0, 30], Validity::AllValid).into_array(),
+            BoolArray::from_iter([false, true, false]).into_array(),
+        ],
+    );
+    assert!(nullable_child.is_err());
+
+    Ok(())
+}
+
+#[test]
+fn structural_operations_preserve_sparse_alignment() -> VortexResult<()> {
+    let array = union_array()?.into_array();
+    let mut ctx = array_session().create_execution_ctx();
+
+    let sliced = array.slice(1..3)?;
+    let filtered = array.filter(Mask::from_iter([true, false, true]))?;
+    let taken = array.take(PrimitiveArray::from_iter([2u32, 1]).into_array())?;
+
+    assert_eq!(
+        sliced.execute_scalar(0, &mut ctx)?,
+        Scalar::union(variants()?, 9, true.into())?
+    );
+    assert_eq!(
+        filtered.execute_scalar(1, &mut ctx)?,
+        Scalar::union(variants()?, 5, 30i32.into())?
+    );
+    assert_eq!(
+        taken.execute_scalar(0, &mut ctx)?,
+        Scalar::union(variants()?, 5, 30i32.into())?
+    );
+
+    Ok(())
+}
+
+#[test]
+fn serde_roundtrip() -> VortexResult<()> {
+    let session = array_session();
+    let mut execution_ctx = session.create_execution_ctx();
+    let array = union_array()?;
+    let dtype = array.dtype().clone();
+    let len = array.len();
+    let array_ctx = ArrayContext::empty();
+    let serialized =
+        array
+            .clone()
+            .into_array()
+            .serialize(&array_ctx, &session, &SerializeOptions::default())?;
+    let mut concat = ByteBufferMut::empty();
+    for buffer in serialized {
+        concat.extend_from_slice(buffer.as_ref());
+    }
+    let decoded = SerializedArray::try_from(concat.freeze())?.decode(
+        &dtype,
+        len,
+        &ReadContext::new(array_ctx.to_ids()),
+        &session,
+    )?;
+    let decoded = decoded.as_::<Union>();
+
+    assert_eq!(decoded.variants(), array.variants());
+    for index in 0..len {
+        assert_eq!(
+            decoded.array().execute_scalar(index, &mut execution_ctx)?,
+            array.execute_scalar(index, &mut execution_ctx)?
+        );
+    }
+
+    Ok(())
+}
+
+#[test]
+fn constant_union_executes_to_sparse_union() -> VortexResult<()> {
+    let scalar = Scalar::union(variants()?, 9, true.into())?;
+    let mut ctx = array_session().create_execution_ctx();
+    let array = ConstantArray::new(scalar.clone(), 3)
+        .into_array()
+        .execute::<UnionArray>(&mut ctx)?;
+
+    assert_eq!(array.len(), 3);
+    assert_eq!(array.execute_scalar(2, &mut ctx)?, scalar);
+
+    Ok(())
+}
+
+#[test]
+fn chunked_union_packs_components() -> VortexResult<()> {
+    let first = union_array()?.into_array().slice(0..1)?;
+    let second = union_array()?.into_array().slice(1..3)?;
+    let dtype = first.dtype().clone();
+    let chunked = ChunkedArray::try_new(vec![first, second], dtype)?.into_array();
+    let mut ctx = array_session().create_execution_ctx();
+    let canonical = chunked.execute::<UnionArray>(&mut ctx)?;
+
+    assert!(canonical.type_ids().is::<Chunked>());
+    assert!(canonical.iter_children().all(|child| child.is::<Chunked>()));
+    assert_eq!(
+        canonical.execute_scalar(2, &mut ctx)?,
+        Scalar::union(variants()?, 5, 30i32.into())?
+    );
+
+    Ok(())
+}
diff --git a/vortex-array/src/arrays/union/vtable/mod.rs b/vortex-array/src/arrays/union/vtable/mod.rs
new file mode 100644
index 00000000000..4023699c256
--- /dev/null
+++ b/vortex-array/src/arrays/union/vtable/mod.rs
@@ -0,0 +1,196 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
+use vortex_error::vortex_ensure;
+use vortex_error::vortex_err;
+use vortex_error::vortex_panic;
+use vortex_session::VortexSession;
+use vortex_session::registry::CachedId;
+
+use crate::ArrayRef;
+use crate::ExecutionCtx;
+use crate::ExecutionResult;
+use crate::array::Array;
+use crate::array::ArrayId;
+use crate::array::ArrayParts;
+use crate::array::ArrayView;
+use crate::array::EmptyArrayData;
+use crate::array::VTable;
+use crate::array::with_empty_buffers;
+use crate::arrays::union::TYPE_IDS_DTYPE;
+use crate::arrays::union::UnionArrayExt;
+use crate::arrays::union::array::CHILDREN_OFFSET;
+use crate::arrays::union::array::TYPE_IDS_SLOT;
+use crate::arrays::union::array::make_union_parts;
+use crate::arrays::union::compute::rules::PARENT_RULES;
+use crate::buffer::BufferHandle;
+use crate::builders::ArrayBuilder;
+use crate::dtype::DType;
+use crate::serde::ArrayChildren;
+
+mod operations;
+mod validity;
+
+/// A canonical sparse [`Union`]-encoded array.
+pub type UnionArray = Array<Union>;
+
+/// The canonical sparse Union encoding.
+#[derive(Clone, Debug)]
+pub struct Union;
+
+impl VTable for Union {
+    type TypedArrayData = EmptyArrayData;
+    type OperationsVTable = Self;
+    type ValidityVTable = Self;
+
+    fn id(&self) -> ArrayId {
+        static ID: CachedId = CachedId::new("vortex.union");
+        *ID
+    }
+
+    fn validate(
+        &self,
+        _data: &EmptyArrayData,
+        dtype: &DType,
+        len: usize,
+        slots: &[Option<ArrayRef>],
+    ) -> VortexResult<()> {
+        let DType::Union(variants) = dtype else {
+            vortex_bail!("Expected union dtype, found {dtype}")
+        };
+        vortex_ensure!(
+            variants.variants().all(|variant| !variant.is_nullable()),
+            "UnionArray children must be non-nullable"
+        );
+        vortex_ensure!(
+            slots.len() == CHILDREN_OFFSET + variants.len(),
+            "UnionArray has {} slots but expected {}",
+            slots.len(),
+            CHILDREN_OFFSET + variants.len()
+        );
+
+        let type_ids = slots[TYPE_IDS_SLOT]
+            .as_ref()
+            .ok_or_else(|| vortex_err!("UnionArray is missing its type_ids slot"))?;
+        vortex_ensure!(
+            type_ids.dtype() == &TYPE_IDS_DTYPE,
+            "UnionArray type_ids must be non-nullable i8, got {}",
+            type_ids.dtype()
+        );
+        vortex_ensure!(
+            type_ids.len() == len,
+            "UnionArray type_ids length {} does not match outer length {len}",
+            type_ids.len()
+        );
+
+        for (index, variant_dtype) in variants.variants().enumerate() {
+            let child = slots[CHILDREN_OFFSET + index]
+                .as_ref()
+                .ok_or_else(|| vortex_err!("UnionArray is missing child {index}"))?;
+            vortex_ensure!(
+                child.len() == len,
+                "UnionArray child {index} length {} does not match outer length {len}",
+                child.len()
+            );
+            vortex_ensure!(
+                child.dtype() == &variant_dtype,
+                "UnionArray child {index} has dtype {} but expected {variant_dtype}",
+                child.dtype()
+            );
+        }
+
+        Ok(())
+    }
+
+    fn nbuffers(_array: ArrayView<'_, Self>) -> usize {
+        0
+    }
+
+    fn buffer(_array: ArrayView<'_, Self>, idx: usize) -> BufferHandle {
+        vortex_panic!("UnionArray buffer index {idx} out of bounds")
+    }
+
+    fn buffer_name(_array: ArrayView<'_, Self>, idx: usize) -> Option<String> {
+        vortex_panic!("UnionArray buffer_name index {idx} out of bounds")
+    }
+
+    fn with_buffers(
+        &self,
+        array: ArrayView<'_, Self>,
+        buffers: &[BufferHandle],
+    ) -> VortexResult<ArrayParts<Self>> {
+        with_empty_buffers(self, array, buffers)
+    }
+
+    fn serialize(
+        _array: ArrayView<'_, Self>,
+        _session: &VortexSession,
+    ) -> VortexResult<Option<Vec<u8>>> {
+        Ok(Some(vec![]))
+    }
+
+    fn deserialize(
+        &self,
+        dtype: &DType,
+        len: usize,
+        metadata: &[u8],
+        buffers: &[BufferHandle],
+        children: &dyn ArrayChildren,
+        _session: &VortexSession,
+    ) -> VortexResult<ArrayParts<Self>> {
+        vortex_ensure!(metadata.is_empty(), "UnionArray expects empty metadata");
+        vortex_ensure!(buffers.is_empty(), "UnionArray expects no buffers");
+        let DType::Union(variants) = dtype else {
+            vortex_bail!("Expected union dtype, found {dtype}")
+        };
+        vortex_ensure!(
+            children.len() == CHILDREN_OFFSET + variants.len(),
+            "UnionArray expected {} children, found {}",
+            CHILDREN_OFFSET + variants.len(),
+            children.len()
+        );
+
+        let type_ids = children.get(TYPE_IDS_SLOT, &TYPE_IDS_DTYPE, len)?;
+        let sparse_children = variants
+            .variants()
+            .enumerate()
+            .map(|(index, dtype)| children.get(CHILDREN_OFFSET + index, &dtype, len))
+            .collect::<VortexResult<Vec<_>>>()?;
+
+        Ok(make_union_parts(
+            type_ids,
+            variants.clone(),
+            &sparse_children,
+        ))
+    }
+
+    fn slot_name(array: ArrayView<'_, Self>, idx: usize) -> String {
+        if idx == TYPE_IDS_SLOT {
+            "type_ids".to_string()
+        } else {
+            array.variants().names()[idx - CHILDREN_OFFSET].to_string()
+        }
+    }
+
+    fn execute(array: Array<Self>, _ctx: &mut ExecutionCtx) -> VortexResult<ExecutionResult> {
+        Ok(ExecutionResult::done(array))
+    }
+
+    fn append_to_builder(
+        _array: ArrayView<'_, Self>,
+        _builder: &mut dyn ArrayBuilder,
+        _ctx: &mut ExecutionCtx,
+    ) -> VortexResult<()> {
+        vortex_bail!("append_to_builder is not supported for UnionArray yet")
+    }
+
+    fn reduce_parent(
+        array: ArrayView<'_, Self>,
+        parent: &ArrayRef,
+        child_idx: usize,
+    ) -> VortexResult<Option<ArrayRef>> {
+        PARENT_RULES.evaluate(array, parent, child_idx)
+    }
+}
diff --git a/vortex-array/src/arrays/union/vtable/operations.rs b/vortex-array/src/arrays/union/vtable/operations.rs
new file mode 100644
index 00000000000..debd1b4ebac
--- /dev/null
+++ b/vortex-array/src/arrays/union/vtable/operations.rs
@@ -0,0 +1,37 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_error::VortexResult;
+use vortex_error::vortex_err;
+
+use crate::ExecutionCtx;
+use crate::array::ArrayView;
+use crate::array::OperationsVTable;
+use crate::arrays::Union;
+use crate::arrays::union::UnionArrayExt;
+use crate::scalar::Scalar;
+
+impl OperationsVTable<Union> for Union {
+    fn scalar_at(
+        array: ArrayView<'_, Union>,
+        index: usize,
+        ctx: &mut ExecutionCtx,
+    ) -> VortexResult<Scalar> {
+        let type_id = array
+            .type_ids()
+            .execute_scalar(index, ctx)?
+            .as_primitive()
+            .typed_value::<i8>()
+            .ok_or_else(|| vortex_err!("UnionArray type ID at index {index} is null"))?;
+        let child_index = array
+            .variants()
+            .tag_to_child_index(type_id)
+            .ok_or_else(|| vortex_err!("Unknown UnionArray type ID {type_id}"))?;
+        let child = array
+            .child(child_index)
+            .ok_or_else(|| vortex_err!("UnionArray is missing child {child_index}"))?;
+        let child_scalar = child.execute_scalar(index, ctx)?;
+
+        Scalar::union(array.variants().clone(), type_id, child_scalar)
+    }
+}
diff --git a/vortex-array/src/arrays/union/vtable/validity.rs b/vortex-array/src/arrays/union/vtable/validity.rs
new file mode 100644
index 00000000000..0231b7a3334
--- /dev/null
+++ b/vortex-array/src/arrays/union/vtable/validity.rs
@@ -0,0 +1,16 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use vortex_error::VortexResult;
+
+use crate::array::ArrayView;
+use crate::array::ValidityVTable;
+use crate::arrays::Union;
+use crate::validity::Validity;
+
+impl ValidityVTable<Union> for Union {
+    fn validity(_array: ArrayView<'_, Union>) -> VortexResult<Validity> {
+        // TODO(connor)[Union]: define how union and child nullability interact.
+        Ok(Validity::NonNullable)
+    }
+}
diff --git a/vortex-array/src/canonical.rs b/vortex-array/src/canonical.rs
index 71da52e1300..2e4851ac15d 100644
--- a/vortex-array/src/canonical.rs
+++ b/vortex-array/src/canonical.rs
@@ -35,6 +35,8 @@ use crate::arrays::Primitive;
 use crate::arrays::PrimitiveArray;
 use crate::arrays::Struct;
 use crate::arrays::StructArray;
+use crate::arrays::Union;
+use crate::arrays::UnionArray;
 use crate::arrays::VarBinView;
 use crate::arrays::VarBinViewArray;
 use crate::arrays::Variant;
@@ -47,6 +49,7 @@ use crate::arrays::listview::ListViewDataParts;
 use crate::arrays::listview::ListViewRebuildMode;
 use crate::arrays::primitive::PrimitiveDataParts;
 use crate::arrays::struct_::StructDataParts;
+use crate::arrays::union::UnionDataParts;
 use crate::arrays::varbinview::VarBinViewDataParts;
 use crate::arrays::variant::VariantArrayExt;
 use crate::dtype::DType;
@@ -69,7 +72,7 @@ use crate::validity::Validity;
 ///
 /// Each `Canonical` variant has a corresponding [`DType`] variant, with the notable exception of
 /// [`Canonical::VarBinView`], which is the canonical encoding for both [`DType::Utf8`] and
-/// [`DType::Binary`]. [`DType::Union`] does not yet have a public canonical array.
+/// [`DType::Binary`].
 ///
 /// # Laziness
 ///
@@ -80,8 +83,9 @@ use crate::validity::Validity;
 ///
 /// # Arrow interoperability
 ///
-/// All of the Vortex canonical encodings have an equivalent Arrow encoding that can be built
-/// zero-copy, and the corresponding Arrow array types can also be built directly.
+/// Except for the currently experimental union encoding, Vortex canonical encodings have an
+/// equivalent Arrow encoding that can be built zero-copy, and the corresponding Arrow array types
+/// can also be built directly.
 ///
 /// The full list of canonical types and their equivalent Arrow array types are:
 ///
@@ -128,6 +132,7 @@ pub enum Canonical {
     List(ListViewArray),
     FixedSizeList(FixedSizeListArray),
     Struct(StructArray),
+    Union(UnionArray),
     /// Canonical storage for extension dtypes, wrapping the canonical form of the storage dtype.
     Extension(ExtensionArray),
     /// Canonical storage for dynamic variant values, optionally with typed shredded paths.
@@ -146,6 +151,7 @@ macro_rules! match_each_canonical {
             Canonical::List($ident) => $eval,
             Canonical::FixedSizeList($ident) => $eval,
             Canonical::Struct($ident) => $eval,
+            Canonical::Union($ident) => $eval,
             Canonical::Variant($ident) => $eval,
             Canonical::Extension($ident) => $eval,
         }
@@ -228,7 +234,7 @@ impl Canonical {
                     Validity::from(n),
                 )
             }),
-            DType::Union(..) => todo!("TODO(connor)[Union]: unimplemented"),
+            DType::Union(variants) => Canonical::Union(UnionArray::empty(variants.clone())),
             DType::Variant(_) => {
                 vortex_panic!(InvalidArgument: "Canonical empty is not supported for Variant")
             }
@@ -401,6 +407,24 @@ impl Canonical {
         }
     }
 
+    /// Return this canonical array as a sparse [`UnionArray`].
+    pub fn as_union(&self) -> &UnionArray {
+        if let Canonical::Union(a) = self {
+            a
+        } else {
+            vortex_panic!("Cannot get UnionArray from {:?}", &self)
+        }
+    }
+
+    /// Unwrap this canonical array as a sparse [`UnionArray`].
+    pub fn into_union(self) -> UnionArray {
+        if let Canonical::Union(a) = self {
+            a
+        } else {
+            vortex_panic!("Cannot unwrap UnionArray from {:?}", &self)
+        }
+    }
+
     pub fn as_extension(&self) -> &ExtensionArray {
         if let Canonical::Extension(a) = self {
             a
@@ -652,6 +676,7 @@ impl Executable for CanonicalValidity {
                     StructArray::new_unchecked(fields, struct_fields, len, validity.execute(ctx)?)
                 })))
             }
+            union @ Canonical::Union(_) => Ok(CanonicalValidity(union)),
             Canonical::Extension(ext) => Ok(CanonicalValidity(Canonical::Extension(
                 ExtensionArray::new(
                     ext.ext_dtype().clone(),
@@ -829,6 +854,27 @@ impl Executable for RecursiveCanonical {
                     )
                 })))
             }
+            Canonical::Union(union) => {
+                let UnionDataParts {
+                    variants,
+                    type_ids,
+                    children,
+                } = union.into_data_parts();
+                let type_ids = type_ids.execute::<RecursiveCanonical>(ctx)?.0.into_array();
+                let children = children
+                    .iter()
+                    .cloned()
+                    .map(|child| {
+                        child
+                            .execute::<RecursiveCanonical>(ctx)
+                            .map(|canonical| canonical.0.into_array())
+                    })
+                    .collect::<VortexResult<Arc<[_]>>>()?;
+
+                Ok(RecursiveCanonical(Canonical::Union(unsafe {
+                    UnionArray::new_unchecked(type_ids, variants, children)
+                })))
+            }
             Canonical::Extension(ext) => Ok(RecursiveCanonical(Canonical::Extension(
                 ExtensionArray::new(
                     ext.ext_dtype().clone(),
@@ -1003,6 +1049,18 @@ impl Executable for StructArray {
     }
 }
 
+/// Execute the array to canonical form and unwrap as a [`UnionArray`].
+///
+/// This will panic if the array's dtype is not union.
+impl Executable for UnionArray {
+    fn execute(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Self> {
+        match array.try_downcast::<Union>() {
+            Ok(union_array) => Ok(union_array),
+            Err(array) => Ok(Canonical::execute(array, ctx)?.into_union()),
+        }
+    }
+}
+
 /// Execute the array to canonical form and unwrap as a [`VariantArray`].
 ///
 /// This will panic if the array's dtype is not variant.
@@ -1032,6 +1090,7 @@ pub enum CanonicalView<'a> {
     List(ArrayView<'a, ListView>),
     FixedSizeList(ArrayView<'a, FixedSizeList>),
     Struct(ArrayView<'a, Struct>),
+    Union(ArrayView<'a, Union>),
     Extension(ArrayView<'a, Extension>),
     Variant(ArrayView<'a, Variant>),
 }
@@ -1047,6 +1106,7 @@ impl From<CanonicalView<'_>> for Canonical {
             CanonicalView::List(a) => Canonical::List(a.into_owned()),
             CanonicalView::FixedSizeList(a) => Canonical::FixedSizeList(a.into_owned()),
             CanonicalView::Struct(a) => Canonical::Struct(a.into_owned()),
+            CanonicalView::Union(a) => Canonical::Union(a.into_owned()),
             CanonicalView::Extension(a) => Canonical::Extension(a.into_owned()),
             CanonicalView::Variant(a) => Canonical::Variant(a.into_owned()),
         }
@@ -1065,6 +1125,7 @@ impl CanonicalView<'_> {
             CanonicalView::List(a) => a.array().clone(),
             CanonicalView::FixedSizeList(a) => a.array().clone(),
             CanonicalView::Struct(a) => a.array().clone(),
+            CanonicalView::Union(a) => a.array().clone(),
             CanonicalView::Extension(a) => a.array().clone(),
             CanonicalView::Variant(a) => a.array().clone(),
         }
@@ -1083,6 +1144,7 @@ impl Matcher for AnyCanonical {
             || array.is::<Primitive>()
             || array.is::<Decimal>()
             || array.is::<Struct>()
+            || array.is::<Union>()
             || array.is::<ListView>()
             || array.is::<FixedSizeList>()
             || array.is::<VarBinView>()
@@ -1102,6 +1164,8 @@ impl Matcher for AnyCanonical {
             Some(CanonicalView::Decimal(a))
         } else if let Some(a) = array.as_opt::<Struct>() {
             Some(CanonicalView::Struct(a))
+        } else if let Some(a) = array.as_opt::<Union>() {
+            Some(CanonicalView::Union(a))
         } else if let Some(a) = array.as_opt::<ListView>() {
             Some(CanonicalView::List(a))
         } else if let Some(a) = array.as_opt::<FixedSizeList>() {
diff --git a/vortex-array/src/dtype/dtype_impl.rs b/vortex-array/src/dtype/dtype_impl.rs
index e3594f818c6..ce52079a94e 100644
--- a/vortex-array/src/dtype/dtype_impl.rs
+++ b/vortex-array/src/dtype/dtype_impl.rs
@@ -312,7 +312,7 @@ impl DType {
                 }
                 Some(sum)
             }
-            Union(..) => todo!("TODO(connor)[Union]: unimplemented"),
+            Union(..) => None,
             Variant(_) => None,
             Extension(ext) => ext.storage_dtype().element_size(),
         }
diff --git a/vortex-array/src/lib.rs b/vortex-array/src/lib.rs
index 9d3a59eb7a0..04be12a6d03 100644
--- a/vortex-array/src/lib.rs
+++ b/vortex-array/src/lib.rs
@@ -40,7 +40,7 @@
 //! # Built-in, Lazy, and Experimental Arrays
 //!
 //! Built-in arrays live in [`arrays`]. Some are canonical (`PrimitiveArray`, `StructArray`,
-//! `VarBinViewArray`); others are utility or lazy arrays such as [`ChunkedArray`],
+//! `UnionArray`, `VarBinViewArray`); others are utility or lazy arrays such as [`ChunkedArray`],
 //! [`ConstantArray`], [`FilterArray`], [`SliceArray`], and [`ScalarFnArray`].
 //! Lazy arrays defer work so compute kernels can operate on encoded data or prune children
 //! before materialization.
diff --git a/vortex-array/src/scalar_fn/fns/cast/mod.rs b/vortex-array/src/scalar_fn/fns/cast/mod.rs
index 20852779d42..1959da25a1d 100644
--- a/vortex-array/src/scalar_fn/fns/cast/mod.rs
+++ b/vortex-array/src/scalar_fn/fns/cast/mod.rs
@@ -185,6 +185,9 @@ fn cast_canonical(
         CanonicalView::List(a) => <ListView as CastKernel>::cast(a, dtype, ctx),
         CanonicalView::FixedSizeList(a) => <FixedSizeList as CastKernel>::cast(a, dtype, ctx),
         CanonicalView::Struct(a) => struct_cast(a, dtype, ctx),
+        CanonicalView::Union(_) => {
+            vortex_bail!("Union arrays don't support casting yet")
+        }
         CanonicalView::Extension(a) => <Extension as CastReduce>::cast(a, dtype),
         CanonicalView::Variant(_) => {
             vortex_bail!("Variant arrays don't support casting")
diff --git a/vortex-array/src/session/mod.rs b/vortex-array/src/session/mod.rs
index 2cb8414cbfd..c844a7b8015 100644
--- a/vortex-array/src/session/mod.rs
+++ b/vortex-array/src/session/mod.rs
@@ -27,6 +27,7 @@ use crate::arrays::Masked;
 use crate::arrays::Null;
 use crate::arrays::Primitive;
 use crate::arrays::Struct;
+use crate::arrays::Union;
 use crate::arrays::VarBin;
 use crate::arrays::VarBinView;
 use crate::arrays::Variant;
@@ -72,6 +73,7 @@ impl Default for ArraySession {
         this.register(ListView);
         this.register(FixedSizeList);
         this.register(Struct);
+        this.register(Union);
         this.register(Variant);
         this.register(Extension);
 
diff --git a/vortex-compressor/src/compressor/cascade.rs b/vortex-compressor/src/compressor/cascade.rs
index 9dbbc611448..d5797b0b37f 100644
--- a/vortex-compressor/src/compressor/cascade.rs
+++ b/vortex-compressor/src/compressor/cascade.rs
@@ -14,6 +14,7 @@ use vortex_array::arrays::ExtensionArray;
 use vortex_array::arrays::FixedSizeListArray;
 use vortex_array::arrays::Masked;
 use vortex_array::arrays::StructArray;
+use vortex_array::arrays::UnionArray;
 use vortex_array::arrays::Variant;
 use vortex_array::arrays::VariantArray;
 use vortex_array::arrays::extension::ExtensionArrayExt;
@@ -23,6 +24,7 @@ use vortex_array::arrays::listview::list_from_list_view;
 use vortex_array::arrays::masked::MaskedArraySlotsExt;
 use vortex_array::arrays::scalar_fn::AnyScalarFn;
 use vortex_array::arrays::struct_::StructArrayExt;
+use vortex_array::arrays::union::UnionArrayExt;
 use vortex_array::arrays::variant::VariantArrayExt;
 use vortex_array::scalar::Scalar;
 use vortex_error::VortexResult;
@@ -130,6 +132,18 @@ impl CascadingCompressor {
                 )?
                 .into_array())
             }
+            Canonical::Union(union_array) => {
+                let type_ids = self.compress(union_array.type_ids(), exec_ctx)?;
+                let children = union_array
+                    .iter_children()
+                    .map(|child| self.compress(child, exec_ctx))
+                    .collect::<VortexResult<Vec<_>>>()?;
+
+                Ok(
+                    UnionArray::try_new(type_ids, union_array.variants().clone(), children)?
+                        .into_array(),
+                )
+            }
             Canonical::List(list_view_array) => {
                 if list_view_array.is_zero_copy_to_list() || list_view_array.elements().is_empty() {
                     let list_array = list_from_list_view(list_view_array, exec_ctx)?;
diff --git a/vortex-duckdb/src/exporter/canonical.rs b/vortex-duckdb/src/exporter/canonical.rs
index 7ee47a40642..0bf9067250c 100644
--- a/vortex-duckdb/src/exporter/canonical.rs
+++ b/vortex-duckdb/src/exporter/canonical.rs
@@ -32,6 +32,9 @@ pub(crate) fn new_exporter(
         Canonical::List(array) => list_view::new_exporter(array, cache, ctx),
         Canonical::FixedSizeList(array) => fixed_size_list::new_exporter(array, cache, ctx),
         Canonical::Struct(array) => struct_::new_exporter(array, cache, ctx),
+        Canonical::Union(_) => {
+            vortex_bail!("Union arrays can't be exported to DuckDB yet")
+        }
         Canonical::Extension(ext) => extension::new_exporter(ext, ctx),
         Canonical::Variant(_) => {
             vortex_bail!("Variant arrays can't be exported to DuckDB")