[PGPRO-3005] Now we search the remote functions in the target list of each node of a query plan.

Some code readability improvements.

Author: danolivo

src/ddl.c

@@ -24,6 +24,7 @@
 #include "tcop/pquery.h"
 #include "utils/snapmgr.h"
 #include "nodes/makefuncs.h"
+#include "nodes/nodeFuncs.h"
 #include "catalog/pg_constraint.h"
 #include "catalog/pg_namespace.h"
 #include "executor/spi.h"
@@ -848,16 +849,14 @@ MtmProcessUtilitySender(PlannedStmt *pstmt, const char *queryString,
 
 		case T_CreateTableSpaceStmt:
 		case T_DropTableSpaceStmt:
-		{
 			skipCommand = true;
 			MtmProcessDDLCommand(stmt_string, false);
 			break;
-		}
 
 		/*
 		 * Explain will not call ProcessUtility for passed CreateTableAsStmt,
-		 * but will run it manually, so we will not catch it in a standart way.
-		 * So catch it in a non-standart way.
+		 * but will run it manually, so we will not catch it in a standard way.
+		 * So catch it in a non-standard way.
 		 */
 		case T_ExplainStmt:
 		{
@@ -948,7 +947,8 @@ MtmProcessUtilitySender(PlannedStmt *pstmt, const char *queryString,
 			DropStmt *stmt = (DropStmt *) parsetree;
 			if (stmt->removeType == OBJECT_INDEX && stmt->concurrent)
 			{
-				if (context == PROCESS_UTILITY_TOPLEVEL) {
+				if (context == PROCESS_UTILITY_TOPLEVEL)
+				{
 					MtmProcessDDLCommand(stmt_string, false);
 					skipCommand = true;
 				}
@@ -971,9 +971,9 @@ MtmProcessUtilitySender(PlannedStmt *pstmt, const char *queryString,
 					if (OidIsValid(relid))
 					{
 						Relation rel = heap_open(relid, ShareLock);
-						if (RelationNeedsWAL(rel)) {
+						if (RelationNeedsWAL(rel))
 							MtmTx.contains_dml = true;
-						}
+
 						heap_close(rel, ShareLock);
 					}
 				}
@@ -1032,37 +1032,70 @@ MtmProcessUtilitySender(PlannedStmt *pstmt, const char *queryString,
 	}
 }
 
+static bool
+targetList_walker(Node *node, void *context)
+{
+	if (node == NULL)
+		return false;
+
+	if (IsA(node, FuncExpr))
+	{
+		Oid func_oid = ((FuncExpr*) node)->funcid;
+
+		if (hash_search(MtmRemoteFunctions, &func_oid, HASH_FIND, NULL))
+			return true;
+	}
+
+	return expression_tree_walker(node, targetList_walker, context);
+}
+
+static bool
+search_for_remote_functions(PlanState *node, void *context)
+{
+	if (node == NULL)
+			return false;
+
+	if (targetList_walker((Node *) node->plan->targetlist, NULL))
+		return true;
+
+	return planstate_tree_walker(node, search_for_remote_functions, NULL);
+}
+
 static void
 MtmExecutorStart(QueryDesc *queryDesc, int eflags)
 {
+	if (PreviousExecutorStartHook != NULL)
+		PreviousExecutorStartHook(queryDesc, eflags);
+	else
+		standard_ExecutorStart(queryDesc, eflags);
+
+	/*
+	 * Explain can contain remote functions. But we don't need to send it to
+	 * another nodes of multimaster.
+	 */
+	if (eflags & EXEC_FLAG_EXPLAIN_ONLY)
+		return;
+
 	if (!MtmIsLogicalReceiver && !MtmDDLStatement && MtmIsEnabled())
 	{
-		ListCell   *tlist;
-
 		if (!MtmRemoteFunctionsValid)
 			MtmInitializeRemoteFunctionsMap();
 
-		foreach(tlist, queryDesc->plannedstmt->planTree->targetlist)
-		{
-			TargetEntry *tle = (TargetEntry *) lfirst(tlist);
-			if (tle->expr && IsA(tle->expr, FuncExpr))
-			{
-				Oid func_oid = ((FuncExpr*) tle->expr)->funcid;
+		Assert(queryDesc->planstate);
 
-				if (hash_search(MtmRemoteFunctions, &func_oid, HASH_FIND, NULL))
-				{
-					MtmProcessDDLCommand(queryDesc->sourceText, true);
-					MtmDDLStatement = queryDesc;
-					MtmTx.contains_ddl = true;
-					break;
-				}
-			}
+		/*
+		 * If any node contains function from the remote functions list in its
+		 * target list than we will send all query as a DDL command.
+		 */
+		if (search_for_remote_functions(queryDesc->planstate, NULL))
+		{
+			MtmProcessDDLCommand(queryDesc->sourceText, true);
+			MtmDDLStatement = queryDesc;
+			MtmTx.contains_ddl = true;
 		}
+		else
+			mtm_log(DDLProcessingTrace, "The query plan don't contain any remote functions");
 	}
-	if (PreviousExecutorStartHook != NULL)
-		PreviousExecutorStartHook(queryDesc, eflags);
-	else
-		standard_ExecutorStart(queryDesc, eflags);
 }
 
 static void
@@ -1082,14 +1115,19 @@ MtmExecutorFinish(QueryDesc *queryDesc)
 			operation == CMD_DELETE || pstmt->hasModifyingCTE)
 		{
 			int i;
-			for (i = 0; i < estate->es_num_result_relations; i++) {
+
+			for (i = 0; i < estate->es_num_result_relations; i++)
+			{
 				Relation rel = estate->es_result_relations[i].ri_RelationDesc;
-				if (RelationNeedsWAL(rel)) {
-					if (MtmIgnoreTablesWithoutPk) {
-						if (!rel->rd_indexvalid) {
+				if (RelationNeedsWAL(rel))
+				{
+					if (MtmIgnoreTablesWithoutPk)
+					{
+						if (!rel->rd_indexvalid)
 							RelationGetIndexList(rel);
-						}
-						if (rel->rd_replidindex == InvalidOid) {
+
+						if (rel->rd_replidindex == InvalidOid)
+						{
 							// XXX
 							MtmMakeRelationLocal(RelationGetRelid(rel), false);
 							continue;
@@ -1417,14 +1455,12 @@ MtmInitializeRemoteFunctionsMap()
 	p = pstrdup(MtmRemoteFunctionsList);
 	do {
 		q = strchr(p, ',');
-		if (q != NULL) {
+		if (q != NULL)
 			*q++ = '\0';
-		}
+
 		clist = FuncnameGetCandidates(stringToQualifiedNameList(p), -1, NIL, false, false, true);
 		if (clist == NULL)
-		{
 			mtm_log(DEBUG1, "Can't resolve function '%s', postponing that", p);
-		}
 		else
 		{
 			while (clist != NULL)
@@ -1438,9 +1474,8 @@ MtmInitializeRemoteFunctionsMap()
 	} while (p != NULL);
 
 	clist = FuncnameGetCandidates(stringToQualifiedNameList("mtm.alter_sequences"), -1, NIL, false, false, true);
-	if (clist != NULL) {
+	if (clist != NULL)
 		hash_search(MtmRemoteFunctions, &clist->oid, HASH_ENTER, NULL);
-	}
 
 	/* restore back current user context */
 	SetUserIdAndSecContext(save_userid, save_sec_context);

src/multimaster.c

@@ -408,7 +408,7 @@ _PG_init(void)
 		"List of function names which should be executed remotely at all multimaster nodes instead of executing them at master and replicating result of their work",
 		NULL,
 		&MtmRemoteFunctionsList,
-		"lo_create,lo_unlink,"
+		"lo_creat,lo_create,lo_unlink,lo_open,loread,lowrite,lo_lseek,lo_tell,lo_close,lo_export"
 		"pathman.create_single_range_partition,pathman.drop_partitions,pathman.create_hash_partitions,pathman.create_range_partitions,pathman.split_range_partition,pathman.drop_range_partition,pathman.add_range_partition,pathman.attach_range_partition,pathman.detach_range_partition,pathman.prepend_range_partition,pathman.append_range_partition,pathman.set_auto,pathman.set_enable_parent,pathman.merge_range_partitions,pathman.add_to_pathman_config,pathman.set_spawn_using_bgw,pathman.set_init_callback,pathman.set_interval,pathman.partition_table_concurrently,"
 		"public.create_single_range_partition,public.drop_partitions,public.create_hash_partitions,public.create_range_partitions,public.split_range_partition,public.drop_range_partition,public.add_range_partition,public.attach_range_partition,public.detach_range_partition,public.prepend_range_partition,public.append_range_partition,public.set_auto,public.set_enable_parent,public.merge_range_partitions,public.add_to_pathman_config,public.set_spawn_using_bgw,public.set_init_callback,public.set_interval,public.partition_table_concurrently"
 		,

src/pglogical_apply.c

@@ -1223,11 +1223,6 @@ process_remote_insert(StringInfo s, Relation rel)
 		ExecStoreTuple(tup, newslot, InvalidBuffer, true);
 		MemoryContextSwitchTo(oldctx);
 
-		/* debug output */
-#ifdef VERBOSE_INSERT
-		log_tuple("INSERT:%s", tupDesc, newslot->tts_tuple);
-#endif
-
 		tmp = estate->es_result_relation_info->ri_TrigDesc;
 		estate->es_result_relation_info->ri_TrigDesc = NULL;
 		ExecSimpleRelationInsert(estate, newslot);

src/pglogical_output.c

@@ -726,6 +726,10 @@ static bool
 pg_filter_decode_txn(LogicalDecodingContext *ctx,
 					   ReorderBufferTXN *txn)
 {
+	/* if txn is NULL, filter it out */
+	if (txn == NULL)
+		return true;
+
 	/* MTM replicates only two-phase transactions */
 	if (txn->gid[0] == '\0')
 		return true;
@@ -734,13 +738,10 @@ pg_filter_decode_txn(LogicalDecodingContext *ctx,
 	 * XXX: that is called per-change and quite expensive for in-progress
 	 * transactions.
 	 */
-	if (txn != NULL &&
-			TransactionIdIsValid(txn->xid) &&
-			TransactionIdDidAbort(txn->xid))
+	if (TransactionIdIsValid(txn->xid) && TransactionIdDidAbort(txn->xid))
 			return true;
 
-	/* if txn is NULL, filter it out */
-	return (txn != NULL)? false:true;
+	return false;
 }
 
 /* ABORT callback */

src/pglogical_proto.c

@@ -152,10 +152,12 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 
 	Assert(hooks_data->is_recovery || txn->origin_id == InvalidRepOriginId);
 
-	if (++MtmSenderTID == InvalidOid) { 
+	if (++MtmSenderTID == InvalidOid)
+	{
 		pglogical_relid_map_reset();
 		MtmSenderTID += 1; /* skip InvalidOid */
 	}
+
 	MtmLastRelId = InvalidOid;
 	MtmCurrentXid = txn->xid;
 	MtmIsFilteredTxn = false;
@@ -597,7 +599,7 @@ pglogical_write_tuple(StringInfo out, PGLogicalOutputData *data,
 		transfer_type = decide_datum_transfer(att, typclass,
 											  data->allow_internal_basetypes,
 											  data->allow_binary_basetypes);
-			
+
         pq_sendbyte(out, transfer_type);
 		switch (transfer_type)
 		{