Fix memory leak issues with age_load

src/backend/utils/adt/agtype_util.c

@@ -2304,3 +2304,84 @@ char *agtype_value_type_to_string(enum agtype_value_type type)
 
     return NULL;
 }
+
+/*
+ * Deallocates the passed agtype_value recursively.
+ */
+void pfree_agtype_value(agtype_value* value)
+{
+    pfree_agtype_value_content(value);
+    pfree(value);
+}
+
+/*
+ * Helper function that recursively deallocates the contents 
+ * of the passed agtype_value only. It does not deallocate
+ * `value` itself.
+ */
+void pfree_agtype_value_content(agtype_value* value)
+{
+    int i;
+
+    // guards against stack overflow due to deeply nested agtype_value
+    check_stack_depth();
+
+    switch (value->type)
+    {
+        case AGTV_NUMERIC:
+            pfree(value->val.numeric);
+            break;
+
+        case AGTV_STRING:
+            /*
+             * The char pointer (val.string.val) is not free'd because
+             * it is not allocated by an agtype helper function.
+             */
+            break;
+
+        case AGTV_ARRAY:
+        case AGTV_PATH:
+            for (i = 0; i < value->val.array.num_elems; i++)
+            {
+                pfree_agtype_value_content(&value->val.array.elems[i]);
+            }
+            pfree(value->val.array.elems);
+            break;
+
+        case AGTV_OBJECT:
+        case AGTV_VERTEX:
+        case AGTV_EDGE:
+            for (i = 0; i < value->val.object.num_pairs; i++)
+            {
+                pfree_agtype_value_content(&value->val.object.pairs[i].key);
+                pfree_agtype_value_content(&value->val.object.pairs[i].value);
+            }
+            pfree(value->val.object.pairs);
+            break;
+
+        case AGTV_BINARY:
+            pfree(value->val.binary.data);
+            break;
+
+        case AGTV_NULL:
+        case AGTV_INTEGER:
+        case AGTV_FLOAT:
+        case AGTV_BOOL:
+            /*
+             * These are deallocated by the calling function.
+             */
+            break;
+
+        default:
+            ereport(ERROR,
+                    (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+                     errmsg("unknown agtype")));
+            break;
+    }
+}
+
+void pfree_agtype_in_state(agtype_in_state* value)
+{
+    pfree_agtype_value(value->res);
+    free(value->parse_state);
+}

src/backend/utils/load/ag_load_edges.c

@@ -105,6 +105,8 @@ void edge_row_cb(int delim __attribute__((unused)), void *data)
         insert_edge_simple(cr->graph_oid, cr->object_name,
                            object_graph_id, start_vertex_graph_id,
                            end_vertex_graph_id, props);
+
+        pfree(props);
     }
 
     for (i = 0; i < n_fields; ++i)

src/backend/utils/load/ag_load_labels.c

@@ -137,6 +137,7 @@ void vertex_row_cb(int delim __attribute__((unused)), void *data)
                                         n_fields, label_id_int);
         insert_vertex_simple(cr->graph_oid, cr->object_name,
                              object_graph_id, props);
+        pfree(props);
     }
 
 

src/backend/utils/load/age_load.c

@@ -39,6 +39,7 @@
 
 agtype *create_empty_agtype(void)
 {
+    agtype* out;
     agtype_in_state result;
 
     memset(&result, 0, sizeof(agtype_in_state));
@@ -47,12 +48,18 @@ agtype *create_empty_agtype(void)
                                    NULL);
     result.res = push_agtype_value(&result.parse_state, WAGT_END_OBJECT, NULL);
 
-    return agtype_value_to_agtype(result.res);
+    out = agtype_value_to_agtype(result.res);
+    pfree_agtype_in_state(&result);
+
+    return out;
 }
 
 agtype *create_agtype_from_list(char **header, char **fields, size_t fields_len,
                                 int64 vertex_id)
 {
+    agtype* out;
+    agtype_value* key_agtype;
+    agtype_value* value_agtype;
     agtype_in_state result;
     int i;
 
@@ -61,33 +68,50 @@ agtype *create_agtype_from_list(char **header, char **fields, size_t fields_len,
     result.res = push_agtype_value(&result.parse_state, WAGT_BEGIN_OBJECT,
                                    NULL);
 
+    key_agtype = string_to_agtype_value("__id__");
     result.res = push_agtype_value(&result.parse_state,
                                    WAGT_KEY,
-                                   string_to_agtype_value("__id__"));
+                                   key_agtype);
+
+    value_agtype = integer_to_agtype_value(vertex_id);
     result.res = push_agtype_value(&result.parse_state,
                                    WAGT_VALUE,
-                                   integer_to_agtype_value(vertex_id));
+                                   value_agtype);
+
+    pfree_agtype_value(key_agtype);
+    pfree_agtype_value(value_agtype);
 
     for (i = 0; i<fields_len; i++)
     {
+        key_agtype = string_to_agtype_value(header[i]);
         result.res = push_agtype_value(&result.parse_state,
                                        WAGT_KEY,
-                                       string_to_agtype_value(header[i]));
+                                       key_agtype);
+
+        value_agtype = string_to_agtype_value(fields[i]);
         result.res = push_agtype_value(&result.parse_state,
                                        WAGT_VALUE,
-                                       string_to_agtype_value(fields[i]));
+                                       value_agtype);
+
+        pfree_agtype_value(key_agtype);
+        pfree_agtype_value(value_agtype);
     }
 
     result.res = push_agtype_value(&result.parse_state,
                                    WAGT_END_OBJECT, NULL);
 
-    return agtype_value_to_agtype(result.res);
+    out = agtype_value_to_agtype(result.res);
+    pfree_agtype_in_state(&result);
+
+    return out;
 }
 
 agtype* create_agtype_from_list_i(char **header, char **fields,
                                   size_t fields_len, size_t start_index)
 {
-
+    agtype* out;
+    agtype_value* key_agtype;
+    agtype_value* value_agtype;
     agtype_in_state result;
     size_t i;
 
@@ -103,18 +127,26 @@ agtype* create_agtype_from_list_i(char **header, char **fields,
 
     for (i = start_index; i < fields_len; i++)
     {
+        key_agtype = string_to_agtype_value(header[i]);
         result.res = push_agtype_value(&result.parse_state,
                                        WAGT_KEY,
-                                       string_to_agtype_value(header[i]));
+                                       key_agtype);
+        value_agtype = string_to_agtype_value(fields[i]);
         result.res = push_agtype_value(&result.parse_state,
                                        WAGT_VALUE,
-                                       string_to_agtype_value(fields[i]));
+                                       value_agtype);
+
+        pfree_agtype_value(key_agtype);
+        pfree_agtype_value(value_agtype);
     }
 
     result.res = push_agtype_value(&result.parse_state,
                                    WAGT_END_OBJECT, NULL);
 
-    return agtype_value_to_agtype(result.res);
+    out = agtype_value_to_agtype(result.res);
+    pfree_agtype_in_state(&result);
+
+    return out;
 }
 
 void insert_edge_simple(Oid graph_oid, char *label_name, graphid edge_id,

src/include/utils/agtype.h

@@ -547,6 +547,9 @@ agtype_value *string_to_agtype_value(char *s);
 agtype_value *integer_to_agtype_value(int64 int_value);
 void add_agtype(Datum val, bool is_null, agtype_in_state *result, Oid val_type,
                 bool key_scalar);
+void pfree_agtype_value(agtype_value* value);
+void pfree_agtype_value_content(agtype_value* value);
+void pfree_agtype_in_state(agtype_in_state* value);
 
 /* Oid accessors for AGTYPE */
 Oid get_AGTYPEOID(void);