Bound per-row memory in writable clauses

regress/expected/cypher_merge.out

@@ -2231,6 +2231,81 @@ $$) AS (src agtype, last agtype);
  "Anchor" | "Anchor"
 (1 row)
 
+-- Verify that consecutive eager MERGE rows preserve row-specific composite
+-- values across context resets in both ON CREATE and ON MATCH paths.
+SELECT * FROM cypher('merge_actions', $$
+  UNWIND [1, 2, 3] AS ident
+  MERGE (n:Batch {id: ident})
+    ON CREATE SET n.payload = {nested: [ident, {active: true}]}
+  RETURN n.id, n.payload
+$$) AS (id agtype, payload agtype);
+ id |              payload              
+----+-----------------------------------
+ 1  | {"nested": [1, {"active": true}]}
+ 2  | {"nested": [2, {"active": true}]}
+ 3  | {"nested": [3, {"active": true}]}
+(3 rows)
+
+SELECT * FROM cypher('merge_actions', $$
+  UNWIND [1, 2, 3] AS ident
+  MERGE (n:Batch {id: ident})
+    ON MATCH SET n.payload = {updated: [ident, false]}
+  RETURN n.id, n.payload
+$$) AS (id agtype, payload agtype);
+ id |         payload         
+----+-------------------------
+ 1  | {"updated": [1, false]}
+ 2  | {"updated": [2, false]}
+ 3  | {"updated": [3, false]}
+(3 rows)
+
+-- A remembered MERGE path must retain its properties across row-context
+-- resets so that a later duplicate reuses the first path.
+SELECT * FROM cypher('merge_actions', $$
+  UNWIND [1, 2, 1] AS ident
+  MERGE (n:DedupBatch {
+    id: ident,
+    payload: {nested: [ident]}
+  })
+  RETURN ident, n.id, n.payload
+$$) AS (input agtype, id agtype, payload agtype);
+ input | id |     payload     
+-------+----+-----------------
+ 1     | 1  | {"nested": [1]}
+ 2     | 2  | {"nested": [2]}
+ 1     | 1  | {"nested": [1]}
+(3 rows)
+
+SELECT * FROM cypher('merge_actions', $$
+  MATCH (n:DedupBatch)
+  RETURN count(n)
+$$) AS (count agtype);
+ count 
+-------
+ 2
+(1 row)
+
+-- Exercise the same retained-path lifetime in terminal MERGE.
+SELECT * FROM cypher('merge_actions', $$
+  UNWIND [3, 4, 3] AS ident
+  MERGE (:DedupBatch {
+    id: ident,
+    payload: {nested: [ident]}
+  })
+$$) AS (result agtype);
+ result 
+--------
+(0 rows)
+
+SELECT * FROM cypher('merge_actions', $$
+  MATCH (n:DedupBatch)
+  RETURN count(n)
+$$) AS (count agtype);
+ count 
+-------
+ 4
+(1 row)
+
 -- cleanup
 SELECT * FROM cypher('merge_actions', $$ MATCH (n) DETACH DELETE n $$) AS (a agtype);
  a 
@@ -2241,12 +2316,14 @@ SELECT * FROM cypher('merge_actions', $$ MATCH (n) DETACH DELETE n $$) AS (a agt
 -- delete graphs
 --
 SELECT drop_graph('merge_actions', true);
-NOTICE:  drop cascades to 5 other objects
+NOTICE:  drop cascades to 7 other objects
 DETAIL:  drop cascades to table merge_actions._ag_label_vertex
 drop cascades to table merge_actions._ag_label_edge
 drop cascades to table merge_actions."Person"
 drop cascades to table merge_actions."KNOWS"
 drop cascades to table merge_actions."on"
+drop cascades to table merge_actions."Batch"
+drop cascades to table merge_actions."DedupBatch"
 NOTICE:  graph "merge_actions" has been dropped
  drop_graph 
 ------------

regress/sql/cypher_merge.sql

@@ -1085,6 +1085,52 @@ SELECT * FROM cypher('merge_actions', $$
   RETURN b.source_name, b.last_seen_by
 $$) AS (src agtype, last agtype);
 
+-- Verify that consecutive eager MERGE rows preserve row-specific composite
+-- values across context resets in both ON CREATE and ON MATCH paths.
+SELECT * FROM cypher('merge_actions', $$
+  UNWIND [1, 2, 3] AS ident
+  MERGE (n:Batch {id: ident})
+    ON CREATE SET n.payload = {nested: [ident, {active: true}]}
+  RETURN n.id, n.payload
+$$) AS (id agtype, payload agtype);
+
+SELECT * FROM cypher('merge_actions', $$
+  UNWIND [1, 2, 3] AS ident
+  MERGE (n:Batch {id: ident})
+    ON MATCH SET n.payload = {updated: [ident, false]}
+  RETURN n.id, n.payload
+$$) AS (id agtype, payload agtype);
+
+-- A remembered MERGE path must retain its properties across row-context
+-- resets so that a later duplicate reuses the first path.
+SELECT * FROM cypher('merge_actions', $$
+  UNWIND [1, 2, 1] AS ident
+  MERGE (n:DedupBatch {
+    id: ident,
+    payload: {nested: [ident]}
+  })
+  RETURN ident, n.id, n.payload
+$$) AS (input agtype, id agtype, payload agtype);
+
+SELECT * FROM cypher('merge_actions', $$
+  MATCH (n:DedupBatch)
+  RETURN count(n)
+$$) AS (count agtype);
+
+-- Exercise the same retained-path lifetime in terminal MERGE.
+SELECT * FROM cypher('merge_actions', $$
+  UNWIND [3, 4, 3] AS ident
+  MERGE (:DedupBatch {
+    id: ident,
+    payload: {nested: [ident]}
+  })
+$$) AS (result agtype);
+
+SELECT * FROM cypher('merge_actions', $$
+  MATCH (n:DedupBatch)
+  RETURN count(n)
+$$) AS (count agtype);
+
 -- cleanup
 SELECT * FROM cypher('merge_actions', $$ MATCH (n) DETACH DELETE n $$) AS (a agtype);
 

src/backend/executor/cypher_delete.c

@@ -149,6 +149,9 @@ static TupleTableSlot *exec_cypher_delete(CustomScanState *node)
          */
         while(true)
         {
+            /* Release scratch retained by the preceding input row. */
+            ResetExprContext(econtext);
+
             /* Process the subtree first */
             Decrement_Estate_CommandId(estate)
             slot = ExecProcNode(node->ss.ps.lefttree);
@@ -168,6 +171,9 @@ static TupleTableSlot *exec_cypher_delete(CustomScanState *node)
     }
     else
     {
+        /* Release scratch retained by the preceding input row. */
+        ResetExprContext(econtext);
+
         /* Process the subtree first */
         Decrement_Estate_CommandId(estate)
         slot = ExecProcNode(node->ss.ps.lefttree);
@@ -386,6 +392,10 @@ static void process_delete_list(CustomScanState *node)
     HASHCTL hashctl;
     HTAB *index_cache = NULL;
     HASHCTL idx_hashctl;
+    MemoryContext old_context;
+
+    /* Keep transient delete state in the current input row's context. */
+    old_context = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
 
     /* Hash table for caching compiled security quals per label */
     MemSet(&hashctl, 0, sizeof(hashctl));
@@ -517,7 +527,8 @@ static void process_delete_list(CustomScanState *node)
                 if (!found_rls)
                 {
                     entry->qualExprs = setup_security_quals(resultRelInfo, estate, node, CMD_DELETE);
-                    entry->slot = ExecInitExtraTupleSlot(estate, RelationGetDescr(rel), &TTSOpsHeapTuple);
+                    entry->slot = MakeSingleTupleTableSlot(
+                        RelationGetDescr(rel), &TTSOpsHeapTuple);
                 }
 
                 ExecStoreHeapTuple(heap_tuple, entry->slot, false);
@@ -566,9 +577,26 @@ static void process_delete_list(CustomScanState *node)
         destroy_entity_result_rel_info(resultRelInfo);
     }
 
+    /* Release standalone RLS slots before destroying their owning cache. */
+    {
+        HASH_SEQ_STATUS seq;
+        RLSCacheEntry *entry;
+
+        hash_seq_init(&seq, qual_cache);
+        while ((entry = (RLSCacheEntry *)hash_seq_search(&seq)) != NULL)
+        {
+            if (entry->slot != NULL)
+            {
+                ExecDropSingleTupleTableSlot(entry->slot);
+            }
+        }
+    }
+
     /* Clean up the cache */
     hash_destroy(qual_cache);
     hash_destroy(index_cache);
+
+    MemoryContextSwitchTo(old_context);
 }
 
 /*

src/backend/executor/cypher_merge.c

@@ -97,6 +97,8 @@ static bool compare_2_paths(path_entry **lhs, path_entry **rhs,
 static path_entry **find_duplicate_path(CustomScanState *node,
                                         path_entry **path_array);
 static void free_path_entry_array(path_entry **path_array, int length);
+static void preserve_path_properties(path_entry **path_array, int length,
+                                     MemoryContext memory_context);
 
 /*
  * Initializes the MERGE Execution Node at the beginning of the execution
@@ -428,6 +430,31 @@ static void free_path_entry_array(path_entry **path_array, int length)
     }
 }
 
+/*
+ * created_paths_list retains one entry per unique path until node shutdown.
+ * Copy row-local properties only for paths added to that list.
+ */
+static void preserve_path_properties(path_entry **path_array, int length,
+                                     MemoryContext memory_context)
+{
+    MemoryContext old_context;
+    int index;
+
+    old_context = MemoryContextSwitchTo(memory_context);
+
+    for (index = 0; index < length; index++)
+    {
+        path_entry *entry = path_array[index];
+
+        if (!entry->actual && !entry->prop_isNull)
+        {
+            entry->prop = datumCopy(entry->prop, false, -1);
+        }
+    }
+
+    MemoryContextSwitchTo(old_context);
+}
+
 /*
  * Helper function to prebuild a path. The user needs to free the returned
  * path_entry when done.
@@ -690,6 +717,9 @@ static TupleTableSlot *exec_cypher_merge(CustomScanState *node)
                 TupleTableSlot *projected;
                 HeapTuple htup;
 
+                /* Release scratch retained by the preceding input row. */
+                ResetExprContext(econtext);
+
                 /* Process the subtree first */
                 Decrement_Estate_CommandId(estate)
                 slot = ExecProcNode(node->ss.ps.lefttree);
@@ -737,6 +767,9 @@ static TupleTableSlot *exec_cypher_merge(CustomScanState *node)
                         created_path *new_path =
                             palloc0(sizeof(created_path));
 
+                        preserve_path_properties(
+                            prebuilt_path_array, path_length,
+                            estate->es_query_cxt);
                         new_path->next = css->created_paths_list;
                         new_path->entry = prebuilt_path_array;
                         css->created_paths_list = new_path;
@@ -795,6 +828,9 @@ static TupleTableSlot *exec_cypher_merge(CustomScanState *node)
          */
         do
         {
+            /* Release scratch retained by the preceding input row. */
+            ResetExprContext(econtext);
+
             /* Process the subtree first */
             Decrement_Estate_CommandId(estate)
             slot = ExecProcNode(node->ss.ps.lefttree);
@@ -837,6 +873,8 @@ static TupleTableSlot *exec_cypher_merge(CustomScanState *node)
                 {
                     created_path *new_path = palloc0(sizeof(created_path));
 
+                    preserve_path_properties(prebuilt_path_array, path_length,
+                                             estate->es_query_cxt);
                     new_path->next = css->created_paths_list;
                     new_path->entry = prebuilt_path_array;
                     css->created_paths_list = new_path;
@@ -912,6 +950,7 @@ static TupleTableSlot *exec_cypher_merge(CustomScanState *node)
          * Process the subtree. The subtree will only consist of the MERGE
          * path.
          */
+        ResetExprContext(econtext);
         Decrement_Estate_CommandId(estate)
         slot = ExecProcNode(node->ss.ps.lefttree);
         Increment_Estate_CommandId(estate)