Changeset: edf1bc1d80f9 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/edf1bc1d80f9
Modified Files:
gdk/gdk.h
monetdb5/modules/mal/heapn.c
monetdb5/modules/mal/mat.c
monetdb5/modules/mal/pipeline.c
monetdb5/modules/mal/pipeline.h
monetdb5/modules/mal/pp_algebra.c
monetdb5/modules/mal/pp_hash.c
monetdb5/modules/mal/pp_mat.c
monetdb5/modules/mal/pp_slicer.c
monetdb5/modules/mal/pp_sort.c
sql/backends/monet5/copy.c
sql/backends/monet5/vaults/parquet/parquet.c
Branch: pp_hashjoin
Log Message:
rename macros of pipeline code and put them in a single place
diffs (truncated from 575 to 300 lines):
diff --git a/gdk/gdk.h b/gdk/gdk.h
--- a/gdk/gdk.h
+++ b/gdk/gdk.h
@@ -438,8 +438,9 @@ typedef struct BAT {
#endif
Heap *torderidx; /* order oid index */
Strimps *tstrimps; /* string imprint index */
+ PROPrec *tprops; /* list of dynamic properties stored in the bat
descriptor */
+
struct pipeline_io *pl_io;
- PROPrec *tprops; /* list of dynamic properties stored in the bat
descriptor */
MT_Lock theaplock; /* lock protecting heap reference changes */
MT_RWLock thashlock; /* lock specifically for hash management */
diff --git a/monetdb5/modules/mal/heapn.c b/monetdb5/modules/mal/heapn.c
--- a/monetdb5/modules/mal/heapn.c
+++ b/monetdb5/modules/mal/heapn.c
@@ -1056,7 +1056,7 @@ _heap_create( int size, bool shared, boo
heapn *h = (heapn*)GDKzalloc(sizeof(heapn));
h->s.destroy = (pipeline_io_destroy)heap_destroy;
- h->s.type = HEAP_SINK;
+ h->s.type = PIPELINE_IO_HEAP;
h->shared = shared;
h->grouped = grouped;
h->size = size;
@@ -1524,7 +1524,7 @@ HEAPtopn(Client cntxt, MalBlkPtr m, MalS
}
private = hps->tprivate_bat;
heapn *hp = (heapn*)hps->pl_io;
- assert(hp && hp->s.type == HEAP_SINK);
+ assert(hp && hp->s.type == PIPELINE_IO_HEAP);
if (((hp->sub && hp->sub->vb == NULL) || (hp->grouped && hp->grpb ==
NULL)) && !_heap_init(hp)) {
BBPreclaim(b);
BBPreclaim(gps);
diff --git a/monetdb5/modules/mal/mat.c b/monetdb5/modules/mal/mat.c
--- a/monetdb5/modules/mal/mat.c
+++ b/monetdb5/modules/mal/mat.c
@@ -61,7 +61,7 @@ MATpackInternal(Client cntxt, MalBlkPtr
bat bid = stk->stk[getArg(p, i)].val.bval;
b = BBPquickdesc(bid);
mat_t *mp = (mat_t *) b->pl_io;
- if (mp && mp->s.type == MAT_SINK) {
+ if (mp && mp->s.type == PIPELINE_IO_MAT) {
bn = pack_mat(b);
if (bn == NULL)
throw(MAL, "mat.pack", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
@@ -146,7 +146,7 @@ MATpackIncrement(Client cntxt, MalBlkPtr
if (getArgType(mb, p, 2) == TYPE_int) {
mat_t *mp = (mat_t *) b->pl_io;
- if (mp && mp->s.type == MAT_SINK) {
+ if (mp && mp->s.type == PIPELINE_IO_MAT) {
bn = pack_mat(b);
if (bn == NULL)
throw(MAL, "mat.pack", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
diff --git a/monetdb5/modules/mal/pipeline.c b/monetdb5/modules/mal/pipeline.c
--- a/monetdb5/modules/mal/pipeline.c
+++ b/monetdb5/modules/mal/pipeline.c
@@ -117,7 +117,6 @@ sleep_ns( int ns)
#endif
}
-#define COUNTER_SINK 98
typedef struct pp_counter_t {
struct pipeline_io s;
@@ -206,9 +205,9 @@ PPcounter_get(Client cntxt, MalBlkPtr mb
BBPunfix(b->batCacheid);
throw(MAL, "pipeline.counter_get", SQLSTATE(HY002) "Missing
source sink");
}
- if (c->s.type != COUNTER_SINK) {
+ if (c->s.type != PIPELINE_IO_COUNTER) {
BBPunfix(b->batCacheid);
- throw(MAL, "pipeline.counter_get", SQLSTATE(HY002) "Invalid
source type %d, expected %d", c->s.type, COUNTER_SINK);
+ throw(MAL, "pipeline.counter_get", SQLSTATE(HY002) "Invalid
source type %d, expected %d", c->s.type, PIPELINE_IO_COUNTER);
}
if (c->sync && c->scnt != (int)p->p->nr_workers) {
MT_lock_set(&p->p->l);
@@ -253,7 +252,7 @@ PPcounter(Client cntxt, MalBlkPtr mb, Ma
return createException(SQL, "pipeline.counter",
SQLSTATE(HY002) RUNTIME_OBJECT_MISSING);
size_t cnt = 0;
hash_table *h = (hash_table*)b->pl_io;
- if (h && h->s.type == OA_HASH_TABLE_SINK) {
+ if (h && h->s.type == PIPELINE_IO_HASH_TABLE) {
cnt = h->size;
} else {
cnt = BATcount(b);
@@ -285,7 +284,7 @@ PPcounter(Client cntxt, MalBlkPtr mb, Ma
}
b->pl_io = (struct pipeline_io*)c;
- c->s.type = COUNTER_SINK;
+ c->s.type = PIPELINE_IO_COUNTER;
c->s.destroy = (pipeline_io_destroy)&counter_free;
c->s.done = (pipeline_io_done)&counter_done;
c->current = 0;
@@ -323,7 +322,6 @@ PPdone(Client cntxt, MalBlkPtr mb, MalSt
return MAL_SUCCEED;
}
-#define CONCAT_SINK 99
typedef struct pp_concat_t {
struct pipeline_io s;
@@ -351,7 +349,7 @@ static int
concat_done( pp_concat *c, int wid, int nr_workers, bool redo )
{
int res = 1;
- assert(c->s.type == CONCAT_SINK);
+ assert(c->s.type == PIPELINE_IO_CONCAT);
MT_lock_set(&c->l);
if (!c->started) {
c->cur = (int*)GDKzalloc(sizeof(int) * nr_workers);
@@ -381,7 +379,7 @@ static int
concat_next( pp_concat *c, int wid)
{
int res = 1;
- assert(c->s.type == CONCAT_SINK);
+ assert(c->s.type == PIPELINE_IO_CONCAT);
MT_lock_set(&c->l);
assert(c->started);
BAT *sb = c->srcs[c->cur[wid]];
@@ -400,7 +398,7 @@ static BAT*
concat_next_bat( pp_concat *c, int wid)
{
BAT *res = NULL;
- assert(c->s.type == CONCAT_SINK);
+ assert(c->s.type == PIPELINE_IO_CONCAT);
MT_lock_set(&c->l);
assert(c->started);
BAT *sb = c->srcs[c->cur[wid]];
@@ -435,7 +433,7 @@ PPconcat_block(Client cntxt, MalBlkPtr m
if (!b)
throw(MAL, "pipeline.concat_block", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
pp_concat *pcat = (pp_concat*)b->pl_io;
- if (pcat->s.type != CONCAT_SINK) {
+ if (pcat->s.type != PIPELINE_IO_CONCAT) {
BBPreclaim(b);
throw(MAL, "pipeline.concat_block", SQLSTATE(HY002) "Invalid
type for a concat source %d", pcat->s.type);
}
@@ -464,7 +462,7 @@ PPconcat_add(Client cntxt, MalBlkPtr mb,
throw(MAL, "pipeline.concat_add", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
}
pp_concat *pcat = (pp_concat*)b->pl_io;
- if (pcat->s.type != CONCAT_SINK) {
+ if (pcat->s.type != PIPELINE_IO_CONCAT) {
BBPreclaim(b);
BBPreclaim(i);
throw(MAL, "pipeline.concat_add", SQLSTATE(HY002) "Invalid type
for a concat source %d", pcat->s.type);
@@ -498,7 +496,7 @@ PPconcat(Client cntxt, MalBlkPtr mb, Mal
throw(SQL, "pipeline.concat", SQLSTATE(HY013) MAL_MALLOC_FAIL);
}
b->pl_io = (struct pipeline_io*)pcat;
- pcat->s.type = CONCAT_SINK;
+ pcat->s.type = PIPELINE_IO_CONCAT;
pcat->s.destroy = (pipeline_io_destroy)&concat_free;
pcat->s.done = (pipeline_io_done)&concat_done;
pcat->s.next = (pipeline_io_next)&concat_next;
diff --git a/monetdb5/modules/mal/pipeline.h b/monetdb5/modules/mal/pipeline.h
--- a/monetdb5/modules/mal/pipeline.h
+++ b/monetdb5/modules/mal/pipeline.h
@@ -11,6 +11,9 @@
#ifndef _PIPELINE_H_
#define _PIPELINE_H_
+#include "gdk.h"
+#include "mal_pipelines.h"
+
#define pipeline_lock(pl) MT_lock_set(&pl->p->l)
#define pipeline_unlock(pl) MT_lock_unset(&pl->p->l)
@@ -23,12 +26,17 @@
#define SLICE_SIZE 100000
// TODO a better way to define/add/register sinks, similar to types
-#define OA_HASH_TABLE_SINK 1
-#define TOPN_SINK 3
-#define HEAP_SINK 4
-#define PART_SINK 5
-#define MAT_SINK 6
-#define COPY_SINK 42
+#define PIPELINE_IO_HASH_TABLE 1
+#define PIPELINE_IO_SOP 2 /* set of ordered parts */
+#define PIPELINE_IO_TOPN 3
+#define PIPELINE_IO_HEAP 4
+#define PIPELINE_IO_PART 5
+#define PIPELINE_IO_MAT 6
+#define PIPELINE_IO_COPY 7
+#define PIPELINE_IO_COUNTER 8
+#define PIPELINE_IO_CONCAT 9
+#define PIPELINE_IO_PARQUET 10
+#define PIPELINE_IO_MPARQUET 10
extern int BATupgrade(BAT *r, BAT *b, bool locked);
extern void BATswap_heaps(BAT *u, BAT *b, Pipeline *p);
diff --git a/monetdb5/modules/mal/pp_algebra.c
b/monetdb5/modules/mal/pp_algebra.c
--- a/monetdb5/modules/mal/pp_algebra.c
+++ b/monetdb5/modules/mal/pp_algebra.c
@@ -1061,7 +1061,7 @@ LALGunique(Client ctx, bat *rid, bat *ui
}
hash_table *h = (hash_table*)u->pl_io;
- assert(h && h->s.type == OA_HASH_TABLE_SINK);
+ assert(h && h->s.type == PIPELINE_IO_HASH_TABLE);
MT_lock_set(&u->theaplock);
MT_lock_set(&b->theaplock);
if (ATOMvarsized(u->ttype) /*&& !VIEWvtparent(b)*/) {
@@ -1313,7 +1313,7 @@ LALGgroup_unique(Client ctx, bat *rid, b
}
hash_table *h = (hash_table*)u->pl_io;
- assert(h && h->s.type == OA_HASH_TABLE_SINK);
+ assert(h && h->s.type == PIPELINE_IO_HASH_TABLE);
MT_lock_set(&u->theaplock);
MT_lock_set(&b->theaplock);
if (ATOMvarsized(u->ttype) /*&& !VIEWvtparent(b)*/) {
@@ -1662,7 +1662,7 @@ LALGgroup(Client ctx, bat *rid, bat *uid
//(void)sid;
hash_table *h = (hash_table*)u->pl_io;
- assert(h && h->s.type == OA_HASH_TABLE_SINK);
+ assert(h && h->s.type == PIPELINE_IO_HASH_TABLE);
MT_lock_set(&u->theaplock);
MT_lock_set(&b->theaplock);
if ((ATOMvarsized(u->ttype) && !VIEWvtparent(b)) ||
@@ -1956,7 +1956,7 @@ LALGderive(Client ctx, bat *rid, bat *ui
//(void)sid;
hash_table *h = (hash_table*)u->pl_io;
- assert(h && h->s.type == OA_HASH_TABLE_SINK);
+ assert(h && h->s.type == PIPELINE_IO_HASH_TABLE);
MT_lock_set(&u->theaplock);
MT_lock_set(&b->theaplock);
if ((ATOMvarsized(u->ttype) && !VIEWvtparent(b)) ||
diff --git a/monetdb5/modules/mal/pp_hash.c b/monetdb5/modules/mal/pp_hash.c
--- a/monetdb5/modules/mal/pp_hash.c
+++ b/monetdb5/modules/mal/pp_hash.c
@@ -48,7 +48,7 @@ _ht_init(hash_table *h)
if (h->vals == NULL || h->gids == NULL)
goto error;
if (h->p) {
- assert(h->s.type == OA_HASH_TABLE_SINK);
+ assert(h->s.type == PIPELINE_IO_HASH_TABLE);
h->pgids = (gid*)GDKmalloc(sizeof(gid)* h->size);
if (h->pgids == NULL)
goto error;
@@ -96,7 +96,7 @@ _ht_create( int type, size_t size, hash_
if (!type)
type = TYPE_oid;
h->s.destroy = (pipeline_io_destroy)&ht_destroy;
- h->s.type = OA_HASH_TABLE_SINK;
+ h->s.type = PIPELINE_IO_HASH_TABLE;
if (bits >= GIDBITS)
bits = GIDBITS-1;
h->bits = bits;
@@ -467,7 +467,7 @@ OAHASHhashmark_init(Client ctx, bat *res
hash_table *h = (hash_table*)ht->pl_io;
if (hp)
h = (hash_table*)hp->pl_io;
- //assert(h && h->s.type == OA_HASH_TABLE_SINK);
+ //assert(h && h->s.type == PIPELINE_IO_HASH_TABLE);
BUN sz = h?h->last:BATcount(ht); /* no hash ie outer cross product case
*/
r = COLnew(0, TYPE_bit, sz, TRANSIENT);
@@ -532,7 +532,7 @@ UHASHext(Client cntxt, MalBlkPtr m, MalS
if (!i)
return createException(MAL, "hash.ext", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
hash_table *h = (hash_table*)i->pl_io;
- if (!h || h->s.type != OA_HASH_TABLE_SINK) {
+ if (!h || h->s.type != PIPELINE_IO_HASH_TABLE) {
BBPreclaim(i);
return createException(MAL, "hash.ext", SQLSTATE(HY002)
"Missing hash table");
}
@@ -876,7 +876,7 @@ OAHASHbuild_tbl(Client ctx, bat *slot_id
goto error;
}
hash_table *h = (hash_table*)u->pl_io;
- assert(h && h->s.type == OA_HASH_TABLE_SINK);
+ assert(h && h->s.type == PIPELINE_IO_HASH_TABLE);
BUN cnt = BATcount(b);
g = COLnew(b->hseqbase, TYPE_oid, cnt, TRANSIENT);
@@ -1278,7 +1278,7 @@ OAHASHbuild_tbl_cmbd(Client ctx, bat *sl
goto error;
}
hash_table *h = (hash_table*)u->pl_io;
- assert(h && h->s.type == OA_HASH_TABLE_SINK);
+ assert(h && h->s.type == PIPELINE_IO_HASH_TABLE);
BUN cnt = BATcount(b);
g = COLnew(b->hseqbase, TYPE_oid, cnt, TRANSIENT);
@@ -3278,7 +3278,7 @@ OAHASHno_slices(Client ctx, int *no_slic
if (!b)
return createException(SQL, "oahash.no_slices", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
hash_table *h = (hash_table*)b->pl_io;
- assert(h && h->s.type == OA_HASH_TABLE_SINK);
+ assert(h && h->s.type == PIPELINE_IO_HASH_TABLE);
if (h->size < SLICE_SIZE )
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]
