Changeset: d6ddf6853b65 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/d6ddf6853b65
Modified Files:
gdk/gdk.h
monetdb5/mal/mal_pipelines.c
monetdb5/modules/mal/heapn.c
monetdb5/modules/mal/mat.c
monetdb5/modules/mal/pipeline.c
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/copy_convert.c
sql/backends/monet5/generator/generator.c
sql/backends/monet5/sql.c
sql/backends/monet5/vaults/parquet/parquet.c
Branch: pp_hashjoin
Log Message:
bat struct tsink rename and callbacks to manage pl_io struct
diffs (truncated from 929 to 300 lines):
diff --git a/gdk/gdk.h b/gdk/gdk.h
--- a/gdk/gdk.h
+++ b/gdk/gdk.h
@@ -336,19 +336,21 @@ gdk_export bool VALisnil(const ValRecord
typedef struct PROPrec PROPrec;
-typedef void (*sink_destroy)(void *sink);
-typedef int (*sink_done)(void *sink, int wid, int nr_workers, bool redo);
-typedef int (*sink_next)(void *sink, int wid);
-typedef void *(*sink_next_bat)(void *sink, int wid);
-typedef struct Sink {
- sink_destroy destroy;
- sink_done done;
- sink_next next; /* counter incrementing sources */
- sink_next_bat next_bat; /* bat generating sources */
+typedef void (*pl_io_destroy)(void *pl_io);
+typedef int (*pl_io_done)(void *pl_io, int wid, int nr_workers, bool redo);
+typedef int (*pl_io_next)(void *pl_io, int wid);
+typedef void *(*pl_io_next_bat)(void *pl_io, int wid);
+
+typedef struct pipeline_io {
+ pl_io_destroy destroy;
+ pl_io_done done;
+ pl_io_next next; /* counter incrementing sources */
+ pl_io_next_bat next_bat; /* bat generating sources */
int type; /* sink/source type */
char *error;
-} Sink;
-#define TSKdestroy(b) if (b->tsink && b->tsink->destroy) {
b->tsink->destroy(b->tsink); b->tsink = NULL; }
+} pl_source, pl_sink, Sink;
+
+#define TSKdestroy(b) if (b->pl_io && b->pl_io->destroy) {
b->pl_io->destroy(b->pl_io); b->pl_io = NULL; }
#define TSKfree(b) TSKdestroy(b)
#define ORDERIDXOFF 3
@@ -436,7 +438,7 @@ typedef struct BAT {
#endif
Heap *torderidx; /* order oid index */
Strimps *tstrimps; /* string imprint index */
- Sink *tsink;
+ struct pipeline_io *pl_io;
PROPrec *tprops; /* list of dynamic properties stored in the bat
descriptor */
MT_Lock theaplock; /* lock protecting heap reference changes */
diff --git a/monetdb5/mal/mal_pipelines.c b/monetdb5/mal/mal_pipelines.c
--- a/monetdb5/mal/mal_pipelines.c
+++ b/monetdb5/mal/mal_pipelines.c
@@ -338,7 +338,7 @@ runMALpipelines(Client cntxt, MalBlkPtr
if (!sb) {
err = createException(SQL, "language.pipeline",
SQLSTATE(HY002) RUNTIME_OBJECT_MISSING);
} else {
- Sink *sink = sb->tsink;
+ Sink *sink = sb->pl_io;
sink->error = err;
BBPreclaim(sb);
}
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
@@ -987,7 +987,7 @@ HEAPnew_topn( MalStkPtr s, InstrPtr p, i
heapn_done(hp);
b = COLnew(0, TYPE_oid, hp->grouped?n*256:n, TRANSIENT);
if (b)
- b->tsink = (Sink*)hp;
+ b->pl_io = (Sink*)hp;
else
heap_destroy(hp);
return b;
@@ -1055,7 +1055,7 @@ _heap_create( int size, bool shared, boo
{
heapn *h = (heapn*)GDKzalloc(sizeof(heapn));
- h->s.destroy = (sink_destroy)heap_destroy;
+ h->s.destroy = (pl_io_destroy)heap_destroy;
h->s.type = HEAP_SINK;
h->shared = shared;
h->grouped = grouped;
@@ -1227,7 +1227,7 @@ HEAPnew_new( MalBlkPtr m, MalStkPtr s, I
}
BAT *b = COLnew(0, TYPE_oid, hp->grouped?(n*256):n, TRANSIENT);
if (b)
- b->tsink = (Sink*)hp;
+ b->pl_io = (Sink*)hp;
else
heap_destroy(hp);
return b;
@@ -1523,7 +1523,7 @@ HEAPtopn(Client cntxt, MalBlkPtr m, MalS
hps = BATdescriptor(*HP);
}
private = hps->tprivate_bat;
- heapn *hp = (heapn*)hps->tsink;
+ heapn *hp = (heapn*)hps->pl_io;
assert(hp && hp->s.type == HEAP_SINK);
if (((hp->sub && hp->sub->vb == NULL) || (hp->grouped && hp->grpb ==
NULL)) && !_heap_init(hp)) {
BBPreclaim(b);
@@ -1832,7 +1832,7 @@ HEAPorder(Client ctx, bat *rid, bat *hb,
if (!hpb)
throw(MAL, "heapn.order", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
- heapn *hp = (heapn*)hpb->tsink;
+ heapn *hp = (heapn*)hpb->pl_io;
r = COLnew(0, TYPE_oid, hp->grouped?hp->gsize*hp->size:hp->size,
TRANSIENT);
if (!r) {
BBPreclaim(hpb);
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
@@ -60,7 +60,7 @@ MATpackInternal(Client cntxt, MalBlkPtr
for (i = 1; i < p->argc; i++) {
bat bid = stk->stk[getArg(p, i)].val.bval;
b = BBPquickdesc(bid);
- mat_t *mp = (mat_t *) b->tsink;
+ mat_t *mp = (mat_t *) b->pl_io;
if (mp && mp->s.type == MAT_SINK) {
bn = pack_mat(b);
if (bn == NULL)
@@ -145,7 +145,7 @@ MATpackIncrement(Client cntxt, MalBlkPtr
throw(MAL, "mat.pack", SQLSTATE(HY002) RUNTIME_OBJECT_MISSING);
if (getArgType(mb, p, 2) == TYPE_int) {
- mat_t *mp = (mat_t *) b->tsink;
+ mat_t *mp = (mat_t *) b->pl_io;
if (mp && mp->s.type == MAT_SINK) {
bn = pack_mat(b);
if (bn == NULL)
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
@@ -201,7 +201,7 @@ PPcounter_get(Client cntxt, MalBlkPtr mb
BAT *b = BATdescriptor(cb);
if (!b)
throw(MAL, "pipeline.counter_get", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
- pp_counter *c = (pp_counter*)b->tsink;
+ pp_counter *c = (pp_counter*)b->pl_io;
if (!c) {
BBPunfix(b->batCacheid);
throw(MAL, "pipeline.counter_get", SQLSTATE(HY002) "Missing
source sink");
@@ -252,7 +252,7 @@ PPcounter(Client cntxt, MalBlkPtr mb, Ma
if (!b)
return createException(SQL, "pipeline.counter",
SQLSTATE(HY002) RUNTIME_OBJECT_MISSING);
size_t cnt = 0;
- hash_table *h = (hash_table*)b->tsink;
+ hash_table *h = (hash_table*)b->pl_io;
if (h && h->s.type == OA_HASH_TABLE_SINK) {
cnt = h->size;
} else {
@@ -284,10 +284,10 @@ PPcounter(Client cntxt, MalBlkPtr mb, Ma
throw(SQL, "pipeline.counter", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
}
- b->tsink = (Sink*)c;
+ b->pl_io = (Sink*)c;
c->s.type = COUNTER_SINK;
- c->s.destroy = (sink_destroy)&counter_free;
- c->s.done = (sink_done)&counter_done;
+ c->s.destroy = (pl_io_destroy)&counter_free;
+ c->s.done = (pl_io_done)&counter_done;
c->current = 0;
c->cur = NULL;
c->nr = nr;
@@ -295,7 +295,7 @@ PPcounter(Client cntxt, MalBlkPtr mb, Ma
if (sync) {
c->sync = true;
c->scnt = 0;
- c->s.done = (sink_done)&sync_counter_done;
+ c->s.done = (pl_io_done)&sync_counter_done;
}
*rb = b->batCacheid;
BBPkeepref(b);
@@ -313,11 +313,11 @@ PPdone(Client cntxt, MalBlkPtr mb, MalSt
(void)cntxt; (void)mb;
BAT *b = BATdescriptor(B);
if (b) {
- if (!b->tsink) {
+ if (!b->pl_io) {
BBPunfix(b->batCacheid);
throw(MAL, "pipeline.done", SQLSTATE(HY002) "Missing
source sink");
}
- *res = b->tsink->done(b->tsink, p->wid, p->p->nr_workers, redo);
+ *res = b->pl_io->done(b->pl_io, p->wid, p->p->nr_workers, redo);
BBPunfix(b->batCacheid);
}
return MAL_SUCCEED;
@@ -359,7 +359,7 @@ concat_done( pp_concat *c, int wid, int
}
BAT *sb = c->srcs[c->cur[wid]];
if (sb) {
- Sink *s = sb->tsink;
+ Sink *s = sb->pl_io;
MT_lock_unset(&c->l);
res = s->done(s, wid, nr_workers, redo);
MT_lock_set(&c->l);
@@ -367,7 +367,7 @@ concat_done( pp_concat *c, int wid, int
sb = c->srcs[c->cur[wid]];
if (!sb)
break;
- s = sb->tsink;
+ s = sb->pl_io;
MT_lock_unset(&c->l);
res = s->done(s, wid, nr_workers, false);
MT_lock_set(&c->l);
@@ -386,7 +386,7 @@ concat_next( pp_concat *c, int wid)
assert(c->started);
BAT *sb = c->srcs[c->cur[wid]];
if (sb) {
- Sink *s = sb->tsink;
+ Sink *s = sb->pl_io;
MT_lock_unset(&c->l);
if (s->next)
res = s->next(s, wid);
@@ -405,7 +405,7 @@ concat_next_bat( pp_concat *c, int wid)
assert(c->started);
BAT *sb = c->srcs[c->cur[wid]];
if (sb) {
- Sink *s = sb->tsink;
+ Sink *s = sb->pl_io;
MT_lock_unset(&c->l);
if (s->next_bat)
res = s->next_bat(s, wid);
@@ -434,7 +434,7 @@ PPconcat_block(Client cntxt, MalBlkPtr m
BAT *b = BATdescriptor(cb);
if (!b)
throw(MAL, "pipeline.concat_block", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
- pp_concat *pcat = (pp_concat*)b->tsink;
+ pp_concat *pcat = (pp_concat*)b->pl_io;
if (pcat->s.type != CONCAT_SINK) {
BBPreclaim(b);
throw(MAL, "pipeline.concat_block", SQLSTATE(HY002) "Invalid
type for a concat source %d", pcat->s.type);
@@ -463,7 +463,7 @@ PPconcat_add(Client cntxt, MalBlkPtr mb,
BBPreclaim(i);
throw(MAL, "pipeline.concat_add", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
}
- pp_concat *pcat = (pp_concat*)b->tsink;
+ pp_concat *pcat = (pp_concat*)b->pl_io;
if (pcat->s.type != CONCAT_SINK) {
BBPreclaim(b);
BBPreclaim(i);
@@ -497,12 +497,12 @@ PPconcat(Client cntxt, MalBlkPtr mb, Mal
GDKfree(pcat);
throw(SQL, "pipeline.concat", SQLSTATE(HY013) MAL_MALLOC_FAIL);
}
- b->tsink = (Sink*)pcat;
+ b->pl_io = (Sink*)pcat;
pcat->s.type = CONCAT_SINK;
- pcat->s.destroy = (sink_destroy)&concat_free;
- pcat->s.done = (sink_done)&concat_done;
- pcat->s.next = (sink_next)&concat_next;
- pcat->s.next_bat = (sink_next_bat)&concat_next_bat;
+ pcat->s.destroy = (pl_io_destroy)&concat_free;
+ pcat->s.done = (pl_io_done)&concat_done;
+ pcat->s.next = (pl_io_next)&concat_next;
+ pcat->s.next_bat = (pl_io_next_bat)&concat_next_bat;
pcat->current = 0;
pcat->max = nr;
pcat->started = false;
@@ -534,8 +534,8 @@ PPresultset(Client cntxt, MalBlkPtr mb,
GDKfree(prs);
throw(SQL, "pipeline.resultset", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
}
- b->tsink = (Sink*)prs;
- prs->s.destroy = (sink_destroy)&GDKfree;
+ b->pl_io = (Sink*)prs;
+ prs->s.destroy = (pl_io_destroy)&GDKfree;
MT_lock_init(&prs->l, "resultset");
*rb = b->batCacheid;
BBPkeepref(b);
@@ -552,7 +552,7 @@ PPclaim(Client cntxt, MalBlkPtr mb, MalS
BAT *b = BATdescriptor(rb);
if (b) {
- pp_resultset *rs = (pp_resultset*)b->tsink;
+ pp_resultset *rs = (pp_resultset*)b->pl_io;
*res = ATOMIC_ADD(&rs->claimed, cnt);
BBPreclaim(b);
}
@@ -577,7 +577,7 @@ PPidentity(Client cntxt, MalBlkPtr mb, M
BBPreclaim(b);
b = BATdescriptor(rb);
if (b) {
- pp_resultset *rs = (pp_resultset*)b->tsink;
+ pp_resultset *rs = (pp_resultset*)b->pl_io;
offset = ATOMIC_ADD(&rs->claimed, cnt);
BBPreclaim(b);
@@ -612,7 +612,7 @@ PPappend(Client cntxt, MalBlkPtr mb, Mal
BBPreclaim(r);
throw(MAL, "bat.append", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
}
- pp_resultset *pp_rs = (pp_resultset*)r->tsink;
+ pp_resultset *pp_rs = (pp_resultset*)r->pl_io;
(void)pp_rs;
if (i && (i->ttype == TYPE_msk || mask_cand(i))) {
@@ -661,7 +661,7 @@ source_next(Client cntxt, MalBlkPtr mb,
if (!s)
throw(MAL, "source.next", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
- Sink *src = s->tsink;
+ Sink *src = s->pl_io;
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]
