mz_repr/

row.rs

// Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

//! In-memory `Tag`-based encoding of a tuple of `Datum`s.
//!
//! See `doc/developer/row-encoding.md` for the size limits this encoding and
//! its datum types impose.

use std::borrow::Borrow;
use std::cell::{Cell, RefCell};
use std::cmp::Ordering;
use std::convert::{TryFrom, TryInto};
use std::fmt::{self, Debug};
use std::hash::{Hash, Hasher};
use std::marker::PhantomData;
use std::mem::{size_of, transmute};
use std::ops::Deref;
use std::str;

use chrono::{DateTime, Datelike, NaiveDate, NaiveDateTime, NaiveTime, Timelike, Utc};
use compact_bytes::CompactBytes;
use mz_ore::cast::{CastFrom, ReinterpretCast};
use mz_ore::soft_assert_no_log;
use mz_ore::vec::Vector;
use mz_persist_types::Codec64;
use num_enum::{IntoPrimitive, TryFromPrimitive};
use ordered_float::OrderedFloat;
#[cfg(any(test, feature = "proptest"))]
use proptest::prelude::*;
#[cfg(any(test, feature = "proptest"))]
use proptest::strategy::{BoxedStrategy, Strategy};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::adt::array::{
    Array, ArrayDimension, ArrayDimensions, InvalidArrayError, MAX_ARRAY_DIMENSIONS,
};
use crate::adt::date::Date;
use crate::adt::interval::Interval;
use crate::adt::mz_acl_item::{AclItem, MzAclItem};
use crate::adt::numeric;
use crate::adt::numeric::Numeric;
use crate::adt::range::{
    self, InvalidRangeError, Range, RangeBound, RangeInner, RangeLowerBound, RangeUpperBound,
};
use crate::adt::timestamp::CheckedTimestamp;
#[cfg(any(test, feature = "proptest"))]
use crate::scalar::arb_datum;
use crate::scalar::{DatumKind, SqlScalarType};
use crate::{Datum, RelationDesc, Timestamp};

pub(crate) mod encode;
pub mod iter;

include!(concat!(env!("OUT_DIR"), "/mz_repr.row.rs"));

/// A packed representation for `Datum`s.
///
/// `Datum` is easy to work with but very space inefficient. A `Datum::Int32(42)`
/// is laid out in memory like this:
///
///   tag: 3
///   padding: 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0
///   data: 0 0 0 42
///   padding: 0 0 0 0 0 0 0 0 0 0 0 0
///
/// For a total of 32 bytes! The second set of padding is needed in case we were
/// to write a 16-byte datum into this location. The first set of padding is
/// needed to align that hypothetical decimal to a 16 bytes boundary.
///
/// A `Row` stores zero or more `Datum`s without any padding. We avoid the need
/// for the first set of padding by only providing access to the `Datum`s via
/// calls to `ptr::read_unaligned`, which on modern x86 is barely penalized. We
/// avoid the need for the second set of padding by not providing mutable access
/// to the `Datum`. Instead, `Row` is append-only.
///
/// A `Row` can be built from a collection of `Datum`s using `Row::pack`, but it
/// is more efficient to use `Row::pack_slice` so that a right-sized allocation
/// can be created. If that is not possible, consider using the row buffer
/// pattern: allocate one row, pack into it, and then call [`Row::clone`] to
/// receive a copy of that row, leaving behind the original allocation to pack
/// future rows.
///
/// Creating a row via [`Row::pack_slice`]:
///
/// ```
/// # use mz_repr::{Row, Datum};
/// let row = Row::pack_slice(&[Datum::Int32(0), Datum::Int32(1), Datum::Int32(2)]);
/// assert_eq!(row.unpack(), vec![Datum::Int32(0), Datum::Int32(1), Datum::Int32(2)])
/// ```
///
/// `Row`s can be unpacked by iterating over them:
///
/// ```
/// # use mz_repr::{Row, Datum};
/// let row = Row::pack_slice(&[Datum::Int32(0), Datum::Int32(1), Datum::Int32(2)]);
/// assert_eq!(row.iter().nth(1).unwrap(), Datum::Int32(1));
/// ```
///
/// If you want random access to the `Datum`s in a `Row`, use `Row::unpack` to create a `Vec<Datum>`
/// ```
/// # use mz_repr::{Row, Datum};
/// let row = Row::pack_slice(&[Datum::Int32(0), Datum::Int32(1), Datum::Int32(2)]);
/// let datums = row.unpack();
/// assert_eq!(datums[1], Datum::Int32(1));
/// ```
///
/// # Performance
///
/// Rows are dynamically sized, but up to a fixed size their data is stored in-line.
/// It is best to re-use a `Row` across multiple `Row` creation calls, as this
/// avoids the allocations involved in `Row::new()`.
#[derive(Default, Eq, PartialEq, Serialize, Deserialize)]
pub struct Row {
    data: CompactBytes,
}

impl Row {
    const SIZE: usize = CompactBytes::MAX_INLINE;

    /// A variant of `Row::from_proto` that allows for reuse of internal allocs
    /// and validates the decoding against a provided [`RelationDesc`].
    pub fn decode_from_proto(
        &mut self,
        proto: &ProtoRow,
        desc: &RelationDesc,
    ) -> Result<(), String> {
        let mut packer = self.packer();
        for (col_idx, _, _) in desc.iter_all() {
            let d = match proto.datums.get(col_idx.to_raw()) {
                Some(x) => x,
                None => {
                    packer.push(Datum::Null);
                    continue;
                }
            };
            packer.try_push_proto(d)?;
        }

        Ok(())
    }

    /// Allocate an empty `Row` with a pre-allocated capacity.
    #[inline]
    pub fn with_capacity(cap: usize) -> Self {
        Self {
            data: CompactBytes::with_capacity(cap),
        }
    }

    /// Create an empty `Row`.
    #[inline]
    pub const fn empty() -> Self {
        Self {
            data: CompactBytes::empty(),
        }
    }

    /// Creates a new row from supplied bytes.
    ///
    /// # Safety
    ///
    /// This method relies on `data` being an appropriate row encoding, and can
    /// result in unsafety if this is not the case.
    pub unsafe fn from_bytes_unchecked(data: &[u8]) -> Self {
        Row {
            data: CompactBytes::new(data),
        }
    }

    /// Constructs a [`RowPacker`] that will pack datums into this row's
    /// allocation.
    ///
    /// This method clears the existing contents of the row, but retains the
    /// allocation.
    pub fn packer(&mut self) -> RowPacker<'_> {
        self.clear();
        RowPacker { row: self }
    }

    /// Take some `Datum`s and pack them into a `Row`.
    ///
    /// This method builds a `Row` by repeatedly increasing the backing
    /// allocation. If the contents of the iterator are known ahead of
    /// time, consider [`Row::with_capacity`] to right-size the allocation
    /// first, and then [`RowPacker::extend`] to populate it with `Datum`s.
    /// This avoids the repeated allocation resizing and copying.
    pub fn pack<'a, I, D>(iter: I) -> Row
    where
        I: IntoIterator<Item = D>,
        D: Borrow<Datum<'a>>,
    {
        let mut row = Row::default();
        row.packer().extend(iter);
        row
    }

    /// Use `self` to pack `iter`, and then clone the result.
    ///
    /// This is a convenience method meant to reduce boilerplate around row
    /// formation.
    pub fn pack_using<'a, I, D>(&mut self, iter: I) -> Row
    where
        I: IntoIterator<Item = D>,
        D: Borrow<Datum<'a>>,
    {
        self.packer().extend(iter);
        self.clone()
    }

    /// Like [`Row::pack`], but the provided iterator is allowed to produce an
    /// error, in which case the packing operation is aborted and the error
    /// returned.
    pub fn try_pack<'a, I, D, E>(iter: I) -> Result<Row, E>
    where
        I: IntoIterator<Item = Result<D, E>>,
        D: Borrow<Datum<'a>>,
    {
        let mut row = Row::default();
        row.packer().try_extend(iter)?;
        Ok(row)
    }

    /// Pack a slice of `Datum`s into a `Row`.
    ///
    /// This method has the advantage over `pack` that it can determine the required
    /// allocation before packing the elements, ensuring only one allocation and no
    /// redundant copies required.
    pub fn pack_slice<'a>(slice: &[Datum<'a>]) -> Row {
        // Pre-allocate the needed number of bytes.
        let mut row = Row::with_capacity(datums_size(slice.iter()));
        row.packer().extend(slice.iter());
        row
    }

    /// Returns the total amount of bytes used by this row.
    pub fn byte_len(&self) -> usize {
        let heap_size = if self.data.spilled() {
            self.data.len()
        } else {
            0
        };
        let inline_size = std::mem::size_of::<Self>();
        inline_size.saturating_add(heap_size)
    }

    /// The length of the encoded row in bytes. Does not include the size of the `Row` struct itself.
    pub fn data_len(&self) -> usize {
        self.data.len()
    }

    /// Returns the total capacity in bytes used by this row.
    pub fn byte_capacity(&self) -> usize {
        self.data.capacity()
    }

    /// Extracts a Row slice containing the entire [`Row`].
    #[inline]
    pub fn as_row_ref(&self) -> &RowRef {
        // SAFETY: `Row` contains valid row data, by construction.
        unsafe { RowRef::from_slice(self.data.as_slice()) }
    }

    /// Clear the contents of the [`Row`], leaving any allocation in place.
    #[inline]
    fn clear(&mut self) {
        self.data.clear();
    }
}

impl Borrow<RowRef> for Row {
    #[inline]
    fn borrow(&self) -> &RowRef {
        self.as_row_ref()
    }
}

impl AsRef<RowRef> for Row {
    #[inline]
    fn as_ref(&self) -> &RowRef {
        self.as_row_ref()
    }
}

impl Deref for Row {
    type Target = RowRef;

    #[inline]
    fn deref(&self) -> &Self::Target {
        self.as_row_ref()
    }
}

// Nothing depends on Row being exactly 24, we just want to add visibility to the size.
static_assertions::const_assert_eq!(std::mem::size_of::<Row>(), 24);

impl Clone for Row {
    fn clone(&self) -> Self {
        Row {
            data: self.data.clone(),
        }
    }

    fn clone_from(&mut self, source: &Self) {
        self.data.clone_from(&source.data);
    }
}

// Row's `Hash` implementation defers to `RowRef` to ensure they hash equivalently.
impl std::hash::Hash for Row {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.as_row_ref().hash(state)
    }
}

#[cfg(any(test, feature = "proptest"))]
impl Arbitrary for Row {
    type Parameters = prop::collection::SizeRange;
    type Strategy = BoxedStrategy<Row>;

    fn arbitrary_with(size: Self::Parameters) -> Self::Strategy {
        prop::collection::vec(arb_datum(true), size)
            .prop_map(|items| {
                let mut row = Row::default();
                let mut packer = row.packer();
                for item in items.iter() {
                    let datum: Datum<'_> = item.into();
                    packer.push(datum);
                }
                row
            })
            .boxed()
    }
}

impl PartialOrd for Row {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for Row {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.as_ref().cmp(other.as_ref())
    }
}

/// A [`Row`] that serializes as protobuf-encoded [`ProtoRow`] bytes.
///
/// `Row`'s own serde impl emits the raw bytes of the in-memory `Tag` based
/// datum encoding, which is free to change between releases. Use this wrapper
/// instead wherever a row is serialized into a durable, cross-version format,
/// such as the stable LIR plan format. `ProtoRow` already carries the needed
/// backward compatibility obligation: it is persist's storage codec for
/// `SourceData`, and `row.proto` is covered by the buf breaking lint.
#[derive(
    Clone,
    Default,
    Eq,
    PartialEq,
    Ord,
    PartialOrd,
    Hash,
    Serialize,
    Deserialize
)]
pub struct StableRow(#[serde(with = "stable_row_proto")] pub Row);

impl From<Row> for StableRow {
    fn from(row: Row) -> Self {
        StableRow(row)
    }
}

impl Deref for StableRow {
    type Target = Row;

    fn deref(&self) -> &Row {
        &self.0
    }
}

impl Debug for StableRow {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        self.0.fmt(f)
    }
}

mod stable_row_proto {
    use mz_proto::RustType;
    use prost::Message;
    use serde::de::Error;
    use serde::{Deserialize, Deserializer, Serializer};

    use crate::row::{ProtoRow, Row};

    pub fn serialize<S: Serializer>(row: &Row, serializer: S) -> Result<S::Ok, S::Error> {
        serializer.serialize_bytes(&row.into_proto().encode_to_vec())
    }

    pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Row, D::Error> {
        let bytes = serde_bytes::ByteBuf::deserialize(deserializer)?;
        let proto = ProtoRow::decode(bytes.as_slice()).map_err(D::Error::custom)?;
        Row::from_proto(proto).map_err(D::Error::custom)
    }
}

#[allow(missing_debug_implementations)]
mod columnation {
    use columnation::{Columnation, Region};
    use mz_ore::region::LgAllocRegion;

    use crate::Row;

    /// Region allocation for `Row` data.
    ///
    /// Content bytes are stored in stable contiguous memory locations,
    /// and then a `Row` referencing them is falsified.
    pub struct RowStack {
        region: LgAllocRegion<u8>,
    }

    impl RowStack {
        const LIMIT: usize = 2 << 20;
    }

    // Implement `Default` manually to specify a region allocation limit.
    impl Default for RowStack {
        fn default() -> Self {
            Self {
                // Limit the region size to 2MiB.
                region: LgAllocRegion::with_limit(Self::LIMIT),
            }
        }
    }

    impl Columnation for Row {
        type InnerRegion = RowStack;
    }

    impl Region for RowStack {
        type Item = Row;
        #[inline]
        fn clear(&mut self) {
            self.region.clear();
        }
        #[inline(always)]
        unsafe fn copy(&mut self, item: &Row) -> Row {
            if item.data.spilled() {
                let bytes = self.region.copy_slice(&item.data[..]);
                Row {
                    data: compact_bytes::CompactBytes::from_raw_parts(
                        bytes.as_mut_ptr(),
                        item.data.len(),
                        item.data.capacity(),
                    ),
                }
            } else {
                item.clone()
            }
        }

        fn reserve_items<'a, I>(&mut self, items: I)
        where
            Self: 'a,
            I: Iterator<Item = &'a Self::Item> + Clone,
        {
            let size = items
                .filter(|row| row.data.spilled())
                .map(|row| row.data.len())
                .sum();
            let size = std::cmp::min(size, Self::LIMIT);
            self.region.reserve(size);
        }

        fn reserve_regions<'a, I>(&mut self, regions: I)
        where
            Self: 'a,
            I: Iterator<Item = &'a Self> + Clone,
        {
            let size = regions.map(|r| r.region.len()).sum();
            let size = std::cmp::min(size, Self::LIMIT);
            self.region.reserve(size);
        }

        fn heap_size(&self, callback: impl FnMut(usize, usize)) {
            self.region.heap_size(callback)
        }
    }
}

mod columnar {
    use columnar::common::PushIndexAs;
    use columnar::{
        AsBytes, Borrow, Clear, Columnar, Container, FromBytes, Index, IndexAs, Len, Push,
    };
    use mz_ore::cast::CastFrom;
    use std::ops::Range;

    use crate::{Row, RowRef};

    #[derive(
        Copy,
        Clone,
        Debug,
        Default,
        PartialEq,
        serde::Serialize,
        serde::Deserialize
    )]
    pub struct Rows<BC = Vec<u64>, VC = Vec<u8>> {
        /// Bounds container; provides indexed access to offsets.
        bounds: BC,
        /// Values container; provides slice access to bytes.
        values: VC,
    }

    impl Columnar for Row {
        #[inline(always)]
        fn copy_from(&mut self, other: columnar::Ref<'_, Self>) {
            self.clear();
            self.data.extend_from_slice(other.data());
        }
        #[inline(always)]
        fn into_owned(other: columnar::Ref<'_, Self>) -> Self {
            other.to_owned()
        }
        type Container = Rows;
        #[inline(always)]
        fn reborrow<'b, 'a: 'b>(thing: columnar::Ref<'a, Self>) -> columnar::Ref<'b, Self>
        where
            Self: 'a,
        {
            thing
        }
    }

    impl<BC: PushIndexAs<u64>> Borrow for Rows<BC, Vec<u8>> {
        type Ref<'a> = &'a RowRef;
        type Borrowed<'a>
            = Rows<BC::Borrowed<'a>, &'a [u8]>
        where
            Self: 'a;
        #[inline(always)]
        fn borrow<'a>(&'a self) -> Self::Borrowed<'a> {
            Rows {
                bounds: self.bounds.borrow(),
                values: self.values.borrow(),
            }
        }
        #[inline(always)]
        fn reborrow<'c, 'a: 'c>(item: Self::Borrowed<'a>) -> Self::Borrowed<'c>
        where
            Self: 'a,
        {
            Rows {
                bounds: BC::reborrow(item.bounds),
                values: item.values,
            }
        }

        fn reborrow_ref<'b, 'a: 'b>(item: Self::Ref<'a>) -> Self::Ref<'b>
        where
            Self: 'a,
        {
            item
        }
    }

    impl<BC: PushIndexAs<u64>> Container for Rows<BC, Vec<u8>> {
        fn extend_from_self(&mut self, other: Self::Borrowed<'_>, range: Range<usize>) {
            if !range.is_empty() {
                // Imported bounds will be relative to this starting offset.
                let values_len: u64 = self.values.len().try_into().expect("must fit");

                // Push all bytes that we can, all at once.
                let other_lower = if range.start == 0 {
                    0
                } else {
                    other.bounds.index_as(range.start - 1)
                };
                let other_upper = other.bounds.index_as(range.end - 1);
                self.values.extend_from_self(
                    other.values,
                    usize::try_from(other_lower).expect("must fit")
                        ..usize::try_from(other_upper).expect("must fit"),
                );

                // Each bound needs to be shifted by `values_len - other_lower`.
                if values_len == other_lower {
                    self.bounds.extend_from_self(other.bounds, range);
                } else {
                    for index in range {
                        let shifted = other.bounds.index_as(index) - other_lower + values_len;
                        self.bounds.push(&shifted)
                    }
                }
            }
        }
        fn reserve_for<'a, I>(&mut self, selves: I)
        where
            Self: 'a,
            I: Iterator<Item = Self::Borrowed<'a>> + Clone,
        {
            self.bounds.reserve_for(selves.clone().map(|r| r.bounds));
            self.values.reserve_for(selves.map(|r| r.values));
        }
    }

    impl<'a, BC: AsBytes<'a>, VC: AsBytes<'a>> AsBytes<'a> for Rows<BC, VC> {
        const SLICE_COUNT: usize = BC::SLICE_COUNT + VC::SLICE_COUNT;
        #[inline(always)]
        fn get_byte_slice(&self, index: usize) -> (u64, &'a [u8]) {
            mz_ore::soft_assert_no_log!(index < Self::SLICE_COUNT);
            if index < BC::SLICE_COUNT {
                self.bounds.get_byte_slice(index)
            } else {
                self.values.get_byte_slice(index - BC::SLICE_COUNT)
            }
        }
    }
    impl<'a, BC: FromBytes<'a>, VC: FromBytes<'a>> FromBytes<'a> for Rows<BC, VC> {
        const SLICE_COUNT: usize = BC::SLICE_COUNT + VC::SLICE_COUNT;
        #[inline(always)]
        fn from_bytes(bytes: &mut impl Iterator<Item = &'a [u8]>) -> Self {
            Self {
                bounds: FromBytes::from_bytes(bytes),
                values: FromBytes::from_bytes(bytes),
            }
        }
    }

    impl<BC: Len, VC> Len for Rows<BC, VC> {
        #[inline(always)]
        fn len(&self) -> usize {
            self.bounds.len()
        }
    }

    impl<'a, BC: Len + IndexAs<u64>> Index for Rows<BC, &'a [u8]> {
        type Ref = &'a RowRef;
        #[inline(always)]
        fn get(&self, index: usize) -> Self::Ref {
            let lower = if index == 0 {
                0
            } else {
                self.bounds.index_as(index - 1)
            };
            let upper = self.bounds.index_as(index);
            let lower = usize::cast_from(lower);
            let upper = usize::cast_from(upper);
            // SAFETY: self.values contains only valid row data, and self.metadata delimits only ranges
            // that correspond to the original rows.
            unsafe { RowRef::from_slice(&self.values[lower..upper]) }
        }
    }
    impl<'a, BC: Len + IndexAs<u64>> Index for &'a Rows<BC, Vec<u8>> {
        type Ref = &'a RowRef;
        #[inline(always)]
        fn get(&self, index: usize) -> Self::Ref {
            let lower = if index == 0 {
                0
            } else {
                self.bounds.index_as(index - 1)
            };
            let upper = self.bounds.index_as(index);
            let lower = usize::cast_from(lower);
            let upper = usize::cast_from(upper);
            // SAFETY: self.values contains only valid row data, and self.metadata delimits only ranges
            // that correspond to the original rows.
            unsafe { RowRef::from_slice(&self.values[lower..upper]) }
        }
    }

    impl<BC: Push<u64>> Push<&Row> for Rows<BC> {
        #[inline(always)]
        fn push(&mut self, item: &Row) {
            self.values.extend_from_slice(item.data.as_slice());
            self.bounds.push(u64::cast_from(self.values.len()));
        }
    }
    impl<BC: Push<u64>> Push<Row> for Rows<BC> {
        #[inline(always)]
        fn push(&mut self, item: Row) {
            self.push(&item);
        }
    }
    impl<BC: for<'a> Push<&'a u64>> Push<&RowRef> for Rows<BC> {
        #[inline(always)]
        fn push(&mut self, item: &RowRef) {
            self.values.extend_from_slice(item.data());
            self.bounds.push(&u64::cast_from(self.values.len()));
        }
    }
    impl<BC: Clear, VC: Clear> Clear for Rows<BC, VC> {
        #[inline(always)]
        fn clear(&mut self) {
            self.bounds.clear();
            self.values.clear();
        }
    }
}

/// A contiguous slice of bytes that are row data.
///
/// A [`RowRef`] is to [`Row`] as [`prim@str`] is to [`String`].
#[derive(PartialEq, Eq, Hash)]
#[repr(transparent)]
pub struct RowRef([u8]);

impl RowRef {
    /// Create a [`RowRef`] from a slice of data.
    ///
    /// # Safety
    ///
    /// We do not check that the provided slice is valid [`Row`] data; the caller is required to
    /// ensure this.
    pub unsafe fn from_slice(row: &[u8]) -> &RowRef {
        #[allow(clippy::as_conversions)]
        let ptr = row as *const [u8] as *const RowRef;
        // SAFETY: We know `ptr` is non-null and aligned because it came from a &[u8].
        unsafe { &*ptr }
    }

    /// Unpack `self` into a `Vec<Datum>` for efficient random access.
    pub fn unpack(&self) -> Vec<Datum<'_>> {
        // It's usually cheaper to unpack twice to figure out the right length than it is to grow the vec as we go
        let len = self.iter().count();
        let mut vec = Vec::with_capacity(len);
        vec.extend(self.iter());
        vec
    }

    /// Return the first [`Datum`] in `self`
    ///
    /// Panics if the [`RowRef`] is empty.
    pub fn unpack_first(&self) -> Datum<'_> {
        self.iter().next().unwrap()
    }

    /// Iterate the [`Datum`] elements of the [`RowRef`].
    pub fn iter(&self) -> DatumListIter<'_> {
        DatumListIter { data: &self.0 }
    }

    /// Return the byte length of this [`RowRef`].
    pub fn byte_len(&self) -> usize {
        self.0.len()
    }

    /// For debugging only.
    pub fn data(&self) -> &[u8] {
        &self.0
    }

    /// True iff there is no data in this [`RowRef`].
    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }
}

impl ToOwned for RowRef {
    type Owned = Row;

    fn to_owned(&self) -> Self::Owned {
        // SAFETY: RowRef has the invariant that the wrapped data must be a valid Row encoding.
        unsafe { Row::from_bytes_unchecked(&self.0) }
    }
}

impl<'a> IntoIterator for &'a RowRef {
    type Item = Datum<'a>;
    type IntoIter = DatumListIter<'a>;

    fn into_iter(self) -> DatumListIter<'a> {
        DatumListIter { data: &self.0 }
    }
}

/// These implementations order first by length, and then by slice contents.
/// This allows many comparisons to complete without dereferencing memory.
/// Warning: These order by the u8 array representation, and NOT by Datum::cmp.
impl PartialOrd for RowRef {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for RowRef {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        match self.0.len().cmp(&other.0.len()) {
            std::cmp::Ordering::Less => std::cmp::Ordering::Less,
            std::cmp::Ordering::Greater => std::cmp::Ordering::Greater,
            std::cmp::Ordering::Equal => self.0.cmp(&other.0),
        }
    }
}

impl fmt::Debug for RowRef {
    /// Debug representation using the internal datums
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("RowRef{")?;
        f.debug_list().entries(&*self).finish()?;
        f.write_str("}")
    }
}

/// Packs datums into a [`Row`].
///
/// Creating a `RowPacker` via [`Row::packer`] starts a packing operation on the
/// row. A packing operation always starts from scratch: the existing contents
/// of the underlying row are cleared.
///
/// To complete a packing operation, drop the `RowPacker`.
#[derive(Debug)]
pub struct RowPacker<'a> {
    row: &'a mut Row,
}

/// Infallible conversion from a [`Datum`] to a typed value.
///
/// Used by [`DatumList::typed_iter`] to yield elements as `T` rather than
/// raw `Datum`s. At runtime, `T` is always `Datum<'a>`, so the conversion
/// is identity.
///
/// See `doc/developer/design/20260311_sqlfunc_generic.md` for the design
/// behind the generic type parameter and type erasure.
///
/// This trait is sealed and cannot be implemented outside of this crate.
pub trait FromDatum<'a>:
    Sized + PartialEq + std::borrow::Borrow<Datum<'a>> + sealed::Sealed
{
    fn from_datum(datum: Datum<'a>) -> Self;
}

mod sealed {
    use crate::Datum;

    pub trait Sealed {}
    impl<'a> Sealed for Datum<'a> {}
}

impl<'a> FromDatum<'a> for Datum<'a> {
    #[inline]
    fn from_datum(datum: Datum<'a>) -> Self {
        datum
    }
}

#[derive(Debug, Clone)]
pub struct DatumListIter<'a> {
    data: &'a [u8],
}

#[derive(Debug, Clone)]
pub struct DatumListTypedIter<'a, T> {
    inner: DatumListIter<'a>,
    _phantom: PhantomData<fn() -> T>,
}

#[derive(Debug, Clone)]
pub struct DatumDictIter<'a> {
    data: &'a [u8],
    prev_key: Option<&'a str>,
}

#[derive(Debug, Clone)]
pub struct DatumDictTypedIter<'a, T> {
    inner: DatumDictIter<'a>,
    _phantom: PhantomData<fn() -> T>,
}

/// `RowArena` is used to hold on to temporary `Row`s for functions like `eval` that need to create complex `Datum`s but don't have a `Row` to put them in yet.
#[derive(Debug)]
pub struct RowArena {
    // A stack of byte regions, used as a bump allocator. Bytes handed to
    // `push_bytes` are *copied* into the active (last) region and a reference
    // into that region is returned.
    //
    // The invariant that keeps returned references valid for the arena's
    // lifetime is that a region is never reallocated once it holds data: when
    // the active region lacks spare capacity for a push we allocate a *new*,
    // larger region rather than growing the current one (which would move its
    // bytes and dangle outstanding references). The outer `Vec` may itself
    // reallocate as regions are added, but that only moves the `Vec<u8>`
    // headers, not the heap buffers they own, so references remain valid.
    //
    // `clear` retains only the largest region (emptied) to right-size the arena
    // for reuse; reusing one region across `clear` cycles makes a steady-state
    // workload (e.g. decoding rows one at a time) allocation-free.
    inner: RefCell<Vec<Vec<u8>>>,
    // A single recycled scratch buffer backing `RowArena::writer`. A writer takes ownership of this
    // buffer (or allocates a fresh one if absent), builds into it, and on drop returns it here for
    // the next writer to reuse — so building values incrementally does not allocate per use once the
    // buffer reaches its high-water mark. Holding `Option` (rather than the buffer directly) means
    // `writer` borrows this cell only transiently, to take and return the buffer, never across the
    // writer's lifetime. That keeps nested writers sound: a writer obtained while another is live
    // finds the slot empty and allocates its own buffer instead of double-borrowing.
    scratch: RefCell<Option<Vec<u8>>>,
    // Optional ceiling on the bytes this arena will hold, and a running total of what it holds.
    // `None` is unbounded, which is what every arena in a dataflow must keep using: a budget that
    // can change mid-run would make dataflow evaluation non-deterministic (see
    // [`RowArena::with_budget`]). A budget is for evaluating a user-authored expression in a shared
    // process, where that expression's memory use must be bounded (see `mz_adapter::webhook`).
    //
    // NOTE: exceeding the budget does not make a push fail. The pushes are infallible, and a
    // refused push would hand back a truncated value, i.e. a corrupt datum. The budget is instead a
    // *reported* condition: `over_budget` is polled by whoever is able to return an error, which
    // for scalar expressions is the evaluator between calls.
    budget: Option<usize>,
    allocated: Cell<usize>,
}

// DatumList and DatumDict defined here rather than near Datum because we need private access to the unsafe data field

/// A sequence of Datums
///
/// The type parameter `T` represents the element type of the list. It is a
/// phantom parameter that carries no runtime data — the actual elements are
/// stored as serialized bytes and `T` is not enforced at runtime. It is up
/// to the caller to ensure `T` matches the actual element type. The default
/// `T = Datum<'a>` means existing code that writes `DatumList<'a>` continues
/// to work unchanged.
///
/// See `doc/developer/design/20260311_sqlfunc_generic.md` for the design
/// behind the generic type parameter.
pub struct DatumList<'a, T = Datum<'a>> {
    /// Points at the serialized datums
    data: &'a [u8],
    _phantom: PhantomData<fn() -> T>,
}

impl<'a, T> DatumList<'a, T> {
    /// Private constructor. All `DatumList` values should be created through
    /// this function to keep the `PhantomData` bookkeeping in one place.
    pub(crate) fn new(data: &'a [u8]) -> Self {
        DatumList {
            data,
            _phantom: PhantomData,
        }
    }
}

impl<'a, T> Clone for DatumList<'a, T> {
    fn clone(&self) -> Self {
        *self
    }
}

impl<'a, T> Copy for DatumList<'a, T> {}

impl<'a, T> Debug for DatumList<'a, T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_list().entries(self.iter()).finish()
    }
}

impl<'a, T> PartialEq for DatumList<'a, T> {
    #[inline(always)]
    fn eq(&self, other: &DatumList<'a, T>) -> bool {
        self.iter().eq(other.iter())
    }
}

impl<'a, T> Eq for DatumList<'a, T> {}

impl<'a, T> Hash for DatumList<'a, T> {
    #[inline(always)]
    fn hash<H: Hasher>(&self, state: &mut H) {
        for d in self.iter() {
            d.hash(state);
        }
    }
}

impl<T> Ord for DatumList<'_, T> {
    #[inline(always)]
    fn cmp(&self, other: &DatumList<'_, T>) -> Ordering {
        // Grow the stack: lists can be arbitrarily deeply nested (e.g. jsonb).
        mz_ore::stack::maybe_grow(|| self.iter().cmp(other.iter()))
    }
}

impl<T> PartialOrd for DatumList<'_, T> {
    #[inline(always)]
    fn partial_cmp(&self, other: &DatumList<'_, T>) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

/// A mapping from string keys to Datums
///
/// The type parameter `T` represents the value type of the map. It is a
/// phantom parameter — the actual values are stored as serialized bytes and
/// `T` is not enforced at runtime. It is up to the caller to ensure `T`
/// matches the actual value type. The default `T = Datum<'a>` means existing
/// code that writes `DatumMap<'a>` continues to work unchanged.
///
/// See `doc/developer/design/20260311_sqlfunc_generic.md` for the design
/// behind the generic type parameter.
pub struct DatumMap<'a, T = Datum<'a>> {
    /// Points at the serialized datums, which should be sorted in key order
    data: &'a [u8],
    _phantom: PhantomData<fn() -> T>,
}

impl<'a, T> DatumMap<'a, T> {
    /// Private constructor. All `DatumMap` values should be created through
    /// this function to keep the `PhantomData` bookkeeping in one place.
    pub(crate) fn new(data: &'a [u8]) -> Self {
        DatumMap {
            data,
            _phantom: PhantomData,
        }
    }
}

impl<'a, T> Clone for DatumMap<'a, T> {
    fn clone(&self) -> Self {
        *self
    }
}

impl<'a, T> Copy for DatumMap<'a, T> {}

impl<'a, T> PartialEq for DatumMap<'a, T> {
    #[inline(always)]
    fn eq(&self, other: &DatumMap<'a, T>) -> bool {
        self.iter().eq(other.iter())
    }
}

impl<'a, T> Eq for DatumMap<'a, T> {}

impl<'a, T> Hash for DatumMap<'a, T> {
    #[inline(always)]
    fn hash<H: Hasher>(&self, state: &mut H) {
        for (k, v) in self.iter() {
            k.hash(state);
            v.hash(state);
        }
    }
}

impl<'a, T> Ord for DatumMap<'a, T> {
    #[inline(always)]
    fn cmp(&self, other: &DatumMap<'a, T>) -> Ordering {
        // Grow the stack: maps can be arbitrarily deeply nested (e.g. jsonb).
        mz_ore::stack::maybe_grow(|| self.iter().cmp(other.iter()))
    }
}

impl<'a, T> PartialOrd for DatumMap<'a, T> {
    #[inline(always)]
    fn partial_cmp(&self, other: &DatumMap<'a, T>) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl<'a> crate::scalar::SqlContainerType for DatumList<'a, Datum<'a>> {
    fn unwrap_element_type(container: &SqlScalarType) -> &SqlScalarType {
        container.unwrap_list_element_type()
    }
    fn wrap_element_type(element: SqlScalarType) -> SqlScalarType {
        SqlScalarType::List {
            element_type: Box::new(element),
            custom_id: None,
        }
    }
}

impl<'a> crate::scalar::SqlContainerType for DatumMap<'a, Datum<'a>> {
    fn unwrap_element_type(container: &SqlScalarType) -> &SqlScalarType {
        container.unwrap_map_value_type()
    }
    fn wrap_element_type(element: SqlScalarType) -> SqlScalarType {
        SqlScalarType::Map {
            value_type: Box::new(element),
            custom_id: None,
        }
    }
}

/// Represents a single `Datum`, appropriate to be nested inside other
/// `Datum`s.
#[derive(Clone, Copy, Eq, PartialEq, Hash)]
pub struct DatumNested<'a> {
    val: &'a [u8],
}

impl<'a> std::fmt::Display for DatumNested<'a> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        std::fmt::Display::fmt(&self.datum(), f)
    }
}

impl<'a> std::fmt::Debug for DatumNested<'a> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("DatumNested")
            .field("val", &self.datum())
            .finish()
    }
}

impl<'a> DatumNested<'a> {
    // Figure out which bytes `read_datum` returns (e.g. including the tag),
    // and then store a reference to those bytes, so we can "replay" this same
    // call later on without storing the datum itself.
    pub fn extract(data: &mut &'a [u8]) -> DatumNested<'a> {
        let prev = *data;
        let _ = unsafe { read_datum(data) };
        DatumNested {
            val: &prev[..(prev.len() - data.len())],
        }
    }

    /// Returns the datum `self` contains.
    pub fn datum(&self) -> Datum<'a> {
        let mut temp = self.val;
        unsafe { read_datum(&mut temp) }
    }
}

impl<'a> Ord for DatumNested<'a> {
    fn cmp(&self, other: &Self) -> Ordering {
        // Grow the stack: this recurses once per level of nested list/map values.
        mz_ore::stack::maybe_grow(|| self.datum().cmp(&other.datum()))
    }
}

impl<'a> PartialOrd for DatumNested<'a> {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

// Prefer adding new tags to the end of the enum. Certain behavior, like row ordering and EXPLAIN
// PHYSICAL PLAN, rely on the ordering of this enum. Neither of these are breaking changes, but
// it's annoying when they change.
#[derive(Debug, Clone, Copy, PartialEq, Eq, IntoPrimitive, TryFromPrimitive)]
#[repr(u8)]
enum Tag {
    Null,
    False,
    True,
    Float32,
    Float64,
    Date,
    Time,
    Timestamp,
    TimestampTz,
    Interval,
    // The length-prefixed tags: for each of the three kinds, four tags ordered by the width of
    // the length prefix in front of the payload, so a tag's distance from its kind's first tag
    // selects the width. Each kind has its own `read_datum` arm naming the `Datum` constructor,
    // since the kind never varies within a column; the width does, so it stays inside the arm.
    BytesTiny,
    BytesShort,
    BytesLong,
    BytesHuge,
    StringTiny,
    StringShort,
    StringLong,
    StringHuge,
    ListTiny,
    ListShort,
    ListLong,
    ListHuge,
    Uuid,
    Array,
    Dict,
    JsonNull,
    Dummy,
    Numeric,
    MzTimestamp,
    Range,
    MzAclItem,
    AclItem,
    // Everything except leap seconds and times beyond the range of
    // i64 nanoseconds. (Note that Materialize does not support leap
    // seconds, but this module does).
    CheapTimestamp,
    // Everything except leap seconds and times beyond the range of
    // i64 nanoseconds. (Note that Materialize does not support leap
    // seconds, but this module does).
    CheapTimestampTz,
    // The next several tags are for variable-length signed integer encoding.
    // The basic idea is that `NonNegativeIntN_K` is used to encode a datum of type
    // IntN whose actual value is positive or zero and fits in K bits, and similarly for
    // NegativeIntN_K with negative values.
    //
    // The order of these tags matters, because we want to be able to choose the
    // tag for a given datum quickly, with arithmetic, rather than slowly, with a
    // stack of `if` statements.
    //
    // Separate tags for non-negative and negative numbers are used to avoid having to
    // waste one bit in the actual data space to encode the sign.
    // A signed family alternates non-negative and negative at each payload width, so a tag
    // splits into both by arithmetic: the width is its distance from the family's first tag
    // shifted right by one, and the sign is that distance's low bit. Keeping the two signs
    // apart would make the width a subtraction from one of two bases, chosen by a compare.
    //
    // The family ends at the width where the payload is the whole integer. There the sign is
    // already the payload's top bit, so one tag serves both and the alternation stops, which is
    // why the fixed-width tag sits there rather than in a family of its own.
    NonNegativeInt16_0, // i.e., 0
    NegativeInt16_0,    // i.e., -1
    NonNegativeInt16_8,
    NegativeInt16_8,
    Int16,

    NonNegativeInt32_0,
    NegativeInt32_0,
    NonNegativeInt32_8,
    NegativeInt32_8,
    NonNegativeInt32_16,
    NegativeInt32_16,
    NonNegativeInt32_24,
    NegativeInt32_24,
    Int32,

    NonNegativeInt64_0,
    NegativeInt64_0,
    NonNegativeInt64_8,
    NegativeInt64_8,
    NonNegativeInt64_16,
    NegativeInt64_16,
    NonNegativeInt64_24,
    NegativeInt64_24,
    NonNegativeInt64_32,
    NegativeInt64_32,
    NonNegativeInt64_40,
    NegativeInt64_40,
    NonNegativeInt64_48,
    NegativeInt64_48,
    NonNegativeInt64_56,
    NegativeInt64_56,
    Int64,

    // These are like the ones above, but for unsigned types. The situation is slightly simpler
    // as we don't have negatives, so the width is the whole distance from the family's first
    // tag, and the fixed-width tag is again the widest member.
    UInt8_0, // i.e., 0
    UInt8,

    UInt16_0,
    UInt16_8,
    UInt16,

    UInt32_0,
    UInt32_8,
    UInt32_16,
    UInt32_24,
    UInt32,

    UInt64_0,
    UInt64_8,
    UInt64_16,
    UInt64_24,
    UInt64_32,
    UInt64_40,
    UInt64_48,
    UInt64_56,
    UInt64,
}

impl Tag {
    /// The tag's discriminant, usable in const context.
    #[allow(clippy::as_conversions)]
    const fn byte(self) -> u8 {
        self as u8
    }
}

/// Assert that the listed tags are consecutive, in the order given.
///
/// Every family below is addressed by arithmetic rather than by name: `push_datum` writes
/// `first + n` for a payload of `n` bytes, and `read_signed_varint`/`read_unsigned_varint`
/// invert that by subtraction. A variant inserted into the middle of a family, or two members
/// swapped, silently changes how many bytes a tag claims, which corrupts the datum rather than
/// failing to compile. These assertions turn that into a compile error at the definition.
macro_rules! assert_consecutive {
    ($($tag:ident),+ $(,)?) => {
        const _: () = {
            let tags: &[u8] = &[$(Tag::$tag.byte()),+];
            let mut i = 1;
            while i < tags.len() {
                assert!(
                    tags[i] == tags[i - 1] + 1,
                    concat!("tags are not consecutive: ", stringify!($($tag),+))
                );
                i += 1;
            }
        };
    };
}

assert_consecutive!(BytesTiny, BytesShort, BytesLong, BytesHuge);
assert_consecutive!(StringTiny, StringShort, StringLong, StringHuge);
assert_consecutive!(ListTiny, ListShort, ListLong, ListHuge);
assert_consecutive!(
    NonNegativeInt16_0,
    NegativeInt16_0,
    NonNegativeInt16_8,
    NegativeInt16_8,
    Int16,
);
assert_consecutive!(
    NonNegativeInt32_0,
    NegativeInt32_0,
    NonNegativeInt32_8,
    NegativeInt32_8,
    NonNegativeInt32_16,
    NegativeInt32_16,
    NonNegativeInt32_24,
    NegativeInt32_24,
    Int32,
);
assert_consecutive!(
    NonNegativeInt64_0,
    NegativeInt64_0,
    NonNegativeInt64_8,
    NegativeInt64_8,
    NonNegativeInt64_16,
    NegativeInt64_16,
    NonNegativeInt64_24,
    NegativeInt64_24,
    NonNegativeInt64_32,
    NegativeInt64_32,
    NonNegativeInt64_40,
    NegativeInt64_40,
    NonNegativeInt64_48,
    NegativeInt64_48,
    NonNegativeInt64_56,
    NegativeInt64_56,
    Int64,
);
assert_consecutive!(UInt8_0, UInt8,);
assert_consecutive!(UInt16_0, UInt16_8, UInt16,);
assert_consecutive!(UInt32_0, UInt32_8, UInt32_16, UInt32_24, UInt32,);
assert_consecutive!(
    UInt64_0, UInt64_8, UInt64_16, UInt64_24, UInt64_32, UInt64_40, UInt64_48, UInt64_56, UInt64,
);

// --------------------------------------------------------------------------------
// reading data

/// Read a byte slice starting at byte `offset`.
///
/// Updates `offset` to point to the first byte after the end of the read region.
fn read_untagged_bytes<'a>(data: &mut &'a [u8]) -> &'a [u8] {
    let len = u64::from_le_bytes(read_byte_array(data));
    let len = usize::cast_from(len);
    let (bytes, next) = data.split_at(len);
    *data = next;
    bytes
}

/// Read a byte slice preceded by its length, the width of the length prefix given by the tag's
/// distance from `first`, its kind's `*Tiny` tag.
///
/// Each arm reads the prefix at a constant width, which is a plain load; deriving the width and
/// reading that many bytes measured twice as slow on long payloads.
///
/// # Safety
///
/// The contents are whatever `push_lengthed_bytes` wrote, so a caller may treat them as UTF-8
/// only for a `String` tag.
#[inline(always)]
fn read_lengthed_bytes<'a>(data: &mut &'a [u8], tag: Tag, first: Tag) -> &'a [u8] {
    let len = match u8::from(tag).wrapping_sub(u8::from(first)) {
        0 => usize::from(read_byte(data)),
        1 => usize::from(u16::from_le_bytes(read_byte_array(data))),
        2 => usize::cast_from(u32::from_le_bytes(read_byte_array(data))),
        _ => usize::cast_from(u64::from_le_bytes(read_byte_array(data))),
    };
    let (bytes, next) = data.split_at(len);
    *data = next;
    bytes
}

#[inline(always)]
fn read_byte(data: &mut &[u8]) -> u8 {
    let byte = data[0];
    *data = &data[1..];
    byte
}

/// The payload of a variable-length integer whose datum ends within eight bytes of the end of
/// `data`, so the wide load in [`read_varint_word`] would run off the end.
///
/// Out of line and cold: this is at most the tail of a row, or of a nested list or map.
#[cold]
#[inline(never)]
fn read_varint_word_tail(data: &[u8], len: usize) -> u64 {
    #[inline(always)]
    fn ext<const L: usize>(data: &[u8]) -> u64 {
        let mut raw = [0; 8];
        raw[..L].copy_from_slice(&data[..L]);
        u64::from_le_bytes(raw)
    }
    // Each arm reads a constant width, which is a load rather than a `memcpy`. This path serves
    // the last datum or two of every row, which at low arity is a real share of all datums.
    match len {
        0 => 0,
        1 => u64::from(data[0]),
        2 => ext::<2>(data),
        3 => ext::<3>(data),
        4 => ext::<4>(data),
        5 => ext::<5>(data),
        6 => ext::<6>(data),
        7 => ext::<7>(data),
        // An eight-byte payload needs eight bytes past the tag, which this path's caller found
        // wanting, so a valid row cannot reach here.
        _ => panic!("payload runs past the end of the row"),
    }
}

/// Read the `len` payload bytes of a variable-length integer, returning them in the low
/// `len * 8` bits of a word. Bits above that are zero or belong to whatever follows in the row,
/// so a caller must mask them off.
///
/// Loading a fixed eight bytes is what keeps `len` out of the load, and so off a branch. Those
/// bytes exist except at the very end of the buffer.
#[inline(always)]
fn read_varint_word(data: &mut &[u8], len: usize) -> u64 {
    let word = match data.first_chunk::<8>() {
        Some(chunk) => u64::from_le_bytes(*chunk),
        None => read_varint_word_tail(data, len),
    };
    *data = &data[len..];
    word
}

/// Mask covering the low `len` bytes of a word.
///
/// `len` reaches eight for a 64-bit value that needs every byte, where a shift of 64 would be
/// undefined, so that case saturates instead.
#[inline(always)]
fn payload_mask(len: usize) -> u64 {
    if len >= 8 {
        u64::MAX
    } else {
        (1u64 << (len * 8)) - 1
    }
}

/// The low `N` bytes of `word`, little-endian.
#[inline(always)]
fn truncate<const N: usize>(word: u64) -> [u8; N] {
    word.to_le_bytes()[..N].try_into().expect("N <= 8")
}

/// Read the payload of a variable-length integer of either sign, extended to `N` bytes.
///
/// A signed family alternates the two signs at each payload width, so the tag's distance from
/// `first` holds the width above its low bit and the sign in it. Both fall out by shifting and
/// masking, with no compare and nothing to dispatch on: a column's tag varies with the magnitude
/// and sign of every value, so any branch on it is one the predictor cannot learn.
///
/// # Correctness
///
/// `tag` must belong to the family starting at `first`, and `data` must hold its payload.
#[inline(always)]
fn read_signed_varint<const N: usize>(data: &mut &[u8], tag: Tag, first: Tag) -> [u8; N] {
    let delta = u8::from(tag).wrapping_sub(u8::from(first));
    let len = usize::from(delta >> 1);
    // All ones for a negative value, so the bytes above the payload sign-extend, and zero
    // otherwise. A `len` of zero leaves the whole word filled, which is the -1 the encoder means.
    let negative = delta & 1 == 1;
    read_varint_payload(data, len, negative)
}

/// Read a `len` byte payload and extend it to `N` bytes, with ones above it when `negative`.
///
/// `len` must not exceed `N`, since `truncate` keeps only the low `N` bytes and a wider payload
/// would lose its top ones.
#[inline(always)]
fn read_varint_payload<const N: usize>(data: &mut &[u8], len: usize, negative: bool) -> [u8; N] {
    let mask = payload_mask(len);
    let fill = 0u64.wrapping_sub(u64::from(negative));
    truncate((read_varint_word(data, len) & mask) | (fill & !mask))
}

/// Read the payload of an unsigned variable-length integer, zero-extended to `N` bytes.
///
/// As [`read_signed_varint`], without the sign.
///
/// # Correctness
///
/// `tag` must belong to the family starting at `first`, and `data` must hold its payload.
#[inline(always)]
fn read_unsigned_varint<const N: usize>(data: &mut &[u8], tag: Tag, first: Tag) -> [u8; N] {
    let len = usize::from(u8::from(tag).wrapping_sub(u8::from(first)));
    read_varint_payload(data, len, false)
}

#[inline(always)]
pub(super) fn read_byte_array<const N: usize>(data: &mut &[u8]) -> [u8; N] {
    let (prev, next) = data.split_first_chunk().unwrap();
    *data = next;
    *prev
}

pub(super) fn read_date(data: &mut &[u8]) -> Date {
    let days = i32::from_le_bytes(read_byte_array(data));
    Date::from_pg_epoch(days).expect("unexpected date")
}

pub(super) fn read_naive_date(data: &mut &[u8]) -> NaiveDate {
    let year = i32::from_le_bytes(read_byte_array(data));
    let ordinal = u32::from_le_bytes(read_byte_array(data));
    NaiveDate::from_yo_opt(year, ordinal).unwrap()
}

pub(super) fn read_time(data: &mut &[u8]) -> NaiveTime {
    let secs = u32::from_le_bytes(read_byte_array(data));
    let nanos = u32::from_le_bytes(read_byte_array(data));
    NaiveTime::from_num_seconds_from_midnight_opt(secs, nanos).unwrap()
}

/// Read a datum starting at byte `offset`.
///
/// Updates `offset` to point to the first byte after the end of the read region.
///
/// # Safety
///
/// This function is safe if a `Datum` was previously written at this offset by `push_datum`.
/// Otherwise it could return invalid values, which is Undefined Behavior.
pub unsafe fn read_datum<'a>(data: &mut &'a [u8]) -> Datum<'a> {
    let tag = Tag::try_from_primitive(read_byte(data)).expect("unknown row tag");
    match tag {
        Tag::Null => Datum::Null,
        Tag::False => Datum::False,
        Tag::True => Datum::True,
        Tag::NonNegativeInt16_0
        | Tag::NegativeInt16_0
        | Tag::NonNegativeInt16_8
        | Tag::NegativeInt16_8
        | Tag::Int16 => Datum::Int16(i16::from_le_bytes(read_signed_varint(
            data,
            tag,
            Tag::NonNegativeInt16_0,
        ))),
        Tag::NonNegativeInt32_0
        | Tag::NegativeInt32_0
        | Tag::NonNegativeInt32_8
        | Tag::NegativeInt32_8
        | Tag::NonNegativeInt32_16
        | Tag::NegativeInt32_16
        | Tag::NonNegativeInt32_24
        | Tag::NegativeInt32_24
        | Tag::Int32 => Datum::Int32(i32::from_le_bytes(read_signed_varint(
            data,
            tag,
            Tag::NonNegativeInt32_0,
        ))),
        Tag::NonNegativeInt64_0
        | Tag::NegativeInt64_0
        | Tag::NonNegativeInt64_8
        | Tag::NegativeInt64_8
        | Tag::NonNegativeInt64_16
        | Tag::NegativeInt64_16
        | Tag::NonNegativeInt64_24
        | Tag::NegativeInt64_24
        | Tag::NonNegativeInt64_32
        | Tag::NegativeInt64_32
        | Tag::NonNegativeInt64_40
        | Tag::NegativeInt64_40
        | Tag::NonNegativeInt64_48
        | Tag::NegativeInt64_48
        | Tag::NonNegativeInt64_56
        | Tag::NegativeInt64_56
        | Tag::Int64 => Datum::Int64(i64::from_le_bytes(read_signed_varint(
            data,
            tag,
            Tag::NonNegativeInt64_0,
        ))),
        Tag::UInt8_0 | Tag::UInt8 => Datum::UInt8(u8::from_le_bytes(read_unsigned_varint(
            data,
            tag,
            Tag::UInt8_0,
        ))),
        Tag::UInt16_0 | Tag::UInt16_8 | Tag::UInt16 => Datum::UInt16(u16::from_le_bytes(
            read_unsigned_varint(data, tag, Tag::UInt16_0),
        )),
        Tag::UInt32_0 | Tag::UInt32_8 | Tag::UInt32_16 | Tag::UInt32_24 | Tag::UInt32 => {
            Datum::UInt32(u32::from_le_bytes(read_unsigned_varint(
                data,
                tag,
                Tag::UInt32_0,
            )))
        }
        Tag::UInt64_0
        | Tag::UInt64_8
        | Tag::UInt64_16
        | Tag::UInt64_24
        | Tag::UInt64_32
        | Tag::UInt64_40
        | Tag::UInt64_48
        | Tag::UInt64_56
        | Tag::UInt64 => Datum::UInt64(u64::from_le_bytes(read_unsigned_varint(
            data,
            tag,
            Tag::UInt64_0,
        ))),

        Tag::Float32 => {
            let f = f32::from_bits(u32::from_le_bytes(read_byte_array(data)));
            Datum::Float32(OrderedFloat::from(f))
        }
        Tag::Float64 => {
            let f = f64::from_bits(u64::from_le_bytes(read_byte_array(data)));
            Datum::Float64(OrderedFloat::from(f))
        }
        Tag::Date => Datum::Date(read_date(data)),
        Tag::Time => Datum::Time(read_time(data)),
        Tag::CheapTimestamp => {
            let ts = i64::from_le_bytes(read_byte_array(data));
            let secs = ts.div_euclid(1_000_000_000);
            let nsecs: u32 = ts.rem_euclid(1_000_000_000).try_into().unwrap();
            let ndt = DateTime::from_timestamp(secs, nsecs)
                .expect("We only write round-trippable timestamps")
                .naive_utc();
            Datum::Timestamp(
                CheckedTimestamp::from_timestamplike(ndt).expect("unexpected timestamp"),
            )
        }
        Tag::CheapTimestampTz => {
            let ts = i64::from_le_bytes(read_byte_array(data));
            let secs = ts.div_euclid(1_000_000_000);
            let nsecs: u32 = ts.rem_euclid(1_000_000_000).try_into().unwrap();
            let dt = DateTime::from_timestamp(secs, nsecs)
                .expect("We only write round-trippable timestamps");
            Datum::TimestampTz(
                CheckedTimestamp::from_timestamplike(dt).expect("unexpected timestamp"),
            )
        }
        Tag::Timestamp => {
            let date = read_naive_date(data);
            let time = read_time(data);
            Datum::Timestamp(
                CheckedTimestamp::from_timestamplike(date.and_time(time))
                    .expect("unexpected timestamp"),
            )
        }
        Tag::TimestampTz => {
            let date = read_naive_date(data);
            let time = read_time(data);
            Datum::TimestampTz(
                CheckedTimestamp::from_timestamplike(DateTime::from_naive_utc_and_offset(
                    date.and_time(time),
                    Utc,
                ))
                .expect("unexpected timestamptz"),
            )
        }
        Tag::Interval => {
            let months = i32::from_le_bytes(read_byte_array(data));
            let days = i32::from_le_bytes(read_byte_array(data));
            let micros = i64::from_le_bytes(read_byte_array(data));
            Datum::Interval(Interval {
                months,
                days,
                micros,
            })
        }
        Tag::BytesTiny | Tag::BytesShort | Tag::BytesLong | Tag::BytesHuge => {
            Datum::Bytes(read_lengthed_bytes(data, tag, Tag::BytesTiny))
        }
        Tag::StringTiny | Tag::StringShort | Tag::StringLong | Tag::StringHuge => {
            // SAFETY: the bytes were written from a `str` under a `String` tag.
            Datum::String(str::from_utf8_unchecked(read_lengthed_bytes(
                data,
                tag,
                Tag::StringTiny,
            )))
        }
        Tag::ListTiny | Tag::ListShort | Tag::ListLong | Tag::ListHuge => Datum::List(
            DatumList::new(read_lengthed_bytes(data, tag, Tag::ListTiny)),
        ),
        Tag::Uuid => Datum::Uuid(Uuid::from_bytes(read_byte_array(data))),
        Tag::Array => {
            // See the comment in `Row::push_array` for details on the encoding
            // of arrays.
            let ndims = read_byte(data);
            let dims_size = usize::from(ndims) * size_of::<u64>() * 2;
            let (dims, next) = data.split_at(dims_size);
            *data = next;
            let bytes = read_untagged_bytes(data);
            Datum::Array(Array {
                dims: ArrayDimensions { data: dims },
                elements: DatumList::new(bytes),
            })
        }
        Tag::Dict => {
            let bytes = read_untagged_bytes(data);
            Datum::Map(DatumMap::new(bytes))
        }
        Tag::JsonNull => Datum::JsonNull,
        Tag::Dummy => Datum::Dummy,
        Tag::Numeric => {
            let digits = read_byte(data).into();
            let exponent = i8::reinterpret_cast(read_byte(data));
            let bits = read_byte(data);

            let lsu_u16_len = Numeric::digits_to_lsu_elements_len(digits);
            let lsu_u8_len = lsu_u16_len * 2;
            let (lsu_u8, next) = data.split_at(lsu_u8_len);
            *data = next;

            // TODO: if we refactor the decimal library to accept the owned
            // array as a parameter to `from_raw_parts` below, we could likely
            // avoid a copy because it is exactly the value we want
            let mut lsu = [0; numeric::NUMERIC_DATUM_WIDTH_USIZE];
            for (i, c) in lsu_u8.chunks(2).enumerate() {
                lsu[i] = u16::from_le_bytes(c.try_into().unwrap());
            }

            let d = Numeric::from_raw_parts(digits, exponent.into(), bits, lsu);
            Datum::from(d)
        }
        Tag::MzTimestamp => {
            let t = Timestamp::decode(read_byte_array(data));
            Datum::MzTimestamp(t)
        }
        Tag::Range => {
            // See notes on `push_range_with` for details about encoding.
            let flag_byte = read_byte(data);
            let flags = range::InternalFlags::from_bits(flag_byte)
                .expect("range flags must be encoded validly");

            if flags.contains(range::InternalFlags::EMPTY) {
                assert!(
                    flags == range::InternalFlags::EMPTY,
                    "empty ranges contain only RANGE_EMPTY flag"
                );

                return Datum::Range(Range { inner: None });
            }

            let lower_bound = if flags.contains(range::InternalFlags::LB_INFINITE) {
                None
            } else {
                Some(DatumNested::extract(data))
            };

            let lower = RangeBound {
                inclusive: flags.contains(range::InternalFlags::LB_INCLUSIVE),
                bound: lower_bound,
            };

            let upper_bound = if flags.contains(range::InternalFlags::UB_INFINITE) {
                None
            } else {
                Some(DatumNested::extract(data))
            };

            let upper = RangeBound {
                inclusive: flags.contains(range::InternalFlags::UB_INCLUSIVE),
                bound: upper_bound,
            };

            Datum::Range(Range {
                inner: Some(RangeInner { lower, upper }),
            })
        }
        Tag::MzAclItem => {
            const N: usize = MzAclItem::binary_size();
            let mz_acl_item =
                MzAclItem::decode_binary(&read_byte_array::<N>(data)).expect("invalid mz_aclitem");
            Datum::MzAclItem(mz_acl_item)
        }
        Tag::AclItem => {
            const N: usize = AclItem::binary_size();
            let acl_item =
                AclItem::decode_binary(&read_byte_array::<N>(data)).expect("invalid aclitem");
            Datum::AclItem(acl_item)
        }
    }
}

// --------------------------------------------------------------------------------
// writing data

fn push_untagged_bytes<D>(data: &mut D, bytes: &[u8])
where
    D: Vector<u8>,
{
    let len = u64::cast_from(bytes.len());
    data.extend_from_slice(&len.to_le_bytes());
    data.extend_from_slice(bytes);
}

fn push_lengthed_bytes<D>(data: &mut D, bytes: &[u8], tag: Tag)
where
    D: Vector<u8>,
{
    match tag {
        Tag::BytesTiny | Tag::StringTiny | Tag::ListTiny => {
            let len = bytes.len().to_le_bytes();
            data.push(len[0]);
        }
        Tag::BytesShort | Tag::StringShort | Tag::ListShort => {
            let len = bytes.len().to_le_bytes();
            data.extend_from_slice(&len[0..2]);
        }
        Tag::BytesLong | Tag::StringLong | Tag::ListLong => {
            let len = bytes.len().to_le_bytes();
            data.extend_from_slice(&len[0..4]);
        }
        Tag::BytesHuge | Tag::StringHuge | Tag::ListHuge => {
            let len = bytes.len().to_le_bytes();
            data.extend_from_slice(&len);
        }
        _ => unreachable!(),
    }
    data.extend_from_slice(bytes);
}

pub(super) fn date_to_array(date: Date) -> [u8; size_of::<i32>()] {
    i32::to_le_bytes(date.pg_epoch_days())
}

fn push_date<D>(data: &mut D, date: Date)
where
    D: Vector<u8>,
{
    data.extend_from_slice(&date_to_array(date));
}

pub(super) fn naive_date_to_arrays(
    date: NaiveDate,
) -> ([u8; size_of::<i32>()], [u8; size_of::<u32>()]) {
    (
        i32::to_le_bytes(date.year()),
        u32::to_le_bytes(date.ordinal()),
    )
}

fn push_naive_date<D>(data: &mut D, date: NaiveDate)
where
    D: Vector<u8>,
{
    let (ds1, ds2) = naive_date_to_arrays(date);
    data.extend_from_slice(&ds1);
    data.extend_from_slice(&ds2);
}

pub(super) fn time_to_arrays(time: NaiveTime) -> ([u8; size_of::<u32>()], [u8; size_of::<u32>()]) {
    (
        u32::to_le_bytes(time.num_seconds_from_midnight()),
        u32::to_le_bytes(time.nanosecond()),
    )
}

fn push_time<D>(data: &mut D, time: NaiveTime)
where
    D: Vector<u8>,
{
    let (ts1, ts2) = time_to_arrays(time);
    data.extend_from_slice(&ts1);
    data.extend_from_slice(&ts2);
}

/// Returns an i64 representing a `NaiveDateTime`, if
/// said i64 can be round-tripped back to a `NaiveDateTime`.
///
/// The only exotic NDTs for which this can't happen are those that
/// are hundreds of years in the future or past, or those that
/// represent a leap second. (Note that Materialize does not support
/// leap seconds, but this module does).
// This function is inspired by `NaiveDateTime::timestamp_nanos`,
// with extra checking.
fn checked_timestamp_nanos(dt: NaiveDateTime) -> Option<i64> {
    let subsec_nanos = dt.and_utc().timestamp_subsec_nanos();
    if subsec_nanos >= 1_000_000_000 {
        return None;
    }
    let as_ns = dt.and_utc().timestamp().checked_mul(1_000_000_000)?;
    as_ns.checked_add(i64::from(subsec_nanos))
}

// This function is extremely hot, so
// we just use `as` to avoid the overhead of
// `try_into` followed by `unwrap`.
// `leading_ones` and `leading_zeros`
// can never return values greater than 64, so the conversion is safe.
#[inline(always)]
#[allow(clippy::as_conversions)]
fn min_bytes_signed<T>(i: T) -> u8
where
    T: Into<i64>,
{
    let i: i64 = i.into();

    // To fit in n bytes, we require that
    // everything but the leading sign bits fits in n*8
    // bits.
    let n_sign_bits = if i.is_negative() {
        i.leading_ones() as u8
    } else {
        i.leading_zeros() as u8
    };

    (64 - n_sign_bits + 7) / 8
}

// In principle we could just use `min_bytes_signed`, rather than
// having a separate function here, as long as we made that one take
// `T: Into<i128>` instead of 64. But LLVM doesn't seem smart enough
// to realize that that function is the same as the current version,
// and generates worse code.
//
// Justification for `as` is the same as in `min_bytes_signed`.
#[inline(always)]
#[allow(clippy::as_conversions)]
fn min_bytes_unsigned<T>(i: T) -> u8
where
    T: Into<u64>,
{
    let i: u64 = i.into();

    let n_sign_bits = i.leading_zeros() as u8;

    (64 - n_sign_bits + 7) / 8
}

const TINY: usize = 1 << 8;
const SHORT: usize = 1 << 16;
const LONG: usize = 1 << 32;

fn push_datum<D>(data: &mut D, datum: Datum)
where
    D: Vector<u8>,
{
    match datum {
        Datum::Null => data.push(Tag::Null.into()),
        Datum::False => data.push(Tag::False.into()),
        Datum::True => data.push(Tag::True.into()),
        Datum::Int16(i) => {
            // The family alternates the signs at each width, so the width is two tags apart and
            // the sign is the low bit. The clamp folds both signs onto the widest tag, where the
            // payload carries its own sign. See `read_signed_varint`, which takes this apart.
            const WIDEST_DELTA: u8 = Tag::Int16.byte() - Tag::NonNegativeInt16_0.byte();
            let mbs = min_bytes_signed(i);
            let delta = ((mbs << 1) + u8::from(i.is_negative())).min(WIDEST_DELTA);
            let tag = u8::from(Tag::NonNegativeInt16_0) + delta;

            data.push(tag);
            data.extend_from_slice(&i.to_le_bytes()[0..usize::from(mbs)]);
        }
        Datum::Int32(i) => {
            // The family alternates the signs at each width, so the width is two tags apart and
            // the sign is the low bit. The clamp folds both signs onto the widest tag, where the
            // payload carries its own sign. See `read_signed_varint`, which takes this apart.
            const WIDEST_DELTA: u8 = Tag::Int32.byte() - Tag::NonNegativeInt32_0.byte();
            let mbs = min_bytes_signed(i);
            let delta = ((mbs << 1) + u8::from(i.is_negative())).min(WIDEST_DELTA);
            let tag = u8::from(Tag::NonNegativeInt32_0) + delta;

            data.push(tag);
            data.extend_from_slice(&i.to_le_bytes()[0..usize::from(mbs)]);
        }
        Datum::Int64(i) => {
            // The family alternates the signs at each width, so the width is two tags apart and
            // the sign is the low bit. The clamp folds both signs onto the widest tag, where the
            // payload carries its own sign. See `read_signed_varint`, which takes this apart.
            const WIDEST_DELTA: u8 = Tag::Int64.byte() - Tag::NonNegativeInt64_0.byte();
            let mbs = min_bytes_signed(i);
            let delta = ((mbs << 1) + u8::from(i.is_negative())).min(WIDEST_DELTA);
            let tag = u8::from(Tag::NonNegativeInt64_0) + delta;

            data.push(tag);
            data.extend_from_slice(&i.to_le_bytes()[0..usize::from(mbs)]);
        }
        Datum::UInt8(i) => {
            let mbu = min_bytes_unsigned(i);
            let tag = u8::from(Tag::UInt8_0) + mbu;
            data.push(tag);
            data.extend_from_slice(&i.to_le_bytes()[0..usize::from(mbu)]);
        }
        Datum::UInt16(i) => {
            let mbu = min_bytes_unsigned(i);
            let tag = u8::from(Tag::UInt16_0) + mbu;
            data.push(tag);
            data.extend_from_slice(&i.to_le_bytes()[0..usize::from(mbu)]);
        }
        Datum::UInt32(i) => {
            let mbu = min_bytes_unsigned(i);
            let tag = u8::from(Tag::UInt32_0) + mbu;
            data.push(tag);
            data.extend_from_slice(&i.to_le_bytes()[0..usize::from(mbu)]);
        }
        Datum::UInt64(i) => {
            let mbu = min_bytes_unsigned(i);
            let tag = u8::from(Tag::UInt64_0) + mbu;
            data.push(tag);
            data.extend_from_slice(&i.to_le_bytes()[0..usize::from(mbu)]);
        }
        Datum::Float32(f) => {
            data.push(Tag::Float32.into());
            data.extend_from_slice(&f.to_bits().to_le_bytes());
        }
        Datum::Float64(f) => {
            data.push(Tag::Float64.into());
            data.extend_from_slice(&f.to_bits().to_le_bytes());
        }
        Datum::Date(d) => {
            data.push(Tag::Date.into());
            push_date(data, d);
        }
        Datum::Time(t) => {
            data.push(Tag::Time.into());
            push_time(data, t);
        }
        Datum::Timestamp(t) => {
            let datetime = t.to_naive();
            if let Some(nanos) = checked_timestamp_nanos(datetime) {
                data.push(Tag::CheapTimestamp.into());
                data.extend_from_slice(&nanos.to_le_bytes());
            } else {
                data.push(Tag::Timestamp.into());
                push_naive_date(data, datetime.date());
                push_time(data, datetime.time());
            }
        }
        Datum::TimestampTz(t) => {
            let datetime = t.to_naive();
            if let Some(nanos) = checked_timestamp_nanos(datetime) {
                data.push(Tag::CheapTimestampTz.into());
                data.extend_from_slice(&nanos.to_le_bytes());
            } else {
                data.push(Tag::TimestampTz.into());
                push_naive_date(data, datetime.date());
                push_time(data, datetime.time());
            }
        }
        Datum::Interval(i) => {
            data.push(Tag::Interval.into());
            data.extend_from_slice(&i.months.to_le_bytes());
            data.extend_from_slice(&i.days.to_le_bytes());
            data.extend_from_slice(&i.micros.to_le_bytes());
        }
        Datum::Bytes(bytes) => {
            let tag = match bytes.len() {
                0..TINY => Tag::BytesTiny,
                TINY..SHORT => Tag::BytesShort,
                SHORT..LONG => Tag::BytesLong,
                _ => Tag::BytesHuge,
            };
            data.push(tag.into());
            push_lengthed_bytes(data, bytes, tag);
        }
        Datum::String(string) => {
            let tag = match string.len() {
                0..TINY => Tag::StringTiny,
                TINY..SHORT => Tag::StringShort,
                SHORT..LONG => Tag::StringLong,
                _ => Tag::StringHuge,
            };
            data.push(tag.into());
            push_lengthed_bytes(data, string.as_bytes(), tag);
        }
        Datum::List(list) => {
            let tag = match list.data.len() {
                0..TINY => Tag::ListTiny,
                TINY..SHORT => Tag::ListShort,
                SHORT..LONG => Tag::ListLong,
                _ => Tag::ListHuge,
            };
            data.push(tag.into());
            push_lengthed_bytes(data, list.data, tag);
        }
        Datum::Uuid(u) => {
            data.push(Tag::Uuid.into());
            data.extend_from_slice(u.as_bytes());
        }
        Datum::Array(array) => {
            // See the comment in `Row::push_array` for details on the encoding
            // of arrays.
            data.push(Tag::Array.into());
            data.push(array.dims.ndims());
            data.extend_from_slice(array.dims.data);
            push_untagged_bytes(data, array.elements.data);
        }
        Datum::Map(dict) => {
            data.push(Tag::Dict.into());
            push_untagged_bytes(data, dict.data);
        }
        Datum::JsonNull => data.push(Tag::JsonNull.into()),
        Datum::MzTimestamp(t) => {
            data.push(Tag::MzTimestamp.into());
            data.extend_from_slice(&t.encode());
        }
        Datum::Dummy => data.push(Tag::Dummy.into()),
        Datum::Numeric(mut n) => {
            // Pseudo-canonical representation of decimal values with
            // insignificant zeroes trimmed. This compresses the number further
            // than `Numeric::trim` by removing all zeroes, and not only those in
            // the fractional component.
            numeric::cx_datum().reduce(&mut n.0);
            let (digits, exponent, bits, lsu) = n.0.to_raw_parts();
            data.push(Tag::Numeric.into());
            data.push(u8::try_from(digits).expect("digits to fit within u8; should not exceed 39"));
            data.push(
                i8::try_from(exponent)
                    .expect("exponent to fit within i8; should not exceed +/- 39")
                    .to_le_bytes()[0],
            );
            data.push(bits);

            let lsu = &lsu[..Numeric::digits_to_lsu_elements_len(digits)];

            // Little endian machines can take the lsu directly from u16 to u8.
            if cfg!(target_endian = "little") {
                // SAFETY: `lsu` (returned by `coefficient_units()`) is a `&[u16]`, so
                // each element can safely be transmuted into two `u8`s.
                let (prefix, lsu_bytes, suffix) = unsafe { lsu.align_to::<u8>() };
                // The `u8` aligned version of the `lsu` should have twice as many
                // elements as we expect for the `u16` version.
                soft_assert_no_log!(
                    lsu_bytes.len() == Numeric::digits_to_lsu_elements_len(digits) * 2,
                    "u8 version of numeric LSU contained the wrong number of elements; expected {}, but got {}",
                    Numeric::digits_to_lsu_elements_len(digits) * 2,
                    lsu_bytes.len()
                );
                // There should be no unaligned elements in the prefix or suffix.
                soft_assert_no_log!(prefix.is_empty() && suffix.is_empty());
                data.extend_from_slice(lsu_bytes);
            } else {
                for u in lsu {
                    data.extend_from_slice(&u.to_le_bytes());
                }
            }
        }
        Datum::Range(range) => {
            // See notes on `push_range_with` for details about encoding.
            data.push(Tag::Range.into());
            data.push(range.internal_flag_bits());

            if let Some(RangeInner { lower, upper }) = range.inner {
                for bound in [lower.bound, upper.bound] {
                    if let Some(bound) = bound {
                        match bound.datum() {
                            Datum::Null => panic!("cannot push Datum::Null into range"),
                            d => push_datum::<D>(data, d),
                        }
                    }
                }
            }
        }
        Datum::MzAclItem(mz_acl_item) => {
            data.push(Tag::MzAclItem.into());
            data.extend_from_slice(&mz_acl_item.encode_binary());
        }
        Datum::AclItem(acl_item) => {
            data.push(Tag::AclItem.into());
            data.extend_from_slice(&acl_item.encode_binary());
        }
    }
}

/// Return the number of bytes these Datums would use if packed as a Row.
pub fn row_size<'a, I>(a: I) -> usize
where
    I: IntoIterator<Item = Datum<'a>>,
{
    // Using datums_size instead of a.data().len() here is safer because it will
    // return the size of the datums if they were packed into a Row. Although
    // a.data().len() happens to give the correct answer (and is faster), data()
    // is documented as for debugging only.
    let sz = datums_size::<_, _>(a);
    let size_of_row = std::mem::size_of::<Row>();
    // The Row struct attempts to inline data until it can't fit in the
    // preallocated size. Otherwise it spills to heap, and uses the Row to point
    // to that.
    if sz > Row::SIZE {
        sz + size_of_row
    } else {
        size_of_row
    }
}

/// Number of bytes required by the datum.
/// This is used to optimistically pre-allocate buffers for packing rows.
pub fn datum_size(datum: &Datum) -> usize {
    match datum {
        Datum::Null => 1,
        Datum::False => 1,
        Datum::True => 1,
        Datum::Int16(i) => 1 + usize::from(min_bytes_signed(*i)),
        Datum::Int32(i) => 1 + usize::from(min_bytes_signed(*i)),
        Datum::Int64(i) => 1 + usize::from(min_bytes_signed(*i)),
        Datum::UInt8(i) => 1 + usize::from(min_bytes_unsigned(*i)),
        Datum::UInt16(i) => 1 + usize::from(min_bytes_unsigned(*i)),
        Datum::UInt32(i) => 1 + usize::from(min_bytes_unsigned(*i)),
        Datum::UInt64(i) => 1 + usize::from(min_bytes_unsigned(*i)),
        Datum::Float32(_) => 1 + size_of::<f32>(),
        Datum::Float64(_) => 1 + size_of::<f64>(),
        Datum::Date(_) => 1 + size_of::<i32>(),
        Datum::Time(_) => 1 + 8,
        Datum::Timestamp(t) => {
            1 + if checked_timestamp_nanos(t.to_naive()).is_some() {
                8
            } else {
                16
            }
        }
        Datum::TimestampTz(t) => {
            1 + if checked_timestamp_nanos(t.naive_utc()).is_some() {
                8
            } else {
                16
            }
        }
        Datum::Interval(_) => 1 + size_of::<i32>() + size_of::<i32>() + size_of::<i64>(),
        Datum::Bytes(bytes) => {
            // We use a variable length representation of slice length.
            let bytes_for_length = match bytes.len() {
                0..TINY => 1,
                TINY..SHORT => 2,
                SHORT..LONG => 4,
                _ => 8,
            };
            1 + bytes_for_length + bytes.len()
        }
        Datum::String(string) => {
            // We use a variable length representation of slice length.
            let bytes_for_length = match string.len() {
                0..TINY => 1,
                TINY..SHORT => 2,
                SHORT..LONG => 4,
                _ => 8,
            };
            1 + bytes_for_length + string.len()
        }
        Datum::Uuid(_) => 1 + size_of::<uuid::Bytes>(),
        Datum::Array(array) => {
            1 + size_of::<u8>()
                + array.dims.data.len()
                + size_of::<u64>()
                + array.elements.data.len()
        }
        Datum::List(list) => 1 + size_of::<u64>() + list.data.len(),
        Datum::Map(dict) => 1 + size_of::<u64>() + dict.data.len(),
        Datum::JsonNull => 1,
        Datum::MzTimestamp(_) => 1 + size_of::<Timestamp>(),
        Datum::Dummy => 1,
        Datum::Numeric(d) => {
            let mut d = d.0.clone();
            // Values must be reduced to determine appropriate number of
            // coefficient units.
            numeric::cx_datum().reduce(&mut d);
            // 4 = 1 bit each for tag, digits, exponent, bits
            4 + (d.coefficient_units().len() * 2)
        }
        Datum::Range(Range { inner }) => {
            // Tag + flags
            2 + match inner {
                None => 0,
                Some(RangeInner { lower, upper }) => [lower.bound, upper.bound]
                    .iter()
                    .map(|bound| match bound {
                        None => 0,
                        Some(bound) => bound.val.len(),
                    })
                    .sum(),
            }
        }
        Datum::MzAclItem(_) => 1 + MzAclItem::binary_size(),
        Datum::AclItem(_) => 1 + AclItem::binary_size(),
    }
}

/// Number of bytes required by a sequence of datums.
///
/// This method can be used to right-size the allocation for a `Row`
/// before calling [`RowPacker::extend`].
pub fn datums_size<'a, I, D>(iter: I) -> usize
where
    I: IntoIterator<Item = D>,
    D: Borrow<Datum<'a>>,
{
    iter.into_iter().map(|d| datum_size(d.borrow())).sum()
}

/// Number of bytes required by a list of datums. This computes the size that would be required if
/// the given datums were packed into a list.
///
/// This is used to optimistically pre-allocate buffers for packing rows.
pub fn datum_list_size<'a, I, D>(iter: I) -> usize
where
    I: IntoIterator<Item = D>,
    D: Borrow<Datum<'a>>,
{
    1 + size_of::<u64>() + datums_size(iter)
}

impl RowPacker<'_> {
    /// Constructs a row packer that will pack additional datums into the
    /// provided row.
    ///
    /// This function is intentionally somewhat inconvenient to call. You
    /// usually want to call [`Row::packer`] instead to start packing from
    /// scratch.
    pub fn for_existing_row(row: &mut Row) -> RowPacker<'_> {
        RowPacker { row }
    }

    /// Extend an existing `Row` with a `Datum`.
    #[inline]
    pub fn push<'a, D>(&mut self, datum: D)
    where
        D: Borrow<Datum<'a>>,
    {
        push_datum(&mut self.row.data, *datum.borrow());
    }

    /// Extend an existing `Row` with additional `Datum`s.
    #[inline]
    pub fn extend<'a, I, D>(&mut self, iter: I)
    where
        I: IntoIterator<Item = D>,
        D: Borrow<Datum<'a>>,
    {
        for datum in iter {
            push_datum(&mut self.row.data, *datum.borrow())
        }
    }

    /// Extend an existing `Row` with additional `Datum`s.
    ///
    /// In the case the iterator produces an error, the pushing of
    /// datums in terminated and the error returned. The `Row` will
    /// be incomplete, but it will be safe to read datums from it.
    #[inline]
    pub fn try_extend<'a, I, E, D>(&mut self, iter: I) -> Result<(), E>
    where
        I: IntoIterator<Item = Result<D, E>>,
        D: Borrow<Datum<'a>>,
    {
        for datum in iter {
            push_datum(&mut self.row.data, *datum?.borrow());
        }
        Ok(())
    }

    /// Appends the datums of an entire `Row`.
    pub fn extend_by_row(&mut self, row: &Row) {
        self.row.data.extend_from_slice(row.data.as_slice());
    }

    /// Appends the datums of an entire `Row`.
    pub fn extend_by_row_ref(&mut self, row: &RowRef) {
        self.row.data.extend_from_slice(row.data());
    }

    /// Appends the slice of data representing an entire `Row`. The data is not validated.
    ///
    /// # Safety
    ///
    /// The requirements from [`Row::from_bytes_unchecked`] apply here, too:
    /// This method relies on `data` being an appropriate row encoding, and can
    /// result in unsafety if this is not the case.
    #[inline]
    pub unsafe fn extend_by_slice_unchecked(&mut self, data: &[u8]) {
        self.row.data.extend_from_slice(data)
    }

    /// Pushes a [`DatumList`] that is built from a closure.
    ///
    /// The supplied closure will be invoked once with a `Row` that can be used
    /// to populate the list. It is valid to call any method on the
    /// [`RowPacker`] except for [`RowPacker::clear`], [`RowPacker::truncate`],
    /// or [`RowPacker::truncate_datums`].
    ///
    /// Returns the value returned by the closure, if any.
    ///
    /// ```
    /// # use mz_repr::{Row, Datum};
    /// let mut row = Row::default();
    /// row.packer().push_list_with(|row| {
    ///     row.push(Datum::String("age"));
    ///     row.push(Datum::Int64(42));
    /// });
    /// assert_eq!(
    ///     row.unpack_first().unwrap_list().iter().collect::<Vec<_>>(),
    ///     vec![Datum::String("age"), Datum::Int64(42)],
    /// );
    /// ```
    #[inline]
    pub fn push_list_with<F, R>(&mut self, f: F) -> R
    where
        F: FnOnce(&mut RowPacker) -> R,
    {
        // First, assume that the list will fit in 255 bytes, and thus the length will fit in
        // 1 byte. If not, we'll fix it up later.
        let start = self.row.data.len();
        self.row.data.push(Tag::ListTiny.into());
        // Write a dummy len, will fix it up later.
        self.row.data.push(0);

        let out = f(self);

        // The `- 1 - 1` is for the tag and the len.
        let len = self.row.data.len() - start - 1 - 1;
        // We now know the real len.
        if len < TINY {
            // If the len fits in 1 byte, we just need to fix up the len.
            self.row.data[start + 1] = len.to_le_bytes()[0];
        } else {
            // Note: We move this code path into its own function, so that the common case can be
            // inlined.
            long_list(&mut self.row.data, start, len);
        }

        /// 1. Fix up the tag.
        /// 2. Move the actual data a bit (for which we also need to make room at the end).
        /// 3. Fix up the len.
        /// `data`: The row's backing data.
        /// `start`: where `push_list_with` started writing in `data`.
        /// `len`: the length of the data, excluding the tag and the length.
        #[cold]
        fn long_list(data: &mut CompactBytes, start: usize, len: usize) {
            // `len_len`: the length of the length. (Possible values are: 2, 4, 8. 1 is handled
            // elsewhere.) The other parameters are the same as for `long_list`.
            let long_list_inner = |data: &mut CompactBytes, len_len| {
                // We'll need memory for the new, bigger length, so make the `CompactBytes` bigger.
                // The `- 1` is because the old length was 1 byte.
                const ZEROS: [u8; 8] = [0; 8];
                data.extend_from_slice(&ZEROS[0..len_len - 1]);
                // Move the data to the end of the `CompactBytes`, to make space for the new length.
                // Originally, it started after the 1-byte tag and the 1-byte length, now it will
                // start after the 1-byte tag and the len_len-byte length.
                //
                // Note that this is the only operation in `long_list` whose cost is proportional
                // to `len`. Since `len` is at least 256 here, the other operations' cost are
                // negligible. `copy_within` is a memmove, which is probably a fair bit faster per
                // Datum than a Datum encoding in the `f` closure.
                data.copy_within(start + 1 + 1..start + 1 + 1 + len, start + 1 + len_len);
                // Write the new length.
                data[start + 1..start + 1 + len_len]
                    .copy_from_slice(&len.to_le_bytes()[0..len_len]);
            };
            match len {
                0..TINY => {
                    unreachable!()
                }
                TINY..SHORT => {
                    data[start] = Tag::ListShort.into();
                    long_list_inner(data, 2);
                }
                SHORT..LONG => {
                    data[start] = Tag::ListLong.into();
                    long_list_inner(data, 4);
                }
                _ => {
                    data[start] = Tag::ListHuge.into();
                    long_list_inner(data, 8);
                }
            };
        }

        out
    }

    /// Pushes a [`DatumMap`] that is built from a closure.
    ///
    /// The supplied closure will be invoked once with a `Row` that can be used
    /// to populate the dict.
    ///
    /// The closure **must** alternate pushing string keys and arbitrary values,
    /// otherwise reading the dict will cause a panic.
    ///
    /// The closure **must** push keys in ascending order, otherwise equality
    /// checks on the resulting `Row` may be wrong and reading the dict IN DEBUG
    /// MODE will cause a panic.
    ///
    /// The closure **must not** call [`RowPacker::clear`],
    /// [`RowPacker::truncate`], or [`RowPacker::truncate_datums`].
    ///
    /// # Example
    ///
    /// ```
    /// # use mz_repr::{Row, Datum};
    /// let mut row = Row::default();
    /// row.packer().push_dict_with(|row| {
    ///
    ///     // key
    ///     row.push(Datum::String("age"));
    ///     // value
    ///     row.push(Datum::Int64(42));
    ///
    ///     // key
    ///     row.push(Datum::String("name"));
    ///     // value
    ///     row.push(Datum::String("bob"));
    /// });
    /// assert_eq!(
    ///     row.unpack_first().unwrap_map().iter().collect::<Vec<_>>(),
    ///     vec![("age", Datum::Int64(42)), ("name", Datum::String("bob"))]
    /// );
    /// ```
    pub fn push_dict_with<F, R>(&mut self, f: F) -> R
    where
        F: FnOnce(&mut RowPacker) -> R,
    {
        self.row.data.push(Tag::Dict.into());
        let start = self.row.data.len();
        // write a dummy len, will fix it up later
        self.row.data.extend_from_slice(&[0; size_of::<u64>()]);

        let res = f(self);

        let len = u64::cast_from(self.row.data.len() - start - size_of::<u64>());
        // fix up the len
        self.row.data[start..start + size_of::<u64>()].copy_from_slice(&len.to_le_bytes());

        res
    }

    /// Like [`RowPacker::push_dict_with`], but accepts a fallible closure.
    pub fn try_push_dict_with<F, E>(&mut self, f: F) -> Result<(), E>
    where
        F: FnOnce(&mut RowPacker) -> Result<(), E>,
    {
        self.push_dict_with(f)
    }

    /// Convenience function to construct an array from an iter of `Datum`s.
    ///
    /// Returns an error if the number of elements in `iter` does not match
    /// the cardinality of the array as described by `dims`, or if the
    /// number of dimensions exceeds [`MAX_ARRAY_DIMENSIONS`]. If an error
    /// occurs, the packer's state will be unchanged.
    pub fn try_push_array<'a, I, D>(
        &mut self,
        dims: &[ArrayDimension],
        iter: I,
    ) -> Result<(), InvalidArrayError>
    where
        I: IntoIterator<Item = D>,
        D: Borrow<Datum<'a>>,
    {
        // SAFETY: The function returns the exact number of elements pushed into the array.
        unsafe {
            self.push_array_with_unchecked(dims, |packer| {
                let mut nelements = 0;
                for datum in iter {
                    packer.push(datum);
                    nelements += 1;
                }
                Ok::<_, InvalidArrayError>(nelements)
            })
        }
    }

    /// Like [`RowPacker::try_push_array`], but accepts a fallible iterator of
    /// elements.
    pub fn try_push_array_fallible<'a, I, D, E>(
        &mut self,
        dims: &[ArrayDimension],
        iter: I,
    ) -> Result<Result<(), E>, InvalidArrayError>
    where
        I: IntoIterator<Item = Result<D, E>>,
        D: Borrow<Datum<'a>>,
    {
        enum Error<E> {
            Usage(InvalidArrayError),
            Inner(E),
        }

        impl<E> From<InvalidArrayError> for Error<E> {
            fn from(e: InvalidArrayError) -> Self {
                Self::Usage(e)
            }
        }

        // SAFETY: The function returns the exact number of elements pushed into the array.
        let result = unsafe {
            self.push_array_with_unchecked(dims, |packer| {
                let mut nelements = 0;
                for datum in iter {
                    packer.push(datum.map_err(Error::Inner)?);
                    nelements += 1;
                }
                Ok(nelements)
            })
        };
        match result {
            Ok(()) => Ok(Ok(())),
            Err(Error::Usage(e)) => Err(e),
            Err(Error::Inner(e)) => Ok(Err(e)),
        }
    }

    /// Convenience function to construct an array from a function. The function must return the
    /// number of elements it pushed into the array. It is undefined behavior if the function returns
    /// a number different to the number of elements it pushed.
    ///
    /// Returns an error if the number of elements pushed by `f` does not match
    /// the cardinality of the array as described by `dims`, or if the
    /// number of dimensions exceeds [`MAX_ARRAY_DIMENSIONS`], or if `f` errors. If an error
    /// occurs, the packer's state will be unchanged.
    pub unsafe fn push_array_with_unchecked<F, E>(
        &mut self,
        dims: &[ArrayDimension],
        f: F,
    ) -> Result<(), E>
    where
        F: FnOnce(&mut RowPacker) -> Result<usize, E>,
        E: From<InvalidArrayError>,
    {
        // Arrays are encoded as follows.
        //
        // u8    ndims
        // u64   dim_0 lower bound
        // u64   dim_0 length
        // ...
        // u64   dim_n lower bound
        // u64   dim_n length
        // u64   element data size in bytes
        // u8    element data, where elements are encoded in row-major order

        if dims.len() > usize::from(MAX_ARRAY_DIMENSIONS) {
            return Err(InvalidArrayError::TooManyDimensions(dims.len()).into());
        }

        let start = self.row.data.len();
        self.row.data.push(Tag::Array.into());

        // Write dimension information.
        self.row
            .data
            .push(dims.len().try_into().expect("ndims verified to fit in u8"));
        for dim in dims {
            self.row
                .data
                .extend_from_slice(&i64::cast_from(dim.lower_bound).to_le_bytes());
            self.row
                .data
                .extend_from_slice(&u64::cast_from(dim.length).to_le_bytes());
        }

        // Write elements.
        let off = self.row.data.len();
        self.row.data.extend_from_slice(&[0; size_of::<u64>()]);
        let nelements = match f(self) {
            Ok(nelements) => nelements,
            Err(e) => {
                self.row.data.truncate(start);
                return Err(e);
            }
        };
        let len = u64::cast_from(self.row.data.len() - off - size_of::<u64>());
        self.row.data[off..off + size_of::<u64>()].copy_from_slice(&len.to_le_bytes());

        // Check that the number of elements written matches the dimension
        // information.
        let cardinality = match dims {
            [] => 0,
            // Saturate the product: a cardinality that overflows `usize` is
            // impossibly large (no array can hold that many elements), so it can
            // never equal the actual `nelements` and the check below rejects it as
            // `WrongCardinality`. A plain `product()` would panic under overflow
            // checks (debug/fuzz) and silently wrap in release — and a wrapped
            // value could even spuriously match `nelements`, accepting a corrupt
            // array (e.g. dims claiming `[2^32, 2^32]` wrap to 0 elements).
            dims => dims
                .iter()
                .map(|d| d.length)
                .fold(1usize, usize::saturating_mul),
        };
        if nelements != cardinality {
            self.row.data.truncate(start);
            return Err(InvalidArrayError::WrongCardinality {
                actual: nelements,
                expected: cardinality,
            }
            .into());
        }

        Ok(())
    }

    /// Pushes an [`Array`] that is built from a closure.
    ///
    /// __WARNING__: This is fairly "sharp" tool that is easy to get wrong. You
    /// should prefer [`RowPacker::try_push_array`] when possible.
    ///
    /// Returns an error if the number of elements pushed does not match
    /// the cardinality of the array as described by `dims`, or if the
    /// number of dimensions exceeds [`MAX_ARRAY_DIMENSIONS`]. If an error
    /// occurs, the packer's state will be unchanged.
    pub fn push_array_with_row_major<F, I>(
        &mut self,
        dims: I,
        f: F,
    ) -> Result<(), InvalidArrayError>
    where
        I: IntoIterator<Item = ArrayDimension>,
        F: FnOnce(&mut RowPacker) -> usize,
    {
        let start = self.row.data.len();
        self.row.data.push(Tag::Array.into());

        // Write dummy dimension length for now, we'll fix it up.
        let dims_start = self.row.data.len();
        self.row.data.push(42);

        let mut num_dims: u8 = 0;
        let mut cardinality: usize = 1;
        for dim in dims {
            num_dims += 1;
            // Saturate: an overflowing cardinality is impossibly large and is
            // rejected by the `nelements` check below. See the matching note in
            // `push_array_with_unchecked`.
            cardinality = cardinality.saturating_mul(dim.length);

            self.row
                .data
                .extend_from_slice(&i64::cast_from(dim.lower_bound).to_le_bytes());
            self.row
                .data
                .extend_from_slice(&u64::cast_from(dim.length).to_le_bytes());
        }

        if num_dims > MAX_ARRAY_DIMENSIONS {
            // Reset the packer state so we don't have invalid data.
            self.row.data.truncate(start);
            return Err(InvalidArrayError::TooManyDimensions(usize::from(num_dims)));
        }
        // Fix up our dimension length.
        self.row.data[dims_start..dims_start + size_of::<u8>()]
            .copy_from_slice(&num_dims.to_le_bytes());

        // Write elements.
        let off = self.row.data.len();
        self.row.data.extend_from_slice(&[0; size_of::<u64>()]);

        let nelements = f(self);

        let len = u64::cast_from(self.row.data.len() - off - size_of::<u64>());
        self.row.data[off..off + size_of::<u64>()].copy_from_slice(&len.to_le_bytes());

        // Check that the number of elements written matches the dimension
        // information.
        let cardinality = match num_dims {
            0 => 0,
            _ => cardinality,
        };
        if nelements != cardinality {
            self.row.data.truncate(start);
            return Err(InvalidArrayError::WrongCardinality {
                actual: nelements,
                expected: cardinality,
            });
        }

        Ok(())
    }

    /// Convenience function to push a `DatumList` from an iter of `Datum`s
    ///
    /// See [`RowPacker::push_dict_with`] if you need to be able to handle errors
    pub fn push_list<'a, I, D>(&mut self, iter: I)
    where
        I: IntoIterator<Item = D>,
        D: Borrow<Datum<'a>>,
    {
        self.push_list_with(|packer| {
            for elem in iter {
                packer.push(*elem.borrow())
            }
        });
    }

    /// Convenience function to push a `DatumMap` from an iter of `(&str, Datum)` pairs
    pub fn push_dict<'a, I, D>(&mut self, iter: I)
    where
        I: IntoIterator<Item = (&'a str, D)>,
        D: Borrow<Datum<'a>>,
    {
        self.push_dict_with(|packer| {
            for (k, v) in iter {
                packer.push(Datum::String(k));
                packer.push(*v.borrow())
            }
        })
    }

    /// Pushes a `Datum::Range` derived from the `Range<Datum<'a>`.
    ///
    /// # Panics
    /// - If lower and upper express finite values and they are datums of
    ///   different types.
    /// - If lower or upper express finite values and are equal to
    ///   `Datum::Null`. To handle `Datum::Null` properly, use
    ///   [`RangeBound::new`].
    ///
    /// # Notes
    /// - This function canonicalizes the range before pushing it to the row.
    /// - Prefer this function over `push_range_with` because of its
    ///   canonicaliztion.
    /// - Prefer creating [`RangeBound`]s using [`RangeBound::new`], which
    ///   handles `Datum::Null` in a SQL-friendly way.
    pub fn push_range<'a>(&mut self, mut range: Range<Datum<'a>>) -> Result<(), InvalidRangeError> {
        range.canonicalize()?;
        match range.inner {
            None => {
                self.row.data.push(Tag::Range.into());
                // Untagged bytes only contains the `RANGE_EMPTY` flag value.
                self.row.data.push(range::InternalFlags::EMPTY.bits());
                Ok(())
            }
            Some(inner) => self.push_range_with(
                RangeLowerBound {
                    inclusive: inner.lower.inclusive,
                    bound: inner
                        .lower
                        .bound
                        .map(|value| move |row: &mut RowPacker| Ok(row.push(value))),
                },
                RangeUpperBound {
                    inclusive: inner.upper.inclusive,
                    bound: inner
                        .upper
                        .bound
                        .map(|value| move |row: &mut RowPacker| Ok(row.push(value))),
                },
            ),
        }
    }

    /// Pushes a `DatumRange` built from the specified arguments.
    ///
    /// # Warning
    /// Unlike `push_range`, `push_range_with` _does not_ canonicalize its
    /// inputs. Consequentially, this means it's possible to generate ranges
    /// that will not reflect the proper ordering and equality.
    ///
    /// # Panics
    /// - If lower or upper expresses a finite value and does not push exactly
    ///   one value into the `RowPacker`.
    /// - If lower and upper express finite values and they are datums of
    ///   different types.
    /// - If lower or upper express finite values and push `Datum::Null`.
    ///
    /// # Notes
    /// - Prefer `push_range_with` over this function. This function should be
    ///   used only when you are not pushing `Datum`s to the inner row.
    /// - Range encoding is `[<flag bytes>,<lower>?,<upper>?]`, where `lower`
    ///   and `upper` are optional, contingent on the flag value expressing an
    ///   empty range (where neither will be present) or infinite bounds (where
    ///   each infinite bound will be absent).
    /// - To push an emtpy range, use `push_range` using `Range { inner: None }`.
    pub fn push_range_with<L, U, E>(
        &mut self,
        lower: RangeLowerBound<L>,
        upper: RangeUpperBound<U>,
    ) -> Result<(), E>
    where
        L: FnOnce(&mut RowPacker) -> Result<(), E>,
        U: FnOnce(&mut RowPacker) -> Result<(), E>,
        E: From<InvalidRangeError>,
    {
        let start = self.row.data.len();
        self.row.data.push(Tag::Range.into());

        let mut flags = range::InternalFlags::empty();

        flags.set(range::InternalFlags::LB_INFINITE, lower.bound.is_none());
        flags.set(range::InternalFlags::UB_INFINITE, upper.bound.is_none());
        flags.set(range::InternalFlags::LB_INCLUSIVE, lower.inclusive);
        flags.set(range::InternalFlags::UB_INCLUSIVE, upper.inclusive);

        let mut expected_datums = 0;

        self.row.data.push(flags.bits());

        let datum_check = self.row.data.len();

        if let Some(value) = lower.bound {
            let start = self.row.data.len();
            value(self)?;
            assert!(
                start < self.row.data.len(),
                "finite values must each push exactly one value; expected 1 but got 0"
            );
            expected_datums += 1;
        }

        if let Some(value) = upper.bound {
            let start = self.row.data.len();
            value(self)?;
            assert!(
                start < self.row.data.len(),
                "finite values must each push exactly one value; expected 1 but got 0"
            );
            expected_datums += 1;
        }

        // Validate the invariants that 0, 1, or 2 elements were pushed, none are Null,
        // and if two are pushed then the second is not less than the first. Panic in
        // some cases and error in others.
        let mut actual_datums = 0;
        let mut seen = None;
        let mut dataz = &self.row.data[datum_check..];
        while !dataz.is_empty() {
            let d = unsafe { read_datum(&mut dataz) };
            // These checks only fail when decoding untrusted/corrupted bytes;
            // valid callers always push consistent, non-null bounds. Return an
            // error rather than asserting so a crafted proto doesn't panic.
            if d == Datum::Null {
                self.row.data.truncate(start);
                return Err(InvalidRangeError::InvalidRangeData.into());
            }

            match seen {
                None => seen = Some(d),
                Some(seen) => {
                    let seen_kind = DatumKind::from(seen);
                    let d_kind = DatumKind::from(d);
                    if seen_kind != d_kind {
                        self.row.data.truncate(start);
                        return Err(InvalidRangeError::InvalidRangeData.into());
                    }

                    if seen > d {
                        self.row.data.truncate(start);
                        return Err(InvalidRangeError::MisorderedRangeBounds.into());
                    }
                }
            }
            actual_datums += 1;
        }

        if actual_datums != expected_datums {
            self.row.data.truncate(start);
            return Err(InvalidRangeError::InvalidRangeData.into());
        }

        Ok(())
    }

    /// Clears the contents of the packer without de-allocating its backing memory.
    pub fn clear(&mut self) {
        self.row.data.clear();
    }

    /// Truncates the underlying storage to the specified byte position.
    ///
    /// # Safety
    ///
    /// `pos` MUST specify a byte offset that lies on a datum boundary.
    /// If `pos` specifies a byte offset that is *within* a datum, the row
    /// packer will produce an invalid row, the unpacking of which may
    /// trigger undefined behavior!
    ///
    /// To find the byte offset of a datum boundary, inspect the packer's
    /// byte length by calling `packer.data().len()` after pushing the desired
    /// number of datums onto the packer.
    pub unsafe fn truncate(&mut self, pos: usize) {
        self.row.data.truncate(pos)
    }

    /// Truncates the underlying row to contain at most the first `n` datums.
    pub fn truncate_datums(&mut self, n: usize) {
        let prev_len = self.row.data.len();
        let mut iter = self.row.iter();
        for _ in iter.by_ref().take(n) {}
        let next_len = iter.data.len();
        // SAFETY: iterator offsets always lie on a datum boundary.
        unsafe { self.truncate(prev_len - next_len) }
    }

    /// Returns the total amount of bytes used by the underlying row.
    pub fn byte_len(&self) -> usize {
        self.row.byte_len()
    }
}

impl<'a> IntoIterator for &'a Row {
    type Item = Datum<'a>;
    type IntoIter = DatumListIter<'a>;
    fn into_iter(self) -> DatumListIter<'a> {
        self.iter()
    }
}

impl fmt::Debug for Row {
    /// Debug representation using the internal datums
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("Row{")?;
        f.debug_list().entries(self.iter()).finish()?;
        f.write_str("}")
    }
}

impl fmt::Display for Row {
    /// Display representation using the internal datums
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("(")?;
        for (i, datum) in self.iter().enumerate() {
            if i != 0 {
                f.write_str(", ")?;
            }
            write!(f, "{}", datum)?;
        }
        f.write_str(")")
    }
}

impl<'a, T> DatumList<'a, T> {
    pub fn iter(&self) -> DatumListIter<'a> {
        DatumListIter { data: self.data }
    }

    /// Iterate elements as typed `T` values rather than raw `Datum`s.
    ///
    /// Each datum is decoded and converted via [`FromDatum`]. Since generic
    /// type parameters in `#[sqlfunc]` are erased to `Datum<'a>` before code
    /// generation, this is monomorphized to an identity conversion at runtime.
    pub fn typed_iter(&self) -> DatumListTypedIter<'a, T>
    where
        T: FromDatum<'a>,
    {
        DatumListTypedIter {
            inner: self.iter(),
            _phantom: PhantomData,
        }
    }

    /// For debugging only
    pub fn data(&self) -> &'a [u8] {
        self.data
    }
}

impl<T> DatumList<'static, T> {
    pub fn empty() -> Self {
        DatumList::new(&[])
    }
}

impl<'a> IntoIterator for DatumList<'a> {
    type Item = Datum<'a>;
    type IntoIter = DatumListIter<'a>;
    fn into_iter(self) -> DatumListIter<'a> {
        self.iter()
    }
}

impl<'a> Iterator for DatumListIter<'a> {
    type Item = Datum<'a>;
    fn next(&mut self) -> Option<Self::Item> {
        if self.data.is_empty() {
            None
        } else {
            Some(unsafe { read_datum(&mut self.data) })
        }
    }
}

impl<'a, T: FromDatum<'a>> Iterator for DatumListTypedIter<'a, T> {
    type Item = T;
    fn next(&mut self) -> Option<Self::Item> {
        self.inner.next().map(T::from_datum)
    }
}

impl<'a, T> DatumMap<'a, T> {
    pub fn iter(&self) -> DatumDictIter<'a> {
        DatumDictIter {
            data: self.data,
            prev_key: None,
        }
    }

    /// Iterate entries as `(&str, T)` pairs rather than `(&str, Datum)`.
    ///
    /// Each value datum is converted via [`FromDatum`]. Since generic type
    /// parameters in `#[sqlfunc]` are erased to `Datum<'a>` before code
    /// generation, this is monomorphized to an identity conversion at runtime.
    pub fn typed_iter(&self) -> DatumDictTypedIter<'a, T>
    where
        T: FromDatum<'a>,
    {
        DatumDictTypedIter {
            inner: self.iter(),
            _phantom: PhantomData,
        }
    }

    /// For debugging only
    pub fn data(&self) -> &'a [u8] {
        self.data
    }
}

impl<T> DatumMap<'static, T> {
    pub fn empty() -> Self {
        DatumMap::new(&[])
    }
}

impl<'a, T> Debug for DatumMap<'a, T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_map().entries(self.iter()).finish()
    }
}

impl<'a> IntoIterator for &'a DatumMap<'a> {
    type Item = (&'a str, Datum<'a>);
    type IntoIter = DatumDictIter<'a>;
    fn into_iter(self) -> DatumDictIter<'a> {
        self.iter()
    }
}

impl<'a> Iterator for DatumDictIter<'a> {
    type Item = (&'a str, Datum<'a>);
    fn next(&mut self) -> Option<Self::Item> {
        if self.data.is_empty() {
            None
        } else {
            let key_tag =
                Tag::try_from_primitive(read_byte(&mut self.data)).expect("unknown row tag");
            assert!(
                key_tag == Tag::StringTiny
                    || key_tag == Tag::StringShort
                    || key_tag == Tag::StringLong
                    || key_tag == Tag::StringHuge,
                "Dict keys must be strings, got {:?}",
                key_tag
            );
            let bytes = read_lengthed_bytes(&mut self.data, key_tag, Tag::StringTiny);
            // SAFETY: the bytes were written from a `str` under a `String` tag.
            let key = unsafe { str::from_utf8_unchecked(bytes) };
            let val = unsafe { read_datum(&mut self.data) };

            // Gate the `prev_key` bookkeeping on the same flag as the assert it feeds, so builds
            // with soft assertions off pay nothing for it.
            if mz_ore::assert::soft_assertions_enabled() {
                if let Some(prev_key) = self.prev_key {
                    mz_ore::soft_assert_no_log!(
                        prev_key < key,
                        "Dict keys must be unique and given in ascending order: {} came before {}",
                        prev_key,
                        key
                    );
                }
                self.prev_key = Some(key);
            }

            Some((key, val))
        }
    }
}

impl<'a, T: FromDatum<'a>> Iterator for DatumDictTypedIter<'a, T> {
    type Item = (&'a str, T);
    fn next(&mut self) -> Option<Self::Item> {
        self.inner.next().map(|(k, v)| (k, T::from_datum(v)))
    }
}

impl RowArena {
    pub fn new() -> Self {
        RowArena {
            inner: RefCell::new(vec![]),
            scratch: RefCell::new(None),
            budget: None,
            allocated: Cell::new(0),
        }
    }

    /// Creates a `RowArena` that reports itself [`RowArena::over_budget`] once it holds more than
    /// `budget` bytes.
    ///
    /// The budget is advisory to the arena itself: pushes still succeed, because handing back a
    /// truncated value would corrupt the datum. It is the caller's job to poll `over_budget` at a
    /// point where it can fail, so the bytes an arena actually reaches is `budget` plus whatever the
    /// operation in flight at the time added.
    ///
    /// NOTE: a budget bounds a single ad-hoc evaluation in a shared process (see
    /// `mz_adapter::webhook`). It must not be given to an arena that feeds a compute dataflow. A
    /// dataflow re-evaluates the same expression against the same input and must return the same
    /// result every time. Whether an evaluation is over budget depends on what else the arena has
    /// accumulated, and the webhook budget is a runtime dyncfg, so a budgeted dataflow arena would
    /// make the result depend on when it ran. Differential then turns a changed-but-not-retracted
    /// result into non-accumulating diffs that corrupt the collection. A dataflow that ever needs a
    /// budget must fix it for the lifetime of a cluster replica.
    pub fn with_budget(budget: usize) -> Self {
        RowArena {
            budget: Some(budget),
            ..RowArena::new()
        }
    }

    /// Bytes this arena currently holds.
    pub fn allocated_bytes(&self) -> usize {
        self.allocated.get()
    }

    /// Whether this arena holds more than its budget. Always false without one.
    pub fn over_budget(&self) -> bool {
        self.budget
            .is_some_and(|budget| self.allocated.get() > budget)
    }

    /// Bytes this arena can still take before it is [`RowArena::over_budget`], or `usize::MAX`
    /// without a budget.
    ///
    /// Intended for an operation that can predict its own size and would rather fail than build a
    /// value it is about to be told is too big.
    pub fn budget_remaining(&self) -> usize {
        match self.budget {
            None => usize::MAX,
            Some(budget) => budget.saturating_sub(self.allocated.get()),
        }
    }

    /// Creates a `RowArena` with an initial region sized to hold `capacity` bytes, to avoid
    /// reallocations as the first datums are created in the arena.
    pub fn with_capacity(capacity: usize) -> Self {
        let mut inner = Vec::new();
        if capacity > 0 {
            inner.push(Vec::with_capacity(capacity));
        }
        RowArena {
            inner: RefCell::new(inner),
            ..RowArena::new()
        }
    }

    /// Ensures the active region can hold at least `additional` more bytes without allocating a
    /// new region. Call this when you expect to push roughly `additional` bytes next.
    pub fn reserve(&self, additional: usize) {
        if additional == 0 {
            return;
        }
        let mut inner = self.inner.borrow_mut();
        match inner.last_mut() {
            // The active region is empty, so nothing references it yet and it is safe to grow it
            // in place (a reallocation cannot dangle a live reference).
            Some(active) if active.is_empty() => {
                if active.capacity() < additional {
                    active.reserve_exact(additional);
                }
            }
            // The active region holds live data; we cannot grow it without moving those bytes, so
            // stage a fresh region. Size it like `push_bytes` does (at least double the current
            // region) so a sequence of small `reserve`s still yields at most log-many regions
            // rather than many small ones.
            Some(active) => {
                let new_cap = std::cmp::max(additional, active.capacity().saturating_mul(2));
                inner.push(Vec::with_capacity(new_cap));
            }
            None => inner.push(Vec::with_capacity(additional)),
        }
    }

    /// Copies `bytes` into the arena and returns a reference valid for its lifetime.
    ///
    /// Accepts anything that derefs to `[u8]` (e.g. `Vec<u8>`, `&[u8]`); the bytes are copied, so
    /// the caller's allocation is not retained.
    #[allow(clippy::transmute_ptr_to_ptr)]
    pub fn push_bytes<'a, B: Deref<Target = [u8]>>(&'a self, bytes: B) -> &'a [u8] {
        let bytes: &[u8] = &bytes;
        let need = bytes.len();
        if need == 0 {
            return &[];
        }
        let mut inner = self.inner.borrow_mut();

        // Find or create a region with spare capacity for `need` bytes, never growing a region
        // that already holds data (see the type-level comment for why this preserves references).
        let has_room = inner
            .last()
            .map_or(false, |region| region.capacity() - region.len() >= need);
        if !has_room {
            let last_cap = inner.last().map_or(0, |region| region.capacity());
            let new_cap = std::cmp::max(need, last_cap.saturating_mul(2));
            inner.push(Vec::with_capacity(new_cap));
        }

        let region = inner.last_mut().expect("region present");
        let start = region.len();
        region.extend_from_slice(bytes);
        self.allocated.set(self.allocated.get() + need);
        let copied = &region[start..];
        unsafe {
            // This is safe because:
            //   * `copied` references bytes inside `region`'s heap buffer, which we just sized to
            //     fit without reallocating; that buffer is never resized again while it holds data
            //     (we allocate a new region instead), so the reference stays valid.
            //   * The buffer lives as long as the arena: regions are only dropped by `clear`/`drop`,
            //     both of which take `&mut`/ownership, so no `&'a self`-tied reference can outlive
            //     them.
            //   * Pushing further regions may reallocate `self.inner`, but that moves only the
            //     `Vec<u8>` headers, not the heap buffers they own.
            transmute::<&[u8], &'a [u8]>(copied)
        }
    }

    /// Moves `bytes` into the arena and returns a reference valid for its lifetime.
    ///
    /// Prefer this to [`RowArena::push_bytes`] whenever the bytes are already owned: a value large
    /// enough that it would get a region to itself has its allocation adopted as that region, rather
    /// than a fresh region being allocated and copied into, which for a large value halves the peak.
    /// Smaller values are copied, so the arena keeps bump allocating.
    pub fn push_owned_bytes<'a>(&'a self, bytes: Vec<u8>) -> &'a [u8] {
        /// Never adopt below this, however empty the arena. `last_cap` alone would let every value
        /// on a fresh or small arena look big enough, and then each gets an exactly-sized region
        /// with no headroom: one region and one `Vec<u8>` header per value.
        const MIN_ADOPT_BYTES: usize = 4 * 1024;

        let need = bytes.len();
        if need == 0 {
            return &[];
        }

        let mut inner = self.inner.borrow_mut();
        // Adopt only when `push_bytes` would have given these bytes a dedicated, `need`-sized region
        // anyway, i.e. when `need` exceeds the `last_cap * 2` it would otherwise allocate. There's
        // no headroom to lose, so adoption saves a copy for free. Below that we copy, because
        // `push_bytes` grows a region *with* headroom that later values reuse. Adopting there would
        // defeat the bump allocator: adoption leaves an empty region (capacity 0) on top, so nothing
        // would ever grow a region with headroom again.
        let last_cap = inner.last().map_or(0, |region| region.capacity());
        let adopt = need > std::cmp::max(MIN_ADOPT_BYTES, last_cap.saturating_mul(2));
        if !adopt {
            drop(inner);
            return self.push_bytes(&bytes[..]);
        }

        // `push_bytes` would allocate a fresh region here and copy into it, so adopt the caller's
        // allocation as that region. Sound for the same reasons as `push_bytes`: the reference
        // points into a heap buffer the arena now owns for `'a`, and the buffer is never resized
        // while it holds data.
        //
        // Inserted *below* the active region rather than appended, because `push_bytes` sizes a new
        // region as twice the last one's capacity: leaving a large adopted buffer on top would make
        // the next push allocate twice its size.
        self.allocated.set(self.allocated.get() + need);
        let idx = inner.len().saturating_sub(1);
        inner.insert(idx, bytes);
        if inner.len() == 1 {
            // There was no active region to insert below, so keep an empty one on top for the same
            // reason. `Vec::new` does not allocate.
            inner.push(Vec::new());
        }
        let adopted = &inner[idx][..];
        unsafe { transmute::<&[u8], &'a [u8]>(adopted) }
    }

    /// Moves `string` into the arena and returns a reference valid for its lifetime.
    pub fn push_string<'a>(&'a self, string: String) -> &'a str {
        let copied = self.push_owned_bytes(string.into_bytes());
        unsafe {
            // This is safe because we just moved in the bytes of a valid `String`.
            std::str::from_utf8_unchecked(copied)
        }
    }

    /// Returns a growable, writeable byte buffer for assembling a value incrementally.
    ///
    /// Write into it with [`RowArenaBuf::push`], [`RowArenaBuf::extend_from_slice`], or
    /// [`std::io::Write`], then call [`RowArenaBuf::finish`] to copy the result into the arena and
    /// obtain a reference valid for the arena's lifetime. The backing buffer is a single scratch
    /// allocation reused across writers, so this lets a producer that builds bytes piecewise (e.g.
    /// decoding a row) avoid managing its own scratch.
    ///
    /// Nested writers are sound but not free: a writer obtained while another is still live can't
    /// reuse the (in-use) scratch, so it allocates its own buffer. Steady-state, non-nested use
    /// stays allocation-free.
    pub fn writer(&self) -> RowArenaBuf<'_> {
        // Take the recycled buffer if one is available, else allocate a fresh one. The cell is
        // borrowed only for this `take`, never for the writer's lifetime, so a nested `writer` call
        // doesn't double-borrow: it simply finds the slot empty and allocates its own buffer.
        let mut buf = self.scratch.borrow_mut().take().unwrap_or_default();
        buf.clear();
        RowArenaBuf { arena: self, buf }
    }

    /// Take ownership of `row` for the lifetime of the arena, returning a
    /// reference to the first datum in the row.
    ///
    /// If we had an owned datum type, this method would be much clearer, and
    /// would be called `push_owned_datum`.
    pub fn push_unary_row<'a>(&'a self, row: Row) -> Datum<'a> {
        let copied = self.push_bytes(row.data());
        unsafe {
            // This is safe because `copied` is a valid encoding of a single datum (we just packed
            // it into `row`), backed by the arena for the lifetime `'a`. Copying the bytes also
            // sidesteps the `Row`'s inline (`SmallVec`) storage entirely.
            let datum = read_datum(&mut &copied[..]);
            transmute::<Datum<'_>, Datum<'a>>(datum)
        }
    }

    /// Equivalent to `push_unary_row` but returns a `DatumNested` rather than a
    /// `Datum`.
    fn push_unary_row_datum_nested<'a>(&'a self, row: Row) -> DatumNested<'a> {
        let copied = self.push_bytes(row.data());
        unsafe {
            // Safe for the same reasons as `push_unary_row`.
            let nested = DatumNested::extract(&mut &copied[..]);
            transmute::<DatumNested<'_>, DatumNested<'a>>(nested)
        }
    }

    /// Convenience function to make a new `Row` containing a single datum, and
    /// take ownership of it for the lifetime of the arena
    ///
    /// ```
    /// # use mz_repr::{RowArena, Datum};
    /// let arena = RowArena::new();
    /// let datum = arena.make_datum(|packer| {
    ///   packer.push_list(&[Datum::String("hello"), Datum::String("world")]);
    /// });
    /// assert_eq!(datum.unwrap_list().iter().collect::<Vec<_>>(), vec![Datum::String("hello"), Datum::String("world")]);
    /// ```
    pub fn make_datum<'a, F>(&'a self, f: F) -> Datum<'a>
    where
        F: FnOnce(&mut RowPacker),
    {
        let mut row = Row::default();
        f(&mut row.packer());
        self.push_unary_row(row)
    }

    /// Convenience function to build a list datum from an iterator of typed
    /// elements and return it as a `DatumList<'a, T>`.
    ///
    /// By accepting an iterator of `T: Borrow<Datum>` instead of a raw
    /// `RowPacker` closure, this guarantees that only elements of type `T`
    /// are pushed.
    pub fn make_datum_list<'a, T: std::borrow::Borrow<Datum<'a>>>(
        &'a self,
        iter: impl IntoIterator<Item = T>,
    ) -> DatumList<'a, T> {
        let datum = self.make_datum(|packer| {
            packer.push_list_with(|packer| {
                for elem in iter {
                    packer.push(*elem.borrow());
                }
            });
        });
        DatumList::new(datum.unwrap_list().data())
    }

    /// Convenience function identical to `make_datum` but instead returns a
    /// `DatumNested`.
    pub fn make_datum_nested<'a, F>(&'a self, f: F) -> DatumNested<'a>
    where
        F: FnOnce(&mut RowPacker),
    {
        let mut row = Row::default();
        f(&mut row.packer());
        self.push_unary_row_datum_nested(row)
    }

    /// Like [`RowArena::make_datum`], but the provided closure can return an error.
    pub fn try_make_datum<'a, F, E>(&'a self, f: F) -> Result<Datum<'a>, E>
    where
        F: FnOnce(&mut RowPacker) -> Result<(), E>,
    {
        let mut row = Row::default();
        f(&mut row.packer())?;
        Ok(self.push_unary_row(row))
    }

    /// Clear the contents of the arena.
    ///
    /// Retains the single largest region (emptied) so the arena can be reused without
    /// reallocating; a workload that clears between uses of similar size becomes allocation-free.
    pub fn clear(&mut self) {
        let inner = self.inner.get_mut();
        // Keep only the largest-capacity region, reset to empty, and drop the rest. Because region
        // capacities only ever grow (each new region at least doubles the previous), the largest is
        // normally the last; we scan for it defensively, which is cheap given log-many regions.
        if let Some(largest) = (0..inner.len()).max_by_key(|&i| inner[i].capacity()) {
            inner.swap(0, largest);
            inner.truncate(1);
            inner[0].clear();
        }
        self.allocated.set(0);
    }
}

impl Default for RowArena {
    fn default() -> RowArena {
        RowArena::new()
    }
}

/// A growable, writeable byte buffer that builds a value into a [`RowArena`].
///
/// Obtained from [`RowArena::writer`]. Behaves like a writeable byte slice (push/extend bytes,
/// read back as `&[u8]`); [`RowArenaBuf::finish`] copies the assembled bytes into the arena and
/// returns a reference valid for the arena's lifetime. The buffer is owned for the writer's
/// lifetime and, on drop, returned to the arena to be reused by the next writer.
#[derive(Debug)]
pub struct RowArenaBuf<'a> {
    arena: &'a RowArena,
    buf: Vec<u8>,
}

impl<'a> RowArenaBuf<'a> {
    /// Appends a single byte.
    pub fn push(&mut self, byte: u8) {
        self.buf.push(byte);
    }

    /// Appends a slice of bytes.
    pub fn extend_from_slice(&mut self, bytes: &[u8]) {
        self.buf.extend_from_slice(bytes);
    }

    /// The bytes written so far.
    pub fn as_slice(&self) -> &[u8] {
        &self.buf
    }

    /// The number of bytes written so far.
    pub fn len(&self) -> usize {
        self.buf.len()
    }

    /// Whether no bytes have been written.
    pub fn is_empty(&self) -> bool {
        self.buf.is_empty()
    }

    /// Copies the written bytes into the arena, returning a reference valid for its lifetime.
    pub fn finish(self) -> &'a [u8] {
        // `self` is dropped at the end of this call, returning `buf` to the arena for reuse; the
        // returned reference points into a committed region, not `buf`, so it stays valid.
        self.arena.push_bytes(self.buf.as_slice())
    }

    /// Like [`RowArenaBuf::finish`], but returns the bytes as a `&str`.
    ///
    /// Intended for buffers written via [`std::fmt::Write`] (e.g. `write!`), whose contents are
    /// valid UTF-8. Panics if the bytes are not valid UTF-8.
    pub fn finish_str(self) -> &'a str {
        let bytes = self.arena.push_bytes(self.buf.as_slice());
        std::str::from_utf8(bytes).expect("RowArenaBuf::finish_str on non-UTF-8 contents")
    }
}

impl<'a> Drop for RowArenaBuf<'a> {
    fn drop(&mut self) {
        // Return the buffer to the arena so the next writer can reuse its allocation. We keep only
        // one buffer: if the slot is already occupied — an outer writer is still live, or a nested
        // writer beat us to it — we drop ours rather than growing an unbounded pool. The borrow is
        // transient and never overlaps a live writer's, so this can't double-borrow.
        let mut slot = self.arena.scratch.borrow_mut();
        if slot.is_none() {
            *slot = Some(std::mem::take(&mut self.buf));
        }
    }
}

impl<'a> std::ops::Deref for RowArenaBuf<'a> {
    type Target = [u8];
    fn deref(&self) -> &[u8] {
        &self.buf
    }
}

impl<'a> std::io::Write for RowArenaBuf<'a> {
    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
        self.buf.extend_from_slice(bytes);
        Ok(bytes.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

impl<'a> std::fmt::Write for RowArenaBuf<'a> {
    fn write_str(&mut self, s: &str) -> std::fmt::Result {
        self.buf.extend_from_slice(s.as_bytes());
        Ok(())
    }
}

/// A thread-local row, which can be borrowed and returned.
/// # Example
///
/// Use this type instead of creating a new row:
/// ```
/// use mz_repr::SharedRow;
///
/// let mut row_builder = SharedRow::get();
/// ```
///
/// This allows us to reuse an existing row allocation instead of creating a new one or retaining
/// an allocation locally. Additionally, we can observe the size of the local row in a central
/// place and potentially reallocate to reduce memory needs.
///
/// # Panic
///
/// [`SharedRow::get`] panics when trying to obtain multiple references to the shared row.
#[derive(Debug)]
pub struct SharedRow(Row);

impl SharedRow {
    thread_local! {
        /// A thread-local slot containing a shared Row that can be temporarily used by a function.
        /// There can be at most one active user of this Row, which is tracked by the state of the
        /// `Option<_>` wrapper. When it is `Some(..)`, the row is available for using. When it
        /// is `None`, it is not, and the constructor will panic if a thread attempts to use it.
        static SHARED_ROW: Cell<Option<Row>> = const { Cell::new(Some(Row::empty())) }
    }

    /// Get the shared row.
    ///
    /// The row's contents are cleared before returning it.
    ///
    /// # Panic
    ///
    /// Panics when the row is already borrowed elsewhere.
    pub fn get() -> Self {
        let mut row = Self::SHARED_ROW
            .take()
            .expect("attempted to borrow already borrowed SharedRow");
        // Clear row
        row.packer();
        Self(row)
    }

    /// Gets the shared row and uses it to pack `iter`.
    pub fn pack<'a, I, D>(iter: I) -> Row
    where
        I: IntoIterator<Item = D>,
        D: Borrow<Datum<'a>>,
    {
        let mut row_builder = Self::get();
        let mut row_packer = row_builder.packer();
        row_packer.extend(iter);
        row_builder.clone()
    }
}

impl std::ops::Deref for SharedRow {
    type Target = Row;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl std::ops::DerefMut for SharedRow {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.0
    }
}

impl Drop for SharedRow {
    fn drop(&mut self) {
        // Take the Row allocation from this instance and put it back in the thread local slot for
        // the next user. The Row in `self` is replaced with an empty Row which does not allocate.
        Self::SHARED_ROW.set(Some(std::mem::take(&mut self.0)))
    }
}

#[cfg(test)]
mod tests {
    use std::cmp::Ordering;
    use std::collections::hash_map::DefaultHasher;
    use std::hash::{Hash, Hasher};

    use chrono::{DateTime, NaiveDate};
    use itertools::Itertools;
    use mz_ore::{assert_err, assert_none};
    use ordered_float::OrderedFloat;

    use crate::SqlScalarType;

    use super::*;

    // StableRow's wire format is proto bytes, not the in-memory datum
    // encoding, so rows of every column type must roundtrip exactly through
    // both a self-describing format (JSON) and a compact binary one
    // (bincode). Equality on Row compares the packed in-memory bytes, so
    // this also catches any datum normalization sneaking into the
    // Row -> ProtoRow -> Row conversion.
    proptest! {
        #![proptest_config(ProptestConfig::with_cases(1000))]

        #[mz_ore::test]
        #[cfg_attr(miri, ignore)] // too slow, and decNumber uses FFI
        fn stable_row_serde_roundtrip(
            stable in crate::relation::arb_relation_desc(1..8)
                .prop_flat_map(|desc| crate::relation::arb_row_for_relation(&desc))
                .prop_map(StableRow)
        ) {
            let json = serde_json::to_string(&stable).expect("serializes to JSON");
            let from_json: StableRow =
                serde_json::from_str(&json).expect("deserializes from JSON");
            prop_assert_eq!(&stable, &from_json);

            let bytes = bincode::serialize(&stable).expect("serializes to bincode");
            let from_bincode: StableRow =
                bincode::deserialize(&bytes).expect("deserializes from bincode");
            prop_assert_eq!(&stable, &from_bincode);
        }
    }

    /// Every width a variable-length integer can take, at both ends of its range and either
    /// side of each byte boundary, plus the values whose payload is empty.
    fn varint_edge_cases() -> Vec<Datum<'static>> {
        let mut datums = vec![
            Datum::Int16(0),
            Datum::Int16(-1),
            Datum::Int16(i16::MIN),
            Datum::Int16(i16::MAX),
            Datum::Int32(0),
            Datum::Int32(-1),
            Datum::Int32(i32::MIN),
            Datum::Int32(i32::MAX),
            Datum::Int64(0),
            Datum::Int64(-1),
            Datum::Int64(i64::MIN),
            Datum::Int64(i64::MAX),
            Datum::UInt8(0),
            Datum::UInt8(u8::MAX),
            Datum::UInt16(0),
            Datum::UInt16(u16::MAX),
            Datum::UInt32(0),
            Datum::UInt32(u32::MAX),
            Datum::UInt64(0),
            Datum::UInt64(u64::MAX),
        ];
        // One below, at, and one above every point where the payload grows a byte.
        for bits in 1..64 {
            let boundary = 1u64 << bits;
            for delta in [-1i64, 0, 1] {
                let Some(v) = boundary.checked_add_signed(delta) else {
                    continue;
                };
                datums.push(Datum::UInt64(v));
                if let Ok(v) = u32::try_from(v) {
                    datums.push(Datum::UInt32(v));
                }
                if let Ok(v) = u16::try_from(v) {
                    datums.push(Datum::UInt16(v));
                }
                if let Ok(v) = u8::try_from(v) {
                    datums.push(Datum::UInt8(v));
                }
                let Ok(v) = i64::try_from(v) else {
                    continue;
                };
                datums.push(Datum::Int64(v));
                datums.push(Datum::Int64(-v));
                if let Ok(v) = i32::try_from(v) {
                    datums.push(Datum::Int32(v));
                    datums.push(Datum::Int32(-v));
                }
                if let Ok(v) = i16::try_from(v) {
                    datums.push(Datum::Int16(v));
                    datums.push(Datum::Int16(-v));
                }
            }
        }
        datums
    }

    /// Both signed families of a width share one match arm, which recovers the payload width by
    /// subtracting the family's first tag. A value whose tag falls outside the family it is
    /// decoded as would read the wrong number of bytes, so check every boundary lands where the
    /// arithmetic expects.
    #[mz_ore::test]
    fn varint_tags_land_in_their_family() {
        for datum in varint_edge_cases() {
            let row = Row::pack_slice(&[datum]);
            let tag = Tag::try_from_primitive(row.data[0]).expect("valid tag");
            let (first, len, negative, widest) = match datum {
                Datum::Int16(i) => (Tag::NonNegativeInt16_0, min_bytes_signed(i), i < 0, 2),
                Datum::Int32(i) => (Tag::NonNegativeInt32_0, min_bytes_signed(i), i < 0, 4),
                Datum::Int64(i) => (Tag::NonNegativeInt64_0, min_bytes_signed(i), i < 0, 8),
                Datum::UInt8(u) => (Tag::UInt8_0, min_bytes_unsigned(u), false, 1),
                Datum::UInt16(u) => (Tag::UInt16_0, min_bytes_unsigned(u), false, 2),
                Datum::UInt32(u) => (Tag::UInt32_0, min_bytes_unsigned(u), false, 4),
                Datum::UInt64(u) => (Tag::UInt64_0, min_bytes_unsigned(u), false, 8),
                other => panic!("not a variable-length integer: {other:?}"),
            };
            let delta = u8::from(tag) - u8::from(first);
            let signed = matches!(datum, Datum::Int16(_) | Datum::Int32(_) | Datum::Int64(_));
            if signed {
                // Interleaved: the width sits above the low bit, the sign in it. At the widest
                // width the alternation stops, one tag serving both signs.
                assert_eq!(delta >> 1, len, "wrong payload width in tag for {datum:?}");
                let sign_bit = if len == widest { 0 } else { u8::from(negative) };
                assert_eq!(delta & 1, sign_bit, "wrong sign in tag for {datum:?}");
            } else {
                assert_eq!(delta, len, "wrong payload width in tag for {datum:?}");
            }
            assert_eq!(row.unpack_first(), datum, "did not round-trip: {datum:?}");
        }
    }

    /// The wide load in `read_varint_word` reads eight bytes whatever the payload's width, so
    /// the datums at the end of a row, where those bytes do not exist, take the tail path.
    #[mz_ore::test]
    fn varint_at_end_of_row_reads_exact_width() {
        for datum in varint_edge_cases() {
            // Alone in a row, and behind padding long enough to push the fast path back in.
            for prefix in [None, Some(Datum::String("0123456789abcdef"))] {
                let row = match prefix {
                    Some(p) => Row::pack_slice(&[p, datum]),
                    None => Row::pack_slice(&[datum]),
                };
                assert_eq!(
                    row.iter().last(),
                    Some(datum),
                    "did not round-trip at end of row: {datum:?}"
                );
            }
            // And nested, where the list's slice ends before the row's data does.
            let mut row = Row::default();
            let mut packer = row.packer();
            packer.push_list_with(|packer| packer.push(datum));
            packer.push(Datum::Int64(1));
            let list = match row.unpack_first() {
                Datum::List(list) => list,
                other => panic!("expected a list, got {other:?}"),
            };
            assert_eq!(list.iter().next(), Some(datum), "did not round-trip nested");
        }
    }

    // Regression: comparing deeply nested list values must not overflow the
    // stack (STACK-7). `Datum` ordering recurses once per nesting level.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn cmp_deep_nested_list_does_not_overflow() {
        fn deep() -> Row {
            // `push_list` byte-copies the inner value, so building does not recurse.
            let mut row = Row::pack_slice(&[Datum::Int64(1)]);
            for _ in 0..50_000 {
                let mut next = Row::default();
                next.packer().push_list([row.unpack_first()]);
                row = next;
            }
            row
        }
        let a = deep();
        let b = deep();
        assert_eq!(a.unpack_first().cmp(&b.unpack_first()), Ordering::Equal);
    }

    fn hash<T: Hash>(t: &T) -> u64 {
        let mut hasher = DefaultHasher::new();
        t.hash(&mut hasher);
        hasher.finish()
    }

    #[mz_ore::test]
    fn test_assumptions() {
        assert_eq!(size_of::<Tag>(), 1);
        #[cfg(target_endian = "big")]
        {
            // if you want to run this on a big-endian cpu, we'll need big-endian versions of the serialization code
            assert!(false);
        }
    }

    #[mz_ore::test]
    fn miri_test_arena() {
        let arena = RowArena::new();

        assert_eq!(arena.push_string("".to_owned()), "");
        assert_eq!(arena.push_string("العَرَبِيَّة".to_owned()), "العَرَبِيَّة");

        let empty: &[u8] = &[];
        assert_eq!(arena.push_bytes(vec![]), empty);
        assert_eq!(arena.push_bytes(vec![0, 2, 1, 255]), &[0, 2, 1, 255]);

        let mut row = Row::default();
        let mut packer = row.packer();
        packer.push_dict_with(|row| {
            row.push(Datum::String("a"));
            row.push_list_with(|row| {
                row.push(Datum::String("one"));
                row.push(Datum::String("two"));
                row.push(Datum::String("three"));
            });
            row.push(Datum::String("b"));
            row.push(Datum::String("c"));
        });
        assert_eq!(arena.push_unary_row(row.clone()), row.unpack_first());
    }

    #[mz_ore::test]
    fn miri_test_arena_growth_keeps_references() {
        // References returned by `push_bytes` must stay valid as later pushes allocate new
        // regions; this exercises the "never resize a region that holds data" invariant.
        let arena = RowArena::new();
        let chunks: Vec<Vec<u8>> = (0..128u16)
            .map(|i| vec![u8::try_from(i % 256).unwrap(); usize::from(i % 13) + 1])
            .collect();
        let refs: Vec<&[u8]> = chunks
            .iter()
            .map(|c| arena.push_bytes(c.as_slice()))
            .collect();
        for (i, r) in refs.iter().enumerate() {
            assert_eq!(*r, chunks[i].as_slice());
        }
    }

    #[mz_ore::test]
    fn miri_test_arena_unary_row_at_offset() {
        // A row pushed after other bytes lands at a non-zero offset within a region; reading it
        // back must not depend on the row starting at offset zero or on any alignment.
        let arena = RowArena::new();
        arena.reserve(4096);
        let _pad = arena.push_bytes(vec![0xAB; 5]);
        let row = Row::pack_slice(&[Datum::String("hello"), Datum::Int64(42), Datum::True]);
        assert_eq!(arena.push_unary_row(row.clone()), row.unpack_first());
    }

    #[mz_ore::test]
    fn miri_test_arena_clear_reuse() {
        // After `clear` the arena retains a region and remains usable across cycles.
        let mut arena = RowArena::new();
        for i in 0..100u8 {
            let _ = arena.push_bytes(vec![i; 16]);
        }
        arena.clear();
        assert_eq!(arena.push_bytes(vec![7u8; 8]), &[7u8; 8]);
        assert_eq!(arena.push_string("after clear".to_owned()), "after clear");
        arena.clear();
        let empty: &[u8] = &[];
        assert_eq!(arena.push_bytes(Vec::<u8>::new()), empty);
    }

    #[mz_ore::test]
    fn miri_test_arena_adopts_owned_bytes_and_keeps_references() {
        // `push_owned_bytes` adopts a buffer too large for the active region instead of copying it,
        // which puts a region the arena never wrote into in the middle of the stack. References
        // handed out before and after that must all stay valid.
        let arena = RowArena::new();
        let before = arena.push_bytes(vec![1u8; 8]);
        let adopted = arena.push_owned_bytes(vec![2u8; 64 * 1024]);
        let after = arena.push_bytes(vec![3u8; 8]);
        // A small buffer fits the active region, so it is copied rather than given a region.
        let small = arena.push_owned_bytes(vec![4u8; 4]);

        assert_eq!(before, &[1u8; 8]);
        assert_eq!(adopted, &vec![2u8; 64 * 1024][..]);
        assert_eq!(after, &[3u8; 8]);
        assert_eq!(small, &[4u8; 4]);

        let empty: &[u8] = &[];
        assert_eq!(arena.push_owned_bytes(vec![]), empty);
    }

    #[mz_ore::test]
    fn test_arena_owned_pushes_keep_bump_allocating() {
        // Adoption never *creates* a region with headroom: it inserts the caller's buffer, whose
        // capacity equals its length, below whatever is on top. Only `push_bytes` grows the arena
        // geometrically (`new_cap = max(need, last_cap * 2)`), so once the active region cannot fit
        // an incoming value it never can again and every later owned push adopts: one retained
        // allocation and one `Vec<u8>` header per value, rather than `O(log n)` regions. That is the
        // default path for every `String`- and `Vec<u8>`-returning scalar function, and the arenas
        // in the MFP and join paths outlive a single row, so the region list grows with the number
        // of string values in a batch. `RowArena::clear` scans every region, so it degrades too.
        const VALUES: usize = 500;
        const VALUE: &str = "0123456789";

        let regions = |arena: &RowArena| arena.inner.borrow().len();
        let push_all = |arena: &RowArena, owned: bool| {
            for _ in 0..VALUES {
                match owned {
                    true => _ = arena.push_string(VALUE.to_string()),
                    false => _ = arena.push_bytes(VALUE.as_bytes()),
                }
            }
        };

        // The bump allocator working as intended, as the baseline to hold the owned path to.
        let copied = RowArena::new();
        push_all(&copied, false);

        let owned = RowArena::new();
        push_all(&owned, true);

        // Seeding with ordinary copies first must not change the answer. The arena never recovers,
        // so this is not just the empty-arena case where the placeholder on top has capacity 0.
        let seeded = RowArena::new();
        let _ = seeded.push_bytes(VALUE.as_bytes());
        push_all(&seeded, true);

        // Compare against the copy path rather than an absolute count, so this pins the property (a
        // run of small owned pushes still ends with a region that has headroom) and leaves the
        // adoption predicate to the fix.
        let (copied, owned, seeded) = (regions(&copied), regions(&owned), regions(&seeded));
        assert!(
            owned <= copied * 2 && seeded <= copied * 2,
            "{VALUES} owned pushes left {owned} regions on an empty arena and {seeded} on a seeded \
             one, against {copied} for the same bytes copied",
        );
    }

    #[mz_ore::test]
    fn miri_test_arena_budget() {
        // Without a budget nothing is ever over it, however much is pushed.
        let arena = RowArena::new();
        let _ = arena.push_bytes(vec![0u8; 1024]);
        assert!(!arena.over_budget());
        assert_eq!(arena.budget_remaining(), usize::MAX);

        let arena = RowArena::with_budget(100);
        assert!(!arena.over_budget());
        assert_eq!(arena.budget_remaining(), 100);

        // Staying within the budget leaves it satisfied, and the remaining count tracks what a
        // caller that predicts its own size would consult.
        let _ = arena.push_bytes(vec![0u8; 60]);
        assert!(!arena.over_budget());
        assert_eq!(arena.budget_remaining(), 40);
        assert_eq!(arena.allocated_bytes(), 60);

        // Crossing it reports, rather than refusing the push: a truncated push would corrupt the
        // datum, so the value is intact and it is the caller's job to fail.
        let pushed = arena.push_bytes(vec![7u8; 80]);
        assert_eq!(pushed, &[7u8; 80]);
        assert!(arena.over_budget());
        assert_eq!(arena.budget_remaining(), 0);

        // An adopted buffer counts against the budget too, or adoption would be a way around it.
        // Large enough to actually be adopted rather than copied.
        let mut arena = RowArena::with_budget(100);
        let _ = arena.push_owned_bytes(vec![0u8; 8 * 1024]);
        assert!(arena.over_budget());

        arena.clear();
        assert!(!arena.over_budget());
        assert_eq!(arena.allocated_bytes(), 0);
    }

    #[mz_ore::test]
    fn miri_test_arena_writer() {
        use std::io::Write;

        let arena = RowArena::new();

        // Build a value incrementally and commit it.
        let mut w = arena.writer();
        let mut expected = Vec::new();
        for i in 0..1000u16 {
            let byte = u8::try_from(i % 256).unwrap();
            w.push(byte);
            expected.push(byte);
            w.extend_from_slice(&[byte, byte]);
            expected.extend_from_slice(&[byte, byte]);
        }
        assert_eq!(w.as_slice(), expected.as_slice());
        assert_eq!(w.len(), expected.len());
        let first = w.finish();
        assert_eq!(first, expected.as_slice());

        // A second writer reuses the scratch; its result is independent of the first, which stays
        // valid because `finish` copied it into the arena.
        let mut w2 = arena.writer();
        write!(w2, "hello").unwrap();
        let second = w2.finish();
        assert_eq!(second, b"hello");
        assert_eq!(first, expected.as_slice());

        // An empty writer commits to an empty slice.
        let empty: &[u8] = &[];
        assert_eq!(arena.writer().finish(), empty);

        // Abandoning a writer without finishing is fine; the next writer starts empty.
        {
            let mut w3 = arena.writer();
            w3.extend_from_slice(b"discarded");
        }
        assert_eq!(arena.writer().as_slice(), empty);
    }

    #[mz_ore::test]
    fn miri_test_arena_writer_nested() {
        // Reentrancy: a writer obtained while another is still live must not panic (no `RefCell`
        // double-borrow) and must not disturb the outer writer. The nested writer just gets its own
        // buffer; the outer one keeps building independently.
        let arena = RowArena::new();

        let mut outer = arena.writer();
        outer.extend_from_slice(b"outer-before-");

        // Take a second writer while `outer` is still live -- the case that double-borrowed before.
        let inner_bytes = {
            let mut inner = arena.writer();
            inner.extend_from_slice(b"inner");
            // The outer writer is unaffected by the nested one.
            assert_eq!(outer.as_slice(), b"outer-before-");
            inner.finish()
        };
        assert_eq!(inner_bytes, b"inner");

        // `outer` is intact and still writable after the nested writer committed.
        outer.extend_from_slice(b"after");
        let outer_bytes = outer.finish();
        assert_eq!(outer_bytes, b"outer-before-after");
        // Both committed slices stay valid and independent.
        assert_eq!(inner_bytes, b"inner");

        // Once all writers have dropped, the recycled buffer is reusable (and cleared on acquire).
        let mut again = arena.writer();
        again.extend_from_slice(b"reused");
        assert_eq!(again.finish(), b"reused");
    }

    #[mz_ore::test]
    fn miri_test_arena_writer_fmt() {
        use std::fmt::Write;

        // Format text into the writer (e.g. building a cast-to-string result) and commit as `&str`.
        let arena = RowArena::new();
        let mut w = arena.writer();
        for i in 0..5 {
            write!(w, "{i},").unwrap();
        }
        assert_eq!(w.finish_str(), "0,1,2,3,4,");
    }

    #[mz_ore::test]
    fn miri_test_round_trip() {
        fn round_trip(datums: Vec<Datum>) {
            let row = Row::pack(datums.clone());

            // When run under miri this catches undefined bytes written to data
            // eg by calling push_copy! on a type which contains undefined padding values
            println!("{:?}", row.data());

            let datums2 = row.iter().collect::<Vec<_>>();
            let datums3 = row.unpack();
            assert_eq!(datums, datums2);
            assert_eq!(datums, datums3);
        }

        round_trip(vec![]);
        round_trip(
            SqlScalarType::enumerate()
                .iter()
                .flat_map(|r#type| r#type.interesting_datums())
                .collect(),
        );
        round_trip(vec![
            Datum::Null,
            Datum::Null,
            Datum::False,
            Datum::True,
            Datum::Int16(-21),
            Datum::Int32(-42),
            Datum::Int64(-2_147_483_648 - 42),
            Datum::UInt8(0),
            Datum::UInt8(1),
            Datum::UInt16(0),
            Datum::UInt16(1),
            Datum::UInt16(1 << 8),
            Datum::UInt32(0),
            Datum::UInt32(1),
            Datum::UInt32(1 << 8),
            Datum::UInt32(1 << 16),
            Datum::UInt32(1 << 24),
            Datum::UInt64(0),
            Datum::UInt64(1),
            Datum::UInt64(1 << 8),
            Datum::UInt64(1 << 16),
            Datum::UInt64(1 << 24),
            Datum::UInt64(1 << 32),
            Datum::UInt64(1 << 40),
            Datum::UInt64(1 << 48),
            Datum::UInt64(1 << 56),
            Datum::Float32(OrderedFloat::from(-42.12)),
            Datum::Float64(OrderedFloat::from(-2_147_483_648.0 - 42.12)),
            Datum::Date(Date::from_pg_epoch(365 * 45 + 21).unwrap()),
            Datum::Timestamp(
                CheckedTimestamp::from_timestamplike(
                    NaiveDate::from_isoywd_opt(2019, 30, chrono::Weekday::Wed)
                        .unwrap()
                        .and_hms_opt(14, 32, 11)
                        .unwrap(),
                )
                .unwrap(),
            ),
            Datum::TimestampTz(
                CheckedTimestamp::from_timestamplike(DateTime::from_timestamp(61, 0).unwrap())
                    .unwrap(),
            ),
            Datum::Interval(Interval {
                months: 312,
                ..Default::default()
            }),
            Datum::Interval(Interval::new(0, 0, 1_012_312)),
            Datum::Bytes(&[]),
            Datum::Bytes(&[0, 2, 1, 255]),
            Datum::String(""),
            Datum::String("العَرَبِيَّة"),
        ]);
    }

    #[mz_ore::test]
    fn test_array() {
        // Construct an array using `Row::push_array` and verify that it unpacks
        // correctly.
        const DIM: ArrayDimension = ArrayDimension {
            lower_bound: 2,
            length: 2,
        };
        let mut row = Row::default();
        let mut packer = row.packer();
        packer
            .try_push_array(&[DIM], vec![Datum::Int32(1), Datum::Int32(2)])
            .unwrap();
        let arr1 = row.unpack_first().unwrap_array();
        assert_eq!(arr1.dims().into_iter().collect::<Vec<_>>(), vec![DIM]);
        assert_eq!(
            arr1.elements().into_iter().collect::<Vec<_>>(),
            vec![Datum::Int32(1), Datum::Int32(2)]
        );

        // Pack a previously-constructed `Datum::Array` and verify that it
        // unpacks correctly.
        let row = Row::pack_slice(&[Datum::Array(arr1)]);
        let arr2 = row.unpack_first().unwrap_array();
        assert_eq!(arr1, arr2);
    }

    #[mz_ore::test]
    fn test_multidimensional_array() {
        let datums = vec![
            Datum::Int32(1),
            Datum::Int32(2),
            Datum::Int32(3),
            Datum::Int32(4),
            Datum::Int32(5),
            Datum::Int32(6),
            Datum::Int32(7),
            Datum::Int32(8),
        ];

        let mut row = Row::default();
        let mut packer = row.packer();
        packer
            .try_push_array(
                &[
                    ArrayDimension {
                        lower_bound: 1,
                        length: 1,
                    },
                    ArrayDimension {
                        lower_bound: 1,
                        length: 4,
                    },
                    ArrayDimension {
                        lower_bound: 1,
                        length: 2,
                    },
                ],
                &datums,
            )
            .unwrap();
        let array = row.unpack_first().unwrap_array();
        assert_eq!(array.elements().into_iter().collect::<Vec<_>>(), datums);
    }

    #[mz_ore::test]
    fn test_array_max_dimensions() {
        let mut row = Row::default();
        let max_dims = usize::from(MAX_ARRAY_DIMENSIONS);

        // An array with one too many dimensions should be rejected.
        let res = row.packer().try_push_array(
            &vec![
                ArrayDimension {
                    lower_bound: 1,
                    length: 1
                };
                max_dims + 1
            ],
            vec![Datum::Int32(4)],
        );
        assert_eq!(res, Err(InvalidArrayError::TooManyDimensions(max_dims + 1)));
        assert!(row.data.is_empty());

        // An array with exactly the maximum allowable dimensions should be
        // accepted.
        row.packer()
            .try_push_array(
                &vec![
                    ArrayDimension {
                        lower_bound: 1,
                        length: 1
                    };
                    max_dims
                ],
                vec![Datum::Int32(4)],
            )
            .unwrap();
    }

    #[mz_ore::test]
    fn test_array_wrong_cardinality() {
        let mut row = Row::default();
        let res = row.packer().try_push_array(
            &[
                ArrayDimension {
                    lower_bound: 1,
                    length: 2,
                },
                ArrayDimension {
                    lower_bound: 1,
                    length: 3,
                },
            ],
            vec![Datum::Int32(1), Datum::Int32(2)],
        );
        assert_eq!(
            res,
            Err(InvalidArrayError::WrongCardinality {
                actual: 2,
                expected: 6,
            })
        );
        assert!(row.data.is_empty());
    }

    #[mz_ore::test]
    fn test_array_cardinality_overflow() {
        // Dimension lengths whose product overflows `usize` must be rejected as
        // a `WrongCardinality` error, not panic (under overflow checks) or wrap
        // (in release, which could spuriously accept a corrupt array). The
        // product saturates to `usize::MAX`, which no real element count matches.
        let mut row = Row::default();
        let res = row.packer().try_push_array(
            &[
                ArrayDimension {
                    lower_bound: 1,
                    length: usize::MAX,
                },
                ArrayDimension {
                    lower_bound: 1,
                    length: 2,
                },
            ],
            vec![Datum::Int32(1), Datum::Int32(2)],
        );
        assert_eq!(
            res,
            Err(InvalidArrayError::WrongCardinality {
                actual: 2,
                expected: usize::MAX,
            })
        );
        assert!(row.data.is_empty());
    }

    #[mz_ore::test]
    fn test_nesting() {
        let mut row = Row::default();
        row.packer().push_dict_with(|row| {
            row.push(Datum::String("favourites"));
            row.push_list_with(|row| {
                row.push(Datum::String("ice cream"));
                row.push(Datum::String("oreos"));
                row.push(Datum::String("cheesecake"));
            });
            row.push(Datum::String("name"));
            row.push(Datum::String("bob"));
        });

        let mut iter = row.unpack_first().unwrap_map().iter();

        let (k, v) = iter.next().unwrap();
        assert_eq!(k, "favourites");
        assert_eq!(
            v.unwrap_list().iter().collect::<Vec<_>>(),
            vec![
                Datum::String("ice cream"),
                Datum::String("oreos"),
                Datum::String("cheesecake"),
            ]
        );

        let (k, v) = iter.next().unwrap();
        assert_eq!(k, "name");
        assert_eq!(v, Datum::String("bob"));
    }

    #[mz_ore::test]
    fn test_dict_errors() -> Result<(), Box<dyn std::error::Error>> {
        let pack = |ok| {
            let mut row = Row::default();
            row.packer().push_dict_with(|row| {
                if ok {
                    row.push(Datum::String("key"));
                    row.push(Datum::Int32(42));
                    Ok(7)
                } else {
                    Err("fail")
                }
            })?;
            Ok(row)
        };

        assert_eq!(pack(false), Err("fail"));

        let row = pack(true)?;
        let mut dict = row.unpack_first().unwrap_map().iter();
        assert_eq!(dict.next(), Some(("key", Datum::Int32(42))));
        assert_eq!(dict.next(), None);

        Ok(())
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decNumberFromInt32` on OS `linux`
    fn test_datum_sizes() {
        let arena = RowArena::new();

        // Test the claims about various datum sizes.
        let values_of_interest = vec![
            Datum::Null,
            Datum::False,
            Datum::Int16(0),
            Datum::Int32(0),
            Datum::Int64(0),
            Datum::UInt8(0),
            Datum::UInt8(1),
            Datum::UInt16(0),
            Datum::UInt16(1),
            Datum::UInt16(1 << 8),
            Datum::UInt32(0),
            Datum::UInt32(1),
            Datum::UInt32(1 << 8),
            Datum::UInt32(1 << 16),
            Datum::UInt32(1 << 24),
            Datum::UInt64(0),
            Datum::UInt64(1),
            Datum::UInt64(1 << 8),
            Datum::UInt64(1 << 16),
            Datum::UInt64(1 << 24),
            Datum::UInt64(1 << 32),
            Datum::UInt64(1 << 40),
            Datum::UInt64(1 << 48),
            Datum::UInt64(1 << 56),
            Datum::Float32(OrderedFloat(0.0)),
            Datum::Float64(OrderedFloat(0.0)),
            Datum::from(numeric::Numeric::from(0)),
            Datum::from(numeric::Numeric::from(1000)),
            Datum::from(numeric::Numeric::from(9999)),
            Datum::Date(
                NaiveDate::from_ymd_opt(1, 1, 1)
                    .unwrap()
                    .try_into()
                    .unwrap(),
            ),
            Datum::Timestamp(
                CheckedTimestamp::from_timestamplike(
                    DateTime::from_timestamp(0, 0).unwrap().naive_utc(),
                )
                .unwrap(),
            ),
            Datum::TimestampTz(
                CheckedTimestamp::from_timestamplike(DateTime::from_timestamp(0, 0).unwrap())
                    .unwrap(),
            ),
            Datum::Interval(Interval::default()),
            Datum::Bytes(&[]),
            Datum::String(""),
            Datum::JsonNull,
            Datum::Range(Range { inner: None }),
            arena.make_datum(|packer| {
                packer
                    .push_range(Range::new(Some((
                        RangeLowerBound::new(Datum::Int32(-1), true),
                        RangeUpperBound::new(Datum::Int32(1), true),
                    ))))
                    .unwrap();
            }),
        ];
        for value in values_of_interest {
            if datum_size(&value) != Row::pack_slice(&[value]).data.len() {
                panic!("Disparity in claimed size for {:?}", value);
            }
        }
    }

    #[mz_ore::test]
    fn test_range_errors() {
        fn test_range_errors_inner<'a>(
            datums: Vec<Vec<Datum<'a>>>,
        ) -> Result<(), InvalidRangeError> {
            let mut row = Row::default();
            let row_len = row.byte_len();
            let mut packer = row.packer();
            let r = packer.push_range_with(
                RangeLowerBound {
                    inclusive: true,
                    bound: Some(|row: &mut RowPacker| {
                        for d in &datums[0] {
                            row.push(d);
                        }
                        Ok(())
                    }),
                },
                RangeUpperBound {
                    inclusive: true,
                    bound: Some(|row: &mut RowPacker| {
                        for d in &datums[1] {
                            row.push(d);
                        }
                        Ok(())
                    }),
                },
            );

            assert_eq!(row_len, row.byte_len());

            r
        }

        // A finite bound whose closure pushes zero values violates the
        // `push_range_with` caller contract and still panics. This is
        // unreachable when decoding a `ProtoRow`: each decoded bound pushes
        // exactly one datum (or fails), so only an in-process caller can hit it.
        for panicking_case in [
            vec![vec![Datum::Int32(1)], vec![]],
            vec![vec![Datum::Int32(1), Datum::Int32(2)], vec![]],
        ] {
            #[allow(clippy::disallowed_methods)] // not using enhanced panic handler in tests
            let result = std::panic::catch_unwind(|| test_range_errors_inner(panicking_case));
            assert_err!(result);
        }

        // Inconsistent bound counts, mismatched datum kinds, and Null bounds are
        // all reachable from a crafted/corrupted `ProtoRow`, so they return an
        // error instead of panicking.
        for error_case in [
            vec![
                vec![Datum::Int32(1), Datum::Int32(2)],
                vec![Datum::Int32(3)],
            ],
            vec![
                vec![Datum::Int32(1)],
                vec![Datum::Int32(2), Datum::Int32(3)],
            ],
            vec![vec![Datum::Int32(1)], vec![Datum::UInt16(2)]],
            vec![vec![Datum::Null], vec![Datum::Int32(2)]],
            vec![vec![Datum::Int32(1)], vec![Datum::Null]],
        ] {
            assert_eq!(
                test_range_errors_inner(error_case),
                Err(InvalidRangeError::InvalidRangeData)
            );
        }

        let e = test_range_errors_inner(vec![vec![Datum::Int32(2)], vec![Datum::Int32(1)]]);
        assert_eq!(e, Err(InvalidRangeError::MisorderedRangeBounds));
    }

    /// Lists have a variable-length encoding for their lengths. We test each case here.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // slow
    fn test_list_encoding() {
        fn test_list_encoding_inner(len: usize) {
            let list_elem = |i: usize| {
                if i % 2 == 0 {
                    Datum::False
                } else {
                    Datum::True
                }
            };
            let mut row = Row::default();
            {
                // Push some stuff.
                let mut packer = row.packer();
                packer.push(Datum::String("start"));
                packer.push_list_with(|packer| {
                    for i in 0..len {
                        packer.push(list_elem(i));
                    }
                });
                packer.push(Datum::String("end"));
            }
            // Check that we read back exactly what we pushed.
            let mut row_it = row.iter();
            assert_eq!(row_it.next().unwrap(), Datum::String("start"));
            match row_it.next().unwrap() {
                Datum::List(list) => {
                    let mut list_it = list.iter();
                    for i in 0..len {
                        assert_eq!(list_it.next().unwrap(), list_elem(i));
                    }
                    assert_none!(list_it.next());
                }
                _ => panic!("expected Datum::List"),
            }
            assert_eq!(row_it.next().unwrap(), Datum::String("end"));
            assert_none!(row_it.next());
        }

        test_list_encoding_inner(0);
        test_list_encoding_inner(1);
        test_list_encoding_inner(10);
        test_list_encoding_inner(TINY - 1); // tiny
        test_list_encoding_inner(TINY + 1); // short
        test_list_encoding_inner(SHORT + 1); // long

        // The biggest one takes 40 s on my laptop, probably not worth it.
        //test_list_encoding_inner(LONG + 1); // huge
    }

    /// Demonstrates that DatumList's Eq (bytewise) and Ord (datum-by-datum) are now consistent.
    /// A list containing -0.0 and one containing +0.0 have different byte representations
    /// (IEEE 754 distinguishes them), originally Eq says they are not equal. But after
    /// using the new Datum::cmp, Eq says they are equal, which matches what Ord
    /// compares via iter().cmp(other.iter()), and them as equal.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn test_datum_list_eq_ord_consistency() {
        // Build list containing +0.0
        let mut row_pos = Row::default();
        row_pos.packer().push_list_with(|p| {
            p.push(Datum::Float64(OrderedFloat::from(0.0)));
        });
        let list_pos = row_pos.unpack_first().unwrap_list();

        // Build list containing -0.0 (distinct bit pattern from +0.0)
        let mut row_neg = Row::default();
        row_neg.packer().push_list_with(|p| {
            p.push(Datum::Float64(OrderedFloat::from(-0.0)));
        });
        let list_neg = row_neg.unpack_first().unwrap_list();

        // Eq is bytewise: different encodings => not equal
        // This was a bug in the past, so we test it.
        assert_eq!(
            list_pos, list_neg,
            "Eq should see different encodings as equal"
        );

        // Ord is datum-by-datum: -0.0 and +0.0 compare equal as Datums
        assert_eq!(
            list_pos.cmp(&list_neg),
            Ordering::Equal,
            "Ord (datum-by-datum) should see -0.0 and +0.0 as equal"
        );
    }

    /// Demonstrates that DatumMap's derived Eq (bytewise) can make maps with equal keys and
    /// values compare equal when values have different encodings (e.g. -0.0 vs +0.0).
    #[mz_ore::test]
    fn test_datum_map_eq_bytewise_consistency() {
        // Build map {"k": +0.0}
        let mut row_pos = Row::default();
        row_pos.packer().push_dict_with(|p| {
            p.push(Datum::String("k"));
            p.push(Datum::Float64(OrderedFloat::from(0.0)));
        });
        let map_pos = row_pos.unpack_first().unwrap_map();

        // Build map {"k": -0.0}
        let mut row_neg = Row::default();
        row_neg.packer().push_dict_with(|p| {
            p.push(Datum::String("k"));
            p.push(Datum::Float64(OrderedFloat::from(-0.0)));
        });
        let map_neg = row_neg.unpack_first().unwrap_map();

        // Same keys and semantically equal values, but Eq (bytewise) says not equal
        assert_eq!(
            map_pos, map_neg,
            "DatumMap Eq is semantic; -0.0 and +0.0 have different encodings but are equal"
        );
        // Verify they have the same logical content
        let entries_pos: Vec<_> = map_pos.iter().collect();
        let entries_neg: Vec<_> = map_neg.iter().collect();
        assert_eq!(entries_pos.len(), entries_neg.len());
        for ((k1, v1), (k2, v2)) in entries_pos.iter().zip_eq(entries_neg.iter()) {
            assert_eq!(k1, k2);
            assert_eq!(
                v1, v2,
                "Datum-level comparison treats -0.0 and +0.0 as equal"
            );
        }
    }

    /// Hash must agree with Eq: equal lists must have the same hash.
    #[mz_ore::test]
    fn test_datum_list_hash_consistency() {
        // Equal lists (including -0.0 vs +0.0) must hash the same
        let mut row_pos = Row::default();
        row_pos.packer().push_list_with(|p| {
            p.push(Datum::Float64(OrderedFloat::from(0.0)));
        });
        let list_pos = row_pos.unpack_first().unwrap_list();

        let mut row_neg = Row::default();
        row_neg.packer().push_list_with(|p| {
            p.push(Datum::Float64(OrderedFloat::from(-0.0)));
        });
        let list_neg = row_neg.unpack_first().unwrap_list();

        assert_eq!(list_pos, list_neg);
        assert_eq!(
            hash(&list_pos),
            hash(&list_neg),
            "equal lists must have same hash"
        );

        // Unequal lists should have different hashes (with asymptotic probability 1)
        let mut row_a = Row::default();
        row_a.packer().push_list_with(|p| {
            p.push(Datum::Int32(1));
            p.push(Datum::Int32(2));
        });
        let list_a = row_a.unpack_first().unwrap_list();

        let mut row_b = Row::default();
        row_b.packer().push_list_with(|p| {
            p.push(Datum::Int32(1));
            p.push(Datum::Int32(3));
        });
        let list_b = row_b.unpack_first().unwrap_list();

        assert_ne!(list_a, list_b);
        assert_ne!(
            hash(&list_a),
            hash(&list_b),
            "unequal lists must have different hashes"
        );
    }

    /// Ord/PartialOrd for DatumList: less, equal, greater.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn test_datum_list_ordering() {
        let mut row_12 = Row::default();
        row_12.packer().push_list_with(|p| {
            p.push(Datum::Int32(1));
            p.push(Datum::Int32(2));
        });
        let list_12 = row_12.unpack_first().unwrap_list();

        let mut row_13 = Row::default();
        row_13.packer().push_list_with(|p| {
            p.push(Datum::Int32(1));
            p.push(Datum::Int32(3));
        });
        let list_13 = row_13.unpack_first().unwrap_list();

        let mut row_123 = Row::default();
        row_123.packer().push_list_with(|p| {
            p.push(Datum::Int32(1));
            p.push(Datum::Int32(2));
            p.push(Datum::Int32(3));
        });
        let list_123 = row_123.unpack_first().unwrap_list();

        // [1, 2] < [1, 3] due to the second element being different
        assert_eq!(list_12.cmp(&list_13), Ordering::Less);
        assert_eq!(list_13.cmp(&list_12), Ordering::Greater);
        assert_eq!(list_12.cmp(&list_12), Ordering::Equal);
        // shorter prefix compares less
        assert_eq!(list_12.cmp(&list_123), Ordering::Less);
    }

    /// Hash must agree with Eq: equal maps must have the same hash.
    #[mz_ore::test]
    fn test_datum_map_hash_consistency() {
        let mut row_pos = Row::default();
        row_pos.packer().push_dict_with(|p| {
            p.push(Datum::String("x"));
            p.push(Datum::Float64(OrderedFloat::from(0.0)));
        });
        let map_pos = row_pos.unpack_first().unwrap_map();

        let mut row_neg = Row::default();
        row_neg.packer().push_dict_with(|p| {
            p.push(Datum::String("x"));
            p.push(Datum::Float64(OrderedFloat::from(-0.0)));
        });
        let map_neg = row_neg.unpack_first().unwrap_map();

        assert_eq!(map_pos, map_neg);
        assert_eq!(
            hash(&map_pos),
            hash(&map_neg),
            "equal maps must have same hash"
        );

        let mut row_a = Row::default();
        row_a.packer().push_dict_with(|p| {
            p.push(Datum::String("a"));
            p.push(Datum::Int32(1));
        });
        let map_a = row_a.unpack_first().unwrap_map();

        let mut row_b = Row::default();
        row_b.packer().push_dict_with(|p| {
            p.push(Datum::String("a"));
            p.push(Datum::Int32(2));
        });
        let map_b = row_b.unpack_first().unwrap_map();

        assert_ne!(map_a, map_b);
        assert_ne!(
            hash(&map_a),
            hash(&map_b),
            "unequal maps must have different hashes"
        );
    }

    /// Ord/PartialOrd for DatumMap: less, equal, greater (by key then value).
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn test_datum_map_ordering() {
        let mut row_a1 = Row::default();
        row_a1.packer().push_dict_with(|p| {
            p.push(Datum::String("a"));
            p.push(Datum::Int32(1));
        });
        let map_a1 = row_a1.unpack_first().unwrap_map();

        let mut row_a2 = Row::default();
        row_a2.packer().push_dict_with(|p| {
            p.push(Datum::String("a"));
            p.push(Datum::Int32(2));
        });
        let map_a2 = row_a2.unpack_first().unwrap_map();

        let mut row_b1 = Row::default();
        row_b1.packer().push_dict_with(|p| {
            p.push(Datum::String("b"));
            p.push(Datum::Int32(1));
        });
        let map_b1 = row_b1.unpack_first().unwrap_map();

        assert_eq!(map_a1.cmp(&map_a2), Ordering::Less);
        assert_eq!(map_a2.cmp(&map_a1), Ordering::Greater);
        assert_eq!(map_a1.cmp(&map_a1), Ordering::Equal);
        assert_eq!(map_a1.cmp(&map_b1), Ordering::Less); // "a" < "b"
    }

    /// Datum puts Null last in the enum so that nulls sort last (PostgreSQL default).
    /// This ordering is used when comparing DatumList/DatumMap (e.g. jsonb_agg tiebreaker).
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn test_datum_list_and_map_null_sorts_last() {
        // DatumList: [1] < [null] so non-null sorts before null
        let mut row_list_1 = Row::default();
        row_list_1
            .packer()
            .push_list_with(|p| p.push(Datum::Int32(1)));
        let list_1 = row_list_1.unpack_first().unwrap_list();

        let mut row_list_null = Row::default();
        row_list_null
            .packer()
            .push_list_with(|p| p.push(Datum::Null));
        let list_null = row_list_null.unpack_first().unwrap_list();

        assert_eq!(list_1.cmp(&list_null), Ordering::Less);
        assert_eq!(list_null.cmp(&list_1), Ordering::Greater);

        // DatumMap: {"k": 1} < {"k": null} so non-null sorts before null (same as jsonb_agg)
        let mut row_map_1 = Row::default();
        row_map_1.packer().push_dict_with(|p| {
            p.push(Datum::String("k"));
            p.push(Datum::Int32(1));
        });
        let map_1 = row_map_1.unpack_first().unwrap_map();

        let mut row_map_null = Row::default();
        row_map_null.packer().push_dict_with(|p| {
            p.push(Datum::String("k"));
            p.push(Datum::Null);
        });
        let map_null = row_map_null.unpack_first().unwrap_map();

        assert_eq!(map_1.cmp(&map_null), Ordering::Less);
        assert_eq!(map_null.cmp(&map_1), Ordering::Greater);
    }
}