/* * 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/htup_details.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 "utils/rel.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_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. */ ExprContext *econtext = node->ss.ps.ps_ExprContext; SubqueryScanState *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; } } } }