mz_transform/

reprtypecheck.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.

//! Check that the visible type of each query has not been changed

use std::collections::BTreeMap;
use std::fmt::Write;
use std::sync::{Arc, Mutex};

use itertools::Itertools;
use mz_expr::explain::{HumanizedExplain, HumanizerMode};
use mz_expr::{
    AggregateExpr, ColumnOrder, Id, JoinImplementation, LocalId, MirRelationExpr, MirScalarExpr,
    RECURSION_LIMIT, non_nullable_columns,
};
use mz_ore::soft_panic_or_log;
use mz_ore::stack::{CheckedRecursion, RecursionGuard, RecursionLimitError};
use mz_repr::adt::range::Range;
use mz_repr::explain::{DummyHumanizer, ExprHumanizer};
use mz_repr::{
    ColumnName, Datum, ReprColumnType, ReprRelationType, ReprScalarBaseType, ReprScalarType,
    SqlColumnType,
};

/// Typechecking contexts as shared by various typechecking passes.
///
/// We use a `RefCell` to ensure that contexts are shared by multiple typechecker passes.
/// Shared contexts help catch consistency issues.
pub type SharedContext = Arc<Mutex<Context>>;

/// Generates an empty context
pub fn empty_context() -> SharedContext {
    Arc::new(Mutex::new(BTreeMap::new()))
}

/// The possible forms of inconsistency/errors discovered during typechecking.
///
/// Every variant has a `source` field identifying the MIR term that is home
/// to the error (though not necessarily the root cause of the error).
#[derive(Clone, Debug)]
pub enum TypeError<'a> {
    /// Unbound identifiers (local or global)
    Unbound {
        /// Expression with the bug
        source: &'a MirRelationExpr,
        /// The (unbound) identifier referenced
        id: Id,
        /// The type `id` was expected to have
        typ: ReprRelationType,
    },
    /// Dereference of a non-existent column
    NoSuchColumn {
        /// Expression with the bug
        source: &'a MirRelationExpr,
        /// Scalar expression that references an invalid column
        expr: &'a MirScalarExpr,
        /// The invalid column referenced
        col: usize,
    },
    /// A single column type does not match
    MismatchColumn {
        /// Expression with the bug
        source: &'a MirRelationExpr,
        /// The column type we found (`sub` type)
        got: ReprColumnType,
        /// The column type we expected (`sup` type)
        expected: ReprColumnType,
        /// The difference between these types
        diffs: Vec<ReprColumnTypeDifference>,
        /// An explanatory message
        message: String,
    },
    /// Relation column types do not match
    MismatchColumns {
        /// Expression with the bug
        source: &'a MirRelationExpr,
        /// The column types we found (`sub` type)
        got: Vec<ReprColumnType>,
        /// The column types we expected (`sup` type)
        expected: Vec<ReprColumnType>,
        /// The difference between these types
        diffs: Vec<ReprRelationTypeDifference>,
        /// An explanatory message
        message: String,
    },
    /// A constant row does not have the correct type
    BadConstantRowLen {
        /// Expression with the bug
        source: &'a MirRelationExpr,
        /// A constant row
        got: usize,
        /// The expected type (which that row does not have)
        expected: Vec<ReprColumnType>,
    },
    /// A constant row does not have the correct type
    BadConstantRow {
        /// Expression with the bug
        source: &'a MirRelationExpr,
        /// The columns that mismatched, with a debug rendering of the datum we got and the expected type
        mismatches: Vec<(usize, DatumTypeDifference)>,
        /// The expected type (which that row does not have)
        expected: Vec<ReprColumnType>,
        // TODO(mgree) with a good way to get the type of a Datum, we could give a detailed diff here
        // this is difficult for empty structures where we may not know the element type
    },
    /// Projection of a non-existent column
    BadProject {
        /// Expression with the bug
        source: &'a MirRelationExpr,
        /// The column projected
        got: Vec<usize>,
        /// The input columns (which don't have that column)
        input_type: Vec<ReprColumnType>,
    },
    /// An equivalence class in a join was malformed
    BadJoinEquivalence {
        /// Expression with the bug
        source: &'a MirRelationExpr,
        /// The join equivalences
        got: Vec<ReprColumnType>,
        /// The problem with the join equivalences
        message: String,
    },
    /// TopK grouping by non-existent column
    BadTopKGroupKey {
        /// Expression with the bug
        source: &'a MirRelationExpr,
        /// The bad column reference in the group key
        k: usize,
        /// The input columns (which don't have that column)
        input_type: Vec<ReprColumnType>,
    },
    /// TopK ordering by non-existent column
    BadTopKOrdering {
        /// Expression with the bug
        source: &'a MirRelationExpr,
        /// The ordering used
        order: ColumnOrder,
        /// The input columns (which don't work for that ordering)
        input_type: Vec<ReprColumnType>,
    },
    /// LetRec bindings are malformed
    BadLetRecBindings {
        /// Expression with the bug
        source: &'a MirRelationExpr,
    },
    /// Local identifiers are shadowed
    Shadowing {
        /// Expression with the bug
        source: &'a MirRelationExpr,
        /// The id that was shadowed
        id: Id,
    },
    /// Recursion depth exceeded
    Recursion {
        /// The error that aborted recursion
        error: RecursionLimitError,
    },
    /// A dummy value was found
    DisallowedDummy {
        /// The expression with the dummy value
        source: &'a MirRelationExpr,
    },
}

impl<'a> From<RecursionLimitError> for TypeError<'a> {
    fn from(error: RecursionLimitError) -> Self {
        TypeError::Recursion { error }
    }
}

type Context = BTreeMap<Id, Vec<ReprColumnType>>;

/// Characterizes differences between relation types
///
/// Each constructor indicates a reason why some type `sub` was not a subtype of another type `sup`
#[derive(Clone, Debug, Hash)]
pub enum ReprRelationTypeDifference {
    /// `sub` and `sup` don't have the same number of columns
    Length {
        /// Length of `sub`
        len_sub: usize,
        /// Length of `sup`
        len_sup: usize,
    },
    /// `sub` and `sup` differ at the indicated column
    Column {
        /// The column at which `sub` and `sup` differ
        col: usize,
        /// The difference between `sub` and `sup`
        diff: ReprColumnTypeDifference,
    },
}

/// Characterizes differences between individual column types
///
/// Each constructor indicates a reason why some type `sub` was not a subtype of another type `sup`
/// There may be multiple reasons, e.g., `sub` may be missing fields and have fields of different types
#[derive(Clone, Debug, Hash)]
pub enum ReprColumnTypeDifference {
    /// The `ReprScalarBaseType` of `sub` doesn't match that of `sup`
    NotSubtype {
        /// Would-be subtype
        sub: ReprScalarType,
        /// Would-be supertype
        sup: ReprScalarType,
    },
    /// `sub` was nullable but `sup` was not
    Nullability {
        /// Would-be subtype
        sub: ReprColumnType,
        /// Would-be supertype
        sup: ReprColumnType,
    },
    /// Both `sub` and `sup` are a list, map, array, or range, but `sub`'s element type differed from `sup`s
    ElementType {
        /// The type constructor (list, array, etc.)
        ctor: String,
        /// The difference in the element type
        element_type: Box<ReprColumnTypeDifference>,
    },
    /// `sub` and `sup` are both records, but `sub` is missing fields present in `sup`
    RecordMissingFields {
        /// The missing fields
        missing: Vec<ColumnName>,
    },
    /// `sub` and `sup` are both records, but some fields in `sub` are not subtypes of fields in `sup`
    RecordFields {
        /// The differences, by field
        fields: Vec<ReprColumnTypeDifference>,
    },
}

impl ReprRelationTypeDifference {
    /// Returns the same type difference, but ignoring nullability
    ///
    /// Returns `None` when _all_ of the differences are due to nullability
    pub fn ignore_nullability(self) -> Option<Self> {
        use ReprRelationTypeDifference::*;

        match self {
            Length { .. } => Some(self),
            Column { col, diff } => diff.ignore_nullability().map(|diff| Column { col, diff }),
        }
    }
}

impl ReprColumnTypeDifference {
    /// Returns the same type difference, but ignoring nullability
    ///
    /// Returns `None` when _all_ of the differences are due to nullability
    pub fn ignore_nullability(self) -> Option<Self> {
        use ReprColumnTypeDifference::*;

        match self {
            Nullability { .. } => None,
            NotSubtype { .. } | RecordMissingFields { .. } => Some(self),
            ElementType { ctor, element_type } => {
                element_type
                    .ignore_nullability()
                    .map(|element_type| ElementType {
                        ctor,
                        element_type: Box::new(element_type),
                    })
            }
            RecordFields { fields } => {
                let fields = fields
                    .into_iter()
                    .flat_map(|diff| diff.ignore_nullability())
                    .collect::<Vec<_>>();

                if fields.is_empty() {
                    None
                } else {
                    Some(RecordFields { fields })
                }
            }
        }
    }
}

/// Returns a list of differences that make `sub` not a subtype of `sup`
///
/// This function returns an empty list when `sub` is a subtype of `sup`
pub fn relation_subtype_difference(
    sub: &[ReprColumnType],
    sup: &[ReprColumnType],
) -> Vec<ReprRelationTypeDifference> {
    let mut diffs = Vec::new();

    if sub.len() != sup.len() {
        diffs.push(ReprRelationTypeDifference::Length {
            len_sub: sub.len(),
            len_sup: sup.len(),
        });

        // TODO(mgree) we could do an edit-distance computation to report more errors
        return diffs;
    }

    diffs.extend(
        sub.iter()
            .zip_eq(sup.iter())
            .enumerate()
            .flat_map(|(col, (sub_ty, sup_ty))| {
                column_subtype_difference(sub_ty, sup_ty)
                    .into_iter()
                    .map(move |diff| ReprRelationTypeDifference::Column { col, diff })
            }),
    );

    diffs
}

/// Returns a list of differences that make `sub` not a subtype of `sup`
///
/// This function returns an empty list when `sub` is a subtype of `sup`
pub fn column_subtype_difference(
    sub: &ReprColumnType,
    sup: &ReprColumnType,
) -> Vec<ReprColumnTypeDifference> {
    let mut diffs = scalar_subtype_difference(&sub.scalar_type, &sup.scalar_type);

    if sub.nullable && !sup.nullable {
        diffs.push(ReprColumnTypeDifference::Nullability {
            sub: sub.clone(),
            sup: sup.clone(),
        });
    }

    diffs
}

/// Returns a list of differences that make `sub` not a subtype of `sup`
///
/// This function returns an empty list when `sub` is a subtype of `sup`
pub fn scalar_subtype_difference(
    sub: &ReprScalarType,
    sup: &ReprScalarType,
) -> Vec<ReprColumnTypeDifference> {
    use ReprScalarType::*;

    let mut diffs = Vec::new();

    match (sub, sup) {
        (
            List {
                element_type: sub_elt,
                ..
            },
            List {
                element_type: sup_elt,
                ..
            },
        )
        | (
            Map {
                value_type: sub_elt,
                ..
            },
            Map {
                value_type: sup_elt,
                ..
            },
        )
        | (
            Range {
                element_type: sub_elt,
                ..
            },
            Range {
                element_type: sup_elt,
                ..
            },
        )
        | (Array(sub_elt), Array(sup_elt)) => {
            let ctor = format!("{:?}", ReprScalarBaseType::from(sub));
            diffs.extend(
                scalar_subtype_difference(sub_elt, sup_elt)
                    .into_iter()
                    .map(|diff| ReprColumnTypeDifference::ElementType {
                        ctor: ctor.clone(),
                        element_type: Box::new(diff),
                    }),
            );
        }
        (
            Record {
                fields: sub_fields, ..
            },
            Record {
                fields: sup_fields, ..
            },
        ) => {
            if sub_fields.len() != sup_fields.len() {
                diffs.push(ReprColumnTypeDifference::NotSubtype {
                    sub: sub.clone(),
                    sup: sup.clone(),
                });
                return diffs;
            }

            for (sub_ty, sup_ty) in sub_fields.iter().zip_eq(sup_fields.iter()) {
                diffs.extend(column_subtype_difference(sub_ty, sup_ty));
            }
        }
        (_, _) => {
            if ReprScalarBaseType::from(sub) != ReprScalarBaseType::from(sup) {
                diffs.push(ReprColumnTypeDifference::NotSubtype {
                    sub: sub.clone(),
                    sup: sup.clone(),
                })
            }
        }
    };

    diffs
}

/// Unions `other` into `typ`, returning a list of differences on failure
///
/// This function returns an empty list when `typ` and `other` are a union
pub fn scalar_union(
    typ: &mut ReprScalarType,
    other: &ReprScalarType,
) -> Vec<ReprColumnTypeDifference> {
    use ReprScalarType::*;

    let mut diffs = Vec::new();

    // precomputing to appease the borrow checker
    let ctor = ReprScalarBaseType::from(&*typ);
    match (typ, other) {
        (
            List {
                element_type: typ_elt,
            },
            List {
                element_type: other_elt,
            },
        )
        | (
            Map {
                value_type: typ_elt,
            },
            Map {
                value_type: other_elt,
            },
        )
        | (
            Range {
                element_type: typ_elt,
            },
            Range {
                element_type: other_elt,
            },
        )
        | (Array(typ_elt), Array(other_elt)) => {
            let res = scalar_union(typ_elt.as_mut(), other_elt.as_ref());
            diffs.extend(
                res.into_iter()
                    .map(|diff| ReprColumnTypeDifference::ElementType {
                        ctor: format!("{ctor:?}"),
                        element_type: Box::new(diff),
                    }),
            );
        }
        (
            Record { fields: typ_fields },
            Record {
                fields: other_fields,
            },
        ) => {
            if typ_fields.len() != other_fields.len() {
                diffs.push(ReprColumnTypeDifference::NotSubtype {
                    sub: ReprScalarType::Record {
                        fields: typ_fields.clone(),
                    },
                    sup: other.clone(),
                });
                return diffs;
            }

            for (typ_ty, other_ty) in typ_fields.iter_mut().zip_eq(other_fields.iter()) {
                diffs.extend(column_union(typ_ty, other_ty));
            }
        }
        (typ, _) => {
            if ctor != ReprScalarBaseType::from(other) {
                diffs.push(ReprColumnTypeDifference::NotSubtype {
                    sub: typ.clone(),
                    sup: other.clone(),
                })
            }
        }
    };

    diffs
}

/// Unions `other` into `typ`, returning a list of differences on failure
///
/// This function returns an empty list when `typ` and `other` are a union
pub fn column_union(
    typ: &mut ReprColumnType,
    other: &ReprColumnType,
) -> Vec<ReprColumnTypeDifference> {
    let diffs = scalar_union(&mut typ.scalar_type, &other.scalar_type);

    if diffs.is_empty() {
        typ.nullable |= other.nullable;
    }

    diffs
}

/// Returns true when it is safe to treat a `sub` row as an `sup` row
///
/// In particular, the core types must be equal, and if a column in `sup` is nullable, that column should also be nullable in `sub`
/// Conversely, it is okay to treat a known non-nullable column as nullable: `sub` may be nullable when `sup` is not
pub fn is_subtype_of(sub: &[ReprColumnType], sup: &[ReprColumnType]) -> bool {
    if sub.len() != sup.len() {
        return false;
    }

    sub.iter().zip_eq(sup.iter()).all(|(got, known)| {
        (!known.nullable || got.nullable) && got.scalar_type == known.scalar_type
    })
}

/// Characterizes how a Datum differs from a ReprColumnType
#[derive(Clone, Debug)]
pub enum DatumTypeDifference {
    /// Datum was null, but expected a non-null value (i.e., a ReprScalarType)
    Null,
    /// Datum's kind doesn't match the expected ReprScalarType
    Mismatch {
        /// The debug rendering of the Datum that we got (we don't store the Datum itself to avoid borrowing issues)
        /// This debug rendering includes the DatumKind.
        got_debug: String,
        /// The representation scalar type that we expected
        expected: ReprScalarType,
    },
    /// Datum's dimensions didn't match the type's dimensions
    MismatchDimensions {
        /// The type constructor (list, array, int2vector, etc.)
        ctor: String,
        /// The dimension of the datum that we got
        got: usize,
        /// The dimension of the type that we expected
        expected: usize,
    },
    /// There was a nested difference in the element type of some container type
    ElementType {
        /// The type constructor (list, array, int2vector, etc.)
        ctor: String,
        /// The difference in the element type
        element_type: Box<DatumTypeDifference>,
    },
}

/// Computes the difference between this datum and the specified (representation) column type.
///
/// See [`Datum<'a>::is_instance_of`] for just getting a yes/no answer.
///
/// See [`Datum<'a>::is_instance_of_sql`] for comparing `Datum`s to `SqlColumnType`s.
fn datum_difference_with_column_type(
    datum: &Datum<'_>,
    column_type: &ReprColumnType,
) -> Result<(), DatumTypeDifference> {
    fn difference_with_scalar_type(
        datum: &Datum<'_>,
        scalar_type: &ReprScalarType,
    ) -> Result<(), DatumTypeDifference> {
        fn mismatch(got: &Datum<'_>, expected: &ReprScalarType) -> Result<(), DatumTypeDifference> {
            Err(DatumTypeDifference::Mismatch {
                // this will be redacted as appropriate
                got_debug: format!("{got:?}"),
                expected: expected.clone(),
            })
        }

        if let ReprScalarType::Jsonb = scalar_type {
            // json type checking
            match datum {
                Datum::Dummy => Ok(()), // allow dummys (unlike is_instance_of)
                Datum::Null => Err(DatumTypeDifference::Null),
                Datum::JsonNull
                | Datum::False
                | Datum::True
                | Datum::Numeric(_)
                | Datum::String(_) => Ok(()),
                Datum::List(list) => {
                    for elem in list.iter() {
                        difference_with_scalar_type(&elem, scalar_type)?;
                    }
                    Ok(())
                }
                Datum::Map(dict) => {
                    for (_, val) in dict.iter() {
                        difference_with_scalar_type(&val, scalar_type)?;
                    }
                    Ok(())
                }
                _ => mismatch(datum, scalar_type),
            }
        } else {
            fn element_type_difference(
                ctor: &str,
                element_type: DatumTypeDifference,
            ) -> DatumTypeDifference {
                DatumTypeDifference::ElementType {
                    ctor: ctor.to_string(),
                    element_type: Box::new(element_type),
                }
            }
            match (datum, scalar_type) {
                (Datum::Dummy, _) => Ok(()), // allow dummys (unlike is_instance_of)
                (Datum::Null, _) => Err(DatumTypeDifference::Null),
                (Datum::False, ReprScalarType::Bool) => Ok(()),
                (Datum::False, _) => mismatch(datum, scalar_type),
                (Datum::True, ReprScalarType::Bool) => Ok(()),
                (Datum::True, _) => mismatch(datum, scalar_type),
                (Datum::Int16(_), ReprScalarType::Int16) => Ok(()),
                (Datum::Int16(_), _) => mismatch(datum, scalar_type),
                (Datum::Int32(_), ReprScalarType::Int32) => Ok(()),
                (Datum::Int32(_), _) => mismatch(datum, scalar_type),
                (Datum::Int64(_), ReprScalarType::Int64) => Ok(()),
                (Datum::Int64(_), _) => mismatch(datum, scalar_type),
                (Datum::UInt8(_), ReprScalarType::UInt8) => Ok(()),
                (Datum::UInt8(_), _) => mismatch(datum, scalar_type),
                (Datum::UInt16(_), ReprScalarType::UInt16) => Ok(()),
                (Datum::UInt16(_), _) => mismatch(datum, scalar_type),
                (Datum::UInt32(_), ReprScalarType::UInt32) => Ok(()),
                (Datum::UInt32(_), _) => mismatch(datum, scalar_type),
                (Datum::UInt64(_), ReprScalarType::UInt64) => Ok(()),
                (Datum::UInt64(_), _) => mismatch(datum, scalar_type),
                (Datum::Float32(_), ReprScalarType::Float32) => Ok(()),
                (Datum::Float32(_), _) => mismatch(datum, scalar_type),
                (Datum::Float64(_), ReprScalarType::Float64) => Ok(()),
                (Datum::Float64(_), _) => mismatch(datum, scalar_type),
                (Datum::Date(_), ReprScalarType::Date) => Ok(()),
                (Datum::Date(_), _) => mismatch(datum, scalar_type),
                (Datum::Time(_), ReprScalarType::Time) => Ok(()),
                (Datum::Time(_), _) => mismatch(datum, scalar_type),
                (Datum::Timestamp(_), ReprScalarType::Timestamp { .. }) => Ok(()),
                (Datum::Timestamp(_), _) => mismatch(datum, scalar_type),
                (Datum::TimestampTz(_), ReprScalarType::TimestampTz { .. }) => Ok(()),
                (Datum::TimestampTz(_), _) => mismatch(datum, scalar_type),
                (Datum::Interval(_), ReprScalarType::Interval) => Ok(()),
                (Datum::Interval(_), _) => mismatch(datum, scalar_type),
                (Datum::Bytes(_), ReprScalarType::Bytes) => Ok(()),
                (Datum::Bytes(_), _) => mismatch(datum, scalar_type),
                (Datum::String(_), ReprScalarType::String) => Ok(()),
                (Datum::String(_), _) => mismatch(datum, scalar_type),
                (Datum::Uuid(_), ReprScalarType::Uuid) => Ok(()),
                (Datum::Uuid(_), _) => mismatch(datum, scalar_type),
                (Datum::Array(array), ReprScalarType::Array(t)) => {
                    for e in array.elements().iter() {
                        if let Datum::Null = e {
                            continue;
                        }

                        difference_with_scalar_type(&e, t)
                            .map_err(|e| element_type_difference("array", e))?;
                    }
                    Ok(())
                }
                (Datum::Array(array), ReprScalarType::Int2Vector) => {
                    if array.dims().len() != 1 {
                        return Err(DatumTypeDifference::MismatchDimensions {
                            ctor: "int2vector".to_string(),
                            got: array.dims().len(),
                            expected: 1,
                        });
                    }

                    for e in array.elements().iter() {
                        difference_with_scalar_type(&e, &ReprScalarType::Int16)
                            .map_err(|e| element_type_difference("int2vector", e))?;
                    }

                    Ok(())
                }
                (Datum::Array(_), _) => mismatch(datum, scalar_type),
                (Datum::List(list), ReprScalarType::List { element_type, .. }) => {
                    for e in list.iter() {
                        if let Datum::Null = e {
                            continue;
                        }

                        difference_with_scalar_type(&e, element_type)
                            .map_err(|e| element_type_difference("list", e))?;
                    }
                    Ok(())
                }
                (Datum::List(list), ReprScalarType::Record { fields, .. }) => {
                    let len = list.iter().count();
                    if len != fields.len() {
                        return Err(DatumTypeDifference::MismatchDimensions {
                            ctor: "record".to_string(),
                            got: len,
                            expected: fields.len(),
                        });
                    }

                    for (e, t) in list.iter().zip_eq(fields) {
                        if let Datum::Null = e {
                            continue;
                        }

                        difference_with_scalar_type(&e, &t.scalar_type)
                            .map_err(|e| element_type_difference("record", e))?;
                    }
                    Ok(())
                }
                (Datum::List(_), _) => mismatch(datum, scalar_type),
                (Datum::Map(map), ReprScalarType::Map { value_type, .. }) => {
                    for (_, v) in map.iter() {
                        if let Datum::Null = v {
                            continue;
                        }

                        difference_with_scalar_type(&v, value_type)
                            .map_err(|e| element_type_difference("map", e))?;
                    }
                    Ok(())
                }
                (Datum::Map(_), _) => mismatch(datum, scalar_type),
                (Datum::JsonNull, _) => mismatch(datum, scalar_type),
                (Datum::Numeric(_), ReprScalarType::Numeric) => Ok(()),
                (Datum::Numeric(_), _) => mismatch(datum, scalar_type),
                (Datum::MzTimestamp(_), ReprScalarType::MzTimestamp) => Ok(()),
                (Datum::MzTimestamp(_), _) => mismatch(datum, scalar_type),
                (Datum::Range(Range { inner }), ReprScalarType::Range { element_type }) => {
                    match inner {
                        None => Ok(()),
                        Some(inner) => {
                            if let Some(b) = inner.lower.bound {
                                difference_with_scalar_type(&b.datum(), element_type)
                                    .map_err(|e| element_type_difference("range", e))?;
                            }
                            if let Some(b) = inner.upper.bound {
                                difference_with_scalar_type(&b.datum(), element_type)
                                    .map_err(|e| element_type_difference("range", e))?;
                            }
                            Ok(())
                        }
                    }
                }
                (Datum::Range(_), _) => mismatch(datum, scalar_type),
                (Datum::MzAclItem(_), ReprScalarType::MzAclItem) => Ok(()),
                (Datum::MzAclItem(_), _) => mismatch(datum, scalar_type),
                (Datum::AclItem(_), ReprScalarType::AclItem) => Ok(()),
                (Datum::AclItem(_), _) => mismatch(datum, scalar_type),
            }
        }
    }
    if column_type.nullable {
        if let Datum::Null = datum {
            return Ok(());
        }
    }
    difference_with_scalar_type(datum, &column_type.scalar_type)
}

fn row_difference_with_column_types<'a>(
    source: &'a MirRelationExpr,
    datums: &Vec<Datum<'_>>,
    column_types: &[ReprColumnType],
) -> Result<(), TypeError<'a>> {
    // correct length
    if datums.len() != column_types.len() {
        return Err(TypeError::BadConstantRowLen {
            source,
            got: datums.len(),
            expected: column_types.to_vec(),
        });
    }

    // correct types
    let mut mismatches = Vec::new();
    for (i, (d, ty)) in datums.iter().zip_eq(column_types.iter()).enumerate() {
        if let Err(e) = datum_difference_with_column_type(d, ty) {
            mismatches.push((i, e));
        }
    }
    if !mismatches.is_empty() {
        return Err(TypeError::BadConstantRow {
            source,
            mismatches,
            expected: column_types.to_vec(),
        });
    }

    Ok(())
}
/// Check that the visible type of each query has not been changed
#[derive(Debug)]
pub struct Typecheck {
    /// The known types of the queries so far
    ctx: SharedContext,
    /// Whether or not this is the first run of the transform
    disallow_new_globals: bool,
    /// Whether or not to be strict about join equivalences having the same nullability
    strict_join_equivalences: bool,
    /// Whether or not to disallow dummy values
    disallow_dummy: bool,
    /// Recursion guard for checked recursion
    recursion_guard: RecursionGuard,
}

impl CheckedRecursion for Typecheck {
    fn recursion_guard(&self) -> &RecursionGuard {
        &self.recursion_guard
    }
}

impl Typecheck {
    /// Creates a typechecking consistency checking pass using a given shared context
    pub fn new(ctx: SharedContext) -> Self {
        Self {
            ctx,
            disallow_new_globals: false,
            strict_join_equivalences: false,
            disallow_dummy: false,
            recursion_guard: RecursionGuard::with_limit(RECURSION_LIMIT),
        }
    }

    /// New non-transient global IDs will be treated as an error
    ///
    /// Only turn this on after the context has been appropriately populated by, e.g., an earlier run
    pub fn disallow_new_globals(mut self) -> Self {
        self.disallow_new_globals = true;
        self
    }

    /// Equivalence classes in joins must not only agree on scalar type, but also on nullability
    ///
    /// Only turn this on before `JoinImplementation`
    pub fn strict_join_equivalences(mut self) -> Self {
        self.strict_join_equivalences = true;

        self
    }

    /// Disallow dummy values
    pub fn disallow_dummy(mut self) -> Self {
        self.disallow_dummy = true;
        self
    }

    /// Returns the type of a relation expression or a type error.
    ///
    /// This function is careful to check validity, not just find out the type.
    ///
    /// It should be linear in the size of the AST.
    ///
    /// ??? should we also compute keys and return a `ReprRelationType`?
    ///   ggevay: Checking keys would have the same problem as checking nullability: key inference
    ///   is very heuristic (even more so than nullability inference), so it's almost impossible to
    ///   reliably keep it stable across transformations.
    pub fn typecheck<'a>(
        &self,
        expr: &'a MirRelationExpr,
        ctx: &Context,
    ) -> Result<Vec<ReprColumnType>, TypeError<'a>> {
        use MirRelationExpr::*;

        self.checked_recur(|tc| match expr {
            Constant { typ, rows } => {
                if let Ok(rows) = rows {
                    for (row, _id) in rows {
                        let datums = row.unpack();

                        row_difference_with_column_types(expr, &datums, &typ.column_types.iter().map(ReprColumnType::from).collect_vec())?;

                        if self.disallow_dummy && datums.iter().any(|d| d == &mz_repr::Datum::Dummy) {
                            return Err(TypeError::DisallowedDummy {
                                source: expr,
                            });
                        }
                    }
                }

                Ok(typ.column_types.iter().map(ReprColumnType::from).collect_vec())
            }
            Get { typ, id, .. } => {
                if let Id::Global(_global_id) = id {
                    if !ctx.contains_key(id) {
                        // TODO(mgree) pass QueryContext through to check these types
                        return Ok(typ.column_types.iter().map(ReprColumnType::from).collect_vec());
                    }
                }

                let ctx_typ = ctx.get(id).ok_or_else(|| TypeError::Unbound {
                    source: expr,
                    id: id.clone(),
                    typ: ReprRelationType::from(typ),
                })?;

                let column_types = typ.column_types.iter().map(ReprColumnType::from).collect_vec();

                // covariant: the ascribed type must be a subtype of the actual type in the context
                let diffs = relation_subtype_difference(&column_types, ctx_typ).into_iter().flat_map(|diff| diff.ignore_nullability()).collect::<Vec<_>>();

                if !diffs.is_empty() {
                    return Err(TypeError::MismatchColumns {
                        source: expr,
                        got: column_types,
                        expected: ctx_typ.clone(),
                        diffs,
                        message: "annotation did not match context type".to_string(),
                    });
                }

                Ok(column_types)
            }
            Project { input, outputs } => {
                let t_in = tc.typecheck(input, ctx)?;

                for x in outputs {
                    if *x >= t_in.len() {
                        return Err(TypeError::BadProject {
                            source: expr,
                            got: outputs.clone(),
                            input_type: t_in,
                        });
                    }
                }

                Ok(outputs.iter().map(|col| t_in[*col].clone()).collect())
            }
            Map { input, scalars } => {
                let mut t_in = tc.typecheck(input, ctx)?;

                for scalar_expr in scalars.iter() {
                    t_in.push(tc.typecheck_scalar(scalar_expr, expr, &t_in)?);

                    if self.disallow_dummy && scalar_expr.contains_dummy() {
                        return Err(TypeError::DisallowedDummy {
                            source: expr,
                        });
                    }
                }

                Ok(t_in)
            }
            FlatMap { input, func, exprs } => {
                let mut t_in = tc.typecheck(input, ctx)?;

                let mut t_exprs = Vec::with_capacity(exprs.len());
                for scalar_expr in exprs {
                    t_exprs.push(tc.typecheck_scalar(scalar_expr, expr, &t_in)?);

                    if self.disallow_dummy && scalar_expr.contains_dummy() {
                        return Err(TypeError::DisallowedDummy {
                            source: expr,
                        });
                    }
                }
                // TODO(mgree) check t_exprs agrees with `func`'s input type

                let t_out = func.output_type().column_types.iter().map(ReprColumnType::from).collect_vec();

                // FlatMap extends the existing columns
                t_in.extend(t_out);
                Ok(t_in)
            }
            Filter { input, predicates } => {
                let mut t_in = tc.typecheck(input, ctx)?;

                // Set as nonnull any columns where null values would cause
                // any predicate to evaluate to null.
                for column in non_nullable_columns(predicates) {
                    t_in[column].nullable = false;
                }

                for scalar_expr in predicates {
                    let t = tc.typecheck_scalar(scalar_expr, expr, &t_in)?;

                    // filter condition must be boolean
                    // ignoring nullability: null is treated as false
                    // NB this behavior is slightly different from columns_match (for which we would set nullable to false in the expected type)
                    if t.scalar_type != ReprScalarType::Bool {
                        let sub = t.scalar_type.clone();

                        return Err(TypeError::MismatchColumn {
                            source: expr,
                            got: t,
                            expected: ReprColumnType {
                                scalar_type: ReprScalarType::Bool,
                                nullable: true,
                            },
                            diffs: vec![ReprColumnTypeDifference::NotSubtype { sub, sup: ReprScalarType::Bool }],
                            message: "expected boolean condition".to_string(),
                        });
                    }

                    if self.disallow_dummy && scalar_expr.contains_dummy() {
                        return Err(TypeError::DisallowedDummy {
                            source: expr,
                        });
                    }
                }

                Ok(t_in)
            }
            Join {
                inputs,
                equivalences,
                implementation,
            } => {
                let mut t_in_global = Vec::new();
                let mut t_in_local = vec![Vec::new(); inputs.len()];

                for (i, input) in inputs.iter().enumerate() {
                    let input_t = tc.typecheck(input, ctx)?;
                    t_in_global.extend(input_t.clone());
                    t_in_local[i] = input_t;
                }

                for eq_class in equivalences {
                    let mut t_exprs: Vec<ReprColumnType> = Vec::with_capacity(eq_class.len());

                    let mut all_nullable = true;

                    for scalar_expr in eq_class {
                        // Note: the equivalences have global column references
                        let t_expr = tc.typecheck_scalar(scalar_expr, expr, &t_in_global)?;

                        if !t_expr.nullable {
                            all_nullable = false;
                        }

                        if let Some(t_first) = t_exprs.get(0) {
                            let diffs = scalar_subtype_difference(&t_expr.scalar_type, &t_first.scalar_type);
                            if !diffs.is_empty() {
                                return Err(TypeError::MismatchColumn {
                                    source: expr,
                                    got: t_expr,
                                    expected: t_first.clone(),
                                    diffs,
                                    message: "equivalence class members have different scalar types".to_string(),
                                });
                            }

                            // equivalences may or may not match on nullability
                            // before JoinImplementation runs, nullability should match.
                            // but afterwards, some nulls may appear that are actually being filtered out elsewhere
                            if self.strict_join_equivalences {
                                if t_expr.nullable != t_first.nullable {
                                    let sub = t_expr.clone();
                                    let sup = t_first.clone();

                                    let err = TypeError::MismatchColumn {
                                        source: expr,
                                        got: t_expr.clone(),
                                        expected: t_first.clone(),
                                        diffs: vec![ReprColumnTypeDifference::Nullability { sub, sup }],
                                        message: "equivalence class members have different nullability (and join equivalence checking is strict)".to_string(),
                                    };

                                    // TODO(mgree) this imprecision should be resolved, but we need to fix the optimizer
                                    ::tracing::debug!("{err}");
                                }
                            }
                        }

                        if self.disallow_dummy && scalar_expr.contains_dummy() {
                            return Err(TypeError::DisallowedDummy {
                                source: expr,
                            });
                        }

                        t_exprs.push(t_expr);
                    }

                    if self.strict_join_equivalences && all_nullable {
                        let err = TypeError::BadJoinEquivalence {
                            source: expr,
                            got: t_exprs,
                            message: "all expressions were nullable (and join equivalence checking is strict)".to_string(),
                        };

                        // TODO(mgree) this imprecision should be resolved, but we need to fix the optimizer
                        ::tracing::debug!("{err}");
                    }
                }

                // check that the join implementation is consistent
                match implementation {
                    JoinImplementation::Differential((start_idx, first_key, _), others) => {
                        if let Some(key) = first_key {
                            for k in key {
                                let _ = tc.typecheck_scalar(k, expr, &t_in_local[*start_idx])?;
                            }
                        }

                        for (idx, key, _) in others {
                            for k in key {
                                let _ = tc.typecheck_scalar(k, expr, &t_in_local[*idx])?;
                            }
                        }
                    }
                    JoinImplementation::DeltaQuery(plans) => {
                        for plan in plans {
                            for (idx, key, _) in plan {
                                for k in key {
                                    let _ = tc.typecheck_scalar(k, expr, &t_in_local[*idx])?;
                                }
                            }
                        }
                    }
                    JoinImplementation::IndexedFilter(_coll_id, _idx_id, key, consts) => {
                        let typ: Vec<ReprColumnType> = key
                            .iter()
                            .map(|k| tc.typecheck_scalar(k, expr, &t_in_global))
                            .collect::<Result<Vec<ReprColumnType>, TypeError>>()?;

                        for row in consts {
                            let datums = row.unpack();

                            row_difference_with_column_types(expr, &datums, &typ)?;
                        }
                    }
                    JoinImplementation::Unimplemented => (),
                }

                Ok(t_in_global)
            }
            Reduce {
                input,
                group_key,
                aggregates,
                monotonic: _,
                expected_group_size: _,
            } => {
                let t_in = tc.typecheck(input, ctx)?;

                let mut t_out = group_key
                    .iter()
                    .map(|scalar_expr| tc.typecheck_scalar(scalar_expr, expr, &t_in))
                    .collect::<Result<Vec<_>, _>>()?;

                    if self.disallow_dummy && group_key.iter().any(|scalar_expr| scalar_expr.contains_dummy()) {
                        return Err(TypeError::DisallowedDummy {
                            source: expr,
                        });
                    }

                for agg in aggregates {
                    t_out.push(tc.typecheck_aggregate(agg, expr, &t_in)?);
                }

                Ok(t_out)
            }
            TopK {
                input,
                group_key,
                order_key,
                limit: _,
                offset: _,
                monotonic: _,
                expected_group_size: _,
            } => {
                let t_in = tc.typecheck(input, ctx)?;

                for &k in group_key {
                    if k >= t_in.len() {
                        return Err(TypeError::BadTopKGroupKey {
                            source: expr,
                            k,
                            input_type: t_in,
                        });
                    }
                }

                for order in order_key {
                    if order.column >= t_in.len() {
                        return Err(TypeError::BadTopKOrdering {
                            source: expr,
                            order: order.clone(),
                            input_type: t_in,
                        });
                    }
                }

                Ok(t_in)
            }
            Negate { input } => tc.typecheck(input, ctx),
            Threshold { input } => tc.typecheck(input, ctx),
            Union { base, inputs } => {
                let mut t_base = tc.typecheck(base, ctx)?;

                for input in inputs {
                    let t_input = tc.typecheck(input, ctx)?;

                    let len_sub = t_base.len();
                    let len_sup = t_input.len();
                    if len_sub != len_sup {
                        return Err(TypeError::MismatchColumns {
                            source: expr,
                            got: t_base.clone(),
                            expected: t_input,
                            diffs: vec![ReprRelationTypeDifference::Length {
                                len_sub,
                                len_sup,
                            }],
                            message: "Union branches have different numbers of columns".to_string(),
                        });
                    }

                    for (base_col, input_col) in t_base.iter_mut().zip_eq(t_input) {
                        let diffs = column_union(base_col, &input_col);
                        if !diffs.is_empty() {
                            return Err(TypeError::MismatchColumn {
                                    source: expr,
                                    got: input_col,
                                    expected: base_col.clone(),
                                    diffs,
                                    message:
                                        "couldn't compute union of column types in Union"
                                    .to_string(),
                            });
                        }

                    }
                }

                Ok(t_base)
            }
            Let { id, value, body } => {
                let t_value = tc.typecheck(value, ctx)?;

                let binding = Id::Local(*id);
                if ctx.contains_key(&binding) {
                    return Err(TypeError::Shadowing {
                        source: expr,
                        id: binding,
                    });
                }

                let mut body_ctx = ctx.clone();
                body_ctx.insert(Id::Local(*id), t_value);

                tc.typecheck(body, &body_ctx)
            }
            LetRec { ids, values, body, limits: _ } => {
                if ids.len() != values.len() {
                    return Err(TypeError::BadLetRecBindings { source: expr });
                }

                // temporary hack: steal info from the Gets inside to learn the expected types
                // if no get occurs in any definition or the body, that means that relation is dead code (which is okay)
                let mut ctx = ctx.clone();
                // calling tc.collect_recursive_variable_types(expr, ...) triggers a panic due to nested letrecs with shadowing IDs
                for inner_expr in values.iter().chain(std::iter::once(body.as_ref())) {
                    tc.collect_recursive_variable_types(inner_expr, ids, &mut ctx)?;
                }

                for (id, value) in ids.iter().zip_eq(values.iter()) {
                    let typ = tc.typecheck(value, &ctx)?;

                    let id = Id::Local(id.clone());
                    if let Some(ctx_typ) = ctx.get_mut(&id) {
                        for (base_col, input_col) in ctx_typ.iter_mut().zip_eq(typ) {
                            // we expect an EXACT match, but don't care about nullability
                            let diffs = column_union(base_col, &input_col);
                            if !diffs.is_empty() {
                                 return Err(TypeError::MismatchColumn {
                                        source: expr,
                                        got: input_col,
                                        expected: base_col.clone(),
                                        diffs,
                                        message:
                                            "couldn't compute union of column types in LetRec"
                                        .to_string(),
                                    })
                            }
                        }
                    } else {
                        // dead code: no `Get` references this relation anywhere. we record the type anyway
                        ctx.insert(id, typ);
                    }
                }

                tc.typecheck(body, &ctx)
            }
            ArrangeBy { input, keys } => {
                let t_in = tc.typecheck(input, ctx)?;

                for key in keys {
                    for k in key {
                        let _ = tc.typecheck_scalar(k, expr, &t_in)?;
                    }
                }

                Ok(t_in)
            }
        })
    }

    /// Traverses a term to collect the types of given ids.
    ///
    /// LetRec doesn't have type info stored in it. Until we change the MIR to track that information explicitly, we have to rebuild it from looking at the term.
    fn collect_recursive_variable_types<'a>(
        &self,
        expr: &'a MirRelationExpr,
        ids: &[LocalId],
        ctx: &mut Context,
    ) -> Result<(), TypeError<'a>> {
        use MirRelationExpr::*;

        self.checked_recur(|tc| {
            match expr {
                Get {
                    id: Id::Local(id),
                    typ,
                    ..
                } => {
                    if !ids.contains(id) {
                        return Ok(());
                    }

                    let id = Id::Local(id.clone());
                    if let Some(ctx_typ) = ctx.get_mut(&id) {
                        let typ = typ
                            .column_types
                            .iter()
                            .map(ReprColumnType::from)
                            .collect_vec();

                        if ctx_typ.len() != typ.len() {
                            let diffs = relation_subtype_difference(&typ, ctx_typ);

                            return Err(TypeError::MismatchColumns {
                                source: expr,
                                got: typ,
                                expected: ctx_typ.clone(),
                                diffs,
                                message: "environment and type annotation did not match"
                                    .to_string(),
                            });
                        }

                        for (base_col, input_col) in ctx_typ.iter_mut().zip_eq(typ) {
                            let diffs = column_union(base_col, &input_col);
                            if !diffs.is_empty() {
                                return Err(TypeError::MismatchColumn {
                                    source: expr,
                                    got: input_col,
                                    expected: base_col.clone(),
                                    diffs,
                                    message:
                                        "couldn't compute union of column types in Get and context"
                                            .to_string(),
                                });
                            }
                        }
                    } else {
                        ctx.insert(
                            id,
                            typ.column_types
                                .iter()
                                .map(ReprColumnType::from)
                                .collect_vec(),
                        );
                    }
                }
                Get {
                    id: Id::Global(..), ..
                }
                | Constant { .. } => (),
                Let { id, value, body } => {
                    tc.collect_recursive_variable_types(value, ids, ctx)?;

                    // we've shadowed the id
                    if ids.contains(id) {
                        return Err(TypeError::Shadowing {
                            source: expr,
                            id: Id::Local(*id),
                        });
                    }

                    tc.collect_recursive_variable_types(body, ids, ctx)?;
                }
                LetRec {
                    ids: inner_ids,
                    values,
                    body,
                    limits: _,
                } => {
                    for inner_id in inner_ids {
                        if ids.contains(inner_id) {
                            return Err(TypeError::Shadowing {
                                source: expr,
                                id: Id::Local(*inner_id),
                            });
                        }
                    }

                    for value in values {
                        tc.collect_recursive_variable_types(value, ids, ctx)?;
                    }

                    tc.collect_recursive_variable_types(body, ids, ctx)?;
                }
                Project { input, .. }
                | Map { input, .. }
                | FlatMap { input, .. }
                | Filter { input, .. }
                | Reduce { input, .. }
                | TopK { input, .. }
                | Negate { input }
                | Threshold { input }
                | ArrangeBy { input, .. } => {
                    tc.collect_recursive_variable_types(input, ids, ctx)?;
                }
                Join { inputs, .. } => {
                    for input in inputs {
                        tc.collect_recursive_variable_types(input, ids, ctx)?;
                    }
                }
                Union { base, inputs } => {
                    tc.collect_recursive_variable_types(base, ids, ctx)?;

                    for input in inputs {
                        tc.collect_recursive_variable_types(input, ids, ctx)?;
                    }
                }
            }

            Ok(())
        })
    }

    fn typecheck_scalar<'a>(
        &self,
        expr: &'a MirScalarExpr,
        source: &'a MirRelationExpr,
        column_types: &[ReprColumnType],
    ) -> Result<ReprColumnType, TypeError<'a>> {
        use MirScalarExpr::*;

        self.checked_recur(|tc| match expr {
            Column(i, _) => match column_types.get(*i) {
                Some(ty) => Ok(ty.clone()),
                None => Err(TypeError::NoSuchColumn {
                    source,
                    expr,
                    col: *i,
                }),
            },
            Literal(row, typ) => {
                let typ = ReprColumnType::from(typ);
                if let Ok(row) = row {
                    let datums = row.unpack();

                    row_difference_with_column_types(source, &datums, std::slice::from_ref(&typ))?;
                }

                Ok(typ)
            }
            CallUnmaterializable(func) => Ok(ReprColumnType::from(&func.output_type())),
            CallUnary { expr, func } => {
                let typ_in = tc.typecheck_scalar(expr, source, column_types)?;
                let typ_out = func.output_type(SqlColumnType::from_repr(&typ_in));
                Ok(ReprColumnType::from(&typ_out))
            }
            CallBinary { expr1, expr2, func } => {
                let typ_in1 = tc.typecheck_scalar(expr1, source, column_types)?;
                let typ_in2 = tc.typecheck_scalar(expr2, source, column_types)?;
                let typ_out = func.output_type(
                    SqlColumnType::from_repr(&typ_in1),
                    SqlColumnType::from_repr(&typ_in2),
                );
                Ok(ReprColumnType::from(&typ_out))
            }
            CallVariadic { exprs, func } => Ok(ReprColumnType::from(
                &func.output_type(
                    exprs
                        .iter()
                        .map(|e| {
                            tc.typecheck_scalar(e, source, column_types)
                                .map(|typ| SqlColumnType::from_repr(&typ))
                        })
                        .collect::<Result<Vec<_>, TypeError>>()?,
                ),
            )),
            If { cond, then, els } => {
                let cond_type = tc.typecheck_scalar(cond, source, column_types)?;

                // condition must be boolean
                // ignoring nullability: null is treated as false
                // NB this behavior is slightly different from columns_match (for which we would set nullable to false in the expected type)
                if cond_type.scalar_type != ReprScalarType::Bool {
                    let sub = cond_type.scalar_type.clone();

                    return Err(TypeError::MismatchColumn {
                        source,
                        got: cond_type,
                        expected: ReprColumnType {
                            scalar_type: ReprScalarType::Bool,
                            nullable: true,
                        },
                        diffs: vec![ReprColumnTypeDifference::NotSubtype {
                            sub,
                            sup: ReprScalarType::Bool,
                        }],
                        message: "expected boolean condition".to_string(),
                    });
                }

                let mut then_type = tc.typecheck_scalar(then, source, column_types)?;
                let else_type = tc.typecheck_scalar(els, source, column_types)?;

                let diffs = column_union(&mut then_type, &else_type);
                if !diffs.is_empty() {
                    return Err(TypeError::MismatchColumn {
                        source,
                        got: then_type,
                        expected: else_type,
                        diffs,
                        message: "couldn't compute union of column types for If".to_string(),
                    });
                }

                Ok(then_type)
            }
        })
    }

    /// Typecheck an `AggregateExpr`
    pub fn typecheck_aggregate<'a>(
        &self,
        expr: &'a AggregateExpr,
        source: &'a MirRelationExpr,
        column_types: &[ReprColumnType],
    ) -> Result<ReprColumnType, TypeError<'a>> {
        self.checked_recur(|tc| {
            let t_in = tc.typecheck_scalar(&expr.expr, source, column_types)?;

            // TODO check that t_in is actually acceptable for `func`

            Ok(ReprColumnType::from(
                &expr.func.output_type(SqlColumnType::from_repr(&t_in)),
            ))
        })
    }
}

/// Detailed type error logging as a warning, with failures in CI and a logged error in production
///
/// type_error(severity, ...) logs a type warning; if `severity` is `true`, it will also log an error (visible in Sentry)
macro_rules! type_error {
    ($severity:expr, $($arg:tt)+) => {{
        if $severity {
          soft_panic_or_log!($($arg)+);
        } else {
          ::tracing::debug!($($arg)+);
        }
    }}
}

impl crate::Transform for Typecheck {
    fn name(&self) -> &'static str {
        "Typecheck"
    }

    fn actually_perform_transform(
        &self,
        relation: &mut MirRelationExpr,
        transform_ctx: &mut crate::TransformCtx,
    ) -> Result<(), crate::TransformError> {
        let mut typecheck_ctx = self.ctx.lock().expect("typecheck ctx");

        let expected = transform_ctx
            .global_id
            .map_or_else(|| None, |id| typecheck_ctx.get(&Id::Global(id)));

        if let Some(id) = transform_ctx.global_id {
            if self.disallow_new_globals
                && expected.is_none()
                && transform_ctx.global_id.is_some()
                && !id.is_transient()
            {
                type_error!(
                    false, // not severe
                    "type warning: new non-transient global id {id}\n{}",
                    relation.pretty()
                );
            }
        }

        let got = self.typecheck(relation, &typecheck_ctx);

        let humanizer = mz_repr::explain::DummyHumanizer;

        match (got, expected) {
            (Ok(got), Some(expected)) => {
                let id = transform_ctx.global_id.unwrap();

                // contravariant: global types can be updated
                let diffs = relation_subtype_difference(expected, &got);
                if !diffs.is_empty() {
                    // SEVERE only if got and expected have true differences, not just nullability
                    let severity = diffs
                        .iter()
                        .any(|diff| diff.clone().ignore_nullability().is_some());

                    let err = TypeError::MismatchColumns {
                        source: relation,
                        got,
                        expected: expected.clone(),
                        diffs,
                        message: format!(
                            "a global id {id}'s type changed (was `expected` which should be a subtype of `got`) "
                        ),
                    };

                    type_error!(severity, "type error in known global id {id}:\n{err}");
                }
            }
            (Ok(got), None) => {
                if let Some(id) = transform_ctx.global_id {
                    typecheck_ctx.insert(Id::Global(id), got);
                }
            }
            (Err(err), _) => {
                let (expected, binding) = match expected {
                    Some(expected) => {
                        let id = transform_ctx.global_id.unwrap();
                        (
                            format!("expected type {}\n", columns_pretty(expected, &humanizer)),
                            format!("known global id {id}"),
                        )
                    }
                    None => ("".to_string(), "transient query".to_string()),
                };

                type_error!(
                    true, // SEVERE: the transformed code is inconsistent
                    "type error in {binding}:\n{err}\n{expected}{}",
                    relation.pretty()
                );
            }
        }

        Ok(())
    }
}

/// Prints a type prettily with a given `ExprHumanizer`
pub fn columns_pretty<H>(cols: &[ReprColumnType], humanizer: &H) -> String
where
    H: ExprHumanizer,
{
    let mut s = String::with_capacity(2 + 3 * cols.len());

    s.push('(');

    let mut it = cols.iter().peekable();
    while let Some(col) = it.next() {
        s.push_str(&humanizer.humanize_column_type_repr(col, false));

        if it.peek().is_some() {
            s.push_str(", ");
        }
    }

    s.push(')');

    s
}

impl ReprRelationTypeDifference {
    /// Pretty prints a type difference
    ///
    /// Always indents two spaces
    pub fn humanize<H>(&self, h: &H, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result
    where
        H: ExprHumanizer,
    {
        use ReprRelationTypeDifference::*;
        match self {
            Length { len_sub, len_sup } => {
                writeln!(
                    f,
                    "  number of columns do not match ({len_sub} != {len_sup})"
                )
            }
            Column { col, diff } => {
                writeln!(f, "  column {col} differs:")?;
                diff.humanize(4, h, f)
            }
        }
    }
}

impl ReprColumnTypeDifference {
    /// Pretty prints a type difference at a given indentation level
    pub fn humanize<H>(
        &self,
        indent: usize,
        h: &H,
        f: &mut std::fmt::Formatter<'_>,
    ) -> std::fmt::Result
    where
        H: ExprHumanizer,
    {
        use ReprColumnTypeDifference::*;

        // indent
        write!(f, "{:indent$}", "")?;

        match self {
            NotSubtype { sub, sup } => {
                let sub = h.humanize_scalar_type_repr(sub, false);
                let sup = h.humanize_scalar_type_repr(sup, false);

                writeln!(f, "{sub} is a not a subtype of {sup}")
            }
            Nullability { sub, sup } => {
                let sub = h.humanize_column_type_repr(sub, false);
                let sup = h.humanize_column_type_repr(sup, false);

                writeln!(f, "{sub} is nullable but {sup} is not")
            }
            ElementType { ctor, element_type } => {
                writeln!(f, "{ctor} element types differ:")?;

                element_type.humanize(indent + 2, h, f)
            }
            RecordMissingFields { missing } => {
                write!(f, "missing column fields:")?;
                for col in missing {
                    write!(f, " {col}")?;
                }
                f.write_char('\n')
            }
            RecordFields { fields } => {
                writeln!(f, "{} record fields differ:", fields.len())?;

                for (i, diff) in fields.iter().enumerate() {
                    writeln!(f, "{:indent$}  field {i}:", "")?;
                    diff.humanize(indent + 4, h, f)?;
                }
                Ok(())
            }
        }
    }
}

impl DatumTypeDifference {
    /// Pretty prints a type difference at a given indentation level
    pub fn humanize<H>(
        &self,
        indent: usize,
        h: &H,
        f: &mut std::fmt::Formatter<'_>,
    ) -> std::fmt::Result
    where
        H: ExprHumanizer,
    {
        // indent
        write!(f, "{:indent$}", "")?;

        match self {
            DatumTypeDifference::Null => writeln!(f, "unexpected null")?,
            DatumTypeDifference::Mismatch {
                got_debug,
                expected,
            } => {
                let expected = h.humanize_scalar_type_repr(expected, false);
                // NB `got_debug` will be redacted as appropriate
                writeln!(
                    f,
                    "got datum {got_debug}, expected representation type {expected}"
                )?;
            }
            DatumTypeDifference::MismatchDimensions {
                ctor,
                got,
                expected,
            } => {
                writeln!(
                    f,
                    "{ctor} dimensions differ: got datum with dimension {got}, expected dimension {expected}"
                )?;
            }
            DatumTypeDifference::ElementType { ctor, element_type } => {
                writeln!(f, "{ctor} element types differ:")?;
                element_type.humanize(indent + 4, h, f)?;
            }
        }

        Ok(())
    }
}

/// Wrapper struct for a `Display` instance for `TypeError`s with a given `ExprHumanizer`
#[allow(missing_debug_implementations)]
pub struct TypeErrorHumanizer<'a, 'b, H>
where
    H: ExprHumanizer,
{
    err: &'a TypeError<'a>,
    humanizer: &'b H,
}

impl<'a, 'b, H> TypeErrorHumanizer<'a, 'b, H>
where
    H: ExprHumanizer,
{
    /// Create a `Display`-shim struct for a given `TypeError`/`ExprHumanizer` pair
    pub fn new(err: &'a TypeError, humanizer: &'b H) -> Self {
        Self { err, humanizer }
    }
}

impl<'a, 'b, H> std::fmt::Display for TypeErrorHumanizer<'a, 'b, H>
where
    H: ExprHumanizer,
{
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.err.humanize(self.humanizer, f)
    }
}

impl<'a> std::fmt::Display for TypeError<'a> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        TypeErrorHumanizer {
            err: self,
            humanizer: &DummyHumanizer,
        }
        .fmt(f)
    }
}

impl<'a> TypeError<'a> {
    /// The source of the type error
    pub fn source(&self) -> Option<&'a MirRelationExpr> {
        use TypeError::*;
        match self {
            Unbound { source, .. }
            | NoSuchColumn { source, .. }
            | MismatchColumn { source, .. }
            | MismatchColumns { source, .. }
            | BadConstantRowLen { source, .. }
            | BadConstantRow { source, .. }
            | BadProject { source, .. }
            | BadJoinEquivalence { source, .. }
            | BadTopKGroupKey { source, .. }
            | BadTopKOrdering { source, .. }
            | BadLetRecBindings { source }
            | Shadowing { source, .. }
            | DisallowedDummy { source, .. } => Some(source),
            Recursion { .. } => None,
        }
    }

    fn humanize<H>(&self, humanizer: &H, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result
    where
        H: ExprHumanizer,
    {
        if let Some(source) = self.source() {
            writeln!(f, "In the MIR term:\n{}\n", source.pretty())?;
        }

        use TypeError::*;
        match self {
            Unbound { source: _, id, typ } => {
                let typ = columns_pretty(&typ.column_types, humanizer);
                writeln!(f, "{id} is unbound\ndeclared type {typ}")?
            }
            NoSuchColumn {
                source: _,
                expr,
                col,
            } => writeln!(f, "{expr} references non-existent column {col}")?,
            MismatchColumn {
                source: _,
                got,
                expected,
                diffs,
                message,
            } => {
                let got = humanizer.humanize_column_type_repr(got, false);
                let expected = humanizer.humanize_column_type_repr(expected, false);
                writeln!(
                    f,
                    "mismatched column types: {message}\n      got {got}\nexpected {expected}"
                )?;

                for diff in diffs {
                    diff.humanize(2, humanizer, f)?;
                }
            }
            MismatchColumns {
                source: _,
                got,
                expected,
                diffs,
                message,
            } => {
                let got = columns_pretty(got, humanizer);
                let expected = columns_pretty(expected, humanizer);

                writeln!(
                    f,
                    "mismatched relation types: {message}\n      got {got}\nexpected {expected}"
                )?;

                for diff in diffs {
                    diff.humanize(humanizer, f)?;
                }
            }
            BadConstantRowLen {
                source: _,
                got,
                expected,
            } => {
                let expected = columns_pretty(expected, humanizer);
                writeln!(
                    f,
                    "bad constant row\n      row has length {got}\nexpected row of type {expected}"
                )?
            }
            BadConstantRow {
                source: _,
                mismatches,
                expected,
            } => {
                let expected = columns_pretty(expected, humanizer);

                let num_mismatches = mismatches.len();
                let plural = if num_mismatches == 1 { "" } else { "es" };
                writeln!(
                    f,
                    "bad constant row\n      got {num_mismatches} mismatch{plural}\nexpected row of type {expected}"
                )?;

                if num_mismatches > 0 {
                    writeln!(f, "")?;
                    for (col, diff) in mismatches.iter() {
                        writeln!(f, "      column #{col}:")?;
                        diff.humanize(8, humanizer, f)?;
                    }
                }
            }
            BadProject {
                source: _,
                got,
                input_type,
            } => {
                let input_type = columns_pretty(input_type, humanizer);

                writeln!(
                    f,
                    "projection of non-existant columns {got:?} from type {input_type}"
                )?
            }
            BadJoinEquivalence {
                source: _,
                got,
                message,
            } => {
                let got = columns_pretty(got, humanizer);

                writeln!(f, "bad join equivalence {got}: {message}")?
            }
            BadTopKGroupKey {
                source: _,
                k,
                input_type,
            } => {
                let input_type = columns_pretty(input_type, humanizer);

                writeln!(
                    f,
                    "TopK group key component references invalid column {k} in columns: {input_type}"
                )?
            }
            BadTopKOrdering {
                source: _,
                order,
                input_type,
            } => {
                let col = order.column;
                let num_cols = input_type.len();
                let are = if num_cols == 1 { "is" } else { "are" };
                let s = if num_cols == 1 { "" } else { "s" };
                let input_type = columns_pretty(input_type, humanizer);

                // TODO(cloud#8196)
                let mode = HumanizedExplain::new(false);
                let order = mode.expr(order, None);

                writeln!(
                    f,
                    "TopK ordering {order} references invalid column {col}\nthere {are} {num_cols} column{s}: {input_type}"
                )?
            }
            BadLetRecBindings { source: _ } => {
                writeln!(f, "LetRec ids and definitions don't line up")?
            }
            Shadowing { source: _, id } => writeln!(f, "id {id} is shadowed")?,
            DisallowedDummy { source: _ } => writeln!(f, "contains a dummy value")?,
            Recursion { error } => writeln!(f, "{error}")?,
        }

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use mz_ore::{assert_err, assert_ok};
    use mz_repr::{arb_datum, arb_datum_for_column};
    use proptest::prelude::*;

    use super::*;

    #[mz_ore::test]
    fn test_datum_type_difference() {
        let datum = Datum::Int16(1);

        assert_ok!(datum_difference_with_column_type(
            &datum,
            &ReprColumnType {
                scalar_type: ReprScalarType::Int16,
                nullable: true,
            }
        ));

        assert_err!(datum_difference_with_column_type(
            &datum,
            &ReprColumnType {
                scalar_type: ReprScalarType::Int32,
                nullable: false,
            }
        ));
    }

    proptest! {
        #![proptest_config(ProptestConfig { cases: 5000, max_global_rejects: 2500, ..Default::default() })]
        #[mz_ore::test]
        #[cfg_attr(miri, ignore)]
        fn datum_type_difference_with_instance_of_on_valid_data((src, datum) in any::<SqlColumnType>().prop_flat_map(|src| {
            let datum = arb_datum_for_column(src.clone());
            (Just(src), datum) }
        )) {
            let typ = ReprColumnType::from(&src);
            let datum = Datum::from(&datum);

            if datum.contains_dummy() {
                return Err(TestCaseError::reject("datum contains a dummy"));
            }

            let diff = datum_difference_with_column_type(&datum, &typ);
            if datum.is_instance_of(&typ) {
                assert_ok!(diff);
            } else {
                assert_err!(diff);
            }
        }
    }

    proptest! {
        // We run many cases because the data are _random_, and we want to be sure
        // that we have covered sufficient cases. We drop dummy data (behavior is different!) so we allow many more global rejects.
        #![proptest_config(ProptestConfig::with_cases(10000))]
        #[mz_ore::test]
        #[cfg_attr(miri, ignore)]
        fn datum_type_difference_agrees_with_is_instance_of_on_random_data(src in any::<SqlColumnType>(), datum in arb_datum(false)) {
            let typ = ReprColumnType::from(&src);
            let datum = Datum::from(&datum);

            assert!(!datum.contains_dummy(), "datum contains a dummy (bug in arb_datum)");

            let diff = datum_difference_with_column_type(&datum, &typ);
            if datum.is_instance_of(&typ) {
                assert_ok!(diff);
            } else {
                assert_err!(diff);
            }
        }
    }
}