| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| #include "postgres.h" |
| |
| #include "access/heapam.h" |
| #include "access/htup_details.h" |
| #include "access/multixact.h" |
| #include "access/table.h" |
| #include "access/xact.h" |
| #include "executor/tuptable.h" |
| #include "nodes/execnodes.h" |
| #include "nodes/extensible.h" |
| #include "nodes/nodes.h" |
| #include "nodes/plannodes.h" |
| #include "parser/parsetree.h" |
| #include "storage/bufmgr.h" |
| #include "utils/rel.h" |
| #include "common/hashfn.h" |
| |
| #include "catalog/ag_label.h" |
| #include "executor/cypher_executor.h" |
| #include "executor/cypher_utils.h" |
| #include "nodes/cypher_nodes.h" |
| #include "utils/agtype.h" |
| #include "utils/graphid.h" |
| |
| static void begin_cypher_delete(CustomScanState *node, EState *estate, |
| int eflags); |
| static TupleTableSlot *exec_cypher_delete(CustomScanState *node); |
| static void end_cypher_delete(CustomScanState *node); |
| static void rescan_cypher_delete(CustomScanState *node); |
| |
| static void process_delete_list(CustomScanState *node); |
| |
| static void check_for_connected_edges(CustomScanState *node); |
| static agtype_value *extract_entity(CustomScanState *node, |
| TupleTableSlot *scanTupleSlot, |
| int entity_position); |
| static void delete_entity(EState *estate, ResultRelInfo *resultRelInfo, |
| HeapTuple tuple); |
| |
| const CustomExecMethods cypher_delete_exec_methods = {DELETE_SCAN_STATE_NAME, |
| begin_cypher_delete, |
| exec_cypher_delete, |
| end_cypher_delete, |
| rescan_cypher_delete, |
| NULL, |
| NULL, |
| NULL, |
| NULL, |
| NULL, |
| NULL, |
| NULL, |
| NULL}; |
| |
| /* |
| * Initialization at the beginning of execution. Setup the child node, |
| * setup its scan tuple slot and projection info, expression context, |
| * collect metadata about visible edges, and alter the commandid for |
| * the transaction. |
| */ |
| static void begin_cypher_delete(CustomScanState *node, EState *estate, |
| int eflags) |
| { |
| cypher_delete_custom_scan_state *css = |
| (cypher_delete_custom_scan_state *)node; |
| Plan *subplan; |
| HASHCTL hashctl; |
| |
| Assert(list_length(css->cs->custom_plans) == 1); |
| |
| // setup child |
| subplan = linitial(css->cs->custom_plans); |
| node->ss.ps.lefttree = ExecInitNode(subplan, estate, eflags); |
| |
| // setup expr context |
| ExecAssignExprContext(estate, &node->ss.ps); |
| |
| // setup scan tuple slot and projection info |
| ExecInitScanTupleSlot(estate, &node->ss, |
| ExecGetResultType(node->ss.ps.lefttree), |
| &TTSOpsHeapTuple); |
| |
| if (!CYPHER_CLAUSE_IS_TERMINAL(css->flags)) |
| { |
| TupleDesc tupdesc = node->ss.ss_ScanTupleSlot->tts_tupleDescriptor; |
| |
| ExecAssignProjectionInfo(&node->ss.ps, tupdesc); |
| } |
| |
| /* |
| * Get all the labels that are visible to this delete clause at this point |
| * in the transaction. To be used later when the delete clause finds |
| * vertices. |
| */ |
| css->edge_labels = get_all_edge_labels_per_graph(estate, css->delete_data->graph_oid); |
| |
| /* init vertex_id_htab */ |
| MemSet(&hashctl, 0, sizeof(hashctl)); |
| hashctl.keysize = sizeof(graphid); |
| hashctl.entrysize = |
| sizeof(graphid); // entries are not used, but entrysize must >= keysize |
| hashctl.hash = tag_hash; |
| css->vertex_id_htab = hash_create(DELETE_VERTEX_HTAB_NAME, |
| DELETE_VERTEX_HTAB_SIZE, &hashctl, |
| HASH_ELEM | HASH_FUNCTION); |
| |
| /* |
| * Postgres does not assign the es_output_cid in queries that do |
| * not write to disk, ie: SELECT commands. We need the command id |
| * for our clauses, and we may need to initialize it. We cannot use |
| * GetCurrentCommandId because there may be other cypher clauses |
| * that have modified the command id. |
| */ |
| if (estate->es_output_cid == 0) |
| estate->es_output_cid = estate->es_snapshot->curcid; |
| |
| Increment_Estate_CommandId(estate); |
| } |
| |
| /* |
| * Called once per tuple. If this is a terminal DELETE clause |
| * process everyone of its child tuple, otherwise process the |
| * next tuple. |
| */ |
| static TupleTableSlot *exec_cypher_delete(CustomScanState *node) |
| { |
| cypher_delete_custom_scan_state *css = |
| (cypher_delete_custom_scan_state *)node; |
| EState *estate = css->css.ss.ps.state; |
| ExprContext *econtext = css->css.ss.ps.ps_ExprContext; |
| TupleTableSlot *slot; |
| |
| if (CYPHER_CLAUSE_IS_TERMINAL(css->flags)) |
| { |
| /* |
| * If the DELETE clause was the final cypher clause written |
| * then we aren't returning anything from this result node. |
| * So the exec_cypher_delete function will only be called once. |
| * Therefore we will process all tuples from the subtree at once. |
| */ |
| while(true) |
| { |
| //Process the subtree first |
| Decrement_Estate_CommandId(estate) |
| slot = ExecProcNode(node->ss.ps.lefttree); |
| Increment_Estate_CommandId(estate) |
| |
| if (TupIsNull(slot)) |
| break; |
| |
| // setup the scantuple that the process_delete_list needs |
| econtext->ecxt_scantuple = |
| node->ss.ps.lefttree->ps_ProjInfo->pi_exprContext->ecxt_scantuple; |
| |
| process_delete_list(node); |
| } |
| |
| return NULL; |
| } |
| else |
| { |
| //Process the subtree first |
| Decrement_Estate_CommandId(estate) |
| slot = ExecProcNode(node->ss.ps.lefttree); |
| Increment_Estate_CommandId(estate) |
| |
| if (TupIsNull(slot)) |
| return NULL; |
| |
| // setup the scantuple that the process_delete_list needs |
| econtext->ecxt_scantuple = |
| node->ss.ps.lefttree->ps_ProjInfo->pi_exprContext->ecxt_scantuple; |
| |
| process_delete_list(node); |
| |
| econtext->ecxt_scantuple = |
| ExecProject(node->ss.ps.lefttree->ps_ProjInfo); |
| |
| return ExecProject(node->ss.ps.ps_ProjInfo); |
| } |
| } |
| |
| /* |
| * Called at the end of execution. Tell its child to |
| * end its execution. |
| */ |
| static void end_cypher_delete(CustomScanState *node) |
| { |
| check_for_connected_edges(node); |
| |
| hash_destroy(((cypher_delete_custom_scan_state *)node)->vertex_id_htab); |
| |
| ExecEndNode(node->ss.ps.lefttree); |
| } |
| |
| /* |
| * Used for rewinding the scan state and reprocessing the results. |
| * |
| * XXX: This is not currently supported. We need to find out |
| * when this function will be called and determine a process |
| * for allowing the Delete clause to run multiple times without |
| * redundant edits to the database. |
| */ |
| static void rescan_cypher_delete(CustomScanState *node) |
| { |
| ereport(ERROR, |
| (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), |
| errmsg("cypher DELETE clause cannot be rescanned"), |
| errhint("its unsafe to use joins in a query with a Cypher DELETE clause"))); |
| } |
| |
| /* |
| * Create the CustomScanState from the CustomScan and pass |
| * necessary metadata. |
| */ |
| Node *create_cypher_delete_plan_state(CustomScan *cscan) |
| { |
| cypher_delete_custom_scan_state *cypher_css = |
| palloc0(sizeof(cypher_delete_custom_scan_state)); |
| cypher_delete_information *delete_data; |
| char *serialized_data; |
| Const *c; |
| |
| cypher_css->cs = cscan; |
| |
| // get the serialized data structure from the Const and deserialize it. |
| c = linitial(cscan->custom_private); |
| serialized_data = (char *)c->constvalue; |
| delete_data = stringToNode(serialized_data); |
| |
| Assert(is_ag_node(delete_data, cypher_delete_information)); |
| |
| cypher_css->delete_data = delete_data; |
| cypher_css->flags = delete_data->flags; |
| |
| cypher_css->css.ss.ps.type = T_CustomScanState; |
| cypher_css->css.methods = &cypher_delete_exec_methods; |
| |
| return (Node *)cypher_css; |
| } |
| |
| /* |
| * Extract the vertex or edge to be deleted, perform some type checking to |
| * validate datum is an agtype vertex or edge. |
| */ |
| static agtype_value *extract_entity(CustomScanState *node, |
| TupleTableSlot *scanTupleSlot, |
| int entity_position) |
| { |
| agtype_value *original_entity_value; |
| agtype *original_entity; |
| TupleDesc tupleDescriptor; |
| |
| tupleDescriptor = scanTupleSlot->tts_tupleDescriptor; |
| |
| // type checking, make sure the entity is an agtype vertex or edge |
| if (tupleDescriptor->attrs[entity_position -1].atttypid != AGTYPEOID) |
| ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), |
| errmsg("DELETE clause can only delete agtype"))); |
| |
| original_entity = DATUM_GET_AGTYPE_P(scanTupleSlot->tts_values[entity_position - 1]); |
| original_entity_value = get_ith_agtype_value_from_container(&original_entity->root, 0); |
| |
| 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"))); |
| |
| return original_entity_value; |
| } |
| |
| /* |
| * Try and delete the entity that is describe by the HeapTuple in the table |
| * described by the resultRelInfo. |
| */ |
| static void delete_entity(EState *estate, ResultRelInfo *resultRelInfo, |
| HeapTuple tuple) |
| { |
| ResultRelInfo **saved_resultRels; |
| LockTupleMode lockmode; |
| TM_FailureData hufd; |
| TM_Result lock_result; |
| TM_Result delete_result; |
| Buffer buffer; |
| |
| // Find the physical tuple, this variable is coming from |
| saved_resultRels = estate->es_result_relations; |
| estate->es_result_relations = &resultRelInfo; |
| |
| lockmode = ExecUpdateLockMode(estate, resultRelInfo); |
| |
| lock_result = heap_lock_tuple(resultRelInfo->ri_RelationDesc, tuple, |
| GetCurrentCommandId(false), lockmode, |
| LockWaitBlock, false, &buffer, &hufd); |
| |
| /* |
| * It is possible the entity may have already been deleted. If the tuple |
| * can be deleted, the lock result will be HeapTupleMayBeUpdated. If the |
| * tuple was already deleted by this DELETE clause, the result would be |
| * TM_SelfModified, if the result was deleted by a previous delete |
| * clause, the result will TM_Invisible. Throw an error if any |
| * other result was returned. |
| */ |
| if (lock_result == TM_Ok) |
| { |
| delete_result = heap_delete(resultRelInfo->ri_RelationDesc, |
| &tuple->t_self, GetCurrentCommandId(true), |
| estate->es_crosscheck_snapshot, true, &hufd, |
| false); |
| |
| /* |
| * Unlike locking, the heap_delete either succeeded |
| * HeapTupleMayBeUpdate, or it failed and returned any other result. |
| */ |
| switch (delete_result) |
| { |
| case TM_Ok: |
| break; |
| case TM_SelfModified: |
| ereport( |
| ERROR, |
| (errcode(ERRCODE_INTERNAL_ERROR), |
| errmsg( |
| "deleting the same entity more than once cannot happen"))); |
| /* ereport never gets here */ |
| break; |
| case TM_Updated: |
| ereport( |
| ERROR, |
| (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), |
| errmsg("could not serialize access due to concurrent update"))); |
| /* ereport never gets here */ |
| break; |
| default: |
| elog(ERROR, "Entity failed to be update"); |
| /* elog never gets here */ |
| break; |
| } |
| /* increment the command counter */ |
| CommandCounterIncrement(); |
| } |
| else if (lock_result != TM_Invisible && lock_result != TM_SelfModified) |
| { |
| ereport(ERROR, |
| (errcode(ERRCODE_INTERNAL_ERROR), |
| errmsg("Entity could not be locked for updating"))); |
| |
| } |
| |
| ReleaseBuffer(buffer); |
| |
| estate->es_result_relations = saved_resultRels; |
| } |
| |
| /* |
| * After the delete's subtress has been processed, we then go through the list |
| * of variables to be deleted. |
| */ |
| static void process_delete_list(CustomScanState *node) |
| { |
| cypher_delete_custom_scan_state *css = |
| (cypher_delete_custom_scan_state *)node; |
| ListCell *lc; |
| ExprContext *econtext = css->css.ss.ps.ps_ExprContext; |
| TupleTableSlot *scanTupleSlot = econtext->ecxt_scantuple; |
| EState *estate = node->ss.ps.state; |
| |
| 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; |
| ResultRelInfo *resultRelInfo; |
| HeapTuple heap_tuple; |
| char *label_name; |
| Integer *pos; |
| int entity_position; |
| |
| item = lfirst(lc); |
| |
| pos = item->entity_position; |
| entity_position = pos->ival; |
| |
| /* skip if the entity is null */ |
| if (scanTupleSlot->tts_isnull[entity_position - 1]) |
| continue; |
| |
| original_entity_value = extract_entity(node, scanTupleSlot, |
| entity_position); |
| |
| id = GET_AGTYPE_VALUE_OBJECT_VALUE(original_entity_value, "id"); |
| label = GET_AGTYPE_VALUE_OBJECT_VALUE(original_entity_value, "label"); |
| label_name = pnstrdup(label->val.string.val, label->val.string.len); |
| |
| resultRelInfo = create_entity_result_rel_info(estate, css->delete_data->graph_name, label_name); |
| |
| /* |
| * Setup the scan key to require the id field on-disc to match the |
| * entity's graphid. |
| */ |
| if (original_entity_value->type == AGTV_VERTEX) |
| { |
| 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) |
| { |
| ScanKeyInit(&scan_keys[0], Anum_ag_label_edge_table_id, |
| BTEqualStrategyNumber, F_GRAPHIDEQ, |
| GRAPHID_GET_DATUM(id->val.int_value)); |
| } |
| else |
| { |
| ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), |
| errmsg("DELETE clause can only delete vertices and edges"))); |
| } |
| |
| /* |
| * Setup the scan description, with the correct snapshot and scan keys. |
| */ |
| estate->es_snapshot->curcid = GetCurrentCommandId(false); |
| estate->es_output_cid = GetCurrentCommandId(false); |
| 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 after this variable |
| * was created) we can delete it. Otherwise, its safe to skip this |
| * delete. |
| */ |
| if (!HeapTupleIsValid(heap_tuple)) |
| { |
| table_endscan(scan_desc); |
| destroy_entity_result_rel_info(resultRelInfo); |
| |
| continue; |
| } |
| |
| /* |
| * For vertices, we insert the vertex ID in the hashtable |
| * vertex_id_htab. This hashtable is used later to process |
| * connected edges. |
| */ |
| if (original_entity_value->type == AGTV_VERTEX) |
| { |
| bool found; |
| 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); |
| |
| /* Close the scan and the relation. */ |
| table_endscan(scan_desc); |
| destroy_entity_result_rel_info(resultRelInfo); |
| } |
| } |
| |
| /* |
| * Scans the edge tables and checks if the deleted vertices are connected to |
| * any edge(s). For DETACH DELETE, the connected edges are deleted. Otherwise, |
| * an error is thrown. |
| */ |
| static void check_for_connected_edges(CustomScanState *node) |
| { |
| ListCell *lc; |
| cypher_delete_custom_scan_state *css = |
| (cypher_delete_custom_scan_state *)node; |
| EState *estate = css->css.ss.ps.state; |
| char *graph_name = css->delete_data->graph_name; |
| |
| /* scans each label from css->edge_labels */ |
| foreach (lc, css->edge_labels) |
| { |
| char *label_name = lfirst(lc); |
| ResultRelInfo *resultRelInfo; |
| TableScanDesc scan_desc; |
| HeapTuple tuple; |
| TupleTableSlot *slot; |
| |
| resultRelInfo = create_entity_result_rel_info(estate, graph_name, |
| label_name); |
| estate->es_snapshot->curcid = GetCurrentCommandId(false); |
| estate->es_output_cid = GetCurrentCommandId(false); |
| scan_desc = table_beginscan(resultRelInfo->ri_RelationDesc, |
| estate->es_snapshot, 0, NULL); |
| slot = ExecInitExtraTupleSlot( |
| estate, RelationGetDescr(resultRelInfo->ri_RelationDesc), |
| &TTSOpsHeapTuple); |
| |
| /* for each row */ |
| while (true) |
| { |
| graphid startid; |
| graphid endid; |
| bool isNull; |
| bool found_startid = false; |
| bool found_endid = false; |
| |
| tuple = heap_getnext(scan_desc, ForwardScanDirection); |
| |
| /* no more tuples to process, break and scan the next label. */ |
| if (!HeapTupleIsValid(tuple)) |
| { |
| break; |
| } |
| |
| ExecStoreHeapTuple(tuple, slot, false); |
| |
| startid = GRAPHID_GET_DATUM(slot_getattr( |
| slot, Anum_ag_label_edge_table_start_id, &isNull)); |
| endid = GRAPHID_GET_DATUM( |
| slot_getattr(slot, Anum_ag_label_edge_table_end_id, &isNull)); |
| |
| hash_search(css->vertex_id_htab, (void *)&startid, HASH_FIND, |
| &found_startid); |
| |
| if (!found_startid) |
| { |
| hash_search(css->vertex_id_htab, (void *)&endid, HASH_FIND, |
| &found_endid); |
| } |
| |
| if (found_startid || found_endid) |
| { |
| if (css->delete_data->detach) |
| { |
| delete_entity(estate, resultRelInfo, tuple); |
| } |
| else |
| { |
| ereport( |
| ERROR, |
| (errcode(ERRCODE_INTERNAL_ERROR), |
| errmsg( |
| "Cannot delete a vertex that has edge(s). " |
| "Delete the edge(s) first, or try DETACH DELETE."))); |
| } |
| } |
| } |
| |
| table_endscan(scan_desc); |
| destroy_entity_result_rel_info(resultRelInfo); |
| } |
| } |
