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

#![allow(clippy::unnecessary_cast)] // helps with integer casting differences between postgres versions
mod cost;
pub mod exec_methods;
pub mod parallel;
pub(crate) mod privdat;
pub mod projections;
mod scan_state;
pub(crate) mod telemetry;

use cost::{
    costable_drive_cost, decide_scan_parallelism, estimate_path_cost, parallel_divisor,
    topk_can_prune_for_method, CostMemo, DriveCost, ScanParallelismInputs, WorkerDecisionReason,
    WorkerPathPolicy,
};

use std::ffi::CStr;
use std::num::NonZeroUsize;
use std::ptr::addr_of_mut;
use std::sync::atomic::Ordering;

use crate::api::operator::{estimate_query_cost, estimate_selectivity_and_cost};
use crate::api::window_aggregate::window_agg_oid;
use crate::api::{HashMap, HashSet, Varno};
use crate::gucs;
use crate::index::fast_fields_helper::WhichFastField;
use crate::index::mvcc::MvccSatisfies;
use crate::index::reader::index::{SearchIndexReader, MAX_TOPK_FEATURES};
use crate::postgres::customscan::basescan::exec_methods::{
    fast_fields, normal::NormalScanExecState, ExecState,
};
use crate::postgres::customscan::basescan::privdat::PrivateData;
use crate::postgres::customscan::basescan::projections::score::uses_scores;
use crate::postgres::customscan::basescan::projections::snippet::{
    snippet_funcoids, snippet_positions_funcoids, snippets_funcoids, uses_snippets, SnippetType,
};
use crate::postgres::customscan::basescan::projections::window_agg::{
    deserialize_window_agg_placeholders, resolve_window_aggregate_filters_at_plan_time,
    WindowAggregateInfo,
};
use crate::postgres::customscan::basescan::scan_state::BaseScanState;
use crate::postgres::customscan::bitmap_intersection;
use crate::postgres::customscan::builders::custom_path::{
    restrict_info, CustomPathBuilder, ExecMethodType, Flags, RestrictInfoType,
};
use crate::postgres::customscan::builders::custom_scan::CustomScanBuilder;
use crate::postgres::customscan::builders::custom_state::{
    CustomScanStateBuilder, CustomScanStateWrapper,
};
use crate::postgres::customscan::explainer::Explainer;
use crate::postgres::customscan::orderby::{
    extract_pathkey_styles_with_sortability_check, PathKeyInfo, UnusableReason,
};
use crate::postgres::customscan::parallel::{
    compute_nworkers, max_useful_workers, segment_view, RowEstimate,
};
use crate::postgres::customscan::projections::{
    inject_placeholders, maybe_needs_const_projections, pullout_funcexprs,
};
use crate::postgres::customscan::qual_inspect::{
    extract_join_predicates, extract_quals, is_subplan, optimize_quals_with_heap_expr,
    PlannerContext, Qual, QualExtractState,
};
use crate::postgres::customscan::score_funcoids;
use crate::postgres::customscan::solve_expr::SolvePostgresExpressions;
use crate::postgres::customscan::{
    self, range_table, CustomScan, CustomScanState, RelPathlistHookArgs,
};
use crate::postgres::heap::{HeapFetchState, VisibilityChecker};
use crate::postgres::rel::PgSearchRelation;
use crate::postgres::rel_get_bm25_index;
use crate::postgres::storage::metadata::MetaPage;
use crate::postgres::utils::{
    filter_implied_predicates, is_unnest_func, missing_partial_index_predicate,
};
use crate::query::pdb_query::pdb;
use crate::query::SearchQueryInput;
use crate::schema::SearchIndexSchema;
use crate::{nodecast, DEFAULT_STARTUP_COST, PARAMETERIZED_SELECTIVITY, UNKNOWN_SELECTIVITY};
use crate::{FULL_RELATION_SELECTIVITY, UNASSIGNED_SELECTIVITY};

use crate::postgres::customscan::limit_offset::LimitOffset;
use pgrx::{pg_sys, FromDatum, IntoDatum, PgList, PgMemoryContexts};
use tantivy::snippet::SnippetGenerator;
use tantivy::Index;

#[derive(Default)]
pub struct BaseScan;

impl BaseScan {
    /// (Re-)initializes the search reader for the current execution context.
    ///
    /// This function handles three distinct execution scenarios:
    ///
    /// 1. **Leader Execution** (`ParallelWorkerNumber == -1`):
    ///    The scan is running in the main backend process. It uses `MvccSatisfies::Snapshot`
    ///    to see all segments visible to the current transaction's snapshot.
    ///
    /// 2. **Parallel-Aware Worker** (Worker with `parallel_state`):
    ///    The scan is part of a `parallel_aware` path (Partial Scan). Workers coordinate
    ///    via shared memory (DSM) to divide segments. It uses `MvccSatisfies::ParallelWorker`
    ///    to ensure it only queries segments explicitly identified and pinned by the leader.
    ///
    /// 3. **Replicated Worker** (Worker with NO `parallel_state`):
    ///    The scan is `parallel_safe` but NOT `parallel_aware`. This happens when a serial
    ///    scan runs inside a worker context (e.g., on the inner side of a Parallel Hash Join).
    ///    In this case, every worker executes the full scan independently using its own
    ///    transaction snapshot (`MvccSatisfies::Snapshot`).
    pub(crate) fn init_search_reader(state: &mut CustomScanStateWrapper<Self>) {
        let planstate = state.planstate();
        let expr_context = state.runtime_context;
        state
            .custom_state_mut()
            .prepare_query_for_execution(planstate, expr_context);

        // Open the index
        let indexrel = state
            .custom_state()
            .indexrel
            .as_ref()
            .expect("custom_state.indexrel should already be open");

        let search_query_input = state.custom_state().search_query_input();
        let need_scores = state.custom_state().need_scores();
        let needs_tokenizer_manager =
            search_query_input.needs_tokenizer() || state.custom_state().need_snippets();

        let search_reader = SearchIndexReader::open_with_context(
            indexrel,
            search_query_input.clone(),
            need_scores,
            unsafe {
                if pg_sys::ParallelWorkerNumber == -1 {
                    // the leader only sees snapshot-visible segments
                    MvccSatisfies::Snapshot
                } else if let Some(parallel_state) = state.custom_state().parallel_state() {
                    // the workers have their own rules, which is literally every segment
                    // this is because the workers pick a specific segment to query that
                    // is known to be held open/pinned by the leader but might not pass a ::Snapshot
                    // visibility test due to concurrent merges/garbage collects
                    MvccSatisfies::ParallelWorker(segment_view(parallel_state))
                } else {
                    // We are in a worker, but this is not a parallel-aware scan (e.g. we are running
                    // a serial scan inside a parallel worker, like in a Parallel Nested Loop Join).
                    // In this case, we behave like a normal snapshot scan.
                    MvccSatisfies::Snapshot
                }
            },
            std::ptr::NonNull::new(expr_context),
            std::ptr::NonNull::new(planstate),
            needs_tokenizer_manager,
        )
        .expect("should be able to open the search index reader");
        state.custom_state_mut().search_reader = Some(search_reader);

        let parallel_aware = unsafe { (*(*state.planstate()).plan).parallel_aware };
        if !parallel_aware {
            if let Some(cell) = state.custom_state().bitmap_cell.clone() {
                if cell.get().is_none() {
                    if let Some(bitmap_exec) = state.custom_state_mut().bitmap_exec.as_mut() {
                        if let Some(source) = unsafe { bitmap_exec.private_source() } {
                            cell.fill(source);
                        }
                    }
                }
            }
        }

        let csstate = addr_of_mut!(state.csstate);
        state.custom_state_mut().init_exec_method(csstate);

        if state.custom_state().need_snippets() {
            let mut snippet_generators: HashMap<SnippetType, Option<SnippetGenerator>> = state
                .custom_state_mut()
                .snippet_generators
                .drain()
                .collect();

            // Pre-compute enhanced queries for snippet generation if we have join predicates
            let enhanced_query_for_snippets =
                if let Some(ref join_predicate) = state.custom_state().join_predicates {
                    // Combine base query with join predicate for snippet generation
                    let base_query = state.custom_state().search_query_input();
                    Some(SearchQueryInput::Boolean {
                        must: vec![base_query.clone()],
                        should: vec![join_predicate.clone()],
                        must_not: vec![],
                        minimum_should_match: None,
                    })
                } else {
                    None
                };

            for (snippet_type, generator) in &mut snippet_generators {
                // Use enhanced query if available, otherwise use base query
                let query_to_use = enhanced_query_for_snippets
                    .as_ref()
                    .unwrap_or_else(|| state.custom_state().search_query_input());

                let mut new_generator = state
                    .custom_state()
                    .search_reader
                    .as_ref()
                    .unwrap()
                    .snippet_generator(
                        snippet_type.field().root(),
                        query_to_use,
                        std::ptr::NonNull::new(expr_context),
                    );

                unsafe {
                    let estate = (*csstate).ss.ps.state;
                    snippet_type.configure_generator(&mut new_generator.1, estate);
                }

                *generator = Some(new_generator.1);
            }

            state.custom_state_mut().snippet_generators = snippet_generators;
        }

        unsafe {
            inject_pdb_placeholders(state);
        }
    }

    #[allow(clippy::too_many_arguments)]
    unsafe fn extract_all_possible_quals(
        builder: &mut CustomPathBuilder<BaseScan>,
        root: *mut pg_sys::PlannerInfo,
        rti: pg_sys::Index,
        restrict_info: PgList<pg_sys::RestrictInfo>,
        ri_type: RestrictInfoType,
        indexrel: &PgSearchRelation,
        uses_score_or_snippet: bool,
        attempt_pushdown: bool,
    ) -> Option<Qual> {
        let mut state = QualExtractState::default();
        let context = PlannerContext::from_planner(root);

        // Filter out predicates that are implied by the partial index predicate.
        // If a partial index has predicate P (e.g., "deleted_at IS NULL"), and the query
        // also has predicate P, we don't need to create a heap filter for P since the
        // partial index already guarantees it.
        let filtered_restrict_info = filter_implied_predicates(indexrel.rd_indpred, &restrict_info);

        let mut quals = extract_quals(
            &context,
            rti,
            filtered_restrict_info.as_ptr().cast(),
            ri_type,
            indexrel,
            false, // Base relation quals should not convert external to all
            &mut state,
            attempt_pushdown,
        );

        // If full extraction failed (e.g., baserestrictinfo contains SubPlan from
        // RLS policies alongside our @@@ operator), try partial extraction: extract
        // each restrict_info item individually, skipping ones we can't handle.
        // Only use partial extraction when ALL skipped clauses are SubPlans
        // (which will be evaluated via plan.qual). If any non-SubPlan clause
        // is skipped, fall back to let PostgreSQL handle the query normally.
        //
        // TODO: We do something similar in `collect_join_sources_base_rel`,
        // is unification possible?
        if quals.is_none() {
            let mut partial_quals = Vec::new();
            let mut partial_state = QualExtractState::default();
            let mut all_skipped_are_subplans = true;
            for ri in filtered_restrict_info.iter_ptr() {
                if let Some(qual) = extract_quals(
                    &context,
                    rti,
                    ri.cast(),
                    ri_type,
                    indexrel,
                    false,
                    &mut partial_state,
                    attempt_pushdown,
                ) {
                    partial_quals.push(qual);
                } else if !is_subplan(ri.cast(), root) {
                    all_skipped_are_subplans = false;
                }
            }
            if !partial_quals.is_empty()
                && partial_state.uses_our_operator
                && all_skipped_are_subplans
            {
                state = partial_state;
                quals = if partial_quals.len() == 1 {
                    partial_quals.pop()
                } else {
                    Some(Qual::And(partial_quals))
                };
            }
        }

        let allow_without_operator =
            gucs::enable_custom_scan_without_operator() || query_has_window_agg_functions(root);

        // If we couldn't push down quals, try to push down quals from the join
        // This is only done if we have a join predicate, and only if we have used our operator
        let quals = if quals.is_none() {
            let joinri: PgList<pg_sys::RestrictInfo> =
                PgList::from_pg(builder.args().rel().joininfo);
            let mut quals = extract_quals(
                &context,
                rti,
                joinri.as_ptr().cast(),
                RestrictInfoType::Join,
                indexrel,
                true, // Join quals should convert external to all
                &mut state,
                attempt_pushdown,
            );

            let quals =
                Self::handle_heap_expr_optimization(&state, &mut quals, allow_without_operator);

            // If we have found something to push down in the join, then we can use the join quals
            // Note: these Join quals won't help in filtering down the data (as they contain
            // external vars, e.g. `b.category_name @@@ "technology"` in
            // `a.name @@@ "abc" OR b.category_name @@@ "technology"`), and we cannot evaluate
            // boolean expressions that contain external vars. That's why, when handling the Join
            // quals, we'd endup scanning the whole tantivy index.
            // However, the Join quals help with scoring and snippet generation, as the documents
            // that match partially the Join quals will be scored and snippets generated. That is
            // why it only makes sense to use the Join quals if we have used our operator and
            // also used pdb.score or pdb.snippet functions in the query.
            if state.uses_our_operator && uses_score_or_snippet {
                quals
            } else {
                None
            }
        } else {
            Self::handle_heap_expr_optimization(&state, &mut quals, allow_without_operator)
        };

        // Finally, decide whether we can actually use the extracted quals.
        // We allow custom scan if:
        // 1. The query uses one of our operators -- i.e. `@@@`. Plain PostgreSQL
        //    operators we can lower into a Tantivy predicate (such as
        //    `path <@ 'Top.Science'::ltree`) do NOT count here: they get pushed
        //    down once the scan is chosen, but they don't justify the scan on
        //    their own, OR
        // 2. enable_custom_scan_without_operator is true, OR
        // 3. The query has window aggregates (pdb.agg()) that we must handle.
        if state.uses_our_operator || allow_without_operator {
            quals
        } else {
            None
        }
    }

    unsafe fn handle_heap_expr_optimization(
        state: &QualExtractState,
        quals: &mut Option<Qual>,
        allow_without_operator: bool,
    ) -> Option<Qual> {
        if state.uses_heap_expr && !state.uses_our_operator && !allow_without_operator {
            return None;
        }

        // Apply HeapExpr optimization to the base relation quals
        if let Some(ref mut q) = quals {
            optimize_quals_with_heap_expr(q);
        }

        quals.clone()
    }
}

/// Check if the query's target list contains window_agg() function calls
///
/// This is called AFTER window function replacement in BaseScan's create_custom_path.
/// It looks for FuncExpr nodes with window_agg() OID, NOT WindowFunc nodes.
///
/// This is different from query_has_window_functions() in hook.rs which looks for WindowFunc
/// nodes BEFORE replacement in the planner hook.
///
/// Used to determine if we should create a custom path even without @@@ operator.
///
/// Also preserves the historical top-level pdb.agg() validation: if a target
/// entry itself is pdb.agg(), the planner hook did not replace it (e.g. not a
/// TopK query), and we reject it. Recursive detection is only for window_agg()
/// placeholders because plan_custom_path deserializes those recursively later.
pub(super) unsafe fn query_has_window_agg_functions(root: *mut pg_sys::PlannerInfo) -> bool {
    use pgrx::pg_guard;
    use pgrx::pg_sys::expression_tree_walker;

    if root.is_null() || (*root).parse.is_null() {
        return false;
    }

    let parse = (*root).parse;
    let window_agg_func_oid = window_agg_oid();

    // If functions don't exist yet (e.g., during extension creation), skip check
    if window_agg_func_oid == pg_sys::InvalidOid {
        return false;
    }

    let paradedb_agg_func_oid = crate::api::agg_funcoid();
    if !(*parse).targetList.is_null() {
        let target_list = PgList::<pg_sys::TargetEntry>::from_pg((*parse).targetList);
        for te in target_list.iter_ptr() {
            let Some(func_expr) = nodecast!(FuncExpr, T_FuncExpr, (*te).expr) else {
                continue;
            };

            let func_oid = (*func_expr).funcid.to_u32();
            if func_oid == window_agg_func_oid.to_u32() {
                return true;
            }
            if func_oid == paradedb_agg_func_oid.to_u32() {
                pgrx::error!(
                    "pdb.agg() can only be used as a window function in Top K queries \
                     (queries with ORDER BY and LIMIT). For GROUP BY aggregates, use standard \
                     SQL aggregates like COUNT(*), SUM(), etc. \
                     Hint: Try using '@@@ pdb.all()' with ORDER BY and LIMIT, \
                     or see https://github.com/paradedb/paradedb/issues for more information."
                );
            }
        }
    }

    struct Context {
        window_agg_func_oid: u32,
        found: bool,
    }

    // window_agg() can appear nested inside CASE expressions, arithmetic,
    // coercions, etc. The deserialize_window_agg_placeholders pass that runs
    // later in plan_custom_path walks the tree recursively; this detector must
    // do the same so the cost-model gate matches that pass's reality.
    #[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.cast::<Context>();
        if (*context).found {
            return true;
        }

        if let Some(func_expr) = nodecast!(FuncExpr, T_FuncExpr, node) {
            let func_oid = (*func_expr).funcid.to_u32();
            if func_oid == (*context).window_agg_func_oid {
                (*context).found = true;
                return true;
            }
        }

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

    let mut context = Context {
        window_agg_func_oid: window_agg_func_oid.to_u32(),
        found: false,
    };

    if !(*parse).targetList.is_null() {
        expression_tree_walker(
            (*parse).targetList.cast(),
            Some(walker),
            (&mut context as *mut Context).cast(),
        );
    }

    context.found
}

/// Classification of any set-returning function found in the target list,
/// used to decide whether pushing LIMIT through this scan is safe. PG sets
/// `limit_tuples == -1.0` whenever any SRF is present, so we walk once and
/// distinguish between "no SRF / safe (unnest of a ParadeDB SRF placeholder) /
/// unsafe (any other SRF, or `unnest` of a user expression)".
///
/// The `Safe` case must stay narrow: `unnest` is only row-preserving when its
/// argument is guaranteed non-empty, and for a user expression that is a
/// data-dependent property unknowable at plan time — an empty or NULL array
/// silently drops rows below the LIMIT with nothing to refill (see #5573).
/// The paradedb snippet SRFs (`pdb.snippets`, `pdb.snippet_positions`, and
/// their legacy `paradedb.*` shims) are placeholders that only fire for rows
/// the CustomScan actually matched; their output vectors are non-empty by
/// construction on matching rows, so `unnest(pdb.snippets(...))` may safely
/// take the LIMIT-pushdown fast path.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TargetListSrf {
    None,
    Safe,
    Unsafe,
}

impl TargetListSrf {
    fn is_safe(self) -> bool {
        matches!(self, TargetListSrf::Safe)
    }
    fn is_unsafe(self) -> bool {
        matches!(self, TargetListSrf::Unsafe)
    }
}

/// Walk the target list once and classify any SRFs. Only `unnest` whose
/// argument is a ParadeDB snippet placeholder function is treated as safe;
/// everything else — arbitrary `unnest` of a user expression, or any other
/// SRF — is unsafe so LIMIT pushdown stops above the scan.
unsafe fn classify_target_list_srf(root: *mut pg_sys::PlannerInfo) -> TargetListSrf {
    if root.is_null() || (*root).parse.is_null() || (*(*root).parse).targetList.is_null() {
        return TargetListSrf::None;
    }
    let target_list = PgList::<pg_sys::TargetEntry>::from_pg((*(*root).parse).targetList);
    let mut found_safe = false;
    for te in target_list.iter_ptr() {
        if (*te).expr.is_null() || !pg_sys::expression_returns_set((*te).expr.cast()) {
            continue;
        }
        match nodecast!(FuncExpr, T_FuncExpr, (*te).expr) {
            Some(func_expr)
                if is_unnest_func((*func_expr).funcid) && unnest_arg_is_paradedb_srf(func_expr) =>
            {
                found_safe = true
            }
            _ => return TargetListSrf::Unsafe,
        }
    }
    if found_safe {
        TargetListSrf::Safe
    } else {
        TargetListSrf::None
    }
}

/// True when the sole argument to an `unnest` FuncExpr is a call to one of
/// the ParadeDB snippet/placeholder SRFs whose output vector is non-empty by
/// construction on matching rows (`pdb.snippets`, `pdb.snippet_positions`,
/// and their legacy `paradedb.*` shims). Any other argument shape — a Var,
/// a CASE, a user function — cannot be proved non-empty at plan time.
unsafe fn unnest_arg_is_paradedb_srf(unnest_expr: *mut pg_sys::FuncExpr) -> bool {
    use crate::postgres::customscan::basescan::projections::snippet::{
        snippet_positions_funcoids, snippets_funcoids,
    };

    let args = PgList::<pg_sys::Node>::from_pg((*unnest_expr).args);
    if args.len() != 1 {
        return false;
    }
    let arg = args.get_ptr(0).unwrap_or(std::ptr::null_mut());
    let Some(inner) = nodecast!(FuncExpr, T_FuncExpr, arg) else {
        return false;
    };
    let inner_oid = (*inner).funcid;
    snippets_funcoids().contains(&inner_oid) || snippet_positions_funcoids().contains(&inner_oid)
}

pub struct BaseScanDeclineReason {
    pub message: &'static str,
}

impl BaseScanDeclineReason {
    pub fn new(message: &'static str) -> Self {
        Self { message }
    }
}

/// Returns `Ok(())` if all predicates in `baserestrictinfo` can be fully
/// evaluated inside custom scan. Returns `Err` if Postgres will need to apply a
/// post-filter above the scan.
///
/// Used to gate:
///   1. LIMIT pushdown: pushing LIMIT below a post-filter would cap output prematurely.
///   2. Window aggregates (`pdb.agg()`): window aggregates require full pushdown to avoid data loss.
unsafe fn has_non_pushable_predicates(
    rel: *mut pg_sys::RelOptInfo,
    quals_pushed: &Option<Qual>,
) -> Result<(), BaseScanDeclineReason> {
    let restrict_list = PgList::<pg_sys::RestrictInfo>::from_pg((*rel).baserestrictinfo);

    for ri in restrict_list.iter_ptr() {
        let clause = (*ri).clause as *mut pg_sys::Node;
        if pg_sys::contain_subplans(clause) {
            return Err(BaseScanDeclineReason::new(
                "WHERE clause contains subqueries which cannot be evaluated in a custom scan",
            ));
        }
        if pg_sys::contain_volatile_functions(clause) {
            return Err(BaseScanDeclineReason::new(
                "WHERE clause contains volatile functions which cannot be safely evaluated in a custom scan",
            ));
        }
    }

    // If qual extraction returned None but restrictions exist,
    // something is being left as a post-filter.
    if quals_pushed.is_none() && !restrict_list.is_empty() {
        let message = if crate::gucs::enable_filter_pushdown() {
            "WHERE clause contains predicates that cannot be pushed down"
        } else {
            "WHERE clause contains predicates that cannot be pushed down, and paradedb.enable_filter_pushdown is disabled"
        };
        return Err(BaseScanDeclineReason::new(message));
    }

    Ok(())
}

/// Returns true if the query's LIMIT can safely be pushed into this scan node.
///
/// Three conditions must hold:
///   1. This rel bounds the output cardinality (single rel, partitioned, or
///      the driving side of a LEFT LATERAL join). Without this, pushing LIMIT
///      to the inner side of a join would silently truncate results.
///   2. No non-pushable predicates sit between this scan and the LIMIT.
///   3. No unsafe SRFs (anything other than `unnest`) in the target list.
///
/// Independent of whether PG's `limit_tuples` is set or the LIMIT involves a
/// `Param` — that's the caller's policy decision.
unsafe fn is_limit_pushdown_safe(
    root: *mut pg_sys::PlannerInfo,
    rel: *mut pg_sys::RelOptInfo,
    baserels: *mut pg_sys::Bitmapset,
    rti: pg_sys::Index,
    quals: &Option<Qual>,
) -> bool {
    let rel_is_single_or_partitioned = pg_sys::bms_equal((*rel).relids, baserels)
        || range_table::is_partitioned_table_setup(root, (*rel).relids, baserels);
    let is_left_driven_lateral =
        is_left_join_lateral(root, rel) && where_clause_only_references_left(root, rti);

    (rel_is_single_or_partitioned || is_left_driven_lateral)
        && has_non_pushable_predicates(rel, quals).is_ok()
        && !classify_target_list_srf(root).is_unsafe()
}

impl CustomScan for BaseScan {
    const NAME: &'static CStr = c"ParadeDB Base Scan";

    type Args = RelPathlistHookArgs;
    type State = BaseScanState;
    type PrivateData = PrivateData;

    fn exec_methods() -> pg_sys::CustomExecMethods {
        pg_sys::CustomExecMethods {
            CustomName: Self::NAME.as_ptr(),
            BeginCustomScan: Some(customscan::exec::begin_custom_scan::<Self>),
            ExecCustomScan: Some(customscan::exec::exec_custom_scan::<Self>),
            EndCustomScan: Some(customscan::exec::end_custom_scan::<Self>),
            ReScanCustomScan: Some(customscan::exec::rescan_custom_scan::<Self>),
            MarkPosCustomScan: None,
            RestrPosCustomScan: None,
            EstimateDSMCustomScan: Some(customscan::dsm::estimate_dsm_custom_scan::<Self>),
            InitializeDSMCustomScan: Some(customscan::dsm::initialize_dsm_custom_scan::<Self>),
            ReInitializeDSMCustomScan: Some(customscan::dsm::reinitialize_dsm_custom_scan::<Self>),
            InitializeWorkerCustomScan: Some(
                customscan::dsm::initialize_worker_custom_scan::<Self>,
            ),
            ShutdownCustomScan: Some(customscan::exec::shutdown_custom_scan::<Self>),
            ExplainCustomScan: Some(customscan::exec::explain_custom_scan::<Self>),
        }
    }

    fn create_custom_path(mut builder: CustomPathBuilder<Self>) -> Vec<pg_sys::CustomPath> {
        let paths = (|| unsafe {
            let (restrict_info, ri_type) = restrict_info(builder.args().rel());
            let rel = builder.args().rel;

            // Check if the query has window aggregates (pdb.agg() or window_agg())
            let has_window_aggs = query_has_window_agg_functions(builder.args().root);

            if matches!(ri_type, RestrictInfoType::None) && !has_window_aggs {
                // this relation has no restrictions (WHERE clause predicates) and no window aggregates,
                // so there's no need for us to do anything
                return None;
            }

            let rti = builder.args().rti;
            let (table, bm25_index) = {
                let rte = builder.args().rte();

                // we only support plain relation and join rte's
                if rte.rtekind != pg_sys::RTEKind::RTE_RELATION
                    && rte.rtekind != pg_sys::RTEKind::RTE_JOIN
                {
                    return None;
                }

                // and we only work on plain relations
                let relkind = pg_sys::get_rel_relkind(rte.relid) as u8;
                if relkind != pg_sys::RELKIND_RELATION && relkind != pg_sys::RELKIND_MATVIEW {
                    return None;
                }

                // and that relation must have a `USING bm25` index
                let (table, bm25_index) = rel_get_bm25_index(rte.relid)?;

                (table, bm25_index)
            };

            let root = builder.args().root;

            // quick look at the target list to see if we might need to do our const projections
            let target_list = (*(*builder.args().root).parse).targetList;
            let maybe_needs_const_projections = maybe_needs_const_projections(target_list.cast());

            //
            // look for quals we can support.  we do this first so that we can get out early if this
            // isn't a query we can support.
            //
            // Opening the Directory and Index down below is expensive, so if we can avoid it,
            // especially for non-SELECT (ie, UPDATE) statements, that's good
            //
            let is_select =
                (*(*builder.args().root).parse).commandType == pg_sys::CmdType::CMD_SELECT;
            let quals = Self::extract_all_possible_quals(
                &mut builder,
                root,
                rti,
                PgList::from_pg(restrict_info.as_ptr()),
                ri_type,
                &bm25_index,
                maybe_needs_const_projections,
                is_select,
            );

            // If window aggregates are present, validate that the WHERE clause contains no
            // non-pushable predicates (e.g. subqueries, volatile functions, or unpushable
            // filters when filter pushdown is disabled) that would cause silent data loss or incorrect results.
            if has_window_aggs {
                if let Err(reason) = has_non_pushable_predicates(rel, &quals) {
                    pgrx::error!("Cannot execute window aggregate: {}", reason.message);
                }
            }

            // Resolve quals for the custom scan path:
            // - `Some(q)`: Use extracted pushdown or HeapExpr quals.
            // - `None` with `has_window_aggs`: No quals were extracted (e.g., no WHERE clause),
            //   but pdb.agg() window functions require ParadeDB custom scan execution, so default to Qual::All.
            // - `None` without `has_window_aggs`: Neither quals nor window aggs exist; decline custom path.
            let quals = match quals {
                Some(q) => q,
                None if has_window_aggs => Qual::All,
                None => return None,
            };

            if missing_partial_index_predicate(bm25_index.rd_indpred, &restrict_info) {
                return None;
            }

            //
            // ===================
            // If we make it this far, we're going to submit a path... it better be a good one!
            // ====================
            //

            // TODO: `impl Default for PrivateData` requires that many fields are in invalid
            // states. Should consider having a separate builder for PrivateData.
            let mut custom_private = PrivateData::default();

            let segment_count = {
                let directory = MvccSatisfies::LargestSegment.directory(&bm25_index);
                let segment_count = directory.total_segment_count(); // return value only valid after the index has been opened
                Index::open(directory).expect("custom_scan: should be able to open index");
                segment_count.load(Ordering::Relaxed)
            };
            let schema = bm25_index
                .schema()
                .expect("custom_scan: should have a schema");
            let index_expressions = bm25_index.index_expressions();
            let topk_pathkey_info =
                pullup_topk_pathkeys(rti, &schema, root, Some(&index_expressions));

            #[cfg(feature = "pg15")]
            let baserels = (*builder.args().root).all_baserels;
            #[cfg(any(feature = "pg16", feature = "pg17", feature = "pg18"))]
            let baserels = (*builder.args().root).all_query_rels;

            // Detect if we are in a join context (more than 1 base relation in the query)
            // If so, we want to be aggressive with parallelism to enable Parallel Hash Join
            let is_join_context = pg_sys::bms_num_members(baserels) > 1;

            // GROUP BY (`groupClause`) or SELECT DISTINCT (`distinctClause`) above this scan: PG
            // under-costs the serial HashAggregate (see `decide_scan_parallelism`), so route these
            // through the row heuristic. Scalar aggregates leave both empty and stay cost-chosen.
            let parse = (*builder.args().root).parse;
            let has_grouping = !parse.is_null()
                && (!(*parse).groupClause.is_null() || !(*parse).distinctClause.is_null());

            // Push the LIMIT/OFFSET into this scan when one of:
            //   - PG already proved it safe (`limit_tuples > -1.0`)
            //   - The value is a Param (PG can't evaluate at plan time but
            //     we can at exec time)
            //   - A row-preserving `unnest` of a ParadeDB snippet SRF zeroed
            //     PG's `limit_tuples`, but `limit + offset` rows here is
            //     still enough (the paradedb SRF is non-empty by
            //     construction on matching rows). `unnest` of a user
            //     expression is NOT covered by this override — see #5573.
            // `is_limit_pushdown_safe` then gates on topology + predicates +
            // unsafe SRFs.
            let raw_limit_offset = LimitOffset::from_root(builder.args().root);
            let limit_offset = raw_limit_offset.filter(|lo| {
                let pg_says_pushable = (*builder.args().root).limit_tuples > -1.0;
                let unnest_override = classify_target_list_srf(builder.args().root).is_safe();

                // PG zeroes `limit_tuples` whenever a WindowAgg sits between the LIMIT and the
                // scan, because in general a window function needs every row. When the window is
                // a bare ranking over the same order as the LIMIT, the top N rows *are* the first
                // N in window order, so feeding the WindowAgg only those N is result-preserving.
                // This is what makes the natural RRF hybrid-search shape a Top K scan. See #5742.
                let window_override = window_limit_pushdown_is_safe(parse);

                (pg_says_pushable || lo.has_any_param() || unnest_override || window_override)
                    && is_limit_pushdown_safe(
                        builder.args().root,
                        rel,
                        baserels,
                        rti,
                        &Some(quals.clone()),
                    )
            });

            // Get all columns referenced by this RTE throughout the entire query
            let referenced_columns = collect_maybe_fast_field_referenced_columns(rti, rel);

            // Save the count of referenced columns for decision-making
            custom_private.set_referenced_columns_count(referenced_columns.len());

            let has_any_limit = limit_offset.is_some();
            let is_maybe_topk = has_any_limit && topk_pathkey_info.is_usable();

            // When collecting which_fast_fields, analyze the entire set of referenced columns,
            // not just those in the target list. To avoid execution-time surprises, the "planned"
            // fast fields must be a superset of the fast fields which are extracted from the
            // execution-time target list: see `assign_exec_method` for more info.
            custom_private.set_planned_which_fast_fields(
                exec_methods::fast_fields::collect_fast_fields(
                    target_list,
                    &referenced_columns,
                    rti,
                    &table,
                    &bm25_index,
                    false,
                )
                .into_iter()
                .collect(),
            );

            let mut query = SearchQueryInput::from(&quals);
            let norm_selec = if restrict_info.len() == 1 {
                (*restrict_info.get_ptr(0).unwrap()).norm_selec
            } else {
                UNASSIGNED_SELECTIVITY
            };

            // Seeded only by the final `else` branch's selectivity open (every other branch leaves
            // it `None`). It feeds the TopK cost memo so the worker decision reuses that open
            // instead of opening the index a second time.
            let mut precomputed_query_cost: Option<u64> = None;

            let selectivity = if norm_selec != UNASSIGNED_SELECTIVITY {
                // we can use the norm_selec that already happened
                norm_selec
            } else if quals.contains_external_var() {
                // if the query has external vars (references to other relations which decide whether the rows in this
                // relation are visible) then we end up returning *everything* from _this_ relation
                FULL_RELATION_SELECTIVITY
            } else if quals.contains_exprs() {
                // if the query has expressions then it's parameterized and we have to guess something
                PARAMETERIZED_SELECTIVITY
            } else {
                // Ask the index. This is the one branch that opens, so reuse that same
                // open's cost for the TopK worker decision instead of opening twice.
                let (sel, cost) = estimate_selectivity_and_cost(&bm25_index, query.clone());
                precomputed_query_cost = cost;
                sel.unwrap_or(UNKNOWN_SELECTIVITY)
            };

            // Use planning_estimate for costing so parameterized limits still
            // contribute a non-zero row estimate (otherwise the parallel-worker
            // estimator collapses).
            let float_limit = limit_offset.as_ref().map(|lo| lo.planning_estimate());

            custom_private.set_heaprelid(table.oid());
            custom_private.set_indexrelid(bm25_index.oid());
            custom_private.set_range_table_index(rti);
            custom_private.set_query(query.clone());
            custom_private.set_limit_offset(limit_offset.clone());
            custom_private.set_segment_count(segment_count);

            // Determine whether we might be able to sort.
            if is_maybe_topk && topk_pathkey_info.pathkeys().is_some() {
                custom_private.set_maybe_orderby_info(topk_pathkey_info.pathkeys());
            }

            // Choose the exec method type, and make claims about whether it is sorted.
            let limit_is_explicit = has_any_limit && !is_minmax_implicit_limit(builder.args().root);

            // calculate the total number of rows that might match the query, and the number of
            // rows that we expect that scan to return: these may be different in the case of a
            // `limit`.
            //
            // Use RowEstimate enum to distinguish between known and unknown row counts.
            // Unknown is used when the table hasn't been ANALYZEd.
            // TODO: Convert to use RowEstimate::from_reltuples.
            let row_estimate = match table.reltuples() {
                Some(reltuples) if reltuples > 0.0 => {
                    let estimated = (reltuples as f64 * selectivity).max(1.0) as u64;
                    RowEstimate::Known(estimated)
                }
                _ => RowEstimate::Unknown,
            };
            let base_result_rows = match row_estimate.known_rows() {
                Some(rows) => rows.min(float_limit.unwrap_or(f64::MAX)),
                None => float_limit.unwrap_or(1.0),
            }
            .max(1.0);

            let harvested_bitmap = bitmap_intersection::BitmapPlanner::from_query(
                root,
                builder.args().rel,
                bm25_index.oid(),
                &quals,
                row_estimate.known_rows().map(|rows| rows as f64),
            )
            .and_then(|planner| planner.harvest());
            if let Some(harvested) = &harvested_bitmap {
                harvested.rewrite_query(&mut query);
                custom_private.set_query(query.clone());
            }

            let exec_method_types = choose_exec_method(
                &custom_private,
                &topk_pathkey_info,
                limit_is_explicit,
                table.name(),
            );

            //
            // finally, we have enough information to set the cost and estimation information, and
            // to decide on parallelism
            //

            let startup_cost =
                DEFAULT_STARTUP_COST + harvested_bitmap.as_ref().map_or(0.0, |h| h.build_cost);
            let mut custom_paths = Vec::new();
            let parallel_leader_participates = pg_sys::parallel_leader_participation;
            // Seed the cost memo from the open create_custom_path already did for selectivity (if
            // any), so the cost computation opens the index at most once per query.
            let mut cost_memo = CostMemo::from_precomputed(precomputed_query_cost);

            // Cost the query once (memoized) for costable scans; `None` marks the scan uncostable, so
            // pg_search forces the worker decision (and an effective-LIMIT scan uses the magnitude in
            // `cost_test_limited`). Skip the open entirely for un-ANALYZEd tables -- the row heuristic
            // decides those without a cost.
            let drive_cost = match row_estimate {
                RowEstimate::Known(_) => costable_drive_cost(
                    &query,
                    &bm25_index,
                    &quals,
                    builder.args().root,
                    &mut cost_memo,
                ),
                RowEstimate::Unknown => None,
            };

            // For each execution method variant, decide a `WorkerPathPolicy` and emit the path(s) it
            // calls for: one serial path, one partial (parallel) path, or -- for a costable no-LIMIT
            // scan -- both, so PostgreSQL costs the Gather and chooses serial-vs-parallel itself (#4664).
            for method in exec_method_types {
                let per_tuple_cost = match &method {
                    // returning fields from fast fields
                    ExecMethodType::Columnar { .. } => pg_sys::cpu_index_tuple_cost,
                    // requires heap access to return fields
                    _ => pg_sys::cpu_tuple_cost,
                };

                let is_sorted = method.declares_sorted_output();
                let consider_parallel_local = (*builder.args().rel).consider_parallel;
                let prunability = topk_can_prune_for_method(&method, builder.args().root, &quals);

                // Decide the policy first: its reason gates the path cost below.
                let policy = decide_scan_parallelism(ScanParallelismInputs {
                    prunability,
                    query: &query,
                    drive_cost,
                    row_estimate,
                    is_sorted,
                    limit: float_limit,
                    base_result_rows,
                    segment_count,
                    consider_parallel: consider_parallel_local,
                    quals: &quals,
                    root: builder.args().root,
                    parallel_leader_participates,
                    is_join_context,
                    has_grouping,
                });
                let reason = policy.reason();

                // Path cost. A prunable TopK excludes `drive_cost`: Block-WAND prunes it sublinear, so
                // the full-docset drive cost would overstate the work; the output cost (~k rows) is
                // the better estimate. (Same Block-WAND blind spot that forces the serial decision,
                // applied to the cost.)
                let path_drive_cost = match reason {
                    WorkerDecisionReason::BlockWandPrunable => None,
                    WorkerDecisionReason::CostModel
                    | WorkerDecisionReason::CostModelLimited
                    | WorkerDecisionReason::SortedPerSegment
                    | WorkerDecisionReason::RowHeuristic => drive_cost,
                };
                let drive = match (path_drive_cost, row_estimate.known_rows()) {
                    (Some(cost), Some(matches)) => Some(DriveCost { cost, matches }),
                    _ => None,
                };
                let cost_basis = estimate_path_cost(
                    is_sorted,
                    per_tuple_cost,
                    base_result_rows,
                    drive,
                    float_limit,
                );

                // We must force this path (not interchangeable with native paths) if we need const
                // projections for scores/snippets, or it's a TopK, or the predicate matches all.
                let force = maybe_needs_const_projections
                    || matches!(method, ExecMethodType::TopK { .. })
                    || quals.contains_all();

                // Pathkeys to declare on every sibling of this method, so a Gather Merge built over
                // the partial sibling preserves the ordering instead of degrading to a plain Gather.
                let topk_pathkeys = matches!(
                    method,
                    ExecMethodType::TopK {
                        orderby_info: Some(..),
                        ..
                    }
                )
                .then(|| (*builder.args().root).query_pathkeys);

                // Build one sibling path. `nworkers == None` => serial (divisor 1.0); `Some` =>
                // parallel-aware partial path. `offer_parallel` marks a partial path PostgreSQL may
                // reject for the serial sibling (see `hook::add_path`).
                let make_path =
                    |nworkers: Option<NonZeroUsize>, offer_parallel: bool, forced: bool| {
                        let divisor = nworkers
                            .map_or(1.0, |n| parallel_divisor(n, parallel_leader_participates));
                        let rows = base_result_rows / divisor;
                        let total_cost = startup_cost + cost_basis.parallelizable_cost / divisor;

                        let mut path_builder = CustomPathBuilder::<Self>::new(
                            builder.args().root,
                            builder.args().rel,
                            *builder.args(),
                        )
                        .set_force_path(forced);

                        // Our BaseScan is always parallel-safe (can run in a worker), even when it's not
                        // parallel-aware (splitting segments).
                        if consider_parallel_local {
                            path_builder = path_builder.set_parallel_safe(true);
                        }
                        if let Some(nworkers) = nworkers {
                            path_builder = path_builder.set_parallel(nworkers.get());
                        }
                        if offer_parallel {
                            path_builder = path_builder.set_flag(Flags::OfferParallel);
                        }
                        path_builder = path_builder
                            .set_rows(rows)
                            .set_startup_cost(startup_cost)
                            .set_total_cost(total_cost)
                            .set_flag(Flags::Projection);

                        if let Some(pathkeys) = topk_pathkeys {
                            path_builder = path_builder.set_pathkeys(pathkeys);
                        }

                        if let Some(harvested) = &harvested_bitmap {
                            let mut children = PgList::<pg_sys::Path>::new();
                            children.push(harvested.path);
                            path_builder = path_builder.set_custom_paths(children);
                        }

                        let mut method_private = custom_private.clone();
                        method_private.set_exec_method_type(method.clone());
                        method_private.set_use_sorted_path(is_sorted);
                        method_private.set_worker_selection_reason(reason);
                        path_builder.build(method_private)
                    };

                match policy {
                    WorkerPathPolicy::SerialOnly { .. } => {
                        custom_paths.push(make_path(None, false, force));
                    }
                    WorkerPathPolicy::ParallelOnly { nworkers, .. } => {
                        custom_paths.push(make_path(Some(nworkers), false, force));
                    }
                    WorkerPathPolicy::CostedBoth { nworkers, .. } => {
                        // Force the serial sibling so the hook clears PostgreSQL's native paths (same
                        // mechanism as TopK / score-snippet / `all()`), leaving only our serial and
                        // partial paths for PostgreSQL to choose between -- the honest scan-work cost
                        // would otherwise let PostgreSQL's own (correct, but fast-field/Block-WAND-
                        // less) index scan over the BM25 index undercut us on cost (see module docs).
                        // Today the only multi-method emitter is Columnar: it emits unsorted first
                        // and the index-sort variant second, and the sorted variant exists only for
                        // an ORDER BY shape where it is the useful survivor. If another multi-method
                        // emitter is added, re-check this forced clear because a later forced serial
                        // sibling clears paths installed by earlier methods.
                        //
                        // Order is load-bearing (see `hook::add_path`): emit the serial sibling
                        // FIRST so the forced clear happens before the partial sibling is added. The
                        // partial sibling (OfferParallel, never forced) then only adds a partial path
                        // and leaves the serial in place; reversing the order would let the forced
                        // serial sibling clear the partial path.
                        custom_paths.push(make_path(None, false, true));
                        custom_paths.push(make_path(Some(nworkers), true, false));
                    }
                }
            }

            Some(custom_paths)
        })();
        paths.unwrap_or_default()
    }

    fn plan_custom_path(mut builder: CustomScanBuilder<Self>) -> pg_sys::CustomScan {
        unsafe {
            bitmap_intersection::keep_bitmap_child_plan(&mut builder);

            let mut tlist = PgList::<pg_sys::TargetEntry>::from_pg(builder.args().tlist.as_ptr());

            // Store the length of the target list
            builder
                .custom_private_mut()
                .set_target_list_len(Some(tlist.len()));

            // Extract window_agg(json) calls from processed_tlist using expression tree walker
            // Similar to how uses_scores/uses_snippets work - walk the tree to find our placeholders
            // Note: This updates target_entry_index to match the processed_tlist positions
            let processed_tlist = (*builder.args().root).processed_tlist;

            let mut window_aggregates = deserialize_window_agg_placeholders(processed_tlist);

            if !window_aggregates.is_empty() {
                // Convert PostgresExpression filters to SearchQueryInput now that we have root
                // Note: root was not available in the planner hook, so we needed to delay this until now.
                let private_data = builder.custom_private();
                if let Some(heaprelid) = private_data.heaprelid() {
                    if let Some((_, bm25_index)) = rel_get_bm25_index(heaprelid) {
                        let root = builder.args().root;
                        let rti = private_data
                            .range_table_index()
                            .expect("range table index should be set");

                        resolve_window_aggregate_filters_at_plan_time(
                            &mut window_aggregates,
                            &bm25_index,
                            root,
                            rti,
                        );

                        // Validate that all fields in window aggregates exist in the index schema
                        // and are supported for aggregate pushdown (not NUMERIC)
                        if let Ok(schema) = crate::schema::SearchIndexSchema::open(&bm25_index) {
                            for window_agg in &window_aggregates {
                                for agg_type in window_agg.targetlist.aggregates() {
                                    if let Err(e) = agg_type.validate_fields(&schema) {
                                        pgrx::error!("{}", e);
                                    }
                                }
                            }
                        }
                    }
                }

                builder
                    .custom_private_mut()
                    .set_window_aggregates(window_aggregates);
            }

            let private_data = builder.custom_private();
            let rti = private_data
                .range_table_index()
                .expect("range table index should have been set")
                .try_into()
                .expect("range table index should not be negative");
            let processed_tlist = PgList::<pg_sys::TargetEntry>::from_pg(processed_tlist);

            let mut attname_lookup = HashMap::default();
            let funcoids: Vec<pg_sys::Oid> = score_funcoids()
                .iter()
                .copied()
                .chain(snippet_funcoids().iter().copied())
                .chain(snippets_funcoids().iter().copied())
                .chain(snippet_positions_funcoids().iter().copied())
                .collect();
            for te in processed_tlist.iter_ptr() {
                let func_vars_at_level =
                    pullout_funcexprs(te.cast(), &funcoids, rti, builder.args().root);

                for (funcexpr, var, attname) in func_vars_at_level {
                    // if we have a tlist, then we need to add the specific function that uses
                    // a Var at our level to that tlist.
                    //
                    // if we don't have a tlist (it's empty), then that means Postgres will later
                    // give us everything we need

                    if !tlist.is_empty() {
                        let te = pg_sys::copyObjectImpl(te.cast()).cast::<pg_sys::TargetEntry>();
                        (*te).resno = (tlist.len() + 1) as _;
                        (*te).expr = funcexpr.cast();

                        tlist.push(te);
                    }

                    // track a triplet of (varno, varattno, attname) as 3 individual
                    // entries in the `attname_lookup` List
                    attname_lookup.insert((rti as Varno, (*var).varattno), attname);
                }
            }

            // Extract join-level snippet predicates for this relation
            // Get values we need before the mutable borrow

            // Extract the indexrelid early to avoid borrow checker issues later
            let indexrelid = private_data.indexrelid().expect("indexrelid should be set");
            let indexrel = PgSearchRelation::with_lock(indexrelid, pg_sys::AccessShareLock as _);

            let join_predicates = extract_join_predicates(
                &PlannerContext::from_planner(builder.args().root),
                rti as pg_sys::Index,
                &indexrel,
                true,
            );

            builder
                .custom_private_mut()
                .set_join_predicates(join_predicates);

            builder
                .custom_private_mut()
                .set_var_attname_lookup(attname_lookup);

            builder
                .custom_private_mut()
                .set_ambulkdelete_epoch(MetaPage::open(&indexrel).ambulkdelete_epoch());

            // Collect subplans that our Custom Scan doesn't handle internally and set them as plan.qual.
            // PostgreSQL's ExecInitCustomScan will call ExecInitQual on plan.qual,
            // which properly initializes SubPlans. We then evaluate these in exec_custom_scan.
            let clauses = PgList::<pg_sys::Node>::from_pg(builder.args().clauses);
            let mut subplan_quals = PgList::<pg_sys::Node>::new();
            for clause in clauses.iter_ptr() {
                if is_subplan(clause, builder.args().root) {
                    // strip RestrictInfo wrapper, plan.qual needs bare expressions
                    let bare_clause = if (*clause).type_ == pg_sys::NodeTag::T_RestrictInfo {
                        let ri = clause as *mut pg_sys::RestrictInfo;
                        (*ri).clause.cast()
                    } else {
                        clause
                    };
                    subplan_quals.push(bare_clause);
                }
            }

            // SubPlan quals require per-tuple heap access for ExecQual, which would
            // negate the benefit of columnar. Fall back to Normal so we don't pay for both
            // batch processing AND per-tuple heap fetches.
            if !subplan_quals.is_empty()
                && matches!(
                    builder.custom_private().exec_method_type(),
                    ExecMethodType::Columnar { .. }
                )
            {
                builder
                    .custom_private_mut()
                    .set_exec_method_type(ExecMethodType::Normal);
            }

            // #5727: collect heap-filter and PostgresExpression nodes so we can
            // hand them to `custom_exprs` below. `finalize_plan` walks that field
            // for Param references; without it, `Gather.initParam` omits InitPlan
            // outputs and parallel workers execute with empty param slots.
            let expr_nodes = builder
                .custom_private_mut()
                .query_mut()
                .as_mut()
                .map(|q| q.collect_expression_nodes())
                .unwrap_or_default();

            let mut scan = builder.build();
            if !subplan_quals.is_empty() {
                scan.scan.plan.qual = subplan_quals.into_pg();
            }
            if !expr_nodes.is_empty() {
                let mut expr_list = PgList::<pg_sys::Node>::new();
                for node in expr_nodes {
                    expr_list.push(node);
                }
                scan.custom_exprs = expr_list.into_pg();
            }
            scan
        }
    }

    fn create_custom_scan_state(
        mut builder: CustomScanStateBuilder<Self, Self::PrivateData>,
    ) -> *mut CustomScanStateWrapper<Self> {
        unsafe {
            builder.custom_state().heaprelid = builder
                .custom_private()
                .heaprelid()
                .expect("heaprelid should have a value");
            builder.custom_state().indexrelid = builder
                .custom_private()
                .indexrelid()
                .expect("indexrelid should have a value");

            builder
                .custom_state()
                .open_relations(pg_sys::AccessShareLock as _);

            builder.custom_state().execution_rti =
                (*builder.args().cscan).scan.scanrelid as pg_sys::Index;

            builder.custom_state().exec_method_type =
                builder.custom_private().exec_method_type().clone();

            builder.custom_state().targetlist_len = builder.target_list().len();

            builder.custom_state().segment_count = builder.custom_private().segment_count();
            builder.custom_state().worker_selection_reason =
                builder.custom_private().worker_selection_reason();
            builder.custom_state().var_attname_lookup = builder
                .custom_private()
                .var_attname_lookup()
                .as_ref()
                .cloned()
                .expect("should have an attribute name lookup");

            let score_funcoids = score_funcoids();
            let snippet_funcoids = snippet_funcoids();
            let snippets_funcoids = snippets_funcoids();
            let snippet_positions_funcoids = snippet_positions_funcoids();

            builder.custom_state().score_funcoids = score_funcoids;
            builder.custom_state().snippet_funcoids = snippet_funcoids;
            builder.custom_state().snippets_funcoids = snippets_funcoids;
            builder.custom_state().snippet_positions_funcoids = snippet_positions_funcoids;
            builder.custom_state().need_scores = uses_scores(
                builder.target_list().as_ptr().cast(),
                score_funcoids,
                builder.custom_state().execution_rti,
            );

            // Store join snippet predicates in the scan state
            builder.custom_state().join_predicates =
                builder.custom_private().join_predicates().clone();

            // Store window aggregates in the scan state
            let window_aggs = builder.custom_private().window_aggregates().clone();
            builder.custom_state().window_aggregates = window_aggs;

            // store our query into our custom state too
            let base_query = builder
                .custom_private()
                .query()
                .clone()
                .expect("should have a SearchQueryInput");
            builder
                .custom_state()
                .set_base_search_query_input(base_query);

            if builder.custom_state().need_scores {
                let state = builder.custom_state();
                // Pre-compute enhanced score query if we have join predicates that could affect scoring
                let mut enhanced_score_query = None;
                if let Some(ref join_predicate) = state.join_predicates {
                    // Check the ORIGINAL base query for this relation, not the modified search_query_input
                    // which may contain simplified join predicates from other relations
                    let original_base_query = state.base_search_query_input();

                    // Only enhance scoring if the base query doesn't already have search predicates
                    // If base query has @@@ conditions, it already provides scoring context
                    if !base_query_has_search_predicates(original_base_query, state.indexrelid) {
                        // Combine base query with join predicate using Boolean structure
                        // This provides enhanced search context for scoring while maintaining
                        // the same filtering behavior as the base query
                        enhanced_score_query = Some(SearchQueryInput::Boolean {
                            must: vec![original_base_query.clone()],
                            should: vec![join_predicate.clone()],
                            must_not: vec![],
                            minimum_should_match: None,
                        });
                    }
                }

                // Store enhanced score query for use during search execution
                // This will be None for single-table queries, which is correct
                if let Some(enhanced_score_query) = enhanced_score_query {
                    builder
                        .custom_state()
                        .set_base_search_query_input(enhanced_score_query);
                }
            }

            let node = builder.target_list().as_ptr().cast();
            builder.custom_state().planning_rti = builder
                .custom_private()
                .range_table_index()
                .expect("range table index should have been set");
            builder.custom_state().snippet_generators = uses_snippets(
                builder.custom_state().planning_rti,
                &builder.custom_state().var_attname_lookup,
                node,
                snippet_funcoids,
                snippets_funcoids,
                snippet_positions_funcoids,
            )
            .into_iter()
            .map(|snippet_type| (snippet_type, None))
            .collect();

            builder.custom_state().ambulkdelete_epoch =
                builder.custom_private().ambulkdelete_epoch();

            assign_exec_method(&mut builder);

            let state = builder.build();

            // Tell ExecInitCustomScan to create the scan slot as BufferHeapTuple
            // (matching what begin_custom_scan will use via table_slot_callbacks).
            // This ensures ExecInitQual compiles plan.qual expressions for the
            // correct slot type, avoiding TTS_IS_VIRTUAL assertion failures.
            #[cfg(any(feature = "pg16", feature = "pg17", feature = "pg18"))]
            {
                (*state).csstate.slotOps =
                    pg_sys::table_slot_callbacks((*state).custom_state().heaprel().as_ptr());
            }

            state
        }
    }

    fn explain_custom_scan(
        state: &CustomScanStateWrapper<Self>,
        _ancestors: *mut pg_sys::List,
        explainer: &mut Explainer,
    ) {
        explainer.add_text("Table", state.custom_state().heaprelname());
        explainer.add_text("Index", state.custom_state().indexrelname());
        if let Some(bitmap_exec) = state.custom_state().bitmap_exec.as_ref() {
            explainer.add_text("Bitmap Intersection", bitmap_exec.index_names().join(", "));
            if explainer.is_analyze() {
                if let Some((exact, lossy, recheck, rejected)) = bitmap_exec.cursor_stats() {
                    explainer.add_unsigned_integer("Bitmap Exact Pages", exact, None);
                    explainer.add_unsigned_integer("Bitmap Lossy Pages", lossy, None);
                    explainer.add_unsigned_integer("Bitmap Recheck Pages", recheck, None);
                    explainer.add_unsigned_integer("Bitmap Rejected Docs", rejected, None);
                }
            }
        }
        if explainer.is_costs() {
            explainer.add_unsigned_integer(
                "Segment Count",
                state.custom_state().segment_count as u64,
                None,
            );
        }

        if explainer.is_verbose() {
            if let Some(reason) = state.custom_state().worker_selection_reason {
                explainer.add_text("Worker Selection", reason.label());
            }
        }

        if explainer.is_analyze() {
            explainer.add_unsigned_integer(
                "Heap Fetches",
                state
                    .custom_state()
                    .visibility_checker
                    .as_ref()
                    .map_or(0, |vc| vc.heap_tuple_check_count) as u64,
                None,
            );
            if explainer.is_verbose() {
                explainer.add_unsigned_integer(
                    "Virtual Tuples",
                    state.custom_state().virtual_tuple_count as u64,
                    None,
                );
                explainer.add_unsigned_integer(
                    "Invisible Tuples",
                    state
                        .custom_state()
                        .visibility_checker
                        .as_ref()
                        .map_or(0, |vc| vc.invisible_tuple_count) as u64,
                    None,
                );
                if let Some(explain_data) = state.custom_state().telemetry.parallel_explain() {
                    explainer.add_json("Parallel Workers", &explain_data.workers);
                }
                let segment_info = state.custom_state().segment_info_for_explain();
                if !segment_info.is_empty() {
                    explainer.add_json("Segment Info", &segment_info);
                }
            }
        }

        explainer.add_text(
            "Exec Method",
            state
                .custom_state()
                .exec_method_name()
                .split("::")
                .last()
                .unwrap(),
        );
        exec_methods::fast_fields::explain(state, explainer);

        explainer.add_bool("Scores", state.custom_state().need_scores());
        if let Some(orderby_info) = state.custom_state().orderby_info().as_ref() {
            explainer.add_text(
                "   TopK Order By",
                orderby_info
                    .iter()
                    .map(|oi| format!("{} {}", oi.feature, oi.direction.as_ref()))
                    .collect::<Vec<_>>()
                    .join(", "),
            );
        }

        if let Some(limit) = state.custom_state().limit() {
            explainer.add_unsigned_integer("   TopK Limit", limit as u64, None);
            if explainer.is_analyze() {
                explainer.add_unsigned_integer(
                    "   Queries",
                    state.custom_state().total_query_count().try_into().unwrap(),
                    None,
                );
            }
        }

        // Add a flag to indicate if the query is a full index scan
        let base_query = state.custom_state().base_search_query_input();

        // Only process the query if it's initialized
        // For EXPLAIN without ANALYZE, the query might not be initialized yet
        if !matches!(base_query, SearchQueryInput::Uninitialized) {
            if base_query.is_full_scan_query() {
                explainer.add_bool("Full Index Scan", true);
            }

            // Show query with integrated estimates if GUC is enabled and verbose
            if gucs::explain_recursive_estimates() && explainer.is_verbose() {
                // Get or create a search reader for estimates.
                // - EXPLAIN ANALYZE: search_reader is already initialized by begin_custom_scan
                // - EXPLAIN (without ANALYZE): search_reader is None, so we create a temporary
                //   reader using MvccSatisfies::LargestSegment for estimation purposes only
                let query_tree =
                    if let Some(search_reader) = state.custom_state().search_reader.as_ref() {
                        // EXPLAIN ANALYZE: use the existing search reader
                        search_reader
                            .build_query_tree_with_estimates(base_query.without_heap_filters())
                            .expect("building query tree with estimates should not fail")
                    } else {
                        // EXPLAIN (without ANALYZE): create a temporary reader for estimates
                        let indexrel = state
                            .custom_state()
                            .indexrel
                            .as_ref()
                            .expect("indexrel should be open");

                        let temp_reader = SearchIndexReader::open_with_context(
                            indexrel,
                            base_query.without_heap_filters(),
                            false,                         // don't need scores for estimates
                            MvccSatisfies::LargestSegment, // Use largest segment for estimation
                            None,                          // No expr_context needed for estimates
                            None,                          // No planstate needed for estimates
                            base_query.needs_tokenizer(),
                        )
                        .expect("opening temporary search reader for estimates should not fail");

                        temp_reader
                            .build_query_tree_with_estimates(base_query.without_heap_filters())
                            .expect("building query tree with estimates should not fail")
                    };

                explainer.add_query_with_estimates(&query_tree);
            } else {
                // Regular display without estimates
                explainer.add_query(base_query);
            }
        } else {
            explainer.add_text("Tantivy Query", "(query not yet initialized)");
        }
    }

    fn begin_custom_scan(
        state: &mut CustomScanStateWrapper<Self>,
        estate: *mut pg_sys::EState,
        eflags: i32,
    ) {
        unsafe {
            // open the heap and index relations with the proper locks
            let rte = pg_sys::exec_rt_fetch(state.custom_state().execution_rti, estate);
            assert!(!rte.is_null());
            let lockmode = (*rte).rellockmode as pg_sys::LOCKMODE;

            state.custom_state_mut().open_relations(lockmode);

            // Initialize the harvested child bitmap scan, if any; registering it in
            // custom_ps lets EXPLAIN render it.
            let cscan = state.csstate.ss.ps.plan.cast::<pg_sys::CustomScan>();
            if let Some(bitmap_exec) = bitmap_intersection::BitmapExec::init(cscan, estate, eflags)
            {
                let mut custom_ps = PgList::<pg_sys::PlanState>::from_pg(state.csstate.custom_ps);
                custom_ps.push(bitmap_exec.planstate());
                state.csstate.custom_ps = custom_ps.into_pg();
                state.custom_state_mut().bitmap_exec = Some(bitmap_exec);
            }

            // For EXPLAIN ANALYZE queries, we need to continue with full initialization
            // For EXPLAIN-only (without ANALYZE), begin_custom_scan is not called at all
            if eflags & (pg_sys::EXEC_FLAG_EXPLAIN_ONLY as i32) != 0 {
                // don't do anything else if we're only explaining the query
                return;
            }

            // setup the structures we need to do mvcc checking and heap fetching
            state.custom_state_mut().visibility_checker =
                Some(VisibilityChecker::with_rel_and_snap(
                    state.custom_state().heaprel(),
                    pg_sys::GetActiveSnapshot(),
                ));
            state.custom_state_mut().doc_from_heap_state =
                Some(HeapFetchState::new(state.custom_state().heaprel()));

            // and finally, get the custom scan itself properly initialized
            let tupdesc = state.custom_state().heaptupdesc();
            let planstate = state.planstate();

            pg_sys::ExecInitScanTupleSlot(
                estate,
                addr_of_mut!(state.csstate.ss),
                tupdesc,
                pg_sys::table_slot_callbacks(state.custom_state().heaprel().as_ptr()),
            );

            // On PG15, ExecInitCustomScan hardcodes &TTSOpsVirtual for the scan slot
            // (there's no slotOps override mechanism). ExecInitQual then compiles
            // plan.qual expressions assuming virtual slots. Since we just replaced
            // the slot with BufferHeapTuple above, we must re-initialize the qual
            // so expressions are compiled for the correct slot type.
            #[cfg(feature = "pg15")]
            {
                let plan = state.csstate.ss.ps.plan;
                if !(*plan).qual.is_null() {
                    state.csstate.ss.ps.qual =
                        pg_sys::ExecInitQual((*plan).qual, state.planstate());
                }
            }

            pg_sys::ExecInitResultTypeTL(addr_of_mut!(state.csstate.ss.ps));
            pg_sys::ExecAssignProjectionInfo(
                state.planstate(),
                (*state.csstate.ss.ss_ScanTupleSlot).tts_tupleDescriptor,
            );

            state
                .custom_state_mut()
                .init_expr_context(estate, planstate);
            state.runtime_context = state.csstate.ss.ps.ps_ExprContext;
        }
    }

    fn rescan_custom_scan(state: &mut CustomScanStateWrapper<Self>) {
        // Drop the previous execution's scorers (whose cursors point into the
        // bitmap) and the stale source cell before the bitmap is freed; only then
        // rebuild for the new params. The exec method itself is preserved and
        // re-bound by `reset()` below. No reader means the scan never executed:
        // there are no scorers, and the exec method may not even be bound yet.
        if state.custom_state().search_reader.is_some() {
            state.custom_state_mut().reset_exec_results();
        }
        drop(state.custom_state_mut().search_reader.take());
        state.custom_state_mut().bitmap_cell = None;
        if let Some(bitmap_exec) = state.custom_state_mut().bitmap_exec.as_mut() {
            unsafe { bitmap_exec.rescan() };
        }
        Self::init_search_reader(state);
        state.custom_state_mut().reset();
    }

    #[allow(clippy::blocks_in_conditions)]
    fn exec_custom_scan(state: &mut CustomScanStateWrapper<Self>) -> *mut pg_sys::TupleTableSlot {
        if state.custom_state().search_reader.is_none() {
            Self::init_search_reader(state);
        }

        loop {
            let exec_method = state.custom_state_mut().exec_method_mut();

            // get the next matching document from our search results and look for it in the heap
            match exec_method.next(state.custom_state_mut()) {
                // reached the end of the SearchResults
                ExecState::Eof => {
                    return std::ptr::null_mut();
                }

                // SearchResults found a match
                ExecState::FromHeap {
                    ctid,
                    score,
                    doc_address: _,
                } => {
                    unsafe {
                        let slot = match check_visibility(state, ctid, state.scanslot().cast()) {
                            // the ctid is visible
                            Some(slot) => {
                                exec_method.increment_visible();
                                slot
                            }

                            // the ctid is not visible
                            None => {
                                continue;
                            }
                        };

                        // Evaluate executor-level quals (e.g., RLS policy SubPlan expressions)
                        // that couldn't be pushed into the tantivy query.
                        // These are set as plan.qual in plan_custom_path.
                        if !satisfies_subplan_quals(state, slot) {
                            continue;
                        }

                        let needs_special_projection = state.custom_state().need_scores()
                            || state.custom_state().need_snippets()
                            || state.custom_state().window_aggregate_results.is_some()
                            || state.custom_state().vector_distance_placeholder;

                        if !needs_special_projection {
                            //
                            // we don't need scores, snippets, or window aggregates
                            // do the projection and return
                            //

                            (*(*state.projection_info()).pi_exprContext).ecxt_scantuple = slot;
                            return pg_sys::ExecProject(state.projection_info());
                        } else {
                            //
                            // we do need scores or snippets
                            //
                            // replace their placeholder values and then rebuild the ProjectionInfo
                            // and project it
                            //

                            let mut per_tuple_context = PgMemoryContexts::For(
                                (*(*state.projection_info()).pi_exprContext).ecxt_per_tuple_memory,
                            );
                            per_tuple_context.reset();

                            if state.custom_state().need_scores() {
                                let const_score_node = state
                                    .custom_state()
                                    .const_score_node
                                    .expect("const_score_node should be set");
                                (*const_score_node).constvalue = score.into_datum().unwrap();
                                (*const_score_node).constisnull = false;
                            }

                            // Update window aggregate values
                            if let Some(agg_results) =
                                &state.custom_state().window_aggregate_results
                            {
                                for (te_idx, datum) in agg_results {
                                    if let Some(const_node) =
                                        state.custom_state().const_window_agg_nodes.get(te_idx)
                                    {
                                        (**const_node).constvalue = *datum;
                                        (**const_node).constisnull = false;
                                    }
                                }
                            }

                            // finally, do the projection
                            return per_tuple_context.switch_to(|_| {
                                // TODO: We go _back_ to the heap to get snippet information here
                                // inside of `make_snippet` and `get_snippet_positions`. It's possible
                                // that we could use a wider tuple slot to fetch the extra columns that
                                // we need during our initial lookup above (but then we'd need to copy
                                // into the correctly shaped slot for this scan).
                                let estate = state.csstate.ss.ps.state;
                                maybe_project_snippets(state.custom_state(), ctid, estate);

                                let planstate = state.planstate();

                                (*(*state.projection_info()).pi_exprContext).ecxt_scantuple = slot;
                                let proj_info = pg_sys::ExecBuildProjectionInfo(
                                    state
                                        .custom_state()
                                        .placeholder_targetlist
                                        .expect("placeholder_targetlist must be set"),
                                    (*planstate).ps_ExprContext,
                                    (*planstate).ps_ResultTupleSlot,
                                    planstate,
                                    (*state.csstate.ss.ss_ScanTupleSlot).tts_tupleDescriptor,
                                );
                                pg_sys::ExecProject(proj_info)
                            });
                        }
                    }
                }

                ExecState::Virtual { slot } => {
                    state.custom_state_mut().virtual_tuple_count += 1;
                    return slot;
                }
            }
        }
    }

    fn shutdown_custom_scan(state: &mut CustomScanStateWrapper<Self>) {
        // Leader-only: last chance to read DSM before Postgres destroys it.
        let scan_state = state.custom_state_mut();
        if let Some(parallel) = scan_state.parallel {
            if parallel.is_leader() {
                parallel.finalize_explain(&mut scan_state.telemetry);
            }
        };
    }

    fn end_custom_scan(state: &mut CustomScanStateWrapper<Self>) {
        // Workers: DSM is still alive; publish local telemetry once.
        // Leader: do not touch DSM — Shutdown already ran (or serial path).
        {
            let scan_state = state.custom_state_mut();
            if let Some(parallel) = scan_state.parallel {
                if !parallel.is_leader() {
                    parallel.publish_telemetry(&scan_state.telemetry);
                }
            }
        }

        // get some things dropped now. Order matters: scorers hold bitmap
        // cursors into the TIDBitmap/DSA, so everything that can hold a scorer
        // drops before the bitmap machinery is torn down.
        state.custom_state_mut().drop_exec_method();
        drop(state.custom_state_mut().visibility_checker.take());
        drop(state.custom_state_mut().doc_from_heap_state.take());
        drop(state.custom_state_mut().search_reader.take());
        drop(std::mem::take(
            &mut state.custom_state_mut().snippet_generators,
        ));
        state.custom_state_mut().bitmap_cell = None;
        if let Some(bitmap_exec) = state.custom_state_mut().bitmap_exec.take() {
            unsafe { bitmap_exec.shutdown() };
        }

        state.custom_state_mut().heaprel.take();
        state.custom_state_mut().indexrel.take();
    }
}

/// Returns true if LIMIT 1 was injected by PostgreSQL's MIN/MAX optimization.
///
/// PostgreSQL rewrites `SELECT MIN(col) FROM t` into a subquery like:
/// `SELECT col AS agg_target FROM t ORDER BY col LIMIT 1`
///
/// This detection is intentionally strict to avoid suppressing warnings for user-written LIMITs.
unsafe fn is_minmax_implicit_limit(root: *mut pg_sys::PlannerInfo) -> bool {
    let Some(root) = root.as_ref() else {
        return false;
    };
    let Some(parse) = root.parse.as_ref() else {
        return false;
    };

    // Must be a subquery (has parent) where parent has aggregates but this subquery doesn't.
    // This matches MIN/MAX rewrite: parent has the aggregate, child is the LIMIT 1 subquery.
    let Some(parent_parse) = root.parent_root.as_ref().and_then(|p| p.parse.as_ref()) else {
        return false;
    };
    if parse.hasAggs || !parent_parse.hasAggs || !parse.limitOffset.is_null() {
        return false;
    }

    // Must have LIMIT 1 as a constant int8 (PostgreSQL's MIN/MAX always uses exactly LIMIT 1).
    let Some(limit_const) = nodecast!(Const, T_Const, parse.limitCount) else {
        return false;
    };
    if (*limit_const).constisnull
        || (*limit_const).consttype != pg_sys::INT8OID
        || i64::from_datum((*limit_const).constvalue, false) != Some(1)
    {
        return false;
    }

    // Must have exactly one target column named "agg_target" - PostgreSQL's internal name
    // for the synthetic column created during MIN/MAX rewrite.
    if parse.targetList.is_null() || parse.sortClause.is_null() {
        return false;
    }
    let target_list = PgList::<pg_sys::TargetEntry>::from_pg(parse.targetList);
    let Some(target_entry) = target_list.get_ptr(0).filter(|_| target_list.len() == 1) else {
        return false;
    };
    if (*target_entry).resname.is_null()
        || CStr::from_ptr((*target_entry).resname).to_bytes() != b"agg_target"
    {
        return false;
    }

    // Sort clause must reference the single target column (ORDER BY for MIN/MAX).
    let sort_list = PgList::<pg_sys::SortGroupClause>::from_pg(parse.sortClause);
    let Some(sort_clause) = sort_list.get_ptr(0).filter(|_| sort_list.len() == 1) else {
        return false;
    };
    (*sort_clause).tleSortGroupRef == (*target_entry).ressortgroupref
}

/// Returns true if the scan may return only `LIMIT + OFFSET` rows even though a WindowAgg sits
/// above it, i.e. the window functions compute the same values over that truncated input.
///
/// Four things all have to hold:
///
///   - The window functions are the *only* reason PG zeroed `limit_tuples`. That one field also
///     stands in for `GROUP BY`, `GROUPING SETS`, `DISTINCT`, aggregates and `HAVING`, each of
///     which collapses rows above the WindowAgg, so the top N of the final result needs more
///     than N rows out of the scan.
///   - Every window function is position-only (`row_number`, `rank`, `dense_rank`). One that reads
///     the whole partition -- `sum(x) OVER ()`, `percent_rank()`, `cume_dist()` -- returns a
///     different value over a truncated input.
///   - No `PARTITION BY`. Partitions draw rows from outside the top N, so truncating first
///     renumbers them.
///   - The window ordering matches the query's `ORDER BY`. Otherwise the top N by the query's
///     ordering are not the first N in window order, and the ranks shift.
unsafe fn window_limit_pushdown_is_safe(parse: *mut pg_sys::Query) -> bool {
    let Some(parse_ref) = parse.as_ref() else {
        return false;
    };
    if parse_ref.windowClause.is_null() || parse_ref.sortClause.is_null() {
        return false;
    }
    if !parse_ref.groupClause.is_null()
        || !parse_ref.groupingSets.is_null()
        || !parse_ref.distinctClause.is_null()
        || !parse_ref.havingQual.is_null()
        || parse_ref.hasAggs
    {
        return false;
    }

    let windows = PgList::<pg_sys::WindowClause>::from_pg(parse_ref.windowClause);
    for wc in windows.iter_ptr() {
        if !(*wc).partitionClause.is_null() {
            return false;
        }
        if !pg_sys::equal(
            (*wc).orderClause.cast::<core::ffi::c_void>(),
            parse_ref.sortClause.cast::<core::ffi::c_void>(),
        ) {
            return false;
        }
    }

    window_funcs_are_position_only(parse)
}

/// Returns true if every window function in `parse` derives solely from a row's position in the
/// window ordering (`row_number`, `rank`, `dense_rank`).
unsafe fn window_funcs_are_position_only(parse: *mut pg_sys::Query) -> bool {
    use pgrx::pg_guard;

    struct Context {
        position_only: bool,
    }

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

        if let Some(wfunc) = nodecast!(WindowFunc, T_WindowFunc, node) {
            if !matches!(
                (*wfunc).winfnoid.to_u32(),
                pg_sys::F_ROW_NUMBER | pg_sys::F_RANK_ | pg_sys::F_DENSE_RANK_
            ) {
                (*ctx).position_only = false;
                return true;
            }
        }

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

    let Some(parse) = parse.as_ref() else {
        return false;
    };
    if parse.targetList.is_null() {
        return false;
    }

    let mut context = Context {
        position_only: true,
    };
    let tlist = PgList::<pg_sys::TargetEntry>::from_pg(parse.targetList);
    for te in tlist.iter_ptr() {
        if (*te).expr.is_null() {
            continue;
        }
        walker(
            (*te).expr.cast::<pg_sys::Node>(),
            addr_of_mut!(context).cast::<core::ffi::c_void>(),
        );
        if !context.position_only {
            return false;
        }
    }
    context.position_only
}

///
/// Validates whether a query that should use Top K scan is actually using it.
///
/// When `paradedb.planner_warnings` is not 'off', this function checks if a query with LIMIT
/// that uses ParadeDB's search operators is using the Top K execution method. If Top K was
/// expected but not chosen, it logs a warning with diagnostic information to help developers
/// identify performance issues.
///
/// # Performance Note
/// This function has minimal overhead as it returns early when the GUC is disabled.
fn validate_topk_expectation(
    privdata: &PrivateData,
    topk_pathkey_info: &PathKeyInfo,
    limit_is_explicit: bool,
    chosen_method: &ExecMethodType,
    table_name: &str,
) {
    // Fast path: if validation is disabled, return immediately
    if crate::gucs::planner_warnings() == crate::gucs::PlannerWarnings::Off {
        return;
    }

    // Check if this query should be using Top K
    let has_limit = privdata.limit_offset().is_some();
    let has_search_query = privdata.query().is_some();
    let no_group_by = privdata.window_aggregates().is_empty();

    // Top K is expected when we have: explicit LIMIT + search query + no GROUP BY
    let should_use_topk = has_limit && limit_is_explicit && has_search_query && no_group_by;

    // Check if we actually got Top K
    let is_using_topk = matches!(chosen_method, ExecMethodType::TopK { .. });

    // If Top K is not expected or we're already using Top K, nothing to warn about
    if !should_use_topk || is_using_topk {
        return;
    }

    // At this point: should_use_topk is true AND we're not using Top K - emit warning
    let limit = privdata.limit_offset().as_ref().unwrap().limit.to_string();
    let method_name = match chosen_method {
        ExecMethodType::Normal => "Normal",
        ExecMethodType::Columnar { .. } => "Columnar",
        ExecMethodType::TopK { .. } => "TopK",
    };

    let (reason, remedies) = match topk_pathkey_info {
        PathKeyInfo::Unusable(UnusableReason::TooManyColumns { count, max }) => (
            format!(
                "ORDER BY has {} columns but Top K supports maximum {}",
                count, max
            ),
            format!("Reduce ORDER BY columns to {} or fewer", max),
        ),
        PathKeyInfo::Unusable(UnusableReason::PrefixOnly { matched }) => (
            format!(
                "only partial prefix of ORDER BY can be pushed down ({} columns matched)",
                matched
            ),
            "Ensure all ORDER BY columns are indexed with pdb::literal tokenizer for strings, \
                 or verify that normalizer/collation matches the index"
                .to_string(),
        ),
        PathKeyInfo::Unusable(UnusableReason::NotSortable) => (
            "ORDER BY columns cannot be pushed down to the index".to_string(),
            "Ensure ORDER BY columns are indexed. Numeric columns are fast by default. \
                 For string columns, use pdb.literal tokenizer"
                .to_string(),
        ),
        PathKeyInfo::Unusable(UnusableReason::VectorMetricMismatch {
            field_metric,
            op_metric,
        }) => (
            format!(
                "ORDER BY uses the {} ({:?}) operator but the index attribute was built with \
                 the {} opclass ({:?})",
                op_metric.operator(),
                op_metric,
                field_metric.opclass_name(),
                field_metric,
            ),
            format!(
                "Either change the ORDER BY operator to {} (matching the index opclass), \
                 or rebuild the index with the {} opclass on the vector column.",
                field_metric.operator(),
                op_metric.opclass_name(),
            ),
        ),
        PathKeyInfo::Unusable(UnusableReason::UnsafeCollation) => (
            "ORDER BY columns whose collation is not byte-ordered (C-like) cannot be pushed down to the index"
                .to_string(),
            "Specify COLLATE \"C\" in your query, or use a byte-ordered collation instead".to_string(),
        ),
        PathKeyInfo::Unusable(UnusableReason::UnsupportedScoreExpression) => (
            "ORDER BY expressions containing pdb.score() cannot be evaluated in the index".to_string(),
            "Only standalone pdb.score() or sums of pdb.score() across tables are supported in ORDER BY".to_string(),
        ),
        PathKeyInfo::UsablePrefix(matched) => (
            format!(
                "only partial prefix of ORDER BY can be pushed down ({} of {} columns)",
                matched.len(),
                privdata
                    .maybe_orderby_info()
                    .as_ref()
                    .map_or(0, |o| o.len()),
            ),
            "Ensure all ORDER BY columns are indexed with pdb::literal tokenizer for strings, \
                 or verify that normalizer/collation matches the index"
                .to_string(),
        ),
        PathKeyInfo::None => (
            // This case should normally use Top K with no ordering
            "unknown reason (no pathkeys but should still use Top K)".to_string(),
            "This is unexpected - please report this issue".to_string(),
        ),
        PathKeyInfo::UsableAll(_) => (
            "unknown reason (pathkeys are usable)".to_string(),
            "This is unexpected - please report this issue".to_string(),
        ),
    };

    BaseScan::add_planner_warning(
        format!(
            "Query has LIMIT {} but is not using Top K scan (using {} instead). \
             Reason: {}. \
             This may cause poor performance on large datasets. \
             Remedies: {}. \
             To disable this warning: SET paradedb.planner_warnings = 'off'",
            limit, method_name, reason, remedies
        ),
        table_name,
    );
}

///
/// Choose and return an ExecMethodType based on the properties of the builder at planning time.
///
/// If the query can return "fast fields", make that determination here, falling back to the
/// [`NormalScanExecState`] if not.
///
/// We support `ColumnarExecState` when there are a mix of string and numeric fast fields.
///
/// If we have failed to extract all relevant information at planning time, then the fast-field
/// execution methods might still fall back to `Normal` at execution time: see the notes in
/// `assign_exec_method` and `compute_exec_which_fast_fields`.
///
/// `pdb.score()`, `ctid`, and `tableoid` are considered fast fields for the purposes of
/// these specialized `ExecMethod`s.
///
fn choose_exec_method(
    privdata: &PrivateData,
    topk_pathkey_info: &PathKeyInfo,
    limit_is_explicit: bool,
    table_name: &str,
) -> Vec<ExecMethodType> {
    // See if we can use Top K.
    // A known limit value or a parameterized limit (value resolved at execution time) both qualify.
    if let Some(lo) = privdata.limit_offset().clone() {
        if let Some(orderby_info) = privdata.maybe_orderby_info() {
            let method = ExecMethodType::TopK {
                heaprelid: privdata.heaprelid().expect("heaprelid must be set"),
                limit_offset: lo.clone(),
                orderby_info: Some(orderby_info.clone()),
                window_aggregates: privdata.window_aggregates().clone(),
            };
            validate_topk_expectation(
                privdata,
                topk_pathkey_info,
                limit_is_explicit,
                &method,
                table_name,
            );
            return vec![method];
        }
        if matches!(topk_pathkey_info, PathKeyInfo::None) {
            let method = ExecMethodType::TopK {
                heaprelid: privdata.heaprelid().expect("heaprelid must be set"),
                limit_offset: lo,
                orderby_info: None,
                window_aggregates: privdata.window_aggregates().clone(),
            };
            validate_topk_expectation(
                privdata,
                topk_pathkey_info,
                limit_is_explicit,
                &method,
                table_name,
            );
            return vec![method];
        }
    }

    // Otherwise, see if we can use a fast fields method.
    let is_capable = fast_fields::is_columnar_capable(privdata);
    if is_capable {
        let mut methods = Vec::new();

        let lo = privdata.limit_offset().clone();

        methods.push(ExecMethodType::Columnar {
            which_fast_fields: privdata.planned_which_fast_fields().clone().unwrap(),
            limit_offset: lo,
        });

        // Validate expectations for the first method (Unsorted)
        validate_topk_expectation(
            privdata,
            topk_pathkey_info,
            limit_is_explicit,
            &methods[0],
            table_name,
        );

        return methods;
    }

    // Else, fall back to normal execution
    let method = ExecMethodType::Normal;
    validate_topk_expectation(
        privdata,
        topk_pathkey_info,
        limit_is_explicit,
        &method,
        table_name,
    );
    vec![method]
}

///
/// Creates and assigns the execution method which was chosen at planning time.
///
/// If a fast-fields execution method was chosen at planning time, we might still fall back to
/// NormalScanExecState if we fail to extract the superset of fields during planning time which was
/// needed at execution time.
///
fn assign_exec_method(builder: &mut CustomScanStateBuilder<BaseScan, PrivateData>) {
    match builder.custom_state_ref().exec_method_type.clone() {
        ExecMethodType::Normal => builder
            .custom_state()
            .assign_exec_method(NormalScanExecState::default(), Some(ExecMethodType::Normal)),
        ExecMethodType::TopK {
            heaprelid,
            limit_offset,
            orderby_info,
            window_aggregates: _,
        } => builder.custom_state().assign_exec_method(
            exec_methods::top_k::TopKScanExecState::new(heaprelid, &limit_offset, orderby_info),
            None,
        ),

        ExecMethodType::Columnar {
            which_fast_fields,
            limit_offset: _,
        } => {
            if let Some(which_fast_fields) =
                compute_exec_which_fast_fields(builder, which_fast_fields)
            {
                builder.custom_state().assign_exec_method(
                    exec_methods::fast_fields::columnar::ColumnarExecState::new(which_fast_fields),
                    None,
                )
            } else {
                builder.custom_state().assign_exec_method(
                    NormalScanExecState::default(),
                    Some(ExecMethodType::Normal),
                )
            }
        }
    }
}

///
/// Computes the execution time `which_fast_fields`, which are validated to be a subset of the
/// planning time `which_fast_fields`. If it's not the case, we return `None` to indicate that
/// we should fall back to the `Normal` execution mode.
///
fn compute_exec_which_fast_fields(
    builder: &mut CustomScanStateBuilder<BaseScan, PrivateData>,
    planned_which_fast_fields: HashSet<WhichFastField>,
) -> Option<Vec<WhichFastField>> {
    let target_list = builder.target_list().as_ptr();
    let exec_which_fast_fields = unsafe {
        let custom_state = builder.custom_state();
        let indexrel = custom_state.indexrel();
        let execution_rti = custom_state.execution_rti;
        let heaprel = custom_state.heaprel();
        //
        // In order for our planned ExecMethodType to be accurate, this must always be a
        // subset of the fast fields which were extracted at planning time.
        exec_methods::fast_fields::collect_fast_fields(
            target_list,
            // At this point, all fast fields which we need to extract are listed directly
            // in our execution-time target list, so there is no need to extract from other
            // positions.
            &HashSet::default(),
            execution_rti,
            heaprel,
            indexrel,
            true,
        )
    };

    if fast_fields::is_all_special_or_junk_fields(&exec_which_fast_fields) {
        // In some cases, enough columns are pruned between planning and execution that there
        // is no point actually using fast fields, and we can fall back to `Normal`.
        //
        // TODO: In order to implement https://github.com/paradedb/paradedb/issues/2623, we will
        // need to differentiate these cases, so that we can always emit the sort order that we
        // claimed.
        return None;
    }

    let missing_fast_fields = exec_which_fast_fields
        .iter()
        .filter(|ff| !planned_which_fast_fields.contains(ff))
        .collect::<Vec<_>>();

    if !missing_fast_fields.is_empty() {
        pgrx::log!(
            "Failed to extract all fast fields at planning time: \
             was missing {missing_fast_fields:?} from {planned_which_fast_fields:?} \
             Falling back to Normal execution.",
        );
        return None;
    }

    Some(exec_which_fast_fields)
}

/// Use the [`VisibilityChecker`] to verify the ctid is visible and fetch the tuple into the slot.
/// Returns the slot if visible, None if the tuple is deleted/not visible.
/// See [`VisibilityChecker`] docs for details on two-layer visibility.
#[inline(always)]
fn check_visibility(
    state: &mut CustomScanStateWrapper<BaseScan>,
    ctid: u64,
    bslot: *mut pg_sys::BufferHeapTupleTableSlot,
) -> Option<*mut pg_sys::TupleTableSlot> {
    state
        .custom_state_mut()
        .visibility_checker()
        .exec_if_visible(ctid, bslot.cast(), move |_| bslot.cast())
}

/// Evaluate executor-level quals (e.g., RLS policy SubPlan expressions) that couldn't be pushed
/// into the tantivy query. These are set as `plan.qual` in `plan_custom_path`.
/// Returns `true` if qual passes (or no qual exists), `false` if the row should be skipped.
#[inline(always)]
unsafe fn satisfies_subplan_quals(
    state: &mut CustomScanStateWrapper<BaseScan>,
    slot: *mut pg_sys::TupleTableSlot,
) -> bool {
    let qual = state.csstate.ss.ps.qual;
    if qual.is_null() {
        return true;
    }
    let econtext = (*state.projection_info()).pi_exprContext;
    (*econtext).ecxt_scantuple = slot;
    pg_sys::slot_getallattrs(slot);
    pg_sys::ExecQual(qual, econtext)
}

/// Inject ParadeDB-specific placeholders (score, snippets, window aggregates) into the tuple slot
unsafe fn inject_pdb_placeholders(state: &mut CustomScanStateWrapper<BaseScan>) {
    let need_scores = state.custom_state().need_scores();
    let need_snippets = state.custom_state().need_snippets();
    let has_window_aggs = !state.custom_state().window_aggregates.is_empty();
    // A TopK scan that orders by `embedding <-> query` also needs the special
    // projection: it lets us blank the junk distance column (see
    // `inject_vector_distance_placeholders`) and skip the per-row
    // `l2_distance` + `detoast_attr`. We only consider TopK here because that
    // is the only exec method that provides the vector ordering itself; in any
    // other plan a Sort consumes the distance column and we must leave it be.
    let is_topk = matches!(
        state.custom_state().exec_method_type,
        ExecMethodType::TopK { .. }
    );

    if !need_scores && !need_snippets && !has_window_aggs && !is_topk {
        // nothing to inject, use whatever we originally setup as our ProjectionInfo
        return;
    }

    // inject score and/or snippet placeholder [`pg_sys::Const`] nodes into what is a copy of the Plan's
    // targetlist.  We store this in our custom state's "placeholder_targetlist" for use during the
    // forced projection we must do later.
    let planstate = state.planstate();

    let (targetlist, const_score_node, const_snippet_nodes) = inject_placeholders(
        (*(*planstate).plan).targetlist,
        state.custom_state().planning_rti,
        state.custom_state().score_funcoids,
        state.custom_state().snippet_funcoids,
        state.custom_state().snippets_funcoids,
        state.custom_state().snippet_positions_funcoids,
        &state.custom_state().var_attname_lookup,
        &state.custom_state().snippet_generators,
    );

    // Now inject window aggregate placeholders
    let (targetlist, const_window_agg_nodes) = if !state.custom_state().window_aggregates.is_empty()
    {
        inject_window_aggregate_placeholders(targetlist, &state.custom_state().window_aggregates)
    } else {
        (targetlist, HashMap::default())
    };

    // Blank out the junk vector-distance ORDER BY column. When the ORDER BY is
    // `embedding <-> query`, pg adds that `OpExpr` to the scan's targetlist as
    // a junk column and evaluates it per output row — triggering `detoast_attr`
    // on the TOAST'd heap vector (~27 % of query time at LIMIT 100). The TopK
    // scan already produced the rows in order, so the column's value is unused
    // and junk-stripped; replacing it with a NULL Const skips the recompute.
    // Only safe for TopK (it owns the ordering); other plans Sort by it.
    let vector_distance_placeholder = is_topk && inject_vector_distance_placeholders(targetlist);

    state.custom_state_mut().placeholder_targetlist = Some(targetlist);
    state.custom_state_mut().const_score_node = Some(const_score_node);
    state.custom_state_mut().const_snippet_nodes = const_snippet_nodes;
    state.custom_state_mut().const_window_agg_nodes = const_window_agg_nodes;
    state.custom_state_mut().vector_distance_placeholder = vector_distance_placeholder;
}

/// Replace every *junk* top-level pgvector distance `OpExpr`
/// (e.g. `embedding <-> q`) in `targetlist` with a NULL `Const(float8)`,
/// mutating the `TargetEntry`'s `expr` in place. Returns `true` if at least one
/// entry was replaced.
///
/// A junk distance column only exists to carry the ORDER BY key; with a TopK
/// scan the rows already arrive ordered and the column is junk-stripped from
/// the result, so its value is never observed. Blanking it to NULL lets
/// `ExecProject` skip `l2_distance(embedding, q)` and the heap-vector detoast.
///
/// Two deliberate restrictions:
/// * `resjunk` only — a SELECT-ed `vec <-> q` must be computed exactly; the
///   TopK score is an approximate, squared/normalized ordering key, not the
///   pgvector distance.
/// * the caller invokes this only for TopK scans — any other plan has a Sort
///   that consumes the distance value, so it cannot be blanked.
///
/// This does NOT use `expression_tree_mutator`: that deep-copies every node it
/// visits, which would orphan the score/snippet/window `Const` pointers already
/// collected for this targetlist. The ORDER BY key is always a top-level
/// `TargetEntry` expr, so an in-place per-entry rewrite is sufficient.
unsafe fn inject_vector_distance_placeholders(targetlist: *mut pg_sys::List) -> bool {
    use crate::vector::metric::VectorMetric;

    let mut replaced = false;
    let tlist = PgList::<pg_sys::TargetEntry>::from_pg(targetlist);
    for te in tlist.iter_ptr() {
        if te.is_null() || !(*te).resjunk {
            continue;
        }
        let Some(opexpr) = nodecast!(OpExpr, T_OpExpr, (*te).expr.cast::<pg_sys::Node>()) else {
            continue;
        };
        if VectorMetric::from_opoid((*opexpr).opno).is_some() {
            let const_node = pg_sys::makeConst(
                pg_sys::FLOAT8OID,
                -1,
                pg_sys::Oid::INVALID,
                size_of::<f64>() as _,
                pg_sys::Datum::null(),
                true,
                true,
            );
            (*te).expr = const_node.cast();
            replaced = true;
        }
    }
    replaced
}

/// Inject placeholder Const nodes for window aggregates at execution time
/// At this point, the WindowFunc has been replaced with paradedb.window_agg(json) calls
/// This function finds those calls (which may be wrapped in other functions)
/// and replaces them with placeholder Const nodes that will be filled in during execution.
unsafe fn inject_window_aggregate_placeholders(
    targetlist: *mut pg_sys::List,
    window_aggs: &[WindowAggregateInfo],
) -> (*mut pg_sys::List, HashMap<usize, *mut pg_sys::Const>) {
    let mut const_nodes = HashMap::default();
    let tlist = PgList::<pg_sys::TargetEntry>::from_pg(targetlist);
    let window_agg_procid = window_agg_oid();

    // If window_agg function doesn't exist yet, return original targetlist
    if window_agg_procid == pg_sys::InvalidOid {
        return (targetlist, const_nodes);
    }

    // Process each window aggregate target entry
    let mut new_tlist = PgList::<pg_sys::TargetEntry>::new();

    for (idx, te) in tlist.iter_ptr().enumerate() {
        // Check if this target entry is one of our window aggregates
        let agg_info = window_aggs.iter().find(|a| a.target_entry_index == idx);

        if let Some(agg_info) = agg_info {
            // This target entry should contain a window_agg call (possibly wrapped)
            let (new_expr, const_node_opt) = replace_window_agg_with_const(
                (*te).expr as *mut pg_sys::Node,
                window_agg_procid,
                agg_info.result_type_oid(),
            );

            // Create a new target entry with the modified expression
            let new_te = pg_sys::flatCopyTargetEntry(te);
            (*new_te).expr = new_expr.cast();
            new_tlist.push(new_te);

            if let Some(const_node) = const_node_opt {
                const_nodes.insert(idx, const_node);
            }
        } else {
            // Not a window aggregate - just copy it
            new_tlist.push(te);
        }
    }

    (new_tlist.into_pg(), const_nodes)
}

// Helper function to recursively search and replace window_agg calls
//
// Note: This follows a similar recursive pattern to replace_in_node() in hook.rs,
// but operates at a different stage:
// - That function: Planning stage - replaces WindowFunc → window_agg() placeholder
// - This function: Execution stage - replaces window_agg() → Const placeholder for value injection
//
// TODO: This duplication could potentially be eliminated by moving to UPPERREL_WINDOW handling.
// See https://github.com/paradedb/paradedb/issues/3455
unsafe fn replace_window_agg_with_const(
    node: *mut pg_sys::Node,
    window_agg_procid: pg_sys::Oid,
    result_type_oid: pg_sys::Oid,
) -> (*mut pg_sys::Node, Option<*mut pg_sys::Const>) {
    if node.is_null() {
        return (node, None);
    }

    // Check if this is the window_agg FuncExpr
    if let Some(funcexpr) = nodecast!(FuncExpr, T_FuncExpr, node) {
        if (*funcexpr).funcid == window_agg_procid {
            // Found it! Replace with a Const node
            let const_node = pg_sys::makeConst(
                result_type_oid,
                -1,
                pg_sys::DEFAULT_COLLATION_OID,
                if result_type_oid == pg_sys::INT8OID {
                    8
                } else {
                    -1
                },
                pg_sys::Datum::null(),
                true,                               // constisnull
                result_type_oid == pg_sys::INT8OID, // constbyval (true for INT8)
            );

            return (const_node.cast(), Some(const_node));
        }

        // Not window_agg, but might have window_agg as an argument
        let args = PgList::<pg_sys::Node>::from_pg((*funcexpr).args);
        let mut new_args = PgList::<pg_sys::Node>::new();
        let mut found_const = None;
        let mut modified = false;

        for arg in args.iter_ptr() {
            let (new_arg, const_opt) =
                replace_window_agg_with_const(arg, window_agg_procid, result_type_oid);
            if const_opt.is_some() {
                found_const = const_opt;
                modified = true;
            }
            if new_arg != arg {
                modified = true;
            }
            new_args.push(new_arg);
        }

        if modified {
            // Create a new FuncExpr with modified arguments
            let new_funcexpr = pg_sys::makeFuncExpr(
                (*funcexpr).funcid,
                (*funcexpr).funcresulttype,
                new_args.into_pg(),
                (*funcexpr).funccollid,
                (*funcexpr).inputcollid,
                (*funcexpr).funcformat,
            );
            return (new_funcexpr.cast(), found_const);
        }
    }

    (node, None)
}

/// Determine whether there are any pathkeys at all, and whether we might be able to push down
/// ordering in Top K.
///
/// If between 1 and 3 pathkeys are declared, and are indexed as fast, then return
/// `UsableAll(Vec<OrderByStyles>)` for them for use in Top K.
///
/// This function must be kept in sync with `validate_topk_compatibility` in `hook.rs` to ensure
/// that queries validated during the planner hook phase can be executed by the custom scan.
unsafe fn pullup_topk_pathkeys(
    rti: pg_sys::Index,
    schema: &SearchIndexSchema,
    root: *mut pg_sys::PlannerInfo,
    index_expressions: Option<&PgList<pg_sys::Expr>>,
) -> PathKeyInfo {
    match extract_pathkey_styles_with_sortability_check(
        root,
        rti,
        schema,
        |search_field| search_field.is_raw_sortable(),
        |search_field| search_field.is_lower_sortable(),
        index_expressions,
    ) {
        PathKeyInfo::UsableAll(styles) if styles.len() <= MAX_TOPK_FEATURES => {
            // Top K is the base scan's only executor which supports sorting, and supports up to
            // MAX_TOPK_FEATURES order-by clauses.
            PathKeyInfo::UsableAll(styles)
        }
        PathKeyInfo::UsableAll(ref styles) => {
            // Too many pathkeys were extracted.
            PathKeyInfo::Unusable(UnusableReason::TooManyColumns {
                count: styles.len(),
                max: MAX_TOPK_FEATURES,
            })
        }
        PathKeyInfo::UsablePrefix(ref prefix) => {
            // Top K cannot execute for a prefix of pathkeys, because it eliminates results before
            // the suffix of the pathkey comes into play.
            PathKeyInfo::Unusable(UnusableReason::PrefixOnly {
                matched: prefix.len(),
            })
        }
        pki @ (PathKeyInfo::None | PathKeyInfo::Unusable(_)) => pki,
    }
}

/// Gather all columns referenced by the specified RTE (Range Table Entry) throughout the query.
/// This gives us a more complete picture than just looking at the target list.
///
/// This function is critical for issue #2505/#2556 where we need to detect all columns used in JOIN
/// conditions to ensure we select the right execution method. Previously, only looking at the
/// target list would miss columns referenced in JOIN conditions, leading to execution-time errors.
///
unsafe fn collect_maybe_fast_field_referenced_columns(
    rte_index: pg_sys::Index,
    rel: *mut pg_sys::RelOptInfo,
) -> HashSet<pg_sys::AttrNumber> {
    let mut referenced_columns = HashSet::default();

    // Check reltarget exprs.
    let reltarget_exprs = PgList::<pg_sys::Expr>::from_pg((*(*rel).reltarget).exprs);
    for rte in reltarget_exprs.iter_ptr() {
        if let Some(var) = nodecast!(Var, T_Var, rte) {
            if (*var).varno as u32 == rte_index {
                referenced_columns.insert((*var).varattno);
            }
        }
        // NOTE: Unless we encounter the fallback in `compute_exec_which_fast_fields`, then we
        // can be reasonably confident that directly inspecting Vars is sufficient. We haven't
        // seen it yet in the wild.
    }

    referenced_columns
}

#[rustfmt::skip]
/// Check if the base query has search predicates for the current table's index
fn base_query_has_search_predicates(
    query: &SearchQueryInput,
    current_index_oid: pg_sys::Oid,
) -> bool {
    match query {
        SearchQueryInput::All => false,
        SearchQueryInput::Uninitialized => false,
        SearchQueryInput::Empty => false,

        SearchQueryInput::WithIndex { oid, query } => {
            // Only consider search predicates for the current table's index
            if *oid == current_index_oid {
                // This is a search predicate for our index
                // Check the inner query directly for range vs search predicates
                base_query_has_search_predicates(query, current_index_oid)
            } else {
                // This is a search predicate for a different index, ignore it
                false
            }
        }

        // Boolean queries need recursive checking
        SearchQueryInput::Boolean {
            must,
            should,
            must_not,
            ..
        } => {
            must.iter()
                .any(|q| base_query_has_search_predicates(q, current_index_oid))
                || should
                    .iter()
                    .any(|q| base_query_has_search_predicates(q, current_index_oid))
                || must_not
                    .iter()
                    .any(|q| base_query_has_search_predicates(q, current_index_oid))
        }

        // Wrapper queries need recursive checking
        SearchQueryInput::Boost { query, .. } => {
            base_query_has_search_predicates(query, current_index_oid)
        }
        SearchQueryInput::ConstScore { query, .. } => {
            base_query_has_search_predicates(query, current_index_oid)
        }
        SearchQueryInput::ScoreFilter {
            query: Some(query), ..
        } => base_query_has_search_predicates(query, current_index_oid),
        SearchQueryInput::ScoreFilter { query: None, .. } => false,
        SearchQueryInput::DisjunctionMax { disjuncts, .. } => disjuncts
            .iter()
            .any(|q| base_query_has_search_predicates(q, current_index_oid)),

        // Despite being part of FieldedQuery, these do not use a field, as far as the user knows
        SearchQueryInput::FieldedQuery {  query: pdb::Query::All, ..} |
        SearchQueryInput::FieldedQuery {  query: pdb::Query::Empty, ..} => false,

        // These are NOT search predicates (they're range/exists/other predicates)
        SearchQueryInput::FieldedQuery { query: pdb::Query::Range { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::RangeContains { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::RangeIntersects { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::RangeTerm { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::RangeWithin { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::Exists, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::FastFieldRangeWeight { .. }, .. }
        | SearchQueryInput::MoreLikeThis { .. } => false,

        // These are search predicates that use the @@@ operator
        SearchQueryInput::FieldedQuery { query: pdb::Query::ParseWithField { query_string, .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::Parse { query_string, .. }, .. } => {
            // For ParseWithField, check if it's a text search or a range query
            !is_range_query_string(query_string)
        }
        SearchQueryInput::Parse { .. }
        | SearchQueryInput::TermSet { .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::UnclassifiedString { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::UnclassifiedArray { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::ScoreAdjusted { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::TermSet { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::Term { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::Phrase { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::PhraseArray { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::Proximity { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::TokenizedPhrase { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::PhrasePrefix { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::FuzzyTerm { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::Match { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::MatchArray { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::Regex { .. }, .. }
        | SearchQueryInput::FieldedQuery { query: pdb::Query::RegexPhrase { .. }, .. } => true,

        // // Term with no field is not a search predicate
        // NB:  We don't support unqualified term queries anymore
        // SearchQueryInput::Term { field: None, .. } => false,

        // Postgres expressions are unknown, assume they could be search predicates
        SearchQueryInput::PostgresExpression { .. } => true,

        // HeapFilter contains search predicates
        SearchQueryInput::HeapFilter { indexed_query, .. } => {
            base_query_has_search_predicates(indexed_query, current_index_oid)
        }
    }
}

/// Check if a query string represents a range query (contains operators like >, <, etc.)
fn is_range_query_string(query_string: &str) -> bool {
    // Range queries typically start with operators
    query_string.trim_start().starts_with('>')
        || query_string.trim_start().starts_with('<')
        || query_string.trim_start().starts_with(">=")
        || query_string.trim_start().starts_with("<=")
        || query_string.contains("..")  // Range syntax like "1..10"
        || query_string.contains(" TO ") // Range syntax like "1 TO 10"
}

/// Project configured snippets (if any).
///
/// Must be called inside the per-tuple `MemoryContext`.
unsafe fn maybe_project_snippets(state: &BaseScanState, ctid: u64, estate: *mut pg_sys::EState) {
    if !state.need_snippets() {
        return;
    }

    for (snippet_type, const_snippet_nodes) in &state.const_snippet_nodes {
        match snippet_type {
            SnippetType::SingleText(_, config, _) => {
                // Resolve start/end tags once per snippet type; for Static
                // values this avoids cloning the String per tuple.
                let start_tag = config.resolve_start_tag(estate);
                let end_tag = config.resolve_end_tag(estate);
                let snippet = state.make_snippet(ctid, snippet_type, &start_tag, &end_tag);

                for const_ in const_snippet_nodes {
                    match &snippet {
                        Some(text) => {
                            (**const_).constvalue = text.into_datum().unwrap();
                            (**const_).constisnull = false;
                        }
                        None => {
                            (**const_).constvalue = pg_sys::Datum::null();
                            (**const_).constisnull = true;
                        }
                    }
                }
            }
            SnippetType::MultipleText(_, config, _, _) => {
                let start_tag = config.resolve_start_tag(estate);
                let end_tag = config.resolve_end_tag(estate);
                let snippets = state.make_snippets(ctid, snippet_type, &start_tag, &end_tag);

                for const_ in const_snippet_nodes {
                    match &snippets {
                        Some(array) => {
                            (**const_).constvalue = array.clone().into_datum().unwrap();
                            (**const_).constisnull = false;
                        }
                        None => {
                            (**const_).constvalue = pg_sys::Datum::null();
                            (**const_).constisnull = true;
                        }
                    }
                }
            }
            SnippetType::Positions(..) => {
                let positions = state.get_snippet_positions(ctid, snippet_type);

                for const_ in const_snippet_nodes {
                    match &positions {
                        Some(positions) => {
                            (**const_).constvalue = positions.clone().into_datum().unwrap();
                            (**const_).constisnull = false;
                        }
                        None => {
                            (**const_).constvalue = pg_sys::Datum::null();
                            (**const_).constisnull = true;
                        }
                    }
                }
            }
        }
    }
}

/// Check if the query contains a LEFT JOIN LATERAL pattern
///
/// This function verifies that the query has the specific structure:
/// `... LEFT JOIN LATERAL (...) ...`
///
/// We verify:
/// 1. The parse tree contains a LEFT JOIN node
/// 2. That LEFT JOIN's right side is marked as LATERAL in the range table
///
/// This enables Top K optimization because LEFT JOIN semantics guarantee all
/// left-side rows are preserved. If WHERE/ORDER BY/LIMIT only reference the
/// left table, we can safely apply Top K to the left scan before the join.
unsafe fn is_left_join_lateral(
    root: *mut pg_sys::PlannerInfo,
    rel: *mut pg_sys::RelOptInfo,
) -> bool {
    // Check if this is a join query
    if !(*root).hasJoinRTEs {
        return false;
    }

    // Check if this is a base relation (not a join itself)
    if (*rel).reloptkind != pg_sys::RelOptKind::RELOPT_BASEREL {
        return false;
    }

    // Check the parse tree for LEFT JOIN patterns with LATERAL
    // We need to verify:
    // 1. There's a LEFT JOIN in the query
    // 2. The right side of that LEFT JOIN is LATERAL
    //
    // We use a combination of checks:
    // - Parse tree: to find LEFT JOIN structure
    // - RTE lateral flag: to confirm the right side is LATERAL

    // First, quickly check if any LATERAL references exist at all
    let simple_rel_array = (*root).simple_rel_array;
    if simple_rel_array.is_null() {
        return false;
    }

    let mut has_lateral = false;
    let simple_rel_array_size = (*root).simple_rel_array_size;
    for i in 1..simple_rel_array_size {
        let other_rel = *simple_rel_array.add(i as usize);
        if !other_rel.is_null() && !(*other_rel).lateral_relids.is_null() {
            has_lateral = true;
            break;
        }
    }

    if !has_lateral {
        return false;
    }

    // Now check the parse tree for LEFT JOIN structure
    let jointree = (*(*root).parse).jointree;
    if jointree.is_null() || (*jointree).fromlist.is_null() {
        return false;
    }

    let fromlist = PgList::<pg_sys::Node>::from_pg((*jointree).fromlist);
    for node in fromlist.iter_ptr() {
        if has_left_join_lateral_pattern(node, (*root).parse) {
            return true;
        }
    }

    false
}

/// Recursively check if a node contains a LEFT JOIN LATERAL pattern
unsafe fn has_left_join_lateral_pattern(
    node: *mut pg_sys::Node,
    query: *mut pg_sys::Query,
) -> bool {
    if node.is_null() {
        return false;
    }

    if let Some(join_expr) = nodecast!(JoinExpr, T_JoinExpr, node) {
        // Check if this is a LEFT JOIN
        if (*join_expr).jointype == pg_sys::JoinType::JOIN_LEFT {
            // Check if the right side has LATERAL
            if is_lateral_subquery((*join_expr).rarg, query) {
                return true;
            }
        }

        // Recursively check nested joins
        if has_left_join_lateral_pattern((*join_expr).larg, query) {
            return true;
        }
        if has_left_join_lateral_pattern((*join_expr).rarg, query) {
            return true;
        }
    }

    false
}

/// Check if a node represents a LATERAL subquery
unsafe fn is_lateral_subquery(node: *mut pg_sys::Node, query: *mut pg_sys::Query) -> bool {
    if node.is_null() || query.is_null() {
        return false;
    }

    // Check if it's a RangeTblRef pointing to a LATERAL RTE
    if let Some(rtref) = nodecast!(RangeTblRef, T_RangeTblRef, node) {
        let rtable = (*query).rtable;
        if !rtable.is_null() {
            let rte = pg_sys::rt_fetch((*rtref).rtindex as pg_sys::Index, rtable);
            if !rte.is_null() && (*rte).lateral {
                return true;
            }
        }
    }

    // For nested joins, recursively check
    if let Some(join_expr) = nodecast!(JoinExpr, T_JoinExpr, node) {
        if is_lateral_subquery((*join_expr).larg, query) {
            return true;
        }
        if is_lateral_subquery((*join_expr).rarg, query) {
            return true;
        }
    }

    false
}

/// Verify WHERE clause only references the left table (current relation)
///
/// This method is used to check whether we can safely push down a LEFT LATERAL JOIN as Top K.
/// Because Top K eliminates rows _before_ the JOIN is actually executed, the WHERE clause (and
/// join condition) may only reference the left hand side of the join to avoid eliminating rows via the
/// limit which would be filtered by conditions on the right hand side.
unsafe fn where_clause_only_references_left(
    root: *mut pg_sys::PlannerInfo,
    rti: pg_sys::Index,
) -> bool {
    // Get WHERE clause
    let quals = if !(*root).parse.is_null()
        && !(*(*root).parse).jointree.is_null()
        && !(*(*(*root).parse).jointree).quals.is_null()
    {
        (*(*(*root).parse).jointree).quals
    } else {
        return true; // No WHERE clause means it only references left
    };

    // Walk the quals to check if they only reference our relation
    #[pgrx::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;
        }

        if let Some(var) = nodecast!(Var, T_Var, node) {
            let rti = *(data as *const pg_sys::Index);
            // If we find a Var that's not from our relation, return true (fail)
            if (*var).varno as i32 != rti as i32 && (*var).varno > 0 {
                return true;
            }
        }

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

    // If walker returns true, it found a reference to another relation
    !walker(quals, &rti as *const _ as *mut _)
}
