Fix database does not exists error when is_member(), schema_id() func…

…tion gets called in parallel query mode. (#289)

This commit solve the issue of parallel worker not being able to obtain correct logical database name. Whenever get_cur_db_name() is called, it returns static char current_db_name[] which is initialized as empty. Now, parallel workers are not aware of the database name. For them, it is empty whenever get_cur_db_name() gets called. Hence, we were getting the following error:

database "" does not exist
So, to communicate the logical database name and more data related to babelfish we have introduced a struct BabelfishFixedParallelState. Currently BabelfishFixedParallelState has only 1 field, i.e., logical_db_name, we can add more fields to it whenever necessary. Then we will write the logical_db_name in DSM during initialization (in InitializeParallelDSM) using a hook. Another hook is used while trying to fetch the logical database name in ParallelWorkerMain().

In this way we are introduction a mechanism through which we can communicate any babelfish related information to parallel workers.

Issues Resolved: 
BABEL-4538, BABEL-4481, BABEL-4421

Signed-off-by: Shameem Ahmed <[email protected]>

src/backend/access/transam/parallel.c

@@ -77,6 +77,10 @@
 #define PARALLEL_KEY_RELMAPPER_STATE		UINT64CONST(0xFFFFFFFFFFFF000D)
 #define PARALLEL_KEY_UNCOMMITTEDENUMS		UINT64CONST(0xFFFFFFFFFFFF000E)
 
+/* Hooks for communicating babelfish related information to parallel worker */
+bbf_InitializeParallelDSM_hook_type bbf_InitializeParallelDSM_hook = NULL;
+bbf_ParallelWorkerMain_hook_type bbf_ParallelWorkerMain_hook = NULL;
+
 /* Fixed-size parallel state. */
 typedef struct FixedParallelState
 {
@@ -291,6 +295,10 @@ InitializeParallelDSM(ParallelContext *pcxt)
 		shm_toc_estimate_chunk(&pcxt->estimator, strlen(pcxt->library_name) +
 							   strlen(pcxt->function_name) + 2);
 		shm_toc_estimate_keys(&pcxt->estimator, 1);
+
+		/* Estimate how much we'll need for the babelfish fixed parallel state */
+		if (MyProcPort->is_tds_conn && bbf_InitializeParallelDSM_hook)
+			(*bbf_InitializeParallelDSM_hook) (pcxt, true);
 	}
 
 	/*
@@ -465,6 +473,10 @@ InitializeParallelDSM(ParallelContext *pcxt)
 		strcpy(entrypointstate, pcxt->library_name);
 		strcpy(entrypointstate + lnamelen + 1, pcxt->function_name);
 		shm_toc_insert(pcxt->toc, PARALLEL_KEY_ENTRYPOINT, entrypointstate);
+
+		/* Initialize babelfish fixed-size state in shared memory. */
+		if (MyProcPort->is_tds_conn && bbf_InitializeParallelDSM_hook)
+			(*bbf_InitializeParallelDSM_hook) (pcxt, false);
 	}
 
 	/* Restore previous memory context. */
@@ -1483,6 +1495,10 @@ ParallelWorkerMain(Datum main_arg)
 										   false);
 	RestoreUncommittedEnums(uncommittedenumsspace);
 
+	/* Hook for babelfish to restore babelfish fixed parallel state */
+	if (MyFixedParallelState->babelfish_context && bbf_ParallelWorkerMain_hook)
+		(*bbf_ParallelWorkerMain_hook) (toc);
+
 	/* Attach to the leader's serializable transaction, if SERIALIZABLE. */
 	AttachSerializableXact(fps->serializable_xact_handle);
 

src/include/access/parallel.h

@@ -82,4 +82,15 @@ extern void ParallelWorkerMain(Datum main_arg);
 /* Below helpers are added to support parallel workers in Babelfish context */
 extern bool IsBabelfishParallelWorker(void);
 
+/* Key for BabelfishFixedParallelState */
+#define BABELFISH_PARALLEL_KEY_FIXED        UINT64CONST(0xBBF0000000000001)
+
+/* Hooks for communicating babelfish related information to parallel worker */
+typedef void (*bbf_InitializeParallelDSM_hook_type)(ParallelContext *pcxt, bool estimate);
+extern PGDLLIMPORT bbf_InitializeParallelDSM_hook_type bbf_InitializeParallelDSM_hook;
+
+typedef void (*bbf_ParallelWorkerMain_hook_type)(shm_toc *toc);
+extern PGDLLIMPORT bbf_ParallelWorkerMain_hook_type bbf_ParallelWorkerMain_hook;
+
+
 #endif							/* PARALLEL_H */