Changeset: 37b3bdec5569 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/37b3bdec5569
Modified Files:
clients/Tests/MAL-signatures-hge.test
clients/Tests/MAL-signatures.test
gdk/gdk_utils.c
monetdb5/modules/mal/pp_hash.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
Branch: default
Log Message:
added implementation for semijoin with hash on the other end.
Uses the rel2bin_*fullouter code generation path.
diffs (truncated from 365 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
@@ -9681,6 +9681,7 @@ iaggr sum command iaggr.sum(X_0:bat[:flt
iaggr sum command iaggr.sum(X_0:bat[:dbl], X_1:bit) (X_2:dbl, X_3:dbl,
X_4:lng) IALGfsum_skipnil Return the Kahan/Neumaier summation or nil.
iaggr sum command iaggr.sum(X_0:bat[:flt], X_1:bit) (X_2:dbl, X_3:dbl,
X_4:lng) IALGfsum_skipnil Return the Kahan/Neumaier summation or nil.
iaggr sum command iaggr.sum(X_0:bat[:flt], X_1:bit) (X_2:flt, X_3:flt,
X_4:lng) IALGfsum_skipnil_flt Return the Kahan/Neumaier summation or nil.
+ialgebra matched command ialgebra.matched(X_0:bat[:bit],
X_1:bat[:oid]):bat[:oid] OAHASHmatched Mark matched rows, only return
rows matched rows on first match.
ialgebra project pattern ialgebra.project(X_0:bat[:oid],
X_1:any_1):bat[:any_1] LALGconstant Project a single value
ialgebra projection command ialgebra.projection(X_0:bat[:oid],
X_1:bat[:any_1]):bat[:any_1] LALGprojection Project.
ialgebra unique command ialgebra.unique(X_0:bat[:any_1], X_1:bat[:oid],
X_2:bat[:oid], X_3:bit) (X_4:bat[:oid], X_5:bat[:any_1])
LALGgroup_unique Unique rows per group.
diff --git a/clients/Tests/MAL-signatures.test
b/clients/Tests/MAL-signatures.test
--- a/clients/Tests/MAL-signatures.test
+++ b/clients/Tests/MAL-signatures.test
@@ -7322,6 +7322,7 @@ iaggr sum command iaggr.sum(X_0:bat[:flt
iaggr sum command iaggr.sum(X_0:bat[:dbl], X_1:bit) (X_2:dbl, X_3:dbl,
X_4:lng) IALGfsum_skipnil Return the Kahan/Neumaier summation or nil.
iaggr sum command iaggr.sum(X_0:bat[:flt], X_1:bit) (X_2:dbl, X_3:dbl,
X_4:lng) IALGfsum_skipnil Return the Kahan/Neumaier summation or nil.
iaggr sum command iaggr.sum(X_0:bat[:flt], X_1:bit) (X_2:flt, X_3:flt,
X_4:lng) IALGfsum_skipnil_flt Return the Kahan/Neumaier summation or nil.
+ialgebra matched command ialgebra.matched(X_0:bat[:bit],
X_1:bat[:oid]):bat[:oid] OAHASHmatched Mark matched rows, only return
rows matched rows on first match.
ialgebra project pattern ialgebra.project(X_0:bat[:oid],
X_1:any_1):bat[:any_1] LALGconstant Project a single value
ialgebra projection command ialgebra.projection(X_0:bat[:oid],
X_1:bat[:any_1]):bat[:any_1] LALGprojection Project.
ialgebra unique command ialgebra.unique(X_0:bat[:any_1], X_1:bat[:oid],
X_2:bat[:oid], X_3:bit) (X_4:bat[:oid], X_5:bat[:any_1])
LALGgroup_unique Unique rows per group.
diff --git a/gdk/gdk_utils.c b/gdk/gdk_utils.c
--- a/gdk/gdk_utils.c
+++ b/gdk/gdk_utils.c
@@ -1189,7 +1189,7 @@ GDKinit(opt *set, int setlen, bool embed
if (GDKnr_threads > THREADS)
GDKnr_threads = THREADS;
#if defined(HAVE_SYSCONF) && defined(_SC_LEVEL3_CACHE_SIZE)
- GDKL3_size = (size_t)sysconf(_SC_LEVEL3_CACHE_SIZE);
+ GDKL3_size = (size_t)sysconf(_SC_LEVEL3_CACHE_SIZE); /* error (-1),
will lead too huge L3_size */
#else
GDKL3_size = 16*1024*1024;
#endif
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
@@ -645,6 +645,66 @@ error:
}
static str
+OAHASHmatched(Client ctx, bat *matched_first, const bat *matched, const bat
*rows)
+{
+ (void)ctx;
+ BAT *m = BATdescriptor(*matched);
+ if (!m)
+ return createException(MAL, "oahash.matched", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
+ BAT *r = BATdescriptor(*rows);
+ if (!r) {
+ BBPreclaim(m);
+ return createException(MAL, "oahash.matched", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
+ }
+ BUN cnt = BATcount(r), k = 0;
+ BAT *mo = COLnew(0, TYPE_oid, cnt, TRANSIENT);
+ if (mo == NULL) {
+ BBPreclaim(m);
+ BBPreclaim(r);
+ return createException(MAL, "oahash.matched", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
+ }
+ QryCtx *qry_ctx = MT_thread_get_qry_ctx();
+ qry_ctx = qry_ctx ? qry_ctx : &(QryCtx) {.endtime = 0};
+ bit *mrk = Tloc(m, 0);
+ oid *row = Tloc(r, 0);
+ oid *matched_only = Tloc(mo, 0);
+
+
+ MT_lock_set(&m->theaplock);
+ if (r->ttype == TYPE_oid) {
+ TIMEOUT_LOOP_IDX_DECL(i, cnt, qry_ctx) {
+ oid j = row[i];
+ if (mrk[j] == FALSE) {
+ mrk[j] = TRUE;
+ matched_only[k++] = i;
+ }
+ }
+ } else { /* handle voids */
+ BUN off = r->tseqbase;
+ TIMEOUT_LOOP_IDX_DECL(i, cnt, qry_ctx) {
+ oid j = off + i;
+ if (mrk[j] == FALSE) {
+ mrk[j] = TRUE;
+ matched_only[k++] = i;
+ }
+ }
+ }
+ MT_lock_unset(&m->theaplock);
+ BATsetcount(mo, k);
+ mo->tnonil = TRUE;
+ mo->tkey = TRUE;
+ mo->tsorted = TRUE;
+ mo->trevsorted = FALSE;
+ str err = MAL_SUCCEED;
+ if (!err)
+ TIMEOUT_CHECK(qry_ctx, err = createException(SQL,
"oahash.build", RUNTIME_QRY_TIMEOUT));
+ BBPreclaim(m);
+ BBPreclaim(r);
+ BBPretain(*matched_first = mo->batCacheid);
+ return err;
+}
+
+static str
UHASHext(Client cntxt, MalBlkPtr m, MalStkPtr s, InstrPtr p)
{
(void)cntxt;
@@ -3738,6 +3798,8 @@ static mel_func oa_hash_init_funcs[] = {
pattern("oahash", "hash", OAHASHhash, false, "Compute hash.", args(1,2,
arg("hash", lng), argany("in",1))),
pattern("batoahash", "hash", OAHASHhash, false, "Compute hash.", args(1,2,
batarg("hash", lng), batargany("in",1))),
+ command("ialgebra", "matched", OAHASHmatched, false, "Mark matched rows, only
return rows matched rows on first match.", args(1,3, batarg("matched_once",
oid), batarg("matched", bit), batarg("rows", oid))),
+
{ .imp=NULL }
};
#include "mal_import.h"
diff --git a/sql/backends/monet5/rel_physical.c
b/sql/backends/monet5/rel_physical.c
--- a/sql/backends/monet5/rel_physical.c
+++ b/sql/backends/monet5/rel_physical.c
@@ -871,9 +871,18 @@ rel_pipeline(visitor *v, sql_rel *rel, b
bool need_all = false;
bool cross = list_empty(eq_exps);
- rel->oahash = 2;
+ sql_rel *l = rel->l, *r = rel->r;
+ if (!cross && rel_getcount(v->sql, l) < rel_getcount(v->sql, r)
&& rel->op == op_semi)
+ rel->oahash = 1;
+ else
+ rel->oahash = 2;
sql_rel *rel_hsh = rel->r, *rel_prb = rel->l;
+ if (rel->oahash == 1) {
+ rel_hsh = rel->l;
+ rel_prb = rel->r;
+ }
+
list *found_exps_cmp_hsh = NULL;
/* get full projection list from parent */
@@ -888,20 +897,25 @@ rel_pipeline(visitor *v, sql_rel *rel, b
} else {
if (rel_is_ref(rel_hsh))
need_all = true;
- rel->r = rel_hsh = rel_buildhash(v, rel_hsh, NULL,
list_empty(eq_exps));
+ rel_hsh = rel_buildhash(v, rel_hsh, NULL,
list_empty(eq_exps));
rel_hsh->flag = (int)(needs_payload?op_join:op_semi);
}
- rel->l = rel_prb = rel_probehash(v, rel_prb, NULL);
+ rel_prb = rel_probehash(v, rel_prb, NULL);
+ rel->r = (rel->oahash==1) ? rel_prb : rel_hsh;
+ rel->l = (rel->oahash==1) ? rel_hsh : rel_prb;
rel_hsh->attr = cross?NULL:exps_cmp_hsh;
rel_prb->attr = exps_cmp_prb;
list *exps_hsh = NULL, *exps_prb = rel_prb->exps =
sa_list(v->sql->sa);
- if (needs_payload)
+ if (needs_payload || rel->oahash == 1)
exps_hsh = sa_list(v->sql->sa);
find_payload_exps(v->sql, &exps_hsh, &exps_prb, p, rel_hsh,
rel_prb);
+ if (rel->oahash == 1)
+ rel_hsh->exps = exps_hsh;
if (need_all && !is_base(rel_hsh->op))
rel_hsh->exps =
!rel_hsh->exps?rel_hsh->attr:list_distinct(list_merge(rel_hsh->exps,
rel_hsh->attr, NULL), (fcmp) exp_equal, NULL);
+
if (needs_payload) { /* TODO improve ie only keep attributes
needed for 'other' exps */
if (!is_base(rel_hsh->op)) {
sql_rel *l = rel_hsh->l;
diff --git a/sql/backends/monet5/rel_pphash.c b/sql/backends/monet5/rel_pphash.c
--- a/sql/backends/monet5/rel_pphash.c
+++ b/sql/backends/monet5/rel_pphash.c
@@ -1160,7 +1160,7 @@ rel2bin_oahash_fullouterjoin(backend *be
{
mvc *sql = be->mvc;
sql_subtype *tpe_oid = sql_fetch_localtype(TYPE_oid);
- bool hf = false;
+ bool hf = false, outer = is_outerjoin(rel->op), anti_join = (rel->op ==
op_anti);
sql_rel *rel_hsh = rel->oahash == 1? rel->l: rel->r;
/* start new parallel block after join. NB get_need_pipeline has side
effect! */
@@ -1197,7 +1197,8 @@ rel2bin_oahash_fullouterjoin(backend *be
* immediately add this concat block to the p_source at the beginning
of
* code generation for this block.
*/
- (void)stmt_concat(be, be->source, 2);
+ if (outer || anti_join)
+ (void)stmt_concat(be, be->source, 2);
/* create all results variables, from rel->l to rel->r */
list *vars = sa_list(sql->sa);
@@ -1208,7 +1209,7 @@ rel2bin_oahash_fullouterjoin(backend *be
append(vars, s);
}
}
- if (!list_empty(((sql_rel*)rel->r)->exps)) {
+ if (outer && !list_empty(((sql_rel*)rel->r)->exps)) {
for (node *n = ((sql_rel*)rel->r)->exps->h; n; n = n->next) {
sql_subtype *st = exp_subtype(n->data);
stmt *s = stmt_bat_declare(be, st);
@@ -1217,7 +1218,9 @@ rel2bin_oahash_fullouterjoin(backend *be
}
/*** first concat_block: the join ***/
- int b = stmt_concat_barrier(be, be->source, 0, 0);
+ int b = 0;
+ if (outer || anti_join)
+ b = stmt_concat_barrier(be, be->source, 0, 0);
// existing leftjoin code, extended to mark the matched payloads
stmt *sub = NULL, *probe_sub = NULL;
@@ -1234,16 +1237,34 @@ rel2bin_oahash_fullouterjoin(backend *be
} else {
sub = rel2bin_oahash_equi_join(be, rel, refs, jexps,
&probed_ids, &probe_sub, NULL /* nulls */, &m /* prb_mrk */, &hsh_mrk,
&probe_side, &hash_side, !list_empty(sexps) /* has_outerselect */);
}
- if (!list_empty(sexps)) {
- stmt *sel = rel2bin_oahash_outerselect(be, sub, sexps, rel,
probed_ids, hash_ids, &m, list_empty(jexps), false);
- if (sel == NULL) return NULL;
+ if (!outer || !list_empty(sexps)) {
+ stmt *sel = NULL;
+ if (outer)
+ sel = rel2bin_oahash_outerselect(be, sub, sexps, rel,
probed_ids, hash_ids, &m, list_empty(jexps), false);
+ else if (!list_empty(sexps))
+ sel = rel2bin_oahash_select(be, sub, sexps, rel, hf);
+ if (sel == NULL && (outer || !list_empty(sexps)))
+ return NULL;
+
+ if (!outer) {
+ if (sel)
+ hsh_mrk = stmt_project(be, sel, hsh_mrk);
+ stmt *matched = stmt_mark_matched(be, hp_mrk, hsh_mrk);
+ if (!sel)
+ sel = matched;
+ else
+ sel = stmt_project(be, matched, sel);
+ //if (anti_join)
+ //sel = /* empty */
+ }
list *lp = sa_list(sql->sa);
for (node *n = probe_side->h; n; n = n->next) {
stmt *c = stmt_project(be, sel, n->data);
const char *cname = column_name(sql->sa, c);
const char *tname = table_name(sql->sa, c);
int label = c->label;
- c = stmt_alias(be, c, label, tname, cname);
+ if (label)
+ c = stmt_alias(be, c, label, tname, cname);
append(lp, c);
}
list *lh = sa_list(sql->sa);
@@ -1252,31 +1273,40 @@ rel2bin_oahash_fullouterjoin(backend *be
const char *cname = column_name(sql->sa, c);
const char *tname = table_name(sql->sa, c);
int label = c->label;
- c = sql_Nop_(be, "ifthenelse", m, c, stmt_atom(be,
atom_general(sql->sa, tail_type(c), NULL, 0)), NULL);
- c = stmt_alias(be, c, label, tname, cname);
+ if (outer)
+ c = sql_Nop_(be, "ifthenelse", m, c,
stmt_atom(be, atom_general(sql->sa, tail_type(c), NULL, 0)), NULL);
+ if (label)
+ c = stmt_alias(be, c, label, tname, cname);
append(lh, c);
}
sub = rel->oahash == 1?
stmt_list(be, list_merge(lh, lp, NULL)):
stmt_list(be, list_merge(lp, lh, NULL));
- /* mark the matched payload */
- assert(sel && m);
- /* X_nn:bat[:oid] := algebra.thetaselect(<m>, nil:bat[:any],
true:bit, "=":str);*/
- stmt *ss = stmt_thetaselect(be, m, NULL, stmt_bool(be, 1),
"==", tpe_oid);
- if (!ss) return NULL;
- stmt *s_oids = stmt_project(be, ss, sel);
- if (!s_oids) return NULL;
- hsh_mrk = stmt_project(be, s_oids, hsh_mrk);
- } else {
+ if (outer) {
+ /* mark the matched payload */
+ assert(sel && m);
+ /* X_nn:bat[:oid] := algebra.thetaselect(<m>,
nil:bat[:any], true:bit, "=":str);*/
+ stmt *ss = stmt_thetaselect(be, m, NULL, stmt_bool(be,
1), "==", tpe_oid);
+ if (!ss)
+ return NULL;
+ stmt *s_oids = stmt_project(be, ss, sel);
+ if (!s_oids)
+ return NULL;
+ hsh_mrk = stmt_project(be, s_oids, hsh_mrk);
+ }
+ } else if (outer) {
atom *a_null = atom_general(sql->sa, tpe_oid, NULL, 0);
- stmt *stmt_null = stmt_atom(be, a_null); /* we just need the
count of unmatched */
+ stmt *stmt_null = stmt_atom(be, a_null);
/* X_nn:bat[:oid] := algebra.thetaselect(<m>, nil:bat[:any],
true:bit, "=":str);*/
stmt *sel = stmt_thetaselect(be, hsh_mrk, NULL, stmt_null,
"ne", tpe_oid);
hsh_mrk = stmt_project(be, sel, hsh_mrk);
}
- if (!hsh_mrk) return NULL;
- hp_mrk = stmt_algebra_project(be, hp_mrk, hsh_mrk, stmt_bool(be,1),
projectRef);
+ if (outer) {
+ if (!hsh_mrk)
+ return NULL;
+ hp_mrk = stmt_algebra_project(be, hp_mrk, hsh_mrk,
stmt_bool(be,1), projectRef);
+ }
// END existing code
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]
