Changeset: f2ec65936609 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/f2ec65936609
Modified Files:
gdk/gdk.h
monetdb5/mal/mal_pipelines.c
monetdb5/modules/mal/heapn.c
monetdb5/modules/mal/pipeline.c
monetdb5/modules/mal/pp_algebra.c
monetdb5/modules/mal/pp_hash.c
monetdb5/modules/mal/pp_hash.h
monetdb5/modules/mal/pp_mat.c
monetdb5/modules/mal/pp_mat.h
monetdb5/modules/mal/pp_slicer.c
monetdb5/modules/mal/pp_sort.c
sql/backends/monet5/copy.c
sql/backends/monet5/copy.h
sql/backends/monet5/generator/generator.c
sql/backends/monet5/sql.c
sql/backends/monet5/vaults/parquet/parquet.c
Branch: pp_hashjoin
Log Message:
Use struct pipeline_io instead of Sink
diffs (truncated from 398 to 300 lines):
diff --git a/gdk/gdk.h b/gdk/gdk.h
--- a/gdk/gdk.h
+++ b/gdk/gdk.h
@@ -348,7 +348,7 @@ typedef struct pipeline_io {
pl_io_next_bat next_bat; /* bat generating sources */
int type; /* sink/source type */
char *error;
-} pl_source, pl_sink, Sink;
+} pl_source, pl_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)
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->pl_io;
+ struct pipeline_io *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
@@ -48,7 +48,7 @@ typedef struct subheap {
} subheap;
typedef struct heapn {
- Sink s;
+ struct pipeline_io s;
size_t size;
size_t used;
bool full;
@@ -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->pl_io = (Sink*)hp;
+ b->pl_io = (struct pipeline_io*)hp;
else
heap_destroy(hp);
return b;
@@ -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->pl_io = (Sink*)hp;
+ b->pl_io = (struct pipeline_io*)hp;
else
heap_destroy(hp);
return b;
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
@@ -119,7 +119,7 @@ sleep_ns( int ns)
#define COUNTER_SINK 98
typedef struct pp_counter_t {
- Sink s;
+ struct pipeline_io s;
MT_Lock l;
int nr;
@@ -284,7 +284,7 @@ PPcounter(Client cntxt, MalBlkPtr mb, Ma
throw(SQL, "pipeline.counter", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
}
- b->pl_io = (Sink*)c;
+ b->pl_io = (struct pipeline_io*)c;
c->s.type = COUNTER_SINK;
c->s.destroy = (pl_io_destroy)&counter_free;
c->s.done = (pl_io_done)&counter_done;
@@ -325,7 +325,7 @@ PPdone(Client cntxt, MalBlkPtr mb, MalSt
#define CONCAT_SINK 99
typedef struct pp_concat_t {
- Sink s;
+ struct pipeline_io s;
MT_Lock l;
int current;
@@ -359,7 +359,7 @@ concat_done( pp_concat *c, int wid, int
}
BAT *sb = c->srcs[c->cur[wid]];
if (sb) {
- Sink *s = sb->pl_io;
+ struct pipeline_io *s = sb->pl_io;
MT_lock_unset(&c->l);
res = s->done(s, wid, nr_workers, redo);
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->pl_io;
+ struct pipeline_io *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->pl_io;
+ struct pipeline_io *s = sb->pl_io;
MT_lock_unset(&c->l);
if (s->next_bat)
res = s->next_bat(s, wid);
@@ -487,7 +487,7 @@ PPconcat(Client cntxt, MalBlkPtr mb, Mal
(void)mb;
bat *rb = getArgReference_bat(stk, pci, 0);
int nr = *getArgReference_int(stk, pci, 1);
- pp_concat *pcat = (pp_concat*)GDKzalloc(sizeof(pp_concat) + (nr+1) *
sizeof(Sink*) );
+ pp_concat *pcat = (pp_concat*)GDKzalloc(sizeof(pp_concat) + (nr+1) *
sizeof(struct pipeline_io*) );
if (!pcat)
throw(SQL, "pipeline.concat", SQLSTATE(HY013) MAL_MALLOC_FAIL);
@@ -497,7 +497,7 @@ PPconcat(Client cntxt, MalBlkPtr mb, Mal
GDKfree(pcat);
throw(SQL, "pipeline.concat", SQLSTATE(HY013) MAL_MALLOC_FAIL);
}
- b->pl_io = (Sink*)pcat;
+ b->pl_io = (struct pipeline_io*)pcat;
pcat->s.type = CONCAT_SINK;
pcat->s.destroy = (pl_io_destroy)&concat_free;
pcat->s.done = (pl_io_done)&concat_done;
@@ -513,7 +513,7 @@ PPconcat(Client cntxt, MalBlkPtr mb, Mal
}
typedef struct pp_resultset_t {
- Sink s;
+ struct pipeline_io s;
ATOMIC_TYPE claimed;
MT_Lock l;
} pp_resultset;
@@ -534,7 +534,7 @@ PPresultset(Client cntxt, MalBlkPtr mb,
GDKfree(prs);
throw(SQL, "pipeline.resultset", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
}
- b->pl_io = (Sink*)prs;
+ b->pl_io = (struct pipeline_io*)prs;
prs->s.destroy = (pl_io_destroy)&GDKfree;
MT_lock_init(&prs->l, "resultset");
*rb = b->batCacheid;
@@ -661,7 +661,7 @@ source_next(Client cntxt, MalBlkPtr mb,
if (!s)
throw(MAL, "source.next", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
- Sink *src = s->pl_io;
+ struct pipeline_io *src = s->pl_io;
if (!src || !src->next_bat) {
BBPreclaim(s);
throw(MAL, "source.next", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
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
@@ -1643,7 +1643,7 @@ LALGgroup(Client ctx, bat *rid, bat *uid
err = createException(MAL, "pp group.group",
SQLSTATE(HY013) MAL_MALLOC_FAIL);
goto error;
}
- u->pl_io = (Sink*)ht_create(b->ttype?b->ttype:TYPE_oid, 1,
NULL);
+ u->pl_io = (struct
pipeline_io*)ht_create(b->ttype?b->ttype:TYPE_oid, 1, NULL);
if (u->pl_io == NULL) {
err = createException(MAL, "pp group.group",
SQLSTATE(HY013) MAL_MALLOC_FAIL);
goto error;
@@ -1936,7 +1936,7 @@ LALGderive(Client ctx, bat *rid, bat *ui
goto error;
}
/* Lookup parent hash */
- u->pl_io = (Sink*)ht_create(b->ttype?b->ttype:TYPE_oid, 1,
(hash_table*)H->pl_io);
+ u->pl_io = (struct
pipeline_io*)ht_create(b->ttype?b->ttype:TYPE_oid, 1, (hash_table*)H->pl_io);
if (u->pl_io == NULL) {
BBPunfix(H->batCacheid);
err = createException(MAL, "pp group.group(derive)",
SQLSTATE(HY013) MAL_MALLOC_FAIL);
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
@@ -428,7 +428,7 @@ OAHASHnew(Client cntxt, MalBlkPtr m, Mal
BBPreclaim(pht);
return createException(MAL, "oahash.new", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
}
- b->pl_io = (Sink*)ht_create(tt, (size_t)size, parent);
+ b->pl_io = (struct pipeline_io*)ht_create(tt, (size_t)size, parent);
BBPreclaim(pht);
if (b->pl_io == NULL) {
BBPunfix(b->batCacheid);
diff --git a/monetdb5/modules/mal/pp_hash.h b/monetdb5/modules/mal/pp_hash.h
--- a/monetdb5/modules/mal/pp_hash.h
+++ b/monetdb5/modules/mal/pp_hash.h
@@ -146,7 +146,7 @@ typedef lng (*fhsh)(const void *v);
typedef size_t (*flen)(const void *v);
typedef struct hash_table {
- Sink s;
+ struct pipeline_io s;
int type;
int width;
fcmp cmp;
diff --git a/monetdb5/modules/mal/pp_mat.c b/monetdb5/modules/mal/pp_mat.c
--- a/monetdb5/modules/mal/pp_mat.c
+++ b/monetdb5/modules/mal/pp_mat.c
@@ -23,7 +23,7 @@
#include "pipeline.h"
typedef struct part_t {
- Sink s;
+ struct pipeline_io s;
int nr;
lng *curpos;
MT_Lock l;
@@ -136,14 +136,14 @@ MATnew(Client cntxt, MalBlkPtr mb, MalSt
break;
BATnegateprops(b);
if (hashsize)
- b->pl_io = (Sink*)ht_create(tt, (size_t)(hashsize),
pmat?(hash_table*)pmat->bat[i]->pl_io:NULL);
+ b->pl_io = (struct pipeline_io*)ht_create(tt,
(size_t)(hashsize), pmat?(hash_table*)pmat->bat[i]->pl_io:NULL);
}
if (i < mat->nr) {
mat_destroy(mat);
BBPunfix(matb->batCacheid);
throw(MAL, "mat.new", SQLSTATE(HY013) MAL_MALLOC_FAIL);
}
- matb->pl_io = (Sink*)mat;
+ matb->pl_io = (struct pipeline_io*)mat;
*mid = matb->batCacheid;
BBPkeepref(matb);
return MAL_SUCCEED;
@@ -177,7 +177,7 @@ PARTnew(Client cntxt, MalBlkPtr mb, MalS
throw(MAL, "part.new", SQLSTATE(HY013) MAL_MALLOC_FAIL);
}
- partb->pl_io = (Sink*)part;
+ partb->pl_io = (struct pipeline_io*)part;
*pid = partb->batCacheid;
BBPkeepref(partb);
return MAL_SUCCEED;
diff --git a/monetdb5/modules/mal/pp_mat.h b/monetdb5/modules/mal/pp_mat.h
--- a/monetdb5/modules/mal/pp_mat.h
+++ b/monetdb5/modules/mal/pp_mat.h
@@ -12,7 +12,7 @@
#define _PP_MAT_H_
typedef struct mat_t {
- Sink s;
+ struct pipeline_io s;
int nr;
int nr_parts;
int *part;
diff --git a/monetdb5/modules/mal/pp_slicer.c b/monetdb5/modules/mal/pp_slicer.c
--- a/monetdb5/modules/mal/pp_slicer.c
+++ b/monetdb5/modules/mal/pp_slicer.c
@@ -17,7 +17,7 @@
#include "pipeline.h"
typedef struct topn_t {
- Sink s;
+ struct pipeline_io s;
lng start;
lng end;
} topn_t;
@@ -66,7 +66,7 @@ LALGsubslice(Client ctx, bat *gid, bat *
msg = createException(SQL, "algebra.subslice",
SQLSTATE(HY013) MAL_MALLOC_FAIL);
goto error;
}
- t->pl_io = (Sink*)n;
+ t->pl_io = (struct pipeline_io*)n;
t->tprivate_bat = 1;
} else {
if ((t = BATdescriptor(*tid)) == NULL) {
@@ -85,7 +85,7 @@ LALGsubslice(Client ctx, bat *gid, bat *
msg = createException(SQL, "algebra.subslice",
SQLSTATE(HY013) MAL_MALLOC_FAIL);
goto error;
}
- t->pl_io = (Sink*)n;
+ t->pl_io = (struct pipeline_io*)n;
}
assert(n && n->s.type == TOPN_SINK);
diff --git a/monetdb5/modules/mal/pp_sort.c b/monetdb5/modules/mal/pp_sort.c
--- a/monetdb5/modules/mal/pp_sort.c
+++ b/monetdb5/modules/mal/pp_sort.c
@@ -687,7 +687,7 @@ typedef struct part_t {
} part_t;
typedef struct sop_t {
- Sink s;
+ struct pipeline_io s;
int nr;
int nr_workers;
MT_Lock l;
@@ -756,7 +756,7 @@ SOPnew(Client cntxt, MalBlkPtr mb, MalSt
GDKfree(q);
throw(MAL, "sop.new", SQLSTATE(HY013) MAL_MALLOC_FAIL);
}
- qb->pl_io = (Sink*)q;
+ qb->pl_io = (struct pipeline_io*)q;
*sop = qb->batCacheid;
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]
