Continue work on bulk inserting

Author: ildus

Makefile

@@ -1,7 +1,7 @@
 # contrib/jsonbc/Makefile
 
 MODULE_big = jsonbc
-OBJS= jsonbc.o jsonbc_worker.o $(WIN32RES)
+OBJS= jsonbc.o jsonbc_worker.o jsonbc_utils.o $(WIN32RES)
 
 EXTENSION = jsonbc
 DATA = jsonbc--0.1.sql

jsonbc.c

@@ -36,23 +36,6 @@ typedef struct
 	MemoryContext	item_mcxt;
 } CompressionThroughBuffers;
 
-#if PG_VERSION_NUM == 110000
-struct shm_mq_alt
-{
-	slock_t		mq_mutex;
-	PGPROC	   *mq_receiver;	/* this one */
-	PGPROC	   *mq_sender;		/* this one */
-	uint64		mq_bytes_read;
-	uint64		mq_bytes_written;
-	Size		mq_ring_size;
-	bool		mq_detached;	/* and this one */
-
-	/* in postgres version there are more attributes, but we don't need them */
-};
-#else
-#error "shm_mq struct in jsonbc is copied from PostgreSQL 11, please correct it according to your version"
-#endif
-
 /* local */
 static MemoryContext compression_mcxt = NULL;
 static CompressionThroughBuffers *compression_buffers = NULL;
@@ -68,44 +51,18 @@ int jsonbc_queue_size = 0;
 static void init_memory_context(bool);
 static void memory_reset_callback(void *arg);
 static void encode_varbyte(uint32 val, unsigned char *ptr, int *len);
-static uint32 decode_varbyte(unsigned char *ptr);
 static char *packJsonbValue(JsonbValue *val, int header_size, int *len);
 static void setup_guc_variables(void);
 static char *jsonbc_get_keys(Oid cmoptoid, uint32 *ids, int nkeys, size_t *buflen);
 static void jsonbc_get_key_ids(Oid cmoptoid, char *buf, int buflen, uint32 *idsbuf, int nkeys);
+static uint32 decode_varbyte(unsigned char *ptr);
 
 static size_t
 jsonbc_get_queue_size(void)
 {
 	return (Size) (jsonbc_queue_size * 1024);
 }
 
-void
-shm_mq_clean_sender(shm_mq *mq)
-{
-	struct shm_mq_alt	*amq = (struct shm_mq_alt *) mq;
-
-	/* check that attributes are same and our struct still compatible with global shm_mq */
-	Assert(shm_mq_get_sender(mq) == amq->mq_sender);
-	Assert(shm_mq_get_receiver(mq) == amq->mq_receiver);
-
-	amq->mq_sender = NULL;
-	amq->mq_detached = false;
-}
-
-void
-shm_mq_clean_receiver(shm_mq *mq)
-{
-	struct shm_mq_alt	*amq = (struct shm_mq_alt *) mq;
-
-	/* check that attributes are same and our struct still compatible with global shm_mq */
-	Assert(shm_mq_get_sender(mq) == amq->mq_sender);
-	Assert(shm_mq_get_receiver(mq) == amq->mq_receiver);
-
-	amq->mq_receiver = NULL;
-	amq->mq_detached = false;
-}
-
 static size_t
 jsonbc_shmem_size(void)
 {
@@ -131,7 +88,6 @@ jsonbc_shmem_size(void)
 static void
 jsonbc_shmem_startup_hook(void)
 {
-	int				mqkey;
 	bool			found;
 	Size			size = jsonbc_shmem_size();
 	jsonbc_shm_hdr *hdr;
@@ -146,6 +102,7 @@ jsonbc_shmem_startup_hook(void)
 	if (!found)
 	{
 		int i;
+		int	mqkey;
 
 		toc = shm_toc_create(JSONBC_SHM_MQ_MAGIC, workers_data, size);
 		hdr = shm_toc_allocate(toc, sizeof(jsonbc_shm_hdr));

jsonbc.h

@@ -30,15 +30,31 @@ typedef struct jsonbc_shm_worker
 	PGPROC			   *proc;
 } jsonbc_shm_worker;
 
+typedef struct jsonbc_pair
+{
+	int32	 id;
+	char	*key;
+} jsonbc_pair;
+
 typedef struct jsonbc_cached_cmopt
 {
-	HTAB	*keys;
-} cmopt_cached;
+	Oid		 cmoptoid;
+	HTAB	*key_cache;
+	HTAB	*id_cache;
+} jsonbc_cached_opt;
 
 typedef struct jsonbc_cached_key
 {
-	HTAB *
-}
+	uint32	 keyhash;
+	uint32	 pairslen;
+	jsonbc_pair	**pairs;	/* for hash collisions */
+} jsonbc_cached_key;
+
+typedef struct jsonbc_cached_id
+{
+	uint32	 id;
+	jsonbc_pair	*pair;
+} jsonbc_cached_id;
 
 extern void _PG_init(void);
 extern void jsonbc_register_worker(int n);
@@ -48,7 +64,4 @@ extern int jsonbc_nworkers;
 extern int jsonbc_cache_size;
 extern int jsonbc_queue_size;
 
-extern void shm_mq_clean_receiver(shm_mq *mq);
-extern void shm_mq_clean_sender(shm_mq *mq);
-
 #endif

jsonbc_utils.c

@@ -0,0 +1,157 @@
+#include "jsonbc.h"
+#include "jsonbc_utils.h"
+
+#include "postgres.h"
+#include "nodes/execnodes.h"
+
+#if PG_VERSION_NUM == 110000
+struct shm_mq_alt
+{
+	slock_t		mq_mutex;
+	PGPROC	   *mq_receiver;	/* this one */
+	PGPROC	   *mq_sender;		/* this one */
+	uint64		mq_bytes_read;
+	uint64		mq_bytes_written;
+	Size		mq_ring_size;
+	bool		mq_detached;	/* and this one */
+
+	/* in postgres version there are more attributes, but we don't need them */
+};
+#else
+#error "shm_mq struct in jsonbc is copied from PostgreSQL 11, please correct it according to your version"
+#endif
+
+RangeTblEntry *
+add_range_table_to_estate(EState *estate, Relation rel)
+{
+	RangeTblEntry *rte = makeNode(RangeTblEntry);
+	char	   *refname = RelationGetRelationName(rel);
+
+	Assert(pstate != NULL);
+
+	rte->rtekind = RTE_RELATION;
+	rte->alias = NULL;
+	rte->relid = RelationGetRelid(rel);
+	rte->relkind = rel->rd_rel->relkind;
+
+	rte->eref = makeAlias(refname, NIL);
+
+	rte->lateral = false;
+	rte->inh = false;
+	rte->inFromCl = false;
+
+	rte->requiredPerms = ACL_SELECT;
+	rte->checkAsUser = InvalidOid;
+	rte->selectedCols = NULL;
+	rte->insertedCols = NULL;
+	rte->updatedCols = NULL;
+
+	estate->es_range_table = lappend(estate->es_range_table, rte);
+
+	return rte;
+}
+
+/**
+ * Get 32-bit Murmur3 hash. Ported from qLibc library.
+ * Added compability with C99, and postgres code style
+ *
+ * @param data      source data
+ * @param nbytes    size of data
+ *
+ * @return 32-bit unsigned hash value.
+ *
+ * @code
+ *  uint32_t hashval = qhashmurmur3_32((void*)"hello", 5);
+ * @endcode
+ *
+ * @code
+ *  MurmurHash3 was created by Austin Appleby  in 2008. The initial
+ *  implementation was published in C++ and placed in the public.
+ *    https://sites.google.com/site/murmurhash/
+ *  Seungyoung Kim has ported its implementation into C language
+ *  in 2012 and published it as a part of qLibc component.
+ * @endcode
+ */
+uint32 qhashmurmur3_32(const void *data, size_t nbytes)
+{
+    int i;
+    uint32 k;
+    const int nblocks;
+    const uint32 *blocks;
+    const uint8 *tail;
+
+    const uint32 c1 = 0xcc9e2d51;
+    const uint32 c2 = 0x1b873593;
+
+    uint32 h = 0;
+
+	Assert(data != NULL && nbytes > 0);
+
+    nblocks = nbytes / 4;
+    blocks = (const uint32 *) (data);
+    tail = (const uint8 *) (data + (nblocks * 4));
+
+    for (i = 0; i < nblocks; i++)
+	{
+        k = blocks[i];
+
+        k *= c1;
+        k = (k << 15) | (k >> (32 - 15));
+        k *= c2;
+
+        h ^= k;
+        h = (h << 13) | (h >> (32 - 13));
+        h = (h * 5) + 0xe6546b64;
+    }
+
+    k = 0;
+    switch (nbytes & 3)
+	{
+        case 3:
+            k ^= tail[2] << 16;
+        case 2:
+            k ^= tail[1] << 8;
+        case 1:
+            k ^= tail[0];
+            k *= c1;
+            k = (k << 15) | (k >> (32 - 15));
+            k *= c2;
+            h ^= k;
+    };
+
+    h ^= nbytes;
+
+    h ^= h >> 16;
+    h *= 0x85ebca6b;
+    h ^= h >> 13;
+    h *= 0xc2b2ae35;
+    h ^= h >> 16;
+
+    return h;
+}
+
+void
+shm_mq_clean_sender(shm_mq *mq)
+{
+	struct shm_mq_alt	*amq = (struct shm_mq_alt *) mq;
+
+	/* check that attributes are same and our struct still compatible with global shm_mq */
+	Assert(shm_mq_get_sender(mq) == amq->mq_sender);
+	Assert(shm_mq_get_receiver(mq) == amq->mq_receiver);
+
+	amq->mq_sender = NULL;
+	amq->mq_detached = false;
+}
+
+void
+shm_mq_clean_receiver(shm_mq *mq)
+{
+	struct shm_mq_alt	*amq = (struct shm_mq_alt *) mq;
+
+	/* check that attributes are same and our struct still compatible with global shm_mq */
+	Assert(shm_mq_get_sender(mq) == amq->mq_sender);
+	Assert(shm_mq_get_receiver(mq) == amq->mq_receiver);
+
+	amq->mq_receiver = NULL;
+	amq->mq_detached = false;
+}

jsonbc_utils.h

@@ -0,0 +1,11 @@
+#ifndef JSONBC_UTILS_H
+#define JSONBC_UTILS_H
+
+#include "postgres.h"
+#include "nodes/execnodes.h"
+
+extern RangeTblEntry * add_range_table_to_estate(EState *estate, Relation rel);
+extern void shm_mq_clean_receiver(shm_mq *mq);
+extern void shm_mq_clean_sender(shm_mq *mq);
+
+#endif