diff --git a/Cargo.lock b/Cargo.lock
index 91a2c5de5b..74d693c0ec 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -16493,16 +16493,20 @@ dependencies = [
  "coerce",
  "datum",
  "detoast",
+ "exectuples",
  "fmgr_core",
  "fmgr_seams",
  "format_type",
  "guc",
  "heapam",
+ "indexam",
  "init_small",
  "inval",
  "lsyscache",
  "mcx",
  "miscinit",
+ "namespace_seams",
+ "objectaccess",
  "pg_constraint",
  "ri_triggers_seams",
  "rls_seams",
@@ -16512,12 +16516,15 @@ dependencies = [
  "superuser",
  "syscache_seams",
  "table",
+ "tableam",
  "types_core",
  "types_error",
  "types_fmgr",
  "types_guc",
  "types_nodes",
  "types_rel",
+ "types_scan",
+ "types_slot",
  "types_snapshot",
  "types_trigger",
  "types_trigger_call",
diff --git a/crates/backend/optimizer/path/costsize/src/lib.rs b/crates/backend/optimizer/path/costsize/src/lib.rs
index 5f3098d959..45b6e764ae 100644
--- a/crates/backend/optimizer/path/costsize/src/lib.rs
+++ b/crates/backend/optimizer/path/costsize/src/lib.rs
@@ -3411,8 +3411,8 @@ pub fn final_cost_nestloop(
         ntuples = outer_path_rows * inner_path_rows;
     }
 
-    let quals = types_pathnodes::relids::pgvec_clone_shallow(run.mcx, &path.jpath.joinrestrictinfo);
-    let restrict_qual_cost = crate::cost_qual_eval(run, &quals)?;
+    // `path` is not yet in the arena, so its lists can be passed while `run` is borrowed mutably.
+    let restrict_qual_cost = crate::cost_qual_eval(run, &path.jpath.joinrestrictinfo)?;
     startup_cost += restrict_qual_cost.startup;
     let cpu_per_tuple = gucs::cpu_tuple_cost() + restrict_qual_cost.per_tuple;
     // mul_add mirrors the C referee's fmadd (GCC fp-contract fuses
@@ -3597,8 +3597,7 @@ pub fn final_cost_hashjoin(
     #[allow(unused_assignments)] // Rust-structural: C declares uninitialized (costsize.c:4294); the by-value seam made this initializer dead
     let mut innerbucketsize = 1.0f64;
     let mut innermcvfreq = 1.0f64;
-    let inner_relids = run.root.rel(inner_parent).relids.clone();
-    let hcls = types_pathnodes::relids::pgvec_clone_shallow(run.mcx, &path.path_hashclauses);
+    let hcls: &[RinfoId] = &path.path_hashclauses;
     let otherclauses = if inner_is_unique_path {
         innerbucketsize = 1.0 / virtualbuckets;
         innermcvfreq = 0.0;
@@ -3607,7 +3606,7 @@ pub fn final_cost_hashjoin(
         let (other, bs) = planner_seams::estimate_multivariate_bucketsize::call(
             run,
             inner_parent,
-            &hcls,
+            hcls,
         )?;
         innerbucketsize = bs;
         other
@@ -3615,7 +3614,8 @@ pub fn final_cost_hashjoin(
     for &hcl in otherclauses.iter() {
         let right_is_inner = {
             let r = run.root.rinfo(hcl);
-            types_pathnodes::relids::relids_is_subset(&r.right_relids, &inner_relids)
+            let inner_relids = &run.root.rel(inner_parent).relids;
+            types_pathnodes::relids::relids_is_subset(&r.right_relids, inner_relids)
         };
         let (thisbucketsize, thismcvfreq) = if right_is_inner {
             let cached = run.root.rinfo(hcl).right_bucketsize;
@@ -3662,10 +3662,8 @@ pub fn final_cost_hashjoin(
         startup_cost += DISABLE_COST;
     }
 
-    let hash_qual_cost = crate::cost_qual_eval(run, &hcls)?;
-    let joinrestrict =
-        types_pathnodes::relids::pgvec_clone_shallow(run.mcx, &path.jpath.joinrestrictinfo);
-    let qp_qual_cost = crate::cost_qual_eval(run, &joinrestrict)?;
+    let hash_qual_cost = crate::cost_qual_eval(run, hcls)?;
+    let qp_qual_cost = crate::cost_qual_eval(run, &path.jpath.joinrestrictinfo)?;
     let qp_startup = qp_qual_cost.startup - hash_qual_cost.startup;
     let qp_per_tuple = qp_qual_cost.per_tuple - hash_qual_cost.per_tuple;
 
@@ -4054,12 +4052,8 @@ pub fn final_cost_mergejoin(
         path.jpath.path.rows = clamp_row_est(path.jpath.path.rows / parallel_divisor);
     }
 
-    let mergeclauses =
-        types_pathnodes::relids::pgvec_clone_shallow(run.mcx, &path.path_mergeclauses);
-    let restrictinfos =
-        types_pathnodes::relids::pgvec_clone_shallow(run.mcx, &path.jpath.joinrestrictinfo);
-    let merge_qual_cost = crate::cost_qual_eval(run, &mergeclauses)?;
-    let mut qp_qual_cost = crate::cost_qual_eval(run, &restrictinfos)?;
+    let merge_qual_cost = crate::cost_qual_eval(run, &path.path_mergeclauses)?;
+    let mut qp_qual_cost = crate::cost_qual_eval(run, &path.jpath.joinrestrictinfo)?;
     qp_qual_cost.startup -= merge_qual_cost.startup;
     qp_qual_cost.per_tuple -= merge_qual_cost.per_tuple;
 
@@ -4070,7 +4064,7 @@ pub fn final_cost_mergejoin(
     path.skip_mark_restore =
         early_stop && path.jpath.joinrestrictinfo.len() == path.path_mergeclauses.len();
 
-    let mergejointuples = approx_tuple_count(run, outer_path, inner_path, &mergeclauses)?;
+    let mergejointuples = approx_tuple_count(run, outer_path, inner_path, &path.path_mergeclauses)?;
 
     let outer_is_unique_path = matches!(
         run.root.path(outer_path),
diff --git a/crates/backend/optimizer/plan/planner/src/joinpath.rs b/crates/backend/optimizer/plan/planner/src/joinpath.rs
index f7faf9be59..930817a064 100644
--- a/crates/backend/optimizer/plan/planner/src/joinpath.rs
+++ b/crates/backend/optimizer/plan/planner/src/joinpath.rs
@@ -12,7 +12,7 @@ use types_pathnodes::{
 
 use crate::gucs;
 use crate::pathkeys::{
-    build_join_pathkeys, find_mergeclauses_for_outer_pathkeys,
+    build_join_pathkeys_len, find_mergeclauses_for_outer_pathkeys,
     get_cheapest_path_for_pathkeys, make_inner_pathkeys_for_merge, pathkeys_contained_in,
     pathkeys_count_contained_in, select_outer_pathkeys_for_merge,
     trim_mergeclauses_for_inner_pathkeys, update_mergeclause_eclasses,
@@ -112,41 +112,26 @@ pub fn add_paths_to_joinrel<'mcx>(
     let param_source_rels = {
         use crate::relnode::{relids_difference, relids_overlap, relids_union};
         let mcx = run.mcx;
-        let joinrelids = if run.root.rel(joinrel).reloptkind == types_pathnodes::RELOPT_OTHER_JOINREL
-        {
-            crate::relnode::relids_copy(mcx, &run.root.rel(joinrel).top_parent_relids)
+        let root = &run.root;
+        let joinrelids = if root.rel(joinrel).reloptkind == types_pathnodes::RELOPT_OTHER_JOINREL {
+            &root.rel(joinrel).top_parent_relids
         } else {
-            crate::relnode::relids_copy(mcx, &run.root.rel(joinrel).relids)
+            &root.rel(joinrel).relids
         };
         let mut psr: Relids<'mcx> = crate::relnode::relids_empty();
-        for i in 0..run.root.join_info_list.len() {
-            let (min_l, min_r, jt) = {
-                let sj = &run.root.join_info_list[i];
-                (
-                    crate::relnode::relids_copy(mcx, &sj.min_lefthand),
-                    crate::relnode::relids_copy(mcx, &sj.min_righthand),
-                    sj.jointype,
-                )
-            };
-            if relids_overlap(&joinrelids, &min_r) && !relids_overlap(&joinrelids, &min_l) {
-                psr = relids_union(
-                    mcx,
-                    &psr,
-                    &relids_difference(mcx, &run.root.all_baserels, &min_r),
-                );
+        for sj in root.join_info_list.iter() {
+            let (min_l, min_r) = (&sj.min_lefthand, &sj.min_righthand);
+            if relids_overlap(joinrelids, min_r) && !relids_overlap(joinrelids, min_l) {
+                psr = relids_union(mcx, &psr, &relids_difference(mcx, &root.all_baserels, min_r));
             }
-            if jt == types_pathnodes::JOIN_FULL
-                && relids_overlap(&joinrelids, &min_l)
-                && !relids_overlap(&joinrelids, &min_r)
+            if sj.jointype == types_pathnodes::JOIN_FULL
+                && relids_overlap(joinrelids, min_l)
+                && !relids_overlap(joinrelids, min_r)
             {
-                psr = relids_union(
-                    mcx,
-                    &psr,
-                    &relids_difference(mcx, &run.root.all_baserels, &min_l),
-                );
+                psr = relids_union(mcx, &psr, &relids_difference(mcx, &root.all_baserels, min_l));
             }
         }
-        relids_union(mcx, &psr, &run.root.rel(joinrel).lateral_relids)
+        relids_union(mcx, &psr, &root.rel(joinrel).lateral_relids)
     };
     if mergejoin_allowed {
         sort_inner_and_outer(
@@ -182,16 +167,16 @@ fn compute_semi_anti_join_factors<'mcx>(
     use types_pathnodes::{JOIN_ANTI, JOIN_SEMI};
     let mcx = run.mcx;
     let joinquals: PgVec<'mcx, RinfoId> = if is_outer_join(jointype) {
-        let joinrelids = crate::relnode::relids_copy(mcx, &run.root.rel(joinrel).relids);
-        let mut v = PgVec::new_in(mcx);
+        let joinrelids = &run.root.rel(joinrel).relids;
+        let mut v = PgVec::with_capacity_in(restrictlist.len(), mcx);
         for &rid in restrictlist {
-            if !crate::joinrels::rinfo_is_pushed_down(run, rid, &joinrelids) {
+            if !crate::joinrels::rinfo_is_pushed_down(run, rid, joinrelids) {
                 v.push(rid);
             }
         }
         v
     } else {
-        let mut v = PgVec::new_in(mcx);
+        let mut v = PgVec::with_capacity_in(restrictlist.len(), mcx);
         v.extend(restrictlist.iter().copied());
         v
     };
@@ -235,11 +220,13 @@ fn select_mergejoin_clauses<'mcx>(
     restrictlist: &[RinfoId],
 ) -> PgResult<(PgVec<'mcx, RinfoId>, bool)> {
     let isouterjoin = is_outer_join(jointype);
-    let joinrelids = crate::relnode::relids_copy(run.mcx, &run.root.rel(joinrel).relids);
+    // joinrel relids are re-borrowed per clause: update_mergeclause_eclasses needs `run` mutably.
     let mut have_nonmergeable_joinclause = false;
     let mut result: PgVec<'mcx, RinfoId> = PgVec::new_in(run.mcx);
     for &rid in restrictlist {
-        if isouterjoin && crate::joinrels::rinfo_is_pushed_down(run, rid, &joinrelids) {
+        if isouterjoin
+            && crate::joinrels::rinfo_is_pushed_down(run, rid, &run.root.rel(joinrel).relids)
+        {
             continue;
         }
         {
@@ -348,8 +335,7 @@ fn sort_inner_and_outer<'mcx>(
         let csi = if run.root.path(inner_path).base().parallel_safe {
             Some(inner_path)
         } else if save_jointype != types_pathnodes::JOIN_UNIQUE_INNER {
-            let pl = crate::relnode::pgvec_clone_shallow(run.mcx, &run.root.rel(innerrel).pathlist);
-            get_cheapest_parallel_safe_total_inner(run, &pl)
+            get_cheapest_parallel_safe_total_inner(run, &run.root.rel(innerrel).pathlist)
         } else {
             None
         };
@@ -361,7 +347,8 @@ fn sort_inner_and_outer<'mcx>(
     let all_pathkeys = select_outer_pathkeys_for_merge(run, mergeclause_list, joinrel)?;
 
     for i in 0..all_pathkeys.len() {
-        let mut outerkeys: PgVec<'mcx, PathKey> = PgVec::new_in(run.mcx);
+        let mut outerkeys: PgVec<'mcx, PathKey> =
+            PgVec::with_capacity_in(all_pathkeys.len(), run.mcx);
         if i == 0 {
             outerkeys.extend(all_pathkeys.iter().copied());
         } else {
@@ -378,27 +365,17 @@ fn sort_inner_and_outer<'mcx>(
             find_mergeclauses_for_outer_pathkeys(run, &outerkeys, mergeclause_list)?;
         debug_assert_eq!(cur_mergeclauses.len(), mergeclause_list.len());
         let innerkeys = make_inner_pathkeys_for_merge(run, &cur_mergeclauses, &outerkeys)?;
-        let merge_pathkeys = build_join_pathkeys(run, joinrel, jointype, &outerkeys)?;
-        let want_partial = cheapest_partial_outer.is_some() && cheapest_safe_inner.is_some();
-        let (pk2, mc2, ok2, ik2) = if want_partial {
-            (
-                crate::relnode::pgvec_clone_shallow(run.mcx, &merge_pathkeys),
-                crate::relnode::pgvec_clone_shallow(run.mcx, &cur_mergeclauses),
-                crate::relnode::pgvec_clone_shallow(run.mcx, &outerkeys),
-                crate::relnode::pgvec_clone_shallow(run.mcx, &innerkeys),
-            )
-        } else {
-            (PgVec::new_in(run.mcx), PgVec::new_in(run.mcx), PgVec::new_in(run.mcx), PgVec::new_in(run.mcx))
-        };
+        let n = build_join_pathkeys_len(run, joinrel, jointype, &outerkeys)?;
+        let merge_pathkeys = &outerkeys[..n];
         try_mergejoin_path(
             run,
             joinrel,
             outer_path,
             inner_path,
             merge_pathkeys,
-            cur_mergeclauses,
-            outerkeys,
-            innerkeys,
+            &cur_mergeclauses,
+            &outerkeys,
+            &innerkeys,
             jointype,
             inner_unique,
             sjinfo,
@@ -407,16 +384,16 @@ fn sort_inner_and_outer<'mcx>(
             semifactors,
             false,
         )?;
-        if want_partial {
+        if let (Some(cpo), Some(csi)) = (cheapest_partial_outer, cheapest_safe_inner) {
             try_partial_mergejoin_path(
                 run,
                 joinrel,
-                cheapest_partial_outer.unwrap(),
-                cheapest_safe_inner.unwrap(),
-                pk2,
-                mc2,
-                ok2,
-                ik2,
+                cpo,
+                csi,
+                merge_pathkeys,
+                &cur_mergeclauses,
+                &outerkeys,
+                &innerkeys,
                 jointype,
                 inner_unique,
                 sjinfo,
@@ -427,13 +404,14 @@ fn sort_inner_and_outer<'mcx>(
     Ok(())
 }
 
-// generate_mergejoin_paths (joinpath.c).
+// generate_mergejoin_paths (joinpath.c); outer_pathkeys is the caller's copy of outerpath's pathkeys.
 #[allow(clippy::too_many_arguments)]
 fn generate_mergejoin_paths<'mcx>(
     run: &mut PlannerRun<'mcx>,
     joinrel: RelId,
     innerrel: RelId,
     outerpath: PathId,
+    outer_pathkeys: &[PathKey],
     mut jointype: u32,
     inner_unique: bool,
     sjinfo: &SpecialJoinInfo<'mcx>,
@@ -446,7 +424,6 @@ fn generate_mergejoin_paths<'mcx>(
     semifactors: Option<SemiAntiJoinFactors>,
     is_partial: bool,
 ) -> PgResult<()> {
-    let mcx = run.mcx;
     let save_jointype = jointype;
     if matches!(
         jointype,
@@ -454,10 +431,9 @@ fn generate_mergejoin_paths<'mcx>(
     ) {
         jointype = JOIN_INNER;
     }
-    let outer_pathkeys =
-        crate::relnode::pgvec_clone_shallow(mcx, &run.root.path(outerpath).base().pathkeys);
+    debug_assert!(*outer_pathkeys == *run.root.path(outerpath).base().pathkeys);
     let mergeclauses =
-        find_mergeclauses_for_outer_pathkeys(run, &outer_pathkeys, mergeclause_list)?;
+        find_mergeclauses_for_outer_pathkeys(run, outer_pathkeys, mergeclause_list)?;
     // FULL may try a clauseless merge join (its only legal plan for e.g.
     // FULL JOIN ON FALSE), per C.
     if mergeclauses.is_empty() && jointype != types_pathnodes::JOIN_FULL {
@@ -467,19 +443,17 @@ fn generate_mergejoin_paths<'mcx>(
         return Ok(());
     }
 
-    let innersortkeys = make_inner_pathkeys_for_merge(run, &mergeclauses, &outer_pathkeys)?;
+    let innersortkeys = make_inner_pathkeys_for_merge(run, &mergeclauses, outer_pathkeys)?;
 
-    let mut mpk: PgVec<'mcx, PathKey> = PgVec::new_in(mcx);
-    mpk.extend(merge_pathkeys.iter().copied());
     try_mergejoin_path(
         run,
         joinrel,
         outerpath,
         inner_cheapest_total,
-        mpk,
-        crate::relnode::pgvec_clone_shallow(mcx, &mergeclauses),
-        PgVec::new_in(mcx),
-        crate::relnode::pgvec_clone_shallow(mcx, &innersortkeys),
+        merge_pathkeys,
+        &mergeclauses,
+        &[],
+        &innersortkeys,
         jointype,
         inner_unique,
         sjinfo,
@@ -510,13 +484,13 @@ fn generate_mergejoin_paths<'mcx>(
 
     for sortkeycnt in (1..=num_sortkeys).rev() {
         let trialsortkeys = &innersortkeys[..sortkeycnt];
-        let inner_pathlist =
-            crate::relnode::pgvec_clone_shallow(mcx, &run.root.rel(innerrel).pathlist);
+        // C's newclauses: trimmed at most once per sortkeycnt, shared by both arms.
         let mut newclauses: Option<PgVec<'mcx, RinfoId>> = None;
 
+        // innerrel's pathlist is read in place: only joinrel gains paths in this loop.
         let innerpath = get_cheapest_path_for_pathkeys(
             run,
-            &inner_pathlist,
+            &run.root.rel(innerrel).pathlist,
             trialsortkeys,
             &crate::relnode::RELIDS_UNSET,
             CostSelector::Total,
@@ -534,25 +508,20 @@ fn generate_mergejoin_paths<'mcx>(
                 }
             };
             if cheaper {
-                let clauses = if sortkeycnt < num_sortkeys {
+                if sortkeycnt < num_sortkeys {
                     let t = trim_mergeclauses_for_inner_pathkeys(run, &mergeclauses, trialsortkeys);
                     debug_assert!(!t.is_empty());
-                    t
-                } else {
-                    crate::relnode::pgvec_clone_shallow(mcx, &mergeclauses)
-                };
-                newclauses = Some(crate::relnode::pgvec_clone_shallow(mcx, &clauses));
-                let mut mpk: PgVec<'mcx, PathKey> = PgVec::new_in(mcx);
-                mpk.extend(merge_pathkeys.iter().copied());
+                    newclauses = Some(t);
+                }
                 try_mergejoin_path(
                     run,
                     joinrel,
                     outerpath,
                     ip,
-                    mpk,
-                    clauses,
-                    PgVec::new_in(mcx),
-                    PgVec::new_in(mcx),
+                    merge_pathkeys,
+                    newclauses.as_deref().unwrap_or(&mergeclauses),
+                    &[],
+                    &[],
                     jointype,
                     inner_unique,
                     sjinfo,
@@ -567,7 +536,7 @@ fn generate_mergejoin_paths<'mcx>(
 
         let innerpath = get_cheapest_path_for_pathkeys(
             run,
-            &inner_pathlist,
+            &run.root.rel(innerrel).pathlist,
             trialsortkeys,
             &crate::relnode::RELIDS_UNSET,
             CostSelector::Startup,
@@ -586,33 +555,21 @@ fn generate_mergejoin_paths<'mcx>(
             };
             if cheaper {
                 if Some(ip) != cheapest_total_inner {
-                    let clauses = match newclauses {
-                        Some(ref c) => crate::relnode::pgvec_clone_shallow(mcx, c),
-                        None => {
-                            if sortkeycnt < num_sortkeys {
-                                let t = trim_mergeclauses_for_inner_pathkeys(
-                                    run,
-                                    &mergeclauses,
-                                    trialsortkeys,
-                                );
-                                debug_assert!(!t.is_empty());
-                                t
-                            } else {
-                                crate::relnode::pgvec_clone_shallow(mcx, &mergeclauses)
-                            }
-                        }
-                    };
-                    let mut mpk: PgVec<'mcx, PathKey> = PgVec::new_in(mcx);
-                    mpk.extend(merge_pathkeys.iter().copied());
+                    if newclauses.is_none() && sortkeycnt < num_sortkeys {
+                        let t =
+                            trim_mergeclauses_for_inner_pathkeys(run, &mergeclauses, trialsortkeys);
+                        debug_assert!(!t.is_empty());
+                        newclauses = Some(t);
+                    }
                     try_mergejoin_path(
                         run,
                         joinrel,
                         outerpath,
                         ip,
-                        mpk,
-                        clauses,
-                        PgVec::new_in(mcx),
-                        PgVec::new_in(mcx),
+                        merge_pathkeys,
+                        newclauses.as_deref().unwrap_or(&mergeclauses),
+                        &[],
+                        &[],
                         jointype,
                         inner_unique,
                         sjinfo,
@@ -687,10 +644,11 @@ fn match_unsorted_outer<'mcx>(
         }
     }
 
-    let outer_paths =
-        crate::relnode::pgvec_clone_shallow(run.mcx, &run.root.rel(outerrel).pathlist);
-    for &raw_outerpath in outer_paths.iter() {
-        let mut outerpath = raw_outerpath;
+    // Rel lists indexed in place: only joinrel gains paths (create_unique_path just caches on the rel).
+    let mut outer_pathkeys: PgVec<'mcx, PathKey> = PgVec::new_in(run.mcx);
+    let n_outer = run.root.rel(outerrel).pathlist.len();
+    for oi in 0..n_outer {
+        let mut outerpath = run.root.rel(outerrel).pathlist[oi];
         if path_param_by_rel(run, outerpath, innerrel) {
             continue;
         }
@@ -701,21 +659,18 @@ fn match_unsorted_outer<'mcx>(
             outerpath = crate::pathnode::create_unique_path(run, outerrel, outerpath, sjinfo)?
                 .expect("unique-ify was proven possible");
         }
-        let outer_pathkeys = crate::relnode::pgvec_clone_shallow(
-            run.mcx,
-            &run.root.path(outerpath).base().pathkeys,
-        );
-        let merge_pathkeys = build_join_pathkeys(run, joinrel, jointype, &outer_pathkeys)?;
+        outer_pathkeys.clear();
+        outer_pathkeys.extend_from_slice(&run.root.path(outerpath).base().pathkeys);
+        let n = build_join_pathkeys_len(run, joinrel, jointype, &outer_pathkeys)?;
+        let merge_pathkeys = &outer_pathkeys[..n];
         if save_jointype == JOIN_UNIQUE_INNER {
             let ict = inner_cheapest_total.expect("checked above");
-            try_nestloop_path(run, joinrel, outerpath, ict, &merge_pathkeys, jointype, inner_unique, sjinfo, restrictlist, param_source_rels, semifactors)?;
+            try_nestloop_path(run, joinrel, outerpath, ict, merge_pathkeys, jointype, inner_unique, sjinfo, restrictlist, param_source_rels, semifactors)?;
         } else if nestjoin_ok {
-            let inner_candidates = crate::relnode::pgvec_clone_shallow(
-                run.mcx,
-                &run.root.rel(innerrel).cheapest_parameterized_paths,
-            );
-            for &innerpath in inner_candidates.iter() {
-                try_nestloop_path(run, joinrel, outerpath, innerpath, &merge_pathkeys, jointype, inner_unique, sjinfo, restrictlist, param_source_rels, semifactors)?;
+            let n_inner = run.root.rel(innerrel).cheapest_parameterized_paths.len();
+            for ii in 0..n_inner {
+                let innerpath = run.root.rel(innerrel).cheapest_parameterized_paths[ii];
+                try_nestloop_path(run, joinrel, outerpath, innerpath, merge_pathkeys, jointype, inner_unique, sjinfo, restrictlist, param_source_rels, semifactors)?;
                 if let Some(mpath) = get_memoize_path(
                     run,
                     innerrel,
@@ -726,11 +681,11 @@ fn match_unsorted_outer<'mcx>(
                     inner_unique,
                     restrictlist,
                 )? {
-                    try_nestloop_path(run, joinrel, outerpath, mpath, &merge_pathkeys, jointype, inner_unique, sjinfo, restrictlist, param_source_rels, semifactors)?;
+                    try_nestloop_path(run, joinrel, outerpath, mpath, merge_pathkeys, jointype, inner_unique, sjinfo, restrictlist, param_source_rels, semifactors)?;
                 }
             }
             if let Some(mp) = matpath {
-                try_nestloop_path(run, joinrel, outerpath, mp, &merge_pathkeys, jointype, inner_unique, sjinfo, restrictlist, param_source_rels, semifactors)?;
+                try_nestloop_path(run, joinrel, outerpath, mp, merge_pathkeys, jointype, inner_unique, sjinfo, restrictlist, param_source_rels, semifactors)?;
             }
         }
         if save_jointype == JOIN_UNIQUE_OUTER {
@@ -746,6 +701,7 @@ fn match_unsorted_outer<'mcx>(
                 joinrel,
                 innerrel,
                 outerpath,
+                &outer_pathkeys,
                 save_jointype,
                 inner_unique,
                 sjinfo,
@@ -753,7 +709,7 @@ fn match_unsorted_outer<'mcx>(
                 mergeclause_list,
                 useallclauses,
                 ict_for_merge,
-                &merge_pathkeys,
+                merge_pathkeys,
                 param_source_rels,
                 semifactors,
                 false,
@@ -787,8 +743,7 @@ fn match_unsorted_outer<'mcx>(
             if save_jointype == JOIN_UNIQUE_INNER {
                 return Ok(());
             }
-            let pl = crate::relnode::pgvec_clone_shallow(run.mcx, &run.root.rel(innerrel).pathlist);
-            pict = get_cheapest_parallel_safe_total_inner(run, &pl);
+            pict = get_cheapest_parallel_safe_total_inner(run, &run.root.rel(innerrel).pathlist);
         }
         if let Some(ict) = pict {
             consider_parallel_mergejoin(
@@ -835,17 +790,18 @@ fn consider_parallel_nestloop<'mcx>(
         matpath = Some(create_material_path(run, innerrel, inner_cheapest_total));
     }
 
-    let outer_paths =
-        crate::relnode::pgvec_clone_shallow(run.mcx, &run.root.rel(outerrel).partial_pathlist);
-    for &outerpath in outer_paths.iter() {
-        let outer_pathkeys =
-            crate::relnode::pgvec_clone_shallow(run.mcx, &run.root.path(outerpath).base().pathkeys);
-        let pathkeys = build_join_pathkeys(run, joinrel, jointype, &outer_pathkeys)?;
-        let inner_candidates = crate::relnode::pgvec_clone_shallow(
-            run.mcx,
-            &run.root.rel(innerrel).cheapest_parameterized_paths,
-        );
-        for &raw_inner in inner_candidates.iter() {
+    // Rel lists indexed in place: only joinrel gains (partial) paths in this loop.
+    let mut outer_pathkeys: PgVec<'mcx, PathKey> = PgVec::new_in(run.mcx);
+    let n_outer = run.root.rel(outerrel).partial_pathlist.len();
+    for oi in 0..n_outer {
+        let outerpath = run.root.rel(outerrel).partial_pathlist[oi];
+        outer_pathkeys.clear();
+        outer_pathkeys.extend_from_slice(&run.root.path(outerpath).base().pathkeys);
+        let n = build_join_pathkeys_len(run, joinrel, jointype, &outer_pathkeys)?;
+        let pathkeys = &outer_pathkeys[..n];
+        let n_inner = run.root.rel(innerrel).cheapest_parameterized_paths.len();
+        for ii in 0..n_inner {
+            let raw_inner = run.root.rel(innerrel).cheapest_parameterized_paths[ii];
             if !run.root.path(raw_inner).base().parallel_safe {
                 continue;
             }
@@ -859,7 +815,7 @@ fn consider_parallel_nestloop<'mcx>(
                 raw_inner
             };
             try_partial_nestloop_path(
-                run, joinrel, outerpath, innerpath, &pathkeys, jointype, inner_unique,
+                run, joinrel, outerpath, innerpath, pathkeys, jointype, inner_unique,
                 sjinfo, restrictlist, param_source_rels, semifactors,
             )?;
             if let Some(mpath) = get_memoize_path(
@@ -867,14 +823,14 @@ fn consider_parallel_nestloop<'mcx>(
                 restrictlist,
             )? {
                 try_partial_nestloop_path(
-                    run, joinrel, outerpath, mpath, &pathkeys, jointype, inner_unique,
+                    run, joinrel, outerpath, mpath, pathkeys, jointype, inner_unique,
                     sjinfo, restrictlist, param_source_rels, semifactors,
                 )?;
             }
         }
         if let Some(mp) = matpath {
             try_partial_nestloop_path(
-                run, joinrel, outerpath, mp, &pathkeys, jointype, inner_unique,
+                run, joinrel, outerpath, mp, pathkeys, jointype, inner_unique,
                 sjinfo, restrictlist, param_source_rels, semifactors,
             )?;
         }
@@ -898,17 +854,21 @@ fn consider_parallel_mergejoin<'mcx>(
     inner_cheapest_total: PathId,
     semifactors: Option<SemiAntiJoinFactors>,
 ) -> PgResult<()> {
-    let outer_paths =
-        crate::relnode::pgvec_clone_shallow(run.mcx, &run.root.rel(outerrel).partial_pathlist);
-    for &outerpath in outer_paths.iter() {
-        let outer_pathkeys =
-            crate::relnode::pgvec_clone_shallow(run.mcx, &run.root.path(outerpath).base().pathkeys);
-        let merge_pathkeys = build_join_pathkeys(run, joinrel, jointype, &outer_pathkeys)?;
+    // outerrel's partial_pathlist indexed in place: only joinrel gains partial paths here.
+    let mut outer_pathkeys: PgVec<'mcx, PathKey> = PgVec::new_in(run.mcx);
+    let n_outer = run.root.rel(outerrel).partial_pathlist.len();
+    for oi in 0..n_outer {
+        let outerpath = run.root.rel(outerrel).partial_pathlist[oi];
+        outer_pathkeys.clear();
+        outer_pathkeys.extend_from_slice(&run.root.path(outerpath).base().pathkeys);
+        let n = build_join_pathkeys_len(run, joinrel, jointype, &outer_pathkeys)?;
+        let merge_pathkeys = &outer_pathkeys[..n];
         generate_mergejoin_paths(
             run,
             joinrel,
             innerrel,
             outerpath,
+            &outer_pathkeys,
             jointype,
             inner_unique,
             sjinfo,
@@ -916,7 +876,7 @@ fn consider_parallel_mergejoin<'mcx>(
             mergeclause_list,
             false,
             inner_cheapest_total,
-            &merge_pathkeys,
+            merge_pathkeys,
             &crate::relnode::RELIDS_UNSET,
             semifactors,
             true,
@@ -1118,48 +1078,36 @@ fn get_memoize_path<'mcx>(
     // Unique joins skip to the next outer tuple on the first match; singlerow
     // caching only works when the whole join condition is parameterized.
     if inner_unique {
-        let serials = {
-            let Some(pi) = &run.root.path(inner_path).base().param_info else {
-                return Ok(None);
-            };
-            crate::relnode::relids_copy(run.mcx, &pi.ppi_serials)
+        let root = &run.root;
+        let Some(pi) = &root.path(inner_path).base().param_info else {
+            return Ok(None);
         };
         for &rid in restrictlist {
-            if !crate::relnode::relids_is_member(run.root.rinfo(rid).rinfo_serial, &serials) {
+            if !crate::relnode::relids_is_member(root.rinfo(rid).rinfo_serial, &pi.ppi_serials) {
                 return Ok(None);
             }
         }
     }
-    // A cache hit would skip volatile-function calls the query expects.
+    // A cache hit would skip volatile calls the query expects; read-only, so lists are walked in place.
     {
-        let exprs = crate::relnode::pgvec_clone_shallow(
-            run.mcx,
-            &run.root.rel_reltarget(innerrel).exprs,
-        );
-        for &e in exprs.iter() {
-            if clauses::contain_volatile_functions(*run.root.expr_node(e))? {
+        let root = &run.root;
+        for &e in root.rel_reltarget(innerrel).exprs.iter() {
+            if clauses::contain_volatile_functions(*root.expr_node(e))? {
                 return Ok(None);
             }
         }
-        let base_rinfos = crate::relnode::pgvec_clone_shallow(
-            run.mcx,
-            &run.root.rel(innerrel).baserestrictinfo,
-        );
-        for &rid in base_rinfos.iter() {
-            let clause = *run.root.expr_node(run.root.rinfo(rid).clause);
+        for &rid in root.rel(innerrel).baserestrictinfo.iter() {
+            let clause = *root.expr_node(root.rinfo(rid).clause);
             if clauses::contain_volatile_functions(clause)? {
                 return Ok(None);
             }
         }
-        let ppi_clauses: PgVec<'mcx, RinfoId> =
-            match &run.root.path(inner_path).base().param_info {
-                Some(pi) => crate::relnode::pgvec_clone_shallow(run.mcx, &pi.ppi_clauses),
-                None => PgVec::new_in(run.mcx),
-            };
-        for &rid in ppi_clauses.iter() {
-            let clause = *run.root.expr_node(run.root.rinfo(rid).clause);
-            if clauses::contain_volatile_functions(clause)? {
-                return Ok(None);
+        if let Some(pi) = &root.path(inner_path).base().param_info {
+            for &rid in pi.ppi_clauses.iter() {
+                let clause = *root.expr_node(root.rinfo(rid).clause);
+                if clauses::contain_volatile_functions(clause)? {
+                    return Ok(None);
+                }
             }
         }
     }
@@ -1202,13 +1150,13 @@ fn try_nestloop_path<'mcx>(
         relids_overlap, relids_union,
     };
     let mcx = run.mcx;
-    let inner_paramrels =
-        relids_copy(mcx, crate::pathnode::path_req_outer(run.root.path(inner_path).base()));
-    let outer_paramrels =
-        relids_copy(mcx, crate::pathnode::path_req_outer(run.root.path(outer_path).base()));
+    // Read-only until costing, so the relids are borrowed; only required_outer is built.
+    let root = &run.root;
+    let inner_paramrels = crate::pathnode::path_req_outer(root.path(inner_path).base());
+    let outer_paramrels = crate::pathnode::path_req_outer(root.path(outer_path).base());
     if sjinfo.ojrelid != 0
-        && (relids_is_member(sjinfo.ojrelid as i32, &inner_paramrels)
-            || relids_is_member(sjinfo.ojrelid as i32, &outer_paramrels))
+        && (relids_is_member(sjinfo.ojrelid as i32, inner_paramrels)
+            || relids_is_member(sjinfo.ojrelid as i32, outer_paramrels))
     {
         return Ok(());
     }
@@ -1216,22 +1164,22 @@ fn try_nestloop_path<'mcx>(
     // (reparameterize_path_by_child runs at create_plan), so the tests below
     // use topmost-parent relids too.
     let top_or_self = |r: RelId| {
-        if relids_is_empty(&run.root.rel(r).top_parent_relids) {
-            relids_copy(mcx, &run.root.rel(r).relids)
+        if relids_is_empty(&root.rel(r).top_parent_relids) {
+            &root.rel(r).relids
         } else {
-            relids_copy(mcx, &run.root.rel(r).top_parent_relids)
+            &root.rel(r).top_parent_relids
         }
     };
-    let innerrelids = top_or_self(run.root.path(inner_path).base().parent);
-    let outerrelids = top_or_self(run.root.path(outer_path).base().parent);
+    let innerrelids = top_or_self(root.path(inner_path).base().parent);
+    let outerrelids = top_or_self(root.path(outer_path).base().parent);
 
     // calc_nestloop_required_outer (pathnode.c).
-    debug_assert!(!relids_overlap(&outer_paramrels, &innerrelids));
-    let required_outer = if relids_is_empty(&inner_paramrels) {
-        relids_copy(mcx, &outer_paramrels)
+    debug_assert!(!relids_overlap(outer_paramrels, innerrelids));
+    let required_outer = if relids_is_empty(inner_paramrels) {
+        relids_copy(mcx, outer_paramrels)
     } else {
-        let u = relids_union(mcx, &outer_paramrels, &inner_paramrels);
-        let d = relids_difference(mcx, &u, &outerrelids);
+        let u = relids_union(mcx, outer_paramrels, inner_paramrels);
+        let d = relids_difference(mcx, &u, outerrelids);
         if relids_is_empty(&d) {
             crate::relnode::relids_empty()
         } else {
@@ -1242,8 +1190,8 @@ fn try_nestloop_path<'mcx>(
         && !relids_overlap(&required_outer, param_source_rels)
     {
         // allow_star_schema_join (joinpath.c).
-        let star = relids_overlap(&inner_paramrels, &outerrelids)
-            && !relids_is_subset(&inner_paramrels, &outerrelids);
+        let star = relids_overlap(inner_paramrels, outerrelids)
+            && !relids_is_subset(inner_paramrels, outerrelids);
         if !star {
             return Ok(());
         }
@@ -1299,22 +1247,21 @@ fn try_partial_nestloop_path<'mcx>(
     _param_source_rels: &Relids<'mcx>,
     semifactors: Option<SemiAntiJoinFactors>,
 ) -> PgResult<()> {
-    use crate::relnode::{relids_copy, relids_is_empty, relids_is_subset};
-    let mcx = run.mcx;
+    use crate::relnode::{relids_is_empty, relids_is_subset};
     debug_assert!(relids_is_empty(&run.root.rel(joinrel).lateral_relids));
     debug_assert!(relids_is_empty(crate::pathnode::path_req_outer(
         run.root.path(outer_path).base()
     )));
-    let inner_paramrels =
-        relids_copy(mcx, crate::pathnode::path_req_outer(run.root.path(inner_path).base()));
-    if !relids_is_empty(&inner_paramrels) {
-        let outer_parent = run.root.path(outer_path).base().parent;
-        let outerrelids = if relids_is_empty(&run.root.rel(outer_parent).top_parent_relids) {
-            relids_copy(mcx, &run.root.rel(outer_parent).relids)
+    let root = &run.root;
+    let inner_paramrels = crate::pathnode::path_req_outer(root.path(inner_path).base());
+    if !relids_is_empty(inner_paramrels) {
+        let outer_parent = root.path(outer_path).base().parent;
+        let outerrelids = if relids_is_empty(&root.rel(outer_parent).top_parent_relids) {
+            &root.rel(outer_parent).relids
         } else {
-            relids_copy(mcx, &run.root.rel(outer_parent).top_parent_relids)
+            &root.rel(outer_parent).top_parent_relids
         };
-        if !relids_is_subset(&inner_paramrels, &outerrelids) {
+        if !relids_is_subset(inner_paramrels, outerrelids) {
             return Ok(());
         }
     }
@@ -1346,10 +1293,10 @@ fn try_mergejoin_path<'mcx>(
     joinrel: RelId,
     outer_path: PathId,
     inner_path: PathId,
-    pathkeys: PgVec<'mcx, PathKey>,
-    mergeclauses: PgVec<'mcx, RinfoId>,
-    mut outersortkeys: PgVec<'mcx, PathKey>,
-    mut innersortkeys: PgVec<'mcx, PathKey>,
+    pathkeys: &[PathKey],
+    mergeclauses: &[RinfoId],
+    mut outersortkeys: &[PathKey],
+    mut innersortkeys: &[PathKey],
     jointype: u32,
     inner_unique: bool,
     sjinfo: &SpecialJoinInfo<'mcx>,
@@ -1389,33 +1336,33 @@ fn try_mergejoin_path<'mcx>(
     let mut outer_presorted_keys = 0usize;
     if !outersortkeys.is_empty() {
         let (contained, n) = pathkeys_count_contained_in(
-            &outersortkeys,
+            outersortkeys,
             &run.root.path(outer_path).base().pathkeys,
         );
         if contained {
-            outersortkeys.clear();
+            outersortkeys = &[];
         } else {
             outer_presorted_keys = n;
         }
     }
     if !innersortkeys.is_empty()
-        && pathkeys_contained_in(&innersortkeys, &run.root.path(inner_path).base().pathkeys)
+        && pathkeys_contained_in(innersortkeys, &run.root.path(inner_path).base().pathkeys)
     {
-        innersortkeys.clear();
+        innersortkeys = &[];
     }
 
     let workspace = initial_cost_mergejoin(
         run,
         jointype,
-        &mergeclauses,
+        mergeclauses,
         outer_path,
         inner_path,
-        &outersortkeys,
-        &innersortkeys,
+        outersortkeys,
+        innersortkeys,
         outer_presorted_keys,
     )?;
 
-    if add_path_precheck(run, joinrel, workspace.disabled_nodes, workspace.startup_cost, workspace.total_cost, &pathkeys, &required_outer) {
+    if add_path_precheck(run, joinrel, workspace.disabled_nodes, workspace.startup_cost, workspace.total_cost, pathkeys, &required_outer) {
         let path = create_mergejoin_path(
             run,
             joinrel,
@@ -1445,10 +1392,10 @@ fn try_partial_mergejoin_path<'mcx>(
     joinrel: RelId,
     outer_path: PathId,
     inner_path: PathId,
-    pathkeys: PgVec<'mcx, PathKey>,
-    mergeclauses: PgVec<'mcx, RinfoId>,
-    mut outersortkeys: PgVec<'mcx, PathKey>,
-    mut innersortkeys: PgVec<'mcx, PathKey>,
+    pathkeys: &[PathKey],
+    mergeclauses: &[RinfoId],
+    mut outersortkeys: &[PathKey],
+    mut innersortkeys: &[PathKey],
     jointype: u32,
     inner_unique: bool,
     sjinfo: &SpecialJoinInfo<'mcx>,
@@ -1466,32 +1413,32 @@ fn try_partial_mergejoin_path<'mcx>(
     let mut outer_presorted_keys = 0usize;
     if !outersortkeys.is_empty() {
         let (contained, n) = pathkeys_count_contained_in(
-            &outersortkeys,
+            outersortkeys,
             &run.root.path(outer_path).base().pathkeys,
         );
         if contained {
-            outersortkeys.clear();
+            outersortkeys = &[];
         } else {
             outer_presorted_keys = n;
         }
     }
     if !innersortkeys.is_empty()
-        && pathkeys_contained_in(&innersortkeys, &run.root.path(inner_path).base().pathkeys)
+        && pathkeys_contained_in(innersortkeys, &run.root.path(inner_path).base().pathkeys)
     {
-        innersortkeys.clear();
+        innersortkeys = &[];
     }
 
     let workspace = initial_cost_mergejoin(
         run,
         jointype,
-        &mergeclauses,
+        mergeclauses,
         outer_path,
         inner_path,
-        &outersortkeys,
-        &innersortkeys,
+        outersortkeys,
+        innersortkeys,
         outer_presorted_keys,
     )?;
-    if !add_partial_path_precheck(run, joinrel, workspace.disabled_nodes, workspace.total_cost, &pathkeys) {
+    if !add_partial_path_precheck(run, joinrel, workspace.disabled_nodes, workspace.total_cost, pathkeys) {
         return Ok(());
     }
     let path = create_mergejoin_path(
@@ -1520,10 +1467,12 @@ fn hash_inner_and_outer<'mcx>(
     use types_pathnodes::{JOIN_UNIQUE_INNER, JOIN_UNIQUE_OUTER};
     let save_jointype = jointype;
     let isouterjoin = is_outer_join(jointype);
-    let joinrelids = crate::relnode::relids_copy(run.mcx, &run.root.rel(joinrel).relids);
+    // joinrel relids are re-borrowed per clause: clause_sides_match_join needs `run` mutably.
     let mut hashclauses: PgVec<'mcx, RinfoId> = PgVec::new_in(run.mcx);
     for &ri in restrictlist {
-        if isouterjoin && crate::joinrels::rinfo_is_pushed_down(run, ri, &joinrelids) {
+        if isouterjoin
+            && crate::joinrels::rinfo_is_pushed_down(run, ri, &run.root.rel(joinrel).relids)
+        {
             continue;
         }
         let r = run.root.rinfo(ri);
@@ -1603,15 +1552,16 @@ fn hash_inner_and_outer<'mcx>(
         )?;
     }
 
-    let outer_params =
-        crate::relnode::pgvec_clone_shallow(run.mcx, &run.root.rel(outerrel).cheapest_parameterized_paths);
-    let inner_params =
-        crate::relnode::pgvec_clone_shallow(run.mcx, &run.root.rel(innerrel).cheapest_parameterized_paths);
-    for &op in outer_params.iter() {
+    // Rel lists indexed in place: only joinrel gains paths in this loop.
+    let n_outer = run.root.rel(outerrel).cheapest_parameterized_paths.len();
+    let n_inner = run.root.rel(innerrel).cheapest_parameterized_paths.len();
+    for oi in 0..n_outer {
+        let op = run.root.rel(outerrel).cheapest_parameterized_paths[oi];
         if path_param_by_rel(run, op, innerrel) {
             continue;
         }
-        for &ip in inner_params.iter() {
+        for ii in 0..n_inner {
+            let ip = run.root.rel(innerrel).cheapest_parameterized_paths[ii];
             if path_param_by_rel(run, ip, outerrel) {
                 continue;
             }
@@ -1654,8 +1604,7 @@ fn hash_inner_and_outer<'mcx>(
         } else if run.root.path(cheapest_total_inner).base().parallel_safe {
             Some(cheapest_total_inner)
         } else if save_jointype != JOIN_UNIQUE_INNER {
-            let pl = crate::relnode::pgvec_clone_shallow(run.mcx, &run.root.rel(innerrel).pathlist);
-            get_cheapest_parallel_safe_total_inner(run, &pl)
+            get_cheapest_parallel_safe_total_inner(run, &run.root.rel(innerrel).pathlist)
         } else {
             None
         };
@@ -1676,25 +1625,25 @@ fn clause_sides_match_join(
     outerrel: RelId,
     innerrel: RelId,
 ) -> bool {
-    let (left, right) = {
-        let r = run.root.rinfo(ri);
-        (r.left_relids.clone(), r.right_relids.clone())
+    use crate::relnode::relids_is_subset;
+    let outer_is_left = {
+        let root = &run.root;
+        let r = root.rinfo(ri);
+        let (outer_relids, inner_relids) = (&root.rel(outerrel).relids, &root.rel(innerrel).relids);
+        if relids_is_subset(&r.left_relids, outer_relids)
+            && relids_is_subset(&r.right_relids, inner_relids)
+        {
+            true
+        } else if relids_is_subset(&r.left_relids, inner_relids)
+            && relids_is_subset(&r.right_relids, outer_relids)
+        {
+            false
+        } else {
+            return false;
+        }
     };
-    let outer_relids = run.root.rel(outerrel).relids.clone();
-    let inner_relids = run.root.rel(innerrel).relids.clone();
-    if crate::relnode::relids_is_subset(&left, &outer_relids)
-        && crate::relnode::relids_is_subset(&right, &inner_relids)
-    {
-        run.root.rinfo_mut(ri).outer_is_left = true;
-        true
-    } else if crate::relnode::relids_is_subset(&left, &inner_relids)
-        && crate::relnode::relids_is_subset(&right, &outer_relids)
-    {
-        run.root.rinfo_mut(ri).outer_is_left = false;
-        true
-    } else {
-        false
-    }
+    run.root.rinfo_mut(ri).outer_is_left = outer_is_left;
+    true
 }
 
 #[allow(clippy::too_many_arguments)]
diff --git a/crates/backend/optimizer/plan/planner/src/pathkeys.rs b/crates/backend/optimizer/plan/planner/src/pathkeys.rs
index ab30fe29bf..32a171695e 100644
--- a/crates/backend/optimizer/plan/planner/src/pathkeys.rs
+++ b/crates/backend/optimizer/plan/planner/src/pathkeys.rs
@@ -1013,21 +1013,21 @@ pub fn trim_mergeclauses_for_inner_pathkeys<'mcx>(
     new_mergeclauses
 }
 
-// build_join_pathkeys (pathkeys.c); FULL/RIGHT/RIGHT_ANTI (NIL result) are
-// loud upstream of add_paths_to_joinrel.
-pub fn build_join_pathkeys<'mcx>(
-    run: &mut PlannerRun<'mcx>,
+// build_join_pathkeys (pathkeys.c) as the kept outer_pathkeys prefix length, so callers borrow a slice.
+// FULL/RIGHT/RIGHT_ANTI (NIL result) are loud upstream of add_paths_to_joinrel.
+pub fn build_join_pathkeys_len(
+    run: &mut PlannerRun<'_>,
     joinrel: types_pathnodes::RelId,
     jointype: u32,
     outer_pathkeys: &[PathKey],
-) -> PgResult<PgVec<'mcx, PathKey>> {
+) -> PgResult<usize> {
     if matches!(
         jointype,
         types_pathnodes::JOIN_FULL | types_pathnodes::JOIN_RIGHT | types_pathnodes::JOIN_RIGHT_ANTI
     ) {
-        return Ok(PgVec::new_in(run.mcx));
+        return Ok(0);
     }
-    truncate_useless_pathkeys(run, joinrel, outer_pathkeys)
+    useful_pathkeys_len(run, joinrel, outer_pathkeys)
 }
 
 pub fn truncate_useless_pathkeys<'mcx>(
@@ -1035,14 +1035,24 @@ pub fn truncate_useless_pathkeys<'mcx>(
     rel: types_pathnodes::RelId,
     pathkeys: &[PathKey],
 ) -> PgResult<PgVec<'mcx, PathKey>> {
+    let nuseful = useful_pathkeys_len(run, rel, pathkeys)?;
+    let mut out: PgVec<'mcx, PathKey> = PgVec::new_in(run.mcx);
+    out.extend(pathkeys[..nuseful].iter().copied());
+    Ok(out)
+}
+
+// truncate_useless_pathkeys' count of leading pathkeys worth keeping.
+fn useful_pathkeys_len(
+    run: &mut PlannerRun<'_>,
+    rel: types_pathnodes::RelId,
+    pathkeys: &[PathKey],
+) -> PgResult<usize> {
     let mut nuseful = pathkeys_useful_for_merging(run, rel, pathkeys)?;
     nuseful = nuseful.max(pathkeys_useful_for_ordering(run, pathkeys));
     nuseful = nuseful.max(pathkeys_useful_for_grouping(run, pathkeys));
     nuseful = nuseful.max(pathkeys_useful_for_distinct(run, pathkeys));
     nuseful = nuseful.max(pathkeys_useful_for_setop(run, pathkeys));
-    let mut out: PgVec<'mcx, PathKey> = PgVec::new_in(run.mcx);
-    out.extend(pathkeys.iter().take(nuseful).copied());
-    Ok(out)
+    Ok(nuseful.min(pathkeys.len()))
 }
 
 fn pathkeys_useful_for_merging(
diff --git a/crates/backend/optimizer/util/pathnode/src/lib.rs b/crates/backend/optimizer/util/pathnode/src/lib.rs
index 17aad6af0e..6837fa909b 100644
--- a/crates/backend/optimizer/util/pathnode/src/lib.rs
+++ b/crates/backend/optimizer/util/pathnode/src/lib.rs
@@ -906,24 +906,28 @@ pub fn calc_non_nestloop_required_outer<'mcx>(
     outer_path: PathId,
     inner_path: PathId,
 ) -> Relids<'mcx> {
-    use types_pathnodes::relids::{relids_copy, relids_is_empty, relids_overlap, relids_union};
-    let outer_paramrels = relids_copy(mcx, path_req_outer(run.root.path(outer_path).base()));
-    let inner_paramrels = relids_copy(mcx, path_req_outer(run.root.path(inner_path).base()));
+    use types_pathnodes::relids::{relids_is_empty, relids_overlap, relids_union};
+    let outer_paramrels = path_req_outer(run.root.path(outer_path).base());
+    let inner_paramrels = path_req_outer(run.root.path(inner_path).base());
     // Input parameterizations refer to topmost parents until
     // reparameterize_path_by_child runs, so the disallowed-parameterization
     // tests use topmost-parent relids.
     let top_or_self = |r: types_pathnodes::RelId| {
         if relids_is_empty(&run.root.rel(r).top_parent_relids) {
-            relids_copy(mcx, &run.root.rel(r).relids)
+            &run.root.rel(r).relids
         } else {
-            relids_copy(mcx, &run.root.rel(r).top_parent_relids)
+            &run.root.rel(r).top_parent_relids
         }
     };
-    let outerrelids = top_or_self(run.root.path(outer_path).base().parent);
-    let innerrelids = top_or_self(run.root.path(inner_path).base().parent);
-    debug_assert!(!relids_overlap(&outer_paramrels, &innerrelids));
-    debug_assert!(!relids_overlap(&inner_paramrels, &outerrelids));
-    relids_union(mcx, &outer_paramrels, &inner_paramrels)
+    debug_assert!(!relids_overlap(
+        outer_paramrels,
+        top_or_self(run.root.path(inner_path).base().parent)
+    ));
+    debug_assert!(!relids_overlap(
+        inner_paramrels,
+        top_or_self(run.root.path(outer_path).base().parent)
+    ));
+    relids_union(mcx, outer_paramrels, inner_paramrels)
 }
 
 // join_clause_is_movable_into (restrictinfo.c). The body lives in
@@ -3733,32 +3737,32 @@ pub fn create_nestloop_path<'mcx>(
     required_outer: &Relids<'mcx>,
     semifactors: Option<SemiAntiJoinFactors>,
 ) -> PgResult<PathId> {
-    use types_pathnodes::relids::{relids_copy, relids_is_empty, relids_is_member, relids_overlap};
+    use types_pathnodes::relids::{relids_is_empty, relids_is_member, relids_overlap};
     let mcx = run.mcx;
 
     // Clauses already enforced inside the parameterized inner path (matched by
     // rinfo_serial) are removed from the join's own restrict list now: the
     // list feeds this path's size and cost estimates.
-    let mut restrict_vec: PgVec<'mcx, RinfoId> = PgVec::new_in(mcx);
-    restrict_vec.extend(restrict_clauses.iter().copied());
-    let inner_req_outer = relids_copy(mcx, path_req_outer(run.root.path(inner_path).base()));
-    let outerrelids = {
-        let r = run.root.path(outer_path).base().parent;
-        if relids_is_empty(&run.root.rel(r).top_parent_relids) {
-            relids_copy(mcx, &run.root.rel(r).relids)
+    let inner_enforces_joinquals = {
+        let outer_rel = run.root.rel(run.root.path(outer_path).base().parent);
+        let outerrelids = if relids_is_empty(&outer_rel.top_parent_relids) {
+            &outer_rel.relids
         } else {
-            relids_copy(mcx, &run.root.rel(r).top_parent_relids)
-        }
+            &outer_rel.top_parent_relids
+        };
+        relids_overlap(path_req_outer(run.root.path(inner_path).base()), outerrelids)
     };
-    if relids_overlap(&inner_req_outer, &outerrelids) {
+    let mut restrict_vec: PgVec<'mcx, RinfoId> =
+        PgVec::with_capacity_in(restrict_clauses.len(), mcx);
+    if inner_enforces_joinquals {
         let enforced_serials = get_param_path_clause_serials(run, inner_path);
-        let mut jclauses: PgVec<'mcx, RinfoId> = PgVec::new_in(mcx);
-        for &rid in restrict_vec.iter() {
+        for &rid in restrict_clauses {
             if !relids_is_member(run.root.rinfo(rid).rinfo_serial, &enforced_serials) {
-                jclauses.push(rid);
+                restrict_vec.push(rid);
             }
         }
-        restrict_vec = jclauses;
+    } else {
+        restrict_vec.extend_from_slice(restrict_clauses);
     }
 
     let param_info = get_joinrel_parampathinfo(
@@ -3884,7 +3888,7 @@ pub fn create_hashjoin_path<'mcx>(
 }
 
 
-// create_mergejoin_path (pathnode.c).
+// create_mergejoin_path (pathnode.c); copies the borrowed lists so precheck-rejected candidates allocate nothing.
 #[allow(clippy::too_many_arguments)]
 pub fn create_mergejoin_path<'mcx>(
     run: &mut PlannerRun<'mcx>,
@@ -3896,11 +3900,11 @@ pub fn create_mergejoin_path<'mcx>(
     outer_path: PathId,
     inner_path: PathId,
     restrict_clauses: &[RinfoId],
-    pathkeys: PgVec<'mcx, PathKey>,
+    pathkeys: &[PathKey],
     required_outer: &Relids<'mcx>,
-    mergeclauses: PgVec<'mcx, RinfoId>,
-    outersortkeys: PgVec<'mcx, PathKey>,
-    innersortkeys: PgVec<'mcx, PathKey>,
+    mergeclauses: &[RinfoId],
+    outersortkeys: &[PathKey],
+    innersortkeys: &[PathKey],
     outer_presorted_keys: usize,
 ) -> PgResult<PathId> {
     let mcx = run.mcx;
@@ -3937,7 +3941,7 @@ pub fn create_mergejoin_path<'mcx>(
         disabled_nodes: 0,
         startup_cost: 0.0,
         total_cost: 0.0,
-        pathkeys,
+        pathkeys: mcx::slice_in(mcx, pathkeys)?,
     };
     let mut node = MergePath {
         jpath: JoinPath {
@@ -3948,9 +3952,9 @@ pub fn create_mergejoin_path<'mcx>(
             innerjoinpath: Some(inner_path),
             joinrestrictinfo,
         },
-        path_mergeclauses: mergeclauses,
-        outersortkeys,
-        innersortkeys,
+        path_mergeclauses: mcx::slice_in(mcx, mergeclauses)?,
+        outersortkeys: mcx::slice_in(mcx, outersortkeys)?,
+        innersortkeys: mcx::slice_in(mcx, innersortkeys)?,
         outer_presorted_keys: outer_presorted_keys as i32,
         skip_mark_restore: false,
         materialize_inner: false,
diff --git a/crates/backend/utils/adt/ri_triggers/Cargo.toml b/crates/backend/utils/adt/ri_triggers/Cargo.toml
index 07a891bf29..18f63f7973 100644
--- a/crates/backend/utils/adt/ri_triggers/Cargo.toml
+++ b/crates/backend/utils/adt/ri_triggers/Cargo.toml
@@ -45,3 +45,10 @@ ruleutils = { path = "../ruleutils" }
 rls_seams = { path = "../../misc/rls_seams" }
 aclchk_seams = { path = "../../../catalog/aclchk_seams" }
 types_nodes = { path = "../../../../_support/types/nodes" }
+indexam = { path = "../../../access/index/indexam" }
+tableam = { path = "../../../access/table/tableam" }
+exectuples = { path = "../../../executor/exectuples" }
+namespace_seams = { path = "../../../catalog/namespace_seams" }
+objectaccess = { path = "../../../catalog/objectaccess" }
+types_scan = { path = "../../../../_support/types/types_scan" }
+types_slot = { path = "../../../../_support/types/types_slot" }
diff --git a/crates/backend/utils/adt/ri_triggers/src/lib.rs b/crates/backend/utils/adt/ri_triggers/src/lib.rs
index c3c6fe4047..8b5d753cdd 100644
--- a/crates/backend/utils/adt/ri_triggers/src/lib.rs
+++ b/crates/backend/utils/adt/ri_triggers/src/lib.rs
@@ -19,9 +19,9 @@ use types_tuple::NameData;
 use types_error::{
     PgError, PgResult, ERRCODE_DATA_CORRUPTED, ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED,
     ERRCODE_FOREIGN_KEY_VIOLATION, ERRCODE_INTERNAL_ERROR, ERRCODE_INVALID_OBJECT_DEFINITION,
-    ERRCODE_RESTRICT_VIOLATION, ERROR,
+    ERRCODE_RESTRICT_VIOLATION, ERRCODE_T_R_SERIALIZATION_FAILURE, ERROR,
 };
-use types_rel::{Relation, RowShareLock, RELKIND_PARTITIONED_TABLE};
+use types_rel::{AccessShareLock, NoLock, Relation, RowShareLock, RELKIND_PARTITIONED_TABLE};
 use types_trigger::{
     Trigger, TRIGGER_FIRED_AFTER, TRIGGER_FIRED_BY_DELETE, TRIGGER_FIRED_BY_INSERT,
     TRIGGER_FIRED_BY_UPDATE, TRIGGER_FIRED_FOR_ROW,
@@ -93,6 +93,26 @@ struct RiConstraintInfo {
     period_contained_by_oper: Oid,
     agged_period_contained_by_oper: Oid,
     period_intersect_oper: Oid,
+    conindid: Oid,
+    fastpath_state: RiFastPathState,
+    fpmeta: Option<FastPathMeta>,
+}
+
+// RI_FastPathState (ri_triggers.c): eligibility checks that need the PK index open.
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum RiFastPathState {
+    Unknown,
+    Usable,
+    Unusable,
+}
+
+// FastPathMeta (ri_triggers.c): per-key scan-key metadata for the direct PK index probe.
+#[derive(Clone, Copy)]
+struct FastPathMeta {
+    regops: [Oid; RI_MAX_NUMKEYS],
+    subtypes: [Oid; RI_MAX_NUMKEYS],
+    strats: [u16; RI_MAX_NUMKEYS],
+    index_attnos: [i16; RI_MAX_NUMKEYS],
 }
 
 fn cache_mcx() -> &'static MemoryContext {
@@ -409,6 +429,14 @@ fn RI_FKey_check<'mcx>(mcx: Mcx<'mcx>, tgdata: &RiTriggerData<'_, 'mcx>) -> PgRe
         _ => {}
     }
 
+    // Direct PK index probe instead of the SPI query; false means a check made under the PK lock requires SPI.
+    if ri_fastpath_is_applicable(&riinfo, &pk_rel)
+        && ri_FastPathCheck(mcx, &riinfo, fk_rel, &pk_rel, newtup)?
+    {
+        // Keep RowShareLock to commit, as the SPI query's range-table lock would.
+        return pk_rel.close(NoLock);
+    }
+
     spi::SPI_connect()?;
 
     let qkey = ri_build_query_key(&riinfo, RI_PLAN_CHECK_LOOKUPPK);
@@ -1448,6 +1476,9 @@ fn ri_LoadConstraintInfo(constraint_oid: Oid) -> PgResult<RiConstraintInfo> {
         period_contained_by_oper: InvalidOid,
         agged_period_contained_by_oper: InvalidOid,
         period_intersect_oper: InvalidOid,
+        conindid: req(Anum_conindid)?.as_oid(),
+        fastpath_state: RiFastPathState::Unknown,
+        fpmeta: None,
     };
 
     // DeconstructFkConstraintRow (pg_constraint.c): 1-D no-null int2[]/oid[].
@@ -1722,6 +1753,398 @@ fn ri_ExtractValues(
     }
 }
 
+// ri_fastpath_is_applicable (ri_triggers.c): the per-row checks that need no index or catalog access.
+fn ri_fastpath_is_applicable(riinfo: &RiConstraintInfo, pk_rel: &Relation<'_>) -> bool {
+    // Partitioned PKs need partition routing and temporal FKs need range_agg(): both stay on SPI.
+    if pk_rel.rd_rel.relkind == RELKIND_PARTITIONED_TABLE || riinfo.hasperiod {
+        return false;
+    }
+    riinfo.fastpath_state != RiFastPathState::Unusable
+}
+
+// Write fast-path state onto the cached entry; a missing entry was invalidated and gets rebuilt.
+fn ri_cache_update(constraint_id: Oid, f: impl FnOnce(&mut RiConstraintInfo)) {
+    RI_CONSTRAINT_CACHE.with(|c| {
+        if let Some(e) = c.borrow_mut().as_mut().and_then(|m| m.get_mut(&constraint_id)) {
+            f(e);
+        }
+    });
+}
+
+// ri_FastPathCheck (ri_triggers.c): probe the PK unique index and lock the row without SPI.
+// Returns false when the index or the FK types require the SPI path; reports violations itself.
+fn ri_FastPathCheck<'mcx>(
+    mcx: Mcx<'mcx>,
+    riinfo: &RiConstraintInfo,
+    fk_rel: &Relation<'mcx>,
+    pk_rel: &Relation<'mcx>,
+    newtup: &HeapTupleData<'_>,
+) -> PgResult<bool> {
+    // ExecCheckXactReadOnly refuses the SPI query's row lock; pgx's executor names the command "SELECT".
+    if xact::XactReadOnly() && !namespace_seams::is_temp_namespace::call(pk_rel.rd_rel.relnamespace) {
+        xact::PreventCommandIfReadOnly("SELECT")?;
+    }
+
+    // SPI increments the command counter before the query, so earlier triggers' effects are seen;
+    // doing it under the PK lock also processes local invalidations before the reload below.
+    xact::CommandCounterIncrement()?;
+    let constraint_id = riinfo.constraint_id;
+    let conindid = ri_LoadConstraintInfo(constraint_id)?.conindid;
+    let idx_rel = indexam::index_open(mcx, conindid, AccessShareLock)?;
+    // Opening the index can have processed further invalidations.
+    let mut riinfo = ri_LoadConstraintInfo(constraint_id)?;
+    if !ri_check_fastpath_index(&mut riinfo, fk_rel, pk_rel, &idx_rel)? {
+        indexam::index_close(idx_rel, NoLock)?;
+        return Ok(false);
+    }
+
+    // Snapshot only after the PK lock is held, as SPI takes it, so a key committed while we waited
+    // is visible; active too, since STABLE equality functions read the active snapshot.
+    let snapshot = snapmgr::RegisterSnapshot(Some(&snapmgr::GetTransactionSnapshot()?))?
+        .expect("registered a snapshot");
+    snapmgr::PushActiveSnapshot(&snapshot)?;
+    let mut slot = tableam::table_slot_create(mcx, pk_rel)?;
+
+    let (_, save_sec_context) = miscinit::GetUserIdAndSecContext();
+    let guard = miscinit::SecContextGuard::set(
+        pk_rel.rd_rel.relowner,
+        save_sec_context | SECURITY_LOCAL_USERID_CHANGE | SECURITY_NOFORCE_RLS,
+    );
+    let found = (|| -> PgResult<bool> {
+        ri_CheckPermissions(mcx, &riinfo, pk_rel)?;
+        let mut scan =
+            indexam::index_beginscan(mcx, pk_rel, &idx_rel, snapshot.clone(), riinfo.nkeys as i32, 0)?;
+        let fpmeta = match riinfo.fpmeta {
+            Some(m) => m,
+            None => ri_populate_fastpath_metadata(&riinfo, &idx_rel)?,
+        };
+        ri_CheckFunctionPermissions(mcx, &riinfo, &fpmeta)?;
+        let mut pk_vals = [Datum::null(); RI_MAX_NUMKEYS];
+        let mut pk_nulls = [false; RI_MAX_NUMKEYS];
+        ri_ExtractValues(fk_rel, newtup, &riinfo, false, &mut pk_vals, &mut pk_nulls);
+        let mut skeys = build_index_scankeys(&riinfo, &fpmeta, &idx_rel, &pk_vals)?;
+        let found =
+            ri_FastPathProbeOne(mcx, pk_rel, &idx_rel, &mut scan, &mut slot, &snapshot, &mut skeys)?;
+        indexam::index_endscan(scan)?;
+        Ok(found)
+    })();
+    guard.restore();
+    exectuples::exec_clear_tuple(&mut slot, mcx);
+    snapmgr::UnregisterSnapshot(Some(&snapshot));
+    snapmgr::PopActiveSnapshot()?;
+
+    if !found? {
+        return Err(ri_ReportViolation(
+            mcx,
+            &riinfo,
+            pk_rel,
+            fk_rel,
+            newtup,
+            None,
+            RI_PLAN_CHECK_LOOKUPPK,
+            false,
+            false,
+        ));
+    }
+    indexam::index_close(idx_rel, NoLock)?;
+    Ok(true)
+}
+
+// ri_FastPathProbeOne (ri_triggers.c): true when a matching PK row was found and KEY SHARE locked.
+fn ri_FastPathProbeOne<'mcx>(
+    mcx: Mcx<'mcx>,
+    pk_rel: &Relation<'mcx>,
+    idx_rel: &Relation<'mcx>,
+    scan: &mut indexam::IndexScanDescData<'mcx>,
+    slot: &mut types_slot::SlotData<'mcx>,
+    snapshot: &snapmgr::Snapshot,
+    skeys: &mut [types_scan::ScanKeyData],
+) -> PgResult<bool> {
+    indexam::index_rescan(scan, Some(skeys), None)?;
+    if !indexam::index_getnext_slot(mcx, scan, types_scan::ScanDirection::ForwardScanDirection, slot)? {
+        return Ok(false);
+    }
+    match ri_LockPKTuple(mcx, pk_rel, slot, snapshot)? {
+        None => Ok(false),
+        Some(false) => Ok(true),
+        Some(true) => recheck_matched_pk_tuple(mcx, idx_rel, skeys, slot),
+    }
+}
+
+// ri_LockPKTuple (ri_triggers.c): ExecLockRows' table_tuple_lock handling for one PK row.
+// None when the row is gone; Some(traversed) when locked, traversed meaning an update chain was followed.
+fn ri_LockPKTuple<'mcx>(
+    mcx: Mcx<'mcx>,
+    pk_rel: &Relation<'mcx>,
+    slot: &mut types_slot::SlotData<'mcx>,
+    snapshot: &snapmgr::Snapshot,
+) -> PgResult<Option<bool>> {
+    use tableam::{LockTupleMode, LockWaitPolicy, TM_FailureData, TM_Result};
+    let uses_xact_snapshot = xact::IsolationUsesXactSnapshot();
+    let mut lockflags = tableam::TUPLE_LOCK_FLAG_LOCK_UPDATE_IN_PROGRESS;
+    if !uses_xact_snapshot {
+        lockflags |= tableam::TUPLE_LOCK_FLAG_FIND_LAST_VERSION;
+    }
+    let tid = slot.base().tts_tid;
+    let snap: tableam::Snapshot<'mcx> = Some(snapshot.clone());
+    let mut tmfd = TM_FailureData::default();
+    // Lock as of the scan snapshot's command id, as ExecLockRows uses es_output_cid: a later id
+    // (advanced by user code in the scan) would see this command's own update as TM_Invisible.
+    let result = tableam::table_tuple_lock(
+        mcx,
+        pk_rel,
+        &tid,
+        &snap,
+        slot,
+        snapshot.curcid.get(),
+        LockTupleMode::LockTupleKeyShare,
+        LockWaitPolicy::LockWaitBlock,
+        lockflags,
+        &mut tmfd,
+    )?;
+    match result {
+        TM_Result::TM_Ok => Ok(Some(tmfd.traversed)),
+        // ExecLockRows reports both concurrent updates and deletes with the "update" text.
+        TM_Result::TM_Deleted | TM_Result::TM_Updated if uses_xact_snapshot => Err(Box::new(
+            PgError::error("could not serialize access due to concurrent update".to_string())
+                .with_sqlstate(ERRCODE_T_R_SERIALIZATION_FAILURE),
+        )),
+        TM_Result::TM_Deleted => Ok(None),
+        TM_Result::TM_Updated => Err(Box::new(PgError::error(format!(
+            "unexpected table_tuple_lock status: {}",
+            result as u32
+        )))),
+        // Updated or deleted by this or a later command of this transaction: skipped like ExecLockRows.
+        TM_Result::TM_SelfModified => Ok(None),
+        TM_Result::TM_Invisible => {
+            Err(Box::new(PgError::error("attempted to lock invisible tuple".to_string())))
+        }
+        other => Err(Box::new(PgError::error(format!(
+            "unrecognized table_tuple_lock status: {}",
+            other as u32
+        )))),
+    }
+}
+
+// ri_check_fastpath_index (ri_triggers.c): index- and type-dependent eligibility, cached on the
+// entry. Checked under the PK lock with the index open, so REINDEX CONCURRENTLY cannot race it.
+fn ri_check_fastpath_index(
+    riinfo: &mut RiConstraintInfo,
+    fk_rel: &Relation<'_>,
+    pk_rel: &Relation<'_>,
+    idx_rel: &Relation<'_>,
+) -> PgResult<bool> {
+    if riinfo.fastpath_state != RiFastPathState::Unknown {
+        return Ok(riinfo.fastpath_state == RiFastPathState::Usable);
+    }
+    let state = if ri_fastpath_index_usable(riinfo, fk_rel, pk_rel, idx_rel)? {
+        RiFastPathState::Usable
+    } else {
+        RiFastPathState::Unusable
+    };
+    riinfo.fastpath_state = state;
+    ri_cache_update(riinfo.constraint_id, |e| e.fastpath_state = state);
+    Ok(state == RiFastPathState::Usable)
+}
+
+fn ri_fastpath_index_usable(
+    riinfo: &RiConstraintInfo,
+    fk_rel: &Relation<'_>,
+    pk_rel: &Relation<'_>,
+    idx_rel: &Relation<'_>,
+) -> PgResult<bool> {
+    // Any amcanunique AM can back a FK; the direct probe is written for btree only.
+    if idx_rel.rd_rel.relam != types_core::BTREE_AM_OID {
+        return Ok(false);
+    }
+    let index = idx_rel.rd_index.as_ref().expect("index relation has rd_index");
+    let nkeyatts = index.indnkeyatts as usize;
+    // Leave index collations differing from the referenced column's to SPI.
+    for i in 0..nkeyatts {
+        let attnum = index.indkey[i];
+        if attnum <= 0 || idx_rel.rd_indcollation[i] != att_collation(pk_rel, attnum) {
+            return Ok(false);
+        }
+    }
+    // The stored PK = FK operator must still be the opfamily's equality member.
+    for i in 0..riinfo.nkeys {
+        let Some(idx_col) = index.indkey[..nkeyatts].iter().position(|&a| a == riinfo.pk_attnums[i])
+        else {
+            return Ok(false);
+        };
+        if lsyscache::get_op_opfamily_strategy(riinfo.pf_eq_oprs[i], idx_rel.rd_opfamily[idx_col])?
+            != types_scan::BTEqualStrategyNumber as i32
+        {
+            return Ok(false);
+        }
+    }
+    // FK values go to the operator unconverted; anything needing a cast function stays on SPI.
+    for i in 0..riinfo.nkeys {
+        let fk_type = att_type(fk_rel, riinfo.fk_attnums[i]);
+        let (_, righttype) = lsyscache::op_input_types(riinfo.pf_eq_oprs[i])?;
+        if !coerce::IsBinaryCoercible(fk_type, righttype)? {
+            return Ok(false);
+        }
+    }
+    Ok(true)
+}
+
+// ri_CheckPermissions (ri_triggers.c): ExecCheckPermissions for the SPI path's
+// SELECT ... FOR KEY SHARE, run as the PK table's owner.
+fn ri_CheckPermissions(mcx: Mcx<'_>, riinfo: &RiConstraintInfo, pk_rel: &Relation<'_>) -> PgResult<()> {
+    use types_nodes::parsenodes::{ObjectType, ACL_SELECT, ACL_SELECT_FOR_UPDATE, ACL_UPDATE, ACL_USAGE};
+    const ACLCHECK_NO_PRIV: i32 = 1;
+    let userid = miscinit::GetUserId();
+    let nspid = pk_rel.rd_rel.relnamespace;
+    let aclresult = aclchk_seams::object_aclcheck::call(
+        types_core::NAMESPACE_RELATION_ID,
+        nspid,
+        userid,
+        ACL_USAGE,
+    )?;
+    if aclresult != ACLCHECK_OK {
+        let nspname = lsyscache::get_namespace_name(mcx, nspid)?;
+        aclchk_seams::aclcheck_error::call(
+            aclresult,
+            ObjectType::OBJECT_SCHEMA as i32,
+            nspname.as_ref().map_or("", |n| n.as_str()),
+        )?;
+    }
+
+    // ExecCheckOneRelPerms: SELECT on the key columns, UPDATE on any column for the row lock.
+    let relid = pk_rel.rd_id;
+    let required = ACL_SELECT | ACL_SELECT_FOR_UPDATE;
+    let remaining = required & !aclchk_seams::pg_class_aclmask::call(relid, userid, required, true)?;
+    let mut ok = true;
+    if remaining & ACL_SELECT != 0 {
+        for i in 0..riinfo.nkeys {
+            if aclchk_seams::pg_attribute_aclcheck::call(relid, riinfo.pk_attnums[i], userid, ACL_SELECT)?
+                != ACLCHECK_OK
+            {
+                ok = false;
+                break;
+            }
+        }
+    }
+    if ok && remaining & ACL_UPDATE != 0 {
+        ok = aclchk_seams::pg_attribute_aclcheck_all::call(relid, userid, ACL_UPDATE, false)?
+            == ACLCHECK_OK;
+    }
+    if !ok {
+        aclchk_seams::aclcheck_error::call(ACLCHECK_NO_PRIV, ObjectType::OBJECT_TABLE as i32, pk_rel.name())?;
+    }
+    Ok(())
+}
+
+// recheck_matched_pk_tuple (ri_triggers.c): after following an update chain, the locked version
+// must still carry the searched key; a key update means the referenced value is gone.
+fn recheck_matched_pk_tuple<'mcx>(
+    mcx: Mcx<'mcx>,
+    idx_rel: &Relation<'_>,
+    skeys: &mut [types_scan::ScanKeyData],
+    slot: &mut types_slot::SlotData<'mcx>,
+) -> PgResult<bool> {
+    let index = idx_rel.rd_index.as_ref().expect("index relation has rd_index");
+    for (i, skey) in skeys.iter_mut().enumerate() {
+        let mut isnull = false;
+        let value = exectuples::slot_getattr(slot, index.indkey[i] as i32, &mut isnull);
+        // A nullable unique column may hold NULL in the newer version: never equal.
+        if isnull {
+            return Ok(false);
+        }
+        let mut fcinfo = types_fmgr::LocalFcinfo::<2>::fresh(skey.sk_collation);
+        // SAFETY: mcx outlives this call.
+        unsafe { fcinfo.set_result_mcx(mcx) };
+        fcinfo.set_arg(0, value);
+        fcinfo.set_arg(1, skey.sk_argument);
+        if !skey.sk_func.invoke(&mut fcinfo)?.as_bool() {
+            return Ok(false);
+        }
+    }
+    Ok(true)
+}
+
+// ri_CheckFunctionPermissions (ri_triggers.c): the EXECUTE check ExecInitFunc makes on the SPI
+// query's equality operator; no cast function is involved on the fast path.
+fn ri_CheckFunctionPermissions(
+    mcx: Mcx<'_>,
+    riinfo: &RiConstraintInfo,
+    fpmeta: &FastPathMeta,
+) -> PgResult<()> {
+    use types_nodes::parsenodes::{ObjectType, ACL_EXECUTE};
+    for &funcid in &fpmeta.regops[..riinfo.nkeys] {
+        let aclresult = aclchk_seams::object_aclcheck::call(
+            types_core::PROCEDURE_RELATION_ID,
+            funcid,
+            miscinit::GetUserId(),
+            ACL_EXECUTE,
+        )?;
+        if aclresult != ACLCHECK_OK {
+            let name = lsyscache::get_func_name(mcx, funcid)?;
+            aclchk_seams::aclcheck_error::call(
+                aclresult,
+                ObjectType::OBJECT_FUNCTION as i32,
+                name.as_ref().map_or("", |n| n.as_str()),
+            )?;
+        }
+        objectaccess::InvokeFunctionExecuteHook(funcid)?;
+    }
+    Ok(())
+}
+
+// build_index_scankeys (ri_triggers.c): ExecIndexBuildScanKeys-style equality keys, each placed at
+// its index column's position because btree wants keys in attribute order.
+fn build_index_scankeys(
+    riinfo: &RiConstraintInfo,
+    fpmeta: &FastPathMeta,
+    idx_rel: &Relation<'_>,
+    pk_vals: &[Datum],
+) -> PgResult<Vec<types_scan::ScanKeyData>> {
+    let mut skeys: Vec<_> = (0..riinfo.nkeys).map(|_| types_scan::ScanKeyData::empty()).collect();
+    for i in 0..riinfo.nkeys {
+        let pos = fpmeta.index_attnos[i] as usize - 1;
+        let key: &mut types_scan::ScanKeyData = &mut skeys[pos];
+        key.sk_attno = fpmeta.index_attnos[i];
+        key.sk_strategy = fpmeta.strats[i];
+        key.sk_subtype = fpmeta.subtypes[i];
+        key.sk_collation = idx_rel.rd_indcollation[pos];
+        fmgr_core::fmgr_info_into(fpmeta.regops[i], &mut key.sk_func)?;
+        key.sk_argument = pk_vals[i];
+    }
+    Ok(skeys)
+}
+
+// ri_populate_fastpath_metadata (ri_triggers.c): look up the per-key operator procedure and
+// strategy/subtype once and cache them on the constraint entry.
+fn ri_populate_fastpath_metadata(
+    riinfo: &RiConstraintInfo,
+    idx_rel: &Relation<'_>,
+) -> PgResult<FastPathMeta> {
+    let index = idx_rel.rd_index.as_ref().expect("index relation has rd_index");
+    let mut fpmeta = FastPathMeta {
+        regops: [InvalidOid; RI_MAX_NUMKEYS],
+        subtypes: [InvalidOid; RI_MAX_NUMKEYS],
+        strats: [0; RI_MAX_NUMKEYS],
+        index_attnos: [0; RI_MAX_NUMKEYS],
+    };
+    for i in 0..riinfo.nkeys {
+        let eq_opr = riinfo.pf_eq_oprs[i];
+        // FK columns need not be listed in index order.
+        let idx_col = index.indkey[..riinfo.nkeys]
+            .iter()
+            .position(|&a| a == riinfo.pk_attnums[i])
+            .expect("referenced column is an index key");
+        fpmeta.index_attnos[i] = idx_col as i16 + 1;
+        fpmeta.regops[i] = lsyscache::get_opcode(eq_opr)?;
+        let (strat, _, subtype) =
+            lsyscache::get_op_opfamily_properties(eq_opr, idx_rel.rd_opfamily[idx_col], false)?;
+        fpmeta.strats[i] = strat as u16;
+        fpmeta.subtypes[i] = subtype;
+    }
+    ri_cache_update(riinfo.constraint_id, |e| e.fpmeta = Some(fpmeta));
+    Ok(fpmeta)
+}
+
 // C errtableconstraint(rel, conname): the schema/table/constraint
 // ErrorResponse fields that constraint-mapping clients (diesel's
 // constraint_name(), sequel's pg_auto_constraint_validations) read.
diff --git a/crates/backend/utils/adt/ri_triggers/src/tests.rs b/crates/backend/utils/adt/ri_triggers/src/tests.rs
index 3ba375f141..b86500bcbf 100644
--- a/crates/backend/utils/adt/ri_triggers/src/tests.rs
+++ b/crates/backend/utils/adt/ri_triggers/src/tests.rs
@@ -26,6 +26,9 @@ fn stub_info(constraint_oid: Oid, root: Oid, oid_hv: u32, root_hv: u32) -> RiCon
         period_contained_by_oper: InvalidOid,
         agged_period_contained_by_oper: InvalidOid,
         period_intersect_oper: InvalidOid,
+        conindid: InvalidOid,
+        fastpath_state: RiFastPathState::Unknown,
+        fpmeta: None,
     }
 }