// Copyright (c) 2023-2026 ParadeDB, Inc.
//
// This file is part of ParadeDB - Postgres for Search and Analytics
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.

use crate::api::operator::row_expr_from_indexed_expr;
use crate::api::tokenizers::definitions::pdb::AliasDatumWithType;
use crate::api::tokenizers::{
    type_can_be_tokenized, type_is_alias, type_is_tokenizer, AliasTypmod, UncheckedTypmod,
};
use crate::api::{FieldName, HashMap};
use crate::index::writer::index::IndexError;
use crate::nodecast;
use crate::postgres::build::is_bm25_index;
use crate::postgres::composite::{
    get_composite_fields_for_index, is_composite_type, CompositeSlotValues,
};
use crate::postgres::customscan::orderby::text_lower_funcoid;
use crate::postgres::deparse::deparse_expr;
use crate::postgres::rel::PgSearchRelation;
use crate::postgres::types::TantivyValue;
use crate::postgres::var::find_vars;
use crate::schema::{CategorizedFieldData, SearchField, SearchFieldType};
use anyhow::{anyhow, Result};
use chrono::{NaiveDate, NaiveTime};
use pgrx::itemptr::{item_pointer_get_both, item_pointer_set_all};
use pgrx::*;
use rustc_hash::FxHashMap;
use std::collections::{BTreeMap, HashSet};

use std::ptr::addr_of_mut;
use std::str::FromStr;
use tantivy::schema::OwnedValue;
use tokenizers::SearchNormalizer;

extern "C-unwind" {
    // SAFETY: `IsTransactionState()` doesn't raise an ERROR.  As such, we can avoid the pgrx
    // sigsetjmp overhead by linking to the function directly.
    pub fn IsTransactionState() -> bool;
}

/// Implements Drop that skips cleanup during panic unwinding.
///
/// Because panics are used to propagate PostgreSQL errors via pgrx, it is almost never
/// safe to interact with PostgreSQL APIs during Drop - doing so can cause a double-panic
/// which results in SIGABRT. This macro ensures cleanup is skipped during unwinding.
///
/// PostgreSQL's transaction abort mechanism will clean up resources (buffers, relations, etc.)
/// when the transaction is aborted due to the error.
///
/// # Example
/// ```ignore
/// impl_safe_drop!(MyStruct, |self| {
///     unsafe {
///         if crate::postgres::utils::IsTransactionState() {
///             pg_sys::some_cleanup_function(self.handle);
///         }
///     }
/// });
/// ```
#[macro_export]
macro_rules! impl_safe_drop {
    ($ty:ty, |$self:ident| $body:block) => {
        impl Drop for $ty {
            fn drop(&mut $self) {
                if std::thread::panicking() {
                    return;
                }
                $body
            }
        }
    };
}

/// RAII guard for PostgreSQL standalone expression context
/// Automatically frees the context when dropped
///
/// let context_guard = ExprContextGuard::new();
/// // Use context_guard.as_ptr() to get the raw pointer
/// // Context is automatically freed when context_guard goes out of scope
#[derive(Debug)]
pub struct ExprContextGuard(*mut pg_sys::ExprContext);

// SAFETY: PostgreSQL doesn't execute within threads, despite Tantivy expecting it.
// The ExprContextGuard is used in Tantivy queries that require Send+Sync, but in practice
// these are never actually shared across threads in our PostgreSQL context.
unsafe impl Send for ExprContextGuard {}
unsafe impl Sync for ExprContextGuard {}

impl ExprContextGuard {
    /// Creates a new standalone expression context
    pub fn new() -> Self {
        unsafe { Self(pg_sys::CreateStandaloneExprContext()) }
    }

    /// Returns the raw pointer to the expression context
    pub fn as_ptr(&self) -> *mut pg_sys::ExprContext {
        self.0
    }
}

impl_safe_drop!(ExprContextGuard, |self| {
    unsafe {
        let is_commit = pg_sys::IsTransactionState();
        pg_sys::FreeExprContext(self.0, is_commit);
    }
});

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

/// Recursively sort all object keys in JSON for deterministic output
pub fn sort_json_keys(value: &mut serde_json::Value) {
    match value {
        serde_json::Value::Object(map) => {
            // Collect entries, sort by key, and rebuild
            let sorted: BTreeMap<String, serde_json::Value> = map
                .iter()
                .map(|(k, v)| {
                    let mut v = v.clone();
                    sort_json_keys(&mut v);
                    (k.clone(), v)
                })
                .collect();
            *map = sorted.into_iter().collect();
        }
        serde_json::Value::Array(arr) => {
            for item in arr.iter_mut() {
                sort_json_keys(item);
            }
        }
        _ => {}
    }
}

/// TransactionIdPrecedesOrEquals --- is id1 logically <= id2?
///
/// Ported to rust from the Postgres sources to avoid the pgrx FFI overhead
#[allow(non_snake_case)]
#[inline]
pub fn TransactionIdPrecedesOrEquals(
    id1: pg_sys::TransactionId,
    id2: pg_sys::TransactionId,
) -> bool {
    //
    // #define TransactionIdIsNormal(xid)		((xid) >= FirstNormalTransactionId)
    //
    // /*
    //  * TransactionIdPrecedesOrEquals --- is id1 logically <= id2?
    //  */
    // bool
    // TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
    // {
    //     /*
    //      * If either ID is a permanent XID then we can just do unsigned
    //      * comparison.  If both are normal, do a modulo-2^32 comparison.
    //      */
    //     int32		diff;
    //
    //     if (!TransactionIdIsNormal(id1) || !TransactionIdIsNormal(id2))
    //         return (id1 <= id2);
    //
    //     diff = (int32) (id1 - id2);
    //     return (diff <= 0);
    // }

    #[inline]
    fn is_transaction_id_normal(xid: pg_sys::TransactionId) -> bool {
        xid >= pg_sys::FirstNormalTransactionId
    }

    if !is_transaction_id_normal(id1) || !is_transaction_id_normal(id2) {
        return id1 <= id2;
    }

    // Compare as i32 to handle wraparound
    unsafe {
        // SAFETY: `pg_sysTransactionId` is a #[repr(transparent)] wrapper around a `u32`
        let id1: i32 = std::mem::transmute(id1);
        let id2: i32 = std::mem::transmute(id2);
        let diff = id1.wrapping_sub(id2);
        diff <= 0
    }
}

#[cfg(any(test, feature = "pg_test"))]
#[pgrx::pg_schema]
mod tests {
    use crate::postgres::utils::TransactionIdPrecedesOrEquals;
    use pgrx::{pg_sys, pg_test};

    #[pg_test]
    fn xid_precedes_or_equals_test() {
        use proptest::prelude::*;

        proptest!(|(xid1 in 0..u32::MAX, xid2 in 0..u32::MAX)| {
            let us = TransactionIdPrecedesOrEquals(pg_sys::TransactionId::from(xid1), pg_sys::TransactionId::from(xid2));
            let pg = unsafe { pg_sys::TransactionIdPrecedesOrEquals(pg_sys::TransactionId::from(xid1), pg_sys::TransactionId::from(xid2)) };
            prop_assert_eq!(us, pg);
        });
    }
}

/// Finds and returns the `USING bm25` index on the specified relation with the
/// highest OID, or [`None`] if there aren't any.
pub fn locate_bm25_index(heaprelid: pg_sys::Oid) -> Option<PgSearchRelation> {
    locate_bm25_index_from_heaprel(&PgSearchRelation::open(heaprelid))
}

/// Finds and returns the `USING bm25` index on the specified relation with the
/// highest OID, or [`None`] if there aren't any.
pub fn locate_bm25_index_from_heaprel(heaprel: &PgSearchRelation) -> Option<PgSearchRelation> {
    unsafe {
        let indices = heaprel.indices(pg_sys::AccessShareLock as _);

        // Find all bm25 indexes and keep the one with highest OID
        indices
            .into_iter()
            .filter(|index| pg_sys::get_index_isvalid(index.oid()) && is_bm25_index(index))
            .max_by_key(|index| index.oid().to_u32())
    }
}

/// Rather than using pgrx' version of this function, we use our own, which doesn't leave 2
/// empty bytes in the middle of the 64bit representation.  A ctid being only 48bits means
/// if we leave the upper 16 bits (2 bytes) empty, tantivy will have a better chance of
/// bitpacking or compressing these values.
#[allow(dead_code)]
#[inline(always)]
pub fn item_pointer_to_u64(ctid: pg_sys::ItemPointerData) -> u64 {
    let (blockno, offno) = item_pointer_get_both(ctid);
    let blockno = blockno as u64;
    let offno = offno as u64;

    // shift the BlockNumber left 16 bits -- the length of the OffsetNumber we OR onto the end
    // pgrx's version shifts left 32, which is wasteful
    (blockno << 16) | offno
}

/// Rather than using pgrx' version of this function, we use our own, which doesn't leave 2
/// empty bytes in the middle of the 64bit representation.  A ctid being only 48bits means
/// if we leave the upper 16 bits (2 bytes) empty, tantivy will have a better chance of
/// bitpacking or compressing these values.
#[inline(always)]
pub fn u64_to_item_pointer(value: u64, tid: &mut pg_sys::ItemPointerData) {
    // shift right 16 bits to pop off the OffsetNumber, leaving only the BlockNumber
    // pgrx's version must shift right 32 bits to be in parity with `item_pointer_to_u64()`
    let blockno = (value >> 16) as pg_sys::BlockNumber;
    let offno = value as pg_sys::OffsetNumber;
    item_pointer_set_all(tid, blockno, offno);
}

/// Returns `true` if the block referenced by `ctid` (u64-packed form) exists
/// in `rel`. A `false` result means VACUUM has truncated the page.
#[inline(always)]
pub unsafe fn ctid_satisfies_nblocks(ctid: u64, rel: pg_sys::Relation) -> bool {
    let blockno = (ctid >> 16) as pg_sys::BlockNumber;
    blockno < pg_sys::RelationGetNumberOfBlocksInFork(rel, pg_sys::ForkNumber::MAIN_FORKNUM)
}

#[derive(Copy, Clone, Debug)]
pub enum FieldSource {
    /// Direct column from heap tuple
    Heap { attno: usize },

    /// Expression index (scalar result)
    Expression { att_idx: usize },

    /// Field extracted from a composite type
    CompositeField {
        /// Index attribute number (slot in values[] array for aminsert/build_callback)
        index_attno: usize,

        /// Expression index (position in expr_results[] array for MVCC build)
        expression_idx: usize,

        /// Field position within the composite (0-indexed)
        field_idx: usize,

        /// OID of the named composite type
        composite_type_oid: pg_sys::Oid,
    },
}

/// Collect composite slot info from categorized fields for upfront unpacking.
///
/// Returns a lazy iterator of (slot_index, datum, is_null, type_oid) for each unique
/// composite slot in the categorized fields.
///
/// # Safety
/// - `values` and `isnull` pointers are dereferenced lazily during iteration,
///   not at call time. Caller must ensure these pointers remain valid until
///   the returned iterator is fully consumed.
/// - The iterator should be consumed immediately (e.g., via `from_composites()`).
///   Storing the iterator and consuming it later risks UB if the underlying
///   slot arrays have been invalidated.
pub unsafe fn collect_composites_for_unpacking<'a>(
    categorized_fields: impl Iterator<Item = &'a CategorizedFieldData> + 'a,
    values: *mut pg_sys::Datum,
    isnull: *mut bool,
) -> impl Iterator<Item = (usize, pg_sys::Datum, bool, pg_sys::Oid)> + 'a {
    let mut seen = rustc_hash::FxHashSet::default();

    categorized_fields.filter_map(move |cat| {
        if let FieldSource::CompositeField {
            index_attno,
            composite_type_oid,
            ..
        } = cat.source
        {
            if seen.insert(index_attno) {
                let datum = *values.add(index_attno);
                let is_null = *isnull.add(index_attno);
                return Some((index_attno, datum, is_null, composite_type_oid));
            }
        }
        None
    })
}

/// Helper to extract field value from values[] array, handling composite fields.
///
/// **Works for**: aminsert (INSERT) and build_callback (CREATE INDEX) paths.
/// **Does NOT work for**: MVCC build (which uses expr_results[] instead).
///
/// # Arguments
/// * `source` - The field source (Heap, Expression, or CompositeField)
/// * `index_attno` - Index attribute position (slot in values[] array)
/// * `values` - Array of datums from PostgreSQL
/// * `isnull` - Array of null flags from PostgreSQL
/// * `unpacked_composites` - Pre-unpacked composite values
///
/// # Returns
/// Tuple of (datum, is_null) for the field
///
/// # Safety
/// Caller must ensure values and isnull pointers are valid.
pub unsafe fn get_field_value(
    source: &FieldSource,
    index_attno: usize,
    values: *mut pg_sys::Datum,
    isnull: *mut bool,
    unpacked_composites: &CompositeSlotValues,
) -> (pg_sys::Datum, bool) {
    match source {
        FieldSource::Heap { .. } | FieldSource::Expression { .. } => {
            (*values.add(index_attno), *isnull.add(index_attno))
        }
        FieldSource::CompositeField {
            index_attno: comp_index_attno,
            field_idx,
            ..
        } => unpacked_composites.get(*comp_index_attno, *field_idx),
    }
}

/// Represents the metadata extracted from an index attribute
#[derive(Debug)]
pub struct ExtractedFieldAttribute {
    /// its ordinal position in the index attribute list
    pub attno: usize,

    /// its source in the heap: either a heap attno, or an expression index
    pub source: FieldSource,

    /// its original Postgres type OID
    pub pg_type: PgOid,

    /// the type we'll use for indexing in tantivy
    pub tantivy_type: SearchFieldType,

    pub inner_typoid: pg_sys::Oid,

    pub normalizer: Option<SearchNormalizer>,
}

/// Recursively strips tokenizer casts (e.g. `pdb.literal`, `pdb.alias`) from an expression.
pub unsafe fn strip_tokenizer_cast(node: *mut pg_sys::Node) -> *mut pg_sys::Node {
    if node.is_null() {
        return node;
    }

    if let Some(func) = nodecast!(FuncExpr, T_FuncExpr, node) {
        if type_is_tokenizer((*func).funcresulttype) {
            let args = PgList::<pg_sys::Node>::from_pg((*func).args);
            if let Some(arg) = args.get_ptr(0) {
                return strip_tokenizer_cast(arg);
            }
        }
    } else if let Some(relabel) = nodecast!(RelabelType, T_RelabelType, node) {
        return strip_tokenizer_cast((*relabel).arg.cast());
    } else if let Some(coerce) = nodecast!(CoerceToDomain, T_CoerceToDomain, node) {
        return strip_tokenizer_cast((*coerce).arg.cast());
    } else if let Some(coerce) = nodecast!(CoerceViaIO, T_CoerceViaIO, node) {
        return strip_tokenizer_cast((*coerce).arg.cast());
    }

    node
}

/// Recursively strips `UNNEST` function calls and basic type coercion wrappers
/// (`RelabelType`, `CoerceToDomain`, `CoerceViaIO`) from an expression.
///
/// Returns a tuple containing the stripped node and a boolean indicating if `UNNEST` was found.
pub unsafe fn strip_unnest_and_relabel(mut node: *mut pg_sys::Node) -> (*mut pg_sys::Node, bool) {
    let mut found_unnest = false;
    loop {
        if node.is_null() {
            return (node, found_unnest);
        }

        if let Some(relabel) = nodecast!(RelabelType, T_RelabelType, node) {
            node = (*relabel).arg.cast();
            continue;
        }
        if let Some(coerce) = nodecast!(CoerceToDomain, T_CoerceToDomain, node) {
            node = (*coerce).arg.cast();
            continue;
        }
        if let Some(coerce) = nodecast!(CoerceViaIO, T_CoerceViaIO, node) {
            node = (*coerce).arg.cast();
            continue;
        }
        if let Some(phv) = nodecast!(PlaceHolderVar, T_PlaceHolderVar, node) {
            node = (*phv).phexpr.cast();
            continue;
        }
        if let Some(func) = nodecast!(FuncExpr, T_FuncExpr, node) {
            if is_unnest_func((*func).funcid) {
                found_unnest = true;
                let args = PgList::<pg_sys::Node>::from_pg((*func).args);
                if let Some(arg) = args.get_ptr(0) {
                    node = arg;
                    continue;
                }
            }
        }
        break;
    }
    (node, found_unnest)
}

/// Identifies if the function identified by `funcid` is `pg_catalog.unnest(anyarray)`.
pub fn is_unnest_func(funcid: pg_sys::Oid) -> bool {
    use std::sync::Once;
    static mut UNNEST_OID: pg_sys::Oid = pg_sys::InvalidOid;
    static UNNEST_OID_ONCE: Once = Once::new();

    unsafe {
        UNNEST_OID_ONCE.call_once(|| {
            if let Some(oid) = direct_function_call::<pg_sys::Oid>(
                pg_sys::regprocedurein,
                &[c"pg_catalog.unnest(anyarray)".into_datum()],
            ) {
                UNNEST_OID = oid;
            }
        });

        if funcid == UNNEST_OID && UNNEST_OID != pg_sys::InvalidOid {
            return true;
        }

        false
    }
}

/// Create a text Const node from a string
///
/// # Safety
/// This function must be called within a PostgreSQL memory context that will persist
/// for the lifetime of the plan tree. The returned Const node will be allocated in the
/// current memory context and should not be freed manually.
pub unsafe fn make_text_const(text: &str) -> *mut pg_sys::Const {
    let text_datum = text
        .into_datum()
        .expect("failed to convert string to datum");

    pg_sys::makeConst(
        pg_sys::TEXTOID,
        -1,
        pg_sys::DEFAULT_COLLATION_OID,
        -1,
        text_datum,
        false, // constisnull
        false, // constbyval (text is not passed by value)
    )
}

/// Extracts the field attributes from the index relation.
/// It returns a vector of tuples containing the field name and its type OID.
pub unsafe fn extract_field_attributes(
    indexrel: pg_sys::Relation,
) -> HashMap<FieldName, ExtractedFieldAttribute> {
    #[inline]
    unsafe fn is_text_lower(expression: *mut pg_sys::Node) -> bool {
        if let Some(coerce) = nodecast!(CoerceViaIO, T_CoerceViaIO, expression) {
            if let Some(func_expr) = nodecast!(FuncExpr, T_FuncExpr, (*coerce).arg) {
                if (*func_expr).funcid == text_lower_funcoid() {
                    return true;
                }
            }
        }

        false
    }

    let heap_relation = PgSearchRelation::from_pg(indexrel).heap_relation().unwrap();
    let heap_tupdesc = heap_relation.tuple_desc();
    let pg_search_indexrel = PgSearchRelation::from_pg(indexrel);
    let index_info = pg_search_indexrel.index_info();
    let expressions = pg_search_indexrel.index_expressions();
    let mut expressions_iter = expressions.iter_ptr().enumerate();
    let mut field_attributes: FxHashMap<FieldName, ExtractedFieldAttribute> = Default::default();
    for attno in 0..(*index_info).ii_NumIndexAttrs {
        let heap_attno = (*index_info).ii_IndexAttrNumbers[attno as usize];
        let (attname, attribute_type_oid, att_typmod, source, expression, inner_typoid, normalizer) =
            if heap_attno == 0 {
                // Is an expression.
                let Some((expression_idx, expression)) = expressions_iter.next() else {
                    panic!("Expected expression for index attribute {attno}.");
                };
                let source = FieldSource::Expression {
                    att_idx: expression_idx,
                };
                let node = expression.cast();

                let typoid = pg_sys::exprType(node);
                let mut attname = None;
                let mut typmod = -1;
                let mut inner_typoid = typoid;
                let mut normalizer = None;

                // Check if this expression is a composite type
                if is_composite_type(typoid) {
                    // Get composite fields (validates for nested composites, etc.)
                    let composite_fields =
                        get_composite_fields_for_index(typoid).unwrap_or_else(|e| panic!("{e}"));
                    let row_args =
                        row_expr_from_indexed_expr(expression.cast()).map(|row_expr| unsafe {
                            PgList::<pg_sys::Node>::from_pg((*row_expr).args)
                        });

                    // Add each field from the composite to the index
                    for comp_field in composite_fields {
                        if comp_field.is_dropped {
                            continue;
                        }

                        // Check for duplicate field name (reuse existing logic)
                        if field_attributes.contains_key(&FieldName::from(&comp_field.field_name)) {
                            panic!(
                                "indexed attribute {} defined more than once",
                                comp_field.field_name
                            );
                        }

                        let pg_type = PgOid::from_untagged(comp_field.type_oid);
                        let tantivy_type = SearchFieldType::try_from((
                            pg_type,
                            comp_field.typmod,
                            comp_field.type_oid,
                        ))
                        .unwrap_or_else(|e| panic!("{e}"));

                        field_attributes.insert(
                            comp_field.field_name.clone().into(),
                            ExtractedFieldAttribute {
                                attno: attno as usize,
                                source: FieldSource::CompositeField {
                                    index_attno: attno as usize,
                                    expression_idx,
                                    field_idx: comp_field.field_index,
                                    composite_type_oid: typoid,
                                },
                                pg_type,
                                tantivy_type,
                                inner_typoid: row_args
                                    .as_ref()
                                    .and_then(|args| args.get_ptr(comp_field.field_index))
                                    .map(|arg| {
                                        let inner_node = strip_tokenizer_cast(arg.cast());
                                        pg_sys::exprType(inner_node)
                                    })
                                    .unwrap_or(comp_field.type_oid),
                                normalizer: None,
                            },
                        );
                    }

                    // Skip normal expression handling for composite types
                    continue;
                }

                if type_is_tokenizer(typoid) {
                    typmod = pg_sys::exprTypmod(node);

                    let parsed_typmod =
                        UncheckedTypmod::try_from(typmod).unwrap_or_else(|e| panic!("{e}"));
                    let vars = find_vars(node);

                    normalizer = parsed_typmod.normalizer();
                    attname = parsed_typmod.alias();

                    // Attempt to determine inner_typoid by peeling the tokenizer cast/function.
                    // This handles cases like `(a || b)::pdb.literal('alias=...')` where vars.len() > 1.
                    if inner_typoid == typoid {
                        let inner_node = strip_tokenizer_cast(expression.cast());
                        inner_typoid = pg_sys::exprType(inner_node);
                    }

                    if attname.is_none() && vars.len() == 1 {
                        let var = vars[0];
                        inner_typoid = pg_sys::exprType(var as *mut pg_sys::Node);
                        if is_text_lower(expression.cast()) {
                            normalizer = Some(SearchNormalizer::Lowercase);
                        } else if let Some(relabel) =
                            nodecast!(RelabelType, T_RelabelType, expression)
                        {
                            if is_a((*relabel).arg.cast(), pg_sys::NodeTag::T_CoerceViaIO) {
                                inner_typoid = pg_sys::exprType((*relabel).arg.cast());
                            }

                            if is_text_lower((*relabel).arg.cast()) {
                                normalizer = Some(SearchNormalizer::Lowercase);
                            }
                        }
                        if attname.is_none() {
                            let heap_attname = heap_relation
                                .tuple_desc()
                                .get((*var).varattno as usize - 1)
                                .unwrap()
                                .name()
                                .to_string();
                            attname = Some(heap_attname);
                        }
                    }

                    if type_is_alias(typoid) {
                        if type_can_be_tokenized(inner_typoid) {
                            panic!("To alias a text or JSON type, cast it to a tokenizer with an `alias` argument instead of `pdb.alias`");
                        } else {
                            // if we have a non text field text to `pdb.alias`, unwrap it to get the inner typoid
                            if let Some(relabel) = nodecast!(RelabelType, T_RelabelType, expression)
                            {
                                if let Some(func) = nodecast!(FuncExpr, T_FuncExpr, (*relabel).arg)
                                {
                                    let args = PgList::<pg_sys::Node>::from_pg((*func).args);
                                    if args.len() == 1 {
                                        if let Some(arg) = args.get_ptr(0) {
                                            if let Some(inner_func) =
                                                nodecast!(FuncExpr, T_FuncExpr, arg)
                                            {
                                                inner_typoid = (*inner_func).funcresulttype;
                                            } else if let Some(op_expr) =
                                                nodecast!(OpExpr, T_OpExpr, arg)
                                            {
                                                inner_typoid = (*op_expr).opresulttype;
                                            }
                                        }
                                    }
                                }
                            }

                            // use the alias name as the field name instead of the heap attribute name
                            let alias_typmod =
                                AliasTypmod::try_from(typmod).unwrap_or_else(|e| panic!("{e}"));
                            attname = alias_typmod.alias();
                        }
                    }
                }

                let Some(attname) = attname else {
                    let expr_str = deparse_expr(None, &heap_relation, expression.cast());
                    panic!(
                        "indexed expression requires a tokenizer cast with an alias: {expr_str}"
                    );
                };

                (
                    attname,
                    typoid,
                    typmod,
                    source,
                    Some(expression),
                    inner_typoid,
                    normalizer,
                )
            } else {
                // Is a field -- get the field name from the heap relation.
                let attno = (heap_attno - 1) as usize;
                let att = heap_tupdesc.get(attno).expect("attribute should exist");
                (
                    att.name().to_owned(),
                    att.type_oid().value(),
                    att.type_mod(),
                    FieldSource::Heap { attno },
                    None,
                    att.type_oid().value(),
                    None,
                )
            };

        if field_attributes.contains_key(&FieldName::from(&attname)) {
            panic!("indexed attribute {attname} defined more than once");
        }

        let pg_type = PgOid::from_untagged(attribute_type_oid);
        let tantivy_type = SearchFieldType::try_from((pg_type, att_typmod, inner_typoid))
            .unwrap_or_else(|e| panic!("{e}"));

        // non-plain-attribute expressions that aren't cast to a tokenizer type are forced to use our `pdb.literal` tokenizer
        let missing_tokenizer_cast = expression.is_some()
            && att_typmod == -1
            && matches!(tantivy_type, SearchFieldType::Text(..));
        if missing_tokenizer_cast {
            let expr_str =
                unsafe { deparse_expr(None, &heap_relation, expression.unwrap().cast()) };
            panic!("indexed expression must be cast to a tokenizer: {expr_str}");
        }

        field_attributes.insert(
            attname.into(),
            ExtractedFieldAttribute {
                attno: attno as usize,
                source,
                pg_type,
                tantivy_type,
                inner_typoid,
                normalizer,
            },
        );
    }
    field_attributes
}

pub unsafe fn row_to_search_document<'a>(
    categorized_fields: impl Iterator<
        Item = (
            pg_sys::Datum,
            bool,
            &'a SearchField,
            &'a CategorizedFieldData,
        ),
    >,
    document: &mut tantivy::TantivyDocument,
) -> Result<(), IndexError> {
    for (
        datum,
        isnull,
        search_field,
        CategorizedFieldData {
            pg_type,
            base_oid,
            is_key_field,
            is_array,
            is_json,
            ..
        },
    ) in categorized_fields
    {
        if isnull && *is_key_field {
            return Err(IndexError::KeyIdNull(search_field.field_name().to_string()));
        }

        if isnull {
            continue;
        }

        // For pdb.alias types, unwrap the datum first before any processing
        // The AliasDatumWithType structure wraps the actual datum for all alias types
        let actual_datum = if type_is_alias(pg_type.value()) {
            unsafe { AliasDatumWithType::extract_datum(datum) }
        } else {
            datum
        };

        if *is_array {
            // Check for NUMERIC array field types that need special handling
            match search_field.field_type() {
                SearchFieldType::Numeric64(_, scale) => {
                    for value in TantivyValue::try_from_numeric_array_i64(actual_datum, scale)
                        .unwrap_or_else(|e| {
                            panic!("could not parse field `{}`: {e}", search_field.field_name())
                        })
                    {
                        document.add_field_value(search_field.field(), &OwnedValue::from(value));
                    }
                }
                SearchFieldType::NumericBytes(..) => {
                    for value in TantivyValue::try_from_numeric_array_bytes(actual_datum)
                        .unwrap_or_else(|e| {
                            panic!("could not parse field `{}`: {e}", search_field.field_name())
                        })
                    {
                        document.add_field_value(search_field.field(), &OwnedValue::from(value));
                    }
                }
                _ => {
                    for value in TantivyValue::try_from_datum_array(actual_datum, *base_oid)
                        .unwrap_or_else(|e| {
                            panic!("could not parse field `{}`: {e}", search_field.field_name())
                        })
                    {
                        document.add_field_value(search_field.field(), &OwnedValue::from(value));
                    }
                }
            }
        } else if *is_json {
            for value in
                TantivyValue::try_from_datum_json(actual_datum, *base_oid).unwrap_or_else(|e| {
                    panic!("could not parse field `{}`: {e}", search_field.field_name())
                })
            {
                document.add_field_value(search_field.field(), &OwnedValue::from(value));
            }
        } else {
            // Check for NUMERIC field types that need special handling
            let tv = match search_field.field_type() {
                SearchFieldType::Numeric64(_, scale) => {
                    TantivyValue::try_from_numeric_i64(actual_datum, scale)
                }
                SearchFieldType::NumericBytes(..) => {
                    TantivyValue::try_from_numeric_bytes(actual_datum)
                }
                _ => TantivyValue::try_from_datum(actual_datum, *base_oid),
            }
            .unwrap_or_else(|e| {
                panic!("could not parse field `{}`: {e}", search_field.field_name())
            });
            document.add_field_value(search_field.field(), &OwnedValue::from(tv));
        }
    }
    Ok(())
}

/// Utility function for easy `f64` to `u32` conversion
fn f64_to_u32(n: f64) -> Result<u32> {
    let truncated = n.trunc();
    if truncated.is_nan()
        || truncated.is_infinite()
        || truncated < 0.0
        || truncated > u32::MAX.into()
    {
        return Err(anyhow!("overflow in f64 to u32"));
    }

    Ok(truncated as u32)
}

/// Seconds are represented by `f64` in pgrx, with a maximum of microsecond precision
fn convert_pgrx_seconds_to_chrono(orig: f64) -> Result<(u32, u32, u32)> {
    let seconds = f64_to_u32(orig)?;
    let microseconds = f64_to_u32((orig * 1_000_000.0) % 1_000_000.0)?;
    let nanoseconds = f64_to_u32((orig * 1_000_000_000.0) % 1_000_000_000.0)?;
    Ok((seconds, microseconds, nanoseconds))
}

pub fn convert_pg_date_string(typeoid: PgOid, date_string: &str) -> tantivy::DateTime {
    use crate::postgres::datetime::micros_to_tantivy_datetime;

    match typeoid {
        PgOid::BuiltIn(PgBuiltInOids::DATEOID | PgBuiltInOids::DATERANGEOID) => {
            let d = pgrx::datum::Date::from_str(date_string)
                .expect("must be valid postgres date format");
            let micros = NaiveDate::from_ymd_opt(d.year(), d.month().into(), d.day().into())
                .expect("must be able to parse date format")
                .and_hms_opt(0, 0, 0)
                .expect("must be able to set date default time")
                .and_utc()
                .timestamp_micros();
            micros_to_tantivy_datetime(micros)
                .expect("date exceeds Tantivy DateTime nanosecond range")
        }
        PgOid::BuiltIn(PgBuiltInOids::TIMESTAMPOID | PgBuiltInOids::TSRANGEOID) => {
            // Since [`pgrx::Timestamp`]s are tied to the Postgres instance's timezone,
            // to figure out *which* timezone it's actually in, we convert to a
            // [`pgrx::TimestampWithTimeZone`].
            // Once the offset is known, we can create and return a [`chrono::NaiveDateTime`]
            // with the appropriate offset.
            let t = pgrx::datum::Timestamp::from_str(date_string)
                .expect("must be a valid postgres timestamp");
            let twtz: datum::TimestampWithTimeZone = t.into();
            let (seconds, micros, _nanos) = convert_pgrx_seconds_to_chrono(twtz.second())
                .expect("must not overflow converting pgrx seconds");
            let micros =
                NaiveDate::from_ymd_opt(twtz.year(), twtz.month().into(), twtz.day().into())
                    .expect("must be able to convert date timestamp")
                    .and_hms_micro_opt(twtz.hour().into(), twtz.minute().into(), seconds, micros)
                    .expect("must be able to parse timestamp format")
                    .and_utc()
                    .timestamp_micros();
            micros_to_tantivy_datetime(micros)
                .expect("timestamp exceeds Tantivy DateTime nanosecond range")
        }
        PgOid::BuiltIn(PgBuiltInOids::TIMESTAMPTZOID | pg_sys::BuiltinOid::TSTZRANGEOID) => {
            let twtz = pgrx::datum::TimestampWithTimeZone::from_str(date_string)
                .expect("must be a valid postgres timestamp with time zone")
                .to_utc();
            let (seconds, micros, _nanos) = convert_pgrx_seconds_to_chrono(twtz.second())
                .expect("must not overflow converting pgrx seconds");
            let micros =
                NaiveDate::from_ymd_opt(twtz.year(), twtz.month().into(), twtz.day().into())
                    .expect("must be able to convert timestamp with timezone")
                    .and_hms_micro_opt(twtz.hour().into(), twtz.minute().into(), seconds, micros)
                    .expect("must be able to parse timestamp with timezone")
                    .and_utc()
                    .timestamp_micros();
            micros_to_tantivy_datetime(micros)
                .expect("timestamptz exceeds Tantivy DateTime nanosecond range")
        }
        PgOid::BuiltIn(PgBuiltInOids::TIMEOID) => {
            let t =
                pgrx::datum::Time::from_str(date_string).expect("must be a valid postgres time");
            let (hour, minute, second, micros) = t.to_hms_micro();
            let naive_time =
                NaiveTime::from_hms_micro_opt(hour.into(), minute.into(), second.into(), micros)
                    .expect("must be able to parse time");
            let naive_date = NaiveDate::from_ymd_opt(1970, 1, 1).expect("default date");
            let micros = naive_date.and_time(naive_time).and_utc().timestamp_micros();
            micros_to_tantivy_datetime(micros)
                .expect("time exceeds Tantivy DateTime nanosecond range")
        }
        PgOid::BuiltIn(PgBuiltInOids::TIMETZOID) => {
            let twtz = pgrx::datum::TimeWithTimeZone::from_str(date_string)
                .expect("must be a valid postgres time with time zone")
                .to_utc();
            let (hour, minute, second, micros) = twtz.to_hms_micro();
            let naive_time =
                NaiveTime::from_hms_micro_opt(hour.into(), minute.into(), second.into(), micros)
                    .expect("must be able to parse time with time zone");
            let naive_date = NaiveDate::from_ymd_opt(1970, 1, 1).expect("default date");
            let micros = naive_date.and_time(naive_time).and_utc().timestamp_micros();
            micros_to_tantivy_datetime(micros)
                .expect("timetz exceeds Tantivy DateTime nanosecond range")
        }
        _ => panic!("Unsupported typeoid: {typeoid:?}"),
    }
}

/// Extract precision and scale from PostgreSQL NUMERIC typmod.
///
/// PostgreSQL encodes NUMERIC precision and scale in the typmod as:
/// `typmod = ((precision << 16) | (scale & 0x7ff)) + VARHDRSZ`
///
/// The scale is stored in 11 bits (0x7ff = 2047), supporting the range [-1000, 1000].
/// Negative scales are stored in two's complement within those 11 bits.
///
/// # Returns
/// - `(precision, Some(scale))` if typmod specifies precision/scale
/// - `(0, None)` if typmod is -1 (unlimited/unspecified precision)
///
/// # Note
/// - For NUMERIC columns declared without precision (e.g., `NUMERIC` instead of `NUMERIC(10,2)`),
///   PostgreSQL uses typmod = -1, indicating arbitrary precision.
/// - PostgreSQL 15+ supports negative scales (e.g., NUMERIC(5,-3) rounds to nearest 1000).
///
/// # Reference
/// See PostgreSQL's `numeric_typmod_scale()` in src/backend/utils/adt/numeric.c:
/// ```c
/// return (((typmod - VARHDRSZ) & 0x7ff) ^ 1024) - 1024;
/// ```
pub fn extract_numeric_precision_scale(typmod: i32) -> (u16, Option<i16>) {
    // PostgreSQL uses typmod = -1 for unlimited precision NUMERIC
    if typmod < 0 {
        return (0, None);
    }

    // Extract precision and scale from typmod
    // See PostgreSQL's make_numeric_typmod() and numeric_typmod_scale()
    let typmod_val = (typmod - pg_sys::VARHDRSZ as i32) as u32;

    // Precision is in upper 16 bits
    let precision = ((typmod_val >> 16) & 0xFFFF) as u16;

    // Scale is stored in lower 11 bits (0x7ff), using two's complement for negatives.
    // The formula (x ^ 1024) - 1024 sign-extends an 11-bit value to a full integer.
    // This works because 1024 = 2^10, the midpoint of the 11-bit range.
    let stored_scale = (typmod_val & 0x7ff) as i16;
    let scale = (stored_scale ^ 1024) - 1024;

    (precision, Some(scale))
}

type IsArray = bool;
/// Returns the base type of the given `oid`, and a boolean indicating if the
/// type is an array.
pub fn resolve_base_type(oid: PgOid) -> Option<(PgOid, IsArray)> {
    fn is_domain_type(oid: pg_sys::Oid) -> bool {
        unsafe { pg_sys::get_typtype(oid) as u8 == pg_sys::TYPTYPE_DOMAIN }
    }

    if matches!(oid, PgOid::Invalid) {
        return None;
    }

    // resolve domain type to its base
    let base_oid = if is_domain_type(oid.value()) {
        let resolved_type = unsafe { pg_sys::getBaseType(oid.value()) };
        if resolved_type == pg_sys::InvalidOid {
            return None;
        }
        resolved_type
    } else {
        oid.value()
    };

    // check if it's an array type
    let array_type = PgOid::from(unsafe { pg_sys::get_element_type(base_oid) });

    match array_type {
        // not an array
        PgOid::Invalid => Some((base_oid.into(), false)),

        // built-in array type or custom array type
        PgOid::BuiltIn(_) | PgOid::Custom(_) => {
            let resolved_array_type = if is_domain_type(array_type.value()) {
                let resolved_type = unsafe { pg_sys::getBaseType(array_type.value()) };
                if resolved_type == pg_sys::InvalidOid {
                    return None;
                }
                resolved_type
            } else {
                array_type.value()
            };

            Some((resolved_array_type.into(), true))
        }
    }
}

pub trait ToPalloc: Sized {
    fn palloc(self) -> *mut Self {
        self.palloc_in(PgMemoryContexts::CurrentMemoryContext)
    }

    fn palloc_in(self, mcxt: PgMemoryContexts) -> *mut Self;
}

impl<T> ToPalloc for T {
    fn palloc_in(mut self, mut mcxt: PgMemoryContexts) -> *mut Self {
        unsafe { mcxt.copy_ptr_into((&mut self as *mut T).cast(), size_of::<T>()) }
    }
}

/// Recursively check if an expression tree contains any of the specified operators
/// Uses PostgreSQL's expression_tree_walker for robust traversal
///
/// NOTE: This logic is duplicated with `extract_quals` in qual_inspect.rs.
/// Both need to traverse expression trees looking for operators, so changes to one
/// should be reflected in the other.
/// TODO: Consider unifying this logic to avoid duplication (see GitHub issue #3455)
pub unsafe fn expr_contains_any_operator(
    node: *mut pg_sys::Node,
    target_opnos: &[pg_sys::Oid],
) -> bool {
    use pgrx::pg_guard;
    use std::ptr::addr_of_mut;

    #[pg_guard]
    unsafe extern "C-unwind" fn walker(
        node: *mut pg_sys::Node,
        data: *mut core::ffi::c_void,
    ) -> bool {
        if node.is_null() {
            return false;
        }

        let context = &*(data as *const Context);
        let node_type = (*node).type_;

        // Check if this node is an OpExpr with one of our target operators
        if node_type == pg_sys::NodeTag::T_OpExpr {
            let opexpr = node as *mut pg_sys::OpExpr;
            if context.target_opnos.contains(&(*opexpr).opno) {
                // Found a match! Set the flag and stop walking
                (*(data as *mut Context)).found = true;
                return true; // Stop walking
            }
        }
        pg_sys::expression_tree_walker(node, Some(walker), data)
    }

    struct Context {
        target_opnos: Vec<pg_sys::Oid>,
        found: bool,
    }

    let mut context = Context {
        target_opnos: target_opnos.to_vec(),
        found: false,
    };

    walker(node, addr_of_mut!(context).cast());
    context.found
}

/// Collects all unique RTIs (range table indices) from Var nodes in an expression tree.
/// Returns a HashSet of RTIs referenced by the expression.
pub unsafe fn expr_collect_rtis(
    node: *mut pg_sys::Node,
) -> std::collections::HashSet<pg_sys::Index> {
    #[pg_guard]
    unsafe extern "C-unwind" fn walker(
        node: *mut pg_sys::Node,
        data: *mut core::ffi::c_void,
    ) -> bool {
        if node.is_null() {
            return false;
        }

        let rtis = &mut *(data as *mut HashSet<pg_sys::Index>);

        if (*node).type_ == pg_sys::NodeTag::T_Var {
            let var = node as *mut pg_sys::Var;
            let varno = (*var).varno as pg_sys::Index;
            // Skip special RTIs like INNER_VAR/OUTER_VAR
            if varno > 0 && varno < pg_sys::INNER_VAR as pg_sys::Index {
                rtis.insert(varno);
            }
        }

        pg_sys::expression_tree_walker(node, Some(walker), data)
    }

    let mut rtis = HashSet::new();
    walker(node, addr_of_mut!(rtis).cast());
    rtis
}

/// A Var reference with its range table index and attribute number.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct VarRef {
    /// Range table index (varno)
    pub rti: pg_sys::Index,
    /// Attribute number (varattno), 1-indexed
    pub attno: pg_sys::AttrNumber,
}

/// Collects all unique Var references (RTI + attribute number) from an expression tree.
/// Returns a Vec of VarRef structs for each column referenced by the expression.
///
/// If `include_special_vars` is true, variables with special varnos (like INDEX_VAR) are included.
/// If false, only variables referencing base relations (varno > 0 and < INNER_VAR) are included.
pub unsafe fn expr_collect_vars(
    node: *mut pg_sys::Node,
    include_special_vars: bool,
) -> Vec<VarRef> {
    #[pg_guard]
    unsafe extern "C-unwind" fn walker(
        node: *mut pg_sys::Node,
        data: *mut core::ffi::c_void,
    ) -> bool {
        if node.is_null() {
            return false;
        }

        let (vars, include_special_vars) = &mut *(data as *mut (Vec<VarRef>, bool));

        if (*node).type_ == pg_sys::NodeTag::T_Var {
            let var = node as *mut pg_sys::Var;
            let varno = (*var).varno as pg_sys::Index;
            let varattno = (*var).varattno;

            // Standard check for base relation var:
            let is_base_rel_var = varno > 0 && varno < pg_sys::INNER_VAR as pg_sys::Index;

            if *include_special_vars {
                // Include if valid attno
                if varattno > 0 {
                    vars.push(VarRef {
                        rti: varno,
                        attno: varattno,
                    });
                }
            } else {
                // Only include base relation vars
                if is_base_rel_var && varattno > 0 {
                    vars.push(VarRef {
                        rti: varno,
                        attno: varattno,
                    });
                }
            }
        }

        pg_sys::expression_tree_walker(node, Some(walker), data)
    }

    let mut context = (Vec::new(), include_special_vars);
    walker(node, addr_of_mut!(context).cast());
    context.0
}

/// Look up a function in the pdb schema by name and argument types.
/// Returns InvalidOid if the function doesn't exist yet (e.g., during extension creation).
pub fn lookup_pdb_function(func_name: &str, arg_types: &[pg_sys::Oid]) -> pg_sys::Oid {
    unsafe {
        // Look up the pdb schema
        let pdb_schema = pg_sys::get_namespace_oid(c"pdb".as_ptr(), true);
        if pdb_schema == pg_sys::InvalidOid {
            return pg_sys::InvalidOid;
        }

        // Build the qualified function name list: pdb.<func_name>
        let mut func_name_list = PgList::<pg_sys::Node>::new();
        func_name_list.push(pg_sys::makeString(c"pdb".as_ptr() as *mut std::ffi::c_char) as *mut _);
        // Convert func_name to CString for makeString
        let func_name_cstr = std::ffi::CString::new(func_name).unwrap();
        func_name_list
            .push(pg_sys::makeString(func_name_cstr.as_ptr() as *mut std::ffi::c_char) as *mut _);

        // LookupFuncName returns InvalidOid if function doesn't exist (with missing_ok = true)
        pg_sys::LookupFuncName(
            func_name_list.as_ptr(),
            arg_types.len() as i32,
            arg_types.as_ptr(),
            true, // missing_ok = true, don't error if not found
        )
    }
}

/// Returns true if the pg_search extension is installed in the current database.
///
/// This is used to guard shared_preload_libraries hooks from touching schemas/types/functions
/// that only exist after `CREATE EXTENSION pg_search`.
pub fn pg_search_extension_installed() -> bool {
    unsafe { pg_sys::get_extension_oid(c"pg_search".as_ptr(), true) != pg_sys::InvalidOid }
}

/// RAII wrapper for `pg_sys::List` that automatically frees the list on drop.
///
/// This is useful when you need to create a temporary PostgreSQL list for use with
/// PostgreSQL functions and want to ensure it's properly freed even if the code
/// returns early or panics.
///
/// # Example
/// ```ignore
/// let temp_list = TempPgList::new();
/// temp_list.push(some_node as *mut std::ffi::c_void);
/// let result = pg_sys::some_function(temp_list.as_ptr());
/// // temp_list is automatically freed when it goes out of scope
/// ```
#[derive(Default)]
pub struct TempPgList(*mut pg_sys::List);

impl TempPgList {
    /// Create a new empty temporary list.
    pub fn new() -> Self {
        Self(std::ptr::null_mut())
    }

    /// Append a cell to the list.
    ///
    /// # Safety
    /// The caller must ensure that `datum` is a valid pointer that can be
    /// stored in a PostgreSQL list.
    pub unsafe fn push(&mut self, datum: *mut std::ffi::c_void) {
        self.0 = pg_sys::lappend(self.0, datum);
    }

    /// Get the raw pointer to the list.
    pub fn as_ptr(&self) -> *mut pg_sys::List {
        self.0
    }
}

impl Drop for TempPgList {
    fn drop(&mut self) {
        unsafe {
            if !self.0.is_null() {
                pg_sys::list_free(self.0);
            }
        }
    }
}

/// Filter out RestrictInfo entries whose clauses are implied by a partial index predicate.
///
/// When using a partial index (e.g., `CREATE INDEX ... WHERE deleted_at IS NULL`),
/// the index only contains rows that satisfy the predicate. If the query's WHERE clause
/// includes the same predicate, we don't need to create a heap filter for it since the
/// partial index already guarantees it.
///
/// This function uses PostgreSQL's `predicate_implied_by` to check if the index predicate
/// implies each query clause. If so, that clause is filtered out.
pub unsafe fn filter_implied_predicates(
    index_predicate: *mut pg_sys::List,
    restrict_info: &PgList<pg_sys::RestrictInfo>,
) -> PgList<pg_sys::RestrictInfo> {
    // If there's no partial index predicate, return the original list unchanged
    if index_predicate.is_null() {
        return PgList::from_pg(restrict_info.as_ptr());
    }

    // Build a new list with only the predicates that are NOT implied by the index predicate
    let mut filtered_list: *mut pg_sys::List = std::ptr::null_mut();

    for ri in restrict_info.iter_ptr() {
        let clause = (*ri).clause;
        let mut clause_list = TempPgList::new();
        clause_list.push(clause as *mut std::ffi::c_void);

        // Check if the index predicate implies this clause
        // predicate_implied_by(A, B, false) returns true if B => A
        let is_implied = pg_sys::predicate_implied_by(clause_list.as_ptr(), index_predicate, false);

        if !is_implied {
            filtered_list = pg_sys::lappend(filtered_list, ri as *mut std::ffi::c_void);
        }
    }

    PgList::from_pg(filtered_list)
}

#[macro_export]
macro_rules! debug1 {
    ($($arg:tt)*) => {{
        if unsafe { pg_sys::message_level_is_interesting(pg_sys::DEBUG1 as _) } {
            pg_sys::debug1!($($arg)*);
        }
    }};
}

#[macro_export]
macro_rules! debug2 {
    ($($arg:tt)*) => {{
        if unsafe { pg_sys::message_level_is_interesting(pg_sys::DEBUG2 as _) } {
            pg_sys::debug2!($($arg)*);
        }
    }};
}
