| /* |
| * 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 "catalog/ag_label.h" |
| #include "executor/cypher_executor.h" |
| #include "executor/cypher_utils.h" |
| |
| static void begin_cypher_merge(CustomScanState *node, EState *estate, |
| int eflags); |
| static TupleTableSlot *exec_cypher_merge(CustomScanState *node); |
| static void end_cypher_merge(CustomScanState *node); |
| static void rescan_cypher_merge(CustomScanState *node); |
| static Datum merge_vertex(cypher_merge_custom_scan_state *css, |
| cypher_target_node *node, ListCell *next, List* list); |
| static void merge_edge(cypher_merge_custom_scan_state *css, |
| cypher_target_node *node, Datum prev_vertex_id, |
| ListCell *next, List *list); |
| static void process_simple_merge(CustomScanState *node); |
| static bool check_path(cypher_merge_custom_scan_state *css, |
| TupleTableSlot *slot); |
| static void process_path(cypher_merge_custom_scan_state *css); |
| static void mark_tts_isnull(TupleTableSlot *slot); |
| |
| const CustomExecMethods cypher_merge_exec_methods = {MERGE_SCAN_STATE_NAME, |
| begin_cypher_merge, |
| exec_cypher_merge, |
| end_cypher_merge, |
| rescan_cypher_merge, |
| NULL, NULL, NULL, NULL, |
| NULL, NULL, NULL, NULL}; |
| |
| /* |
| * Initializes the MERGE Execution Node at the beginning of the execution |
| * phase. |
| */ |
| static void begin_cypher_merge(CustomScanState *node, EState *estate, |
| int eflags) |
| { |
| cypher_merge_custom_scan_state *css = |
| (cypher_merge_custom_scan_state *)node; |
| ListCell *lc = NULL; |
| Plan *subplan = NULL; |
| |
| Assert(list_length(css->cs->custom_plans) == 1); |
| |
| // initialize the subplan |
| subplan = linitial(css->cs->custom_plans); |
| node->ss.ps.lefttree = ExecInitNode(subplan, estate, eflags); |
| |
| /* TODO is this necessary? Removing it seems to not have an impact */ |
| ExecAssignExprContext(estate, &node->ss.ps); |
| |
| ExecInitScanTupleSlot(estate, &node->ss, |
| ExecGetResultType(node->ss.ps.lefttree), |
| &TTSOpsVirtual); |
| |
| /* |
| * When MERGE is not the last clause in a cypher query. Setup projection |
| * information to pass to the parent execution node. |
| */ |
| if (!CYPHER_CLAUSE_IS_TERMINAL(css->flags)) |
| { |
| TupleDesc tupdesc = node->ss.ss_ScanTupleSlot->tts_tupleDescriptor; |
| |
| ExecAssignProjectionInfo(&node->ss.ps, tupdesc); |
| } |
| |
| /* |
| * For each vertex and edge in the path, setup the information |
| * needed if we need to create them. |
| */ |
| foreach(lc, css->path->target_nodes) |
| { |
| cypher_target_node *cypher_node = |
| (cypher_target_node *)lfirst(lc); |
| Relation rel = NULL; |
| |
| /* |
| * This entity is references an entity that is already declared. Either |
| * by a previous clause or an entity earlier in the MERGE path. In both |
| * cases, this target_entry will not create data, only reference data |
| * that already exists. |
| */ |
| if (!CYPHER_TARGET_NODE_INSERT_ENTITY(cypher_node->flags)) |
| { |
| continue; |
| } |
| |
| // Open relation and acquire a row exclusive lock. |
| rel = table_open(cypher_node->relid, RowExclusiveLock); |
| |
| // Initialize resultRelInfo for the vertex |
| cypher_node->resultRelInfo = makeNode(ResultRelInfo); |
| InitResultRelInfo(cypher_node->resultRelInfo, rel, |
| list_length(estate->es_range_table), NULL, |
| estate->es_instrument); |
| |
| // Open all indexes for the relation |
| ExecOpenIndices(cypher_node->resultRelInfo, false); |
| |
| // Setup the relation's tuple slot |
| cypher_node->elemTupleSlot = ExecInitExtraTupleSlot( |
| estate, |
| RelationGetDescr(cypher_node->resultRelInfo->ri_RelationDesc), |
| &TTSOpsHeapTuple); |
| |
| if (cypher_node->id_expr != NULL) |
| { |
| cypher_node->id_expr_state = |
| ExecInitExpr(cypher_node->id_expr, (PlanState *)node); |
| } |
| |
| if (cypher_node->prop_expr != NULL) |
| { |
| cypher_node->prop_expr_state = ExecInitExpr(cypher_node->prop_expr, |
| (PlanState *)node); |
| } |
| } |
| |
| /* |
| * 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; |
| } |
| |
| /* store the currentCommandId for this instance */ |
| css->base_currentCommandId = GetCurrentCommandId(false); |
| |
| Increment_Estate_CommandId(estate); |
| } |
| |
| /* |
| * Checks the subtree to see if the lateral join representing the MERGE path |
| * found results. Returns true if the path does not exist and must be created, |
| * false otherwise. |
| */ |
| static bool check_path(cypher_merge_custom_scan_state *css, |
| TupleTableSlot *slot) |
| { |
| cypher_create_path *path = css->path; |
| ListCell *lc = NULL; |
| |
| foreach(lc, path->target_nodes) |
| { |
| cypher_target_node *node = lfirst(lc); |
| |
| /* |
| * If target_node as a valid attribute number and is a node not |
| * declared in a previous clause, check the tuple position in the |
| * slot. If the slot is null, the path was not found. The rules |
| * state that if one part of the path does not exists, the whole |
| * path must be created. |
| */ |
| if (node->tuple_position != InvalidAttrNumber || |
| ((node->flags & CYPHER_TARGET_NODE_MERGE_EXISTS) == 0)) |
| { |
| /* |
| * Attribute number is 1 indexed and tts_values is 0 indexed, |
| * offset by 1. |
| */ |
| if (slot->tts_isnull[node->tuple_position - 1]) |
| { |
| |
| return true; |
| } |
| } |
| } |
| return false; |
| } |
| |
| static void process_path(cypher_merge_custom_scan_state *css) |
| { |
| cypher_create_path *path = css->path; |
| List *list = path->target_nodes; |
| ListCell *lc = list_head(list); |
| |
| /* |
| * Create the first vertex. The create_vertex function will |
| * create the rest of the path, if necessary. |
| */ |
| merge_vertex(css, lfirst(lc), lnext(list, lc), list); |
| |
| /* |
| * If this path is a variable, take the list that was accumulated |
| * in the vertex/edge creation, create a path datum, and add to the |
| * scantuple slot. |
| */ |
| if (path->path_attr_num != InvalidAttrNumber) |
| { |
| ExprContext *econtext = css->css.ss.ps.ps_ExprContext; |
| TupleTableSlot *scantuple = econtext->ecxt_scantuple; |
| Datum result; |
| int tuple_position = path->path_attr_num - 1; |
| bool debug_flag = false; |
| |
| /* |
| * We need to make sure that the tuple_position is within the |
| * boundaries of the tuple's number of attributes. Otherwise, it |
| * will corrupt memory. The cases where it doesn't fit within are |
| * usually due to a variable that is specified but there isn't a RETURN |
| * clause. In these cases we just don't bother to store the |
| * value. |
| */ |
| if (!debug_flag && |
| (tuple_position < scantuple->tts_tupleDescriptor->natts || |
| scantuple->tts_tupleDescriptor->natts != 1)) |
| { |
| result = make_path(css->path_values); |
| |
| /* store the result */ |
| scantuple->tts_values[tuple_position] = result; |
| scantuple->tts_isnull[tuple_position] = false; |
| } |
| } |
| } |
| |
| /* |
| * Function that handles the case where MERGE is the only clause in the query. |
| */ |
| static void process_simple_merge(CustomScanState *node) |
| { |
| cypher_merge_custom_scan_state *css = |
| (cypher_merge_custom_scan_state *)node; |
| EState *estate = css->css.ss.ps.state; |
| TupleTableSlot *slot = NULL; |
| |
| /*Process the subtree first */ |
| Decrement_Estate_CommandId(estate) |
| slot = ExecProcNode(node->ss.ps.lefttree); |
| Increment_Estate_CommandId(estate) |
| |
| if (TupIsNull(slot)) |
| { |
| ExprContext *econtext = node->ss.ps.ps_ExprContext; |
| SubqueryScanState *sss = (SubqueryScanState *)node->ss.ps.lefttree; |
| |
| /* our child execution node should be a subquery */ |
| Assert(IsA(sss, SubqueryScanState)); |
| |
| /* setup the scantuple that the process_path needs */ |
| econtext->ecxt_scantuple = sss->ss.ss_ScanTupleSlot; |
| |
| process_path(css); |
| } |
| } |
| |
| /* |
| * Iterate through the TupleTableSlot's tts_values and marks the isnull field |
| * with true. |
| */ |
| static void mark_tts_isnull(TupleTableSlot *slot) |
| { |
| int numberOfAttributes = slot->tts_tupleDescriptor->natts; |
| int i; |
| |
| for (i = 0; i < numberOfAttributes; i++) |
| { |
| Datum val; |
| |
| val = slot->tts_values[i]; |
| |
| if (val == (Datum)NULL) |
| { |
| slot->tts_isnull[i] = true; |
| } |
| } |
| } |
| |
| /* |
| * Function that is called mid-execution. This function will call |
| * its subtree in the execution tree, and depending on the results |
| * create the new path, and depending on the context of the MERGE |
| * within the query pass data to the parent execution node. |
| * |
| * Returns a TupleTableSlot with the next tuple to it parent or |
| * Returns NULL when MERGE has no more tuples to emit. |
| */ |
| static TupleTableSlot *exec_cypher_merge(CustomScanState *node) |
| { |
| cypher_merge_custom_scan_state *css = |
| (cypher_merge_custom_scan_state *)node; |
| EState *estate = css->css.ss.ps.state; |
| ExprContext *econtext = css->css.ss.ps.ps_ExprContext; |
| TupleTableSlot *slot = NULL; |
| bool terminal = CYPHER_CLAUSE_IS_TERMINAL(css->flags); |
| |
| /* |
| * There are three cases that dictate the flow of the execution logic. |
| * |
| * 1. MERGE is not the first clause in the cypher query. |
| * 2. MERGE is the first clause and there are no following clauses. |
| * 3. MERGE is the first clause and there are following clauses. |
| * CYPHER_CLAUSE_FLAG_PREVIOUS_CLAUSE |
| */ |
| if (CYPHER_CLAUSE_HAS_PREVIOUS_CLAUSE(css->flags)) |
| { |
| /* |
| * Case 1: MERGE is not the first clause in the cypher query. |
| * |
| * For this case, we need to process all tuples given to us by the |
| * previous clause. When we receive a tuple from the previous clause: |
| * check to see if the left lateral join found the pattern already. If |
| * it did, we don't need to create the pattern. If the lateral join did |
| * not find the whole path, create the whole path. |
| * |
| * If this is a terminal clause, process all tuples. If not, pass the |
| * tuple to up the execution tree. |
| */ |
| do |
| { |
| /*Process the subtree first */ |
| Decrement_Estate_CommandId(estate) |
| slot = ExecProcNode(node->ss.ps.lefttree); |
| Increment_Estate_CommandId(estate) |
| |
| /* |
| * We are done processing the subtree, mark as terminal |
| * so the function returns NULL. |
| */ |
| if (TupIsNull(slot)) |
| { |
| terminal = true; |
| break; |
| } |
| |
| /* setup the scantuple that the process_path needs */ |
| econtext->ecxt_scantuple = |
| node->ss.ps.lefttree->ps_ProjInfo->pi_exprContext->ecxt_scantuple; |
| |
| if (check_path(css, econtext->ecxt_scantuple)) |
| { |
| process_path(css); |
| } |
| |
| } while (terminal); |
| |
| /* if this was a terminal MERGE just return NULL */ |
| if (terminal) |
| { |
| return NULL; |
| } |
| |
| econtext->ecxt_scantuple = ExecProject(node->ss.ps.lefttree->ps_ProjInfo); |
| return ExecProject(node->ss.ps.ps_ProjInfo); |
| |
| } |
| else if (terminal) |
| { |
| /* |
| * Case 2: MERGE is the first clause and there are no following clauses |
| * |
| * For case 2, check to see if we found the pattern, if not create it. |
| * Return NULL in either cases, because no rows are expected. |
| */ |
| process_simple_merge(node); |
| |
| /* |
| * Case 2 always returns NULL the first time exec_cypher_merge is |
| * called. |
| */ |
| return NULL; |
| } |
| else |
| { |
| /* |
| * Case 3: MERGE is the first clause and there are following clauses. |
| * |
| * Case three has two subcases: |
| * |
| * 1. The already path exists. |
| * 2. The path does not exist. |
| */ |
| |
| /* |
| * Part of Case 2. |
| * |
| * If created_new_path is marked as true. The path did not exist and |
| * MERGE created it. We have already passed that information up the |
| * execution tree, and now we tell MERGE's parents in the execution |
| * tree there is no more tuples to pass. |
| */ |
| if (css->created_new_path) |
| { |
| /* |
| * If the created_new_path is set to true. Then MERGE should not |
| * have found a path, because this should only be set to true if |
| * merge found sub-case 1 |
| */ |
| Assert(css->found_a_path == false); |
| |
| return NULL; |
| } |
| |
| /* |
| * Process the subtree. The subtree will only consist of the MERGE |
| * path. |
| */ |
| Decrement_Estate_CommandId(estate) |
| slot = ExecProcNode(node->ss.ps.lefttree); |
| Increment_Estate_CommandId(estate) |
| |
| if (!TupIsNull(slot)) |
| { |
| /* |
| * Part of Sub-Case 1. |
| * |
| * If we found a path, mark the found_a_path flag to true and |
| * pass the tuple to the next execution tree. When the path |
| * exists, we don't need to create/modify anything. |
| */ |
| css->found_a_path = true; |
| |
| econtext->ecxt_scantuple = ExecProject(node->ss.ps.lefttree->ps_ProjInfo); |
| return ExecProject(node->ss.ps.ps_ProjInfo); |
| } |
| else if (TupIsNull(slot) && css->found_a_path) |
| { |
| /* |
| * Part of Sub-Case 2. |
| * |
| * MERGE found the path(s) that already exists and we are done passing |
| * all the found path(s) up the execution tree. |
| */ |
| return NULL; |
| } |
| else |
| { |
| /* |
| * Part of Sub-Case 1. |
| * |
| * MERGE could not find the path in memory and the sub-node in the |
| * execution tree returned NULL. We need to create the path and |
| * pass the tuple to the next execution node. The subtrees will |
| * begin its cleanup process when there are no more tuples found. |
| * So we will need to create a TupleTableSlot and populate with the |
| * information from the newly created path that the query needs. |
| */ |
| SubqueryScanState *sss = NULL; |
| econtext = node->ss.ps.ps_ExprContext; |
| sss = (SubqueryScanState *)node->ss.ps.lefttree; |
| |
| /* |
| * Our child execution node is always a subquery. If not there |
| * is an issue. |
| */ |
| Assert(IsA(sss, SubqueryScanState)); |
| |
| /* |
| * found_a_path should only be set to true if MERGE is following |
| * sub-case 2. |
| */ |
| Assert(css->found_a_path == false); |
| |
| /* |
| * This block of sub-case 1 should only be executed once. To |
| * create the single path if the path does not exist. If we find |
| * ourselves here again, the internal state of the MERGE execution |
| * node was incorrectly altered. |
| */ |
| Assert(css->created_new_path == false); |
| |
| /* |
| * Postgres cleared the child tuple table slot, we need to remake |
| * it. |
| */ |
| ExecInitScanTupleSlot(estate, &sss->ss, |
| ExecGetResultType(sss->subplan), |
| &TTSOpsVirtual); |
| |
| /* setup the scantuple that the process_path needs */ |
| econtext->ecxt_scantuple = sss->ss.ss_ScanTupleSlot; |
| |
| // create the path |
| process_path(css); |
| |
| // mark the create_new_path flag to true. |
| css->created_new_path = true; |
| |
| /* |
| * find the tts_values that process_path did not populate and |
| * mark as null. |
| */ |
| mark_tts_isnull(econtext->ecxt_scantuple); |
| |
| // store the heap tuble |
| ExecStoreVirtualTuple(econtext->ecxt_scantuple); |
| |
| /* |
| * make the subquery's projection scan slot be the tuple table we |
| * created and run the projection logic. |
| */ |
| sss->ss.ps.ps_ProjInfo->pi_exprContext->ecxt_scantuple = |
| econtext->ecxt_scantuple; |
| |
| // assign this to be our scantuple |
| econtext->ecxt_scantuple = ExecProject(node->ss.ps.lefttree->ps_ProjInfo); |
| |
| /* |
| * run the merge's projection logic and pass to its parent |
| * execution node |
| */ |
| return ExecProject(node->ss.ps.ps_ProjInfo); |
| } |
| } |
| } |
| |
| |
| /* |
| * Function called at the end of the execution phase to cleanup |
| * MERGE. |
| */ |
| static void end_cypher_merge(CustomScanState *node) |
| { |
| cypher_merge_custom_scan_state *css = |
| (cypher_merge_custom_scan_state *)node; |
| cypher_create_path *path = css->path; |
| ListCell *lc = NULL; |
| |
| // increment the command counter |
| CommandCounterIncrement(); |
| |
| ExecEndNode(node->ss.ps.lefttree); |
| |
| foreach (lc, path->target_nodes) |
| { |
| cypher_target_node *cypher_node = (cypher_target_node *)lfirst(lc); |
| |
| if (!CYPHER_TARGET_NODE_INSERT_ENTITY(cypher_node->flags)) |
| { |
| continue; |
| } |
| |
| // close all indices for the node |
| ExecCloseIndices(cypher_node->resultRelInfo); |
| |
| // close the relation itself |
| table_close(cypher_node->resultRelInfo->ri_RelationDesc, |
| RowExclusiveLock); |
| } |
| } |
| |
| /* |
| * Rescan is mostly used by join execution nodes, and several others. |
| * Since we are creating data here its not safe to rescan the node. Throw |
| * an error and try to help the uer understand what went wrong. |
| */ |
| static void rescan_cypher_merge(CustomScanState *node) |
| { |
| ereport(ERROR, |
| (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), |
| errmsg("cypher merge clause cannot be rescanned"), |
| errhint("its unsafe to use joins in a query with a Cypher MERGE clause"))); |
| } |
| |
| /* |
| * Extracts the metadata information that MERGE needs from the |
| * merge_custom_scan node and creates the cypher_merge_custom_scan_state |
| * for the execution phase. |
| */ |
| Node *create_cypher_merge_plan_state(CustomScan *cscan) |
| { |
| cypher_merge_custom_scan_state *cypher_css = |
| palloc0(sizeof(cypher_merge_custom_scan_state)); |
| cypher_merge_information *merge_information; |
| char *serialized_data = NULL; |
| Const *c = NULL; |
| |
| 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; |
| merge_information = stringToNode(serialized_data); |
| |
| Assert(is_ag_node(merge_information, cypher_merge_information)); |
| |
| cypher_css->merge_information = merge_information; |
| cypher_css->flags = merge_information->flags; |
| cypher_css->merge_function_attr = merge_information->merge_function_attr; |
| cypher_css->path = merge_information->path; |
| cypher_css->created_new_path = false; |
| cypher_css->found_a_path = false; |
| cypher_css->graph_oid = merge_information->graph_oid; |
| |
| cypher_css->css.ss.ps.type = T_CustomScanState; |
| cypher_css->css.methods = &cypher_merge_exec_methods; |
| |
| return (Node *)cypher_css; |
| } |
| |
| /* |
| * Creates the vertex entity, returns the vertex's id in case the caller is |
| * the create_edge function. |
| */ |
| static Datum merge_vertex(cypher_merge_custom_scan_state *css, |
| cypher_target_node *node, ListCell *next, List *list) |
| { |
| bool isNull; |
| Datum id; |
| EState *estate = css->css.ss.ps.state; |
| ExprContext *econtext = css->css.ss.ps.ps_ExprContext; |
| ResultRelInfo *resultRelInfo = node->resultRelInfo; |
| TupleTableSlot *elemTupleSlot = node->elemTupleSlot; |
| TupleTableSlot *scanTupleSlot = econtext->ecxt_scantuple; |
| |
| Assert(node->type == LABEL_KIND_VERTEX); |
| |
| /* |
| * Vertices in a path might already exists. If they do get the id |
| * to pass to the edges before and after it. Otherwise, insert the |
| * new vertex into it's table and then pass the id along. |
| */ |
| if (CYPHER_TARGET_NODE_INSERT_ENTITY(node->flags)) |
| { |
| ResultRelInfo **old_estate_es_result_relations = NULL; |
| Datum prop; |
| |
| /* |
| * Set estate's result relation to the vertex's result |
| * relation. |
| * |
| * Note: This obliterates what was their previously |
| */ |
| |
| /* save the old result relation info */ |
| old_estate_es_result_relations = estate->es_result_relations; |
| |
| estate->es_result_relations = &resultRelInfo; |
| |
| ExecClearTuple(elemTupleSlot); |
| |
| /* get the next graphid for this vertex */ |
| id = ExecEvalExpr(node->id_expr_state, econtext, &isNull); |
| elemTupleSlot->tts_values[vertex_tuple_id] = id; |
| elemTupleSlot->tts_isnull[vertex_tuple_id] = isNull; |
| |
| /* get the properties for this vertex */ |
| prop = ExecEvalExpr(node->prop_expr_state, econtext, &isNull); |
| elemTupleSlot->tts_values[vertex_tuple_properties] = prop; |
| elemTupleSlot->tts_isnull[vertex_tuple_properties] = isNull; |
| |
| /* |
| * Insert the new vertex. |
| * |
| * Depending on the currentCommandId, we need to do this one of two |
| * different ways - |
| * |
| * 1) If they are equal, the currentCommandId hasn't been used for an |
| * update, or it hasn't been incremented after being used. In either |
| * case, we need to use the current one and then increment it so that |
| * the following commands will have visibility of this update. Note, |
| * it isn't our job to update the currentCommandId first and then do |
| * this check. |
| * |
| * 2) If they are not equal, the currentCommandId has been used and/or |
| * updated. In this case, we can't use it. Otherwise our update won't |
| * be visible to anything that follows, until the currentCommandId is |
| * updated again. Remember, visibility is, greater than but not equal |
| * to, the currentCommandID used for the update. So, in this case we |
| * need to use the original currentCommandId when begin_cypher_merge |
| * was initiated as everything under this instance of merge needs to |
| * be based off of that initial currentCommandId. This allows the |
| * following command to see the updates generated by this instance of |
| * merge. |
| */ |
| if (css->base_currentCommandId == GetCurrentCommandId(false)) |
| { |
| insert_entity_tuple(resultRelInfo, elemTupleSlot, estate); |
| |
| /* |
| * Increment the currentCommandId since we processed an update. We |
| * don't want to do this outside of this block because we don't want |
| * to inadvertently or unnecessarily update the commandCounterId of |
| * another command. |
| */ |
| CommandCounterIncrement(); |
| } |
| else |
| { |
| insert_entity_tuple_cid(resultRelInfo, elemTupleSlot, estate, |
| css->base_currentCommandId); |
| } |
| |
| /* restore the old result relation info */ |
| estate->es_result_relations = old_estate_es_result_relations; |
| |
| /* |
| * When the vertex is used by clauses higher in the execution tree |
| * we need to create a vertex datum. When the vertex is a variable, |
| * add to the scantuple slot. When the vertex is part of a path |
| * variable, add to the list. |
| */ |
| if (CYPHER_TARGET_NODE_OUTPUT(node->flags)) |
| { |
| Datum result; |
| |
| /* make the vertex agtype */ |
| result = make_vertex(id, CStringGetDatum(node->label_name), prop); |
| |
| /* append to the path list */ |
| if (CYPHER_TARGET_NODE_IN_PATH(node->flags)) |
| { |
| css->path_values = lappend(css->path_values, |
| DatumGetPointer(result)); |
| } |
| |
| /* |
| * Put the vertex in the correct spot in the scantuple, so parent |
| * execution nodes can reference the newly created variable. |
| */ |
| if (CYPHER_TARGET_NODE_IS_VARIABLE(node->flags)) |
| { |
| bool debug_flag = false; |
| int tuple_position = node->tuple_position - 1; |
| |
| /* |
| * We need to make sure that the tuple_position is within the |
| * boundaries of the tuple's number of attributes. Otherwise, it |
| * will corrupt memory. The cases where it doesn't fall within |
| * are usually due to a variable that is specified but there |
| * isn't a RETURN clause. In these cases we just don't bother to |
| * store the value. |
| */ |
| if (!debug_flag && |
| (tuple_position < scanTupleSlot->tts_tupleDescriptor->natts || |
| scanTupleSlot->tts_tupleDescriptor->natts != 1)) |
| { |
| /* store the result */ |
| scanTupleSlot->tts_values[tuple_position] = result; |
| scanTupleSlot->tts_isnull[tuple_position] = false; |
| } |
| } |
| } |
| } |
| else |
| { |
| agtype *a = NULL; |
| Datum d; |
| agtype_value *v = NULL; |
| agtype_value *id_value = NULL; |
| |
| /* check that the variable isn't NULL */ |
| if (scanTupleSlot->tts_isnull[node->tuple_position - 1]) |
| { |
| ereport(ERROR, |
| (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), |
| errmsg("Existing variable %s cannot be NULL in MERGE clause", |
| node->variable_name))); |
| } |
| |
| /* get the vertex agtype in the scanTupleSlot */ |
| d = scanTupleSlot->tts_values[node->tuple_position - 1]; |
| a = DATUM_GET_AGTYPE_P(d); |
| |
| /* Convert to an agtype value */ |
| v = get_ith_agtype_value_from_container(&a->root, 0); |
| |
| if (v->type != AGTV_VERTEX) |
| { |
| ereport(ERROR, |
| (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), |
| errmsg("agtype must resolve to a vertex"))); |
| } |
| |
| /* extract the id agtype field */ |
| id_value = GET_AGTYPE_VALUE_OBJECT_VALUE(v, "id"); |
| |
| /* extract the graphid and cast to a Datum */ |
| id = GRAPHID_GET_DATUM(id_value->val.int_value); |
| |
| /* |
| * Its possible the variable has already been deleted. There are two |
| * ways this can happen. One is the query explicitly deleted the |
| * variable, the is_deleted flag will catch that. However, it is |
| * possible the user deleted the vertex using another variable name. We |
| * need to scan the table to find the vertex's current status relative |
| * to this CREATE clause. If the variable was initially created in this |
| * clause, we can skip this check, because the transaction system |
| * guarantees that nothing can happen to that tuple, as far as we are |
| * concerned with at this time. |
| */ |
| if (!SAFE_TO_SKIP_EXISTENCE_CHECK(node->flags)) |
| { |
| if (!entity_exists(estate, css->graph_oid, DATUM_GET_GRAPHID(id))) |
| { |
| ereport(ERROR, |
| (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), |
| errmsg("vertex assigned to variable %s was deleted", |
| node->variable_name))); |
| } |
| } |
| |
| /* |
| * Add the Datum to the list of entities for creating the path variable |
| */ |
| if (CYPHER_TARGET_NODE_IN_PATH(node->flags)) |
| { |
| Datum vertex = scanTupleSlot->tts_values[node->tuple_position - 1]; |
| css->path_values = lappend(css->path_values, |
| DatumGetPointer(vertex)); |
| } |
| } |
| |
| /* If the path continues, create the next edge, passing the vertex's id. */ |
| if (next != NULL) |
| { |
| merge_edge(css, lfirst(next), id, lnext(list, next), list); |
| } |
| |
| return id; |
| } |
| |
| /* |
| * Create the edge entity. |
| */ |
| static void merge_edge(cypher_merge_custom_scan_state *css, |
| cypher_target_node *node, Datum prev_vertex_id, |
| ListCell *next, List *list) |
| { |
| bool isNull; |
| EState *estate = css->css.ss.ps.state; |
| ExprContext *econtext = css->css.ss.ps.ps_ExprContext; |
| ResultRelInfo *resultRelInfo = node->resultRelInfo; |
| ResultRelInfo **old_estate_es_result_relations = NULL; |
| TupleTableSlot *elemTupleSlot = node->elemTupleSlot; |
| Datum id; |
| Datum start_id, end_id, next_vertex_id; |
| List *prev_path = css->path_values; |
| Datum prop; |
| |
| Assert(node->type == LABEL_KIND_EDGE); |
| Assert(lfirst(next) != NULL); |
| |
| /* |
| * Create the next vertex before creating the edge. We need the |
| * next vertex's id. |
| */ |
| css->path_values = NIL; |
| next_vertex_id = merge_vertex(css, lfirst(next), lnext(list, next), list); |
| |
| /* |
| * Set the start and end vertex ids |
| */ |
| if (node->dir == CYPHER_REL_DIR_RIGHT || node->dir == CYPHER_REL_DIR_NONE) |
| { |
| // create pattern (prev_vertex)-[edge]->(next_vertex) |
| start_id = prev_vertex_id; |
| end_id = next_vertex_id; |
| } |
| else if (node->dir == CYPHER_REL_DIR_LEFT) |
| { |
| // create pattern (prev_vertex)<-[edge]-(next_vertex) |
| start_id = next_vertex_id; |
| end_id = prev_vertex_id; |
| } |
| else |
| { |
| ereport(ERROR, |
| (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), |
| errmsg("edge direction must be specified in a MERGE clause"))); |
| } |
| |
| /* |
| * Set estate's result relation to the vertex's result |
| * relation. |
| * |
| * Note: This obliterates what was their previously |
| */ |
| |
| /* save the old result relation info */ |
| old_estate_es_result_relations = estate->es_result_relations; |
| |
| estate->es_result_relations = &resultRelInfo; |
| |
| ExecClearTuple(elemTupleSlot); |
| |
| // Graph Id for the edge |
| id = ExecEvalExpr(node->id_expr_state, econtext, &isNull); |
| elemTupleSlot->tts_values[edge_tuple_id] = id; |
| elemTupleSlot->tts_isnull[edge_tuple_id] = isNull; |
| |
| // Graph id for the starting vertex |
| elemTupleSlot->tts_values[edge_tuple_start_id] = start_id; |
| elemTupleSlot->tts_isnull[edge_tuple_start_id] = false; |
| |
| // Graph id for the ending vertex |
| elemTupleSlot->tts_values[edge_tuple_end_id] = end_id; |
| elemTupleSlot->tts_isnull[edge_tuple_end_id] = false; |
| |
| // Edge's properties map |
| prop = ExecEvalExpr(node->prop_expr_state, econtext, &isNull); |
| elemTupleSlot->tts_values[edge_tuple_properties] = prop; |
| elemTupleSlot->tts_isnull[edge_tuple_properties] = isNull; |
| |
| // Insert the new edge |
| insert_entity_tuple(resultRelInfo, elemTupleSlot, estate); |
| |
| /* restore the old result relation info */ |
| estate->es_result_relations = old_estate_es_result_relations; |
| |
| /* |
| * When the edge is used by clauses higher in the execution tree |
| * we need to create an edge datum. When the edge is a variable, |
| * add to the scantuple slot. When the edge is part of a path |
| * variable, add to the list. |
| */ |
| if (CYPHER_TARGET_NODE_OUTPUT(node->flags)) |
| { |
| Datum result; |
| |
| result = make_edge(id, start_id, end_id, |
| CStringGetDatum(node->label_name), prop); |
| |
| // add the Datum to the list of entities for creating the path variable |
| if (CYPHER_TARGET_NODE_IN_PATH(node->flags)) |
| { |
| prev_path = lappend(prev_path, DatumGetPointer(result)); |
| css->path_values = list_concat(prev_path, css->path_values); |
| } |
| |
| // Add the entity to the TupleTableSlot if necessary |
| if (CYPHER_TARGET_NODE_IS_VARIABLE(node->flags)) |
| { |
| TupleTableSlot *scantuple = econtext->ecxt_scantuple; |
| bool debug_flag = false; |
| int tuple_position = node->tuple_position - 1; |
| |
| /* |
| * We need to make sure that the tuple_position is within the |
| * boundaries of the tuple's number of attributes. Otherwise, it |
| * will corrupt memory. The cases where it doesn't fall within are |
| * usually due to a variable that is specified but there isn't a |
| * RETURN clause. In these cases we just don't bother to store the |
| * value. |
| */ |
| if (!debug_flag && |
| (tuple_position < scantuple->tts_tupleDescriptor->natts || |
| scantuple->tts_tupleDescriptor->natts != 1)) |
| { |
| /* store the result */ |
| scantuple->tts_values[tuple_position] = result; |
| scantuple->tts_isnull[tuple_position] = false; |
| } |
| } |
| } |
| } |