Changeset: 5b999aa09e6f for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/5b999aa09e6f
Modified Files:
clients/Tests/MAL-signatures-hge.test
monetdb5/modules/mal/pp_hash.c
monetdb5/modules/mal/pp_hash.h
sql/backends/monet5/rel_bin.c
sql/backends/monet5/rel_physical.c
sql/backends/monet5/rel_pphash.c
sql/backends/monet5/sql_pp_statement.c
sql/backends/monet5/sql_pp_statement.h
sql/server/rel_basetable.c
sql/server/rel_optimize_sel.c
sql/server/rel_rel.c
sql/server/rel_rel.h
sql/test/bincopy/Tests/very_large_date_file.SQL.py
sql/test/bincopy/Tests/very_large_file.SQL.py
Branch: default
Log Message:
implemented fetch joins in pipeline mode
diffs (truncated from 883 to 300 lines):
diff --git a/clients/Tests/MAL-signatures-hge.test
b/clients/Tests/MAL-signatures-hge.test
--- a/clients/Tests/MAL-signatures-hge.test
+++ b/clients/Tests/MAL-signatures-hge.test
@@ -9125,6 +9125,9 @@ dict join pattern dict.join(X_0:bat[:any
dict renumber pattern dict.renumber(X_0:bat[:any_1],
X_1:bat[:any_1]):bat[:any_1] DICTrenumber renumber offsets
dict select pattern dict.select(X_0:bat[:any], X_1:bat[:oid],
X_2:bat[:any_1], X_3:any_1, X_4:any_1, X_5:bit, X_6:bit, X_7:bit,
X_8:bit):bat[:oid] DICTselect value - range select on a dictionary
dict thetaselect pattern dict.thetaselect(X_0:bat[:any], X_1:bat[:oid],
X_2:bat[:any_1], X_3:any_1, X_4:str):bat[:oid] DICTthetaselect thetaselect on
a dictionary
+fhash build pattern fhash.build(X_0:bat[:any_1], X_1:bit) (X_2:bat[:oid],
X_3:bat[:any_1]) FHASHbuild Add `keys` to the `hashtable`. If
`need_has_nil`, denote if `keys` contains NULL. Returns corresponding `gids` of
the `keys` and the updated `hashtable`
+fhash new pattern fhash.new(X_0:any_1, X_1:lng, X_2:lng):bat[:any_1]
FHASHnew (empty)
+fhash probe command fhash.probe(X_0:bat[:any_1], X_1:bat[:any_1], X_2:bit,
X_3:bit) (X_4:bat[:oid], X_5:bat[:oid]) FHASHprobe Probe the `key` in the
hash table. For a matched key, return its OID in the 'key' column and the slot
ID in the hash table
fits attach unsafe pattern fits.attach(X_0:str):void FITSattach
Open a FITS file and return catalog of the table HDUs
fits export pattern fits.export(X_0:str):void FITSexportTable Export
a table to a FITS file
fits fitstest command fits.fitstest(X_0:str):int FITStest
Returns the type of first extension in the FITS file filename
@@ -10112,10 +10115,6 @@ oahash hash pattern oahash.hash(X_0:any_
oahash hashmark_init command oahash.hashmark_init(X_0:bat[:any_1],
X_1:bat[:any_2]):bat[:bit] OAHASHhashmark_init (empty)
oahash mprobe command oahash.mprobe(X_0:bat[:any_1], X_1:bat[:any_1],
X_2:bit, X_3:bit) (X_4:bat[:oid], X_5:bat[:oid], X_6:bat[:bit]) OAHASHmprobe
Probe the `key`-s in the hash table. For a matched key, return its OID in the
'key' column and the slot ID in the hash table
oahash mprobe command oahash.mprobe(X_0:bat[:any_1], X_1:bat[:any_1],
X_2:bat[:lng], X_3:bit, X_4:bit) (X_5:bat[:oid], X_6:bat[:oid], X_7:bat[:bit])
OAHASHmprobe_single Probe the `key`-s in the hash table. For a matched key,
return its OID in the 'key' column and the slot ID in the hash table
-oahash new pattern oahash.new(X_0:any_1, X_1:int):bat[:any_1]
OAHASHnew (empty)
-oahash new pattern oahash.new(X_0:any_1, X_1:int,
X_2:bat[:any_2]):bat[:any_1] OAHASHnew (empty)
-oahash new pattern oahash.new(X_0:any_1, X_1:int, X_2:bat[:any_2],
X_3:int):bat[:any_1] OAHASHnew (empty)
-oahash new pattern oahash.new(X_0:any_1, X_1:int, X_2:int):bat[:any_1]
OAHASHnew (empty)
oahash new pattern oahash.new(X_0:any_1, X_1:lng):bat[:any_1]
OAHASHnew (empty)
oahash new pattern oahash.new(X_0:any_1, X_1:lng,
X_2:bat[:any_2]):bat[:any_1] OAHASHnew (empty)
oahash new pattern oahash.new(X_0:any_1, X_1:lng, X_2:bat[:any_2],
X_3:int):bat[:any_1] OAHASHnew (empty)
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
@@ -494,19 +494,11 @@ OAHASHnew(Client cntxt, MalBlkPtr m, Mal
bat *res = getArgReference_bat(s, p, 0);
int tt = getArgType(m, p, 1);
- int tt2 = getArgType(m, p, 2);
- lng size = 0;
+ lng size = *getArgReference_lng(s, p, 2);
hash_table *parent = NULL;
BAT *pht = NULL;
int vkey = 0;
- if (tt2 == TYPE_int) {
- assert(0);
- size = (lng) *getArgReference_int(s, p, 2);
- } else {
- assert(tt2 == TYPE_lng);
- size = *getArgReference_lng(s, p, 2);
- }
/* multiply with the magic estimation while avoiding overflow */
size = size > ((dbl)INT64_MAX / 1.2 / 2.1)? INT64_MAX : (lng)(size *
1.2 * 2.1);
@@ -540,6 +532,39 @@ OAHASHnew(Client cntxt, MalBlkPtr m, Mal
}
static str
+FHASHnew(Client cntxt, MalBlkPtr m, MalStkPtr s, InstrPtr p)
+{
+ (void)cntxt;
+
+ bat *res = getArgReference_bat(s, p, 0);
+ int tt = getArgType(m, p, 1);
+ lng size = *getArgReference_lng(s, p, 2);
+ lng max = *getArgReference_lng(s, p, 3);
+ hash_table *parent = NULL;
+ BAT *pht = NULL;
+
+ /* multiply with the magic estimation while avoiding overflow */
+ size = size > ((dbl)INT64_MAX / 1.2 / 2.1)? INT64_MAX : (lng)(size *
1.2 * 2.1);
+
+ BAT *b = COLnew(0, tt, 0, TRANSIENT);
+ if (b == NULL) {
+ BBPreclaim(pht);
+ return createException(MAL, "fhash.new", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
+ }
+ hash_table *ht = ht_create(tt, (size_t)max, parent, 0);
+ ht->perfect = true;
+ b->pl_io = (struct pipeline_io*)ht;
+ BBPreclaim(pht);
+ if (b->pl_io == NULL) {
+ BBPunfix(b->batCacheid);
+ return createException(MAL, "fhash.new", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
+ }
+ *res = b->batCacheid;
+ BBPkeepref(b);
+ return MAL_SUCCEED;
+}
+
+static str
OAHASHhashmark_init(Client ctx, bat *res, const bat *ht_sink, const bat
*payload)
{
(void)ctx;
@@ -1471,6 +1496,124 @@ error:
return err;
}
+#define vfetch() \
+ do { \
+ slot = ATOMIC_ADD_GID(&h->last, cnt); \
+ oid *hgids = (oid*)h->gids; \
+ struct canditer ci; \
+ canditer_init(&ci, NULL, b); \
+ cnt = ci.ncand; \
+ \
+ TIMEOUT_LOOP_IDX_DECL(i, cnt, qry_ctx) { \
+ oid bpi = canditer_next(&ci); \
+ assert(bpi != oid_nil); \
+ gid g = ++slot; \
+ hgids[bpi] = g; \
+ gp[i] = g-1; \
+ } \
+ } while (0)
+
+#define fetch() \
+ do { \
+ slot = ATOMIC_ADD_GID(&h->last, cnt); \
+ oid *bp = Tloc(b, 0); \
+ oid *hgids = (oid*)h->gids; \
+ \
+ TIMEOUT_LOOP_IDX_DECL(i, cnt, qry_ctx) { \
+ oid bpi = bp[i]; \
+ assert(bpi != oid_nil); \
+ gid g = ++slot; \
+ hgids[bpi] = g; \
+ gp[i] = g-1; \
+ } \
+ } while (0)
+
+static str
+FHASHbuild(Client ctx, MalBlkPtr m, MalStkPtr s, InstrPtr p)
+{
+ (void)ctx;
+ (void)m;
+
+ bat *gids = getArgReference_bat(s, p, 0);
+ bat *hashtable = getArgReference_bat(s, p, 1);
+ bat *keys = getArgReference_bat(s, p, 2);
+ bit need_has_nil = *getArgReference_bit(s, p, 3);
+ assert(need_has_nil == FALSE);
+ (void)need_has_nil;
+
+ Pipeline *pl = pipeline_get_thread_private_pipeline();
+ str err = NULL;
+ BAT *u = NULL, *b = NULL, *g = NULL;
+
+ /* we only work with shared hashtable in the build phase */
+ assert(*hashtable && !is_bat_nil(*hashtable));
+
+ b = BATdescriptor(*keys);
+ u = BATdescriptor(*hashtable);
+ if (!b || !u) {
+ err = createException(MAL, "fhash.build", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
+ goto error;
+ }
+ hash_table *h = (hash_table*)u->pl_io;
+ assert(h && h->pl_io.type == PIPELINE_IO_HASH_TABLE);
+
+ BUN cnt = BATcount(b);
+ g = COLnew(b->hseqbase, TYPE_oid, cnt, TRANSIENT);
+ if (g == NULL) {
+ err = createException(MAL, "fhash.build", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
+ goto error;
+ }
+
+ if (cnt) {
+ int tt = b->ttype;
+ gid *gp = Tloc(g, 0);
+
+ h->empty = false;
+ gid slot = 0;
+ QryCtx *qry_ctx = MT_thread_get_qry_ctx();
+ qry_ctx = qry_ctx ? qry_ctx : &(QryCtx) {.endtime = 0};
+
+ assert(tt == TYPE_void || tt == TYPE_oid); /* for now only oids
*/
+
+ if (tt == TYPE_void) {
+ vfetch();
+ } else if (tt == TYPE_oid) {
+ if (BATtdense(b))
+ vfetch();
+ else
+ fetch();
+ }
+ if (!err)
+ TIMEOUT_CHECK(qry_ctx, err = createException(SQL,
"fhash.build", RUNTIME_QRY_TIMEOUT));
+ }
+ if (err || pl->p->status) {
+ if (!err)
+ err = createException(MAL, "fhash.build", "pipeline
execution error");
+ goto error;
+ }
+ BATsetcount(g, cnt);
+ pipeline_lock2(g);
+ BATnegateprops(g);
+ pipeline_unlock2(g);
+ gid last = ATOMIC_GET_GID(&h->last);
+ g->tmaxval = last;
+ g->tkey = FALSE;
+ *gids = g->batCacheid;
+ //skip propcheck
+ //BBPkeepref(u);
+ BBPretain(u->batCacheid);
+ BBPunfix(u->batCacheid);
+ BBPkeepref(g);
+
+ BBPunfix(b->batCacheid);
+ return MAL_SUCCEED;
+error:
+ BBPreclaim(u);
+ BBPreclaim(b);
+ BBPreclaim(g);
+ return err;
+}
+
static str
OAHASHadd_freq(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
{
@@ -1864,6 +2007,120 @@ error:
return err;
}
+#define BATvlookup() \
+ do { \
+ struct canditer ci; \
+ canditer_init(&ci, NULL, k); \
+ keycnt = ci.ncand; \
+ oid *gids = (oid*)ht->gids; \
+ \
+ TIMEOUT_LOOP_IDX_DECL(i, keycnt, qry_ctx) { \
+ oid ky = canditer_next(&ci); \
+ assert(ky != oid_nil); \
+ gid k = ky < ht->size ? gids[ky] : 0; \
+ if (k) { \
+ oid_mtd[mtdcnt] = off+i; \
+ slt[mtdcnt] = (oid)(k - 1); \
+ mtdcnt++; \
+ } \
+ } \
+ } while (0)
+
+#define BATlookup() \
+ do { \
+ oid *gids = (oid*)ht->gids; \
+ oid *kp = Tloc(k, 0); \
+ \
+ TIMEOUT_LOOP_IDX_DECL(i, keycnt, qry_ctx) { \
+ oid kpi = kp[i]; \
+ /*assert(kpi != oid_nil);*/ \
+ gid k = kpi < ht->size ? gids[kpi] : 0; \
+ if (k) { \
+ oid_mtd[mtdcnt] = off+i; \
+ slt[mtdcnt] = (oid)(k - 1); \
+ mtdcnt++; \
+ } \
+ } \
+ } while (0)
+
+static str
+FHASHprobe(Client ctx, bat *PRB_oid, bat *HSH_slotid, const bat *PRB_key,
const bat *HSH_ht, const bit *single, const bit *semantics)
+{
+ (void)ctx;
+ BAT *o = NULL, *s = NULL, *k = NULL, *t = NULL;
+ BUN keycnt, mtdcnt = 0;
+ str err = NULL;
+
+ k = BATdescriptor(*PRB_key);
+ t = BATdescriptor(*HSH_ht);
+ if (!k || !t) {
+ err = createException(SQL, "oahash.probe", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
+ goto error;
+ }
+ assert(!*single && !*semantics);
+ (void)single;
+ (void)semantics;
+
+ keycnt = BATcount(k);
+ o = COLnew(0, TYPE_oid, keycnt, TRANSIENT);
+ s = COLnew(0, TYPE_oid, keycnt, TRANSIENT);
+ if (!o || !s) {
+ err = createException(SQL, "oahash.probe", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
+ goto error;
+ }
+
+ if (keycnt) {
+ if (t->pl_io->error) {
+ err = t->pl_io->error;
+ goto error;
+ }
+
+ hash_table *ht = (hash_table*)t->pl_io;
+ int tt = k->ttype;
+ QryCtx *qry_ctx = MT_thread_get_qry_ctx();
+ qry_ctx = qry_ctx ? qry_ctx : &(QryCtx) {.endtime = 0};
+
+ oid *oid_mtd = Tloc(o, 0);
+ oid *slt = Tloc(s, 0);
+
+ BUN off = k->hseqbase;
+ if(tt == TYPE_void) {
+ BATvlookup();
+ } else if(tt == TYPE_oid) {
+ BATlookup();
+ } else {
+ err = createException(MAL, "oahash.probe",
SQLSTATE(HY000) TYPE_NOT_SUPPORTED);
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]
