Fix working with indexes

Author: ildus

jsonbc.c

@@ -276,6 +276,7 @@ jsonbc_communicate(shm_mq_iovec *iov, int iov_len,
 	shm_mq_result		resmq;
 	shm_mq_handle	   *mqh;
 	jsonbc_shm_hdr	   *hdr;
+	jsonbc_shm_worker  *wd;
 
 	char			   *res;
 	Size				reslen;
@@ -289,50 +290,53 @@ jsonbc_communicate(shm_mq_iovec *iov, int iov_len,
 
 	for (i = 0; i < jsonbc_nworkers; i++)
 	{
-		jsonbc_shm_worker *wd = shm_toc_lookup(toc, i + 1, false);
-		if (!pg_atomic_test_set_flag(&wd->busy))
-			continue;
+		wd = shm_toc_lookup(toc, i + 1, false);
+		if (pg_atomic_test_set_flag(&wd->busy))
+			break;
+	}
+
+	if (i == jsonbc_nworkers)
+	{
+		sem_post(&hdr->workers_sem);
+		elog(ERROR, "jsonbc: could not make a connection with workers");
+	}
 
-		/* send data */
-		shm_mq_set_sender(wd->mqin, MyProc);
-		shm_mq_set_receiver(wd->mqout, MyProc);
+	/* send data */
+	shm_mq_set_sender(wd->mqin, MyProc);
+	shm_mq_set_receiver(wd->mqout, MyProc);
 
-		mqh = shm_mq_attach(wd->mqin, NULL, NULL);
-		resmq = shm_mq_sendv(mqh, iov, iov_len, false);
+	mqh = shm_mq_attach(wd->mqin, NULL, NULL);
+	resmq = shm_mq_sendv(mqh, iov, iov_len, false);
+	if (resmq != SHM_MQ_SUCCESS)
+		detached = true;
+	shm_mq_detach(mqh);
+
+	/* get data */
+	if (!detached)
+	{
+		mqh = shm_mq_attach(wd->mqout, NULL, NULL);
+		resmq = shm_mq_receive(mqh, &reslen, (void **) &res, false);
 		if (resmq != SHM_MQ_SUCCESS)
 			detached = true;
-		shm_mq_detach(mqh);
 
-		/* get data */
 		if (!detached)
-		{
-			mqh = shm_mq_attach(wd->mqout, NULL, NULL);
-			resmq = shm_mq_receive(mqh, &reslen, (void **) &res, false);
-			if (resmq != SHM_MQ_SUCCESS)
-				detached = true;
+			callback_succeded = callback(res, reslen, callback_arg);
 
-			if (!detached)
-				callback_succeded = callback(res, reslen, callback_arg);
-
-			shm_mq_detach(mqh);
-		}
-
-		/* clean self as receiver and unlock mq */
-		shm_mq_clean_sender(wd->mqin);
-		shm_mq_clean_receiver(wd->mqout);
-		pg_atomic_clear_flag(&wd->busy);
+		shm_mq_detach(mqh);
+	}
 
-		if (detached)
-			elog(ERROR, "jsonbc: worker has detached");
+	/* clean self as receiver and unlock mq */
+	shm_mq_clean_sender(wd->mqin);
+	shm_mq_clean_receiver(wd->mqout);
+	pg_atomic_clear_flag(&wd->busy);
+	sem_post(&hdr->workers_sem);
 
-		if (!callback_succeded)
-			elog(ERROR, "jsonbc: communication error");
+	if (detached)
+		elog(ERROR, "jsonbc: worker has detached");
 
-		/* we're done here */
-		break;
-	}
+	if (!callback_succeded)
+		elog(ERROR, "jsonbc: communication error");
 
-	sem_post(&hdr->workers_sem);
 }
 
 /* Get key IDs using workers */

jsonbc_worker.c

@@ -225,12 +225,12 @@ jsonbc_get_key(Relation rel, Relation indrel, Oid cmoptoid, uint32 key_id)
 	char		   *res;
 
 	ScanKeyInit(&skey[0],
-				JSONBC_DICTIONARY_REL_ATT_CMOPT,
+				1,
 				BTEqualStrategyNumber,
 				F_OIDEQ,
 				ObjectIdGetDatum(cmoptoid));
 	ScanKeyInit(&skey[1],
-				JSONBC_DICTIONARY_REL_ATT_ID,
+				2,
 				BTEqualStrategyNumber,
 				F_INT4EQ,
 				Int32GetDatum(key_id));
@@ -327,12 +327,12 @@ jsonbc_get_key_id(Relation rel, Relation indrel, Oid cmoptoid, char *key)
 	uint32			result = 0;
 
 	ScanKeyInit(&skey[0],
-				JSONBC_DICTIONARY_REL_ATT_CMOPT,
+				1,
 				BTEqualStrategyNumber,
 				F_OIDEQ,
 				ObjectIdGetDatum(cmoptoid));
 	ScanKeyInit(&skey[1],
-				JSONBC_DICTIONARY_REL_ATT_KEY,
+				2,
 				BTEqualStrategyNumber,
 				F_TEXTEQ,
 				CStringGetTextDatum(key));
@@ -360,6 +360,7 @@ static void
 jsonbc_get_key_ids(Oid cmoptoid, char *buf, uint32 *idsbuf, int nkeys)
 {
 	Relation	rel = NULL;
+	Relation	indrel;
 	int			i;
 	Oid			relid = jsonbc_get_dictionary_relid();
 	bool		spi_on = false;
@@ -371,7 +372,6 @@ jsonbc_get_key_ids(Oid cmoptoid, char *buf, uint32 *idsbuf, int nkeys)
 	{
 		uint32		hkey;
 		bool		found;
-		Relation	indrel;
 		MemoryContext	oldcontext;
 		jsonbc_cached_key		*ckey;
 		jsonbc_pair				*pair;
@@ -409,9 +409,9 @@ jsonbc_get_key_ids(Oid cmoptoid, char *buf, uint32 *idsbuf, int nkeys)
 		{
 			start_xact_command();
 			rel = relation_open(relid, AccessShareLock);
+			indrel = index_open(jsonbc_keys_indoid, AccessShareLock);
 		}
 
-		indrel = index_open(jsonbc_keys_indoid, AccessShareLock);
 		idsbuf[i] = jsonbc_get_key_id(rel, indrel, cmoptoid, buf);
 
 		if (idsbuf[i] == 0)
@@ -452,24 +452,21 @@ jsonbc_get_key_ids(Oid cmoptoid, char *buf, uint32 *idsbuf, int nkeys)
 			}
 			index_close(indrel2, ExclusiveLock);
 		}
-		index_close(indrel, AccessShareLock);
 
 		pair->id = idsbuf[i];
 next:
 		Assert(idsbuf[i] > 0);
 
 		/* move to next key */
-		while (*buf != '\0')
-			buf++;
-
-		buf++;
+		while (*buf++ != '\0');
 	}
 
 	if (spi_on)
 		SPI_finish();
 
 	if (rel)
 	{
+		index_close(indrel, AccessShareLock);
 		relation_close(rel, AccessShareLock);
 		finish_xact_command();
 	}