JSON_TABLE

This feature allows jsonb data to be treated as a table and thus used in
a FROM clause like other tabular data. Data can be selected from the
jsonb using jsonpath expressions, and hoisted out of nested structures
in the jsonb to form multiple rows, more or less like an outer join.

Nikita Glukhov

Reviewers have included (in no particular order) Andres Freund, Alexander
Korotkov, Pavel Stehule, Andrew Alsup, Erik Rijkers, Zhihong Yu (whose
name I previously misspelled), Himanshu Upadhyaya, Daniel Gustafsson,
Justin Pryzby.

Discussion: https://postgr.es/m/7e2cb85d-24cf-4abb-30a5-1a33715959bd@postgrespro.ru

1 files changed, 436 insertions, 0 deletions

diff --git a/src/backend/utils/adt/jsonpath_exec.c b/src/backend/utils/adt/jsonpath_exec.c
index 7811fa31e07..c55b3aae027 100644
--- a/src/backend/utils/adt/jsonpath_exec.c
+++ b/src/backend/utils/adt/jsonpath_exec.c
@@ -61,9 +61,11 @@
 
 #include "catalog/pg_collation.h"
 #include "catalog/pg_type.h"
+#include "executor/execExpr.h"
 #include "funcapi.h"
 #include "lib/stringinfo.h"
 #include "miscadmin.h"
+#include "nodes/nodeFuncs.h"
 #include "regex/regex.h"
 #include "utils/builtins.h"
 #include "utils/date.h"
@@ -74,6 +76,8 @@
 #include "utils/guc.h"
 #include "utils/json.h"
 #include "utils/jsonpath.h"
+#include "utils/lsyscache.h"
+#include "utils/memutils.h"
 #include "utils/timestamp.h"
 #include "utils/varlena.h"
 
@@ -155,6 +159,57 @@ typedef struct JsonValueListIterator
 	ListCell   *next;
 } JsonValueListIterator;
 
+/* Structures for JSON_TABLE execution  */
+typedef struct JsonTableScanState JsonTableScanState;
+typedef struct JsonTableJoinState JsonTableJoinState;
+
+struct JsonTableScanState
+{
+	JsonTableScanState *parent;
+	JsonTableJoinState *nested;
+	MemoryContext mcxt;
+	JsonPath   *path;
+	List	   *args;
+	JsonValueList found;
+	JsonValueListIterator iter;
+	Datum		current;
+	int			ordinal;
+	bool		currentIsNull;
+	bool		errorOnError;
+	bool		advanceNested;
+	bool		reset;
+};
+
+struct JsonTableJoinState
+{
+	union
+	{
+		struct
+		{
+			JsonTableJoinState *left;
+			JsonTableJoinState *right;
+			bool		advanceRight;
+		}			join;
+		JsonTableScanState scan;
+	}			u;
+	bool		is_join;
+};
+
+/* random number to identify JsonTableContext */
+#define JSON_TABLE_CONTEXT_MAGIC	418352867
+
+typedef struct JsonTableContext
+{
+	int			magic;
+	struct
+	{
+		ExprState  *expr;
+		JsonTableScanState *scan;
+	}		   *colexprs;
+	JsonTableScanState root;
+	bool		empty;
+} JsonTableContext;
+
 /* strict/lax flags is decomposed into four [un]wrap/error flags */
 #define jspStrictAbsenseOfErrors(cxt)	(!(cxt)->laxMode)
 #define jspAutoUnwrap(cxt)				((cxt)->laxMode)
@@ -245,6 +300,7 @@ static JsonPathExecResult getArrayIndex(JsonPathExecContext *cxt,
 										JsonPathItem *jsp, JsonbValue *jb, int32 *index);
 static JsonBaseObjectInfo setBaseObject(JsonPathExecContext *cxt,
 										JsonbValue *jbv, int32 id);
+static void JsonValueListClear(JsonValueList *jvl);
 static void JsonValueListAppend(JsonValueList *jvl, JsonbValue *jbv);
 static int	JsonValueListLength(const JsonValueList *jvl);
 static bool JsonValueListIsEmpty(JsonValueList *jvl);
@@ -262,6 +318,12 @@ static JsonbValue *wrapItemsInArray(const JsonValueList *items);
 static int	compareDatetime(Datum val1, Oid typid1, Datum val2, Oid typid2,
 							bool useTz, bool *have_error);
 
+
+static JsonTableJoinState *JsonTableInitPlanState(JsonTableContext *cxt,
+									Node *plan, JsonTableScanState *parent);
+static bool JsonTableNextRow(JsonTableScanState *scan);
+
+
 /****************** User interface to JsonPath executor ********************/
 
 /*
@@ -2459,6 +2521,13 @@ setBaseObject(JsonPathExecContext *cxt, JsonbValue *jbv, int32 id)
 }
 
 static void
+JsonValueListClear(JsonValueList *jvl)
+{
+	jvl->singleton = NULL;
+	jvl->list = NULL;
+}
+
+static void
 JsonValueListAppend(JsonValueList *jvl, JsonbValue *jbv)
 {
 	if (jvl->singleton)
@@ -3067,3 +3136,370 @@ JsonItemFromDatum(Datum val, Oid typid, int32 typmod, JsonbValue *res)
 							"casted to supported jsonpath types.")));
 	}
 }
+
+/************************ JSON_TABLE functions ***************************/
+
+/*
+ * Returns private data from executor state. Ensure validity by check with
+ * MAGIC number.
+ */
+static inline JsonTableContext *
+GetJsonTableContext(TableFuncScanState *state, const char *fname)
+{
+	JsonTableContext *result;
+
+	if (!IsA(state, TableFuncScanState))
+		elog(ERROR, "%s called with invalid TableFuncScanState", fname);
+	result = (JsonTableContext *) state->opaque;
+	if (result->magic != JSON_TABLE_CONTEXT_MAGIC)
+		elog(ERROR, "%s called with invalid TableFuncScanState", fname);
+
+	return result;
+}
+
+/* Recursively initialize JSON_TABLE scan state */
+static void
+JsonTableInitScanState(JsonTableContext *cxt, JsonTableScanState *scan,
+					   JsonTableParent *node, JsonTableScanState *parent,
+					   List *args, MemoryContext mcxt)
+{
+	int			i;
+
+	scan->parent = parent;
+	scan->errorOnError = node->errorOnError;
+	scan->path = DatumGetJsonPathP(node->path->constvalue);
+	scan->args = args;
+	scan->mcxt = AllocSetContextCreate(mcxt, "JsonTableContext",
+									   ALLOCSET_DEFAULT_SIZES);
+	scan->nested = node->child ?
+		JsonTableInitPlanState(cxt, node->child, scan) : NULL;
+	scan->current = PointerGetDatum(NULL);
+	scan->currentIsNull = true;
+
+	for (i = node->colMin; i <= node->colMax; i++)
+		cxt->colexprs[i].scan = scan;
+}
+
+/* Recursively initialize JSON_TABLE scan state */
+static JsonTableJoinState *
+JsonTableInitPlanState(JsonTableContext *cxt, Node *plan,
+					   JsonTableScanState *parent)
+{
+	JsonTableJoinState *state = palloc0(sizeof(*state));
+
+	if (IsA(plan, JsonTableSibling))
+	{
+		JsonTableSibling *join = castNode(JsonTableSibling, plan);
+
+		state->is_join = true;
+		state->u.join.left = JsonTableInitPlanState(cxt, join->larg, parent);
+		state->u.join.right = JsonTableInitPlanState(cxt, join->rarg, parent);
+	}
+	else
+	{
+		JsonTableParent *node = castNode(JsonTableParent, plan);
+
+		state->is_join = false;
+
+		JsonTableInitScanState(cxt, &state->u.scan, node, parent,
+							   parent->args, parent->mcxt);
+	}
+
+	return state;
+}
+
+/*
+ * JsonTableInitOpaque
+ *		Fill in TableFuncScanState->opaque for JsonTable processor
+ */
+static void
+JsonTableInitOpaque(TableFuncScanState *state, int natts)
+{
+	JsonTableContext *cxt;
+	PlanState  *ps = &state->ss.ps;
+	TableFuncScan  *tfs = castNode(TableFuncScan, ps->plan);
+	TableFunc  *tf = tfs->tablefunc;
+	JsonExpr   *ci = castNode(JsonExpr, tf->docexpr);
+	JsonTableParent *root = castNode(JsonTableParent, tf->plan);
+	List	   *args = NIL;
+	ListCell   *lc;
+	int			i;
+
+	cxt = palloc0(sizeof(JsonTableContext));
+	cxt->magic = JSON_TABLE_CONTEXT_MAGIC;
+
+	if (ci->passing_values)
+	{
+		ListCell   *exprlc;
+		ListCell   *namelc;
+
+		forboth(exprlc, ci->passing_values,
+				namelc, ci->passing_names)
+		{
+			Expr	   *expr = (Expr *) lfirst(exprlc);
+			String	   *name = lfirst_node(String, namelc);
+			JsonPathVariableEvalContext *var = palloc(sizeof(*var));
+
+			var->name = pstrdup(name->sval);
+			var->typid = exprType((Node *) expr);
+			var->typmod = exprTypmod((Node *) expr);
+			var->estate = ExecInitExpr(expr, ps);
+			var->econtext = ps->ps_ExprContext;
+			var->mcxt = CurrentMemoryContext;
+			var->evaluated = false;
+			var->value = (Datum) 0;
+			var->isnull = true;
+
+			args = lappend(args, var);
+		}
+	}
+
+	cxt->colexprs = palloc(sizeof(*cxt->colexprs) *
+						   list_length(tf->colvalexprs));
+
+	JsonTableInitScanState(cxt, &cxt->root, root, NULL, args,
+						   CurrentMemoryContext);
+
+	i = 0;
+
+	foreach(lc, tf->colvalexprs)
+	{
+		Expr	   *expr = lfirst(lc);
+
+		cxt->colexprs[i].expr =
+			ExecInitExprWithCaseValue(expr, ps,
+									  &cxt->colexprs[i].scan->current,
+									  &cxt->colexprs[i].scan->currentIsNull);
+
+		i++;
+	}
+
+	state->opaque = cxt;
+}
+
+/* Reset scan iterator to the beginning of the item list */
+static void
+JsonTableRescan(JsonTableScanState *scan)
+{
+	JsonValueListInitIterator(&scan->found, &scan->iter);
+	scan->current = PointerGetDatum(NULL);
+	scan->currentIsNull = true;
+	scan->advanceNested = false;
+	scan->ordinal = 0;
+}
+
+/* Reset context item of a scan, execute JSON path and reset a scan */
+static void
+JsonTableResetContextItem(JsonTableScanState *scan, Datum item)
+{
+	MemoryContext oldcxt;
+	JsonPathExecResult res;
+	Jsonb		*js = (Jsonb *) DatumGetJsonbP(item);
+
+	JsonValueListClear(&scan->found);
+
+	MemoryContextResetOnly(scan->mcxt);
+
+	oldcxt = MemoryContextSwitchTo(scan->mcxt);
+
+	res = executeJsonPath(scan->path, scan->args, EvalJsonPathVar, js,
+						  scan->errorOnError, &scan->found, false /* FIXME */);
+
+	MemoryContextSwitchTo(oldcxt);
+
+	if (jperIsError(res))
+	{
+		Assert(!scan->errorOnError);
+		JsonValueListClear(&scan->found);	/* EMPTY ON ERROR case */
+	}
+
+	JsonTableRescan(scan);
+}
+
+/*
+ * JsonTableSetDocument
+ *		Install the input document
+ */
+static void
+JsonTableSetDocument(TableFuncScanState *state, Datum value)
+{
+	JsonTableContext *cxt = GetJsonTableContext(state, "JsonTableSetDocument");
+
+	JsonTableResetContextItem(&cxt->root, value);
+}
+
+/*
+ * Fetch next row from a union joined scan.
+ *
+ * Returns false at the end of a scan, true otherwise.
+ */
+static bool
+JsonTableNextJoinRow(JsonTableJoinState *state)
+{
+	if (!state->is_join)
+		return JsonTableNextRow(&state->u.scan);
+
+	if (!state->u.join.advanceRight)
+	{
+		/* fetch next outer row */
+		if (JsonTableNextJoinRow(state->u.join.left))
+			return true;
+
+		state->u.join.advanceRight = true;	/* next inner row */
+	}
+
+	/* fetch next inner row */
+	return JsonTableNextJoinRow(state->u.join.right);
+}
+
+/* Recursively set 'reset' flag of scan and its child nodes */
+static void
+JsonTableJoinReset(JsonTableJoinState *state)
+{
+	if (state->is_join)
+	{
+		JsonTableJoinReset(state->u.join.left);
+		JsonTableJoinReset(state->u.join.right);
+		state->u.join.advanceRight = false;
+	}
+	else
+	{
+		state->u.scan.reset = true;
+		state->u.scan.advanceNested = false;
+
+		if (state->u.scan.nested)
+			JsonTableJoinReset(state->u.scan.nested);
+	}
+}
+
+/*
+ * Fetch next row from a simple scan with outer joined nested subscans.
+ *
+ * Returns false at the end of a scan, true otherwise.
+ */
+static bool
+JsonTableNextRow(JsonTableScanState *scan)
+{
+	JsonbValue *jbv;
+	MemoryContext oldcxt;
+
+	/* reset context item if requested */
+	if (scan->reset)
+	{
+		Assert(!scan->parent->currentIsNull);
+		JsonTableResetContextItem(scan, scan->parent->current);
+		scan->reset = false;
+	}
+
+	if (scan->advanceNested)
+	{
+		/* fetch next nested row */
+		if (JsonTableNextJoinRow(scan->nested))
+			return true;
+
+		scan->advanceNested = false;
+	}
+
+	/* fetch next row */
+	jbv = JsonValueListNext(&scan->found, &scan->iter);
+
+	if (!jbv)
+	{
+		scan->current = PointerGetDatum(NULL);
+		scan->currentIsNull = true;
+		return false;	/* end of scan */
+	}
+
+	/* set current row item */
+	oldcxt = MemoryContextSwitchTo(scan->mcxt);
+	scan->current = JsonbPGetDatum(JsonbValueToJsonb(jbv));
+	scan->currentIsNull = false;
+	MemoryContextSwitchTo(oldcxt);
+
+	scan->ordinal++;
+
+	if (scan->nested)
+	{
+		JsonTableJoinReset(scan->nested);
+		scan->advanceNested = JsonTableNextJoinRow(scan->nested);
+	}
+
+	return true;
+}
+
+/*
+ * JsonTableFetchRow
+ *		Prepare the next "current" tuple for upcoming GetValue calls.
+ *		Returns FALSE if the row-filter expression returned no more rows.
+ */
+static bool
+JsonTableFetchRow(TableFuncScanState *state)
+{
+	JsonTableContext *cxt = GetJsonTableContext(state, "JsonTableFetchRow");
+
+	if (cxt->empty)
+		return false;
+
+	return JsonTableNextRow(&cxt->root);
+}
+
+/*
+ * JsonTableGetValue
+ *		Return the value for column number 'colnum' for the current row.
+ *
+ * This leaks memory, so be sure to reset often the context in which it's
+ * called.
+ */
+static Datum
+JsonTableGetValue(TableFuncScanState *state, int colnum,
+				  Oid typid, int32 typmod, bool *isnull)
+{
+	JsonTableContext *cxt = GetJsonTableContext(state, "JsonTableGetValue");
+	ExprContext *econtext = state->ss.ps.ps_ExprContext;
+	ExprState  *estate = cxt->colexprs[colnum].expr;
+	JsonTableScanState *scan = cxt->colexprs[colnum].scan;
+	Datum		result;
+
+	if (scan->currentIsNull) /* NULL from outer/union join */
+	{
+		result = (Datum) 0;
+		*isnull = true;
+	}
+	else if (estate)	/* regular column */
+	{
+		result = ExecEvalExpr(estate, econtext, isnull);
+	}
+	else
+	{
+		result = Int32GetDatum(scan->ordinal);	/* ordinality column */
+		*isnull = false;
+	}
+
+	return result;
+}
+
+/*
+ * JsonTableDestroyOpaque
+ */
+static void
+JsonTableDestroyOpaque(TableFuncScanState *state)
+{
+	JsonTableContext *cxt = GetJsonTableContext(state, "JsonTableDestroyOpaque");
+
+	/* not valid anymore */
+	cxt->magic = 0;
+
+	state->opaque = NULL;
+}
+
+const TableFuncRoutine JsonbTableRoutine =
+{
+	JsonTableInitOpaque,
+	JsonTableSetDocument,
+	NULL,
+	NULL,
+	NULL,
+	JsonTableFetchRow,
+	JsonTableGetValue,
+	JsonTableDestroyOpaque
+};