1//! Planning: find the calls, wrap every path, build the plan node.
2
3use std::ffi::{CString, c_char};
4use std::sync::OnceLock;
5
6use pgrx::datum::Internal;
7use pgrx::pg_sys::panic::ErrorReport;
8use pgrx::{PgLogLevel, PgSqlErrorCode, pg_guard, pg_sys};
9
10use super::ffi::{append, is_a, len, ptrs, walk};
11use super::{Arg, Judge, exec, reach};
12
13/// The three method tables, built once per process by [`register`].
14pub(super) struct Tables {
15    pub path: pg_sys::CustomPathMethods,
16    pub scan: pg_sys::CustomScanMethods,
17    pub exec: pg_sys::CustomExecMethods,
18}
19
20// SAFETY: the tables hold a `&'static CStr`'s pointer and function
21// pointers, and are never written after `register` publishes them.
22unsafe impl Send for Tables {}
23unsafe impl Sync for Tables {}
24
25static TABLES: OnceLock<Tables> = OnceLock::new();
26static PREVIOUS: OnceLock<pg_sys::set_rel_pathlist_hook_type> = OnceLock::new();
27static PREVIOUS_UPPER: OnceLock<pg_sys::create_upper_paths_hook_type> = OnceLock::new();
28
29pub(super) fn tables() -> &'static Tables {
30    TABLES.get().expect("scan::register ran in _PG_init")
31}
32
33/// Installs the planner hooks and the executor check for `J`. Call once, from `_PG_init`; it
34/// creates nothing, so it is safe in the postmaster.
35pub fn register<J: Judge>() {
36    let built = Tables {
37        path: pg_sys::CustomPathMethods {
38            CustomName: J::NAME.as_ptr(),
39            PlanCustomPath: Some(plan_custom_path::<J>),
40            ReparameterizeCustomPathByChild: None,
41        },
42        scan: pg_sys::CustomScanMethods {
43            CustomName: J::NAME.as_ptr(),
44            CreateCustomScanState: Some(exec::create_state::<J>),
45        },
46        exec: exec::methods::<J>(),
47    };
48    if TABLES.set(built).is_err() {
49        return;
50    }
51    // SAFETY: _PG_init runs single-threaded before any planning in this
52    // process; the previous hook is kept and called first.
53    unsafe {
54        pg_sys::RegisterCustomScanMethods(&tables().scan);
55        let _ = PREVIOUS.set(pg_sys::set_rel_pathlist_hook);
56        pg_sys::set_rel_pathlist_hook = Some(set_rel_pathlist::<J>);
57        let _ = PREVIOUS_UPPER.set(pg_sys::create_upper_paths_hook);
58        pg_sys::create_upper_paths_hook = Some(create_upper_paths::<J>);
59        pg_sys::CacheRegisterSyscacheCallback(
60            pg_sys::SysCacheIdentifier::PROCOID as i32,
61            Some(forget_oids),
62            pg_sys::Datum::from(0usize),
63        );
64        super::guard::install::<J>();
65        super::lift::install::<J>();
66    }
67}
68
69/// Planner weight of one call, in the planner's cost units: a network
70/// round trip dwarfs any local qual, so these sort last and plans that
71/// judge fewer rows win. A weight, not an estimate; EXPLAIN's price is
72/// computed separately.
73const PER_CALL_COST: f64 = 1000.0;
74
75/// The functions' planner support function (`SUPPORT` in their SQL).
76/// Being called at all is half its job: `eval_const_expressions` asks it
77/// to simplify each call, which loads this library, and so installs the
78/// hook, before paths are built for the first query in a backend.
79pub fn support(request: Internal) -> Internal {
80    let not_handled = Internal::from(Some(pg_sys::Datum::from(0usize)));
81    let Some(datum) = request.unwrap() else { return not_handled };
82    let node = datum.cast_mut_ptr::<pg_sys::Node>();
83    // SAFETY: the SQL signature is `(internal) RETURNS internal`, which
84    // only the planner can call, and it passes a SupportRequest* node.
85    unsafe {
86        if is_a(node.cast(), pg_sys::NodeTag::T_SupportRequestCost) {
87            let cost = node.cast::<pg_sys::SupportRequestCost>();
88            (*cost).startup = 0.0;
89            (*cost).per_tuple = PER_CALL_COST;
90            return Internal::from(Some(pg_sys::Datum::from(cost as usize)));
91        }
92    }
93    not_handled
94}
95
96thread_local! {
97    /// `function_oids`' answer, `Some(None)` meaning "not installed".
98    /// Dropped whenever any pg_proc row changes (CREATE, DROP or ALTER
99    /// EXTENSION all change them), so it is never stale.
100    static OIDS: std::cell::RefCell<Option<Option<Vec<pg_sys::Oid>>>> = const { std::cell::RefCell::new(None) };
101}
102
103#[pg_guard]
104unsafe extern "C-unwind" fn forget_oids(_arg: pg_sys::Datum, _cache: i32, _hash: u32) {
105    OIDS.with(|o| *o.borrow_mut() = None);
106}
107
108/// The OIDs of `J::FUNCTIONS`, in order, or `None` if the extension is
109/// not installed in this database. The planner hooks and the executor
110/// check ask on every query, so the answer is cached per backend.
111///
112/// # Safety
113/// Called inside a transaction.
114pub(super) unsafe fn function_oids<J: Judge>() -> Option<Vec<pg_sys::Oid>> {
115    if let Some(known) = OIDS.with(|o| o.borrow().clone()) {
116        return known;
117    }
118    let found = unsafe { lookup_oids::<J>() };
119    OIDS.with(|o| *o.borrow_mut() = Some(found.clone()));
120    found
121}
122
123unsafe fn lookup_oids<J: Judge>() -> Option<Vec<pg_sys::Oid>> {
124    unsafe {
125        let extension = pg_sys::get_extension_oid(J::EXTENSION.as_ptr(), true);
126        if extension == pg_sys::InvalidOid {
127            return None;
128        }
129        let schema = pg_sys::get_namespace_name(pg_sys::get_extension_schema(extension));
130        let oids = J::FUNCTIONS
131            .iter()
132            .map(|f| {
133                let name = CString::new(f.name).expect("function names have no NUL");
134                let names = append(
135                    std::ptr::null_mut(),
136                    [pg_sys::makeString(schema), pg_sys::makeString(pg_sys::pstrdup(name.as_ptr()))],
137                );
138                let types: Vec<pg_sys::Oid> = f
139                    .args
140                    .iter()
141                    .map(|a| match a {
142                        Arg::Row => pg_sys::ANYELEMENTOID,
143                        Arg::Text => pg_sys::TEXTOID,
144                        Arg::Float8 => pg_sys::FLOAT8OID,
145                        Arg::TextArray => pg_sys::TEXTARRAYOID,
146                    })
147                    .collect();
148                pg_sys::LookupFuncName(names, types.len() as i32, types.as_ptr(), true)
149            })
150            .collect();
151        Some(oids)
152    }
153}
154
155/// The outermost calls to `oids` in `node`; calls nested in a claimed
156/// call's arguments are evaluated as its arguments.
157unsafe fn calls_in(node: *mut pg_sys::Node, oids: &[pg_sys::Oid], found: &mut Vec<*mut pg_sys::FuncExpr>) {
158    unsafe {
159        walk(node, &mut |n| {
160            if is_a(n.cast(), pg_sys::NodeTag::T_FuncExpr) {
161                let call = n.cast::<pg_sys::FuncExpr>();
162                if oids.contains(&(*call).funcid) {
163                    found.push(call);
164                    return false;
165                }
166            }
167            true
168        })
169    };
170}
171
172pub(super) unsafe fn contains_call(node: *mut pg_sys::Node, oids: &[pg_sys::Oid]) -> bool {
173    let mut found = Vec::new();
174    unsafe { calls_in(node, oids, &mut found) };
175    !found.is_empty()
176}
177
178#[pg_guard]
179unsafe extern "C-unwind" fn set_rel_pathlist<J: Judge>(
180    root: *mut pg_sys::PlannerInfo,
181    rel: *mut pg_sys::RelOptInfo,
182    rti: pg_sys::Index,
183    rte: *mut pg_sys::RangeTblEntry,
184) {
185    // SAFETY: the planner's arguments, valid for the call.
186    unsafe {
187        if let Some(Some(previous)) = PREVIOUS.get() {
188            previous(root, rel, rti, rte);
189        }
190        wrap_paths::<J>(root, rel, rte);
191    }
192}
193
194/// Replaces every path of `rel` with the scan over it, if `rel` has calls
195/// the scan can evaluate.
196unsafe fn wrap_paths<J: Judge>(root: *mut pg_sys::PlannerInfo, rel: *mut pg_sys::RelOptInfo, rte: *mut pg_sys::RangeTblEntry) {
197    unsafe {
198        // An appendrel parent (a partitioned or inherited table) is not
199        // wrapped: each member is, with the quals and select list
200        // translated to it, and the members' scans share the statement's
201        // dedupe and budget. Under DML and row locks the scan is also
202        // an EvalPlanQual recheck plan, answered without sending
203        // (`exec.rs`, `recheck`).
204        let member = (*rel).reloptkind == pg_sys::RelOptKind::RELOPT_OTHER_MEMBER_REL;
205        if !(member || (*rel).reloptkind == pg_sys::RelOptKind::RELOPT_BASEREL) || (*rte).inh || pg_sys::is_dummy_rel(rel)
206        {
207            return;
208        }
209        let Some(oids) = function_oids::<J>() else { return };
210
211        let mut found = Vec::new();
212        for clause in ptrs::<pg_sys::RestrictInfo>((*rel).baserestrictinfo) {
213            calls_in((*clause).clause.cast(), &oids, &mut found);
214        }
215        // The select list is this relation's to compute only when the
216        // relation is the whole query. A call that ends up evaluated
217        // above the scan anyway (under an aggregate, past a LIMIT) is
218        // simply never referenced, and the scan skips it.
219        // A member's is its topmost parent's, translated to the member's
220        // columns: the same expressions the planner pushes down to it
221        // (`apply_scanjoin_target_to_paths`), so the scan computes them.
222        let whole = if member {
223            !(*rel).top_parent.is_null() && pg_sys::bms_equal((*(*rel).top_parent).relids, (*root).all_query_rels)
224        } else {
225            pg_sys::bms_equal((*rel).relids, (*root).all_query_rels)
226        };
227        if whole {
228            // Not under an aggregate or window function: the grouping or
229            // window node evaluates those, and `lift.rs` gives it a scan.
230            for entry in ptrs::<pg_sys::TargetEntry>((*root).processed_tlist) {
231                walk((*entry).expr.cast(), &mut |n| {
232                    if is_a(n.cast(), pg_sys::NodeTag::T_Aggref) || is_a(n.cast(), pg_sys::NodeTag::T_WindowFunc) {
233                        return false;
234                    }
235                    if is_a(n.cast(), pg_sys::NodeTag::T_FuncExpr) && oids.contains(&(*n.cast::<pg_sys::FuncExpr>()).funcid) {
236                        let call = if member {
237                            pg_sys::adjust_appendrel_attrs_multilevel(root, n, rel, (*rel).top_parent)
238                        } else {
239                            n
240                        };
241                        found.push(call.cast());
242                        return false;
243                    }
244                    true
245                });
246            }
247        }
248        let mut calls: Vec<*mut pg_sys::FuncExpr> = Vec::new();
249        for call in found {
250            let node = call.cast::<pg_sys::Node>();
251            let claimable = !pg_sys::contain_agg_clause(node)
252                && !pg_sys::contain_window_function(node)
253                && !pg_sys::expression_returns_set(node)
254                && pg_sys::bms_is_subset(pg_sys::pull_varnos(root, node), (*rel).relids);
255            if claimable && !calls.iter().any(|&c| pg_sys::equal(c.cast(), call.cast())) {
256                calls.push(call);
257            }
258        }
259        if calls.is_empty() {
260            return;
261        }
262
263        let private = append(std::ptr::null_mut(), calls.iter().copied());
264        // The clauses holding calls run in this node, so it reads their
265        // other columns too (`id` in `id <= 5 OR jev(t, …)`).
266        let mut reads = private;
267        for clause in ptrs::<pg_sys::RestrictInfo>((*rel).baserestrictinfo) {
268            if contains_call((*clause).clause.cast(), &oids) {
269                reads = append(pg_sys::list_copy(reads), [(*clause).clause]);
270            }
271        }
272        let needed = pg_sys::pull_var_clause(reads.cast(), pg_sys::PVC_RECURSE_PLACEHOLDERS as i32);
273        let mut wrapped = std::ptr::null_mut();
274        for child in ptrs::<pg_sys::Path>((*rel).pathlist) {
275            provide(child, needed);
276            wrapped = append(wrapped, [custom_path(rel, child, (*rel).reltarget, private)]);
277        }
278        (*rel).pathlist = wrapped;
279        // Partial paths would be gathered above the scan, with the calls
280        // evaluated per row in the workers.
281        (*rel).partial_pathlist = std::ptr::null_mut();
282    }
283}
284
285#[pg_guard]
286unsafe extern "C-unwind" fn create_upper_paths<J: Judge>(
287    root: *mut pg_sys::PlannerInfo,
288    stage: pg_sys::UpperRelationKind::Type,
289    input_rel: *mut pg_sys::RelOptInfo,
290    output_rel: *mut pg_sys::RelOptInfo,
291    extra: *mut std::ffi::c_void,
292) {
293    // SAFETY: the planner's arguments, valid for the call.
294    unsafe {
295        if let Some(Some(previous)) = PREVIOUS_UPPER.get() {
296            previous(root, stage, input_rel, output_rel, extra);
297        }
298        if stage == pg_sys::UpperRelationKind::UPPERREL_ORDERED {
299            wrap_projections::<J>(root, output_rel);
300        }
301    }
302}
303
304/// With ORDER BY, the planner moves volatile select-list expressions past
305/// the Sort (`make_sort_input_target`), into a projection over the sorted
306/// rows. Each such projection that computes calls becomes the scan over
307/// the Sort: rows are judged in sorted order, and a LIMIT above still
308/// stops the judging.
309unsafe fn wrap_projections<J: Judge>(_root: *mut pg_sys::PlannerInfo, rel: *mut pg_sys::RelOptInfo) {
310    unsafe {
311        let Some(oids) = function_oids::<J>() else { return };
312        for i in 0..len((*rel).pathlist) {
313            let cell = (*(*rel).pathlist).elements.add(i);
314            let path = (*cell).ptr_value.cast::<pg_sys::Path>();
315            if !is_a(path.cast(), pg_sys::NodeTag::T_ProjectionPath) {
316                continue;
317            }
318            let projection = path.cast::<pg_sys::ProjectionPath>();
319            let child = (*projection).subpath;
320            let mut calls: Vec<*mut pg_sys::FuncExpr> = Vec::new();
321            for expr in ptrs::<pg_sys::Node>((*(*path).pathtarget).exprs) {
322                let mut found = Vec::new();
323                calls_in(expr, &oids, &mut found);
324                for call in found {
325                    if !pg_sys::expression_returns_set(call.cast())
326                        && !calls.iter().any(|&c| pg_sys::equal(c.cast(), call.cast()))
327                    {
328                        calls.push(call);
329                    }
330                }
331            }
332            if calls.is_empty() {
333                continue;
334            }
335            // The child cannot be made to compute more (a Sort does not
336            // project), so it must already carry what the calls read.
337            let private = append(std::ptr::null_mut(), calls.iter().copied());
338            let needed = pg_sys::pull_var_clause(
339                private.cast(),
340                (pg_sys::PVC_INCLUDE_AGGREGATES | pg_sys::PVC_INCLUDE_WINDOWFUNCS | pg_sys::PVC_INCLUDE_PLACEHOLDERS) as i32,
341            );
342            if ptrs::<pg_sys::Node>(needed).any(|n| !pg_sys::list_member((*(*child).pathtarget).exprs, n.cast())) {
343                continue;
344            }
345            (*cell).ptr_value = custom_path(rel, child, (*path).pathtarget, private).cast();
346        }
347    }
348}
349
350/// Makes `child` output every Var in `needed`: a call's arguments may use
351/// columns the query needs nowhere else (`WHERE jev(t, …)` needs `t.*`).
352unsafe fn provide(child: *mut pg_sys::Path, needed: *mut pg_sys::List) {
353    unsafe {
354        let target = (*child).pathtarget;
355        let missing: Vec<*mut pg_sys::Expr> =
356            ptrs::<pg_sys::Expr>(needed).filter(|&v| !pg_sys::list_member((*target).exprs, v.cast())).collect();
357        if missing.is_empty() {
358            return;
359        }
360        // Copied: paths share the relation's target.
361        let target = pg_sys::copy_pathtarget(target);
362        for var in missing {
363            pg_sys::add_new_column_to_pathtarget(target, var);
364        }
365        (*child).pathtarget = target;
366    }
367}
368
369/// The scan over `child`, computing `target`.
370unsafe fn custom_path(
371    rel: *mut pg_sys::RelOptInfo,
372    child: *mut pg_sys::Path,
373    target: *mut pg_sys::PathTarget,
374    calls: *mut pg_sys::List,
375) -> *mut pg_sys::CustomPath {
376    unsafe {
377        let path = pg_sys::palloc0(size_of::<pg_sys::CustomPath>()).cast::<pg_sys::CustomPath>();
378        let p = &mut (*path).path;
379        p.type_ = pg_sys::NodeTag::T_CustomPath;
380        p.pathtype = pg_sys::NodeTag::T_CustomScan;
381        p.parent = rel;
382        p.pathtarget = target;
383        p.param_info = (*child).param_info;
384        p.parallel_aware = false;
385        p.parallel_safe = false;
386        p.parallel_workers = 0;
387        // Over a relation, the child's estimate already counts the calls'
388        // selectivity and cost (they are in its restriction clauses).
389        // Rows come out in the child's order.
390        p.rows = (*child).rows;
391        // PG18 counts disabled nodes separately from the cost.
392        #[cfg(not(feature = "pg17"))]
393        {
394            p.disabled_nodes = (*child).disabled_nodes;
395        }
396        p.startup_cost = (*child).startup_cost;
397        p.total_cost = (*child).total_cost;
398        p.pathkeys = (*child).pathkeys;
399        (*path).flags = pg_sys::CUSTOMPATH_SUPPORT_PROJECTION;
400        (*path).custom_paths = append(std::ptr::null_mut(), [child]);
401        (*path).custom_private = calls;
402        (*path).methods = &tables().path;
403        path
404    }
405}
406
407/// Builds the node. `custom_scan_tlist` is the child's columns, then one
408/// column per call; `custom_exprs` holds each call's arguments (a row
409/// argument is written as JSON by `crate::row::RowJson`), then each
410/// call's reach (`reach.rs`), and `custom_private` the number of
411/// child columns and each call's index in `J::FUNCTIONS`. Only Nodes go
412/// in the plan: it is copied and cached.
413#[pg_guard]
414unsafe extern "C-unwind" fn plan_custom_path<J: Judge>(
415    root: *mut pg_sys::PlannerInfo,
416    rel: *mut pg_sys::RelOptInfo,
417    best_path: *mut pg_sys::CustomPath,
418    tlist: *mut pg_sys::List,
419    clauses: *mut pg_sys::List,
420    custom_plans: *mut pg_sys::List,
421) -> *mut pg_sys::Plan {
422    // SAFETY: the planner's arguments, valid for the call; the path is
423    // one `wrap_paths` built.
424    unsafe {
425        let Some(oids) = function_oids::<J>() else {
426            pgrx::error!("{} is no longer installed", J::EXTENSION.to_string_lossy());
427        };
428        let calls: Vec<*mut pg_sys::FuncExpr> = ptrs((*best_path).custom_private).collect();
429        let child = ptrs::<pg_sys::Plan>(custom_plans).next().expect("one child plan");
430        refuse_shipping_servers::<J>(child);
431
432        // The child ran every restriction clause, the calls' included;
433        // those are now this node's, evaluated on the answers.
434        take_calls(child, &oids);
435        let mut quals = std::ptr::null_mut();
436        let (mut judged, mut kept) = (std::ptr::null_mut(), std::ptr::null_mut());
437        for clause in ptrs::<pg_sys::RestrictInfo>(clauses) {
438            if (*clause).pseudoconstant {
439                continue;
440            }
441            if contains_call((*clause).clause.cast(), &oids) {
442                quals = append(quals, [(*clause).clause]);
443                judged = append(judged, [clause]);
444            } else {
445                kept = append(kept, [clause]);
446            }
447        }
448        if !judged.is_null() {
449            (*child).plan_rows = rows_before(root, rel, (*child).plan_rows, judged, kept);
450        }
451
452        let mut scan_tlist = std::ptr::null_mut();
453        let mut resno: i16 = 0;
454        let mut entry = |expr: *mut pg_sys::Expr| {
455            resno += 1;
456            pg_sys::makeTargetEntry(expr, resno, std::ptr::null_mut::<c_char>(), false)
457        };
458        for column in ptrs::<pg_sys::TargetEntry>((*child).targetlist) {
459            scan_tlist = append(scan_tlist, [entry(pg_sys::copyObjectImpl((*column).expr.cast()).cast())]);
460        }
461        let prefix = len(scan_tlist) as i32;
462        let mut private = pg_sys::lappend_int(std::ptr::null_mut(), prefix);
463        let mut exprs = std::ptr::null_mut();
464        for &call in &calls {
465            scan_tlist = append(scan_tlist, [entry(call.cast())]);
466            let index = oids.iter().position(|&o| o == (*call).funcid).expect("a claimed call");
467            private = pg_sys::lappend_int(private, index as i32);
468            for arg in ptrs::<pg_sys::Expr>((*call).args) {
469                exprs = append(exprs, [pg_sys::copyObjectImpl(arg.cast()).cast::<pg_sys::Expr>()]);
470            }
471        }
472        // Each call's reach, set by the post-pass once the final quals and
473        // projection are known (`reach.rs`).
474        exprs = append(exprs, calls.iter().map(|_| reach::always()));
475
476        let scan = pg_sys::palloc0(size_of::<pg_sys::CustomScan>()).cast::<pg_sys::CustomScan>();
477        (*scan).scan.plan.type_ = pg_sys::NodeTag::T_CustomScan;
478        (*scan).scan.plan.targetlist = tlist;
479        (*scan).scan.plan.qual = quals;
480        // No relation of its own: setrefs maps the tlist and quals onto
481        // custom_scan_tlist (INDEX_VAR), matching each call by equal().
482        (*scan).scan.scanrelid = 0;
483        (*scan).flags = (*best_path).flags;
484        (*scan).custom_plans = custom_plans;
485        (*scan).custom_exprs = exprs;
486        (*scan).custom_private = private;
487        (*scan).custom_scan_tlist = scan_tlist;
488        (*scan).methods = &tables().scan;
489        scan.cast()
490    }
491}
492
493/// The rows the child emits once the calls' quals (`judged`) are taken
494/// off it: every row the scan judges. Its estimate counted their
495/// selectivity, and `jev.max_rows`, `jev.max_cost` and EXPLAIN read it.
496/// A table's is recomputed as the planner computes it
497/// (`set_baserel_size_estimates`, `get_parameterized_baserel_size`):
498/// its tuples times the selectivity of the clauses it still runs. Any
499/// other child's (a foreign table's remote estimate, a subquery's) is
500/// divided by the calls' selectivity, which a rounded row count makes
501/// approximate.
502unsafe fn rows_before(
503    root: *mut pg_sys::PlannerInfo,
504    rel: *mut pg_sys::RelOptInfo,
505    rows: f64,
506    judged: *mut pg_sys::List,
507    kept: *mut pg_sys::List,
508) -> f64 {
509    unsafe {
510        let table = matches!(
511            (*rel).reloptkind,
512            pg_sys::RelOptKind::RELOPT_BASEREL | pg_sys::RelOptKind::RELOPT_OTHER_MEMBER_REL
513        ) && (*rel).rtekind == pg_sys::RTEKind::RTE_RELATION
514            && (*rel).fdwroutine.is_null()
515            && (*rel).tuples > 0.0;
516        let before = if table {
517            (*rel).tuples * pg_sys::clauselist_selectivity(root, kept, 0, pg_sys::JoinType::JOIN_INNER, std::ptr::null_mut())
518        } else {
519            let selectivity =
520                pg_sys::clauselist_selectivity(root, judged, 0, pg_sys::JoinType::JOIN_INNER, std::ptr::null_mut());
521            if selectivity > 0.0 { rows / selectivity } else { rows }
522        };
523        pg_sys::clamp_row_est(before).max(rows)
524    }
525}
526
527/// Removes the calls' quals from the child: the scan node under any
528/// gating Result. Volatile, they are never index or remote conditions,
529/// so `qual` (and a Result's `resconstantqual`) is where they are.
530/// Refuses if a call survives anywhere it would run per row.
531unsafe fn take_calls(child: *mut pg_sys::Plan, oids: &[pg_sys::Oid]) {
532    unsafe {
533        let without = |list: *mut pg_sys::List| {
534            let mut kept = std::ptr::null_mut();
535            for expr in ptrs::<pg_sys::Node>(list) {
536                if !contains_call(expr, oids) {
537                    kept = append(kept, [expr]);
538                }
539            }
540            kept
541        };
542        let mut node = child;
543        loop {
544            (*node).qual = without((*node).qual);
545            if is_a(node.cast(), pg_sys::NodeTag::T_Result) {
546                let result = node.cast::<pg_sys::Result>();
547                (*result).resconstantqual = without((*result).resconstantqual.cast()).cast();
548            }
549            if contains_call((*node).targetlist.cast(), oids) {
550                pgrx::error!("a jev call is in the output of the scan's child plan");
551            }
552            match (*node).lefttree {
553                next if !next.is_null() && is_a(node.cast(), pg_sys::NodeTag::T_Result) => node = next,
554                _ => break,
555            }
556        }
557    }
558}
559
560/// Refuses to wrap a `ForeignScan` whose server lists this extension in
561/// its `extensions` option (contract, *Sidecar invariants*). postgres_fdw
562/// ships a listed extension's IMMUTABLE functions to the remote; the
563/// calls are VOLATILE today, so this guards the day a helper is not, and
564/// a remote that judged rows would do so per row, outside any request.
565unsafe fn refuse_shipping_servers<J: Judge>(plan: *mut pg_sys::Plan) {
566    unsafe {
567        if plan.is_null() {
568            return;
569        }
570        if is_a(plan.cast(), pg_sys::NodeTag::T_ForeignScan) {
571            let server = (*plan.cast::<pg_sys::ForeignScan>()).fs_server;
572            if server != pg_sys::InvalidOid && lists_extension::<J>(server) {
573                let name = std::ffi::CStr::from_ptr((*pg_sys::GetForeignServer(server)).servername).to_string_lossy();
574                let extension = J::EXTENSION.to_string_lossy();
575                ErrorReport::new(
576                    PgSqlErrorCode::ERRCODE_INVALID_PARAMETER_VALUE,
577                    format!("foreign server \"{name}\" lists {extension} in its extensions option"),
578                    "jev scan",
579                )
580                .set_detail("Its calls could be shipped to the remote and judged there, one row at a time. Nothing was sent.")
581                .set_hint(format!("Remove {extension} from the server's extensions option."))
582                .report(PgLogLevel::ERROR);
583            }
584        }
585        refuse_shipping_servers::<J>((*plan).lefttree);
586        refuse_shipping_servers::<J>((*plan).righttree);
587        let subplans = if is_a(plan.cast(), pg_sys::NodeTag::T_Append) {
588            (*plan.cast::<pg_sys::Append>()).appendplans
589        } else if is_a(plan.cast(), pg_sys::NodeTag::T_MergeAppend) {
590            (*plan.cast::<pg_sys::MergeAppend>()).mergeplans
591        } else if is_a(plan.cast(), pg_sys::NodeTag::T_CustomScan) {
592            (*plan.cast::<pg_sys::CustomScan>()).custom_plans
593        } else if is_a(plan.cast(), pg_sys::NodeTag::T_SubqueryScan) {
594            append(std::ptr::null_mut(), [(*plan.cast::<pg_sys::SubqueryScan>()).subplan])
595        } else {
596            std::ptr::null_mut()
597        };
598        for sub in ptrs::<pg_sys::Plan>(subplans) {
599            refuse_shipping_servers::<J>(sub);
600        }
601    }
602}
603
604/// Whether `server`'s `extensions` option names `J::EXTENSION`, split as
605/// postgres_fdw splits it (`ExtractExtensionList`: a comma-separated
606/// identifier list, unquoted names folded to lower case).
607unsafe fn lists_extension<J: Judge>(server: pg_sys::Oid) -> bool {
608    unsafe {
609        let options = (*pg_sys::GetForeignServer(server)).options;
610        for option in ptrs::<pg_sys::DefElem>(options) {
611            if std::ffi::CStr::from_ptr((*option).defname) != c"extensions" {
612                continue;
613            }
614            let raw = pg_sys::pstrdup(pg_sys::defGetString(option));
615            let mut names: *mut pg_sys::List = std::ptr::null_mut();
616            if !pg_sys::SplitIdentifierString(raw, b',' as c_char, &mut names) {
617                // postgres_fdw's validator refuses such a value, so it
618                // cannot name anything; neither does it here.
619                continue;
620            }
621            if ptrs::<c_char>(names).any(|n| std::ffi::CStr::from_ptr(n) == J::EXTENSION) {
622                return true;
623            }
624        }
625        false
626    }
627}