supabase-wrappers 0.1.27

//! Provides interface types and trait to develop Postgres foreign data wrapper
//!

use crate::FdwRoutine;
use crate::instance::ForeignServer;
use pgrx::pg_sys::panic::ErrorReport;
use pgrx::prelude::{Date, Interval, Time, Timestamp, TimestampWithTimeZone};
use pgrx::{
    AllocatedByRust, AnyNumeric, FromDatum, IntoDatum, JsonB, PgBuiltInOids, PgOid,
    datum::Uuid,
    fcinfo,
    pg_sys::{self, BuiltinOid, Datum, Expr, ExprState, Oid, bytea},
};
use std::collections::HashMap;
use std::ffi::CStr;
use std::fmt;
use std::iter::Zip;
use std::mem;
use std::slice::Iter;
use std::sync::{Arc, Mutex};

// fdw system catalog oids
// https://doxygen.postgresql.org/pg__foreign__data__wrapper_8h.html
// https://doxygen.postgresql.org/pg__foreign__server_8h.html
// https://doxygen.postgresql.org/pg__foreign__table_8h.html

/// Constant can be used in [validator](ForeignDataWrapper::validator)
pub const FOREIGN_DATA_WRAPPER_RELATION_ID: Oid = BuiltinOid::ForeignDataWrapperRelationId.value();

/// Constant can be used in [validator](ForeignDataWrapper::validator)
pub const FOREIGN_SERVER_RELATION_ID: Oid = BuiltinOid::ForeignServerRelationId.value();

/// Constant can be used in [validator](ForeignDataWrapper::validator)
pub const FOREIGN_TABLE_RELATION_ID: Oid = BuiltinOid::ForeignTableRelationId.value();

/// A data cell in a data row
#[derive(Debug)]
pub enum Cell {
    Bool(bool),
    I8(i8),
    I16(i16),
    F32(f32),
    I32(i32),
    F64(f64),
    I64(i64),
    Numeric(AnyNumeric),
    String(String),
    Date(Date),
    Time(Time),
    Timestamp(Timestamp),
    Timestamptz(TimestampWithTimeZone),
    Interval(Interval),
    Json(JsonB),
    Bytea(*mut bytea),
    Uuid(Uuid),
    BoolArray(Vec<Option<bool>>),
    I16Array(Vec<Option<i16>>),
    I32Array(Vec<Option<i32>>),
    I64Array(Vec<Option<i64>>),
    F32Array(Vec<Option<f32>>),
    F64Array(Vec<Option<f64>>),
    StringArray(Vec<Option<String>>),
}

impl Cell {
    /// Check if cell is an array type
    pub fn is_array(&self) -> bool {
        matches!(
            self,
            Cell::BoolArray(_)
                | Cell::I16Array(_)
                | Cell::I32Array(_)
                | Cell::I64Array(_)
                | Cell::F32Array(_)
                | Cell::F64Array(_)
                | Cell::StringArray(_)
        )
    }
}

unsafe impl Send for Cell {}

impl Clone for Cell {
    fn clone(&self) -> Self {
        match self {
            Cell::Bool(v) => Cell::Bool(*v),
            Cell::I8(v) => Cell::I8(*v),
            Cell::I16(v) => Cell::I16(*v),
            Cell::F32(v) => Cell::F32(*v),
            Cell::I32(v) => Cell::I32(*v),
            Cell::F64(v) => Cell::F64(*v),
            Cell::I64(v) => Cell::I64(*v),
            Cell::Numeric(v) => Cell::Numeric(v.clone()),
            Cell::String(v) => Cell::String(v.clone()),
            Cell::Date(v) => Cell::Date(*v),
            Cell::Time(v) => Cell::Time(*v),
            Cell::Timestamp(v) => Cell::Timestamp(*v),
            Cell::Timestamptz(v) => Cell::Timestamptz(*v),
            Cell::Interval(v) => Cell::Interval(*v),
            Cell::Json(v) => Cell::Json(JsonB(v.0.clone())),
            Cell::Bytea(v) => Cell::Bytea(*v),
            Cell::Uuid(v) => Cell::Uuid(*v),
            Cell::BoolArray(v) => Cell::BoolArray(v.clone()),
            Cell::I16Array(v) => Cell::I16Array(v.clone()),
            Cell::I32Array(v) => Cell::I32Array(v.clone()),
            Cell::I64Array(v) => Cell::I64Array(v.clone()),
            Cell::F32Array(v) => Cell::F32Array(v.clone()),
            Cell::F64Array(v) => Cell::F64Array(v.clone()),
            Cell::StringArray(v) => Cell::StringArray(v.clone()),
        }
    }
}

fn write_array<T: std::fmt::Display>(
    array: &[Option<T>],
    f: &mut fmt::Formatter<'_>,
) -> fmt::Result {
    let res = array
        .iter()
        .map(|e| match e {
            Some(val) => format!("{val}",),
            None => "null".to_owned(),
        })
        .collect::<Vec<String>>()
        .join(",");
    write!(f, "[{res}]",)
}

impl fmt::Display for Cell {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Cell::Bool(v) => write!(f, "{v}"),
            Cell::I8(v) => write!(f, "{v}"),
            Cell::I16(v) => write!(f, "{v}"),
            Cell::F32(v) => write!(f, "{v}"),
            Cell::I32(v) => write!(f, "{v}"),
            Cell::F64(v) => write!(f, "{v}"),
            Cell::I64(v) => write!(f, "{v}"),
            Cell::Numeric(v) => write!(f, "{v}"),
            Cell::String(v) => write!(f, "'{v}'"),
            Cell::Date(v) => unsafe {
                let dt =
                    fcinfo::direct_function_call_as_datum(pg_sys::date_out, &[(*v).into_datum()])
                        .expect("cell should be a valid date");
                let dt_cstr = CStr::from_ptr(dt.cast_mut_ptr());
                write!(
                    f,
                    "'{}'",
                    dt_cstr.to_str().expect("date should be a valid string")
                )
            },
            Cell::Time(v) => unsafe {
                let ts =
                    fcinfo::direct_function_call_as_datum(pg_sys::time_out, &[(*v).into_datum()])
                        .expect("cell should be a valid time");
                let ts_cstr = CStr::from_ptr(ts.cast_mut_ptr());
                write!(
                    f,
                    "'{}'",
                    ts_cstr.to_str().expect("time hould be a valid string")
                )
            },
            Cell::Timestamp(v) => unsafe {
                let ts = fcinfo::direct_function_call_as_datum(
                    pg_sys::timestamp_out,
                    &[(*v).into_datum()],
                )
                .expect("cell should be a valid timestamp");
                let ts_cstr = CStr::from_ptr(ts.cast_mut_ptr());
                write!(
                    f,
                    "'{}'",
                    ts_cstr
                        .to_str()
                        .expect("timestamp should be a valid string")
                )
            },
            Cell::Timestamptz(v) => unsafe {
                let ts = fcinfo::direct_function_call_as_datum(
                    pg_sys::timestamptz_out,
                    &[(*v).into_datum()],
                )
                .expect("cell should be a valid timestamptz");
                let ts_cstr = CStr::from_ptr(ts.cast_mut_ptr());
                write!(
                    f,
                    "'{}'",
                    ts_cstr
                        .to_str()
                        .expect("timestamptz should be a valid string")
                )
            },
            Cell::Interval(v) => write!(f, "{v}"),
            Cell::Json(v) => write!(f, "{v:?}"),
            Cell::Bytea(v) => {
                let byte_u8 = unsafe { pgrx::varlena::varlena_to_byte_slice(*v) };
                let hex = byte_u8
                    .iter()
                    .map(|b| format!("{b:02X}"))
                    .collect::<Vec<String>>()
                    .join("");
                if hex.is_empty() {
                    write!(f, "''")
                } else {
                    write!(f, r#"'\x{hex}'"#,)
                }
            }
            Cell::Uuid(v) => write!(f, "'{v}'",),
            Cell::BoolArray(v) => write_array(v, f),
            Cell::I16Array(v) => write_array(v, f),
            Cell::I32Array(v) => write_array(v, f),
            Cell::I64Array(v) => write_array(v, f),
            Cell::F32Array(v) => write_array(v, f),
            Cell::F64Array(v) => write_array(v, f),
            Cell::StringArray(v) => write_array(v, f),
        }
    }
}

impl IntoDatum for Cell {
    fn into_datum(self) -> Option<Datum> {
        match self {
            Cell::Bool(v) => v.into_datum(),
            Cell::I8(v) => v.into_datum(),
            Cell::I16(v) => v.into_datum(),
            Cell::F32(v) => v.into_datum(),
            Cell::I32(v) => v.into_datum(),
            Cell::F64(v) => v.into_datum(),
            Cell::I64(v) => v.into_datum(),
            Cell::Numeric(v) => v.into_datum(),
            Cell::String(v) => v.into_datum(),
            Cell::Date(v) => v.into_datum(),
            Cell::Time(v) => v.into_datum(),
            Cell::Timestamp(v) => v.into_datum(),
            Cell::Timestamptz(v) => v.into_datum(),
            Cell::Interval(v) => v.into_datum(),
            Cell::Json(v) => v.into_datum(),
            Cell::Bytea(v) => Some(Datum::from(v)),
            Cell::Uuid(v) => v.into_datum(),
            Cell::BoolArray(v) => v.into_datum(),
            Cell::I16Array(v) => v.into_datum(),
            Cell::I32Array(v) => v.into_datum(),
            Cell::I64Array(v) => v.into_datum(),
            Cell::F32Array(v) => v.into_datum(),
            Cell::F64Array(v) => v.into_datum(),
            Cell::StringArray(v) => v.into_datum(),
        }
    }

    fn type_oid() -> Oid {
        Oid::INVALID
    }

    fn is_compatible_with(other: Oid) -> bool {
        Self::type_oid() == other
            || other == pg_sys::BOOLOID
            || other == pg_sys::CHAROID
            || other == pg_sys::INT2OID
            || other == pg_sys::FLOAT4OID
            || other == pg_sys::INT4OID
            || other == pg_sys::FLOAT8OID
            || other == pg_sys::INT8OID
            || other == pg_sys::NUMERICOID
            || other == pg_sys::TEXTOID
            || other == pg_sys::DATEOID
            || other == pg_sys::TIMEOID
            || other == pg_sys::TIMESTAMPOID
            || other == pg_sys::TIMESTAMPTZOID
            || other == pg_sys::INTERVALOID
            || other == pg_sys::JSONBOID
            || other == pg_sys::BYTEAOID
            || other == pg_sys::UUIDOID
            || other == pg_sys::BOOLARRAYOID
            || other == pg_sys::INT2ARRAYOID
            || other == pg_sys::INT4ARRAYOID
            || other == pg_sys::INT8ARRAYOID
            || other == pg_sys::FLOAT4ARRAYOID
            || other == pg_sys::FLOAT8ARRAYOID
            || other == pg_sys::TEXTARRAYOID
    }
}

impl FromDatum for Cell {
    unsafe fn from_polymorphic_datum(datum: Datum, is_null: bool, typoid: Oid) -> Option<Self>
    where
        Self: Sized,
    {
        unsafe {
            let oid = PgOid::from(typoid);
            match oid {
                PgOid::BuiltIn(PgBuiltInOids::BOOLOID) => {
                    bool::from_datum(datum, is_null).map(Cell::Bool)
                }
                PgOid::BuiltIn(PgBuiltInOids::CHAROID) => {
                    i8::from_datum(datum, is_null).map(Cell::I8)
                }
                PgOid::BuiltIn(PgBuiltInOids::INT2OID) => {
                    i16::from_datum(datum, is_null).map(Cell::I16)
                }
                PgOid::BuiltIn(PgBuiltInOids::FLOAT4OID) => {
                    f32::from_datum(datum, is_null).map(Cell::F32)
                }
                PgOid::BuiltIn(PgBuiltInOids::INT4OID) => {
                    i32::from_datum(datum, is_null).map(Cell::I32)
                }
                PgOid::BuiltIn(PgBuiltInOids::FLOAT8OID) => {
                    f64::from_datum(datum, is_null).map(Cell::F64)
                }
                PgOid::BuiltIn(PgBuiltInOids::INT8OID) => {
                    i64::from_datum(datum, is_null).map(Cell::I64)
                }
                PgOid::BuiltIn(PgBuiltInOids::NUMERICOID) => {
                    AnyNumeric::from_datum(datum, is_null).map(Cell::Numeric)
                }
                PgOid::BuiltIn(PgBuiltInOids::TEXTOID) => {
                    String::from_datum(datum, is_null).map(Cell::String)
                }
                PgOid::BuiltIn(PgBuiltInOids::DATEOID) => {
                    Date::from_datum(datum, is_null).map(Cell::Date)
                }
                PgOid::BuiltIn(PgBuiltInOids::TIMEOID) => {
                    Time::from_datum(datum, is_null).map(Cell::Time)
                }
                PgOid::BuiltIn(PgBuiltInOids::TIMESTAMPOID) => {
                    Timestamp::from_datum(datum, is_null).map(Cell::Timestamp)
                }
                PgOid::BuiltIn(PgBuiltInOids::TIMESTAMPTZOID) => {
                    TimestampWithTimeZone::from_datum(datum, is_null).map(Cell::Timestamptz)
                }
                PgOid::BuiltIn(PgBuiltInOids::INTERVALOID) => {
                    Interval::from_datum(datum, is_null).map(Cell::Interval)
                }
                PgOid::BuiltIn(PgBuiltInOids::JSONBOID) => {
                    JsonB::from_datum(datum, is_null).map(Cell::Json)
                }
                PgOid::BuiltIn(PgBuiltInOids::BYTEAOID) => {
                    if is_null {
                        None
                    } else {
                        Some(Cell::Bytea(datum.cast_mut_ptr::<bytea>()))
                    }
                }
                PgOid::BuiltIn(PgBuiltInOids::UUIDOID) => {
                    Uuid::from_datum(datum, is_null).map(Cell::Uuid)
                }
                PgOid::BuiltIn(PgBuiltInOids::BOOLARRAYOID) => {
                    Vec::<Option<bool>>::from_datum(datum, false).map(Cell::BoolArray)
                }
                PgOid::BuiltIn(PgBuiltInOids::INT2ARRAYOID) => {
                    Vec::<Option<i16>>::from_datum(datum, false).map(Cell::I16Array)
                }
                PgOid::BuiltIn(PgBuiltInOids::INT4ARRAYOID) => {
                    Vec::<Option<i32>>::from_datum(datum, false).map(Cell::I32Array)
                }
                PgOid::BuiltIn(PgBuiltInOids::INT8ARRAYOID) => {
                    Vec::<Option<i64>>::from_datum(datum, false).map(Cell::I64Array)
                }
                PgOid::BuiltIn(PgBuiltInOids::FLOAT4ARRAYOID) => {
                    Vec::<Option<f32>>::from_datum(datum, false).map(Cell::F32Array)
                }
                PgOid::BuiltIn(PgBuiltInOids::FLOAT8ARRAYOID) => {
                    Vec::<Option<f64>>::from_datum(datum, false).map(Cell::F64Array)
                }
                PgOid::BuiltIn(PgBuiltInOids::TEXTARRAYOID) => {
                    Vec::<Option<String>>::from_datum(datum, false).map(Cell::StringArray)
                }
                PgOid::Custom(_) => {
                    if is_null {
                        None
                    } else {
                        Some(Cell::Bytea(datum.cast_mut_ptr::<bytea>()))
                    }
                }
                _ => None,
            }
        }
    }
}

pub trait CellFormatter {
    fn fmt_cell(&mut self, cell: &Cell) -> String;
}

struct DefaultFormatter {}

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

impl CellFormatter for DefaultFormatter {
    fn fmt_cell(&mut self, cell: &Cell) -> String {
        format!("{cell}",)
    }
}

/// A data row in a table
///
/// The row contains a column name list and cell list with same number of
/// elements.
#[derive(Debug, Clone, Default)]
pub struct Row {
    /// column names
    pub cols: Vec<String>,

    /// column cell list, should match with cols
    pub cells: Vec<Option<Cell>>,
}

impl Row {
    /// Create an empty row
    pub fn new() -> Self {
        Self::default()
    }

    /// Push a cell with column name to this row
    pub fn push(&mut self, col: &str, cell: Option<Cell>) {
        self.cols.push(col.to_owned());
        self.cells.push(cell);
    }

    /// Return a zipped <column_name, cell> iterator
    pub fn iter(&self) -> Zip<Iter<'_, String>, Iter<'_, Option<Cell>>> {
        self.cols.iter().zip(self.cells.iter())
    }

    /// Remove a cell at the specified index
    pub fn retain<F>(&mut self, f: F)
    where
        F: FnMut((&String, &Option<Cell>)) -> bool,
    {
        let keep: Vec<bool> = self.iter().map(f).collect();
        let mut iter = keep.iter();
        self.cols.retain(|_| *iter.next().unwrap_or(&true));
        iter = keep.iter();
        self.cells.retain(|_| *iter.next().unwrap_or(&true));
    }

    /// Replace `self` with the source row
    #[inline]
    pub fn replace_with(&mut self, src: Row) {
        let _ = mem::replace(self, src);
    }

    /// Clear the row, removing all column names and cells
    pub fn clear(&mut self) {
        self.cols.clear();
        self.cells.clear();
    }
}

/// A column definition in a table
///
/// The column represents a column definition in a table.
#[derive(Debug, Clone, Default)]
pub struct Column {
    /// column name
    pub name: String,

    /// 1-based column number
    pub num: usize,

    /// column type OID, can be used to match pg_sys::BuiltinOid
    pub type_oid: Oid,
}

/// A restiction value used in [`Qual`], either a [`Cell`] or an array of [`Cell`]
#[derive(Debug, Clone)]
pub enum Value {
    Cell(Cell),
    Array(Vec<Cell>),
}

// Struct for parameter expression value evaluation
#[derive(Debug, Clone)]
pub(super) struct ExprEval {
    pub(super) expr: *mut Expr,
    pub(super) expr_state: *mut ExprState,
}

unsafe impl Send for ExprEval {}

/// Query parameter
#[derive(Debug, Clone)]
pub struct Param {
    /// parameter kind
    pub kind: pg_sys::ParamKind::Type,

    /// if the parameter kind is `PARAM_EXTERN`, parameter id is from 1 to n,
    /// otherwise the parameter id is zero
    pub id: usize,

    /// parameter type OID
    pub type_oid: Oid,

    /// parameter value which is evaluated during query execution
    pub eval_value: Arc<Mutex<Option<Value>>>,

    // internal variables for expression evaluation
    pub(super) expr_eval: ExprEval,
}

/// Query restrictions, a.k.a conditions in `WHERE` clause
///
/// A Qual defines a simple condition wich can be used by the FDW to restrict the number
/// of the results.
///
/// <div class="example-wrap" style="display:inline-block"><pre class="compile_fail" style="white-space:normal;font:inherit;">
/// <strong>Warning</strong>: Currently only simple conditions are supported, see below for examples. Other kinds of conditions, like JSON attribute filter e.g. `where json_col->>'key' = 'value'`, are not supported yet.
/// </pre></div>
///
/// ## Examples
///
/// ```sql
/// where id = 1;
/// -- [Qual { field: "id", operator: "=", value: Cell(I32(1)), use_or: false }]
/// ```
///
/// ```sql
/// where id in (1, 2);
/// -- [Qual { field: "id", operator: "=", value: Array([I64(1), I64(2)]), use_or: true }]
/// ```
///
/// ```sql
/// where col is null
/// -- [Qual { field: "col", operator: "is", value: Cell(String("null")), use_or: false }]
/// ```
///
/// ```sql
/// where bool_col
/// -- [Qual { field: "bool_col", operator: "=", value: Cell(Bool(true)), use_or: false }]
/// ```
///
/// ```sql
/// where bool_col is true
/// -- [Qual { field: "bool_col", operator: "is", value: Cell(Bool(true)), use_or: false }]
/// ```
///
/// ```sql
/// where id > 1 and col = 'foo';
/// -- [
/// --   Qual { field: "id", operator: ">", value: Cell(I32(1)), use_or: false },
/// --   Qual { field: "col", operator: "=", value: Cell(String("foo")), use_or: false }
/// -- ]
/// ```
#[derive(Debug, Clone)]
pub struct Qual {
    pub field: String,
    pub operator: String,
    pub value: Value,
    pub use_or: bool,
    pub param: Option<Param>,
}

impl Qual {
    pub fn deparse(&self) -> String {
        let mut formatter = DefaultFormatter::new();
        self.deparse_with_fmt(&mut formatter)
    }

    pub fn deparse_with_fmt<T: CellFormatter>(&self, t: &mut T) -> String {
        if self.use_or {
            match &self.value {
                Value::Cell(_) => unreachable!(),
                Value::Array(cells) => {
                    let conds: Vec<String> = cells
                        .iter()
                        .map(|cell| {
                            format!("{} {} {}", self.field, self.operator, t.fmt_cell(cell))
                        })
                        .collect();
                    conds.join(" or ")
                }
            }
        } else {
            match &self.value {
                Value::Cell(cell) => match self.operator.as_str() {
                    "is" | "is not" => match cell {
                        Cell::String(cell) if cell == "null" => {
                            format!("{} {} null", self.field, self.operator)
                        }
                        _ => format!("{} {} {}", self.field, self.operator, t.fmt_cell(cell)),
                    },
                    "~~" => format!("{} like {}", self.field, t.fmt_cell(cell)),
                    "!~~" => format!("{} not like {}", self.field, t.fmt_cell(cell)),
                    _ => format!("{} {} {}", self.field, self.operator, t.fmt_cell(cell)),
                },
                Value::Array(_) => unreachable!(),
            }
        }
    }
}

/// Query sort, a.k.a `ORDER BY` clause
///
/// ## Examples
///
/// ```sql
/// order by id;
/// -- [Sort { field: "id", field_no: 1, reversed: false, nulls_first: false, collate: None]
/// ```
///
/// ```sql
/// order by id desc;
/// -- [Sort { field: "id", field_no: 1, reversed: true, nulls_first: true, collate: None]
/// ```
///
/// ```sql
/// order by id desc, col;
/// -- [
/// --   Sort { field: "id", field_no: 1, reversed: true, nulls_first: true, collate: None },
/// --   Sort { field: "col", field_no: 2, reversed: false, nulls_first: false, collate: None }
/// -- ]
/// ```
///
/// ```sql
/// order by id collate "de_DE";
/// -- [Sort { field: "col", field_no: 2, reversed: false, nulls_first: false, collate: Some("de_DE") }]
/// ```
#[derive(Debug, Clone, Default)]
pub struct Sort {
    pub field: String,
    pub field_no: usize,
    pub reversed: bool,
    pub nulls_first: bool,
    pub collate: Option<String>,
}

impl Sort {
    pub fn deparse(&self) -> String {
        let mut sql = self.field.to_string();

        if self.reversed {
            sql.push_str(" desc");
        } else {
            sql.push_str(" asc");
        }

        if self.nulls_first {
            sql.push_str(" nulls first")
        } else {
            sql.push_str(" nulls last")
        }

        sql
    }

    pub fn deparse_with_collate(&self) -> String {
        let mut sql = self.deparse();

        if let Some(collate) = &self.collate {
            sql.push_str(&format!(" collate {collate}"));
        }

        sql
    }
}

/// Query limit, a.k.a `LIMIT count OFFSET offset` clause
///
/// ## Examples
///
/// ```sql
/// limit 42;
/// -- Limit { count: 42, offset: 0 }
/// ```
///
/// ```sql
/// limit 42 offset 7;
/// -- Limit { count: 42, offset: 7 }
/// ```
#[derive(Debug, Clone, Default)]
pub struct Limit {
    pub count: i64,
    pub offset: i64,
}

impl Limit {
    pub fn deparse(&self) -> String {
        format!("limit {} offset {}", self.count, self.offset)
    }
}

/// Represents the type of aggregate function for pushdown
///
/// This enum is used to declare which aggregate functions an FDW supports
/// for remote execution via [`ForeignDataWrapper::supported_aggregates`].
///
/// ## Examples
///
/// ```rust,no_run
/// use supabase_wrappers::prelude::*;
///
/// fn get_supported() -> Vec<AggregateKind> {
///     vec![
///         AggregateKind::Count,
///         AggregateKind::Sum,
///         AggregateKind::Avg,
///     ]
/// }
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AggregateKind {
    /// COUNT(*) - count all rows
    Count,
    /// COUNT(column) - count non-null values in a column
    CountColumn,
    /// SUM(column) - sum of values
    Sum,
    /// AVG(column) - average of values
    Avg,
    /// MIN(column) - minimum value
    Min,
    /// MAX(column) - maximum value
    Max,
}

impl AggregateKind {
    /// Returns the SQL function name for this aggregate kind
    ///
    /// ## Examples
    ///
    /// ```rust,no_run
    /// use supabase_wrappers::prelude::*;
    ///
    /// assert_eq!(AggregateKind::Count.sql_name(), "COUNT");
    /// assert_eq!(AggregateKind::Sum.sql_name(), "SUM");
    /// ```
    pub fn sql_name(&self) -> &'static str {
        match self {
            AggregateKind::Count => "COUNT",
            AggregateKind::CountColumn => "COUNT",
            AggregateKind::Sum => "SUM",
            AggregateKind::Avg => "AVG",
            AggregateKind::Min => "MIN",
            AggregateKind::Max => "MAX",
        }
    }
}

/// Represents a single aggregate operation in a pushed-down query
///
/// This struct contains all information needed for an FDW to execute an
/// aggregate function remotely.
///
/// ## Examples
///
/// ```rust,no_run
/// use supabase_wrappers::prelude::*;
///
/// // COUNT(*) aggregate
/// let count_all = Aggregate {
///     kind: AggregateKind::Count,
///     column: None,
///     distinct: false,
///     alias: "count".to_string(),
///     type_oid: pgrx::pg_sys::INT8OID,
/// };
///
/// // SUM(amount) aggregate
/// let sum_amount = Aggregate {
///     kind: AggregateKind::Sum,
///     column: Some(Column {
///         name: "amount".to_string(),
///         num: 1,
///         type_oid: pgrx::pg_sys::FLOAT8OID,
///     }),
///     distinct: false,
///     alias: "total_amount".to_string(),
///     type_oid: pgrx::pg_sys::FLOAT8OID,
/// };
/// ```
#[derive(Debug, Clone)]
pub struct Aggregate {
    /// Type of aggregate function
    pub kind: AggregateKind,

    /// Column being aggregated (None for COUNT(*))
    pub column: Option<Column>,

    /// Whether DISTINCT modifier is applied (e.g., COUNT(DISTINCT col))
    pub distinct: bool,

    /// Output column name/alias for the aggregate result
    pub alias: String,

    /// Result type OID of the aggregate function (from Aggref::aggtype)
    pub type_oid: Oid,
}

impl Aggregate {
    /// Deparses the aggregate into SQL syntax
    ///
    /// ## Examples
    ///
    /// ```rust,no_run
    /// use supabase_wrappers::prelude::*;
    ///
    /// let count_all = Aggregate {
    ///     kind: AggregateKind::Count,
    ///     column: None,
    ///     distinct: false,
    ///     alias: "cnt".to_string(),
    ///     type_oid: pgrx::pg_sys::INT8OID,
    /// };
    /// assert_eq!(count_all.deparse(), "COUNT(*)");
    ///
    /// let sum_col = Aggregate {
    ///     kind: AggregateKind::Sum,
    ///     column: Some(Column { name: "price".to_string(), num: 1, type_oid: pgrx::pg_sys::Oid::INVALID }),
    ///     distinct: false,
    ///     alias: "total".to_string(),
    ///     type_oid: pgrx::pg_sys::FLOAT8OID,
    /// };
    /// assert_eq!(sum_col.deparse(), "SUM(price)");
    /// ```
    pub fn deparse(&self) -> String {
        let func_name = self.kind.sql_name();
        match self.kind {
            AggregateKind::Count => format!("{func_name}(*)"),
            _ => {
                let col_name = self.column.as_ref().map(|c| c.name.as_str()).unwrap_or("*");
                if self.distinct {
                    format!("{func_name}(DISTINCT {col_name})")
                } else {
                    format!("{func_name}({col_name})")
                }
            }
        }
    }

    /// Deparses the aggregate with its alias for use in SELECT
    ///
    /// ## Examples
    ///
    /// ```rust,no_run
    /// use supabase_wrappers::prelude::*;
    ///
    /// let count_all = Aggregate {
    ///     kind: AggregateKind::Count,
    ///     column: None,
    ///     distinct: false,
    ///     alias: "cnt".to_string(),
    ///     type_oid: pgrx::pg_sys::INT8OID,
    /// };
    /// assert_eq!(count_all.deparse_with_alias(), "COUNT(*) AS cnt");
    /// ```
    pub fn deparse_with_alias(&self) -> String {
        format!("{} AS {}", self.deparse(), self.alias)
    }
}

/// The Foreign Data Wrapper trait
///
/// This is the main interface for your foreign data wrapper. Required functions
/// are listed below, all the others are optional.
///
/// 1. new
/// 2. begin_scan
/// 3. iter_scan
/// 4. end_scan
///
/// See the module-level document for more details.
///
pub trait ForeignDataWrapper<E: Into<ErrorReport>> {
    /// Create a FDW instance
    ///
    /// `options` is the key-value pairs defined in `CREATE SERVER` SQL. For example,
    ///
    /// ```sql
    /// create server my_helloworld_server
    ///   foreign data wrapper wrappers_helloworld
    ///   options (
    ///     foo 'bar'
    /// );
    /// ```
    ///
    /// `options` passed here will be a hashmap { 'foo' -> 'bar' }.
    ///
    /// You can do any initalization in this function, like saving connection
    /// info or API url in an variable, but don't do heavy works like database
    /// connection or API call.
    fn new(server: ForeignServer) -> Result<Self, E>
    where
        Self: Sized;

    /// Obtain relation size estimates for a foreign table
    ///
    /// Return the expected number of rows and row size (in bytes) by the
    /// foreign table scan.
    ///
    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-SCAN).
    fn get_rel_size(
        &mut self,
        _quals: &[Qual],
        _columns: &[Column],
        _sorts: &[Sort],
        _limit: &Option<Limit>,
        _options: &HashMap<String, String>,
    ) -> Result<(i64, i32), E> {
        Ok((0, 0))
    }

    /// Called when begin executing a foreign scan
    ///
    /// - `quals` - `WHERE` clause pushed down
    /// - `columns` - target columns to be queried
    /// - `sorts` - `ORDER BY` clause pushed down
    /// - `limit` - `LIMIT` clause pushed down
    /// - `options` - the options defined when `CREATE FOREIGN TABLE`
    ///
    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-SCAN).
    fn begin_scan(
        &mut self,
        quals: &[Qual],
        columns: &[Column],
        sorts: &[Sort],
        limit: &Option<Limit>,
        options: &HashMap<String, String>,
    ) -> Result<(), E>;

    /// Called when fetch one row from the foreign source
    ///
    /// FDW must save fetched foreign data into the [`Row`], or return `None` if no more rows to read.
    ///
    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-SCAN).
    fn iter_scan(&mut self, row: &mut Row) -> Result<Option<()>, E>;

    /// Called when restart the scan from the beginning.
    ///
    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-SCAN).
    fn re_scan(&mut self) -> Result<(), E> {
        Ok(())
    }

    /// Called when end the scan
    ///
    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-SCAN).
    fn end_scan(&mut self) -> Result<(), E>;

    /// Called when begin executing a foreign table modification operation.
    ///
    /// - `options` - the options defined when `CREATE FOREIGN TABLE`
    ///
    /// The foreign table must include a `rowid_column` option which specify
    /// the unique identification column of the foreign table to enable data
    /// modification.
    ///
    /// For example,
    ///
    /// ```sql
    /// create foreign table my_foreign_table (
    ///   id bigint,
    ///   name text
    /// )
    ///   server my_server
    ///   options (
    ///     rowid_column 'id'
    ///   );
    /// ```
    ///
    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-UPDATE).
    fn begin_modify(&mut self, _options: &HashMap<String, String>) -> Result<(), E> {
        Ok(())
    }

    /// Called when insert one row into the foreign table
    ///
    /// - row - the new row to be inserted
    ///
    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-UPDATE).
    fn insert(&mut self, _row: &Row) -> Result<(), E> {
        Ok(())
    }

    /// Called when update one row into the foreign table
    ///
    /// - rowid - the `rowid_column` cell
    /// - new_row - the new row with updated cells
    ///
    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-UPDATE).
    fn update(&mut self, _rowid: &Cell, _new_row: &Row) -> Result<(), E> {
        Ok(())
    }

    /// Called when delete one row into the foreign table
    ///
    /// - rowid - the `rowid_column` cell
    ///
    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-UPDATE).
    fn delete(&mut self, _rowid: &Cell) -> Result<(), E> {
        Ok(())
    }

    /// Called when end the table update
    ///
    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-UPDATE).
    fn end_modify(&mut self) -> Result<(), E> {
        Ok(())
    }

    // =========================================================================
    // Aggregate Pushdown Methods
    // =========================================================================

    /// Returns the list of aggregate functions this FDW can push down to the
    /// remote data source.
    ///
    /// Default implementation returns an empty vector (no aggregate pushdown).
    /// Override this method to enable aggregate pushdown for your FDW.
    ///
    /// When this method returns a non-empty vector, the framework will:
    /// 1. Check if the query's aggregates are all supported
    /// 2. Call [`get_aggregate_rel_size`](Self::get_aggregate_rel_size) for cost estimation
    /// 3. Create an aggregate path in the query planner
    /// 4. Call [`begin_aggregate_scan`](Self::begin_aggregate_scan) if the aggregate path is chosen
    ///
    /// ## Examples
    ///
    /// ```rust,ignore
    /// use supabase_wrappers::prelude::*;
    ///
    /// fn supported_aggregates(&self) -> Vec<AggregateKind> {
    ///     vec![
    ///         AggregateKind::Count,
    ///         AggregateKind::CountColumn,
    ///         AggregateKind::Sum,
    ///         AggregateKind::Avg,
    ///         AggregateKind::Min,
    ///         AggregateKind::Max,
    ///     ]
    /// }
    /// ```
    fn supported_aggregates(&self) -> Vec<AggregateKind> {
        vec![]
    }

    /// Returns whether this FDW supports GROUP BY pushdown alongside aggregates.
    ///
    /// Only relevant if [`supported_aggregates`](Self::supported_aggregates) returns a non-empty vector.
    /// Default implementation returns `false`.
    ///
    /// When `true`, GROUP BY columns will be passed to [`begin_aggregate_scan`](Self::begin_aggregate_scan).
    ///
    /// ## Examples
    ///
    /// ```rust,ignore
    /// fn supports_group_by(&self) -> bool {
    ///     true
    /// }
    /// ```
    fn supports_group_by(&self) -> bool {
        false
    }

    /// Estimate the size of aggregate query results for query planning.
    ///
    /// Called during query planning when aggregate pushdown is being considered.
    /// Returns `(estimated_rows, mean_row_width_bytes)`.
    ///
    /// Default implementation estimates:
    /// - 1 row for ungrouped aggregates (e.g., `SELECT COUNT(*) FROM t`)
    /// - 100 rows for grouped aggregates (e.g., `SELECT dept, COUNT(*) FROM t GROUP BY dept`)
    ///
    /// Override this method if you can provide better estimates based on your
    /// knowledge of the remote data source.
    ///
    /// ## Parameters
    ///
    /// - `aggregates`: List of aggregate operations to be computed
    /// - `group_by`: Columns in GROUP BY clause (empty if none)
    /// - `quals`: WHERE clause conditions
    /// - `options`: Foreign table options
    ///
    /// ## Examples
    ///
    /// ```rust,ignore
    /// fn get_aggregate_rel_size(
    ///     &mut self,
    ///     aggregates: &[Aggregate],
    ///     group_by: &[Column],
    ///     quals: &[Qual],
    ///     options: &HashMap<String, String>,
    /// ) -> Result<(i64, i32), MyFdwError> {
    ///     let rows = if group_by.is_empty() { 1 } else { 50 };
    ///     let width = 8 * aggregates.len() as i32;
    ///     Ok((rows, width))
    /// }
    /// ```
    fn get_aggregate_rel_size(
        &mut self,
        aggregates: &[Aggregate],
        group_by: &[Column],
        _quals: &[Qual],
        _options: &HashMap<String, String>,
    ) -> Result<(i64, i32), E> {
        // Default: 1 row if no GROUP BY, otherwise estimate 100 rows
        let rows = if group_by.is_empty() { 1 } else { 100 };
        let width = 8 * aggregates.len() as i32;
        Ok((rows, width))
    }

    /// Called when beginning execution of a pushed-down aggregate query.
    ///
    /// This method is called instead of [`begin_scan`](Self::begin_scan) when
    /// the query planner chooses the aggregate pushdown path. The FDW should
    /// prepare to return aggregate results based on the provided parameters.
    ///
    /// After this method returns successfully, [`iter_scan`](Self::iter_scan)
    /// will be called to retrieve the aggregate results.
    ///
    /// ## Parameters
    ///
    /// - `aggregates`: List of aggregate operations to compute
    /// - `group_by`: Columns to group by (empty for ungrouped aggregates)
    /// - `quals`: WHERE clause conditions to apply before aggregation
    /// - `options`: Foreign table options
    ///
    /// ## Result Row Format
    ///
    /// The rows returned by [`iter_scan`](Self::iter_scan) should contain:
    /// 1. GROUP BY column values (in order specified)
    /// 2. Aggregate results (in order specified, using `alias` as column name)
    ///
    /// ## Examples
    ///
    /// ```rust,ignore
    /// fn begin_aggregate_scan(
    ///     &mut self,
    ///     aggregates: &[Aggregate],
    ///     group_by: &[Column],
    ///     quals: &[Qual],
    ///     options: &HashMap<String, String>,
    /// ) -> Result<(), MyFdwError> {
    ///     // Build remote query with aggregates
    ///     let mut select_items = Vec::new();
    ///
    ///     // Add GROUP BY columns
    ///     for col in group_by {
    ///         select_items.push(col.name.clone());
    ///     }
    ///
    ///     // Add aggregate expressions
    ///     for agg in aggregates {
    ///         select_items.push(agg.deparse_with_alias());
    ///     }
    ///
    ///     let query = format!("SELECT {} FROM ...", select_items.join(", "));
    ///
    ///     // Execute and store results for iter_scan
    ///     self.results = self.execute_query(&query)?;
    ///     Ok(())
    /// }
    /// ```
    fn begin_aggregate_scan(
        &mut self,
        _aggregates: &[Aggregate],
        _group_by: &[Column],
        _quals: &[Qual],
        _options: &HashMap<String, String>,
    ) -> Result<(), E> {
        // This should not be called unless supported_aggregates() returns non-empty.
        // If we reach here, the FDW declared aggregate support but didn't override
        // this method — abort the transaction to prevent wrong results.
        crate::utils::report_error(
            pgrx::PgSqlErrorCode::ERRCODE_FDW_ERROR,
            "begin_aggregate_scan() not implemented; \
             override this method when supported_aggregates() is non-empty",
        );
        unreachable!()
    }

    /// Obtain a list of foreign table creation commands
    ///
    /// Return a list of string, each of which must contain a CREATE FOREIGN TABLE
    /// which will be executed by the core server.
    ///
    /// [See more details](https://www.postgresql.org/docs/current/fdw-callbacks.html#FDW-CALLBACKS-IMPORT).
    fn import_foreign_schema(
        &mut self,
        _stmt: crate::import_foreign_schema::ImportForeignSchemaStmt,
    ) -> Result<Vec<String>, E> {
        Ok(Vec::new())
    }

    /// Returns a FdwRoutine for the FDW
    ///
    /// Not to be used directly, use [`wrappers_fdw`](crate::wrappers_fdw) macro instead.
    fn fdw_routine() -> FdwRoutine
    where
        Self: Sized,
    {
        unsafe {
            use crate::{import_foreign_schema, modify, scan};
            let mut fdw_routine =
                FdwRoutine::<AllocatedByRust>::alloc_node(pg_sys::NodeTag::T_FdwRoutine);

            // import foreign schema
            fdw_routine.ImportForeignSchema =
                Some(import_foreign_schema::import_foreign_schema::<E, Self>);

            // plan phase
            fdw_routine.GetForeignRelSize = Some(scan::get_foreign_rel_size::<E, Self>);
            fdw_routine.GetForeignPaths = Some(scan::get_foreign_paths::<E, Self>);
            fdw_routine.GetForeignPlan = Some(scan::get_foreign_plan::<E, Self>);
            fdw_routine.ExplainForeignScan = Some(scan::explain_foreign_scan::<E, Self>);

            // upper path planning (aggregate pushdown)
            #[cfg(any(
                feature = "pg13",
                feature = "pg14",
                feature = "pg15",
                feature = "pg16",
                feature = "pg17",
                feature = "pg18"
            ))]
            {
                use crate::upper;
                fdw_routine.GetForeignUpperPaths = Some(upper::get_foreign_upper_paths::<E, Self>);
            }

            // scan phase
            fdw_routine.BeginForeignScan = Some(scan::begin_foreign_scan::<E, Self>);
            fdw_routine.IterateForeignScan = Some(scan::iterate_foreign_scan::<E, Self>);
            fdw_routine.ReScanForeignScan = Some(scan::re_scan_foreign_scan::<E, Self>);
            fdw_routine.EndForeignScan = Some(scan::end_foreign_scan::<E, Self>);

            // modify phase
            fdw_routine.AddForeignUpdateTargets = Some(modify::add_foreign_update_targets);
            fdw_routine.PlanForeignModify = Some(modify::plan_foreign_modify::<E, Self>);
            fdw_routine.BeginForeignModify = Some(modify::begin_foreign_modify::<E, Self>);
            fdw_routine.ExecForeignInsert = Some(modify::exec_foreign_insert::<E, Self>);
            fdw_routine.ExecForeignDelete = Some(modify::exec_foreign_delete::<E, Self>);
            fdw_routine.ExecForeignUpdate = Some(modify::exec_foreign_update::<E, Self>);
            fdw_routine.EndForeignModify = Some(modify::end_foreign_modify::<E, Self>);

            Self::fdw_routine_hook(&mut fdw_routine);
            fdw_routine.into_pg_boxed()
        }
    }

    /// Additional FwdRoutine setup, called by default `Self::fdw_routine()`
    /// after completing its initialization.
    fn fdw_routine_hook(_routine: &mut FdwRoutine<AllocatedByRust>) {}

    /// Validator function for validating options given in `CREATE` and `ALTER`
    /// commands for its foreign data wrapper, as well as foreign servers, user
    /// mappings, and foreign tables using the wrapper.
    ///
    /// [See more details about validator](https://www.postgresql.org/docs/current/fdw-functions.html)
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use pgrx::pg_sys::Oid;
    /// use supabase_wrappers::prelude::check_options_contain;
    ///
    /// use pgrx::pg_sys::panic::ErrorReport;
    /// use pgrx::PgSqlErrorCode;
    ///
    /// enum FdwError {
    ///     InvalidFdwOption,
    ///     InvalidServerOption,
    ///     InvalidTableOption,
    /// }
    ///
    /// impl From<FdwError> for ErrorReport {
    ///     fn from(value: FdwError) -> Self {
    ///         let error_message = match value {
    ///             FdwError::InvalidFdwOption => "invalid foreign data wrapper option",
    ///             FdwError::InvalidServerOption => "invalid foreign server option",
    ///             FdwError::InvalidTableOption => "invalid foreign table option",
    ///         };
    ///         ErrorReport::new(PgSqlErrorCode::ERRCODE_FDW_ERROR, error_message, "")
    ///     }
    /// }
    ///
    /// fn validator(opt_list: Vec<Option<String>>, catalog: Option<Oid>) -> Result<(), FdwError> {
    ///     if let Some(oid) = catalog {
    ///         match oid {
    ///             FOREIGN_DATA_WRAPPER_RELATION_ID => {
    ///                 // check a required option when create foreign data wrapper
    ///                 check_options_contain(&opt_list, "foreign_data_wrapper_required_option");
    ///             }
    ///             FOREIGN_SERVER_RELATION_ID => {
    ///                 // check option here when create server
    ///                 check_options_contain(&opt_list, "foreign_server_required_option");
    ///             }
    ///             FOREIGN_TABLE_RELATION_ID => {
    ///                 // check option here when create foreign table
    ///                 check_options_contain(&opt_list, "foreign_table_required_option");
    ///             }
    ///             _ => {}
    ///         }
    ///     }
    ///
    ///     Ok(())
    /// }
    /// ```
    fn validator(_options: Vec<Option<String>>, _catalog: Option<Oid>) -> Result<(), E> {
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn assert_cell_clone(cell: Cell) {
        let cell_clone = cell.clone();

        match (cell, cell_clone) {
            (Cell::Bool(left), Cell::Bool(right)) => assert_eq!(left, right),
            (Cell::I8(left), Cell::I8(right)) => assert_eq!(left, right),
            (Cell::I16(left), Cell::I16(right)) => assert_eq!(left, right),
            (Cell::F32(left), Cell::F32(right)) => assert_eq!(left, right),
            (Cell::I32(left), Cell::I32(right)) => assert_eq!(left, right),
            (Cell::F64(left), Cell::F64(right)) => assert_eq!(left, right),
            (Cell::I64(left), Cell::I64(right)) => assert_eq!(left, right),
            (Cell::String(left), Cell::String(right)) => assert_eq!(left, right),
            (Cell::BoolArray(left), Cell::BoolArray(right)) => assert_eq!(left, right),
            (Cell::I16Array(left), Cell::I16Array(right)) => assert_eq!(left, right),
            (Cell::I32Array(left), Cell::I32Array(right)) => assert_eq!(left, right),
            (Cell::I64Array(left), Cell::I64Array(right)) => assert_eq!(left, right),
            (Cell::F32Array(left), Cell::F32Array(right)) => assert_eq!(left, right),
            (Cell::F64Array(left), Cell::F64Array(right)) => assert_eq!(left, right),
            (Cell::StringArray(left), Cell::StringArray(right)) => assert_eq!(left, right),
            (left, right) => panic!("cell clone variant mismatch: left={left:?}, right={right:?}",),
        }
    }

    // ==========================================================================
    // Tests for Cell
    // ==========================================================================
    #[test]
    fn test_cell_clone() {
        let cell = Cell::String("hello".to_string());
        assert_cell_clone(cell);
    }

    #[test]
    fn test_cell_clone_primitives() {
        let cases = vec![
            Cell::Bool(true),
            Cell::I8(-8),
            Cell::I16(-16),
            Cell::F32(123.456f32),
            Cell::I32(32),
            Cell::F64(654.321f64),
            Cell::I64(64),
            Cell::String("supabase".to_string()),
        ];

        for cell in cases {
            assert_cell_clone(cell);
        }
    }

    #[test]
    fn test_cell_clone_array_variants() {
        let cases = vec![
            Cell::BoolArray(vec![Some(true), None, Some(false)]),
            Cell::I16Array(vec![Some(-1), None, Some(2)]),
            Cell::I32Array(vec![Some(-10), None, Some(20)]),
            Cell::I64Array(vec![Some(-100), None, Some(200)]),
            Cell::F32Array(vec![Some(1.5), None, Some(2.5)]),
            Cell::F64Array(vec![Some(10.5), None, Some(20.5)]),
            Cell::StringArray(vec![Some("a".to_string()), None, Some("b".to_string())]),
        ];

        for cell in cases {
            let cell_clone = cell.clone();
            assert_cell_clone(cell);
            assert!(cell_clone.is_array());
        }
    }

    #[test]
    fn test_cell_clone_deep_copy_for_owned_types() {
        let mut string_cell = Cell::String("hello".to_string());
        let string_cell_clone = string_cell.clone();
        if let Cell::String(value) = &mut string_cell {
            value.push_str(" world");
        }
        match string_cell_clone {
            Cell::String(value) => assert_eq!(value, "hello"),
            other => panic!("expected Cell::String clone, got {other:?}"),
        }
        match string_cell {
            Cell::String(value) => assert_eq!(value, "hello world"),
            other => panic!("expected mutated Cell::String, got {other:?}"),
        }

        let mut string_array_cell =
            Cell::StringArray(vec![Some("foo".to_string()), None, Some("bar".to_string())]);
        let string_array_cell_clone = string_array_cell.clone();
        if let Cell::StringArray(values) = &mut string_array_cell {
            values[0] = Some("baz".to_string());
        }
        match string_array_cell_clone {
            Cell::StringArray(values) => {
                assert_eq!(
                    values,
                    vec![Some("foo".to_string()), None, Some("bar".to_string())]
                )
            }
            other => panic!("expected Cell::StringArray clone, got {other:?}"),
        }
        match string_array_cell {
            Cell::StringArray(values) => {
                assert_eq!(
                    values,
                    vec![Some("baz".to_string()), None, Some("bar".to_string())]
                )
            }
            other => panic!("expected mutated Cell::StringArray, got {other:?}"),
        }
    }

    #[test]
    fn test_cell_display_primitives_and_string() {
        assert_eq!(format!("{}", Cell::Bool(true)), "true");
        assert_eq!(format!("{}", Cell::I8(-8)), "-8");
        assert_eq!(format!("{}", Cell::I16(16)), "16");
        assert_eq!(format!("{}", Cell::I32(32)), "32");
        assert_eq!(format!("{}", Cell::I64(64)), "64");
        assert_eq!(format!("{}", Cell::F32(3.5)), "3.5");
        assert_eq!(format!("{}", Cell::F64(7.25)), "7.25");
        assert_eq!(format!("{}", Cell::String("hello".to_string())), "'hello'");
    }

    #[test]
    fn test_cell_display_arrays_with_nulls() {
        assert_eq!(
            format!("{}", Cell::BoolArray(vec![Some(true), None, Some(false)])),
            "[true,null,false]"
        );
        assert_eq!(
            format!("{}", Cell::I32Array(vec![Some(1), None, Some(3)])),
            "[1,null,3]"
        );
        assert_eq!(
            format!(
                "{}",
                Cell::StringArray(vec![Some("foo".to_string()), None, Some("bar".to_string())])
            ),
            "[foo,null,bar]"
        );
    }

    #[test]
    fn test_cell_display_empty_arrays() {
        assert_eq!(format!("{}", Cell::BoolArray(vec![])), "[]");
        assert_eq!(format!("{}", Cell::I16Array(vec![])), "[]");
        assert_eq!(format!("{}", Cell::I32Array(vec![])), "[]");
        assert_eq!(format!("{}", Cell::I64Array(vec![])), "[]");
        assert_eq!(format!("{}", Cell::F32Array(vec![])), "[]");
        assert_eq!(format!("{}", Cell::F64Array(vec![])), "[]");
        assert_eq!(format!("{}", Cell::StringArray(vec![])), "[]");
    }

    #[cfg(all(feature = "pg_test", pgrx_embed))]
    #[test]
    fn test_cell_into_datum_scalars_round_trip() {
        let bool_datum = Cell::Bool(true).into_datum().expect("bool should convert");
        let bool_value =
            unsafe { bool::from_datum(bool_datum, false) }.expect("bool should decode");
        assert!(bool_value);

        let i32_datum = Cell::I32(42).into_datum().expect("i32 should convert");
        let i32_value = unsafe { i32::from_datum(i32_datum, false) }.expect("i32 should decode");
        assert_eq!(i32_value, 42);

        let f64_datum = Cell::F64(12.5).into_datum().expect("f64 should convert");
        let f64_value = unsafe { f64::from_datum(f64_datum, false) }.expect("f64 should decode");
        assert_eq!(f64_value, 12.5);

        let string_datum = Cell::String("hello".to_string())
            .into_datum()
            .expect("string should convert");
        let string_value =
            unsafe { String::from_datum(string_datum, false) }.expect("string should decode");
        assert_eq!(string_value, "hello");
    }

    #[cfg(all(feature = "pg_test", pgrx_embed))]
    #[test]
    fn test_cell_into_datum_arrays_round_trip() {
        let bool_array_datum = Cell::BoolArray(vec![Some(true), None, Some(false)])
            .into_datum()
            .expect("bool array should convert");
        let bool_array_value = unsafe { Vec::<Option<bool>>::from_datum(bool_array_datum, false) }
            .expect("bool array should decode");
        assert_eq!(bool_array_value, vec![Some(true), None, Some(false)]);

        let i64_array_datum = Cell::I64Array(vec![Some(1), None, Some(3)])
            .into_datum()
            .expect("i64 array should convert");
        let i64_array_value = unsafe { Vec::<Option<i64>>::from_datum(i64_array_datum, false) }
            .expect("i64 array should decode");
        assert_eq!(i64_array_value, vec![Some(1), None, Some(3)]);

        let string_array_datum =
            Cell::StringArray(vec![Some("foo".to_string()), None, Some("bar".to_string())])
                .into_datum()
                .expect("string array should convert");
        let string_array_value =
            unsafe { Vec::<Option<String>>::from_datum(string_array_datum, false) }
                .expect("string array should decode");
        assert_eq!(
            string_array_value,
            vec![Some("foo".to_string()), None, Some("bar".to_string())]
        );
    }

    #[test]
    fn test_cell_into_datum_type_oid_is_invalid() {
        assert_eq!(Cell::type_oid(), Oid::INVALID);
    }
}