Changeset: 55b38da1cf7e for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/55b38da1cf7e
Modified Files:
sql/backends/monet5/for.c
sql/backends/monet5/for.h
sql/storage/bat/bat_storage.c
sql/storage/bat/bat_storage.h
sql/storage/sql_storage.h
Branch: Jan2022
Log Message:
also allow updates (appends) on ST_FOR
diffs (truncated from 495 to 300 lines):
diff --git a/sql/backends/monet5/for.c b/sql/backends/monet5/for.c
--- a/sql/backends/monet5/for.c
+++ b/sql/backends/monet5/for.c
@@ -29,6 +29,59 @@ BATnegateprops(BAT *b)
b->tnokey[1] = 0;
}
+BAT *
+FORdecompress_(BAT *o, lng minval, int type, role_t role)
+{
+ BAT *b = COLnew(o->hseqbase, type, BATcount(o), role);
+
+ if (!b)
+ return NULL;
+ BUN cnt = BATcount(o);
+#ifdef HAVE_HGE
+ if (type == TYPE_hge) {
+ if (o->ttype == TYPE_bte) {
+ hge *ov = Tloc(b, 0);
+ bte *iv = Tloc(o, 0);
+ for(BUN i = 0; i<cnt; i++)
+ ov[i] = minval + iv[i];
+ } else {
+ hge *ov = Tloc(b, 0);
+ sht *iv = Tloc(o, 0);
+ for(BUN i = 0; i<cnt; i++)
+ ov[i] = minval + iv[i];
+ }
+ } else
+#endif
+ if (type == TYPE_lng) {
+ if (o->ttype == TYPE_bte) {
+ lng *ov = Tloc(b, 0);
+ bte *iv = Tloc(o, 0);
+ for(BUN i = 0; i<cnt; i++)
+ ov[i] = minval + iv[i];
+ } else {
+ lng *ov = Tloc(b, 0);
+ sht *iv = Tloc(o, 0);
+ for(BUN i = 0; i<cnt; i++)
+ ov[i] = minval + iv[i];
+ }
+ } else if (type == TYPE_int) {
+ if (o->ttype == TYPE_bte) {
+ int *ov = Tloc(b, 0);
+ bte *iv = Tloc(o, 0);
+ for(BUN i = 0; i<cnt; i++)
+ ov[i] = minval + iv[i];
+ } else {
+ int *ov = Tloc(b, 0);
+ sht *iv = Tloc(o, 0);
+ for(BUN i = 0; i<cnt; i++)
+ ov[i] = minval + iv[i];
+ }
+ }
+ BATsetcount(b, cnt);
+ BATnegateprops(b);
+ return b;
+}
+
str
FORdecompress(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
{
@@ -54,37 +107,46 @@ FORdecompress(Client cntxt, MalBlkPtr mb
throw(SQL, "for.decompress", SQLSTATE(3F000) "for decompress:
invalid type");
}
- BUN cnt = BATcount(o);
- if (tt == TYPE_lng) {
- lng minval = *getArgReference_lng(stk, pci, 2);
+ lng minval = *getArgReference_lng(stk, pci, 2);
- b = COLnew(o->hseqbase, TYPE_lng, cnt, PERSISTENT);
- if (!b) {
- bat_destroy(o);
- throw(SQL, "for.decompress", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
- }
- if (o->ttype == TYPE_bte) {
- lng *ov = Tloc(b, 0);
- bte *iv = Tloc(o, 0);
- for(BUN i = 0; i<cnt; i++)
- ov[i] = minval + iv[i];
- } else {
- lng *ov = Tloc(b, 0);
- sht *iv = Tloc(o, 0);
- for(BUN i = 0; i<cnt; i++)
- ov[i] = minval + iv[i];
- }
- } else {
+ b = FORdecompress_(o, minval, tt, TRANSIENT);
+ if (!b) {
bat_destroy(o);
- throw(SQL, "for.decompress", SQLSTATE(3F000) "offset type %s
not yet implemented", ATOMname(tt));
+ throw(SQL, "for.decompress", SQLSTATE(HY013) MAL_MALLOC_FAIL);
}
bat_destroy(o);
- BATsetcount(b, cnt);
- BATnegateprops(b);
BBPkeepref(*r = b->batCacheid);
return MAL_SUCCEED;
}
+static BAT *
+FORcompress_(BAT *b, lng min_val, lng max_val, role_t role)
+{
+ BAT *o;
+ BUN cnt = BATcount(b);
+
+ if ((max_val-min_val) < GDK_bte_max/2) {
+ o = COLnew(b->hseqbase, TYPE_bte, cnt, role);
+ if (!o)
+ return NULL;
+ bte *ov = Tloc(o, 0);
+ lng *iv = Tloc(b, 0);
+ for(BUN i = 0; i<cnt; i++)
+ ov[i] = (bte)(iv[i] - min_val);
+ } else {
+ o = COLnew(b->hseqbase, TYPE_sht, cnt, role);
+ if (!o)
+ return NULL;
+ sht *ov = Tloc(o, 0);
+ lng *iv = Tloc(b, 0);
+ for(BUN i = 0; i<cnt; i++)
+ ov[i] = (sht)(iv[i] - min_val);
+ }
+ BATsetcount(o, cnt);
+ BATnegateprops(o);
+ return o;
+}
+
static str
FORcompress_intern(char **comp_min_val, BAT **r, BAT *b)
{
@@ -122,23 +184,9 @@ FORcompress_intern(char **comp_min_val,
throw(SQL, "for.compress", SQLSTATE(3F000) "for
compress: for 'for' compression column's cannot have NULL's");
if ((max_val-min_val) > GDK_sht_max)
throw(SQL, "for.compress", SQLSTATE(3F000) "for
compress: too large value spread for 'for' compression");
- if ((max_val-min_val) < GDK_bte_max/2) {
- o = COLnew(b->hseqbase, TYPE_bte, cnt, PERSISTENT);
- if (!o)
- throw(SQL, "for.compress", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
- bte *ov = Tloc(o, 0);
- lng *iv = Tloc(b, 0);
- for(BUN i = 0; i<cnt; i++)
- ov[i] = (bte)(iv[i] - min_val);
- } else {
- o = COLnew(b->hseqbase, TYPE_sht, cnt, PERSISTENT);
- if (!o)
- throw(SQL, "for.compress", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
- sht *ov = Tloc(o, 0);
- lng *iv = Tloc(b, 0);
- for(BUN i = 0; i<cnt; i++)
- ov[i] = (sht)(iv[i] - min_val);
- }
+ o = FORcompress_(b, min_val, max_val, PERSISTENT);
+ if (!o)
+ throw(SQL, "for.compress", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
snprintf(buf, 64, "FOR-" LLFMT, min_val);
} else {
GDKfree(mn);
@@ -149,8 +197,6 @@ FORcompress_intern(char **comp_min_val,
bat_destroy(o);
throw(SQL, "for.compress", SQLSTATE(HY013) MAL_MALLOC_FAIL);
}
- BATsetcount(o, cnt);
- BATnegateprops(o);
*r = o;
return NULL;
}
@@ -236,3 +282,42 @@ FORcompress_col(Client cntxt, MalBlkPtr
}
return msg;
}
+
+int
+FORprepare4append(BAT **noffsets, BAT *b, lng minval, int tt)
+{
+ ptr mn = NULL, mx = NULL;
+ *noffsets = NULL;
+
+ if (!(mn = BATmin(b, NULL)))
+ return -1;
+ if (!(mx = BATmax(b, NULL))) {
+ GDKfree(mn);
+ return -1;
+ }
+ if (b->ttype == TYPE_lng) {
+ lng min_val = *(lng*)mn;
+ lng max_val = *(lng*)mx;
+ lng maxcnt = (tt == TYPE_bte)?GDK_bte_max/2:GDK_sht_max;
+
+ GDKfree(mn);
+ GDKfree(mx);
+ if (min_val < minval || max_val < minval || (max_val - minval)
> maxcnt)
+ return 0; /* decompress */
+
+ *noffsets = FORcompress_(b, minval, max_val, TRANSIENT);
+ }
+ return 0;
+}
+
+int
+FORprepare4append_vals(void **noffsets, void *vals, BUN cnt, lng minval, int
vtype, int tt)
+{
+ *noffsets = NULL;
+ (void)vals;
+ (void)cnt;
+ (void)minval;
+ (void)vtype;
+ (void)tt;
+ return 0;
+}
diff --git a/sql/backends/monet5/for.h b/sql/backends/monet5/for.h
--- a/sql/backends/monet5/for.h
+++ b/sql/backends/monet5/for.h
@@ -4,6 +4,7 @@
#include "sql.h"
+//extern BAT *FORdecompress_(BAT *o, lng minval, int type, role_t role);
extern str FORcompress_col(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
extern str FORdecompress(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
diff --git a/sql/storage/bat/bat_storage.c b/sql/storage/bat/bat_storage.c
--- a/sql/storage/bat/bat_storage.c
+++ b/sql/storage/bat/bat_storage.c
@@ -1170,6 +1170,62 @@ dict_append_bat(column_storage *cs, BAT
return i;
}
+static BAT *
+for_append_bat(column_storage *cs, BAT *i, char *storage_type)
+{
+ lng offsetval = strtoll(storage_type+4, NULL, 10);
+ BAT *newoffsets = NULL;
+ BAT *b = NULL, *n = NULL;
+
+ if (!(b = temp_descriptor(cs->bid)))
+ return NULL;
+
+ if (FORprepare4append(&newoffsets, i, offsetval, b->ttype) < 0) {
+ assert(0);
+ } else {
+ /* returns new offset bat if values within min/max, else
decompress */
+ if (!newoffsets) { /* decompress */
+ if (cs->ucnt) {
+ BAT *ui = NULL, *uv = NULL;
+ BAT *nb = COLcopy(b, b->ttype, true, TRANSIENT);
+ bat_destroy(b);
+ if (!nb || cs_real_update_bats(cs, &ui, &uv) !=
LOG_OK) {
+ bat_destroy(nb);
+ return NULL;
+ }
+ b = nb;
+ if (BATupdate(b, ui, uv, true) != GDK_SUCCEED) {
+ bat_destroy(ui);
+ bat_destroy(uv);
+ bat_destroy(b);
+ }
+ bat_destroy(ui);
+ bat_destroy(uv);
+ }
+ n = FORdecompress_(b, offsetval, i->ttype, PERSISTENT);
+ bat_destroy(b);
+ if (!n)
+ return NULL;
+ if (cs->bid)
+ temp_destroy(cs->bid);
+ cs->bid = temp_create(n);
+ cs->ucnt = 0;
+ if (cs->uibid)
+ temp_destroy(cs->uibid);
+ if (cs->uvbid)
+ temp_destroy(cs->uvbid);
+ cs->uibid = cs->uvbid = 0;
+ cs->st = ST_DEFAULT;
+ cs->cleared = true;
+ b = n;
+ } else { /* append */
+ i = newoffsets;
+ }
+ }
+ bat_destroy(b);
+ return i;
+}
+
/*
* Returns LOG_OK, LOG_ERR or LOG_CONFLICT
*/
@@ -1519,14 +1575,13 @@ dict_append_val(column_storage *cs, void
return NULL;
}
n = DICTdecompress_(b, u, PERSISTENT);
- /* TODO also decrompress updates if any */
+ /* TODO decompress updates if any */
bat_destroy(b);
assert(newoffsets == NULL);
if (!n) {
bat_destroy(u);
return NULL;
_______________________________________________
checkin-list mailing list
checkin-list@monetdb.org
https://www.monetdb.org/mailman/listinfo/checkin-list
