MERGE support on hypertables

.unreleased/PR_5150

@@ -0,0 +1 @@
+Implements: #5150 MERGE support on hypertables

CHANGELOG.md

@@ -32,6 +32,7 @@ This release includes these noteworthy features:
 * #5584 Reduce decompression during constraint checking
 * #5530 Optimize compressed chunk resorting
 * #5639 Support sending telemetry event reports
+* #5150 MERGE support on hypertables
 
 **Bugfixes**
 * #5396 Fix SEGMENTBY columns predicates to be pushed down

src/import/ht_hypertable_modify.c

@@ -766,9 +766,31 @@ ht_ExecMergeNotMatched(ModifyTableContext * context, ResultRelInfo * resultRelIn
 			 */
 			newslot = ExecProject(action->mas_proj);
 			context->relaction = action;
-
-			(void)
-				ExecInsert(context, mtstate->rootResultRelInfo, newslot, canSetTag);
+			if (cds->is_dropped_attr_exists)
+			{
+				AttrMap *map;
+				TupleDesc parenttupdesc, chunktupdesc;
+				TupleTableSlot *chunk_slot = NULL;
+
+				parenttupdesc = RelationGetDescr(resultRelInfo->ri_RelationDesc);
+				chunktupdesc = RelationGetDescr(cds->rri->ri_RelationDesc);
+				/* map from parent to chunk */
+				map = build_attrmap_by_name_if_req(parenttupdesc, chunktupdesc);
+				if (map != NULL)
+					chunk_slot =
+						execute_attr_map_slot(map,
+												newslot,
+												MakeSingleTupleTableSlot(chunktupdesc,
+																		&TTSOpsVirtual));
+				(void) ExecInsert(context,
+									cds->rri,
+									(chunk_slot ? chunk_slot : newslot),
+									canSetTag);
+				if (chunk_slot)
+					ExecDropSingleTupleTableSlot(chunk_slot);
+			}
+			else
+				(void) ExecInsert(context, cds->rri, newslot, canSetTag);
 			mtstate->mt_merge_inserted = 1;
 			break;
 		case CMD_NOTHING:

src/nodes/chunk_dispatch/chunk_dispatch.c

@@ -11,6 +11,7 @@
 #include <nodes/nodeFuncs.h>
 #include <parser/parsetree.h>
 #include <utils/rel.h>
+#include <utils/syscache.h>
 #include <catalog/pg_type.h>
 
 #include "compat/compat.h"
@@ -20,6 +21,7 @@
 #include "subspace_store.h"
 #include "dimension.h"
 #include "guc.h"
+#include "nodes/hypertable_modify.h"
 #include "ts_catalog/chunk_data_node.h"
 
 static Node *chunk_dispatch_state_create(CustomScan *cscan);
@@ -244,6 +246,15 @@
 	cscan->custom_scan_tlist = tlist;
 	cscan->scan.plan.targetlist = tlist;
 
+#if PG15_GE
+	if (root->parse->mergeUseOuterJoin)
+	{
+		/* replace expressions of ROWID_VAR */
+		tlist = ts_replace_rowid_vars(root, tlist, relopt->relid);
+		cscan->scan.plan.targetlist = tlist;
+		cscan->custom_scan_tlist = tlist;
+	}
+#endif
 	return &cscan->scan.plan;
 }
 
@@ -339,8 +350,61 @@
 	/* Switch to the executor's per-tuple memory context */
 	old = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
 
+#if PG15_GE
+	TupleTableSlot *newslot = NULL;
+	if (dispatch->dispatch_state->mtstate->operation == CMD_MERGE)
+	{
+		HeapTuple tp;
+		AttrNumber natts;
+		AttrNumber attno;
+
+		tp = SearchSysCache1(RELOID, ObjectIdGetDatum(ht->main_table_relid));
+		if (!HeapTupleIsValid(tp))
+			elog(ERROR, "cache lookup failed for relation %u", ht->main_table_relid);
+		natts = ((Form_pg_class) GETSTRUCT(tp))->relnatts;
+		ReleaseSysCache(tp);
+		for (attno = 1; attno <= natts; attno++)
+		{
+			tp = SearchSysCache2(ATTNUM,
+								 ObjectIdGetDatum(ht->main_table_relid),
+								 Int16GetDatum(attno));
+			if (!HeapTupleIsValid(tp))
+				continue;
+			Form_pg_attribute att_tup = (Form_pg_attribute) GETSTRUCT(tp);
+			ReleaseSysCache(tp);
+			if (att_tup->attisdropped || att_tup->atthasmissing)
+			{
+				state->is_dropped_attr_exists = true;
+				continue;
+			}
+		}
+		for (int i = 0; i < ht->space->num_dimensions; i++)
+		{
+			List *actionStates =
+				dispatch->dispatch_state->mtstate->resultRelInfo->ri_notMatchedMergeAction;
+			ListCell *l;
+			foreach (l, actionStates)
+			{
+				MergeActionState *action = (MergeActionState *) lfirst(l);
+				CmdType commandType = action->mas_action->commandType;
+				if (commandType == CMD_INSERT)
+				{
+					/* fetch full projection list */
+					action->mas_proj->pi_exprContext->ecxt_innertuple = slot;
+					newslot = ExecProject(action->mas_proj);
+					break;
+				}
+			}
+			if (newslot)
+				break;
+		}
+	}
 	/* Calculate the tuple's point in the N-dimensional hyperspace */
+	point = ts_hyperspace_calculate_point(ht->space, (newslot ? newslot : slot));
+
+#else
 	point = ts_hyperspace_calculate_point(ht->space, slot);
+#endif
 
 	/* Save the main table's (hypertable's) ResultRelInfo */
 	if (!dispatch->hypertable_result_rel_info)
@@ -375,7 +439,7 @@
 	MemoryContextSwitchTo(old);
 
 	/* Convert the tuple to the chunk's rowtype, if necessary */
-	if (cis->hyper_to_chunk_map != NULL)
+	if (cis->hyper_to_chunk_map != NULL && state->is_dropped_attr_exists == false)
 		slot = execute_attr_map_slot(cis->hyper_to_chunk_map->attrMap, slot, cis->slot);
 
 	return slot;

src/nodes/chunk_dispatch/chunk_dispatch.h

@@ -72,6 +72,8 @@ typedef struct ChunkDispatchState
 	 */
 	ChunkDispatch *dispatch;
 	ResultRelInfo *rri;
+	/* flag to represent dropped attributes */
+	bool is_dropped_attr_exists;
 } ChunkDispatchState;
 
 extern TSDLLEXPORT bool ts_is_chunk_dispatch_state(PlanState *state);