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 {