From 560609e8092b0050dd026cb01e84fb2e6a1aff2a Mon Sep 17 00:00:00 2001 From: crdv7 <15974123760@163.com> Date: Tue, 11 Aug 2026 16:33:05 +0800 Subject: [PATCH] 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 both CustomScan and EState per-tuple contexts at the corresponding row boundaries while retaining node-lifetime slots in the EState. Initialize stored and virtual generated-column metadata with the same lifetime as the one-row ResultRelInfo used by SET, REMOVE, and DELETE. Copy properties for paths retained by MERGE's cross-row de-duplication state into the query-lifetime context, and fully release rejected path arrays. Reset connected-edge RLS expression scratch for each edge in both DETACH DELETE scan paths. Add generated-column batch and DELETE coverage, eager multi-row MERGE coverage, terminal and non-terminal repeated-path coverage, and function-based connected-edge RLS coverage for both endpoint indexes and the sequential-scan fallback. --- regress/expected/cypher_merge.out | 111 ++++++++++++++++++- regress/expected/generated_columns.out | 144 ++++++++++++++++++++++++- regress/expected/security.out | 85 +++++++++++++++ regress/sql/cypher_merge.sql | 67 ++++++++++++ regress/sql/generated_columns.sql | 77 +++++++++++++ regress/sql/security.sql | 64 +++++++++++ src/backend/executor/cypher_create.c | 3 + src/backend/executor/cypher_delete.c | 44 +++++++- src/backend/executor/cypher_merge.c | 49 ++++++++- src/backend/executor/cypher_set.c | 54 ++++++++-- src/backend/executor/cypher_utils.c | 47 ++++++++ src/include/executor/cypher_utils.h | 2 + 12 files changed, 734 insertions(+), 13 deletions(-) diff --git a/regress/expected/cypher_merge.out b/regress/expected/cypher_merge.out index ac08a971a..21f79c6f4 100644 --- a/regress/expected/cypher_merge.out +++ b/regress/expected/cypher_merge.out @@ -2231,6 +2231,113 @@ $$) AS (src agtype, last agtype); "Anchor" | "Anchor" (1 row) +-- Exercise eager MERGE ON CREATE and ON MATCH SET over consecutive rows. +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) + +-- Reuse a path created by an earlier input row. +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) + +-- Reuse a path created by an earlier input row 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) + +-- Exercise repeated duplicate candidates in eager and terminal MERGE. +SELECT * FROM cypher('merge_actions', $$ + UNWIND range(1, 1000) AS ident + MERGE (n:DedupBatch { + id: 20, + payload: {nested: [20]} + }) + RETURN count(n) +$$) AS (count agtype); + count +------- + 1000 +(1 row) + +SELECT * FROM cypher('merge_actions', $$ + UNWIND range(1, 1000) AS ident + MERGE (:DedupBatch { + id: 21, + payload: {nested: [21]} + }) +$$) AS (result agtype); + result +-------- +(0 rows) + +SELECT * FROM cypher('merge_actions', $$ + MATCH (n:DedupBatch) + RETURN count(n) +$$) AS (count agtype); + count +------- + 6 +(1 row) + -- cleanup SELECT * FROM cypher('merge_actions', $$ MATCH (n) DETACH DELETE n $$) AS (a agtype); a @@ -2241,12 +2348,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 ------------ diff --git a/regress/expected/generated_columns.out b/regress/expected/generated_columns.out index 4e929110c..24f478317 100644 --- a/regress/expected/generated_columns.out +++ b/regress/expected/generated_columns.out @@ -104,6 +104,58 @@ SELECT category, properties FROM generated_columns."Product" ORDER BY category N | {"type": "no-cat"} (3 rows) +-- Exercise generated columns through multi-row CREATE, SET, and MERGE. +SELECT * FROM cypher('generated_columns', $$ + UNWIND ['batch-a', 'batch-b', 'batch-c'] AS category + CREATE (p:Product {category: category, batch: 'create'}) + RETURN count(p) +$$) AS (count agtype); + count +------- + 3 +(1 row) + +SELECT * FROM cypher('generated_columns', $$ + MATCH (p:Product) + SET p.checked = true + RETURN count(p) +$$) AS (count agtype); + count +------- + 6 +(1 row) + +SELECT * FROM cypher('generated_columns', $$ + UNWIND ['merge-a', 'merge-b', 'merge-c'] AS category + MERGE (p:Product {category: category}) + ON CREATE SET p.batch = 'merge-create' + RETURN count(p) +$$) AS (count agtype); + count +------- + 3 +(1 row) + +SELECT * FROM cypher('generated_columns', $$ + UNWIND ['merge-a', 'merge-b', 'merge-c'] AS category + MERGE (p:Product {category: category}) + ON MATCH SET p.batch = 'merge-match' + RETURN count(p) +$$) AS (count agtype); + count +------- + 3 +(1 row) + +SELECT count(*) AS rows, + bool_and(category IS NOT DISTINCT FROM + agtype_access_operator(properties, '"category"')::varchar(25)) AS synchronized +FROM generated_columns."Product"; + rows | synchronized +------+-------------- + 12 | t +(1 row) + -- -- GENERATED ALWAYS ... STORED column on an edge label -- @@ -123,7 +175,7 @@ SELECT * FROM cypher('generated_columns', $$ $$) AS (r agtype); r ---------------------------------------------------------------------------------------------------------------------------------------- - {"id": 1125899906842625, "label": "REL", "end_id": 844424930131973, "start_id": 844424930131972, "properties": {"kind": "link"}}::edge + {"id": 1125899906842625, "label": "REL", "end_id": 844424930131982, "start_id": 844424930131981, "properties": {"kind": "link"}}::edge (1 row) SELECT kind FROM generated_columns."REL"; @@ -203,17 +255,105 @@ FROM generated_columns."T"; t (1 row) +-- Exercise SET and DELETE on stored and virtual generated columns. +SELECT create_vlabel('generated_columns', 'DeleteStored'); +NOTICE: VLabel "DeleteStored" has been created + create_vlabel +--------------- + +(1 row) + +ALTER TABLE generated_columns."DeleteStored" + ADD COLUMN marker integer GENERATED ALWAYS AS (1) STORED; +SELECT * FROM cypher('generated_columns', $$ + UNWIND [1, 2, 3] AS ident + CREATE (n:DeleteStored {id: ident}) + RETURN count(n) +$$) AS (count agtype); + count +------- + 3 +(1 row) + +SELECT * FROM cypher('generated_columns', $$ + MATCH (n:DeleteStored) + DELETE n +$$) AS (n agtype); + n +--- +(0 rows) + +SELECT count(*) FROM generated_columns."DeleteStored"; + count +------- + 0 +(1 row) + +SELECT create_vlabel('generated_columns', 'DeleteVirtual'); +NOTICE: VLabel "DeleteVirtual" has been created + create_vlabel +--------------- + +(1 row) + +ALTER TABLE generated_columns."DeleteVirtual" + ADD COLUMN marker integer GENERATED ALWAYS AS (1) VIRTUAL; +SELECT * FROM cypher('generated_columns', $$ + UNWIND [1, 2, 3] AS ident + CREATE (n:DeleteVirtual {id: ident}) + RETURN count(n) +$$) AS (count agtype); + count +------- + 3 +(1 row) + +SELECT * FROM cypher('generated_columns', $$ + MATCH (n:DeleteVirtual) + SET n.checked = true + RETURN count(n) +$$) AS (count agtype); + count +------- + 3 +(1 row) + +SELECT bool_and(marker = 1) AS virtual_ok +FROM generated_columns."DeleteVirtual"; + virtual_ok +------------ + t +(1 row) + +SELECT * FROM cypher('generated_columns', $$ + MATCH (n:DeleteVirtual) + DELETE n + RETURN count(n) +$$) AS (count agtype); + count +------- + 3 +(1 row) + +SELECT count(*) FROM generated_columns."DeleteVirtual"; + count +------- + 0 +(1 row) + -- -- Cleanup -- SELECT drop_graph('generated_columns', true); -NOTICE: drop cascades to 6 other objects +NOTICE: drop cascades to 8 other objects DETAIL: drop cascades to table generated_columns._ag_label_vertex drop cascades to table generated_columns._ag_label_edge drop cascades to table generated_columns."Product" drop cascades to table generated_columns."REL" drop cascades to table generated_columns."Plain" drop cascades to table generated_columns."T" +drop cascades to table generated_columns."DeleteStored" +drop cascades to table generated_columns."DeleteVirtual" NOTICE: graph "generated_columns" has been dropped drop_graph ------------ diff --git a/regress/expected/security.out b/regress/expected/security.out index 521b76b62..e700e8a50 100644 --- a/regress/expected/security.out +++ b/regress/expected/security.out @@ -1440,6 +1440,91 @@ $$) AS (since agtype); ------- (0 rows) +-- Exercise function-based edge RLS through the default endpoint indexes. +SELECT * FROM cypher('rls_graph', $$ + CREATE (:Person {name: 'DetachFnIndexHub', owner: 'rls_user1', department: 'DetachFn'}) +$$) AS (a agtype); + a +--- +(0 rows) + +SELECT * FROM cypher('rls_graph', $$ + MATCH (hub:Person {name: 'DetachFnIndexHub'}) + UNWIND range(1, 16) AS ident + CREATE (hub)-[:KNOWS {owner: 'rls_user1'}]-> + (:Person {owner: 'rls_user1', department: 'DetachFn', leaf: ident}) +$$) AS (a agtype); + a +--- +(0 rows) + +SET ROLE rls_user1; +SELECT * FROM cypher('rls_graph', $$ + MATCH (p:Person {name: 'DetachFnIndexHub'}) DETACH DELETE p +$$) AS (a agtype); + a +--- +(0 rows) + +RESET ROLE; +SELECT * FROM cypher('rls_graph', $$ + MATCH ()-[k:KNOWS]->() WHERE k.owner = 'rls_user1' RETURN count(k) +$$) AS (count agtype); + count +------- + 0 +(1 row) + +-- Drop the endpoint indexes to exercise the connected-edge sequential scan. +DO $$ +DECLARE + index_to_drop regclass; +BEGIN + FOR index_to_drop IN + SELECT indexrelid + FROM pg_index + WHERE indrelid = 'rls_graph."KNOWS"'::regclass + AND indnatts = 1 + AND indkey[0] IN (2, 3) + LOOP + EXECUTE format('DROP INDEX %s', index_to_drop); + END LOOP; +END +$$; +SELECT * FROM cypher('rls_graph', $$ + CREATE (:Person {name: 'DetachFnSeqHub', owner: 'rls_user1', department: 'DetachFn'}) +$$) AS (a agtype); + a +--- +(0 rows) + +SELECT * FROM cypher('rls_graph', $$ + MATCH (hub:Person {name: 'DetachFnSeqHub'}) + UNWIND range(1, 16) AS ident + CREATE (hub)-[:KNOWS {owner: 'rls_user1'}]-> + (:Person {owner: 'rls_user1', department: 'DetachFn', seq_leaf: ident}) +$$) AS (a agtype); + a +--- +(0 rows) + +SET ROLE rls_user1; +SELECT * FROM cypher('rls_graph', $$ + MATCH (p:Person {name: 'DetachFnSeqHub'}) DETACH DELETE p +$$) AS (a agtype); + a +--- +(0 rows) + +RESET ROLE; +SELECT * FROM cypher('rls_graph', $$ + MATCH ()-[k:KNOWS]->() WHERE k.owner = 'rls_user1' RETURN count(k) +$$) AS (count agtype); + count +------- + 0 +(1 row) + -- cleanup DROP POLICY detach_fn_knows_owner ON rls_graph."KNOWS"; DROP POLICY detach_fn_person_all ON rls_graph."Person"; diff --git a/regress/sql/cypher_merge.sql b/regress/sql/cypher_merge.sql index 86b3e0235..c1a5c72f9 100644 --- a/regress/sql/cypher_merge.sql +++ b/regress/sql/cypher_merge.sql @@ -1085,6 +1085,73 @@ SELECT * FROM cypher('merge_actions', $$ RETURN b.source_name, b.last_seen_by $$) AS (src agtype, last agtype); +-- Exercise eager MERGE ON CREATE and ON MATCH SET over consecutive rows. +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); + +-- Reuse a path created by an earlier input row. +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); + +-- Reuse a path created by an earlier input row 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); + +-- Exercise repeated duplicate candidates in eager and terminal MERGE. +SELECT * FROM cypher('merge_actions', $$ + UNWIND range(1, 1000) AS ident + MERGE (n:DedupBatch { + id: 20, + payload: {nested: [20]} + }) + RETURN count(n) +$$) AS (count agtype); + +SELECT * FROM cypher('merge_actions', $$ + UNWIND range(1, 1000) AS ident + MERGE (:DedupBatch { + id: 21, + payload: {nested: [21]} + }) +$$) 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); diff --git a/regress/sql/generated_columns.sql b/regress/sql/generated_columns.sql index 5229c28f1..7dc37f946 100644 --- a/regress/sql/generated_columns.sql +++ b/regress/sql/generated_columns.sql @@ -71,6 +71,38 @@ $$) AS (p agtype); -- The stored generated column always mirrors properties.category SELECT category, properties FROM generated_columns."Product" ORDER BY category NULLS LAST; +-- Exercise generated columns through multi-row CREATE, SET, and MERGE. +SELECT * FROM cypher('generated_columns', $$ + UNWIND ['batch-a', 'batch-b', 'batch-c'] AS category + CREATE (p:Product {category: category, batch: 'create'}) + RETURN count(p) +$$) AS (count agtype); + +SELECT * FROM cypher('generated_columns', $$ + MATCH (p:Product) + SET p.checked = true + RETURN count(p) +$$) AS (count agtype); + +SELECT * FROM cypher('generated_columns', $$ + UNWIND ['merge-a', 'merge-b', 'merge-c'] AS category + MERGE (p:Product {category: category}) + ON CREATE SET p.batch = 'merge-create' + RETURN count(p) +$$) AS (count agtype); + +SELECT * FROM cypher('generated_columns', $$ + UNWIND ['merge-a', 'merge-b', 'merge-c'] AS category + MERGE (p:Product {category: category}) + ON MATCH SET p.batch = 'merge-match' + RETURN count(p) +$$) AS (count agtype); + +SELECT count(*) AS rows, + bool_and(category IS NOT DISTINCT FROM + agtype_access_operator(properties, '"category"')::varchar(25)) AS synchronized +FROM generated_columns."Product"; + -- -- GENERATED ALWAYS ... STORED column on an edge label -- @@ -123,6 +155,51 @@ $$) AS (n agtype); SELECT tbl = 'generated_columns."T"'::regclass::oid AS tableoid_ok FROM generated_columns."T"; +-- Exercise SET and DELETE on stored and virtual generated columns. +SELECT create_vlabel('generated_columns', 'DeleteStored'); +ALTER TABLE generated_columns."DeleteStored" + ADD COLUMN marker integer GENERATED ALWAYS AS (1) STORED; + +SELECT * FROM cypher('generated_columns', $$ + UNWIND [1, 2, 3] AS ident + CREATE (n:DeleteStored {id: ident}) + RETURN count(n) +$$) AS (count agtype); + +SELECT * FROM cypher('generated_columns', $$ + MATCH (n:DeleteStored) + DELETE n +$$) AS (n agtype); + +SELECT count(*) FROM generated_columns."DeleteStored"; + +SELECT create_vlabel('generated_columns', 'DeleteVirtual'); +ALTER TABLE generated_columns."DeleteVirtual" + ADD COLUMN marker integer GENERATED ALWAYS AS (1) VIRTUAL; + +SELECT * FROM cypher('generated_columns', $$ + UNWIND [1, 2, 3] AS ident + CREATE (n:DeleteVirtual {id: ident}) + RETURN count(n) +$$) AS (count agtype); + +SELECT * FROM cypher('generated_columns', $$ + MATCH (n:DeleteVirtual) + SET n.checked = true + RETURN count(n) +$$) AS (count agtype); + +SELECT bool_and(marker = 1) AS virtual_ok +FROM generated_columns."DeleteVirtual"; + +SELECT * FROM cypher('generated_columns', $$ + MATCH (n:DeleteVirtual) + DELETE n + RETURN count(n) +$$) AS (count agtype); + +SELECT count(*) FROM generated_columns."DeleteVirtual"; + -- -- Cleanup -- diff --git a/regress/sql/security.sql b/regress/sql/security.sql index 65910cb5f..4f834aba3 100644 --- a/regress/sql/security.sql +++ b/regress/sql/security.sql @@ -1246,6 +1246,70 @@ SELECT * FROM cypher('rls_graph', $$ MATCH ()-[k:KNOWS]->(b:Person {name: 'DetachFn2'}) RETURN k.since $$) AS (since agtype); +-- Exercise function-based edge RLS through the default endpoint indexes. +SELECT * FROM cypher('rls_graph', $$ + CREATE (:Person {name: 'DetachFnIndexHub', owner: 'rls_user1', department: 'DetachFn'}) +$$) AS (a agtype); + +SELECT * FROM cypher('rls_graph', $$ + MATCH (hub:Person {name: 'DetachFnIndexHub'}) + UNWIND range(1, 16) AS ident + CREATE (hub)-[:KNOWS {owner: 'rls_user1'}]-> + (:Person {owner: 'rls_user1', department: 'DetachFn', leaf: ident}) +$$) AS (a agtype); + +SET ROLE rls_user1; + +SELECT * FROM cypher('rls_graph', $$ + MATCH (p:Person {name: 'DetachFnIndexHub'}) DETACH DELETE p +$$) AS (a agtype); + +RESET ROLE; + +SELECT * FROM cypher('rls_graph', $$ + MATCH ()-[k:KNOWS]->() WHERE k.owner = 'rls_user1' RETURN count(k) +$$) AS (count agtype); + +-- Drop the endpoint indexes to exercise the connected-edge sequential scan. +DO $$ +DECLARE + index_to_drop regclass; +BEGIN + FOR index_to_drop IN + SELECT indexrelid + FROM pg_index + WHERE indrelid = 'rls_graph."KNOWS"'::regclass + AND indnatts = 1 + AND indkey[0] IN (2, 3) + LOOP + EXECUTE format('DROP INDEX %s', index_to_drop); + END LOOP; +END +$$; + +SELECT * FROM cypher('rls_graph', $$ + CREATE (:Person {name: 'DetachFnSeqHub', owner: 'rls_user1', department: 'DetachFn'}) +$$) AS (a agtype); + +SELECT * FROM cypher('rls_graph', $$ + MATCH (hub:Person {name: 'DetachFnSeqHub'}) + UNWIND range(1, 16) AS ident + CREATE (hub)-[:KNOWS {owner: 'rls_user1'}]-> + (:Person {owner: 'rls_user1', department: 'DetachFn', seq_leaf: ident}) +$$) AS (a agtype); + +SET ROLE rls_user1; + +SELECT * FROM cypher('rls_graph', $$ + MATCH (p:Person {name: 'DetachFnSeqHub'}) DETACH DELETE p +$$) AS (a agtype); + +RESET ROLE; + +SELECT * FROM cypher('rls_graph', $$ + MATCH ()-[k:KNOWS]->() WHERE k.owner = 'rls_user1' RETURN count(k) +$$) AS (count agtype); + -- cleanup DROP POLICY detach_fn_knows_owner ON rls_graph."KNOWS"; DROP POLICY detach_fn_person_all ON rls_graph."Person"; diff --git a/src/backend/executor/cypher_create.c b/src/backend/executor/cypher_create.c index 36ef61b32..6a37a7026 100644 --- a/src/backend/executor/cypher_create.c +++ b/src/backend/executor/cypher_create.c @@ -209,6 +209,9 @@ static TupleTableSlot *exec_cypher_create(CustomScanState *node) */ do { + /* Release generated-column scratch from the preceding row. */ + ResetPerTupleExprContext(estate); + /*Process the subtree first */ Decrement_Estate_CommandId(estate) slot = ExecProcNode(node->ss.ps.lefttree); diff --git a/src/backend/executor/cypher_delete.c b/src/backend/executor/cypher_delete.c index e2161e66e..c70966b3d 100644 --- a/src/backend/executor/cypher_delete.c +++ b/src/backend/executor/cypher_delete.c @@ -149,6 +149,10 @@ static TupleTableSlot *exec_cypher_delete(CustomScanState *node) */ while(true) { + /* Release CustomScan and EState scratch from the preceding row. */ + ResetExprContext(econtext); + ResetPerTupleExprContext(estate); + /* Process the subtree first */ Decrement_Estate_CommandId(estate) slot = ExecProcNode(node->ss.ps.lefttree); @@ -168,6 +172,10 @@ static TupleTableSlot *exec_cypher_delete(CustomScanState *node) } else { + /* Release CustomScan and EState scratch from the preceding row. */ + ResetExprContext(econtext); + ResetPerTupleExprContext(estate); + /* Process the subtree first */ Decrement_Estate_CommandId(estate) slot = ExecProcNode(node->ss.ps.lefttree); @@ -301,6 +309,12 @@ static void delete_entity(EState *estate, ResultRelInfo *resultRelInfo, saved_resultRels = estate->es_result_relations; estate->es_result_relations = &resultRelInfo; + /* + * Initialize generated-column state in the per-tuple context before + * lock-mode selection can initialize it in the query context. + */ + init_result_rel_info_generated(resultRelInfo, estate); + lockmode = ExecUpdateLockMode(estate, resultRelInfo); lock_result = heap_lock_tuple(resultRelInfo->ri_RelationDesc, tuple, @@ -386,6 +400,10 @@ static void process_delete_list(CustomScanState *node) HASHCTL hashctl; HTAB *index_cache = NULL; HASHCTL idx_hashctl; + MemoryContext old_context; + + /* Allocate transient delete state in the per-tuple 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 +535,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 +585,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); } /* @@ -645,6 +681,9 @@ static void process_edges_by_index(Oid index_oid, /* Check RLS security quals (USING policy) before delete */ if (rls_enabled) { + /* Reset RLS expression scratch for each edge. */ + ResetExprContext(econtext); + if (!check_security_quals(qualExprs, slot, econtext)) { ereport(ERROR, @@ -823,6 +862,9 @@ static void check_for_connected_edges(CustomScanState *node) /* Check RLS security quals (USING policy) before delete */ if (rls_enabled) { + /* Reset RLS expression scratch for each edge. */ + ResetExprContext(econtext); + /* * For DETACH DELETE, error out if edge RLS check fails. * Unlike normal DELETE which silently skips, we cannot diff --git a/src/backend/executor/cypher_merge.c b/src/backend/executor/cypher_merge.c index 9c52073c2..cbd866444 100644 --- a/src/backend/executor/cypher_merge.c +++ b/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 @@ -426,6 +428,35 @@ static void free_path_entry_array(path_entry **path_array, int length) { pfree_if_not_null(path_array[index]); } + + /* free up the array container */ + pfree_if_not_null(path_array); +} + +/* + * Copy new-entity properties into created_paths_list's memory context. + * prebuild_path() evaluates these Datums in the per-row expression context, + * but created_paths_list is retained until node shutdown. + */ +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); } /* @@ -690,6 +721,10 @@ static TupleTableSlot *exec_cypher_merge(CustomScanState *node) TupleTableSlot *projected; HeapTuple htup; + /* Release scratch retained by the preceding input row. */ + ResetExprContext(econtext); + ResetPerTupleExprContext(estate); + /* Process the subtree first */ Decrement_Estate_CommandId(estate) slot = ExecProcNode(node->ss.ps.lefttree); @@ -737,6 +772,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 +833,10 @@ static TupleTableSlot *exec_cypher_merge(CustomScanState *node) */ do { + /* Release scratch retained by the preceding input row. */ + ResetExprContext(econtext); + ResetPerTupleExprContext(estate); + /* Process the subtree first */ Decrement_Estate_CommandId(estate) slot = ExecProcNode(node->ss.ps.lefttree); @@ -837,6 +879,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 +956,8 @@ static TupleTableSlot *exec_cypher_merge(CustomScanState *node) * Process the subtree. The subtree will only consist of the MERGE * path. */ + ResetExprContext(econtext); + ResetPerTupleExprContext(estate); Decrement_Estate_CommandId(estate) slot = ExecProcNode(node->ss.ps.lefttree); Increment_Estate_CommandId(estate) @@ -1084,9 +1130,6 @@ static void end_cypher_merge(CustomScanState *node) /* free up the path array elements */ free_path_entry_array(entry, path_length); - /* free up the array container */ - pfree_if_not_null(entry); - /* free up the created_path container */ pfree_if_not_null(css->created_paths_list); diff --git a/src/backend/executor/cypher_set.c b/src/backend/executor/cypher_set.c index 7a0d48f0c..37c49d3b7 100644 --- a/src/backend/executor/cypher_set.c +++ b/src/backend/executor/cypher_set.c @@ -113,6 +113,12 @@ static HeapTuple update_entity_tuple(ResultRelInfo *resultRelInfo, estate->es_result_relations = &resultRelInfo; + /* + * Initialize generated-column state in the per-tuple context before + * lock-mode selection can initialize it in the query context. + */ + init_result_rel_info_generated(resultRelInfo, estate); + lockmode = ExecUpdateLockMode(estate, resultRelInfo); lock_result = heap_lock_tuple(resultRelInfo->ri_RelationDesc, old_tuple, @@ -237,18 +243,24 @@ static HeapTuple update_entity_tuple(ResultRelInfo *resultRelInfo, } /* - * When the CREATE clause is the last cypher clause, consume all input from the - * previous clause(s) in the first call of exec_cypher_create. + * When SET or REMOVE is the last Cypher clause, consume all input from the + * previous clauses in the first call of exec_cypher_set. */ static void process_all_tuples(CustomScanState *node) { cypher_set_custom_scan_state *css = (cypher_set_custom_scan_state *)node; TupleTableSlot *slot; EState *estate = css->css.ss.ps.state; + ExprContext *econtext = css->css.ss.ps.ps_ExprContext; do { + /* Release scratch retained by the preceding input row. */ + ResetExprContext(econtext); process_update_list(node); + + /* Release generated-column scratch before reading the next row. */ + ResetPerTupleExprContext(estate); Decrement_Estate_CommandId(estate) slot = ExecProcNode(node->ss.ps.lefttree); Increment_Estate_CommandId(estate) @@ -415,6 +427,10 @@ void apply_update_list(CustomScanState *node, HASHCTL hashctl; HTAB *index_cache = NULL; HASHCTL idx_hashctl; + MemoryContext old_context; + + /* Allocate transient update state in the per-tuple context. */ + old_context = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory); /* allocate an array to hold the last update index of each 'entity' */ luindex = palloc0(sizeof(int) * scanTupleSlot->tts_nvalid); @@ -625,8 +641,9 @@ void apply_update_list(CustomScanState *node, } index_oid = idx_entry->index_oid; - slot = ExecInitExtraTupleSlot( - estate, RelationGetDescr(resultRelInfo->ri_RelationDesc), + /* Do not accumulate per-row slots in estate->es_tupleTable. */ + slot = MakeSingleTupleTableSlot( + RelationGetDescr(resultRelInfo->ri_RelationDesc), &TTSOpsHeapTuple); /* Setup RLS policies if RLS is enabled */ @@ -648,8 +665,10 @@ void apply_update_list(CustomScanState *node, /* Setup security quals */ entry->qualExprs = setup_security_quals(resultRelInfo, estate, node, CMD_UPDATE); - entry->slot = ExecInitExtraTupleSlot( - estate, RelationGetDescr(resultRelInfo->ri_RelationDesc), + + /* The per-row RLS cache owns this standalone slot. */ + entry->slot = MakeSingleTupleTableSlot( + RelationGetDescr(resultRelInfo->ri_RelationDesc), &TTSOpsHeapTuple); } else @@ -843,6 +862,7 @@ void apply_update_list(CustomScanState *node, } estate->es_snapshot->curcid = cid; + ExecDropSingleTupleTableSlot(slot); /* close relation */ ExecCloseIndices(resultRelInfo); table_close(resultRelInfo->ri_RelationDesc, RowExclusiveLock); @@ -851,12 +871,29 @@ void apply_update_list(CustomScanState *node, lidx++; } + /* 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); /* free our lookup array */ pfree_if_not_null(luindex); + + MemoryContextSwitchTo(old_context); } static void process_update_list(CustomScanState *node) @@ -876,6 +913,9 @@ static TupleTableSlot *exec_cypher_set(CustomScanState *node) saved_resultRels = estate->es_result_relations; + /* Release EState per-tuple scratch before fetching the next input row. */ + ResetPerTupleExprContext(estate); + /* Process the subtree first */ Decrement_Estate_CommandId(estate); slot = ExecProcNode(node->ss.ps.lefttree); @@ -904,6 +944,8 @@ static TupleTableSlot *exec_cypher_set(CustomScanState *node) return NULL; } + /* Release scratch allocated by this CustomScan for the preceding row. */ + ResetExprContext(econtext); process_update_list(node); /* increment the command counter to reflect the updates */ diff --git a/src/backend/executor/cypher_utils.c b/src/backend/executor/cypher_utils.c index c697a560c..d51a3c256 100644 --- a/src/backend/executor/cypher_utils.c +++ b/src/backend/executor/cypher_utils.c @@ -32,6 +32,7 @@ #include "rewrite/rewriteManip.h" #include "rewrite/rowsecurity.h" #include "utils/acl.h" +#include "utils/memutils.h" #include "utils/rls.h" #include "catalog/ag_label.h" @@ -129,6 +130,52 @@ void destroy_entity_result_rel_info(ResultRelInfo *result_rel_info) table_close(result_rel_info->ri_RelationDesc, RowExclusiveLock); } +/* + * Initialize generated-column state in the EState per-tuple context. + * + * PostgreSQL's ModifyTable keeps this state in es_query_cxt because it reuses + * each ResultRelInfo for the statement. SET/REMOVE and DELETE instead create + * a temporary ResultRelInfo for each input row, so allocating the state there + * would retain one copy per row. + */ +void init_result_rel_info_generated(ResultRelInfo *result_rel_info, + EState *estate) +{ + TupleConstr *constr = + result_rel_info->ri_RelationDesc->rd_att->constr; + MemoryContext old_query_context; + MemoryContext per_tuple_context; + + if (constr == NULL || + (!constr->has_generated_stored && !constr->has_generated_virtual) || + result_rel_info->ri_extraUpdatedCols_valid) + { + return; + } + + /* Direct ExecInitGenerated() allocations to the per-tuple context. */ + old_query_context = estate->es_query_cxt; + per_tuple_context = GetPerTupleMemoryContext(estate); + + PG_TRY(); + { + estate->es_query_cxt = per_tuple_context; + ExecInitGenerated(result_rel_info, estate, CMD_UPDATE); + + Assert(result_rel_info->ri_GeneratedExprsU == NULL || + GetMemoryChunkContext(result_rel_info->ri_GeneratedExprsU) == + per_tuple_context); + Assert(result_rel_info->ri_extraUpdatedCols == NULL || + GetMemoryChunkContext(result_rel_info->ri_extraUpdatedCols) == + per_tuple_context); + } + PG_FINALLY(); + { + estate->es_query_cxt = old_query_context; + } + PG_END_TRY(); +} + /* * Clear an entity slot and mark every attribute NULL before AGE fills in the * columns it manages (id/start_id/end_id/properties). diff --git a/src/include/executor/cypher_utils.h b/src/include/executor/cypher_utils.h index 8b65bc964..28aae933f 100644 --- a/src/include/executor/cypher_utils.h +++ b/src/include/executor/cypher_utils.h @@ -129,6 +129,8 @@ TupleTableSlot *populate_edge_tts( ResultRelInfo *create_entity_result_rel_info(EState *estate, char *graph_name, char *label_name); void destroy_entity_result_rel_info(ResultRelInfo *result_rel_info); +void init_result_rel_info_generated(ResultRelInfo *result_rel_info, + EState *estate); bool entity_exists(EState *estate, Oid graph_oid, graphid id); HeapTuple insert_entity_tuple(ResultRelInfo *resultRelInfo,