Skip to content

Commit 1ebaf28

Browse files
crdv7crdv7
committed
Bound per-row memory in writable clauses
SET and REMOVE create EState-owned tuple slots for every updated entity and run transient builders in an executor-lifetime memory context. DELETE RLS slots have the same ownership problem, while MERGE SET shares the update path. Large writable statements therefore retain slots and scratch until executor shutdown. Use standalone slots for per-row writes and RLS checks, release them explicitly, and allocate update/delete scratch in the writable CustomScan's per-tuple context. Reset that context before each input row while retaining node-lifetime slots in the EState. Copy properties for paths retained by MERGE's cross-row de-duplication state into the query-lifetime context. Add eager multi-row MERGE coverage for ON CREATE SET and ON MATCH SET, plus terminal and non-terminal repeated-path coverage across row-context resets.
1 parent 8014174 commit 1ebaf28

5 files changed

Lines changed: 230 additions & 8 deletions

File tree

‎regress/expected/cypher_merge.out‎

Lines changed: 78 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2231,6 +2231,81 @@ $$) AS (src agtype, last agtype);
22312231
"Anchor" | "Anchor"
22322232
(1 row)
22332233

2234+
-- Verify that consecutive eager MERGE rows preserve row-specific composite
2235+
-- values across context resets in both ON CREATE and ON MATCH paths.
2236+
SELECT * FROM cypher('merge_actions', $$
2237+
UNWIND [1, 2, 3] AS ident
2238+
MERGE (n:Batch {id: ident})
2239+
ON CREATE SET n.payload = {nested: [ident, {active: true}]}
2240+
RETURN n.id, n.payload
2241+
$$) AS (id agtype, payload agtype);
2242+
id | payload
2243+
----+-----------------------------------
2244+
1 | {"nested": [1, {"active": true}]}
2245+
2 | {"nested": [2, {"active": true}]}
2246+
3 | {"nested": [3, {"active": true}]}
2247+
(3 rows)
2248+
2249+
SELECT * FROM cypher('merge_actions', $$
2250+
UNWIND [1, 2, 3] AS ident
2251+
MERGE (n:Batch {id: ident})
2252+
ON MATCH SET n.payload = {updated: [ident, false]}
2253+
RETURN n.id, n.payload
2254+
$$) AS (id agtype, payload agtype);
2255+
id | payload
2256+
----+-------------------------
2257+
1 | {"updated": [1, false]}
2258+
2 | {"updated": [2, false]}
2259+
3 | {"updated": [3, false]}
2260+
(3 rows)
2261+
2262+
-- A remembered MERGE path must retain its properties across row-context
2263+
-- resets so that a later duplicate reuses the first path.
2264+
SELECT * FROM cypher('merge_actions', $$
2265+
UNWIND [1, 2, 1] AS ident
2266+
MERGE (n:DedupBatch {
2267+
id: ident,
2268+
payload: {nested: [ident]}
2269+
})
2270+
RETURN ident, n.id, n.payload
2271+
$$) AS (input agtype, id agtype, payload agtype);
2272+
input | id | payload
2273+
-------+----+-----------------
2274+
1 | 1 | {"nested": [1]}
2275+
2 | 2 | {"nested": [2]}
2276+
1 | 1 | {"nested": [1]}
2277+
(3 rows)
2278+
2279+
SELECT * FROM cypher('merge_actions', $$
2280+
MATCH (n:DedupBatch)
2281+
RETURN count(n)
2282+
$$) AS (count agtype);
2283+
count
2284+
-------
2285+
2
2286+
(1 row)
2287+
2288+
-- Exercise the same retained-path lifetime in terminal MERGE.
2289+
SELECT * FROM cypher('merge_actions', $$
2290+
UNWIND [3, 4, 3] AS ident
2291+
MERGE (:DedupBatch {
2292+
id: ident,
2293+
payload: {nested: [ident]}
2294+
})
2295+
$$) AS (result agtype);
2296+
result
2297+
--------
2298+
(0 rows)
2299+
2300+
SELECT * FROM cypher('merge_actions', $$
2301+
MATCH (n:DedupBatch)
2302+
RETURN count(n)
2303+
$$) AS (count agtype);
2304+
count
2305+
-------
2306+
4
2307+
(1 row)
2308+
22342309
-- cleanup
22352310
SELECT * FROM cypher('merge_actions', $$ MATCH (n) DETACH DELETE n $$) AS (a agtype);
22362311
a
@@ -2241,12 +2316,14 @@ SELECT * FROM cypher('merge_actions', $$ MATCH (n) DETACH DELETE n $$) AS (a agt
22412316
-- delete graphs
22422317
--
22432318
SELECT drop_graph('merge_actions', true);
2244-
NOTICE: drop cascades to 5 other objects
2319+
NOTICE: drop cascades to 7 other objects
22452320
DETAIL: drop cascades to table merge_actions._ag_label_vertex
22462321
drop cascades to table merge_actions._ag_label_edge
22472322
drop cascades to table merge_actions."Person"
22482323
drop cascades to table merge_actions."KNOWS"
22492324
drop cascades to table merge_actions."on"
2325+
drop cascades to table merge_actions."Batch"
2326+
drop cascades to table merge_actions."DedupBatch"
22502327
NOTICE: graph "merge_actions" has been dropped
22512328
drop_graph
22522329
------------

‎regress/sql/cypher_merge.sql‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1085,6 +1085,52 @@ SELECT * FROM cypher('merge_actions', $$
10851085
RETURN b.source_name, b.last_seen_by
10861086
$$) AS (src agtype, last agtype);
10871087

1088+
-- Verify that consecutive eager MERGE rows preserve row-specific composite
1089+
-- values across context resets in both ON CREATE and ON MATCH paths.
1090+
SELECT * FROM cypher('merge_actions', $$
1091+
UNWIND [1, 2, 3] AS ident
1092+
MERGE (n:Batch {id: ident})
1093+
ON CREATE SET n.payload = {nested: [ident, {active: true}]}
1094+
RETURN n.id, n.payload
1095+
$$) AS (id agtype, payload agtype);
1096+
1097+
SELECT * FROM cypher('merge_actions', $$
1098+
UNWIND [1, 2, 3] AS ident
1099+
MERGE (n:Batch {id: ident})
1100+
ON MATCH SET n.payload = {updated: [ident, false]}
1101+
RETURN n.id, n.payload
1102+
$$) AS (id agtype, payload agtype);
1103+
1104+
-- A remembered MERGE path must retain its properties across row-context
1105+
-- resets so that a later duplicate reuses the first path.
1106+
SELECT * FROM cypher('merge_actions', $$
1107+
UNWIND [1, 2, 1] AS ident
1108+
MERGE (n:DedupBatch {
1109+
id: ident,
1110+
payload: {nested: [ident]}
1111+
})
1112+
RETURN ident, n.id, n.payload
1113+
$$) AS (input agtype, id agtype, payload agtype);
1114+
1115+
SELECT * FROM cypher('merge_actions', $$
1116+
MATCH (n:DedupBatch)
1117+
RETURN count(n)
1118+
$$) AS (count agtype);
1119+
1120+
-- Exercise the same retained-path lifetime in terminal MERGE.
1121+
SELECT * FROM cypher('merge_actions', $$
1122+
UNWIND [3, 4, 3] AS ident
1123+
MERGE (:DedupBatch {
1124+
id: ident,
1125+
payload: {nested: [ident]}
1126+
})
1127+
$$) AS (result agtype);
1128+
1129+
SELECT * FROM cypher('merge_actions', $$
1130+
MATCH (n:DedupBatch)
1131+
RETURN count(n)
1132+
$$) AS (count agtype);
1133+
10881134
-- cleanup
10891135
SELECT * FROM cypher('merge_actions', $$ MATCH (n) DETACH DELETE n $$) AS (a agtype);
10901136

‎src/backend/executor/cypher_delete.c‎

Lines changed: 29 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -149,6 +149,9 @@ static TupleTableSlot *exec_cypher_delete(CustomScanState *node)
149149
*/
150150
while(true)
151151
{
152+
/* Release scratch retained by the preceding input row. */
153+
ResetExprContext(econtext);
154+
152155
/* Process the subtree first */
153156
Decrement_Estate_CommandId(estate)
154157
slot = ExecProcNode(node->ss.ps.lefttree);
@@ -168,6 +171,9 @@ static TupleTableSlot *exec_cypher_delete(CustomScanState *node)
168171
}
169172
else
170173
{
174+
/* Release scratch retained by the preceding input row. */
175+
ResetExprContext(econtext);
176+
171177
/* Process the subtree first */
172178
Decrement_Estate_CommandId(estate)
173179
slot = ExecProcNode(node->ss.ps.lefttree);
@@ -386,6 +392,10 @@ static void process_delete_list(CustomScanState *node)
386392
HASHCTL hashctl;
387393
HTAB *index_cache = NULL;
388394
HASHCTL idx_hashctl;
395+
MemoryContext old_context;
396+
397+
/* Keep transient delete state in the current input row's context. */
398+
old_context = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
389399

390400
/* Hash table for caching compiled security quals per label */
391401
MemSet(&hashctl, 0, sizeof(hashctl));
@@ -517,7 +527,8 @@ static void process_delete_list(CustomScanState *node)
517527
if (!found_rls)
518528
{
519529
entry->qualExprs = setup_security_quals(resultRelInfo, estate, node, CMD_DELETE);
520-
entry->slot = ExecInitExtraTupleSlot(estate, RelationGetDescr(rel), &TTSOpsHeapTuple);
530+
entry->slot = MakeSingleTupleTableSlot(
531+
RelationGetDescr(rel), &TTSOpsHeapTuple);
521532
}
522533

523534
ExecStoreHeapTuple(heap_tuple, entry->slot, false);
@@ -566,9 +577,26 @@ static void process_delete_list(CustomScanState *node)
566577
destroy_entity_result_rel_info(resultRelInfo);
567578
}
568579

580+
/* Release standalone RLS slots before destroying their owning cache. */
581+
{
582+
HASH_SEQ_STATUS seq;
583+
RLSCacheEntry *entry;
584+
585+
hash_seq_init(&seq, qual_cache);
586+
while ((entry = (RLSCacheEntry *)hash_seq_search(&seq)) != NULL)
587+
{
588+
if (entry->slot != NULL)
589+
{
590+
ExecDropSingleTupleTableSlot(entry->slot);
591+
}
592+
}
593+
}
594+
569595
/* Clean up the cache */
570596
hash_destroy(qual_cache);
571597
hash_destroy(index_cache);
598+
599+
MemoryContextSwitchTo(old_context);
572600
}
573601

574602
/*

‎src/backend/executor/cypher_merge.c‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,8 @@ static bool compare_2_paths(path_entry **lhs, path_entry **rhs,
9797
static path_entry **find_duplicate_path(CustomScanState *node,
9898
path_entry **path_array);
9999
static void free_path_entry_array(path_entry **path_array, int length);
100+
static void preserve_path_properties(path_entry **path_array, int length,
101+
MemoryContext memory_context);
100102

101103
/*
102104
* 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)
428430
}
429431
}
430432

433+
/*
434+
* created_paths_list retains one entry per unique path until node shutdown.
435+
* Copy row-local properties only for paths added to that list.
436+
*/
437+
static void preserve_path_properties(path_entry **path_array, int length,
438+
MemoryContext memory_context)
439+
{
440+
MemoryContext old_context;
441+
int index;
442+
443+
old_context = MemoryContextSwitchTo(memory_context);
444+
445+
for (index = 0; index < length; index++)
446+
{
447+
path_entry *entry = path_array[index];
448+
449+
if (!entry->actual && !entry->prop_isNull)
450+
{
451+
entry->prop = datumCopy(entry->prop, false, -1);
452+
}
453+
}
454+
455+
MemoryContextSwitchTo(old_context);
456+
}
457+
431458
/*
432459
* Helper function to prebuild a path. The user needs to free the returned
433460
* path_entry when done.
@@ -690,6 +717,9 @@ static TupleTableSlot *exec_cypher_merge(CustomScanState *node)
690717
TupleTableSlot *projected;
691718
HeapTuple htup;
692719

720+
/* Release scratch retained by the preceding input row. */
721+
ResetExprContext(econtext);
722+
693723
/* Process the subtree first */
694724
Decrement_Estate_CommandId(estate)
695725
slot = ExecProcNode(node->ss.ps.lefttree);
@@ -737,6 +767,9 @@ static TupleTableSlot *exec_cypher_merge(CustomScanState *node)
737767
created_path *new_path =
738768
palloc0(sizeof(created_path));
739769

770+
preserve_path_properties(
771+
prebuilt_path_array, path_length,
772+
estate->es_query_cxt);
740773
new_path->next = css->created_paths_list;
741774
new_path->entry = prebuilt_path_array;
742775
css->created_paths_list = new_path;
@@ -795,6 +828,9 @@ static TupleTableSlot *exec_cypher_merge(CustomScanState *node)
795828
*/
796829
do
797830
{
831+
/* Release scratch retained by the preceding input row. */
832+
ResetExprContext(econtext);
833+
798834
/* Process the subtree first */
799835
Decrement_Estate_CommandId(estate)
800836
slot = ExecProcNode(node->ss.ps.lefttree);
@@ -837,6 +873,8 @@ static TupleTableSlot *exec_cypher_merge(CustomScanState *node)
837873
{
838874
created_path *new_path = palloc0(sizeof(created_path));
839875

876+
preserve_path_properties(prebuilt_path_array, path_length,
877+
estate->es_query_cxt);
840878
new_path->next = css->created_paths_list;
841879
new_path->entry = prebuilt_path_array;
842880
css->created_paths_list = new_path;
@@ -912,6 +950,7 @@ static TupleTableSlot *exec_cypher_merge(CustomScanState *node)
912950
* Process the subtree. The subtree will only consist of the MERGE
913951
* path.
914952
*/
953+
ResetExprContext(econtext);
915954
Decrement_Estate_CommandId(estate)
916955
slot = ExecProcNode(node->ss.ps.lefttree);
917956
Increment_Estate_CommandId(estate)

0 commit comments

Comments
 (0)