rust_rel8/

lib.rs

// `rustdoc_internals` is needed for `#[doc(fake_variadics)]`
#![allow(internal_features)]
#![cfg_attr(any(docsrs, docsrs_dep), feature(rustdoc_internals))]
// #![deny(missing_docs)]

/*!
Welcome to Rust Rel8

This is a port of Haskell's excellent
[Rel8](https://rel8.readthedocs.io/en/latest/cookbook.html) library which
provides a type safe and expressive interface for constructing SQL queries.

Unlike other ORMs and query builders, this library does not provide a
builder pattern on top of the AST for a SQL query, but instead allows you
to write queries as if the tables themselves were just arrays in rust.

# Overview

The fundamental pieces of this crate are:

 * [`Query<T>`]: A sql query which produces rows of type `T`.

   The simplest usage is [`Query::each`] and [`Query::values`] which correspond
   to `SELECT * FROM <table>` and `SELECT * FROM (VALUES <...>)` respectively.

   You can use [Query::all] to perform the query and decode the results into a `Vec<T>`.

 * [Q]: A type you get from [query], it allows for queries to be composed together.

 * [query]: A function which gives you [Q] in a callback, which you can use to
   compose together an filter queries.

 * [`Expr<T>`]: Some SQL expression representing a scalar value.

 * [Table]: A trait representing a row, can be one of:

   - [`Expr<T>`]: each `Expr<T>` can be thought of as a singleton row.
   - `(T, ...)` where `T: Table`: A tuple of tables is itself a tuple.
   - Your type which uses the [TableStruct] derive macro.

You have probably already noticed that this library is doing a lot of type level
magick, but you'll get the hang of it quickly.

The main thing is that every time you see `T` with bounds of
[`ForLifetimeTable`] (or `T::WithLt<'...>`), you can think of `T` as being any
[Table]. It's just that the library needs to be able to talk about the table
with a different lifetime.

Likewise with [TableHKT], which is needed for the library to mention some user
defined table in different [TableMode]s.

*/

pub mod is_nullable;

use std::{
    borrow::Cow,
    marker::PhantomData,
    sync::{Arc, atomic::AtomicU32},
};

use bytemuck::TransparentWrapper as _;
use sea_query::{ExprTrait, OverStatement};

#[derive(Debug, Clone, Copy, PartialEq, PartialOrd, Ord, Eq, Hash)]
struct Binder(u32);

static BINDER_COUNT: AtomicU32 = AtomicU32::new(0);
impl Binder {
    fn new() -> Self {
        Self(BINDER_COUNT.fetch_add(1, std::sync::atomic::Ordering::SeqCst))
    }

    #[allow(unused)]
    fn reset() {
        BINDER_COUNT.store(0, std::sync::atomic::Ordering::SeqCst);
    }
}

#[derive(Clone, Eq, PartialEq)]
struct TableName {
    binder: Binder,
    name: String,
}

impl std::fmt::Debug for TableName {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.name)
    }
}

impl<'a> From<TableName> for Cow<'a, str> {
    fn from(val: TableName) -> Self {
        format!("t{}", val.binder.0).into()
    }
}

impl TableName {
    fn new(binder: Binder) -> Self {
        Self {
            binder,
            name: format!("t{}", binder.0),
        }
    }
}

impl sea_query::Iden for TableName {
    fn unquoted(&self) -> &str {
        &self.name
    }
}

#[derive(Clone, PartialEq, Eq, Hash)]
struct ColumnName {
    name: String,
    rendered: String,
}

impl std::fmt::Display for ColumnName {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.rendered)
    }
}

impl std::fmt::Debug for ColumnName {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.rendered)
    }
}

impl sea_query::Iden for ColumnName {
    fn unquoted(&self) -> &str {
        &self.rendered
    }
}

impl ColumnName {
    fn new(binder: Binder, name: String) -> Self {
        Self {
            rendered: format!("{}{}", name, binder.0),
            name,
        }
    }
}

/// A trait abstracting the mode of a user defined table, which allows us to
/// talk about the same table in two different modes.
///
/// ```rs
/// impl<'scope, T: Column<'scope>> TableHKT<'scope> for MyTable<'scope, T> {
///     type InMode<Mode: Column<'scope>> = MyTable<'scope, Mode>;
///
///     type Mode = T;
/// }
/// ```
pub trait TableHKT {
    /// The current mode of this table.
    type Mode: TableMode;

    /// Replace the mode with another.
    type InMode<Mode: TableMode>;
}

#[cfg(feature = "sqlx")]
/// Proxy module allowing us to have a supertrait conditional on feature flags.
mod value_sqlx {
    /// Proxy trait for [sqlx::Decode] that we can switch off if not enabled
    pub trait Value: for<'r> sqlx::Decode<'r, sqlx::Postgres> + sqlx::Type<sqlx::Postgres> {}
    impl<T> Value for T where T: for<'r> sqlx::Decode<'r, sqlx::Postgres> + sqlx::Type<sqlx::Postgres> {}
}

#[cfg(not(feature = "sqlx"))]
/// Proxy module allowing us to have a supertrait conditional on feature flags.
mod value_sqlx {
    /// Proxy trait for [sqlx::Decode] that we can switch off if not enabled
    pub trait Value {}
    impl<T> Value for T where T: ?Sized {}
}

pub use value_sqlx::Value as SqlxValueIfEnabled;

/// This trait represents values we know to encode and decode from their database type.
///
/// Depending on features, this will have supertraits of the encode/decode
/// traits of the backends.
pub trait Value: Into<sea_query::Value> + SqlxValueIfEnabled + IsNullable {}

impl<T> Value for T where T: Into<sea_query::Value> + value_sqlx::Value + IsNullable {}

/// This trait allows us to write a mapping function between two column modes.
///
/// We use a trait as we need this to work for all types of the type parameter `V`.
pub trait ModeMapper<'scope, SrcMode: TableMode, DestMode: TableMode> {
    /// Map from `SrcMode` to `DestMode`, consuming the value.
    fn map_mode<V>(&mut self, src: SrcMode::T<'scope, V>) -> DestMode::T<'scope, V>
    where
        V: Value;
}

/// This trait allows us to write a mapping function between two column modes.
///
/// We use a trait as we need this to work for all types of the type parameter `V`.
pub trait ModeMapperRef<'scope, SrcMode: TableMode, DestMode: TableMode> {
    /// Map from `SrcMode` to `DestMode`, taking the value as a reference
    fn map_mode_ref<V>(&mut self, src: &SrcMode::T<'scope, V>) -> DestMode::T<'scope, V>
    where
        V: Value;
}

/// This trait allows us to write a mapping function between two column modes.
///
/// We use a trait as we need this to work for all types of the type parameter `V`.
pub trait ModeMapperMut<'scope, SrcMode: TableMode, DestMode: TableMode> {
    /// Map from `SrcMode` to `DestMode`, taking the value as a mutable reference
    fn map_mode_mut<V>(&mut self, src: &mut SrcMode::T<'scope, V>) -> DestMode::T<'scope, V>
    where
        V: Value;
}

/// This trait allows us to change the mode of a table by mapping all the fields
/// with [ModeMapper], [ModeMapperRef], or [ModeMapperMut].
pub trait MapTable<'scope>: TableHKT {
    /// Map each field of the table
    ///
    /// The order and number of fields visited must always remain the same,
    /// across: [Table::visit], [Table::visit_mut], and all methods of [MapTable].
    fn map_modes<Mapper, DestMode>(self, mapper: &mut Mapper) -> Self::InMode<DestMode>
    where
        Mapper: ModeMapper<'scope, Self::Mode, DestMode>,
        DestMode: TableMode;

    /// Map each field of the table, with a reference
    ///
    /// The order and number of fields visited must always remain the same,
    /// across: [Table::visit], [Table::visit_mut], and all methods of [MapTable].
    fn map_modes_ref<Mapper, DestMode>(&self, mapper: &mut Mapper) -> Self::InMode<DestMode>
    where
        Mapper: ModeMapperRef<'scope, Self::Mode, DestMode>,
        DestMode: TableMode;

    /// Map each field of the table, with a mutable reference
    ///
    /// The order and number of fields visited must always remain the same,
    /// across: [Table::visit], [Table::visit_mut], and all methods of [MapTable].
    fn map_modes_mut<Mapper, DestMode>(&mut self, mapper: &mut Mapper) -> Self::InMode<DestMode>
    where
        Mapper: ModeMapperMut<'scope, Self::Mode, DestMode>,
        DestMode: TableMode;
}

/// A mapper which reads the column names of a table in [NameMode] and adds them to a select,
/// then yields an expression referencing the column.
struct NameToExprMapper {
    binder: Binder,
    query: sea_query::SelectStatement,
}

impl<'scope> ModeMapperRef<'scope, NameMode, ExprMode> for NameToExprMapper {
    fn map_mode_ref<V>(
        &mut self,
        src: &<NameMode as TableMode>::T<'scope, V>,
    ) -> <ExprMode as TableMode>::T<'scope, V> {
        let col_name = ColumnName::new(self.binder, src.to_string());
        self.query
            .expr_as(sea_query::Expr::column(*src), col_name.clone());

        Expr::new(ExprInner::Column(TableName::new(self.binder), col_name))
    }
}

struct LitMapper {}

impl<'scope> ModeMapper<'scope, ValueMode, ExprMode> for LitMapper {
    fn map_mode<V>(
        &mut self,
        src: <ValueMode as TableMode>::T<'scope, V>,
    ) -> <ExprMode as TableMode>::T<'scope, V>
    where
        V: Value,
    {
        Expr::new(ExprInner::Raw(sea_query::Expr::value(src.into())))
    }
}

struct ExprCollectorMapper {
    idx: usize,
    table_binder: Binder,
    columns: Vec<ColumnName>,
    values: Vec<sea_query::Value>,
}

impl<'scope> ModeMapper<'scope, ValueMode, ExprMode> for ExprCollectorMapper {
    fn map_mode<V>(
        &mut self,
        src: <ValueMode as TableMode>::T<'scope, V>,
    ) -> <ExprMode as TableMode>::T<'scope, V>
    where
        V: Value,
    {
        let idx = self.idx;
        self.idx += 1;

        self.values.push(src.into());

        let column_name = ColumnName::new(self.table_binder, format!("values_{idx}_"));

        self.columns.push(column_name.clone());

        Expr::new(ExprInner::Column(
            TableName::new(self.table_binder),
            column_name,
        ))
    }
}

struct ExprCollectorRemainingMapper {
    values: Vec<sea_query::Value>,
}

impl<'scope> ModeMapper<'scope, ValueMode, EmptyMode> for ExprCollectorRemainingMapper {
    fn map_mode<V>(
        &mut self,
        src: <ValueMode as TableMode>::T<'scope, V>,
    ) -> <EmptyMode as TableMode>::T<'scope, V>
    where
        V: Value,
    {
        self.values.push(src.into());
    }
}

struct NameCollectorMapper {
    names: Vec<&'static str>,
}

impl<'scope> ModeMapperRef<'scope, NameMode, EmptyMode> for NameCollectorMapper {
    fn map_mode_ref<V>(
        &mut self,
        src: &<NameMode as TableMode>::T<'scope, V>,
    ) -> <EmptyMode as TableMode>::T<'scope, V> {
        self.names.push(*src);
    }
}

/// A mapper which visits each expression of a table in [ExprMode].
struct VisitTableMapper<'a, F> {
    f: &'a mut F,
    mode: VisitTableMode,
}

impl<'a, 'scope, F> ModeMapperRef<'scope, ExprMode, EmptyMode> for VisitTableMapper<'a, F>
where
    F: FnMut(&ErasedExpr),
{
    fn map_mode_ref<V>(
        &mut self,
        src: &<ExprMode as TableMode>::T<'scope, V>,
    ) -> <EmptyMode as TableMode>::T<'scope, V>
    where
        V: Value,
    {
        match self.mode {
            VisitTableMode::All => (self.f)(src.as_erased()),
            VisitTableMode::NonNull => {
                if !<V as IsNullable>::IS_NULLABLE {
                    (self.f)(src.as_erased());
                }
            }
        }
    }
}

/// A mapper which mutably visits each expression of a table in [ExprMode].
struct VisitTableMapperMut<'a, F> {
    f: &'a mut F,
    mode: VisitTableMode,
}

impl<'a, 'scope, F> ModeMapperMut<'scope, ExprMode, EmptyMode> for VisitTableMapperMut<'a, F>
where
    F: FnMut(&mut ErasedExpr),
{
    fn map_mode_mut<V>(
        &mut self,
        src: &mut <ExprMode as TableMode>::T<'scope, V>,
    ) -> <EmptyMode as TableMode>::T<'scope, V>
    where
        V: Value,
    {
        match self.mode {
            VisitTableMode::All => (self.f)(src.as_erased_mut()),
            VisitTableMode::NonNull => {
                if !<V as IsNullable>::IS_NULLABLE {
                    (self.f)(src.as_erased_mut());
                }
            }
        }
    }
}

#[cfg(feature = "sqlx")]
/// A mapper which decodes a table from a result row, turning it from [ExprMode] to [ValueMode].
struct LoadingMapper<'a, IT> {
    it: &'a mut IT,
}

#[cfg(feature = "sqlx")]
impl<'a, 'b, 'scope, IT> ModeMapperRef<'scope, ExprMode, ValueMode> for LoadingMapper<'a, IT>
where
    IT: Iterator<Item = sqlx::postgres::PgValueRef<'b>>,
{
    fn map_mode_ref<V>(
        &mut self,
        _src: &<ExprMode as TableMode>::T<'scope, V>,
    ) -> <ValueMode as TableMode>::T<'scope, V>
    where
        V: Value,
    {
        <_ as sqlx::Decode<sqlx::Postgres>>::decode(self.it.next().unwrap()).unwrap()
    }
}

#[cfg(feature = "sqlx")]
/// A mapper which decodes a table from a result row, turning it from [ExprMode] to [ValueMode].
struct LoadingManyMapper<'a, IT> {
    it: &'a mut IT,
}

#[cfg(feature = "sqlx")]
impl<'a, 'b, 'scope, IT> ModeMapperRef<'scope, ExprMode, ValueManyMode>
    for LoadingManyMapper<'a, IT>
where
    IT: Iterator<Item = sqlx::postgres::PgValueRef<'b>>,
{
    fn map_mode_ref<V>(
        &mut self,
        _src: &<ExprMode as TableMode>::T<'scope, V>,
    ) -> <ValueManyMode as TableMode>::T<'scope, V>
    where
        V: Value,
    {
        <Vec<V> as sqlx::Decode<sqlx::Postgres>>::decode(self.it.next().unwrap()).unwrap()
    }
}

#[cfg(feature = "sqlx")]
/// A mapper which skips loading a table from a result row.
struct SkippingMapper<'a, IT> {
    it: &'a mut IT,
}

#[cfg(feature = "sqlx")]
impl<'a, 'b, 'scope, IT> ModeMapperRef<'scope, ExprMode, EmptyMode> for SkippingMapper<'a, IT>
where
    IT: Iterator<Item = sqlx::postgres::PgValueRef<'b>>,
{
    fn map_mode_ref<V>(
        &mut self,
        _src: &<ExprMode as TableMode>::T<'scope, V>,
    ) -> <EmptyMode as TableMode>::T<'scope, V> {
        self.it.next().unwrap();
    }
}

#[cfg(feature = "sqlx")]
struct ManyRemainingMapper {
    is_empty: bool,
}

#[cfg(feature = "sqlx")]
impl<'scope> ModeMapperRef<'scope, ValueManyMode, EmptyMode> for ManyRemainingMapper {
    fn map_mode_ref<V>(
        &mut self,
        src: &<ValueManyMode as TableMode>::T<'scope, V>,
    ) -> <EmptyMode as TableMode>::T<'scope, V>
    where
        V: Value,
    {
        self.is_empty &= src.is_empty();
    }
}

#[cfg(feature = "sqlx")]
struct SpreadingManyMapper {}

#[cfg(feature = "sqlx")]
impl<'a, 'scope> ModeMapperMut<'scope, ValueManyMode, ValueMode> for SpreadingManyMapper {
    fn map_mode_mut<V>(
        &mut self,
        src: &mut <ValueManyMode as TableMode>::T<'scope, V>,
    ) -> <ValueMode as TableMode>::T<'scope, V>
    where
        V: Value,
    {
        src.pop().unwrap()
    }
}

/// Trait for tables containing values which we can turn into literals
pub trait LitTable: TableHKT<Mode = ValueMode> {
    /// Lift a [ValueMode] table to an [ExprMode] table
    fn lit(self) -> Self::InMode<ExprMode>;
}

impl<'scope, T: TableHKT<Mode = ValueMode> + MapTable<'scope>> LitTable for T {
    fn lit(self) -> Self::InMode<ExprMode> {
        self.map_modes(&mut LitMapper {})
    }
}

/// A table's name and column names.
pub struct TableSchema<Table> {
    /// The name of the table.
    pub name: &'static str,

    /// The table columns, this should be some table in [NameMode].
    pub columns: Table,
}

/// The modes a table can be in
pub mod table_modes {
    /// Name mode, where all columns are [`&'static str`], representing the column names.
    pub enum NameMode {}

    #[derive(Debug, PartialEq)]
    /// Value mode, representing a table row that has been loaded from the query.
    ///
    /// This enum implements [Debug] and [PartialEq] so that your types can
    /// derive them without a baseless failure from a type parameter not
    /// implementing trait despite not appearing in the data type.
    pub enum ValueMode {}

    pub enum ValueManyMode {}

    // /// Value mode, but the value might be null.
    // pub enum ValueNullifiedMode {}

    /// Expr mode, the columns are [crate::Expr]s.
    pub enum ExprMode {}

    // /// Expr mode, but the value might be null, the columns are [crate::Expr]s.
    // pub enum ExprNullifiedMode {}

    /// Empty mode, all fields are `()`. This is used when a mapper doesn't
    /// want to produce a value.
    pub enum EmptyMode {}
}

pub use table_modes::*;

/// Table modes, this trait is used to switch the types of a rust structs fields.
///
/// You should use it in table struct like so:
///
/// ```rust
/// use rust_rel8::*;
///
/// struct MyTable<'scope, Mode: TableMode> {
///   id: Mode::T<'scope, i32>,
///   name: Mode::T<'scope, String>,
///   age: Mode::T<'scope, i32>,
/// }
/// ```
pub trait TableMode {
    /// A Gat, the resultant type may or may not incorporate `V`.
    type T<'scope, V>;
}

impl TableMode for NameMode {
    /// a string representing the column name.
    type T<'scope, V> = &'static str;
}

impl TableMode for ValueMode {
    type T<'scope, V> = V;
}

impl TableMode for ValueManyMode {
    type T<'scope, V> = Vec<V>;
}

impl TableMode for ExprMode {
    type T<'scope, V> = Expr<'scope, V>;
}

impl TableMode for EmptyMode {
    type T<'scope, V> = ();
}

#[derive(bytemuck::TransparentWrapper)]
#[repr(transparent)]
/// A wrapper which implements [Table] for any type implementing [MapTable] in [ExprMode].
pub struct TableUsingMapper<T>(pub T);

#[allow(missing_docs)]
impl<T> TableUsingMapper<T> {
    pub fn wrap(t: T) -> Self {
        <Self as bytemuck::TransparentWrapper<T>>::wrap(t)
    }

    pub fn wrap_ref(t: &T) -> &Self {
        <Self as bytemuck::TransparentWrapper<T>>::wrap_ref(t)
    }

    pub fn wrap_mut(t: &mut T) -> &mut Self {
        <Self as bytemuck::TransparentWrapper<T>>::wrap_mut(t)
    }
}

impl<'scope, T> Table for TableUsingMapper<T>
where
    T: Table + MapTable<'scope> + TableHKT<Mode = ExprMode>,
{
    type Result = T::InMode<ValueMode>;

    fn visit(&self, f: &mut impl FnMut(&ErasedExpr), mode: VisitTableMode) {
        let mut mapper = VisitTableMapper { f, mode };
        self.0.map_modes_ref(&mut mapper);
    }

    fn visit_mut(&mut self, f: &mut impl FnMut(&mut ErasedExpr), mode: VisitTableMode) {
        let mut mapper = VisitTableMapperMut { f, mode };
        self.0.map_modes_mut(&mut mapper);
    }
}

#[cfg(feature = "sqlx")]
impl<T> TableLoaderSqlx for TableUsingMapper<T>
where
    T: Table + MapTable<'static> + TableHKT<Mode = ExprMode>,
{
    fn load<'a>(
        &self,
        values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>,
    ) -> Self::Result {
        let mut mapper = LoadingMapper { it: values };
        self.0.map_modes_ref(&mut mapper)
    }

    fn skip<'a>(&self, values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>) {
        let mut mapper = SkippingMapper { it: values };
        self.0.map_modes_ref(&mut mapper);
    }
}

#[cfg(feature = "sqlx")]
impl<T> TableLoaderManySqlx for TableUsingMapper<T>
where
    T: Table + MapTable<'static> + TableHKT<Mode = ExprMode>,
    T::InMode<ValueManyMode>: MapTable<'static> + TableHKT<Mode = ValueManyMode>,
    // sorry
    <<T as TableHKT>::InMode<table_modes::ValueManyMode> as TableHKT>::InMode<
        table_modes::ValueMode,
    >: type_equalities::IsEqual<Self::Result>,
{
    fn load_many<'a>(
        &self,
        values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>,
    ) -> Vec<Self::Result> {
        let mut mapper = LoadingManyMapper { it: values };
        let mut collected = self.0.map_modes_ref(&mut mapper);

        let mut results: Vec<Self::Result> = vec![];

        while {
            let mut mapper = ManyRemainingMapper { is_empty: true };
            collected.map_modes_ref(&mut mapper);
            !mapper.is_empty
        } {
            let mut mapper = SpreadingManyMapper {};
            results.push(type_equalities::coerce(
                collected.map_modes_mut(&mut mapper),
                type_equalities::trivial_eq(),
            ));
        }

        results
    }
}

/// How to visit the columns in a table
#[derive(Debug, Copy, Clone)]
pub enum VisitTableMode {
    /// Visit all the columns in the table
    All,
    /// Visit only the non-nullable columns in the table
    NonNull,
}

/// A trait that represents a database result row.
///
/// If you implement [Table] on your type, you must also implement [ForLifetimeTable].
pub trait Table {
    /// The value a row of this table has when loaded from the database.
    type Result;

    /// Visit each expr in the table.
    ///
    /// The order and number of expressions visited must always remain the same,
    /// across: [Table::visit], [Table::visit_mut], and all methods of [MapTable].
    fn visit(&self, f: &mut impl FnMut(&ErasedExpr), mode: VisitTableMode);

    /// Visit each expr in the table, with a mutable reference.
    ///
    /// The order and number of expressions visited must always remain the same,
    /// across: [Table::visit], [Table::visit_mut], and all methods of [MapTable].
    fn visit_mut(&mut self, f: &mut impl FnMut(&mut ErasedExpr), mode: VisitTableMode);
}

#[cfg(feature = "sqlx")]
/// Trait allowing loading of [Table]s when using sqlx.
pub trait TableLoaderSqlx: Table {
    /// Load the table given an iterator over a row's values
    fn load<'a>(
        &self,
        values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>,
    ) -> Self::Result;

    /// discard N columns from the iterator ofer a row
    /// This is used when this value was discarded by a [MaybeTable].
    fn skip<'a>(&self, values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>);
}

#[cfg(feature = "sqlx")]
/// Trait allowing loading of many [Table]s when using sqlx.
pub trait TableLoaderManySqlx: Table + TableLoaderSqlx {
    /// Load the table given an iterator over a vec of row's values
    fn load_many<'a>(
        &self,
        values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>,
    ) -> Vec<Self::Result>;
}

/// attach each of the tables columns to the select, and renames the table to
/// use the new names.
///
/// This embeds expressions, so after this the table contains only Column exprs.
fn subst_table<'scope, T: Table>(
    table: &mut T,
    table_name: TableName,
    dest_select: &mut sea_query::SelectStatement,
) {
    table.visit_mut(
        &mut |ErasedExpr(inner)| {
            let new_column_name = match inner {
                ExprInner::Raw(..) => ColumnName::new(Binder::new(), "lit".to_owned()),
                ExprInner::Column(_table_name, column_name) => {
                    ColumnName::new(Binder::new(), column_name.name.clone())
                }
                ExprInner::BinOp(..) => ColumnName::new(Binder::new(), "expr".to_owned()),
                ExprInner::NOp(..) => ColumnName::new(Binder::new(), "expr".to_owned()),
            };
            let r = inner.render();
            dest_select.expr_as(r, new_column_name.clone());
            *inner = ExprInner::Column(table_name.clone(), new_column_name);
        },
        VisitTableMode::All,
    )
}

// not sure if we should do this by having `q` wrap the query and `subst_table` everything, or this
fn insert_table_name<'scope, T: Table>(table: &mut T, new_table_name: TableName) {
    table.visit_mut(
        &mut |ErasedExpr(inner)| match inner {
            ExprInner::Raw(_) => {}
            ExprInner::Column(table_name, _column_name) => {
                *table_name = new_table_name.clone();
            }
            ExprInner::BinOp(_, expr_inner, expr_inner1) => {
                expr_inner.visit_mut(&mut |table_name, _| *table_name = new_table_name.clone());
                expr_inner1.visit_mut(&mut |table_name, _| *table_name = new_table_name.clone());
            }
            ExprInner::NOp(_, expr_inners) => {
                for expr_inner in expr_inners {
                    expr_inner.visit_mut(&mut |table_name, _| *table_name = new_table_name.clone());
                }
            }
        },
        VisitTableMode::All,
    )
}

fn collect_exprs<'scope, T: Table>(table: &T) -> Vec<ExprInner> {
    let mut exprs = vec![];

    table.visit(
        &mut |ErasedExpr(e)| {
            exprs.push(e.clone());
        },
        VisitTableMode::All,
    );

    exprs
}

/// Private type used to restrict calls of [ForLifetimeTable::with_lt] to the
/// library.
#[non_exhaustive]
pub struct WithLtMarker {}

impl WithLtMarker {
    fn new() -> Self {
        Self {}
    }
}

/// A helper trait that allows us to talk about a [Table] with different
/// lifetimes. Conceptually it is a type level function of `lt -> T where T:
/// Table`.
///
/// If you implement [Table] on your type, you must also implement [ForLifetimeTable].
pub trait ForLifetimeTable {
    /// Substitute the lifetime of this table with `'lt`.
    type WithLt<'lt>: ForLifetimeTable + Table + Sized;

    /// Coerce the lifetime of this table. This is used internally by the library.
    fn with_lt<'lt>(self, marker: &mut WithLtMarker) -> Self::WithLt<'lt>;
    fn unwith_lt<'lt>(with_lt: Self::WithLt<'lt>, marker: &mut WithLtMarker) -> Self;
}

impl<'scope, T> Table for Expr<'scope, T>
where
    T: Value,
{
    type Result = T;

    fn visit(&self, f: &mut impl FnMut(&ErasedExpr), mode: VisitTableMode) {
        match mode {
            VisitTableMode::All => f(self.as_erased()),
            VisitTableMode::NonNull => {
                if !<T as IsNullable>::IS_NULLABLE {
                    f(self.as_erased());
                }
            }
        }
    }

    fn visit_mut(&mut self, f: &mut impl FnMut(&mut ErasedExpr), mode: VisitTableMode) {
        match mode {
            VisitTableMode::All => f(self.as_erased_mut()),
            VisitTableMode::NonNull => {
                if !<T as IsNullable>::IS_NULLABLE {
                    f(self.as_erased_mut());
                }
            }
        }
    }
}

impl<'scope, T: Value> ForLifetimeTable for Expr<'scope, T> {
    type WithLt<'lt> = Expr<'lt, T>;

    fn with_lt<'lt>(self, _marker: &mut WithLtMarker) -> Self::WithLt<'lt> {
        Expr::new(self.expr)
    }

    fn unwith_lt<'lt>(with_lt: Self::WithLt<'lt>, _marker: &mut WithLtMarker) -> Self {
        Expr::new(with_lt.expr)
    }
}

impl<'scope, T> ShortenLifetime for Expr<'scope, T> {
    type Shortened<'small>
        = Expr<'small, T>
    where
        Self: 'small;

    fn shorten_lifetime<'small, 'large: 'small>(self) -> Self::Shortened<'small>
    where
        Self: 'large,
    {
        Expr::new(self.expr)
    }
}

#[cfg(feature = "sqlx")]
impl<T: Value> TableLoaderSqlx for Expr<'static, T> {
    fn load<'a>(
        &self,
        values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>,
    ) -> Self::Result {
        T::decode(values.next().unwrap()).unwrap()
    }

    fn skip<'a>(&self, values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>) {
        let _ = values.next().unwrap();
    }
}

#[cfg(feature = "sqlx")]
impl<T: Value> TableLoaderManySqlx for Expr<'static, T> {
    fn load_many<'a>(
        &self,
        values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>,
    ) -> Vec<Self::Result> {
        <Vec<T> as sqlx::Decode<'_, sqlx::Postgres>>::decode(values.next().unwrap())
            .unwrap_or_default()
    }
}

macro_rules! izip_priv {
    ( @closure $p:pat => $tup:expr ) => {
        |$p| $tup
    };

    ( @closure $p:pat => ( $($tup:tt)* ) , $_iter:expr $( , $tail:expr )* ) => {
        izip_priv!(@closure ($p, b) => ( $($tup)*, b ) $( , $tail )*)
    };

    ($first:expr $(,)*) => {
        $first.into_iter().map(|a| (a,))
    };

    ( $first:expr $( , $rest:expr )* $(,)* ) => {
        {
            let iter = $first.into_iter();
            $(
                let iter = iter.zip($rest);
            )*
            iter.map(izip_priv!(@closure a => (a) $( , $rest )*))
        }
    };
}

macro_rules! impl_tuples {
    ($(#[$meta:meta])* $(($idx:tt, $ty:ident)),*) => {

        $(#[$meta])*
        impl<'scope, $($ty,)*> Table for ($($ty,)*)
            where $($ty: Table,)*
        {
            type Result = ($($ty::Result,)*);

            fn visit(&self, f: &mut impl FnMut(&ErasedExpr), mode: VisitTableMode) {
                $(self.$idx.visit(f, mode);)*
            }

            fn visit_mut(&mut self, f: &mut impl FnMut(&mut ErasedExpr), mode: VisitTableMode) {
                $(self.$idx.visit_mut(f, mode);)*
            }
        }

        $(#[$meta])*
        impl<$($ty,)*> ShortenLifetime for ($($ty,)*)
            where $($ty: ShortenLifetime,)*
        {
            type Shortened<'small>
                = ($($ty::Shortened<'small>,)*)
            where
                Self: 'small;

            fn shorten_lifetime<'small, 'large: 'small>(self) -> Self::Shortened<'small>
            where
                Self: 'large,
            {
                ($(self.$idx.shorten_lifetime(),)*)
            }
        }

        $(#[$meta])*
        impl<$($ty,)*> ForLifetimeTable for ($($ty,)*)
            where $($ty: ForLifetimeTable,)*
        {
            type WithLt<'lt>
                = ($($ty::WithLt<'lt>,)*);

            fn with_lt<'lt>(self, marker: &mut WithLtMarker) -> Self::WithLt<'lt> {
                ($(self.$idx.with_lt(marker),)*)
            }

            fn unwith_lt<'lt>(with_lt: Self::WithLt<'lt>, marker: &mut WithLtMarker) -> Self {
                ($(<$ty>::unwith_lt(with_lt.$idx, marker),)*)
            }
        }

        #[cfg(feature = "sqlx")]
        $(#[$meta])*
        impl<$($ty,)*> TableLoaderSqlx for ($($ty,)*)
            where $($ty: TableLoaderSqlx,)*
        {
            fn load<'a>(
                &self,
                values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>,
            ) -> Self::Result {
                (
                    $(self.$idx.load(values),)*
                )
            }

            fn skip<'a>(&self, values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>) {
                $(self.$idx.skip(values);)*
            }
        }

        #[cfg(feature = "sqlx")]
        $(#[$meta])*
        impl<$($ty,)*> TableLoaderManySqlx for ($($ty,)*)
        where $($ty: TableLoaderManySqlx,)*
        {
            fn load_many<'a>(
                &self,
                values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>,
        ) -> Vec<Self::Result> {
            izip_priv!($(self.$idx.load_many(values),)*).collect()
        }
    }
}
}

variadics_please::all_tuples_enumerated!(
    #[doc(fake_variadic)]
    impl_tuples,
    1,
    15,
    T
);

/// A table which contains a tag indicating whether a query returned rows.
/// Use [Query::optional] to construct this.
pub struct MaybeTable<'scope, T> {
    tag: Expr<'scope, Option<bool>>,
    inner: T,
}

impl<'scope, T: Table> MaybeTable<'scope, T> {
    /// Project out an expression from the maybe table
    pub fn project<U, F>(&self, f: F) -> Expr<'scope, Option<U>>
    where
        F: FnOnce(&T) -> Expr<U>,
    {
        let tag = self.tag.clone();
        let e = f(&self.inner);

        let cb = Arc::new(|cond: sea_query::SimpleExpr, expr: sea_query::SimpleExpr| {
            sea_query::CaseStatement::new()
                .case(cond.is_not_null(), expr)
                .finally(sea_query::Expr::null())
                .into()
        });

        Expr::new(ExprInner::BinOp(cb, Box::new(tag.expr), Box::new(e.expr)))
    }

    /// Test if the table is None, if it is, use the `fallback`, otherwise call
    /// `f`.
    pub fn maybe<U: Table, F>(self, fallback: U, f: F) -> U
    where
        F: FnOnce(T) -> U,
    {
        let tag = self.tag.clone();
        let mut e = f(self.inner);

        let mut defaults = collect_exprs(&fallback).into_iter();

        e.visit_mut(
            &mut |ErasedExpr(inner)| {
                let tag = tag.expr.clone();
                let default = defaults.next().unwrap();
                let cb = Arc::new(|exprs: Vec<sea_query::Expr>| {
                    let [tag, expr, default] = exprs.try_into().unwrap();
                    sea_query::CaseStatement::new()
                        .case(tag.is_not_null(), expr)
                        .finally(default)
                        .into()
                });

                *inner = ExprInner::NOp(cb, vec![tag, inner.clone(), default]);
            },
            VisitTableMode::All,
        );

        e
    }
}

impl<'scope, T: Table> Table for MaybeTable<'scope, T> {
    type Result = Option<T::Result>;

    fn visit(&self, f: &mut impl FnMut(&ErasedExpr), mode: VisitTableMode) {
        self.tag.visit(f, mode);
        self.inner.visit(f, mode);
    }

    fn visit_mut(&mut self, f: &mut impl FnMut(&mut ErasedExpr), mode: VisitTableMode) {
        self.tag.visit_mut(f, mode);
        self.inner.visit_mut(f, mode);
    }
}

impl<'scope, T: ForLifetimeTable + Table> ForLifetimeTable for MaybeTable<'scope, T>
where
    for<'lt> T::WithLt<'lt>: Table,
{
    type WithLt<'lt> = MaybeTable<'lt, T::WithLt<'lt>>;

    fn with_lt<'lt>(self, marker: &mut WithLtMarker) -> Self::WithLt<'lt> {
        MaybeTable {
            tag: self.tag.with_lt(marker),
            inner: self.inner.with_lt(marker),
        }
    }

    fn unwith_lt<'lt>(with_lt: Self::WithLt<'lt>, marker: &mut WithLtMarker) -> Self {
        MaybeTable {
            tag: ForLifetimeTable::unwith_lt(with_lt.tag, marker),
            inner: ForLifetimeTable::unwith_lt(with_lt.inner, marker),
        }
    }
}

impl<'scope, T: ShortenLifetime> ShortenLifetime for MaybeTable<'scope, T> {
    type Shortened<'small>
        = MaybeTable<'small, T::Shortened<'small>>
    where
        Self: 'small;

    fn shorten_lifetime<'small, 'large: 'small>(self) -> Self::Shortened<'small>
    where
        Self: 'large,
    {
        MaybeTable {
            tag: self.tag.shorten_lifetime(),
            inner: self.inner.shorten_lifetime(),
        }
    }
}

#[cfg(feature = "sqlx")]
impl<T: TableLoaderSqlx> TableLoaderSqlx for MaybeTable<'static, T> {
    fn load<'a>(
        &self,
        values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>,
    ) -> Self::Result {
        let tag =
            <Option<bool> as sqlx::Decode<sqlx::Postgres>>::decode(values.next().unwrap()).unwrap();

        if tag == Some(true) {
            Some(self.inner.load(values))
        } else {
            self.inner.skip(values);
            None
        }
    }

    fn skip<'a>(&self, values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>) {
        // the tag
        values.next().unwrap();
        self.inner.skip(values);
    }
}

/// A table wrapping another. This table's rows are assumed null if any of the
/// non-nullable columns in `T` are null.
///
/// Use [Query::nullable] to construct this.
pub struct NullTable<'scope, T> {
    // nulltable still has a tag, but it is derived from the non-null columns of
    // inner
    tag: Expr<'scope, Option<bool>>,
    inner: T,
}

impl<'scope, T: Table> NullTable<'scope, T> {
    /// Project out an expression from the null table
    pub fn project<U, F>(&self, f: F) -> Expr<'scope, Option<U>>
    where
        F: FnOnce(&T) -> Expr<U>,
    {
        let e = f(&self.inner);

        Expr::new(e.expr)
    }

    /// Test if the table is None, if it is, use the `fallback`, otherwise call
    /// `f`.
    pub fn maybe<U: Table, F>(self, fallback: U, f: F) -> U
    where
        F: FnOnce(T) -> U,
    {
        let tag = self.tag.clone();
        let mut e = f(self.inner);

        let mut defaults = collect_exprs(&fallback).into_iter();

        e.visit_mut(
            &mut |ErasedExpr(inner)| {
                let tag = tag.expr.clone();
                let default = defaults.next().unwrap();
                let cb = Arc::new(|exprs: Vec<sea_query::Expr>| {
                    let [tag, expr, default] = exprs.try_into().unwrap();
                    sea_query::CaseStatement::new()
                        .case(tag.is_not_null(), expr)
                        .finally(default)
                        .into()
                });

                *inner = ExprInner::NOp(cb, vec![tag, inner.clone(), default]);
            },
            VisitTableMode::All,
        );

        e
    }
}

impl<'scope, T: Table> Table for NullTable<'scope, T> {
    type Result = Option<T::Result>;

    fn visit(&self, f: &mut impl FnMut(&ErasedExpr), mode: VisitTableMode) {
        self.tag.visit(f, mode);
        self.inner.visit(f, mode);
    }

    fn visit_mut(&mut self, f: &mut impl FnMut(&mut ErasedExpr), mode: VisitTableMode) {
        self.tag.visit_mut(f, mode);
        self.inner.visit_mut(f, mode);
    }
}

impl<'scope, T: ForLifetimeTable + Table> ForLifetimeTable for NullTable<'scope, T>
where
    for<'lt> T::WithLt<'lt>: Table,
{
    type WithLt<'lt> = NullTable<'lt, T::WithLt<'lt>>;

    fn with_lt<'lt>(self, marker: &mut WithLtMarker) -> Self::WithLt<'lt> {
        NullTable {
            tag: self.tag.with_lt(marker),
            inner: self.inner.with_lt(marker),
        }
    }

    fn unwith_lt<'lt>(with_lt: Self::WithLt<'lt>, marker: &mut WithLtMarker) -> Self {
        NullTable {
            tag: ForLifetimeTable::unwith_lt(with_lt.tag, marker),
            inner: ForLifetimeTable::unwith_lt(with_lt.inner, marker),
        }
    }
}

impl<'scope, T: ShortenLifetime> ShortenLifetime for NullTable<'scope, T> {
    type Shortened<'small>
        = NullTable<'small, T::Shortened<'small>>
    where
        Self: 'small;

    fn shorten_lifetime<'small, 'large: 'small>(self) -> Self::Shortened<'small>
    where
        Self: 'large,
    {
        NullTable {
            tag: self.tag.shorten_lifetime(),
            inner: self.inner.shorten_lifetime(),
        }
    }
}

#[cfg(feature = "sqlx")]
impl<T: TableLoaderSqlx> TableLoaderSqlx for NullTable<'static, T> {
    fn load<'a>(
        &self,
        values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>,
    ) -> Self::Result {
        let tag =
            <Option<bool> as sqlx::Decode<sqlx::Postgres>>::decode(values.next().unwrap()).unwrap();

        if tag == Some(true) {
            Some(self.inner.load(values))
        } else {
            self.inner.skip(values);
            None
        }
    }

    fn skip<'a>(&self, values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>) {
        // the tag
        values.next().unwrap();
        self.inner.skip(values);
    }
}

/// A table representing an array aggregation.
///
/// Use [Query::many] or [Query::aggregate] to construct this.
pub struct ListTable<T> {
    inner: T,
}

impl<'scope, T: Table> Table for ListTable<T> {
    type Result = Vec<T::Result>;

    fn visit(&self, f: &mut impl FnMut(&ErasedExpr), mode: VisitTableMode) {
        self.inner.visit(f, mode);
    }

    fn visit_mut(&mut self, f: &mut impl FnMut(&mut ErasedExpr), mode: VisitTableMode) {
        self.inner.visit_mut(f, mode);
    }
}

impl<'scope, T: ForLifetimeTable + Table> ForLifetimeTable for ListTable<T>
where
    for<'lt> T::WithLt<'lt>: Table,
{
    type WithLt<'lt> = ListTable<T::WithLt<'lt>>;

    fn with_lt<'lt>(self, marker: &mut WithLtMarker) -> Self::WithLt<'lt> {
        ListTable {
            inner: self.inner.with_lt(marker),
        }
    }

    fn unwith_lt<'lt>(with_lt: Self::WithLt<'lt>, marker: &mut WithLtMarker) -> Self {
        ListTable {
            inner: ForLifetimeTable::unwith_lt(with_lt.inner, marker),
        }
    }
}

impl<T: ShortenLifetime> ShortenLifetime for ListTable<T> {
    type Shortened<'small>
        = ListTable<T::Shortened<'small>>
    where
        Self: 'small;

    fn shorten_lifetime<'small, 'large: 'small>(self) -> Self::Shortened<'small>
    where
        Self: 'large,
    {
        ListTable {
            inner: self.inner.shorten_lifetime(),
        }
    }
}

#[cfg(feature = "sqlx")]
impl<T: TableLoaderManySqlx> TableLoaderSqlx for ListTable<T> {
    fn load<'a>(
        &self,
        values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>,
    ) -> Self::Result {
        self.inner.load_many(values)
    }

    fn skip<'a>(&self, values: &mut impl Iterator<Item = sqlx::postgres::PgValueRef<'a>>) {
        self.inner.skip(values);
    }
}

/// A value representing a sql select statement which produces rows of type `T`.
#[derive(Clone)]
pub struct Query<T> {
    // Unique ID used to make the table name and columns unique
    binder: Binder,
    expr: sea_query::SelectStatement,
    inner: T,
    siblings_need_random: bool,
}

impl<'scope, T> Query<T>
where
    T: Table + ForLifetimeTable,
{
    fn new(binder: Binder, expr: sea_query::SelectStatement, inner: T) -> Self {
        Self {
            binder,
            expr,
            inner,
            siblings_need_random: false,
        }
    }

    fn into_volatile(mut self) -> Self {
        self.siblings_need_random = true;
        self
    }

    fn with_volatile(mut self, volatility: bool) -> Self {
        self.siblings_need_random = volatility;
        self
    }

    fn erased(self) -> (ErasedQuery, T) {
        let r = ErasedQuery {
            expr: self.expr,
            siblings_need_random: self.siblings_need_random,
        };

        (r, self.inner)
    }

    /// Enter a context in which [A] can be used to build aggregations.
    ///
    /// The callback `f` will receive an [A], the methods of which can be used
    /// to build a table in the `'outer` scope.
    pub fn aggregate<U>(
        self,
        f: impl for<'inner, 'outer> FnOnce(
            &mut A<'inner, 'outer>,
            T::WithLt<'inner>,
        ) -> U::WithLt<'outer>,
    ) -> Query<U::WithLt<'scope>>
    where
        U: ForLifetimeTable,
    {
        let mut a = A {
            group_by: vec![],
            _phantom: PhantomData,
        };

        let mut r = f(&mut a, self.inner.with_lt(&mut WithLtMarker::new()));

        let binder = Binder::new();
        let mut expr = sea_query::SelectStatement::new();

        expr.from_subquery(self.expr, TableName::new(self.binder));

        subst_table(&mut r, TableName::new(binder), &mut expr);

        expr.add_group_by(a.group_by.iter().map(ExprInner::render));

        Query::new(binder, expr, r)
    }

    /// Construct a simple list aggregation for this query.
    ///
    /// That is, all rows of this query will be aggregated into an array.
    pub fn many(mut self) -> Query<ListTable<T>> {
        let binder = Binder::new();
        let mut expr = sea_query::SelectStatement::new();

        expr.from_subquery(self.expr, TableName::new(self.binder));

        self.inner.visit_mut(
            &mut |ErasedExpr(inner)| {
                *inner = ExprInner::NOp(
                    Arc::new(|inners| {
                        let [inner] = inners.try_into().unwrap();

                        sea_query::PgFunc::array_agg(inner).into()
                    }),
                    vec![inner.clone()],
                )
            },
            VisitTableMode::All,
        );

        subst_table(&mut self.inner, TableName::new(binder), &mut expr);

        Query::new(binder, expr, ListTable { inner: self.inner })
            .with_volatile(self.siblings_need_random)
    }

    /// Enter a context in which [W] can be used to insert expressions using window functions.
    ///
    /// The callback `f` will receive an `F`, the methods of which can be used
    /// to build a table in the `'outer` scope.
    pub fn window<U>(
        self,
        f: impl for<'inner, 'outer> FnOnce(
            &mut W<'inner, 'outer>,
            T::WithLt<'inner>,
        ) -> U::WithLt<'outer>,
    ) -> Query<U::WithLt<'scope>>
    where
        U: ForLifetimeTable + Table,
    {
        let inner_binder = Binder::new();
        let middle_binder = Binder::new();
        let outer_binder = Binder::new();

        let mut w = W {
            inner_query: sea_query::SelectStatement::new(),
            inner_table: TableName::new(inner_binder),
            middle_query: sea_query::SelectStatement::new(),
            middle_table: TableName::new(middle_binder),
            _phantom: PhantomData,
        };

        let mut r = f(&mut w, self.inner.with_lt(&mut WithLtMarker::new()));

        w.inner_query
            .from_subquery(self.expr, TableName::new(self.binder));
        w.middle_query.from_subquery(w.inner_query, w.inner_table);

        let mut outer_query = sea_query::SelectStatement::new();
        outer_query.from_subquery(w.middle_query, w.middle_table.clone());

        subst_table(&mut r, w.middle_table, &mut outer_query);

        Query::new(outer_binder, outer_query, r)
    }

    /// Lift a table into a query which ensures side effects happen and are not shared.
    pub fn evaluate(mut table: T) -> Self {
        let binder = Binder::new();
        let mut expr = sea_query::SelectStatement::new();
        subst_table(&mut table, TableName::new(binder), &mut expr);
        Self::new(binder, expr, table).into_volatile()
    }

    /// Transform this query into one which produces rows of either [`Some<T>`] or [None].
    ///
    /// That is, this turns the query into a left join.
    pub fn optional(mut self) -> Query<MaybeTable<'scope, T>> {
        let binder = Binder::new();

        let mut filler = sea_query::Query::select()
            .from_values(
                vec![sea_query::ValueTuple::One(true.into())],
                TableName::new(binder),
            )
            .to_owned();

        let tag = ColumnName::new(binder, "tag".to_owned());

        let mut expr = self.expr;
        expr.expr_as(sea_query::Value::Bool(Some(true)), tag.clone());

        let table_name = TableName::new(self.binder);

        filler.join_subquery(
            sea_query::JoinType::LeftJoin,
            expr,
            table_name.clone(),
            sea_query::Condition::all(),
        );

        // important: the tag must be the first column
        filler.expr_as(
            sea_query::Expr::column((table_name.clone(), tag.clone())),
            tag.clone(),
        );

        // the rest can come later
        subst_table(&mut self.inner, table_name, &mut filler);

        let maybe_table = MaybeTable {
            inner: self.inner,
            tag: Expr {
                expr: ExprInner::Column(TableName::new(binder), tag),
                _phantom: PhantomData,
            },
        };

        Query {
            binder,
            expr: filler,
            inner: maybe_table,
            siblings_need_random: self.siblings_need_random,
        }
    }

    /// Make this table nullable. Unlike [MaybeTable], this isn't a left join,
    /// but instead considers the table null if any of the non-nullable columns
    /// are null.
    ///
    /// If the table only contains `Option<T>` then this table cannot distinguish.
    pub fn nullable(mut self) -> Query<NullTable<'scope, T>> {
        let binder = Binder::new();

        let tag = ColumnName::new(binder, "tag".to_owned());

        let mut non_null_exprs = vec![];
        self.inner.visit(
            &mut |ErasedExpr(e)| {
                non_null_exprs.push(e.render().is_not_null());
            },
            VisitTableMode::NonNull,
        );

        let non_null_expr = non_null_exprs
            .into_iter()
            .fold(sea_query::Condition::all(), |a, b| a.add(b));

        let mut expr = self.expr;
        expr.expr_as(non_null_expr, tag.clone());

        let table_name = TableName::new(self.binder);

        let mut query = sea_query::Query::select();
        query.from_subquery(expr, table_name.clone());

        // important: the tag must be the first column
        query.expr_as(
            sea_query::Expr::column((table_name.clone(), tag.clone())),
            tag.clone(),
        );

        // the rest can come later
        subst_table(&mut self.inner, table_name, &mut query);

        let null_table = NullTable {
            inner: self.inner,
            tag: Expr::new(ExprInner::Column(TableName::new(binder), tag)),
        };

        Query {
            binder,
            expr: query,
            inner: null_table,
            siblings_need_random: self.siblings_need_random,
        }
    }

    fn map<F>(self, f: F) -> Self
    where
        F: FnOnce(&T, &mut sea_query::SelectStatement),
    {
        let binder = Binder::new();

        let mut outer = sea_query::Query::select();
        outer.from_subquery(self.expr, TableName::new(self.binder));

        f(&self.inner, &mut outer);

        let mut e = self.inner;
        subst_table(&mut e, TableName::new(binder), &mut outer);

        Query {
            binder,
            expr: outer,
            inner: e,
            siblings_need_random: self.siblings_need_random,
        }
    }

    /// Add an order by clause to the query, the given function should return a
    /// table, the query will be ordered by each column of the table.
    pub fn order_by<U, F>(self, f: F) -> Self
    where
        U: Table,
        F: FnOnce(&T) -> (U, sea_query::Order),
    {
        self.map(|expr, select| {
            let (order_expr, order) = f(expr);

            order_expr.visit(
                &mut |ErasedExpr(e)| {
                    select.order_by_expr(e.render(), order.clone());
                },
                VisitTableMode::All,
            );
        })
    }

    /// Filter the results of this query. This is analagous to [Q::where_], but
    /// allows for chained usage.
    pub fn filter<F>(self, f: F) -> Self
    where
        F: FnOnce(&T) -> Expr<bool>,
    {
        self.map(|expr, select| {
            let bool_expr = f(expr);

            select.and_where(bool_expr.expr.render());
        })
    }

    pub fn limit(self, n: u64) -> Self {
        self.map(|_expr, select| {
            select.limit(n);
        })
    }

    pub fn offset(self, n: u64) -> Self {
        self.map(|_expr, select| {
            select.offset(n);
        })
    }
}

impl<'scope, T> Query<T>
where
    T: MapTable<'scope> + TableHKT<Mode = NameMode>,
    T::InMode<ExprMode>: ForLifetimeTable + Table,
{
    /// Given a [TableSchema], build a query that selects all columns of every row.
    pub fn each(schema: &TableSchema<T>) -> Query<T::InMode<ExprMode>> {
        let binder = Binder::new();
        let mut query = sea_query::Query::select();
        query.from(schema.name);

        let mut mapper = NameToExprMapper { binder, query };

        let expr = schema.columns.map_modes_ref(&mut mapper);

        Query::new(binder, mapper.query, expr)
    }
}

impl<'scope, T: TableHKT<Mode = ValueMode>> Query<T> {
    /// Construct a query yielding the given rows
    pub fn values(vals: impl IntoIterator<Item = T>) -> Query<T::InMode<ExprMode>>
    where
        T: MapTable<'scope>,
        T::InMode<ExprMode>: ForLifetimeTable + Table,
    {
        let binder = Binder::new();
        let mut iter = vals.into_iter();

        let Some(first) = iter.next() else {
            panic!("Nein danke");
        };

        let mut mapper = ExprCollectorMapper {
            idx: 0,
            table_binder: binder,
            columns: Vec::new(),
            values: Vec::new(),
        };

        let result_expr = first.map_modes(&mut mapper);

        let mut all_values = vec![mapper.values];

        for v in iter {
            let mut mapper = ExprCollectorRemainingMapper { values: Vec::new() };
            v.map_modes(&mut mapper);
            all_values.push(mapper.values);
        }

        let mut select = sea_query::Query::select();
        for (idx, col) in mapper.columns.into_iter().enumerate() {
            select.expr_as(sea_query::Expr::column(format!("column{}", idx + 1)), col);
        }

        select.from_values(
            all_values
                .into_iter()
                .map(|v| sea_query::ValueTuple::Many(v)),
            TableName::new(binder),
        );

        Query::new(binder, select, result_expr)
    }
}

#[cfg(feature = "sqlx")]
impl<T: TableLoaderSqlx> Query<T> {
    /// Load all rows of the query
    pub async fn all(&self, pool: &mut sqlx::PgConnection) -> sqlx::Result<Vec<T::Result>> {
        use sea_query::PostgresQueryBuilder;
        use sea_query_sqlx::SqlxBinder as _;
        use sqlx::Row as _;

        let (sql, values) = self.expr.build_sqlx(PostgresQueryBuilder);

        let all = sqlx::query_with(&sql, values).fetch_all(pool).await?;

        Ok(all
            .into_iter()
            .map(|row| {
                let len = row.len();
                let mut it = (0..len).map(|x| row.try_get_raw(x).unwrap());
                self.inner.load(&mut it)
            })
            .collect::<Vec<_>>())
    }
}

#[derive(Debug)]
struct ErasedQuery {
    expr: sea_query::SelectStatement,
    siblings_need_random: bool,
}

/// A publicly exposed opaque type that is used by [Table::visit] and
/// [Table::visit_mut]. Its purpose is to allow you to store [Expr]s in your
/// types which implement the [Table] trait.
#[derive(bytemuck::TransparentWrapper)]
#[repr(transparent)]
pub struct ErasedExpr(ExprInner);

#[derive(Clone)]
enum ExprInner {
    Raw(sea_query::Expr),
    Column(TableName, ColumnName),
    BinOp(
        Arc<dyn Fn(sea_query::SimpleExpr, sea_query::SimpleExpr) -> sea_query::SimpleExpr>,
        Box<ExprInner>,
        Box<ExprInner>,
    ),
    NOp(
        Arc<dyn Fn(Vec<sea_query::SimpleExpr>) -> sea_query::SimpleExpr>,
        Vec<ExprInner>,
    ),
}

impl ExprInner {
    fn visit_mut(&mut self, f: &mut impl FnMut(&mut TableName, &mut ColumnName)) {
        match self {
            ExprInner::Column(table_name, column_name) => f(table_name, column_name),
            ExprInner::BinOp(_, expr_inner, expr_inner1) => {
                expr_inner.visit_mut(f);
                expr_inner1.visit_mut(f);
            }
            ExprInner::NOp(_, inners) => {
                for inner in inners {
                    inner.visit_mut(f);
                }
            }
            ExprInner::Raw(_) => {}
        }
    }

    fn render(&self) -> sea_query::SimpleExpr {
        match self {
            ExprInner::Raw(value) => value.clone(),
            ExprInner::Column(table_name, column_name) => {
                sea_query::Expr::column((table_name.clone(), column_name.clone()))
            }
            ExprInner::BinOp(cb, expr_inner, expr_inner1) => {
                cb(expr_inner.render(), expr_inner1.render())
            }
            ExprInner::NOp(cb, inners) => cb(inners.into_iter().map(|n| n.render()).collect()),
        }
    }
}

/// A type representing an expression in the query, can be passed around on the
/// rust side to wire things up
#[derive(Clone)]
pub struct Expr<'scope, T> {
    expr: ExprInner,
    _phantom: PhantomData<(&'scope (), T)>,
}

impl<'scope, T> Expr<'scope, T> {
    fn new(expr: ExprInner) -> Self {
        Self {
            expr,
            _phantom: PhantomData,
        }
    }

    /// Construct a literal value from any value from any value that can be encoded.
    pub fn lit(value: T) -> Self
    where
        T: Into<sea_query::Value>,
    {
        Self::new(ExprInner::Raw(sea_query::Expr::value(value.into())))
    }

    fn binop<U>(
        self,
        other: Self,
        binop: Arc<dyn Fn(sea_query::SimpleExpr, sea_query::SimpleExpr) -> sea_query::SimpleExpr>,
    ) -> Expr<'scope, U> {
        Expr::new(ExprInner::BinOp(
            binop,
            Box::new(self.expr),
            Box::new(other.expr),
        ))
    }

    /// SQL equality
    pub fn equals(self, other: Self) -> Expr<'scope, bool> {
        self.binop(
            other,
            Arc::new(|a, b| a.binary(sea_query::BinOper::Equal, b)),
        )
    }

    fn as_erased(&self) -> &ErasedExpr {
        ErasedExpr::wrap_ref(&self.expr)
    }

    fn as_erased_mut(&mut self) -> &mut ErasedExpr {
        ErasedExpr::wrap_mut(&mut self.expr)
    }
}

pub trait DBNum {}
impl DBNum for i32 {}
impl DBNum for i64 {}
impl DBNum for f32 {}

macro_rules! binop {
    ($name:ident, $oper:tt) => {
        pub fn $name(self, other: Self) -> Self {
            self.binop(
                other,
                Arc::new(|a, b| a.binary(sea_query::BinOper::$oper, b)),
            )
        }
    };
}

impl<'scope, T: DBNum> Expr<'scope, T> {
    binop!(add, Add);
    binop!(sub, Sub);
    binop!(mul, Mul);
    binop!(div, Div);
    binop!(modulo, Mod);
}

pub trait DBBool {}
impl DBNum for bool {}

impl<'scope, T: DBBool> Expr<'scope, T> {
    binop!(and, And);
    binop!(or, Or);
}

pub trait DBLike {}
impl DBLike for String {}

impl<'scope, T: DBLike> Expr<'scope, T> {
    binop!(like, Like);
    binop!(not_like, NotLike);
}

pub trait DBOrd {}
impl DBOrd for i32 {}
impl DBOrd for i64 {}
impl DBOrd for f32 {}
impl DBOrd for bool {}
impl DBOrd for String {}

impl<'scope, T: DBOrd> Expr<'scope, T> {
    binop!(lt, SmallerThan);
    binop!(leq, SmallerThanOrEqual);
    binop!(gt, GreaterThan);
    binop!(geq, GreaterThanOrEqual);
}

pub trait DBBit {}
impl DBBit for i32 {}
impl DBBit for i64 {}

impl<'scope, T: DBBit> Expr<'scope, T> {
    binop!(bit_and, BitAnd);
    binop!(bit_or, BitOr);
    binop!(lshift, LShift);
    binop!(rshift, RShift);
}

impl<'scope> Expr<'scope, i32> {
    /// generate `nextval('name')`, this must be used within [`Query::evaluate`]
    /// for it to behave properly.
    ///
    /// # Example
    ///
    /// ```rust
    /// use rust_rel8::{helper_tables::One, *};
    ///
    /// query::<(Expr<i32>, Expr<i32>)>(|q| {
    ///   let id = q.q(Query::evaluate(Expr::nextval("table_id_seq")));
    ///   let v = q.q(Query::values([1, 2, 3].map(|a| One { a })));
    ///   (id, v.a)
    /// });
    /// ```
    pub fn nextval(name: &str) -> Self {
        Self::new(ExprInner::Raw(
            sea_query::Func::cust("nextval").arg(name.to_owned()).into(),
        ))
    }
}

/// An opaque value you can use to compose together queries.
///
/// To get a value of this type, use [query].
pub struct Q<'scope> {
    queries: Vec<(TableName, ErasedQuery)>,
    filters: Vec<ExprInner>,
    binder: Binder,
    _phantom: PhantomData<&'scope ()>,
}

impl<'scope> Q<'scope> {
    /// Bind a query and give you a value representing each row it produces.
    ///
    /// The `'scope` lifetime prevents this value leaking out of its context,
    /// which would result in invalid queries.
    pub fn q<T: ForLifetimeTable + Table>(&mut self, query: Query<T>) -> T::WithLt<'scope> {
        let binder = Binder::new();
        let name = TableName::new(binder);
        let (erased, mut inner) = query.erased();
        self.queries.push((name.clone(), erased));
        insert_table_name(&mut inner, name);
        inner.with_lt(&mut WithLtMarker {})
    }

    /// Introduce a where clause for this query.
    ///
    /// If you introduce a clause that looks like `a.id = b.a_id` then you
    /// effectively create an inner join.
    pub fn where_<'a>(&mut self, expr: Expr<'a, bool>)
    where
        'scope: 'a,
    {
        self.filters.push(expr.expr);
    }
}

/// Open a context allowing you to manipulate a query.
///
/// Inside you can use [`Q::q`] on as many [`Query<T>`] values as you wish, the
/// result of each call can be thought of as each value the query yields.
///
/// You can think of this as cross joining each query together, to create inner
/// joins or left joins, simply use [Q::where_] or [`Query<T>::optional`]
/// respectively.
///
/// Unfortunately, rustc isn't able to infer the return type of this function as
/// there seems to be no good way to express that `T::WithLt<'a>` is the same type
/// as `T::WithLt<'b>` modulo lifetimes.
pub fn query<'outer, T: ForLifetimeTable + Table + 'outer>(
    f: impl for<'scope> FnOnce(&mut Q<'scope>) -> T::WithLt<'scope>,
) -> Query<T> {
    let mut q = Q {
        binder: Binder::new(),
        filters: Vec::new(),
        queries: Vec::new(),
        _phantom: PhantomData,
    };

    let mut e = f(&mut q);

    // if one of the selects needs the parents to have dummy columns to prevent
    // postgres evaluating it only once, we add to each query a `random() as dummy`.
    // Then, in those selects needing them,
    let needs_random = q.queries.iter().any(|(_, q)| q.siblings_need_random);

    let mut random_binders: Vec<sea_query::Expr> = Vec::new();

    let mut insert_dummy = |mut stmt: sea_query::SelectStatement, table: &TableName| {
        if needs_random {
            stmt.expr_as(sea_query::Func::random(), "dummy");

            for binder in &random_binders {
                stmt.and_where(binder.clone());
            }

            random_binders
                .push(sea_query::Expr::column((table.clone(), "dummy".to_string())).is_not_null())
        }
        stmt
    };

    let mut iter = q.queries.into_iter();
    let mut table = sea_query::Query::select();

    if let Some((first_table_name, first)) = iter.next() {
        let expr = insert_dummy(first.expr, &first_table_name);
        table.from_subquery(expr, first_table_name);
    };

    for (table_name, q) in iter {
        let expr = insert_dummy(q.expr, &table_name);
        table.join_lateral(
            // normally a cross join, but sea_query doesn't support omitting the `ON` for cross joins (:
            // and CROSS JOIN is INNER JOIN ON TRUE
            sea_query::JoinType::InnerJoin,
            expr,
            table_name,
            sea_query::Condition::all(),
        );
    }

    for filter in q.filters {
        table.and_where(filter.render());
    }

    subst_table(&mut e, TableName::new(q.binder), &mut table);

    let e = ForLifetimeTable::unwith_lt(e, &mut WithLtMarker {});

    // if we needed to add random calls, so do parents
    let mut q = Query::new(q.binder, table.to_owned(), e);
    q.siblings_need_random = needs_random;
    q
}

/// Construct an insert statement, the result of the query `rows` will be inserted into `into`.
pub struct Insert<T: TableHKT<Mode = NameMode>> {
    /// The table to insert the rows into.
    pub into: TableSchema<T>,
    /// Query producing the rows to insert.
    pub rows: Query<T::InMode<ExprMode>>,
}

#[cfg(feature = "sqlx")]
impl<T: TableHKT<Mode = NameMode>> Insert<T>
where
    T: MapTable<'static>,
{
    /// Run the insert statement
    pub async fn run(
        &self,
        pool: &mut sqlx::PgConnection,
    ) -> sqlx::Result<sqlx::postgres::PgQueryResult> {
        use sea_query::PostgresQueryBuilder;
        use sea_query_sqlx::SqlxBinder as _;

        let mut insert = sea_query::Query::insert()
            .into_table(self.into.name)
            .to_owned();

        let mut mapper = NameCollectorMapper { names: Vec::new() };
        self.into.columns.map_modes_ref(&mut mapper);

        insert.columns(mapper.names);
        insert.select_from(self.rows.expr.clone()).unwrap();

        let (sql, values) = insert.build_sqlx(PostgresQueryBuilder);

        let all = sqlx::query_with(&sql, values).execute(pool).await?;

        Ok(all)
    }
}

/// A struct allowing manipulation of an aggregation
///
/// # Lifetimes
///
/// The `'inner` lifetime represents the lifetime of the query you are aggregating on.
/// The `'outer` lifetime represents the lifetime of the aggregation query being build.
///
/// The two are separate to ensure that any expression in the output table is
/// either used in an aggregation function, or part of a group by clause.
pub struct A<'inner, 'outer> {
    group_by: Vec<ExprInner>,
    _phantom: PhantomData<(&'inner (), &'outer ())>,
}

impl<'inner, 'outer> A<'inner, 'outer> {
    /// Group the aggregation by the given table.
    pub fn group_by<T: Table + ForLifetimeTable>(&mut self, expr: T) -> T::WithLt<'outer> {
        expr.visit(
            &mut |ErasedExpr(e)| {
                self.group_by.push(e.clone());
            },
            VisitTableMode::All,
        );

        expr.with_lt(&mut WithLtMarker::new())
    }

    /// Aggregate rows of the given table into an array
    pub fn array_agg<T: Table + ForLifetimeTable>(
        &mut self,
        mut expr: T,
    ) -> ListTable<T::WithLt<'outer>> {
        expr.visit_mut(
            &mut |ErasedExpr(inner)| {
                *inner = ExprInner::NOp(
                    Arc::new(|inners| {
                        let [inner] = inners.try_into().unwrap();

                        sea_query::PgFunc::array_agg(inner).into()
                    }),
                    vec![inner.clone()],
                )
            },
            VisitTableMode::All,
        );

        ListTable {
            inner: expr.with_lt(&mut WithLtMarker::new()),
        }
    }

    fn simple_agg_fn<T: Table + ForLifetimeTable>(
        &mut self,
        mut expr: T,
        f: impl Fn(sea_query::Expr) -> sea_query::Expr + Clone + 'static,
    ) -> T::WithLt<'outer> {
        expr.visit_mut(
            &mut |ErasedExpr(inner)| {
                let f = f.clone();
                *inner = ExprInner::NOp(
                    Arc::new(move |inners| {
                        let [inner] = inners.try_into().unwrap();

                        f(inner)
                    }),
                    vec![inner.clone()],
                )
            },
            VisitTableMode::All,
        );

        expr.with_lt(&mut WithLtMarker::new())
    }

    /// Average the values of this table in the group.
    pub fn avg<T: Table + ForLifetimeTable>(&mut self, expr: T) -> T::WithLt<'outer> {
        self.simple_agg_fn(expr, |e| e.avg())
    }

    /// Bit and the values of this table in the group.
    pub fn bit_and<T: Table + ForLifetimeTable>(&mut self, expr: T) -> T::WithLt<'outer> {
        self.simple_agg_fn(expr, |e| sea_query::Func::bit_and(e).into())
    }

    /// Bit or the values of this table in the group.
    pub fn bit_or<T: Table + ForLifetimeTable>(&mut self, expr: T) -> T::WithLt<'outer> {
        self.simple_agg_fn(expr, |e| sea_query::Func::bit_or(e).into())
    }

    /// Bit xor the values of this table in the group.
    pub fn bit_xor<T: Table + ForLifetimeTable>(&mut self, expr: T) -> T::WithLt<'outer> {
        self.simple_agg_fn(expr, |e| sea_query::Func::cust("BIT_XOR").arg(e).into())
    }

    /// Boolean and the values of this table in the group.
    pub fn bool_and<T: Table + ForLifetimeTable>(&mut self, expr: T) -> T::WithLt<'outer> {
        self.simple_agg_fn(expr, |e| sea_query::Func::cust("BOOL_AND").arg(e).into())
    }

    /// Boolean or the values of this table in the group.
    pub fn bool_or<T: Table + ForLifetimeTable>(&mut self, expr: T) -> T::WithLt<'outer> {
        self.simple_agg_fn(expr, |e| sea_query::Func::cust("BOOL_OR").arg(e).into())
    }

    /// Count the number of rows in the group, including nulls.
    pub fn count_star(&mut self) -> Expr<'outer, i32> {
        Expr::new(ExprInner::Raw(sea_query::Expr::cust("*").count()))
    }

    /// Count the number of rows in the group, excluding where the table is null.
    pub fn count<T>(&mut self, expr: Expr<'inner, T>) -> Expr<'outer, i32> {
        Expr::new(ExprInner::NOp(
            Arc::new(|inners| {
                let [inner] = inners.try_into().unwrap();

                inner.count()
            }),
            vec![expr.expr],
        ))
    }

    /// Count the distinct number of rows in the group, excluding where the table is null.
    pub fn count_distinct<T>(&mut self, expr: Expr<'inner, T>) -> Expr<'outer, i32> {
        Expr::new(ExprInner::NOp(
            Arc::new(|inners| {
                let [inner] = inners.try_into().unwrap();

                inner.count_distinct()
            }),
            vec![expr.expr],
        ))
    }

    /// The maximum value of each column of this table in its group.
    pub fn max<T: Table + ForLifetimeTable>(&mut self, expr: T) -> T::WithLt<'outer> {
        self.simple_agg_fn(expr, |e| e.max())
    }

    /// The minimum value of each column of this table in its group.
    pub fn min<T: Table + ForLifetimeTable>(&mut self, expr: T) -> T::WithLt<'outer> {
        self.simple_agg_fn(expr, |e| e.min())
    }

    /// The summation of the values of each column of this table in its group.
    pub fn sum<T: Table + ForLifetimeTable>(&mut self, expr: T) -> T::WithLt<'outer> {
        self.simple_agg_fn(expr, |e| e.sum())
    }
}

/// A struct representing an `OVER` used by a window function.
#[derive(Clone)]
pub struct Over<'scope> {
    partitions: Vec<ExprInner>,
    orderings: Vec<(ExprInner, sea_query::Order)>,
    _phantom: PhantomData<&'scope ()>,
}

impl<'scope> Over<'scope> {
    /// Create a new [Over] for use with [W] in [Query::window]
    pub fn new() -> Self {
        Self {
            partitions: Vec::new(),
            orderings: Vec::new(),
            _phantom: PhantomData,
        }
    }

    /// Add an `ORDER BY` clause to this `OVER`
    pub fn order_by<T: Table>(mut self, expr: T, order: sea_query::Order) -> Self {
        expr.visit(
            &mut |ErasedExpr(e)| self.orderings.push((e.clone(), order.clone())),
            VisitTableMode::All,
        );
        self
    }

    /// Add a `PARTITION BY` clause to this `OVER`
    pub fn partition_by<T: Table>(mut self, expr: T) -> Self {
        expr.visit(
            &mut |ErasedExpr(e)| self.partitions.push(e.clone()),
            VisitTableMode::All,
        );
        self
    }
}

/// A struct allowing insertion of window functions
///
/// # Lifetimes
///
/// The `'inner` lifetime represents the lifetime of the query you are starting with.
/// The `'outer` lifetime represents the lifetime of the the new query containing the window function expressions.
///
/// Ideally this would not be used and you would just use something like
/// `Expr::row_number().over(...)`, but the query builder being used requires
/// that `PARTITION BY` takes only column names, therefore to support any
/// arbitrary expression we need to place everything into a subquery in which we
/// just alias all expressions to single tables.
///
/// If you have a table in the `'inner` scope, you can freely use [W::id] to bring it to `'outer`.
pub struct W<'inner, 'outer> {
    // inner select, starts empty. when consuming a partition, or by use of
    // `id`, we add the contained expressions to this table.
    inner_query: sea_query::SelectStatement,
    inner_table: TableName,
    middle_query: sea_query::SelectStatement,
    middle_table: TableName,
    _phantom: PhantomData<(&'inner (), &'outer ())>,
}

impl<'inner, 'outer> W<'inner, 'outer> {
    fn generic_window(&mut self, expr: sea_query::Expr, over: Over<'inner>) -> ExprInner {
        let mut window_statement = sea_query::WindowStatement::new();

        for expr in over.partitions {
            let name = ColumnName::new(Binder::new(), "part".to_owned());
            self.inner_query.expr_as(expr.render(), name.clone());
            window_statement.partition_by((self.inner_table.clone(), name));
        }

        for (expr, order) in over.orderings {
            let name = ColumnName::new(Binder::new(), "ord".to_owned());
            self.inner_query.expr_as(expr.render(), name.clone());
            window_statement.order_by((self.inner_table.clone(), name), order);
        }

        let column = ColumnName::new(Binder::new(), "win".to_owned());

        self.middle_query
            .expr_window_as(expr, window_statement, column.clone());

        ExprInner::Column(self.middle_table.clone(), column)
    }

    pub fn row_number(&mut self, over: Over<'inner>) -> Expr<'outer, i64> {
        Expr::new(self.generic_window(sea_query::Func::cust("row_number").into(), over))
    }

    pub fn rank(&mut self, over: Over<'inner>) -> Expr<'outer, i64> {
        Expr::new(self.generic_window(sea_query::Func::cust("rank").into(), over))
    }

    pub fn dense_rank(&mut self, over: Over<'inner>) -> Expr<'outer, i64> {
        Expr::new(self.generic_window(sea_query::Func::cust("dense_rank").into(), over))
    }

    pub fn percent_rank(&mut self, over: Over<'inner>) -> Expr<'outer, f64> {
        Expr::new(self.generic_window(sea_query::Func::cust("percent_rank").into(), over))
    }

    pub fn cume_dist(&mut self, over: Over<'inner>) -> Expr<'outer, f64> {
        Expr::new(self.generic_window(sea_query::Func::cust("cume_dist").into(), over))
    }

    pub fn ntile(
        &mut self,
        num_buckets: Expr<'outer, i64>,
        over: Over<'inner>,
    ) -> Expr<'outer, f64> {
        Expr::new(
            self.generic_window(
                sea_query::Func::cust("ntlile")
                    .arg(num_buckets.expr.render())
                    .into(),
                over,
            ),
        )
    }

    fn lag_lead_generic<T: Table + ForLifetimeTable>(
        &mut self,
        fn_: &'static str,
        mut value: T,
        offset: Option<Expr<'outer, i64>>,
        default: Option<T>,
        over: Over<'inner>,
    ) -> T::WithLt<'outer> {
        let mut default_exprs = default.as_ref().map(collect_exprs).map(|x| x.into_iter());

        let mut window_statement = sea_query::WindowStatement::new();

        for expr in over.partitions {
            let name = ColumnName::new(Binder::new(), "part".to_owned());
            self.inner_query.expr_as(expr.render(), name.clone());
            window_statement.partition_by((self.inner_table.clone(), name));
        }

        for (expr, order) in over.orderings {
            let name = ColumnName::new(Binder::new(), "ord".to_owned());
            self.inner_query.expr_as(expr.render(), name.clone());
            window_statement.order_by((self.inner_table.clone(), name), order);
        }

        subst_table(&mut value, self.inner_table.clone(), &mut self.inner_query);

        value.visit_mut(
            &mut |ErasedExpr(inner)| {
                let offset = offset.clone();
                let default = default_exprs.as_mut().map(|x| x.next().unwrap());

                let r = sea_query::Func::cust(fn_).arg(inner.clone().render());
                let (r, set_default) = if let Some(offset) = offset {
                    (r.arg(offset.expr.render()), true)
                } else {
                    (r, false)
                };
                let r = if let Some(default) = default {
                    let r = if !set_default {
                        r.arg(sea_query::Expr::value(1))
                    } else {
                        r
                    };
                    r.arg(default.render())
                } else {
                    r
                };

                let column = ColumnName::new(Binder::new(), "win".to_owned());

                self.middle_query
                    .expr_window_as(r, window_statement.clone(), column.clone());

                *inner = ExprInner::Column(self.middle_table.clone(), column);
            },
            VisitTableMode::All,
        );

        value.with_lt(&mut WithLtMarker::new())
    }

    pub fn lag<T: Table + ForLifetimeTable>(
        &mut self,
        value: T,
        offset: Option<Expr<'outer, i64>>,
        default: Option<T>,
        over: Over<'inner>,
    ) -> T::WithLt<'outer> {
        self.lag_lead_generic("lag", value, offset, default, over)
    }

    pub fn lead<T: Table + ForLifetimeTable>(
        &mut self,
        value: T,
        offset: Option<Expr<'outer, i64>>,
        default: Option<T>,
        over: Over<'inner>,
    ) -> T::WithLt<'outer> {
        self.lag_lead_generic("lead", value, offset, default, over)
    }

    pub fn first_value<T: Table + ForLifetimeTable>(
        &mut self,
        value: T,
        over: Over<'inner>,
    ) -> T::WithLt<'outer> {
        self.lag_lead_generic("first_value", value, None, None, over)
    }

    pub fn last_value<T: Table + ForLifetimeTable>(
        &mut self,
        value: T,
        over: Over<'inner>,
    ) -> T::WithLt<'outer> {
        self.lag_lead_generic("first_value", value, None, None, over)
    }

    // needs some thought
    // pub fn nth_value<T: Table>(
    //     &mut self,
    //     value: T,
    //     offset: Option<Expr<'outer, i64>>,
    //     over: Over<'inner>,
    // ) -> NullTable {
    //     self.lag_lead_generic("lag", value, offset, None, over)
    // }

    /// Bring a table in the `'inner` scope to the `'outer` scope.
    ///
    /// Ideally this wouldn't be needed, but the sql builder used internally
    /// doesn't support arbitrary expressions in window function partitions, so
    /// we have to move everything into a subquery. This function is then used
    /// to reselect the input table from this subquery.
    pub fn id<T: Table + ForLifetimeTable>(&mut self, mut table: T) -> T::WithLt<'outer> {
        subst_table(&mut table, self.inner_table.clone(), &mut self.inner_query);
        subst_table(
            &mut table,
            self.middle_table.clone(),
            &mut self.middle_query,
        );
        table.with_lt(&mut WithLtMarker::new())
    }
}

/// A set of helper tables that are equivalent to tuples.
///
/// We need these as an alternative to just tuples in order to be parameterised
/// by the [TableMode].
pub mod helper_tables {
    use super::*;
    use rust_rel8_derive::TableStruct;

    #[derive(TableStruct)]
    #[table(crate = "crate")]
    #[perfect_derive::perfect_derive(Debug, PartialEq, Clone)]
    /// A helper table with one field.
    pub struct One<'scope, Mode: TableMode, #[table(proxy)] A: Value> {
        pub a: Mode::T<'scope, A>,
    }

    #[derive(TableStruct)]
    #[table(crate = "crate")]
    #[perfect_derive::perfect_derive(Debug, PartialEq, Clone)]
    /// A helper table with two fields.
    pub struct Two<'scope, Mode: TableMode, #[table(proxy)] A: Value, #[table(proxy)] B: Value> {
        pub a: Mode::T<'scope, A>,
        pub b: Mode::T<'scope, B>,
    }

    #[derive(TableStruct)]
    #[table(crate = "crate")]
    #[perfect_derive::perfect_derive(Debug, PartialEq, Clone)]
    /// A helper table with three fields.
    pub struct Three<
        'scope,
        Mode: TableMode,
        #[table(proxy)] A: Value,
        #[table(proxy)] B: Value,
        #[table(proxy)] C: Value,
    > {
        pub a: Mode::T<'scope, A>,
        pub b: Mode::T<'scope, B>,
        pub c: Mode::T<'scope, C>,
    }
}

pub mod helper_utilities {
    use std::{collections::HashMap, hash::Hash};

    /// A helper trait which implements `shorten_lifetime` for some common wrapper types.
    pub trait ShortenLifetime {
        type Shortened<'small>
        where
            Self: 'small;

        /// Shorten a lifetime, normally rust does this automatically, but if
        /// the lifetime is invariant due to being used in a Gat or trait, we
        /// need to do it manually.
        ///
        /// If rust complains about a lifetime being invariant, you should call
        /// this method at the use site where the lifetime error is generated.
        fn shorten_lifetime<'small, 'large: 'small>(self) -> Self::Shortened<'small>
        where
            Self: 'large;
    }

    impl<T: ShortenLifetime, const N: usize> ShortenLifetime for [T; N] {
        type Shortened<'small>
            = [T::Shortened<'small>; N]
        where
            Self: 'small;

        fn shorten_lifetime<'small, 'large: 'small>(self) -> Self::Shortened<'small>
        where
            Self: 'large,
        {
            self.map(ShortenLifetime::shorten_lifetime)
        }
    }

    impl<T: ShortenLifetime> ShortenLifetime for Vec<T> {
        type Shortened<'small>
            = Vec<T::Shortened<'small>>
        where
            Self: 'small;

        fn shorten_lifetime<'small, 'large: 'small>(self) -> Self::Shortened<'small>
        where
            Self: 'large,
        {
            self.into_iter()
                .map(ShortenLifetime::shorten_lifetime)
                .collect::<Vec<_>>()
        }
    }

    impl<K: Hash + Eq, T: ShortenLifetime> ShortenLifetime for HashMap<K, T> {
        type Shortened<'small>
            = HashMap<K, T::Shortened<'small>>
        where
            Self: 'small;

        fn shorten_lifetime<'small, 'large: 'small>(self) -> Self::Shortened<'small>
        where
            Self: 'large,
        {
            self.into_iter()
                .map(|(k, v)| (k, v.shorten_lifetime()))
                .collect::<HashMap<_, _>>()
        }
    }
}

pub use helper_utilities::ShortenLifetime;

#[cfg(feature = "derive")]
pub use rust_rel8_derive::TableStruct;

use self::is_nullable::IsNullable;