Add framework for LOAD CSV

Author: JoshInnis

Makefile

@@ -108,6 +108,7 @@ REGRESS = new_cypher \
           cypher_merge \
           cypher_unwind \
           cypher_vle \
+          load_csv \
           order_by \
           cypher_setop \
           aggregation \

regress/expected/load_csv.out

@@ -0,0 +1,28 @@
+
+/*
+ * Copyright (C) 2023-2025 PostGraphDB
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */ 
+
+LOAD 'postgraph';
+
+CREATE GRAPH load_csv;
+USE GRAPH load_csv;
+
+
+CYPHER LOAD CSV '~/postgraph/regress/test.csv'
+RETURN *;
+
+DROP GRAPH load_csv;

regress/sql/load_csv.sql

@@ -0,0 +1,5 @@
+booker12,9012,Rachel,Booker
+grey07,2070,Laura,Grey
+johnson81,4081,Craig,Johnson
+jenkins46,9346,Mary,Jenkins
+smith79,5079,Jamie,Smith

regress/test.csv

@@ -515,6 +515,13 @@ PARALLEL SAFE
 AS 'MODULE_PATHNAME', 'edge_unnest';
 
 
+CREATE FUNCTION cypher_load_csv (gtype)
+RETURNS SETOF gtype
+LANGUAGE C 
+STABLE 
+RETURNS NULL ON NULL INPUT 
+PARALLEL UNSAFE 
+AS 'MODULE_PATHNAME', 'cypher_load_csv';
 
 CREATE FUNCTION vle (IN gtype, IN vertex, IN vertex, IN gtype, IN gtype,
                      IN gtype, IN gtype, IN gtype, OUT edges variable_edge)

sql/postgraph.sql.in

@@ -43,6 +43,7 @@ const char *node_names[] = {
     "cypher_delete",
     "cypher_unwind",
     "cypher_merge",
+    "cypher_load_csv",
     "cypher_path",
     "cypher_node",
     "cypher_relationship",
@@ -111,6 +112,7 @@ const ExtensibleNodeMethods node_methods[] = {
     DEFINE_NODE_METHODS(cypher_delete),
     DEFINE_NODE_METHODS(cypher_unwind),
     DEFINE_NODE_METHODS(cypher_merge),
+    DEFINE_NODE_METHODS(cypher_load_csv),
     DEFINE_NODE_METHODS(cypher_path),
     DEFINE_NODE_METHODS(cypher_node),
     DEFINE_NODE_METHODS(cypher_relationship),

src/backend/nodes/ag_nodes.c

@@ -212,6 +212,14 @@ void out_cypher_merge(StringInfo str, const ExtensibleNode *node)
     WRITE_NODE_FIELD(path);
 }
 
+// serialization function for the cypher_delete ExtensibleNode.
+void out_cypher_load_csv(StringInfo str, const ExtensibleNode *node)
+{
+    DEFINE_AG_NODE(cypher_load_csv);
+
+    WRITE_STRING_FIELD(file);
+}
+
 // serialization function for the cypher_path ExtensibleNode.
 void out_cypher_path(StringInfo str, const ExtensibleNode *node)
 {

src/backend/nodes/cypher_outfuncs.c

@@ -153,6 +153,8 @@ static TargetEntry *placeholder_edge(cypher_parsestate *cpstate, char *name);
 static TargetEntry *placeholder_traversal(cypher_parsestate *cpstate, char *name);
 // call
 static Query *transform_cypher_call(cypher_parsestate *cpstate, cypher_clause *clause);
+// load csv
+static Query *transform_cypher_load_csv(cypher_parsestate *cpstate, cypher_clause *clause);
 // transform
 #define PREV_CYPHER_CLAUSE_ALIAS    "_"
 #define CYPHER_OPT_RIGHT_ALIAS      "_R"
@@ -221,6 +223,8 @@ Query *transform_cypher_clause(cypher_parsestate *cpstate, cypher_clause *clause
         result = transform_cypher_sub_pattern(cpstate, clause);
     } else if (is_ag_node(self, cypher_unwind)) {
         result = transform_cypher_unwind(cpstate, clause);
+    } else if (is_ag_node(self, cypher_load_csv)) {
+        result = transform_cypher_load_csv(cpstate, clause);
     } else {
         ereport(ERROR, (errmsg_internal("unexpected Node for cypher_clause")));
     }
@@ -328,6 +332,61 @@ static Node *transform_srf_function(cypher_parsestate *cpstate, Node *n, RangeTb
     return NULL;
 }
 
+static Query *transform_cypher_load_csv(cypher_parsestate *cpstate, cypher_clause *clause) {
+    cypher_parsestate *child_cpstate = make_cypher_parsestate(cpstate);
+    ParseState *pstate = (ParseState *) child_cpstate;
+    cypher_load_csv *load = clause->self;
+    TargetEntry *te;
+    ParseNamespaceItem *pnsi;
+    FuncCall *fc = makeFuncCall(list_make2(makeString("postgraph"), makeString("cypher_load_csv")),
+                                list_make1(makeString(load->file)),
+                                COERCE_SQL_SYNTAX,
+                                -1);
+
+    Query *query = makeNode(Query);
+    query->commandType = CMD_SELECT;
+
+    /*if (clause->prev) {
+        int rtindex;
+
+        pnsi = transform_prev_cypher_clause(child_cpstate, clause->prev, true);
+        rtindex = list_length(pstate->p_rtable);
+        Assert(rtindex == 1); // rte is the first RangeTblEntry in pstate
+        query->targetList = expandNSItemAttrs(pstate, pnsi, 0, -1);
+    }*/
+
+    Expr *expr = NULL;
+    //if (!call->where)
+     {
+        expr = transform_cypher_expr(child_cpstate, fc, EXPR_KIND_SELECT_TARGET);
+        //te = makeTargetEntry((Expr *) expr, (AttrNumber) pstate->p_next_resno++, call->alias, false);
+        te = makeTargetEntry((Expr *) expr, (AttrNumber) pstate->p_next_resno++, "test", false);
+
+        expr = NULL;
+        query->targetList = lappend(query->targetList, te);
+    } /*else {
+       pnsi = add_srf_to_query(child_cpstate, call->func, call->alias);
+       Node *var = scanNSItemForColumn(pstate, pnsi, 0, call->alias, -1);
+       te = makeTargetEntry((Expr *) var, (AttrNumber) pstate->p_next_resno++, call->alias, false);
+       query->targetList = lappend(query->targetList, te);
+
+       transform_entity *entity = make_transform_entity(child_cpstate, ENT_FUNC_CALL, NULL, (Node *)var, call->alias);
+       child_cpstate->entities = lappend(child_cpstate->entities, entity);
+
+
+       expr = transform_cypher_expr(child_cpstate, call->where, EXPR_KIND_WHERE);
+    }*/
+
+    query->rtable = pstate->p_rtable;
+    query->jointree = makeFromExpr(pstate->p_joinlist, expr);
+    query->hasTargetSRFs = pstate->p_hasTargetSRFs;
+
+    assign_query_collations(pstate, query);
+
+    free_cypher_parsestate(child_cpstate);
+    return query;
+}
+
 static Query *transform_cypher_call(cypher_parsestate *cpstate, cypher_clause *clause) {
     cypher_parsestate *child_cpstate = make_cypher_parsestate(cpstate);
     ParseState *pstate = (ParseState *) child_cpstate;

src/backend/parser/cypher_clause.c

@@ -581,6 +581,8 @@ makeSimpleCypherA_Expr(A_Expr_Kind kind, char *name,
 %type <node> call_stmt yield_item
 %type <list> yield_item_list
 
+%type <node> load_stmt
+
 /* common */
 %type <node> cypher_where_opt
 
@@ -11803,7 +11805,8 @@ cypher_query_start:
     | CYPHER with { $$ = $2; }
     | merge
     | CYPHER call_stmt { $$ = $2; }
-    | return
+    | CYPHER load_stmt { $$ = $2; }
+	| return
     | unwind
 ;
 
@@ -11827,11 +11830,21 @@ clause:
     | delete
     | merge
     | call_stmt
+	| load_stmt
     | return
     | unwind
     ;
 
 
+load_stmt:
+	LOAD CSV Sconst
+	{
+		cypher_load_csv *n = make_ag_node(cypher_load_csv);
+        n->file = $3;
+
+        $$ = n;
+	};
+
 UseGraphStmt:
     USE GRAPH BareColLabel
     {

src/backend/parser/cypher_gram.y

@@ -3084,3 +3084,58 @@ generate_series_step_gtype(PG_FUNCTION_ARGS)
 		/* do when there is no more left */
 		SRF_RETURN_DONE(funcctx);
 }
+
+typedef struct
+{
+
+} cypher_load_csv_ctx;
+
+PG_FUNCTION_INFO_V1(cypher_load_csv);
+
+Datum
+cypher_load_csv(PG_FUNCTION_ARGS)
+{
+	FuncCallContext *funcctx;
+	cypher_load_csv_ctx *fctx;
+	int64		result;
+	MemoryContext oldcontext;
+
+	/* stuff done only on the first call of the function */
+	if (SRF_IS_FIRSTCALL())
+	{
+		/* create a function context for cross-call persistence */
+		funcctx = SRF_FIRSTCALL_INIT();
+
+		/* allocate memory for user context */
+		fctx = (generate_series_fctx *) palloc(sizeof(cypher_load_csv_ctx));
+
+
+		funcctx->user_fctx = fctx;
+		/*
+		 * switch to memory context appropriate for multiple function calls
+		 */
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+
+		MemoryContextSwitchTo(oldcontext);
+	}
+
+	/* stuff done on every call of the function */
+	funcctx = SRF_PERCALL_SETUP();
+
+	/*
+	 * get the saved state and use current as the result for this iteration
+	 */
+	fctx = funcctx->user_fctx;
+
+	/*if ((fctx->step > 0 && fctx->current <= fctx->finish) ||
+		(fctx->step < 0 && fctx->current >= fctx->finish))
+	{
+            //gtype_value gtv = { .type = AGTV_INTEGER, .val.int_value = result };
+
+
+		SRF_RETURN_NEXT(funcctx, GTYPE_P_GET_DATUM(gtype_value_to_gtype(&gtv)));
+	}*/
+	//else
+		SRF_RETURN_DONE(funcctx);
+}

src/backend/utils/adt/gtype.c

@@ -45,6 +45,7 @@ typedef enum ag_node_tag
     cypher_delete_t,
     cypher_unwind_t,
     cypher_merge_t,
+    cypher_load_csv_t,
     // pattern
     cypher_path_t,
     cypher_node_t,

src/include/nodes/ag_nodes.h

@@ -165,6 +165,13 @@ typedef struct cypher_call
     Node *query_tree;
 } cypher_call;
 
+typedef struct cypher_load_csv
+{
+    ExtensibleNode extensible;
+    char *file;
+} cypher_load_csv;
+
+
 /*
  * pattern
  */

src/include/nodes/cypher_nodes.h

@@ -38,6 +38,7 @@ void out_cypher_set_item(StringInfo str, const ExtensibleNode *node);
 void out_cypher_delete(StringInfo str, const ExtensibleNode *node);
 void out_cypher_unwind(StringInfo str, const ExtensibleNode *node);
 void out_cypher_merge(StringInfo str, const ExtensibleNode *node);
+void out_cypher_load_csv(StringInfo str, const ExtensibleNode *node);
 
 // pattern
 void out_cypher_path(StringInfo str, const ExtensibleNode *node);