From 2634589c385942ec6b1e95e8b956b2ae2262231c Mon Sep 17 00:00:00 2001 From: crdv7 <15974123760@163.com> Date: Wed, 29 Jul 2026 14:50:27 +0800 Subject: [PATCH] Optimize writable entity lookup and preserve same-row alias updates Propagate hidden CTID hints through eligible MATCH and WITH clauses to avoid repeated graphid lookups during SET, REMOVE, and DELETE. Treat CTID as a validated physical hint. Fall back to a usable non-partial B-tree index whose first key is the entity id, and use a sequential scan only when no suitable index exists. Reuse lookup slots and index relations for the lifetime of SET, DELETE, and MERGE statements. Fix a pre-existing issue where sequential SET or REMOVE clauses through different aliases of the same entity could build an update from stale properties and overwrite an earlier change. Refresh the base properties from the current heap tuple when an alias carries a missing or stale CTID, and synchronize top-level aliases and paths with the same graphid. Avoid propagating hidden CTID entries through aggregation, DISTINCT, computed expressions, or internal parser variables, where they could change query semantics or become associated with the wrong entity. Add regression coverage for stale CTID fallback, same-entity vertex and edge aliases, dependent right-hand-side expressions, REMOVE, DELETE, WITH aliases, aggregation, DISTINCT, and user-created edge id indexes. Closes apache/age#2484 Closes apache/age#2485 --- regress/expected/cypher_delete.out | 63 +++- regress/expected/cypher_remove.out | 43 ++- regress/expected/cypher_set.out | 239 +++++++++++++++ regress/sql/cypher_delete.sql | 16 + regress/sql/cypher_remove.sql | 23 ++ regress/sql/cypher_set.sql | 122 ++++++++ src/backend/executor/cypher_delete.c | 124 ++------ src/backend/executor/cypher_merge.c | 7 + src/backend/executor/cypher_set.c | 396 +++++++++++++------------ src/backend/executor/cypher_utils.c | 331 +++++++++++++++++++++ src/backend/nodes/cypher_copyfuncs.c | 6 + src/backend/nodes/cypher_outfuncs.c | 9 + src/backend/nodes/cypher_readfuncs.c | 6 + src/backend/parser/cypher_analyze.c | 19 ++ src/backend/parser/cypher_clause.c | 173 +++++++++++ src/backend/parser/cypher_parse_node.c | 6 + src/backend/utils/adt/agtype.c | 9 + src/include/executor/cypher_utils.h | 45 +++ src/include/nodes/cypher_nodes.h | 11 + src/include/parser/cypher_parse_node.h | 7 + src/include/utils/agtype.h | 14 + 21 files changed, 1379 insertions(+), 290 deletions(-) diff --git a/regress/expected/cypher_delete.out b/regress/expected/cypher_delete.out index 3cdbbf8bb..465dc05b1 100644 --- a/regress/expected/cypher_delete.out +++ b/regress/expected/cypher_delete.out @@ -812,15 +812,76 @@ SELECT id as "expected: 0 rows" FROM setdelete._ag_label_edge; ------------------ (0 rows) +-- stale ctid fallback - SET one alias, then DELETE the same entity via another alias. +SELECT * FROM cypher('setdelete', $$ CREATE (:B) $$) as ("CREATE" agtype); + CREATE +-------- +(0 rows) + +SELECT * FROM cypher('setdelete', $$ MATCH (n:B), (m:B) WHERE id(n) = id(m) SET n.flag = true DELETE m $$) as ("SET + DELETE alias" agtype); + SET + DELETE alias +-------------------- +(0 rows) + +SELECT count(*) as "expected: 0 rows" FROM setdelete."B"; + expected: 0 rows +------------------ + 0 +(1 row) + +-- hidden ctid propagation through WITH. +SELECT * FROM cypher('setdelete', $$ CREATE (:C) $$) as ("CREATE" agtype); + CREATE +-------- +(0 rows) + +SELECT * FROM cypher('setdelete', $$ MATCH (n:C) WITH n DELETE n $$) as ("WITH DELETE" agtype); + WITH DELETE +------------- +(0 rows) + +SELECT count(*) as "expected: 0 rows" FROM setdelete."C"; + expected: 0 rows +------------------ + 0 +(1 row) + +-- stale edge ctid fallback with DETACH DELETE through another alias. +SELECT * FROM cypher('setdelete', $$ CREATE (:D)-[:de]->(:D) $$) AS ("CREATE" agtype); + CREATE +-------- +(0 rows) + +SELECT * FROM cypher('setdelete', $$ MATCH (n)-[e:de]->(m), (x)-[f:de]->(y) WHERE id(e) = id(f) SET e.i = 1 DETACH DELETE x $$) AS ("SET + DETACH alias" agtype); + SET + DETACH alias +-------------------- +(0 rows) + +SELECT count(*) as "expected: 1 row" FROM setdelete."D"; + expected: 1 row +----------------- + 1 +(1 row) + +SELECT count(*) as "expected: 0 rows" FROM setdelete.de; + expected: 0 rows +------------------ + 0 +(1 row) + -- clean up SELECT drop_graph('setdelete', true); -NOTICE: drop cascades to 6 other objects +NOTICE: drop cascades to 10 other objects DETAIL: drop cascades to table setdelete._ag_label_vertex drop cascades to table setdelete._ag_label_edge drop cascades to table setdelete."A" drop cascades to table setdelete.n drop cascades to table setdelete.e drop cascades to table setdelete.m +drop cascades to table setdelete."B" +drop cascades to table setdelete."C" +drop cascades to table setdelete."D" +drop cascades to table setdelete.de NOTICE: graph "setdelete" has been dropped drop_graph ------------ diff --git a/regress/expected/cypher_remove.out b/regress/expected/cypher_remove.out index ca2b94ae6..59bf67179 100644 --- a/regress/expected/cypher_remove.out +++ b/regress/expected/cypher_remove.out @@ -457,6 +457,45 @@ SELECT * FROM cypher('cypher_remove', $$MATCH ()-[e:edge_multi_property]-() REMO {"id": 3659174697238529, "label": "edge_multi_property", "end_id": 281474976710664, "start_id": 281474976710663, "properties": {}}::edge (2 rows) +-- Hidden ctid columns must propagate through WITH for REMOVE. +SELECT * FROM cypher('cypher_remove', $$ + CREATE (:ctid_remove {drop_me: 1}) +$$) AS (a agtype); + a +--- +(0 rows) + +SELECT * FROM cypher('cypher_remove', $$ + MATCH (n:ctid_remove) + WITH n + REMOVE n.drop_me + RETURN n +$$) AS (a agtype); + a +---------------------------------------------------------------------------- + {"id": 3940649673949185, "label": "ctid_remove", "properties": {}}::vertex +(1 row) + +-- REMOVE must fall back from a stale ctid after an earlier SET moves the row. +SELECT * FROM cypher('cypher_remove', $$ + CREATE (:ctid_remove_stale {drop_me: 1}) +$$) AS (a agtype); + a +--- +(0 rows) + +SELECT * FROM cypher('cypher_remove', $$ + MATCH (n:ctid_remove_stale), (m:ctid_remove_stale) + WHERE id(n) = id(m) + SET n.changed = true + REMOVE m.drop_me + RETURN m +$$) AS (a agtype); + a +------------------------------------------------------------------------------------------------- + {"id": 4222124650659841, "label": "ctid_remove_stale", "properties": {"changed": true}}::vertex +(1 row) + --Errors SELECT * FROM cypher('cypher_remove', $$REMOVE n.i$$) AS (a agtype); ERROR: REMOVE cannot be the first clause in a Cypher query @@ -475,7 +514,7 @@ LINE 1: SELECT * FROM cypher('cypher_remove', $$MATCH (n) REMOVE wro... -- DROP FUNCTION remove_test; SELECT drop_graph('cypher_remove', true); -NOTICE: drop cascades to 13 other objects +NOTICE: drop cascades to 15 other objects DETAIL: drop cascades to table cypher_remove._ag_label_vertex drop cascades to table cypher_remove._ag_label_edge drop cascades to table cypher_remove.test_1 @@ -489,6 +528,8 @@ drop cascades to table cypher_remove.test_6 drop cascades to table cypher_remove.e drop cascades to table cypher_remove.test_7 drop cascades to table cypher_remove.edge_multi_property +drop cascades to table cypher_remove.ctid_remove +drop cascades to table cypher_remove.ctid_remove_stale NOTICE: graph "cypher_remove" has been dropped drop_graph ------------ diff --git a/regress/expected/cypher_set.out b/regress/expected/cypher_set.out index 239234ed6..61157153f 100644 --- a/regress/expected/cypher_set.out +++ b/regress/expected/cypher_set.out @@ -1227,6 +1227,226 @@ $$) AS (a agtype); {"id": 5066549580791809, "label": "TestE2", "properties": {"pathRels": [{"id": 5348024557502465, "label": "E2REL", "end_id": 5066549580791810, "start_id": 5066549580791809, "properties": {}}::edge], "pathNodes": [{"id": 5066549580791809, "label": "TestE2", "properties": {}}::vertex, {"id": 5066549580791810, "label": "TestE2", "properties": {}}::vertex]}}::vertex (1 row) +-- ============================================================================ +-- ctid lookup fallback tests +-- ============================================================================ +SELECT create_graph('ctid_set'); +NOTICE: graph "ctid_set" has been created + create_graph +-------------- + +(1 row) + +-- Same entity referenced through two aliases: the second SET must fall back +-- from the stale ctid to graphid lookup after the first SET moves the tuple. +SELECT * FROM cypher('ctid_set', $$CREATE (:v {k: 1})$$) AS (a agtype); + a +--- +(0 rows) + +SELECT * FROM cypher('ctid_set', $$ + MATCH (n:v), (m:v) + WHERE id(n) = id(m) + SET n.a = 1 + SET m.a = 2 + RETURN n.a, m.a +$$) AS (n agtype, m agtype); + n | m +---+--- + 2 | 2 +(1 row) + +SELECT * FROM cypher('ctid_set', $$MATCH (n:v) RETURN n.a$$) AS (a agtype); + a +--- + 2 +(1 row) + +-- Updating the same entity through different aliases must preserve earlier +-- changes, including when a later RHS depends on an earlier SET. +SELECT * FROM cypher('ctid_set', $$ + CREATE (:alias_vertex {k: 1}) +$$) AS (a agtype); + a +--- +(0 rows) + +SELECT * FROM cypher('ctid_set', $$ + MATCH (n:alias_vertex), (m:alias_vertex) + WHERE id(n) = id(m) + SET n.a = 1 + SET m.b = n.a + 1 + RETURN n.a, m.a, m.b +$$) AS (na agtype, ma agtype, mb agtype); + na | ma | mb +----+----+---- + 1 | 1 | 2 +(1 row) + +SELECT * FROM cypher('ctid_set', $$ + MATCH (n:alias_vertex) + RETURN n.a, n.b +$$) AS (a agtype, b agtype); + a | b +---+--- + 1 | 2 +(1 row) + +-- The same alias synchronization is required for edges. +SELECT * FROM cypher('ctid_set', $$ + CREATE ()-[:alias_edge {k: 1}]->() +$$) AS (a agtype); + a +--- +(0 rows) + +SELECT * FROM cypher('ctid_set', $$ + MATCH ()-[n:alias_edge]->(), ()-[m:alias_edge]->() + WHERE id(n) = id(m) + SET n.a = 1 + SET m.b = n.a + 1 + RETURN n.a, m.a, m.b +$$) AS (na agtype, ma agtype, mb agtype); + na | ma | mb +----+----+---- + 1 | 1 | 2 +(1 row) + +SELECT * FROM cypher('ctid_set', $$ + MATCH ()-[n:alias_edge]->() + RETURN n.a, n.b +$$) AS (a agtype, b agtype); + a | b +---+--- + 1 | 2 +(1 row) + +-- Hidden ctid columns must propagate through WITH. +SELECT * FROM cypher('ctid_set', $$CREATE (:w {k: 1})$$) AS (a agtype); + a +--- +(0 rows) + +SELECT * FROM cypher('ctid_set', $$ + MATCH (n:w) + WITH n + SET n.k = 2 + RETURN n.k +$$) AS (a agtype); + a +--- + 2 +(1 row) + +SELECT * FROM cypher('ctid_set', $$MATCH (n:w) RETURN n.k$$) AS (a agtype); + a +--- + 2 +(1 row) + +-- Hidden ctid columns must follow simple WITH aliases too. +SELECT * FROM cypher('ctid_set', $$ + MATCH (n:w) + WITH n AS m + SET m.k = 3 + RETURN m.k +$$) AS (a agtype); + a +--- + 3 +(1 row) + +SELECT * FROM cypher('ctid_set', $$MATCH (n:w) RETURN n.k$$) AS (a agtype); + a +--- + 3 +(1 row) + +-- Computed WITH expressions must not inherit a ctid from one of their inputs. +SELECT * FROM cypher('ctid_set', $$ + CREATE (:expr {k: 1}), (:expr {k: 2}) +$$) AS (a agtype); + a +--- +(0 rows) + +SELECT * FROM cypher('ctid_set', $$ + MATCH (a:expr {k: 1}), (b:expr {k: 2}) + WITH CASE WHEN a.k = 1 THEN a ELSE b END AS n + SET n.mark = 1 + RETURN n.k, n.mark +$$) AS (k agtype, mark agtype); + k | mark +---+------ + 1 | 1 +(1 row) + +-- Aggregating WITH clauses must not leak pre-aggregation ctid values. +SELECT * FROM cypher('ctid_set', $$ + CREATE (:aggregate {k: 1}), (:aggregate {k: 2}) +$$) AS (a agtype); + a +--- +(0 rows) + +SELECT * FROM cypher('ctid_set', $$ + MATCH (n:aggregate) + WITH collect(n) AS nodes, count(*) AS total + UNWIND nodes AS n + SET n.total = total + RETURN n.k, n.total +$$) AS (k agtype, total agtype); + k | total +---+------- + 1 | 2 + 2 | 2 +(2 rows) + +-- DISTINCT is a row-shaping boundary and must preserve its semantics. +SELECT * FROM cypher('ctid_set', $$ + CREATE (:distinct_source), (:distinct_source), + (:distinct_target {hits: 0}) +$$) AS (a agtype); + a +--- +(0 rows) + +SELECT * FROM cypher('ctid_set', $$ + MATCH (:distinct_source), (n:distinct_target) + WITH DISTINCT n + SET n.hits = n.hits + 1 + RETURN n.hits +$$) AS (hits agtype); + hits +------ + 1 +(1 row) + +-- Edge labels have no inherited primary key. A user-created B-tree index on +-- id must support fallback when DISTINCT intentionally suppresses the ctid. +SELECT * FROM cypher('ctid_set', $$ + CREATE ()-[:indexed_edge {k: 1}]->(), + ()-[:indexed_edge {k: 2}]->() +$$) AS (a agtype); + a +--- +(0 rows) + +CREATE INDEX ctid_set_indexed_edge_id_idx + ON ctid_set.indexed_edge (id); +SELECT * FROM cypher('ctid_set', $$ + MATCH ()-[e:indexed_edge]->() + WITH DISTINCT e + SET e.indexed = true + RETURN e.k, e.indexed + ORDER BY e.k +$$) AS (k agtype, indexed agtype); + k | indexed +---+--------- + 1 | true + 2 | true +(2 rows) + -- -- Clean up -- @@ -1304,6 +1524,25 @@ NOTICE: graph "issue_1884" has been dropped (1 row) +SELECT drop_graph('ctid_set', true); +NOTICE: drop cascades to 11 other objects +DETAIL: drop cascades to table ctid_set._ag_label_vertex +drop cascades to table ctid_set._ag_label_edge +drop cascades to table ctid_set.v +drop cascades to table ctid_set.alias_vertex +drop cascades to table ctid_set.alias_edge +drop cascades to table ctid_set.w +drop cascades to table ctid_set.expr +drop cascades to table ctid_set.aggregate +drop cascades to table ctid_set.distinct_source +drop cascades to table ctid_set.distinct_target +drop cascades to table ctid_set.indexed_edge +NOTICE: graph "ctid_set" has been dropped + drop_graph +------------ + +(1 row) + -- -- End -- diff --git a/regress/sql/cypher_delete.sql b/regress/sql/cypher_delete.sql index 4ddea60c2..f9dd30233 100644 --- a/regress/sql/cypher_delete.sql +++ b/regress/sql/cypher_delete.sql @@ -299,6 +299,22 @@ SELECT * FROM cypher('setdelete', $$ MATCH (n)-[e]->(m) SET e.i = 1 DETACH DELET SELECT id as "expected: 2 rows (m vertices)" FROM setdelete._ag_label_vertex; SELECT id as "expected: 0 rows" FROM setdelete._ag_label_edge; +-- stale ctid fallback - SET one alias, then DELETE the same entity via another alias. +SELECT * FROM cypher('setdelete', $$ CREATE (:B) $$) as ("CREATE" agtype); +SELECT * FROM cypher('setdelete', $$ MATCH (n:B), (m:B) WHERE id(n) = id(m) SET n.flag = true DELETE m $$) as ("SET + DELETE alias" agtype); +SELECT count(*) as "expected: 0 rows" FROM setdelete."B"; + +-- hidden ctid propagation through WITH. +SELECT * FROM cypher('setdelete', $$ CREATE (:C) $$) as ("CREATE" agtype); +SELECT * FROM cypher('setdelete', $$ MATCH (n:C) WITH n DELETE n $$) as ("WITH DELETE" agtype); +SELECT count(*) as "expected: 0 rows" FROM setdelete."C"; + +-- stale edge ctid fallback with DETACH DELETE through another alias. +SELECT * FROM cypher('setdelete', $$ CREATE (:D)-[:de]->(:D) $$) AS ("CREATE" agtype); +SELECT * FROM cypher('setdelete', $$ MATCH (n)-[e:de]->(m), (x)-[f:de]->(y) WHERE id(e) = id(f) SET e.i = 1 DETACH DELETE x $$) AS ("SET + DETACH alias" agtype); +SELECT count(*) as "expected: 1 row" FROM setdelete."D"; +SELECT count(*) as "expected: 0 rows" FROM setdelete.de; + -- clean up SELECT drop_graph('setdelete', true); diff --git a/regress/sql/cypher_remove.sql b/regress/sql/cypher_remove.sql index 1bd4c059c..b05296eab 100644 --- a/regress/sql/cypher_remove.sql +++ b/regress/sql/cypher_remove.sql @@ -136,6 +136,29 @@ SELECT * FROM cypher('cypher_remove', $$CREATE ()-[:edge_multi_property { i: 5, SELECT * FROM cypher('cypher_remove', $$MATCH ()-[e:edge_multi_property]-() RETURN e$$) AS (a agtype); SELECT * FROM cypher('cypher_remove', $$MATCH ()-[e:edge_multi_property]-() REMOVE e.i, e.j RETURN e$$) AS (a agtype); +-- Hidden ctid columns must propagate through WITH for REMOVE. +SELECT * FROM cypher('cypher_remove', $$ + CREATE (:ctid_remove {drop_me: 1}) +$$) AS (a agtype); +SELECT * FROM cypher('cypher_remove', $$ + MATCH (n:ctid_remove) + WITH n + REMOVE n.drop_me + RETURN n +$$) AS (a agtype); + +-- REMOVE must fall back from a stale ctid after an earlier SET moves the row. +SELECT * FROM cypher('cypher_remove', $$ + CREATE (:ctid_remove_stale {drop_me: 1}) +$$) AS (a agtype); +SELECT * FROM cypher('cypher_remove', $$ + MATCH (n:ctid_remove_stale), (m:ctid_remove_stale) + WHERE id(n) = id(m) + SET n.changed = true + REMOVE m.drop_me + RETURN m +$$) AS (a agtype); + --Errors SELECT * FROM cypher('cypher_remove', $$REMOVE n.i$$) AS (a agtype); diff --git a/regress/sql/cypher_set.sql b/regress/sql/cypher_set.sql index e745d5d6e..7fd39463e 100644 --- a/regress/sql/cypher_set.sql +++ b/regress/sql/cypher_set.sql @@ -542,6 +542,127 @@ SELECT * FROM cypher('issue_1884', $$ RETURN a $$) AS (a agtype); +-- ============================================================================ +-- ctid lookup fallback tests +-- ============================================================================ + +SELECT create_graph('ctid_set'); + +-- Same entity referenced through two aliases: the second SET must fall back +-- from the stale ctid to graphid lookup after the first SET moves the tuple. +SELECT * FROM cypher('ctid_set', $$CREATE (:v {k: 1})$$) AS (a agtype); +SELECT * FROM cypher('ctid_set', $$ + MATCH (n:v), (m:v) + WHERE id(n) = id(m) + SET n.a = 1 + SET m.a = 2 + RETURN n.a, m.a +$$) AS (n agtype, m agtype); +SELECT * FROM cypher('ctid_set', $$MATCH (n:v) RETURN n.a$$) AS (a agtype); + +-- Updating the same entity through different aliases must preserve earlier +-- changes, including when a later RHS depends on an earlier SET. +SELECT * FROM cypher('ctid_set', $$ + CREATE (:alias_vertex {k: 1}) +$$) AS (a agtype); +SELECT * FROM cypher('ctid_set', $$ + MATCH (n:alias_vertex), (m:alias_vertex) + WHERE id(n) = id(m) + SET n.a = 1 + SET m.b = n.a + 1 + RETURN n.a, m.a, m.b +$$) AS (na agtype, ma agtype, mb agtype); +SELECT * FROM cypher('ctid_set', $$ + MATCH (n:alias_vertex) + RETURN n.a, n.b +$$) AS (a agtype, b agtype); + +-- The same alias synchronization is required for edges. +SELECT * FROM cypher('ctid_set', $$ + CREATE ()-[:alias_edge {k: 1}]->() +$$) AS (a agtype); +SELECT * FROM cypher('ctid_set', $$ + MATCH ()-[n:alias_edge]->(), ()-[m:alias_edge]->() + WHERE id(n) = id(m) + SET n.a = 1 + SET m.b = n.a + 1 + RETURN n.a, m.a, m.b +$$) AS (na agtype, ma agtype, mb agtype); +SELECT * FROM cypher('ctid_set', $$ + MATCH ()-[n:alias_edge]->() + RETURN n.a, n.b +$$) AS (a agtype, b agtype); + +-- Hidden ctid columns must propagate through WITH. +SELECT * FROM cypher('ctid_set', $$CREATE (:w {k: 1})$$) AS (a agtype); +SELECT * FROM cypher('ctid_set', $$ + MATCH (n:w) + WITH n + SET n.k = 2 + RETURN n.k +$$) AS (a agtype); +SELECT * FROM cypher('ctid_set', $$MATCH (n:w) RETURN n.k$$) AS (a agtype); + +-- Hidden ctid columns must follow simple WITH aliases too. +SELECT * FROM cypher('ctid_set', $$ + MATCH (n:w) + WITH n AS m + SET m.k = 3 + RETURN m.k +$$) AS (a agtype); +SELECT * FROM cypher('ctid_set', $$MATCH (n:w) RETURN n.k$$) AS (a agtype); + +-- Computed WITH expressions must not inherit a ctid from one of their inputs. +SELECT * FROM cypher('ctid_set', $$ + CREATE (:expr {k: 1}), (:expr {k: 2}) +$$) AS (a agtype); +SELECT * FROM cypher('ctid_set', $$ + MATCH (a:expr {k: 1}), (b:expr {k: 2}) + WITH CASE WHEN a.k = 1 THEN a ELSE b END AS n + SET n.mark = 1 + RETURN n.k, n.mark +$$) AS (k agtype, mark agtype); + +-- Aggregating WITH clauses must not leak pre-aggregation ctid values. +SELECT * FROM cypher('ctid_set', $$ + CREATE (:aggregate {k: 1}), (:aggregate {k: 2}) +$$) AS (a agtype); +SELECT * FROM cypher('ctid_set', $$ + MATCH (n:aggregate) + WITH collect(n) AS nodes, count(*) AS total + UNWIND nodes AS n + SET n.total = total + RETURN n.k, n.total +$$) AS (k agtype, total agtype); + +-- DISTINCT is a row-shaping boundary and must preserve its semantics. +SELECT * FROM cypher('ctid_set', $$ + CREATE (:distinct_source), (:distinct_source), + (:distinct_target {hits: 0}) +$$) AS (a agtype); +SELECT * FROM cypher('ctid_set', $$ + MATCH (:distinct_source), (n:distinct_target) + WITH DISTINCT n + SET n.hits = n.hits + 1 + RETURN n.hits +$$) AS (hits agtype); + +-- Edge labels have no inherited primary key. A user-created B-tree index on +-- id must support fallback when DISTINCT intentionally suppresses the ctid. +SELECT * FROM cypher('ctid_set', $$ + CREATE ()-[:indexed_edge {k: 1}]->(), + ()-[:indexed_edge {k: 2}]->() +$$) AS (a agtype); +CREATE INDEX ctid_set_indexed_edge_id_idx + ON ctid_set.indexed_edge (id); +SELECT * FROM cypher('ctid_set', $$ + MATCH ()-[e:indexed_edge]->() + WITH DISTINCT e + SET e.indexed = true + RETURN e.k, e.indexed + ORDER BY e.k +$$) AS (k agtype, indexed agtype); + -- -- Clean up -- @@ -551,6 +672,7 @@ SELECT drop_graph('cypher_set', true); SELECT drop_graph('cypher_set_1', true); SELECT drop_graph('issue_1634', true); SELECT drop_graph('issue_1884', true); +SELECT drop_graph('ctid_set', true); -- -- End diff --git a/src/backend/executor/cypher_delete.c b/src/backend/executor/cypher_delete.c index e2161e66e..b3c4d091e 100644 --- a/src/backend/executor/cypher_delete.c +++ b/src/backend/executor/cypher_delete.c @@ -123,6 +123,9 @@ static void begin_cypher_delete(CustomScanState *node, EState *estate, if (estate->es_output_cid == 0) estate->es_output_cid = estate->es_snapshot->curcid; + /* Build the per-statement index and fetch-slot lookup cache. */ + css->entity_lookup_cache = create_entity_lookup_cache(); + Increment_Estate_CommandId(estate); } @@ -205,6 +208,10 @@ static void end_cypher_delete(CustomScanState *node) hash_destroy(((cypher_delete_custom_scan_state *)node)->vertex_id_htab); + /* Release cached index handles and fetch slots. */ + destroy_entity_lookup_cache( + ((cypher_delete_custom_scan_state *)node)->entity_lookup_cache); + ExecEndNode(node->ss.ps.lefttree); } @@ -384,8 +391,6 @@ static void process_delete_list(CustomScanState *node) EState *estate = node->ss.ps.state; HTAB *qual_cache = NULL; HASHCTL hashctl; - HTAB *index_cache = NULL; - HASHCTL idx_hashctl; /* Hash table for caching compiled security quals per label */ MemSet(&hashctl, 0, sizeof(hashctl)); @@ -395,19 +400,10 @@ static void process_delete_list(CustomScanState *node) qual_cache = hash_create("delete_qual_cache", 8, &hashctl, HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); - MemSet(&idx_hashctl, 0, sizeof(idx_hashctl)); - idx_hashctl.keysize = sizeof(Oid); - idx_hashctl.entrysize = sizeof(IndexCacheEntry); - idx_hashctl.hcxt = CurrentMemoryContext; - index_cache = hash_create("delete_index_cache", 8, &idx_hashctl, - HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); - foreach(lc, css->delete_data->delete_items) { cypher_delete_item *item; agtype_value *original_entity_value, *id, *label; - ScanKeyData scan_keys[1]; - TableScanDesc scan_desc = NULL; ResultRelInfo *resultRelInfo; HeapTuple heap_tuple = NULL; char *label_name; @@ -415,14 +411,6 @@ static void process_delete_list(CustomScanState *node) int entity_position; Oid relid; Relation rel; - int id_attr_num; - Oid index_oid = InvalidOid; - TupleTableSlot *slot = NULL; - Relation index_rel = NULL; - IndexScanDesc index_scan_desc = NULL; - bool shouldFree = false; - IndexCacheEntry *idx_entry; - bool found_idx_entry; item = lfirst(lc); @@ -448,119 +436,62 @@ static void process_delete_list(CustomScanState *node) * Setup the scan key to require the id field on-disc to match the * entity's graphid. */ - if (original_entity_value->type == AGTV_VERTEX) - { - id_attr_num = Anum_ag_label_vertex_table_id; - ScanKeyInit(&scan_keys[0], Anum_ag_label_vertex_table_id, - BTEqualStrategyNumber, F_GRAPHIDEQ, - GRAPHID_GET_DATUM(id->val.int_value)); - } - else if (original_entity_value->type == AGTV_EDGE) - { - id_attr_num = Anum_ag_label_edge_table_id; - ScanKeyInit(&scan_keys[0], Anum_ag_label_edge_table_id, - BTEqualStrategyNumber, F_GRAPHIDEQ, - GRAPHID_GET_DATUM(id->val.int_value)); - } - else + if (original_entity_value->type != AGTV_VERTEX && + original_entity_value->type != AGTV_EDGE) { ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("DELETE clause can only delete vertices and edges"))); } - idx_entry = hash_search(index_cache, &relid, HASH_ENTER, &found_idx_entry); - - if (!found_idx_entry) - { - idx_entry->index_oid = find_usable_btree_index_for_attr(rel, id_attr_num); - } - - index_oid = idx_entry->index_oid; - /* * Setup the scan description, with the correct snapshot and scan keys. */ estate->es_snapshot->curcid = GetCurrentCommandId(false); estate->es_output_cid = GetCurrentCommandId(false); - if (OidIsValid(index_oid)) - { - slot = table_slot_create(rel, NULL); - - index_rel = index_open(index_oid, RowExclusiveLock); - index_scan_desc = index_beginscan(rel, index_rel, estate->es_snapshot, NULL, 1, 0); - index_rescan(index_scan_desc, scan_keys, 1, NULL, 0); - - if (index_getnext_slot(index_scan_desc, ForwardScanDirection, slot)) - { - heap_tuple = ExecFetchSlotHeapTuple(slot, true, &shouldFree); - } - } - else - { - scan_desc = table_beginscan(rel, estate->es_snapshot, 1, scan_keys); - /* Retrieve the tuple. */ - heap_tuple = heap_getnext(scan_desc, ForwardScanDirection); - } - + heap_tuple = find_entity_tuple( + rel, estate->es_snapshot, id->val.int_value, scanTupleSlot, + item->ctid_position, css->entity_lookup_cache); if (HeapTupleIsValid(heap_tuple)) { bool passed_rls = true; - /* Check RLS security quals (USING policy) before delete */ if (check_enable_rls(relid, InvalidOid, true) == RLS_ENABLED) { RLSCacheEntry *entry; bool found_rls; - entry = hash_search(qual_cache, &relid, HASH_ENTER, &found_rls); + entry = hash_search(qual_cache, &relid, HASH_ENTER, + &found_rls); if (!found_rls) { - entry->qualExprs = setup_security_quals(resultRelInfo, estate, node, CMD_DELETE); - entry->slot = ExecInitExtraTupleSlot(estate, RelationGetDescr(rel), &TTSOpsHeapTuple); + entry->qualExprs = setup_security_quals( + resultRelInfo, estate, node, CMD_DELETE); + entry->slot = ExecInitExtraTupleSlot( + estate, RelationGetDescr(rel), &TTSOpsHeapTuple); } - ExecStoreHeapTuple(heap_tuple, entry->slot, false); - - if (!check_security_quals(entry->qualExprs, entry->slot, econtext)) - { - passed_rls = false; - } + passed_rls = check_security_quals(entry->qualExprs, + entry->slot, econtext); } - if (passed_rls) { /* - * For vertices, we insert the vertex ID in the hashtable - * vertex_id_htab. This hashtable is used later to process - * connected edges. + * Save deleted vertex IDs for the later connected-edge check. + * DETACH DELETE removes those edges; plain DELETE reports an + * error if any remain. */ if (original_entity_value->type == AGTV_VERTEX) { bool found; - hash_search(css->vertex_id_htab, (void *)&(id->val.int_value), + + hash_search(css->vertex_id_htab, + (void *)&id->val.int_value, HASH_ENTER, &found); } - - /* At this point, we are ready to delete the node/vertex. */ delete_entity(estate, resultRelInfo, heap_tuple); } - - if (shouldFree) - { - heap_freetuple(heap_tuple); - } - } - - if (OidIsValid(index_oid)) - { - ExecDropSingleTupleTableSlot(slot); - index_endscan(index_scan_desc); - index_close(index_rel, RowExclusiveLock); - } - else - { - table_endscan(scan_desc); + heap_freetuple(heap_tuple); } destroy_entity_result_rel_info(resultRelInfo); @@ -568,7 +499,6 @@ static void process_delete_list(CustomScanState *node) /* Clean up the cache */ hash_destroy(qual_cache); - hash_destroy(index_cache); } /* diff --git a/src/backend/executor/cypher_merge.c b/src/backend/executor/cypher_merge.c index 9c52073c2..af49afd96 100644 --- a/src/backend/executor/cypher_merge.c +++ b/src/backend/executor/cypher_merge.c @@ -120,6 +120,10 @@ static void begin_cypher_merge(CustomScanState *node, EState *estate, /* TODO is this necessary? Removing it seems to not have an impact */ ExecAssignExprContext(estate, &node->ss.ps); + /* Build the lookup cache only when MERGE can execute SET items. */ + if (css->on_match_set_info != NULL || css->on_create_set_info != NULL) + css->entity_lookup_cache = create_entity_lookup_cache(); + ExecInitScanTupleSlot(estate, &node->ss, ExecGetResultType(node->ss.ps.lefttree), &TTSOpsVirtual); @@ -1056,6 +1060,9 @@ static void end_cypher_merge(CustomScanState *node) increment_graph_version(css->graph_oid); } + /* Release cached index handles and fetch slots. */ + destroy_entity_lookup_cache(css->entity_lookup_cache); + ExecEndNode(node->ss.ps.lefttree); foreach (lc, path->target_nodes) diff --git a/src/backend/executor/cypher_set.c b/src/backend/executor/cypher_set.c index 7a0d48f0c..f7e7f94f3 100644 --- a/src/backend/executor/cypher_set.c +++ b/src/backend/executor/cypher_set.c @@ -25,10 +25,11 @@ #include "storage/bufmgr.h" #include "utils/rls.h" +#include "catalog/ag_graph.h" +#include "catalog/ag_label.h" #include "executor/cypher_executor.h" #include "executor/cypher_utils.h" #include "utils/age_global_graph.h" -#include "catalog/ag_graph.h" #include "utils/agtype.h" static void begin_cypher_set(CustomScanState *node, EState *estate, @@ -93,6 +94,9 @@ static void begin_cypher_set(CustomScanState *node, EState *estate, estate->es_output_cid = estate->es_snapshot->curcid; } + /* Build the per-statement index and fetch-slot lookup cache. */ + css->entity_lookup_cache = create_entity_lookup_cache(); + Increment_Estate_CommandId(estate); } @@ -245,9 +249,11 @@ 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 { + ResetExprContext(econtext); process_update_list(node); Decrement_Estate_CommandId(estate) slot = ExecProcNode(node->ss.ps.lefttree); @@ -344,15 +350,12 @@ static agtype_value *replace_entity_in_path(agtype_value *path, } /* - * When a vertex or edge is updated, we need to update the vertex - * or edge if it is contained within a path. Scan through scanTupleSlot - * to find all paths and check if they need to be updated. + * Keep every reference to an updated entity in the current row consistent. + * An entity can appear directly under another variable or within a path. */ -static void update_all_paths(CustomScanState *node, graphid id, - agtype *updated_entity) +static void update_entity_references(TupleTableSlot *scanTupleSlot, graphid id, + agtype *updated_entity) { - ExprContext *econtext = node->ss.ps.ps_ExprContext; - TupleTableSlot *scanTupleSlot = econtext->ecxt_scantuple; int i; for (i = 0; i < scanTupleSlot->tts_tupleDescriptor->natts; i++) @@ -374,7 +377,7 @@ static void update_all_paths(CustomScanState *node, graphid id, original_entity = DATUM_GET_AGTYPE_P(scanTupleSlot->tts_values[i]); - /* if the value is not a scalar type, its not a path */ + /* Vertices, edges, and paths are represented as scalar agtype values. */ if (!AGTYPE_CONTAINER_IS_SCALAR(&original_entity->root)) { continue; @@ -382,8 +385,19 @@ static void update_all_paths(CustomScanState *node, graphid id, original_entity_value = get_ith_agtype_value_from_container(&original_entity->root, 0); - /* we found a path */ - if (original_entity_value->type == AGTV_PATH) + if (original_entity_value->type == AGTV_VERTEX || + original_entity_value->type == AGTV_EDGE) + { + agtype_value *original_id = GET_AGTYPE_VALUE_OBJECT_VALUE( + original_entity_value, "id"); + + if (original_id->val.int_value == id) + { + scanTupleSlot->tts_values[i] = + AGTYPE_P_GET_DATUM(updated_entity); + } + } + else if (original_entity_value->type == AGTV_PATH) { /* check if the path contains the entity. */ if (check_path(original_entity_value, id)) @@ -397,6 +411,49 @@ static void update_all_paths(CustomScanState *node, graphid id, } } +static agtype_value *get_tuple_properties(Relation rel, HeapTuple tuple, + bool is_vertex) +{ + AttrNumber properties_attribute; + agtype_iterator *iterator; + agtype_iterator_token token; + agtype_parse_state *parse_state = NULL; + agtype_value value; + agtype_value *properties = NULL; + agtype *properties_agtype; + Datum properties_datum; + bool isnull; + + properties_attribute = is_vertex + ? Anum_ag_label_vertex_table_properties + : Anum_ag_label_edge_table_properties; + properties_datum = heap_getattr(tuple, properties_attribute, + RelationGetDescr(rel), &isnull); + if (isnull) + { + return NULL; + } + + properties_agtype = DATUM_GET_AGTYPE_P(properties_datum); + iterator = agtype_iterator_init(&properties_agtype->root); + while ((token = agtype_iterator_next(&iterator, &value, true)) != + WAGT_DONE) + { + properties = push_agtype_value( + &parse_state, token, + token < WAGT_BEGIN_ARRAY ? &value : NULL); + } + + if (properties == NULL || properties->type != AGTV_OBJECT) + { + ereport(ERROR, + (errcode(ERRCODE_DATA_EXCEPTION), + errmsg("entity properties must be an agtype object"))); + } + + return properties; +} + /* * Core SET logic that can be called from any executor (SET, MERGE, etc.). * Takes the CustomScanState for expression context and a @@ -405,19 +462,36 @@ static void update_all_paths(CustomScanState *node, graphid id, void apply_update_list(CustomScanState *node, cypher_update_information *set_info) { + EntityLookupCache *entity_lookup_cache = NULL; ExprContext *econtext = node->ss.ps.ps_ExprContext; TupleTableSlot *scanTupleSlot = econtext->ecxt_scantuple; ListCell *lc; EState *estate = node->ss.ps.state; int *luindex = NULL; + bool *seen_entity_positions = NULL; int lidx = 0; HTAB *qual_cache = NULL; HASHCTL hashctl; - HTAB *index_cache = NULL; - HASHCTL idx_hashctl; + + if (node->methods == &cypher_set_exec_methods) + { + cypher_set_custom_scan_state *css = + (cypher_set_custom_scan_state *)node; + + entity_lookup_cache = css->entity_lookup_cache; + } + else if (node->methods == &cypher_merge_exec_methods) + { + cypher_merge_custom_scan_state *css = + (cypher_merge_custom_scan_state *)node; + + entity_lookup_cache = css->entity_lookup_cache; + } /* allocate an array to hold the last update index of each 'entity' */ luindex = palloc0(sizeof(int) * scanTupleSlot->tts_nvalid); + seen_entity_positions = + palloc0(sizeof(bool) * scanTupleSlot->tts_nvalid); /* Hash table for caching compiled security quals per label */ MemSet(&hashctl, 0, sizeof(hashctl)); @@ -427,13 +501,6 @@ void apply_update_list(CustomScanState *node, qual_cache = hash_create("update_qual_cache", 8, &hashctl, HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); - MemSet(&idx_hashctl, 0, sizeof(idx_hashctl)); - idx_hashctl.keysize = sizeof(Oid); - idx_hashctl.entrysize = sizeof(IndexCacheEntry); - idx_hashctl.hcxt = CurrentMemoryContext; - index_cache = hash_create("update_index_cache", 8, &idx_hashctl, - HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); - /* * Iterate through the SET items list and store the loop index of each * 'entity' update. As there is only one entry for each entity, this will @@ -469,20 +536,18 @@ void apply_update_list(CustomScanState *node, agtype *new_property_value = NULL; TupleTableSlot *slot; ResultRelInfo *resultRelInfo; - ScanKeyData scan_keys[1]; - TableScanDesc scan_desc; bool remove_property; char *label_name; cypher_update_item *update_item; Datum new_entity; HeapTuple heap_tuple; + HeapTuple found_tuple = NULL; char *clause_name = set_info->clause_name; int cid; - Oid index_oid = InvalidOid; Relation rel; Oid relid; - IndexCacheEntry *idx_entry; - bool found_idx_entry; + bool first_update; + bool last_update; update_item = (cypher_update_item *)lfirst(lc); @@ -584,47 +649,12 @@ void apply_update_list(CustomScanState *node, new_property_value = DATUM_GET_AGTYPE_P(scanTupleSlot->tts_values[update_item->prop_position - 1]); } - /* Alter the properties Agtype value. */ - if (update_item->prop_name != NULL && - strcmp(update_item->prop_name, "") != 0) - { - altered_properties = alter_property_value(original_properties, - update_item->prop_name, - new_property_value, - remove_property); - } - else - { - altered_properties = alter_properties( - update_item->is_add ? original_properties : NULL, - new_property_value); - - /* - * For SET clause with plus-equal operator, nulls are not removed - * from the map during transformation because they are required in - * the executor to alter (merge) properties correctly. Only after - * that step, they can be removed. - */ - if (update_item->is_add) - { - remove_null_from_agtype_object(altered_properties); - } - } - resultRelInfo = create_entity_result_rel_info( estate, set_info->graph_name, label_name); rel = resultRelInfo->ri_RelationDesc; relid = RelationGetRelid(rel); - idx_entry = hash_search(index_cache, &relid, HASH_ENTER, &found_idx_entry); - if (!found_idx_entry) - { - /* Check if there is a valid index on the 'id' column */ - idx_entry->index_oid = find_usable_btree_index_for_attr(rel, 1); - } - index_oid = idx_entry->index_oid; - slot = ExecInitExtraTupleSlot( estate, RelationGetDescr(resultRelInfo->ri_RelationDesc), &TTSOpsHeapTuple); @@ -660,6 +690,69 @@ void apply_update_list(CustomScanState *node, } } + first_update = + !seen_entity_positions[update_item->entity_position - 1]; + last_update = + luindex[update_item->entity_position - 1] == lidx; + + cid = estate->es_snapshot->curcid; + estate->es_snapshot->curcid = GetCurrentCommandId(false); + + /* + * The entity value carried by an alias may predate an earlier SET + * clause in the same query. Use the current heap tuple as the base for + * the first update of each entity position. The last update also needs + * the tuple for the physical write, so reuse that lookup below. + */ + if (first_update || last_update) + { + found_tuple = find_entity_tuple( + rel, estate->es_snapshot, id->val.int_value, scanTupleSlot, + update_item->ctid_position, entity_lookup_cache); + } + if (first_update && HeapTupleIsValid(found_tuple)) + { + ItemPointerData ctid_hint; + + if (!get_entity_ctid_hint(scanTupleSlot, + update_item->ctid_position, + &ctid_hint) || + !ItemPointerEquals(&ctid_hint, &found_tuple->t_self)) + { + original_properties = get_tuple_properties( + rel, found_tuple, + original_entity_value->type == AGTV_VERTEX); + } + } + seen_entity_positions[update_item->entity_position - 1] = true; + + /* Alter the properties Agtype value. */ + if (update_item->prop_name != NULL && + strcmp(update_item->prop_name, "") != 0) + { + altered_properties = alter_property_value(original_properties, + update_item->prop_name, + new_property_value, + remove_property); + } + else + { + altered_properties = alter_properties( + update_item->is_add ? original_properties : NULL, + new_property_value); + + /* + * For SET clause with plus-equal operator, nulls are not removed + * from the map during transformation because they are required in + * the executor to alter (merge) properties correctly. Only after + * that step, they can be removed. + */ + if (update_item->is_add) + { + remove_null_from_agtype_object(altered_properties); + } + } + /* * Now that we have the updated properties, create a either a vertex or * edge Datum for the in-memory update, and setup the tupleTableSlot @@ -697,153 +790,69 @@ void apply_update_list(CustomScanState *node, /* place the datum in its tuple table slot position. */ scanTupleSlot->tts_values[update_item->entity_position - 1] = new_entity; - /* - * If the tuple table slot has paths, we need to inspect them to see if - * the updated entity is contained within them and replace the entity - * if it is. - */ - update_all_paths(node, - id->val.int_value, DATUM_GET_AGTYPE_P(new_entity)); + /* Keep aliases and paths in the current row in sync. */ + update_entity_references(scanTupleSlot, id->val.int_value, + DATUM_GET_AGTYPE_P(new_entity)); /* * If the last update index for the entity is equal to the current loop * index, then update this tuple. */ - cid = estate->es_snapshot->curcid; - estate->es_snapshot->curcid = GetCurrentCommandId(false); - - if (luindex[update_item->entity_position - 1] == lidx) + if (last_update) { - if (OidIsValid(index_oid)) - { - Relation index_rel; - IndexScanDesc idx_scan_desc; - TupleTableSlot *index_slot; - - index_rel = index_open(index_oid, RowExclusiveLock); - - /* - * Setup the scan key to require the id field on-disc to match the - * entity's graphid. - */ - ScanKeyInit(&scan_keys[0], 1, BTEqualStrategyNumber, F_GRAPHIDEQ, - GRAPHID_GET_DATUM(id->val.int_value)); - - index_slot = table_slot_create(rel, NULL); - idx_scan_desc = index_beginscan(rel, index_rel, estate->es_snapshot, NULL, 1, 0); - index_rescan(idx_scan_desc, scan_keys, 1, NULL, 0); + if (HeapTupleIsValid(found_tuple)) + { + bool should_update = true; - if (index_getnext_slot(idx_scan_desc, ForwardScanDirection, index_slot)) + if (check_enable_rls(relid, InvalidOid, true) == RLS_ENABLED) { - bool shouldFree; - - /* Retrieve the tuple from the slot */ - heap_tuple = ExecFetchSlotHeapTuple(index_slot, true, &shouldFree); - - if (HeapTupleIsValid(heap_tuple)) - { - bool should_update = true; - HeapTuple original_tuple = heap_tuple; - - /* Check RLS security quals (USING policy) before update */ - if (check_enable_rls(relid, InvalidOid, true) == RLS_ENABLED) - { - RLSCacheEntry *entry; - - /* Entry was already created earlier when setting up WCOs */ - entry = hash_search(qual_cache, &relid, HASH_FIND, NULL); - if (!entry) - { - ereport(ERROR, - (errcode(ERRCODE_INTERNAL_ERROR), - errmsg("missing RLS cache entry for relation %u", - relid))); - } - - ExecStoreHeapTuple(heap_tuple, entry->slot, false); - should_update = check_security_quals(entry->qualExprs, - entry->slot, - econtext); - } - - /* Silently skip if USING policy filters out this row */ - if (should_update) - { - heap_tuple = update_entity_tuple(resultRelInfo, slot, estate, - original_tuple); - } - - if (shouldFree) - { - heap_freetuple(original_tuple); - } - } + RLSCacheEntry *entry; + + entry = hash_search(qual_cache, &relid, HASH_FIND, NULL); + if (entry == NULL) + elog(ERROR, "missing RLS cache entry for relation %u", + relid); + ExecStoreHeapTuple(found_tuple, entry->slot, false); + should_update = check_security_quals(entry->qualExprs, + entry->slot, econtext); } - - ExecDropSingleTupleTableSlot(index_slot); - index_endscan(idx_scan_desc); - index_close(index_rel, RowExclusiveLock); - } - else - { - /* - * Setup the scan key to require the id field on-disc to match the - * entity's graphid. - */ - ScanKeyInit(&scan_keys[0], 1, BTEqualStrategyNumber, F_GRAPHIDEQ, - GRAPHID_GET_DATUM(id->val.int_value)); - /* - * Setup the scan description, with the correct snapshot and scan - * keys. - */ - scan_desc = table_beginscan(resultRelInfo->ri_RelationDesc, - estate->es_snapshot, 1, scan_keys); - /* Retrieve the tuple. */ - heap_tuple = heap_getnext(scan_desc, ForwardScanDirection); - - /* - * If the heap tuple still exists (It wasn't deleted between the - * match and this SET/REMOVE) update the heap_tuple. - */ - if (HeapTupleIsValid(heap_tuple)) + if (should_update) { - bool should_update = true; - - /* Check RLS security quals (USING policy) before update */ - if (check_enable_rls(relid, InvalidOid, true) == RLS_ENABLED) - { - RLSCacheEntry *entry; - - /* Entry was already created earlier when setting up WCOs */ - entry = hash_search(qual_cache, &relid, HASH_FIND, NULL); - if (!entry) - { - ereport(ERROR, - (errcode(ERRCODE_INTERNAL_ERROR), - errmsg("missing RLS cache entry for relation %u", - relid))); - } - - ExecStoreHeapTuple(heap_tuple, entry->slot, false); - should_update = check_security_quals(entry->qualExprs, - entry->slot, - econtext); - } - - /* Silently skip if USING policy filters out this row */ - if (should_update) + heap_tuple = update_entity_tuple(resultRelInfo, slot, estate, + found_tuple); + if (heap_tuple != NULL && + update_item->ctid_position > 0 && + update_item->ctid_position <= + scanTupleSlot->tts_tupleDescriptor->natts) { - heap_tuple = update_entity_tuple(resultRelInfo, slot, estate, - heap_tuple); + agtype *ctid_agtype; + agtype_value ctid_value; + MemoryContext old; + + ctid_value.type = AGTV_INTEGER; + ctid_value.val.int_value = AGE_CTID_PACK( + ItemPointerGetBlockNumber(&heap_tuple->t_self), + ItemPointerGetOffsetNumber(&heap_tuple->t_self)); + old = MemoryContextSwitchTo( + econtext->ecxt_per_tuple_memory); + ctid_agtype = agtype_value_to_agtype(&ctid_value); + MemoryContextSwitchTo(old); + scanTupleSlot->tts_values[ + update_item->ctid_position - 1] = + AGTYPE_P_GET_DATUM(ctid_agtype); + scanTupleSlot->tts_isnull[ + update_item->ctid_position - 1] = false; } } - /* close the ScanDescription */ - table_endscan(scan_desc); } } + if (HeapTupleIsValid(found_tuple)) + { + heap_freetuple(found_tuple); + } estate->es_snapshot->curcid = cid; - /* close relation */ + /* close relation */ ExecCloseIndices(resultRelInfo); table_close(resultRelInfo->ri_RelationDesc, RowExclusiveLock); @@ -853,10 +862,10 @@ void apply_update_list(CustomScanState *node, /* Clean up the cache */ hash_destroy(qual_cache); - hash_destroy(index_cache); /* free our lookup array */ pfree_if_not_null(luindex); + pfree_if_not_null(seen_entity_positions); } static void process_update_list(CustomScanState *node) @@ -904,6 +913,7 @@ static TupleTableSlot *exec_cypher_set(CustomScanState *node) return NULL; } + ResetExprContext(econtext); process_update_list(node); /* increment the command counter to reflect the updates */ @@ -921,6 +931,10 @@ static TupleTableSlot *exec_cypher_set(CustomScanState *node) static void end_cypher_set(CustomScanState *node) { + /* Release cached index handles and fetch slots. */ + destroy_entity_lookup_cache( + ((cypher_set_custom_scan_state *)node)->entity_lookup_cache); + ExecEndNode(node->ss.ps.lefttree); } diff --git a/src/backend/executor/cypher_utils.c b/src/backend/executor/cypher_utils.c index c697a560c..4f9c49d84 100644 --- a/src/backend/executor/cypher_utils.c +++ b/src/backend/executor/cypher_utils.c @@ -24,6 +24,8 @@ #include "postgres.h" +#include "access/genam.h" +#include "access/tableam.h" #include "executor/executor.h" #include "executor/nodeModifyTable.h" #include "miscadmin.h" @@ -32,6 +34,9 @@ #include "rewrite/rewriteManip.h" #include "rewrite/rowsecurity.h" #include "utils/acl.h" +#include "utils/hsearch.h" +#include "utils/memutils.h" +#include "utils/relcache.h" #include "utils/rls.h" #include "catalog/ag_label.h" @@ -39,6 +44,11 @@ #include "executor/cypher_utils.h" #include "utils/ag_cache.h" +StaticAssertDecl(Anum_ag_label_vertex_table_id == + Anum_ag_label_edge_table_id, + "vertex and edge id columns must have the same attribute " + "number"); + /* RLS helper function declarations */ static void get_policies_for_relation(Relation relation, CmdType cmd, Oid user_id, List **permissive_policies, @@ -218,6 +228,327 @@ TupleTableSlot *populate_edge_tts( return elemTupleSlot; } +/* Create the per-statement lookup cache in its own memory context. */ +EntityLookupCache *create_entity_lookup_cache(void) +{ + EntityLookupCache *cache; + MemoryContext mcxt; + MemoryContext old; + HASHCTL hashctl; + + mcxt = AllocSetContextCreate(CurrentMemoryContext, + "AGE entity lookup cache", + ALLOCSET_SMALL_SIZES); + old = MemoryContextSwitchTo(mcxt); + + cache = palloc0(sizeof(EntityLookupCache)); + cache->mcxt = mcxt; + + MemSet(&hashctl, 0, sizeof(hashctl)); + hashctl.keysize = sizeof(Oid); + hashctl.entrysize = sizeof(EntityLookupCacheEntry); + hashctl.hcxt = mcxt; + cache->htab = hash_create("AGE entity lookup htab", 8, &hashctl, + HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); + + MemoryContextSwitchTo(old); + return cache; +} + +/* Release slots and index handles before deleting the memory context. */ +void destroy_entity_lookup_cache(EntityLookupCache *cache) +{ + HASH_SEQ_STATUS seq; + EntityLookupCacheEntry *entry; + MemoryContext mcxt; + + if (cache == NULL) + return; + + mcxt = cache->mcxt; + hash_seq_init(&seq, cache->htab); + while ((entry = (EntityLookupCacheEntry *) hash_seq_search(&seq)) != NULL) + { + if (entry->fetch_slot != NULL) + ExecDropSingleTupleTableSlot(entry->fetch_slot); + if (entry->id_index != NULL) + index_close(entry->id_index, AccessShareLock); + } + + MemoryContextDelete(mcxt); +} + +static EntityLookupCacheEntry *get_entity_lookup_cache_entry( + EntityLookupCache *cache, Relation rel) +{ + Oid relid = RelationGetRelid(rel); + bool found; + EntityLookupCacheEntry *entry; + + entry = hash_search(cache->htab, &relid, HASH_ENTER, &found); + if (!found) + { + entry->relid = relid; + entry->index_initialized = false; + entry->fetch_slot = NULL; + entry->id_index = NULL; + } + + return entry; +} + +bool get_entity_ctid_hint(TupleTableSlot *scanTupleSlot, + AttrNumber ctid_position, ItemPointer ctid) +{ + agtype *ctid_agt; + agtype_value *ctid_val; + OffsetNumber ctid_offset; + bool ctid_isnull; + Datum ctid_datum; + int64 packed; + + if (ctid_position <= 0 || scanTupleSlot == NULL || + ctid_position > scanTupleSlot->tts_tupleDescriptor->natts) + return false; + + ctid_datum = slot_getattr(scanTupleSlot, ctid_position, &ctid_isnull); + if (ctid_isnull) + return false; + + /* + * These checks validate the internal ctid transport format. A + * mismatch here means parser/targetlist metadata is broken, not that the + * hint merely went stale, so fail the statement instead of falling back. + */ + if (TupleDescAttr(scanTupleSlot->tts_tupleDescriptor, + ctid_position - 1)->atttypid != AGTYPEOID) + elog(ERROR, "unexpected ctid hint type"); + + ctid_agt = DATUM_GET_AGTYPE_P(ctid_datum); + if (!AGTYPE_CONTAINER_IS_SCALAR(&ctid_agt->root)) + elog(ERROR, "unexpected non-scalar ctid hint"); + + ctid_val = get_ith_agtype_value_from_container(&ctid_agt->root, 0); + if (ctid_val == NULL || ctid_val->type != AGTV_INTEGER) + elog(ERROR, "unexpected non-integer ctid hint"); + + packed = ctid_val->val.int_value; + ctid_offset = AGE_CTID_UNPACK_OFFSET(packed); + if (ctid_offset == InvalidOffsetNumber) + elog(ERROR, "unexpected invalid ctid hint offset"); + + ItemPointerSetBlockNumber(ctid, AGE_CTID_UNPACK_BLOCK(packed)); + ItemPointerSetOffsetNumber(ctid, ctid_offset); + + return true; +} + +/* Try a tuple lookup using the hidden packed ctid hint. */ +static HeapTuple try_ctid_fetch_tuple(Relation rel, Snapshot snapshot, + graphid entity_id, + TupleTableSlot *scanTupleSlot, + AttrNumber ctid_position, + EntityLookupCache *cache) +{ + HeapTuple result = NULL; + ItemPointerData ctid_data; + EntityLookupCacheEntry *entry = NULL; + TupleTableSlot *fetch_slot = NULL; + TupleTableSlot *slot; + bool id_isnull; + Datum id_datum; + + if (!get_entity_ctid_hint(scanTupleSlot, ctid_position, &ctid_data)) + return NULL; + + if (cache != NULL) + { + entry = get_entity_lookup_cache_entry(cache, rel); + if (entry->fetch_slot == NULL) + { + MemoryContext old = MemoryContextSwitchTo(cache->mcxt); + + entry->fetch_slot = MakeSingleTupleTableSlot( + RelationGetDescr(rel), &TTSOpsBufferHeapTuple); + MemoryContextSwitchTo(old); + } + fetch_slot = entry->fetch_slot; + } + + slot = fetch_slot; + if (slot == NULL) + slot = MakeSingleTupleTableSlot(RelationGetDescr(rel), + &TTSOpsBufferHeapTuple); + else + ExecClearTuple(slot); + + if (table_tuple_fetch_row_version(rel, &ctid_data, snapshot, slot)) + { + id_datum = slot_getattr(slot, Anum_ag_label_vertex_table_id, + &id_isnull); + + /* + * table_tuple_fetch_row_version() only proves that the tuple at + * this TID is visible to the current snapshot. AGE writable clauses do + * not run the full PG UPDATE/DELETE recheck chain here, so Tier-1 must + * still verify logical identity itself. The ctid hint can be unreliable + * when an earlier writable clause in the same statement has already + * moved the tuple version, or when parser/alias propagation paired a + * well-formed but wrong ctid with this entity. Since ctid is only a + * physical locator, a visible tuple fetched here must still match + * entity_id before Tier-1 can use it. + */ + if (id_isnull || DATUM_GET_GRAPHID(id_datum) != entity_id) + result = NULL; + else + result = heap_copytuple( + ExecFetchSlotHeapTuple(slot, false, NULL)); + } + + if (fetch_slot == NULL) + ExecDropSingleTupleTableSlot(slot); + else + ExecClearTuple(slot); + + return result; +} + +/* + * Try ctid lookup first, then a usable B-tree index on id. Use SeqScan only + * when the relation has no such index. + * + * The ctid hint must come from the same statement. If the tuple at that ctid + * is no longer visible, fall back to graphid lookup. A visible tuple fetched + * by ctid must still match entity_id. + * + * Returns a heap_copytuple'd tuple, or NULL if not found. + * The caller must heap_freetuple() the result when done. + */ +HeapTuple find_entity_tuple(Relation rel, Snapshot snapshot, + graphid entity_id, + TupleTableSlot *scanTupleSlot, + AttrNumber ctid_position, + EntityLookupCache *cache) +{ + HeapTuple result = NULL; + EntityLookupCacheEntry *entry = NULL; + TupleTableSlot *fetch_slot = NULL; + Relation id_index = NULL; + bool own_index = false; + + /* Tier 1: ctid direct fetch */ + if (ctid_position > 0 && scanTupleSlot != NULL && + ctid_position <= scanTupleSlot->tts_tupleDescriptor->natts) + { + result = try_ctid_fetch_tuple(rel, snapshot, entity_id, + scanTupleSlot, ctid_position, cache); + if (result != NULL) + return result; + } + + /* + * Tier 2: use any valid, non-partial B-tree index whose first key is id. + * Edge labels do not inherit a primary key, but users can create an id + * index explicitly. + */ + if (cache != NULL) + { + entry = get_entity_lookup_cache_entry(cache, rel); + if (!entry->index_initialized) + { + Oid index_oid; + + index_oid = find_usable_btree_index_for_attr( + rel, Anum_ag_label_vertex_table_id); + if (OidIsValid(index_oid)) + entry->id_index = index_open(index_oid, AccessShareLock); + entry->index_initialized = true; + } + id_index = entry->id_index; + } + else + { + Oid index_oid; + + index_oid = find_usable_btree_index_for_attr( + rel, Anum_ag_label_vertex_table_id); + if (OidIsValid(index_oid)) + { + id_index = index_open(index_oid, AccessShareLock); + own_index = true; + } + } + + if (id_index != NULL) + { + ScanKeyData index_keys[1]; + IndexScanDesc index_scan; + TupleTableSlot *slot; + + /* Attribute 1 is the first key of the selected id index. */ + ScanKeyInit(&index_keys[0], 1, BTEqualStrategyNumber, F_GRAPHIDEQ, + GRAPHID_GET_DATUM(entity_id)); + + if (cache != NULL) + { + if (entry->fetch_slot == NULL) + { + MemoryContext old = MemoryContextSwitchTo(cache->mcxt); + + entry->fetch_slot = MakeSingleTupleTableSlot( + RelationGetDescr(rel), &TTSOpsBufferHeapTuple); + MemoryContextSwitchTo(old); + } + fetch_slot = entry->fetch_slot; + } + + slot = fetch_slot; + if (slot == NULL) + slot = MakeSingleTupleTableSlot(RelationGetDescr(rel), + &TTSOpsBufferHeapTuple); + else + ExecClearTuple(slot); + + index_scan = index_beginscan(rel, id_index, snapshot, NULL, 1, 0); + index_rescan(index_scan, index_keys, 1, NULL, 0); + + if (index_getnext_slot(index_scan, ForwardScanDirection, slot)) + result = heap_copytuple(ExecFetchSlotHeapTuple(slot, false, NULL)); + + index_endscan(index_scan); + + if (fetch_slot == NULL) + ExecDropSingleTupleTableSlot(slot); + else + ExecClearTuple(slot); + if (own_index) + index_close(id_index, AccessShareLock); + + return result; + } + + /* Tier 3: SeqScan fallback for relations without a usable id index. */ + { + ScanKeyData heap_keys[1]; + TableScanDesc scan_desc; + HeapTuple tuple; + + /* attno here is the table column number of the id column */ + ScanKeyInit(&heap_keys[0], Anum_ag_label_vertex_table_id, + BTEqualStrategyNumber, F_GRAPHIDEQ, + GRAPHID_GET_DATUM(entity_id)); + + scan_desc = table_beginscan(rel, snapshot, 1, heap_keys); + tuple = heap_getnext(scan_desc, ForwardScanDirection); + + if (HeapTupleIsValid(tuple)) + result = heap_copytuple(tuple); + + table_endscan(scan_desc); + } + + return result; +} /* * Find out if the entity still exists. This is for 'implicit' deletion diff --git a/src/backend/nodes/cypher_copyfuncs.c b/src/backend/nodes/cypher_copyfuncs.c index 549218759..666d8def7 100644 --- a/src/backend/nodes/cypher_copyfuncs.c +++ b/src/backend/nodes/cypher_copyfuncs.c @@ -131,6 +131,9 @@ void copy_cypher_update_item(ExtensibleNode *newnode, const ExtensibleNode *from COPY_SCALAR_FIELD(prop_position); COPY_SCALAR_FIELD(entity_position); + + /* copy ctid position for direct tuple fetch */ + COPY_SCALAR_FIELD(ctid_position); COPY_STRING_FIELD(var_name); COPY_STRING_FIELD(prop_name); COPY_NODE_FIELD(qualified_name); @@ -159,6 +162,9 @@ void copy_cypher_delete_item(ExtensibleNode *newnode, const ExtensibleNode *from COPY_NODE_FIELD(entity_position); COPY_STRING_FIELD(var_name); + + /* copy ctid position for direct tuple fetch */ + COPY_SCALAR_FIELD(ctid_position); } /* copy function for cypher_merge_information */ diff --git a/src/backend/nodes/cypher_outfuncs.c b/src/backend/nodes/cypher_outfuncs.c index 4a35be02f..101b27cf7 100644 --- a/src/backend/nodes/cypher_outfuncs.c +++ b/src/backend/nodes/cypher_outfuncs.c @@ -109,6 +109,9 @@ void out_cypher_return(StringInfo str, const ExtensibleNode *node) WRITE_ENUM_FIELD(op, SetOperation); WRITE_NODE_FIELD(larg); WRITE_NODE_FIELD(rarg); + + /* serialize is-WITH flag for ctid propagation through WITH */ + WRITE_BOOL_FIELD(is_with); } /* serialization function for the cypher_with ExtensibleNode. */ @@ -447,6 +450,9 @@ void out_cypher_update_item(StringInfo str, const ExtensibleNode *node) WRITE_INT32_FIELD(prop_position); WRITE_INT32_FIELD(entity_position); + + /* serialize ctid position for direct tuple fetch */ + WRITE_INT32_FIELD(ctid_position); WRITE_STRING_FIELD(var_name); WRITE_STRING_FIELD(prop_name); WRITE_NODE_FIELD(qualified_name); @@ -475,6 +481,9 @@ void out_cypher_delete_item(StringInfo str, const ExtensibleNode *node) WRITE_NODE_FIELD(entity_position); WRITE_STRING_FIELD(var_name); + + /* serialize ctid position for direct tuple fetch */ + WRITE_INT32_FIELD(ctid_position); } /* serialization function for the cypher_merge_information ExtensibleNode. */ diff --git a/src/backend/nodes/cypher_readfuncs.c b/src/backend/nodes/cypher_readfuncs.c index a9a2ffabd..3d735c84e 100644 --- a/src/backend/nodes/cypher_readfuncs.c +++ b/src/backend/nodes/cypher_readfuncs.c @@ -264,6 +264,9 @@ void read_cypher_update_item(struct ExtensibleNode *node) READ_INT_FIELD(prop_position); READ_INT_FIELD(entity_position); + + /* deserialize ctid position for direct tuple fetch */ + READ_INT_FIELD(ctid_position); READ_STRING_FIELD(var_name); READ_STRING_FIELD(prop_name); READ_NODE_FIELD(qualified_name); @@ -298,6 +301,9 @@ void read_cypher_delete_item(struct ExtensibleNode *node) READ_NODE_FIELD(entity_position); READ_STRING_FIELD(var_name); + + /* deserialize ctid position for direct tuple fetch */ + READ_INT_FIELD(ctid_position); } /* diff --git a/src/backend/parser/cypher_analyze.c b/src/backend/parser/cypher_analyze.c index 5dd53dcd0..eb76fd2f0 100644 --- a/src/backend/parser/cypher_analyze.c +++ b/src/backend/parser/cypher_analyze.c @@ -1009,6 +1009,25 @@ static Query *analyze_cypher(List *stmt, ParseState *parent_pstate, cpstate->default_alias_num = 0; cpstate->entities = NIL; cpstate->subquery_where_flag = false; + + /* + * Inject hidden ctid columns during MATCH only when the statement contains + * a clause that needs to relocate entity tuples. Read-only statements skip + * the extra columns entirely. + */ + cpstate->has_writable_clause = false; + foreach (lc, stmt) + { + Node *clause_node = lfirst(lc); + + if (is_ag_node(clause_node, cypher_set) || + is_ag_node(clause_node, cypher_delete)) + { + cpstate->has_writable_clause = true; + break; + } + } + /* * install error context callback to adjust an error position since * locations in stmt are 0 based diff --git a/src/backend/parser/cypher_clause.c b/src/backend/parser/cypher_clause.c index 147e3e74e..a045c42a8 100644 --- a/src/backend/parser/cypher_clause.c +++ b/src/backend/parser/cypher_clause.c @@ -80,6 +80,7 @@ #define AGE_VARNAME_ID AGE_DEFAULT_VARNAME_PREFIX"id" #define AGE_VARNAME_SET_CLAUSE AGE_DEFAULT_VARNAME_PREFIX"set_clause" #define AGE_VARNAME_SET_VALUE AGE_DEFAULT_VARNAME_PREFIX"set_value" +#define AGE_VARNAME_CTID_FORMAT AGE_DEFAULT_VARNAME_PREFIX"ctid_%s" /* * In the transformation stage, we need to track @@ -266,6 +267,18 @@ static Expr *add_volatile_wrapper(Expr *node); static bool variable_exists(cypher_parsestate *cpstate, char *name); static void add_volatile_wrapper_to_target_entry(List *target_list, int resno); static int get_target_entry_resno(List *target_list, char *name); + +/* Helpers for injecting and resolving ctid columns for direct tuple fetch. */ +static AttrNumber resolve_ctid_position(List *target_list, + const char *var_name); +static bool is_internal_name(const char *name); +static char *get_ctid_source_name(ParseState *pstate, TargetEntry *te); +static void inject_ctid_target_entry(ParseState *pstate, + ParseNamespaceItem *pnsi, + const char *entity_name, + List **target_list); +static void inject_ctid_columns_for_with(ParseState *pstate, Query *query); + static void handle_prev_clause(cypher_parsestate *cpstate, Query *query, cypher_clause *clause, bool first_rte); static TargetEntry *placeholder_target_entry(cypher_parsestate *cpstate, @@ -2853,6 +2866,9 @@ static List *transform_cypher_delete_item_list(cypher_parsestate *cpstate, item->var_name = val->sval; item->entity_position = pos; + /* Resolve the ctid attribute position for direct tuple fetch. */ + item->ctid_position = resolve_ctid_position(query->targetList, + val->sval); items = lappend(items, item); } @@ -3038,6 +3054,9 @@ cypher_update_information *transform_cypher_remove_item_list( property_name = property_node->sval; item->prop_name = property_name; + /* Resolve the ctid attribute position for direct tuple fetch. */ + item->ctid_position = resolve_ctid_position(query->targetList, + variable_name); info->set_items = lappend(info->set_items, item); } @@ -3235,6 +3254,10 @@ cypher_update_information *transform_cypher_set_item_list( target_item->expr = add_volatile_wrapper(target_item->expr); query->targetList = lappend(query->targetList, target_item); + + /* Resolve the ctid attribute position for direct tuple fetch. */ + item->ctid_position = resolve_ctid_position(query->targetList, + variable_name); info->set_items = lappend(info->set_items, item); } @@ -3520,6 +3543,15 @@ static Query *transform_cypher_return(cypher_parsestate *cpstate, &groupClause, EXPR_KIND_SELECT_TARGET); + /* + * Propagate ctid columns through WITH only when the statement contains SET + * or DELETE. Skip aggregation and DISTINCT because hidden ctid must not + * affect grouping or deduplication semantics. + */ + if (self->is_with && cpstate->has_writable_clause && + !pstate->p_hasAggs && groupClause == NIL && !self->distinct) + inject_ctid_columns_for_with(pstate, query); + markTargetListOrigins(pstate, query->targetList); /* ORDER BY */ @@ -3700,6 +3732,8 @@ static Query *transform_cypher_with(cypher_parsestate *cpstate, return_clause->skip = self->skip; return_clause->limit = self->limit; + /* Mark as WITH so ctid columns propagate to downstream clauses. */ + return_clause->is_with = true; wrapper = palloc(sizeof(*wrapper)); wrapper->self = (Node *)return_clause; wrapper->prev = clause->prev; @@ -6875,6 +6909,14 @@ static Expr *transform_cypher_edge(cypher_parsestate *cpstate, { te = makeTargetEntry((Expr *)expr, resno, rel->name, false); *target_list = lappend(*target_list, te); + + /* + * Add a hidden ctid column for direct tuple fetch. This is only useful + * when the statement contains SET, REMOVE, or DELETE. + */ + if (valid_label && cpstate->has_writable_clause && + !is_internal_name(rel->name)) + inject_ctid_target_entry(pstate, pnsi, rel->name, target_list); } return (Expr *)expr; @@ -7163,6 +7205,13 @@ static Expr *transform_cypher_node(cypher_parsestate *cpstate, te = makeTargetEntry(expr, resno, node->name, false); *target_list = lappend(*target_list, te); + /* + * Add a hidden ctid column for direct tuple fetch. This is only useful + * when the statement contains SET, REMOVE, or DELETE. + */ + if (valid_label && cpstate->has_writable_clause && + !is_internal_name(node->name)) + inject_ctid_target_entry(pstate, pnsi, node->name, target_list); return expr; } @@ -7739,6 +7788,130 @@ static void add_volatile_wrapper_to_target_entry(List *target_list, int resno) errmsg("add_volatile_wrapper_to_target_entry: resno not found"))); } +/* Find the ctid target entry for a variable and return its resno. */ +static AttrNumber resolve_ctid_position(List *target_list, + const char *var_name) +{ + char *ctid_name; + AttrNumber pos; + + ctid_name = psprintf(AGE_VARNAME_CTID_FORMAT, var_name); + + pos = get_target_entry_resno(target_list, ctid_name); + pfree(ctid_name); + + if (pos == -1) + return 0; + + add_volatile_wrapper_to_target_entry(target_list, pos); + return pos; +} + +/* Return true when name was generated for an internal pattern entity. */ +static bool is_internal_name(const char *name) +{ + return strncmp(name, AGE_DEFAULT_PREFIX, + strlen(AGE_DEFAULT_PREFIX)) == 0; +} + +/* Resolve the source column name for hidden ctid propagation. */ +static char *get_ctid_source_name(ParseState *pstate, TargetEntry *te) +{ + /* + * Resolve the source ctid from the underlying variable rather than the + * WITH alias. + */ + if (te->expr != NULL && IsA(te->expr, Var)) + { + Var *var = (Var *)te->expr; + + if (var->varattno > 0 && var->varlevelsup == 0) + { + RangeTblEntry *rte = rt_fetch(var->varno, pstate->p_rtable); + + return get_rte_attribute_name(rte, var->varattno); + } + } + + return NULL; +} + +/* Add a hidden CTID target entry for a vertex or edge. */ +static void inject_ctid_target_entry(ParseState *pstate, + ParseNamespaceItem *pnsi, + const char *entity_name, + List **target_list) +{ + Node *ctid_var; + + ctid_var = scanNSItemForColumn(pstate, pnsi, 0, "ctid", -1); + if (ctid_var != NULL) + { + char *ctid_name = psprintf(AGE_VARNAME_CTID_FORMAT, entity_name); + TargetEntry *ctid_te = makeTargetEntry((Expr *)ctid_var, + pstate->p_next_resno++, + ctid_name, false); + *target_list = lappend(*target_list, ctid_te); + } +} + +/* Preserve hidden ctid columns across WITH for writable clauses. */ +static void inject_ctid_columns_for_with(ParseState *pstate, Query *query) +{ + List *new_entries = NIL; + ListCell *lc; + int var_prefix_len = strlen(AGE_DEFAULT_VARNAME_PREFIX); + + foreach (lc, query->targetList) + { + TargetEntry *te = (TargetEntry *)lfirst(lc); + char *source_ctid_name; + char *source_name; + char *ctid_name; + Node *ctid_var; + + if (te->resjunk || te->resname == NULL) + continue; + + if (strncmp(te->resname, AGE_DEFAULT_VARNAME_PREFIX, + var_prefix_len) == 0) + continue; + + /* + * Keep the output hidden ctid column named after te->resname, + * but resolve its value from the source variable. + */ + source_name = get_ctid_source_name(pstate, te); + if (source_name == NULL) + continue; + + ctid_name = psprintf(AGE_VARNAME_CTID_FORMAT, te->resname); + + if (get_target_entry_resno(query->targetList, ctid_name) != -1) + { + pfree(ctid_name); + continue; + } + + source_ctid_name = psprintf(AGE_VARNAME_CTID_FORMAT, source_name); + ctid_var = colNameToVar(pstate, source_ctid_name, false, -1); + pfree(source_ctid_name); + if (ctid_var != NULL) + { + TargetEntry *ctid_te = makeTargetEntry( + (Expr *)ctid_var, (AttrNumber)pstate->p_next_resno++, + ctid_name, false); + new_entries = lappend(new_entries, ctid_te); + } + else + { + pfree(ctid_name); + } + } + + query->targetList = list_concat(query->targetList, new_entries); +} + /* * Transform logic for a previously declared variable in a CREATE clause. * All we need from the variable node is its id, and whether we can skip diff --git a/src/backend/parser/cypher_parse_node.c b/src/backend/parser/cypher_parse_node.c index 48c50d4fd..aacd072e1 100644 --- a/src/backend/parser/cypher_parse_node.c +++ b/src/backend/parser/cypher_parse_node.c @@ -60,6 +60,12 @@ cypher_parsestate *make_cypher_parsestate(cypher_parsestate *parent_cpstate) cpstate->graph_oid = parent_cpstate->graph_oid; cpstate->params = parent_cpstate->params; cpstate->subquery_where_flag = parent_cpstate->subquery_where_flag; + + /* + * Keep the ctid injection decision consistent across the whole + * statement, including subquery parse states. + */ + cpstate->has_writable_clause = parent_cpstate->has_writable_clause; } return cpstate; diff --git a/src/backend/utils/adt/agtype.c b/src/backend/utils/adt/agtype.c index cc9cc7717..fa000b84a 100644 --- a/src/backend/utils/adt/agtype.c +++ b/src/backend/utils/adt/agtype.c @@ -13236,6 +13236,15 @@ Datum agtype_volatile_wrapper(PG_FUNCTION_ARGS) agtv_result.val.string.val = text_to_cstring(DatumGetTextPP(arg)); agtv_result.val.string.len = strlen(agtv_result.val.string.val); } + else if (type == TIDOID) + { + ItemPointer tid = (ItemPointer) DatumGetPointer(arg); + + agtv_result.type = AGTV_INTEGER; + agtv_result.val.int_value = + AGE_CTID_PACK(ItemPointerGetBlockNumberNoCheck(tid), + ItemPointerGetOffsetNumberNoCheck(tid)); + } else if (type == VERTEXOID) { PG_RETURN_DATUM(DirectFunctionCall1(vertex_to_agtype, arg)); diff --git a/src/include/executor/cypher_utils.h b/src/include/executor/cypher_utils.h index 8b65bc964..599bd6e7a 100644 --- a/src/include/executor/cypher_utils.h +++ b/src/include/executor/cypher_utils.h @@ -59,12 +59,33 @@ typedef struct cypher_create_custom_scan_state Oid graph_oid; } cypher_create_custom_scan_state; +/* + * Per-statement lookup cache keyed by table relid. The fetch slot and usable + * B-tree index on id are initialized lazily. + */ +typedef struct EntityLookupCacheEntry +{ + Oid relid; /* hash key */ + bool index_initialized; /* includes a negative lookup result */ + Relation id_index; /* usable id index opened once */ + TupleTableSlot *fetch_slot; /* reusable TTSOpsBufferHeapTuple slot */ +} EntityLookupCacheEntry; + +typedef struct EntityLookupCache +{ + HTAB *htab; /* relid -> EntityLookupCacheEntry */ + MemoryContext mcxt; /* context for cached handles and slots */ +} EntityLookupCache; + typedef struct cypher_set_custom_scan_state { CustomScanState css; CustomScan *cs; cypher_update_information *set_list; int flags; + + /* Per-statement index and fetch-slot cache for tuple lookup. */ + EntityLookupCache *entity_lookup_cache; } cypher_set_custom_scan_state; typedef struct cypher_delete_custom_scan_state @@ -92,6 +113,9 @@ typedef struct cypher_delete_custom_scan_state * and end_id column. */ HTAB *vertex_id_htab; + + /* Per-statement index and fetch-slot cache for tuple lookup. */ + EntityLookupCache *entity_lookup_cache; } cypher_delete_custom_scan_state; typedef struct cypher_merge_custom_scan_state @@ -113,6 +137,9 @@ typedef struct cypher_merge_custom_scan_state bool eager_buffer_filled; cypher_update_information *on_match_set_info; /* NULL if not specified */ cypher_update_information *on_create_set_info; /* NULL if not specified */ + + /* Per-statement index and fetch-slot cache for ON MATCH/CREATE SET. */ + EntityLookupCache *entity_lookup_cache; } cypher_merge_custom_scan_state; /* Reusable SET logic callable from MERGE executor */ @@ -130,6 +157,24 @@ ResultRelInfo *create_entity_result_rel_info(EState *estate, char *graph_name, char *label_name); void destroy_entity_result_rel_info(ResultRelInfo *result_rel_info); +EntityLookupCache *create_entity_lookup_cache(void); +void destroy_entity_lookup_cache(EntityLookupCache *cache); + +/* Decode a hidden packed ctid hint. Return false when no hint is available. */ +bool get_entity_ctid_hint(TupleTableSlot *scanTupleSlot, + AttrNumber ctid_position, ItemPointer ctid); + +/* + * Look up an entity by ctid, then by a usable B-tree index on id. Use a + * sequential scan only when no such index exists. cache may be NULL for + * one-shot use. + */ +HeapTuple find_entity_tuple(Relation rel, Snapshot snapshot, + graphid entity_id, + TupleTableSlot *scanTupleSlot, + AttrNumber ctid_position, + EntityLookupCache *cache); + bool entity_exists(EState *estate, Oid graph_oid, graphid id); HeapTuple insert_entity_tuple(ResultRelInfo *resultRelInfo, TupleTableSlot *elemTupleSlot, diff --git a/src/include/nodes/cypher_nodes.h b/src/include/nodes/cypher_nodes.h index 5efbe95f7..0fef81b02 100644 --- a/src/include/nodes/cypher_nodes.h +++ b/src/include/nodes/cypher_nodes.h @@ -62,6 +62,10 @@ typedef struct cypher_return SetOperation op; List *larg; /* lefthand argument of the unions */ List *rarg; /*righthand argument of the unions */ + + /* Distinguish WITH from RETURN for hidden ctid propagation. */ + bool is_with; + } cypher_return; typedef struct cypher_with @@ -487,6 +491,10 @@ typedef struct cypher_update_item ExtensibleNode extensible; AttrNumber prop_position; AttrNumber entity_position; + + /* Attribute number of the packed ctid column in the scan tuple. */ + AttrNumber ctid_position; + char *var_name; char *prop_name; List *qualified_name; @@ -512,6 +520,9 @@ typedef struct cypher_delete_item ExtensibleNode extensible; Integer *entity_position; char *var_name; + + /* Attribute number of the packed ctid column in the scan tuple. */ + AttrNumber ctid_position; } cypher_delete_item; typedef struct cypher_merge_information diff --git a/src/include/parser/cypher_parse_node.h b/src/include/parser/cypher_parse_node.h index 263ea197b..b024637c9 100644 --- a/src/include/parser/cypher_parse_node.h +++ b/src/include/parser/cypher_parse_node.h @@ -50,6 +50,13 @@ typedef struct cypher_parsestate */ bool exprHasAgg; bool p_opt_match; + + /* + * True if the Cypher statement contains SET, REMOVE, or DELETE, making + * hidden ctid columns useful during MATCH transforms. + */ + bool has_writable_clause; + } cypher_parsestate; typedef struct errpos_ecb_state diff --git a/src/include/utils/agtype.h b/src/include/utils/agtype.h index 4bddbe4f6..d12eb911d 100644 --- a/src/include/utils/agtype.h +++ b/src/include/utils/agtype.h @@ -37,6 +37,20 @@ #include "utils/graphid.h" +/* + * Pack and unpack a TID into an int64 for agtype transport. + * Layout: BlockNumber (32 bits) << 16 | OffsetNumber (16 bits). + */ +StaticAssertDecl(sizeof(OffsetNumber) == 2 && sizeof(BlockNumber) == 4, + "AGE_CTID_PACK assumes 16-bit OffsetNumber and " + "32-bit BlockNumber"); +#define AGE_CTID_PACK(blk, off) \ + (((int64)(blk) << 16) | (int64)(off)) +#define AGE_CTID_UNPACK_BLOCK(packed) \ + ((BlockNumber)((packed) >> 16)) +#define AGE_CTID_UNPACK_OFFSET(packed) \ + ((OffsetNumber)((packed) & 0xFFFF)) + /* Tokens used when sequentially processing an agtype value */ typedef enum {