fix: serialize data instead of pointer in plan_foreign_modify (supabase#548)

* fix: serialize data instead of pointer in plan_foreign_modify

Fix PostgreSQL server crash when using prepared statements with foreign
tables. PostgreSQL caches query plans after ~5-6 executions. The previous
implementation stored a raw pointer to FdwModifyState in fdw_private,
which became invalid after end_foreign_modify freed the state.

Solution:
- Add FdwModifyPrivate struct holding serializable reconstruction data
- Serialize table OID, rowid info (not pointer) in plan_foreign_modify
- Create fresh FdwModifyState in begin_foreign_modify for each execution

Also includes stress test for issue supabase#482 that executes 15 parameterized
INSERT operations to validate the fix.

Fixes supabase#482
Related to supabase#237

* fix: serialize data instead of pointer in plan_foreign_modify

Fixes PostgreSQL server crash when using prepared statements with
foreign tables (issue supabase#482).

The root cause was that PostgreSQL caches query plans after ~5-6
executions (generic plan optimization). The previous implementation
stored a raw pointer to FdwModifyState in fdw_private, which became
invalid after end_foreign_modify freed the state. Subsequent executions
with cached plans dereferenced this stale pointer, causing a crash.

Solution:
- Add FdwModifyPrivate struct holding serializable reconstruction data
- Serialize table OID, rowid info (not pointer) in plan_foreign_modify
- Create fresh FdwModifyState in begin_foreign_modify for each execution

This ensures each query execution gets a valid FDW instance and state,
even when PostgreSQL reuses a cached query plan and skips planning.

* test(clickhouse): add prepared statement stress test

Validates fix for issue supabase#482 by executing 15 parameterized INSERT
operations - enough to trigger PostgreSQL's generic plan caching.
Before the fix, this would crash around iteration 7.

* fix(clippy): inline format args in stress test

Fix uninlined-format-args clippy warning by using inline variable
in format string: format!("stress_{i}") instead of format!("stress_{}", i)

* update Cargo.lock

* format code

---------

Co-authored-by: Bo Lu <[email protected]>
Co-authored-by: Bo Lu <[email protected]>

Author: JohnCari

supabase-wrappers/src/modify.rs

@@ -1,14 +1,16 @@
 use pgrx::pg_sys::panic::ErrorReport;
 use pgrx::{
     debug2,
+    list::List,
     memcxt::PgMemoryContexts,
     pg_sys::{MemoryContext, MemoryContextData, Oid},
     prelude::*,
     rel::PgRelation,
     tupdesc::PgTupleDesc,
-    FromDatum, PgSqlErrorCode,
+    FromDatum, IntoDatum, PgSqlErrorCode,
 };
 use std::collections::HashMap;
+use std::ffi::c_void;
 use std::marker::PhantomData;
 use std::os::raw::c_int;
 use std::ptr;
@@ -20,6 +22,154 @@ use super::memctx;
 use super::polyfill;
 use super::utils;
 
+/// Serializable data for fdw_private in modify operations.
+/// This struct contains only data that can be safely serialized to a PostgreSQL List
+/// and survives query plan caching. The FdwModifyState is reconstructed from this
+/// data in begin_foreign_modify for each execution.
+struct FdwModifyPrivate {
+    /// Foreign table OID - used to create FDW instance and fetch options
+    foreigntableid: Oid,
+    /// Row identifier column name
+    rowid_name: String,
+    /// Row identifier type OID
+    rowid_typid: Oid,
+    /// Update columns (pg13 only)
+    #[cfg(feature = "pg13")]
+    update_cols: Vec<String>,
+}
+
+impl FdwModifyPrivate {
+    /// Serialize FdwModifyPrivate to a PostgreSQL List of Const nodes.
+    /// Format:
+    /// - [0] INT4: foreigntableid as i32
+    /// - [1] TEXT: rowid_name
+    /// - [2] INT4: rowid_typid as i32
+    /// - [3] INT4: update_cols_count (pg13 only)
+    /// - [4..N] TEXT: update_cols entries (pg13 only)
+    unsafe fn serialize_to_list(&self) -> *mut pg_sys::List {
+        pgrx::memcx::current_context(|mcx| {
+            let mut ret = List::<*mut c_void>::Nil;
+
+            // [0] foreigntableid as i32
+            let cst = pg_sys::makeConst(
+                pg_sys::INT4OID,
+                -1,
+                pg_sys::InvalidOid,
+                4,
+                (self.foreigntableid.to_u32() as i32).into_datum().unwrap(),
+                false,
+                true,
+            );
+            ret.unstable_push_in_context(cst as _, mcx);
+
+            // [1] rowid_name as TEXT
+            let cst = pg_sys::makeConst(
+                pg_sys::TEXTOID,
+                -1,
+                pg_sys::InvalidOid,
+                -1,
+                self.rowid_name.clone().into_datum().unwrap(),
+                false,
+                false,
+            );
+            ret.unstable_push_in_context(cst as _, mcx);
+
+            // [2] rowid_typid as i32
+            let cst = pg_sys::makeConst(
+                pg_sys::INT4OID,
+                -1,
+                pg_sys::InvalidOid,
+                4,
+                (self.rowid_typid.to_u32() as i32).into_datum().unwrap(),
+                false,
+                true,
+            );
+            ret.unstable_push_in_context(cst as _, mcx);
+
+            #[cfg(feature = "pg13")]
+            {
+                // [3] update_cols_count as i32
+                let cst = pg_sys::makeConst(
+                    pg_sys::INT4OID,
+                    -1,
+                    pg_sys::InvalidOid,
+                    4,
+                    (self.update_cols.len() as i32).into_datum().unwrap(),
+                    false,
+                    true,
+                );
+                ret.unstable_push_in_context(cst as _, mcx);
+
+                // [4..N] update_cols as TEXT
+                for col in &self.update_cols {
+                    let cst = pg_sys::makeConst(
+                        pg_sys::TEXTOID,
+                        -1,
+                        pg_sys::InvalidOid,
+                        -1,
+                        col.clone().into_datum().unwrap(),
+                        false,
+                        false,
+                    );
+                    ret.unstable_push_in_context(cst as _, mcx);
+                }
+            }
+
+            ret.into_ptr()
+        })
+    }
+
+    /// Deserialize FdwModifyPrivate from a PostgreSQL List of Const nodes.
+    unsafe fn deserialize_from_list(list: *mut pg_sys::List) -> Option<Self> {
+        pgrx::memcx::current_context(|mcx| {
+            let list = List::<*mut c_void>::downcast_ptr_in_memcx(list, mcx)?;
+
+            // [0] foreigntableid
+            let cst_ptr = *list.get(0)? as *mut pg_sys::Const;
+            let cst = *cst_ptr;
+            let foreigntableid_i32 = i32::from_datum(cst.constvalue, cst.constisnull)?;
+            let foreigntableid = Oid::from(foreigntableid_i32 as u32);
+
+            // [1] rowid_name
+            let cst_ptr = *list.get(1)? as *mut pg_sys::Const;
+            let cst = *cst_ptr;
+            let rowid_name = String::from_datum(cst.constvalue, cst.constisnull)?;
+
+            // [2] rowid_typid
+            let cst_ptr = *list.get(2)? as *mut pg_sys::Const;
+            let cst = *cst_ptr;
+            let rowid_typid_i32 = i32::from_datum(cst.constvalue, cst.constisnull)?;
+            let rowid_typid = Oid::from(rowid_typid_i32 as u32);
+
+            #[cfg(feature = "pg13")]
+            let update_cols = {
+                // [3] update_cols_count
+                let cst_ptr = *list.get(3)? as *mut pg_sys::Const;
+                let cst = *cst_ptr;
+                let count = i32::from_datum(cst.constvalue, cst.constisnull)? as usize;
+
+                // [4..N] update_cols
+                let mut cols = Vec::with_capacity(count);
+                for i in 0..count {
+                    let cst_ptr = *list.get(4 + i)? as *mut pg_sys::Const;
+                    let cst = *cst_ptr;
+                    let col = String::from_datum(cst.constvalue, cst.constisnull)?;
+                    cols.push(col);
+                }
+                cols
+            };
+
+            Some(FdwModifyPrivate {
+                foreigntableid,
+                rowid_name,
+                rowid_typid,
+                #[cfg(feature = "pg13")]
+                update_cols,
+            })
+        })
+    }
+}
+
 // Fdw private state for modify
 struct FdwModifyState<E: Into<ErrorReport>, W: ForeignDataWrapper<E>> {
     // foreign data wrapper instance
@@ -43,20 +193,6 @@ struct FdwModifyState<E: Into<ErrorReport>, W: ForeignDataWrapper<E>> {
 }
 
 impl<E: Into<ErrorReport>, W: ForeignDataWrapper<E>> FdwModifyState<E, W> {
-    unsafe fn new(foreigntableid: Oid, tmp_ctx: MemoryContext) -> Self {
-        Self {
-            instance: Some(instance::create_fdw_instance_from_table_id(foreigntableid)),
-            rowid_name: String::default(),
-            rowid_attno: 0,
-            rowid_typid: Oid::INVALID,
-            opts: HashMap::new(),
-            tmp_ctx,
-            _phantom: PhantomData,
-            #[cfg(feature = "pg13")]
-            update_cols: Vec::new(),
-        }
-    }
-
     fn begin_modify(&mut self) -> Result<(), E> {
         if let Some(ref mut instance) = self.instance {
             instance.begin_modify(&self.opts)
@@ -98,8 +234,6 @@ impl<E: Into<ErrorReport>, W: ForeignDataWrapper<E>> FdwModifyState<E, W> {
     }
 }
 
-impl<E: Into<ErrorReport>, W: ForeignDataWrapper<E>> utils::SerdeList for FdwModifyState<E, W> {}
-
 impl<E: Into<ErrorReport>, W: ForeignDataWrapper<E>> Drop for FdwModifyState<E, W> {
     fn drop(&mut self) {
         // drop foreign data wrapper instance
@@ -208,6 +342,7 @@ pub(super) extern "C-unwind" fn add_foreign_update_targets(
 }
 
 #[pg_guard]
+#[allow(clippy::extra_unused_type_parameters)]
 pub(super) extern "C-unwind" fn plan_foreign_modify<
     E: Into<ErrorReport>,
     W: ForeignDataWrapper<E>,
@@ -233,17 +368,7 @@ pub(super) extern "C-unwind" fn plan_foreign_modify<
         let rel = PgRelation::with_lock((*rte).relid, pg_sys::NoLock as _);
 
         let ftable = pg_sys::GetForeignTable(rel.oid());
-        let mut opts = options_to_hashmap((*ftable).options).report_unwrap();
-
-        // add additional metadata to the options
-        opts.insert(
-            "wrappers.fserver_oid".into(),
-            (*ftable).serverid.to_u32().to_string(),
-        );
-        opts.insert(
-            "wrappers.ftable_oid".into(),
-            (*ftable).relid.to_u32().to_string(),
-        );
+        let opts = options_to_hashmap((*ftable).options).report_unwrap();
 
         // check if the rowid column name is specified in table options
         let rowid_name = opts.get("rowid_column");
@@ -256,27 +381,17 @@ pub(super) extern "C-unwind" fn plan_foreign_modify<
         }
         let rowid_name = rowid_name.unwrap();
 
-        // search for rowid attribute in tuple descrition
+        // search for rowid attribute in tuple description
         let tup_desc = PgTupleDesc::from_relation(&rel);
         for attr in tup_desc.iter().filter(|a| !a.attisdropped) {
             let attname = pgrx::name_data_to_str(&attr.attname);
             if attname == rowid_name {
-                let ftable_id = rel.oid();
-
-                // create memory context for modify
-                let ctx_name = format!("Wrappers_modify_{}", ftable_id.to_u32());
-                let ctx = memctx::create_wrappers_memctx(&ctx_name);
-
-                // create modify state
-                let mut state = FdwModifyState::<E, W>::new(ftable_id, ctx);
-
-                state.rowid_name = rowid_name.to_string();
-                state.rowid_typid = attr.atttypid;
-                state.opts = opts;
+                let foreigntableid = rel.oid();
 
+                // Collect update columns for pg13
                 #[cfg(feature = "pg13")]
-                {
-                    // get update column list
+                let update_cols = {
+                    let mut cols = Vec::new();
                     let tgts: pgrx::PgList<pg_sys::TargetEntry> =
                         pgrx::PgList::from_pg((*(*root).parse).targetList);
                     for tgt in tgts.iter_ptr() {
@@ -285,22 +400,28 @@ pub(super) extern "C-unwind" fn plan_foreign_modify<
                             .unwrap()
                             .to_owned();
                         if !(*tgt).resjunk {
-                            state.update_cols.push(col_name);
+                            cols.push(col_name);
                         }
                     }
-                }
-
-                // box the modify state and 'serialize' state to a list, the state
-                // pointer is stored as an integer constant in the list, so it can be
-                // `deserialized` when executing the plan later.
-                // Note that the state itself is not serialized to any memory contexts,
-                // it just sits in Rust managed Box'ed memory and will be dropped when
-                // end_foreign_modify() is called.
-                return PgMemoryContexts::For(state.tmp_ctx).switch_to(|_| {
-                    let p = Box::leak(Box::new(state)) as *mut FdwModifyState<E, W>;
-                    let state = PgBox::<FdwModifyState<E, W>>::from_pg(p as _);
-                    FdwModifyState::serialize_to_list(state)
-                });
+                    cols
+                };
+
+                // Create FdwModifyPrivate with serializable data only.
+                // This data will survive PostgreSQL's query plan caching because
+                // we serialize actual data values, not pointers.
+                let private = FdwModifyPrivate {
+                    foreigntableid,
+                    rowid_name: rowid_name.to_string(),
+                    rowid_typid: attr.atttypid,
+                    #[cfg(feature = "pg13")]
+                    update_cols,
+                };
+
+                // Serialize the data to a PostgreSQL List.
+                // The actual FdwModifyState will be created in begin_foreign_modify
+                // for each execution, ensuring fresh state even when the query plan
+                // is cached and this planning stage is skipped.
+                return private.serialize_to_list();
             }
         }
 
@@ -331,8 +452,54 @@ pub(super) extern "C-unwind" fn begin_foreign_modify<
     }
 
     unsafe {
-        let mut state = FdwModifyState::<E, W>::deserialize_from_list(fdw_private as _);
-        assert!(!state.is_null());
+        // Deserialize FdwModifyPrivate from the cached fdw_private list.
+        // This contains the data needed to reconstruct the FdwModifyState.
+        let private = FdwModifyPrivate::deserialize_from_list(fdw_private);
+        if private.is_none() {
+            report_error(
+                PgSqlErrorCode::ERRCODE_FDW_ERROR,
+                "invalid fdw_private data in begin_foreign_modify",
+            );
+            return;
+        }
+        let private = private.unwrap();
+
+        // Create a fresh memory context for this execution.
+        // This ensures proper cleanup even when the query plan was cached.
+        let ctx_name = format!("Wrappers_modify_{}", private.foreigntableid.to_u32());
+        let tmp_ctx = memctx::create_wrappers_memctx(&ctx_name);
+
+        // Create a fresh FDW instance from the foreign table ID.
+        // This is done here (not in plan_foreign_modify) so that we always have
+        // a valid instance even when PostgreSQL reuses a cached query plan.
+        let fdw_instance: W = instance::create_fdw_instance_from_table_id(private.foreigntableid);
+
+        // Fetch foreign table options fresh for this execution
+        let ftable = pg_sys::GetForeignTable(private.foreigntableid);
+        let mut opts = options_to_hashmap((*ftable).options).report_unwrap();
+
+        // add additional metadata to the options
+        opts.insert(
+            "wrappers.fserver_oid".into(),
+            (*ftable).serverid.to_u32().to_string(),
+        );
+        opts.insert(
+            "wrappers.ftable_oid".into(),
+            (*ftable).relid.to_u32().to_string(),
+        );
+
+        // Create the FdwModifyState with fresh data
+        let mut state = FdwModifyState::<E, W> {
+            instance: Some(fdw_instance),
+            rowid_name: private.rowid_name,
+            rowid_attno: 0, // Will be set below
+            rowid_typid: private.rowid_typid,
+            opts,
+            tmp_ctx,
+            _phantom: PhantomData,
+            #[cfg(feature = "pg13")]
+            update_cols: private.update_cols,
+        };
 
         // search for rowid attribute number
         #[cfg(feature = "pg13")]
@@ -343,6 +510,10 @@ pub(super) extern "C-unwind" fn begin_foreign_modify<
         state.rowid_attno =
             pg_sys::ExecFindJunkAttributeInTlist((*subplan).targetlist, rowid_name_c);
 
+        // Box the state and call begin_modify
+        let state_ptr = Box::leak(Box::new(state));
+        let mut state = PgBox::<FdwModifyState<E, W>>::from_pg(state_ptr as _);
+
         let result = state.begin_modify();
         if result.is_err() {
             drop_fdw_modify_state(state.as_ptr());