Changeset: 13cfc888fcfb for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/13cfc888fcfb
Modified Files:
clients/Tests/exports.stable.out
gdk/gdk.h
gdk/gdk_align.c
gdk/gdk_bat.c
gdk/gdk_batop.c
gdk/gdk_bbp.c
gdk/gdk_sketch.c
gdk/gdk_storage.c
monetdb5/modules/kernel/aggr.c
sql/backends/monet5/sql_statistics.c
sql/scripts/80_statistics.sql
sql/storage/bat/bat_storage.c
Branch: persistent_sketches
Log Message:
Persistent sketches for persistent BATs. WIP.
Handling heap load/unload. Handling some appends/updates. Missing deletes.
diffs (truncated from 1131 to 300 lines):
diff --git a/clients/Tests/exports.stable.out b/clients/Tests/exports.stable.out
--- a/clients/Tests/exports.stable.out
+++ b/clients/Tests/exports.stable.out
@@ -597,8 +597,10 @@ const ptr ptr_nil;
ssize_t shtFromStr(allocator *ma, const char *src, size_t *len, sht **dst,
bool external);
ssize_t shtToStr(allocator *ma, char **dst, size_t *len, const sht *src, bool
external);
const sht sht_nil;
-double sketch_estimate(uint8_t cnt_sketch[BUCKETS][CLZ_BUCKETS]);
-int sketch_populate(BAT *n, BATiter *ni, struct canditer *nci, uint8_t
cnting_sketch[BUCKETS][CLZ_BUCKETS]);
+gdk_export gdk_return sketch_update_many(BATiter *bi, uint8_t
(*sketch)[CLZ_BUCKETS], const void *values, size_t count);
+gdk_export gdk_return sketch_merge(uint8_t (*dst)[CLZ_BUCKETS], uint8_t
(*src)[CLZ_BUCKETS]);
+gdk_export double sketch_estimate(uint8_t (*sketch)[CLZ_BUCKETS]);
+gdk_return bat_sketch(BATiter *bi, struct canditer *bci, uint8_t
(*sketch)[CLZ_BUCKETS]);
ssize_t sql_daytime_fromstr(allocator *ma, const char *buf, daytime *d, long
tz_sec, bool tzlocal);
ssize_t sql_timestamp_fromstr(allocator *ma, const char *buf, timestamp *d,
long tz_sec, bool tzlocal);
ssize_t strFromStr(allocator *ma, const char *restrict src, size_t *restrict
len, char **restrict dst, bool external);
diff --git a/gdk/gdk.h b/gdk/gdk.h
--- a/gdk/gdk.h
+++ b/gdk/gdk.h
@@ -1549,12 +1549,13 @@ typedef struct pipeline_io {
#define ORDERIDXOFF 3
-#define HLL_BITS 6
+#define HLL_BITS 6
#define BITS_MASK ((1ULL << HLL_BITS) - 1)
#define MAX_CLZ (64 - HLL_BITS)
#define BUCKETS (1U << HLL_BITS)
#define CLZ_BUCKETS (MAX_CLZ + 1)
#define HLLSEED 0xadc83b19ULL
+#define SKETCH_SIZE (BUCKETS * CLZ_BUCKETS)
/* assert that atom width is power of 2, i.e., width == 1<<shift */
#define assert_shift_width(shift,width) \
@@ -1631,6 +1632,7 @@ typedef struct BAT {
Heap *theap; /* space for the column. */
BUN tbaseoff; /* offset in heap->base (in whole items) */
Heap *tvheap; /* space for the varsized data. */
+ Heap *tsketch;
Hash *thash; /* hash table */
#ifdef HAVE_RTREE
RTree *trtree; /* rtree geometric index */
@@ -2082,6 +2084,20 @@ gdk_export const inet6 inet6_nil;
#define ATOMtype(t) ((t) == TYPE_void ? TYPE_oid : (t))
#define ATOMextern(t) (ATOMstorage(t) >= TYPE_str)
+static inline bool
+ATOMsketchable(int type)
+{
+ switch (type) {
+ case TYPE_void:
+ case TYPE_bit:
+ case TYPE_msk:
+ case TYPE_oid:
+ case TYPE_ptr:
+ return false;
+ }
+ return true;
+}
+
/* The base type is the storage type if the comparison function, the
* hash function, and the nil value are the same as those of the
* storage type; otherwise it is the type itself. */
@@ -2308,6 +2324,7 @@ typedef struct BATiter {
Heap *h;
void *base;
Heap *vh;
+ Heap *skh;
BUN count;
BUN baseoff;
oid tseq;
@@ -2355,6 +2372,7 @@ bat_iterator_nolock(BAT *b)
: NULL,
.baseoff = b->tbaseoff,
.vh = b->tvheap,
+ .skh = b->tsketch,
.count = b->batCount,
.width = b->twidth,
.shift = b->tshift,
@@ -3421,11 +3439,11 @@ gdk_export void STRMPdestroy(BAT *b);
gdk_export bool BAThasstrimps(BAT *b);
gdk_export gdk_return BATsetstrimps(BAT *b);
-gdk_export int sketch_populate(BAT* n, BATiter *ni, struct canditer *nci,
- uint8_t cnting_sketch[BUCKETS][CLZ_BUCKETS]);
-/* gdk_export void sketch_merge(BAT* b, BAT* n); */
-gdk_export double sketch_estimate(uint8_t cnt_sketch[BUCKETS][CLZ_BUCKETS]);
-gdk_export double bat_guess_uniques(BAT *b, BATiter *bi, struct canditer *bci);
+/* Sketches exported functions */
+gdk_export gdk_return sketch_update_many(BATiter *bi, uint8_t
(*sketch)[CLZ_BUCKETS], const void *values, size_t count);
+gdk_export gdk_return sketch_merge(uint8_t (*dst)[CLZ_BUCKETS], uint8_t
(*src)[CLZ_BUCKETS]);
+gdk_export double sketch_estimate(uint8_t (*sketch)[CLZ_BUCKETS]);
+gdk_return bat_sketch(BATiter *bi, struct canditer *bci, uint8_t
(*sketch)[CLZ_BUCKETS]);
/* Rtree structure functions */
#ifdef HAVE_RTREE
@@ -3900,7 +3918,8 @@ enum heaptype {
hashheap,
orderidxheap,
strimpheap,
- dataheap
+ dataheap,
+ sketchheap
};
gdk_export int BBPselectfarm(role_t role, int type, enum heaptype hptype);
diff --git a/gdk/gdk_align.c b/gdk/gdk_align.c
--- a/gdk/gdk_align.c
+++ b/gdk/gdk_align.c
@@ -190,6 +190,7 @@ VIEWcreate(oid seq, BAT *b, BUN l, BUN h
bn->theap = bi.h;
bn->tbaseoff = bi.baseoff;
bn->tvheap = bi.vh;
+ bn->tsketch = bi.skh;
tp = VIEWtparent(b);
if (tp == 0 && b->ttype != TYPE_void)
@@ -198,12 +199,16 @@ VIEWcreate(oid seq, BAT *b, BUN l, BUN h
HEAPincref(bi.h);
if (bi.vh)
HEAPincref(bi.vh);
+ if (bi.skh)
+ HEAPincref(bi.skh);
if (l != 0 || h < bi.count)
VIEWboundsbi(&bi, bn, l, h);
MT_lock_unset(&b->theaplock);
BBPcacheit(bn, true);
BBPretain(bn->theap->parentid);
+ if (bn->tsketch)
+ BBPretain(bn->tsketch->parentid);
if (bn->tvheap)
BBPretain(bn->tvheap->parentid);
if (bn->ustr)
diff --git a/gdk/gdk_bat.c b/gdk/gdk_bat.c
--- a/gdk/gdk_bat.c
+++ b/gdk/gdk_bat.c
@@ -44,7 +44,6 @@
#include "monetdb_config.h"
#include "gdk.h"
#include "gdk_private.h"
-#include "mutils.h"
#ifdef ALIGN
#undef ALIGN
@@ -118,7 +117,7 @@ BATcreatedesc(oid hseq, int tt, bool hea
{
bat bid;
BAT *bn;
- Heap *h = NULL, *vh = NULL;
+ Heap *h = NULL, *vh = NULL, *skh = NULL;
/*
* Alloc space for the BAT and its dependent records.
@@ -148,12 +147,26 @@ BATcreatedesc(oid hseq, int tt, bool hea
.refs = ATOMIC_VAR_INIT(1),
};
}
+
+ if (ATOMsketchable(tt) && role == PERSISTENT) {
+ if ((skh = GDKmalloc(sizeof(Heap))) == NULL) {
+ GDKfree(h);
+ GDKfree(vh);
+ return NULL;
+ }
+ *skh = (Heap) {
+ .farmid = BBPselectfarm(role, tt, sketchheap),
+ .dirty = true,
+ .refs = ATOMIC_VAR_INIT(1),
+ };
+ }
}
bid = BBPallocbat(tt);
if (bid == 0) {
GDKfree(h);
GDKfree(vh);
+ GDKfree(skh);
return NULL;
}
bn = BBP_desc(bid);
@@ -183,6 +196,7 @@ BATcreatedesc(oid hseq, int tt, bool hea
.batRestricted = BAT_WRITE,
.theap = h,
.tvheap = vh,
+ .tsketch = skh,
.creator_tid = MT_getpid(),
.qc = role == TRANSIENT ? MT_thread_get_qry_ctx() : NULL,
};
@@ -192,6 +206,12 @@ BATcreatedesc(oid hseq, int tt, bool hea
const char *nme = BBP_physical(bn->batCacheid);
settailname(bn->theap, nme, tt, width);
+ if (bn->tsketch) {
+ bn->tsketch->parentid = bn->batCacheid;
+ strtconcat(bn->tsketch->filename,
+ sizeof(bn->tsketch->filename),
+ nme, ".tsketch", NULL);
+ }
if (bn->tvheap) {
bn->tvheap->parentid = bn->batCacheid;
strtconcat(bn->tvheap->filename,
@@ -334,10 +354,21 @@ COLnew2(oid hseq, int tt, BUN cap, role_
goto bailout;
}
+ if (ATOMsketchable(tt) && role == PERSISTENT) {
+ assert(bn->tsketch);
+ if (HEAPalloc(bn->tsketch, SKETCH_SIZE, sizeof(uint8_t)) !=
GDK_SUCCEED) {
+ HEAPfree(bn->theap, true);
+ goto bailout;
+ }
+ memset(bn->tsketch->base, 0, SKETCH_SIZE);
+ }
+
if (bn->tvheap && width == 0 && ATOMheap(tt, bn->tvheap, cap) !=
GDK_SUCCEED) {
HEAPfree(bn->theap, true);
+ HEAPfree(bn->tsketch, true);
goto bailout;
}
+
BBPcacheit(bn, true);
TRC_DEBUG(ALGO, "-> " ALGOBATFMT "\n", ALGOBATPAR(bn));
return bn;
@@ -539,6 +570,11 @@ BATclear(BAT *b, bool force)
}
}
+ if (b->tsketch) {
+ memset(b->tsketch->base, 0, SKETCH_SIZE);
+ b->tsketch->dirty = true;
+ }
+
b->batInserted = 0;
b->batCount = 0;
if (b->ttype == TYPE_void)
@@ -610,6 +646,19 @@ BATfree(BAT *b)
assert(b->tvheap->parentid == b->batCacheid);
HEAPfree(b->tvheap, false);
}
+ /* wait until there are no other references to the heap; a
+ * reference is possible in e.g. BBPsync that uses a
+ * bat_iterator directly on the BBP_desc, i.e. without fix */
+ while (b->tsketch && (ATOMIC_GET(&b->tsketch->refs) & HEAPREFS) > 1) {
+ MT_lock_unset(&b->theaplock);
+ MT_sleep_ms(1);
+ MT_lock_set(&b->theaplock);
+ }
+ if (b->tsketch) {
+ assert((ATOMIC_GET(&b->tsketch->refs) & HEAPREFS) == 1);
+ assert(b->tsketch->parentid == b->batCacheid);
+ HEAPfree(b->tsketch, false);
+ }
MT_lock_unset(&b->theaplock);
}
@@ -627,6 +676,9 @@ BATdestroy(BAT *b)
if (b->theap) {
GDKfree(b->theap);
}
+ if (b->tsketch) {
+ GDKfree(b->tsketch);
+ }
if (b->oldtail) {
ATOMIC_AND(&b->oldtail->refs, ~DELAYEDREMOVE);
/* the bat has not been committed, so we cannot remove
@@ -872,6 +924,10 @@ COLcopy2(BAT *b, int tt, bool writable,
memcpy(bn->theap->base, bi.base, bi.hfree);
bn->theap->free = bi.hfree;
bn->theap->dirty = true;
+ if (bn->tsketch) {
+ memcpy(bn->theap->base, bi.skh, SKETCH_SIZE);
+ bn->tsketch->dirty = true;
+ }
/* make sure we use the correct capacity */
if (ATOMstorage(bn->ttype) == TYPE_msk)
@@ -886,10 +942,13 @@ COLcopy2(BAT *b, int tt, bool writable,
TIMEOUT_LOOP_IDX_DECL(p, bi.count, qry_ctx) {
const void *t = BUNtail(&bi, p);
-
- if (bunfastapp_nocheck(bn, t) != GDK_SUCCEED) {
+ if (bunfastapp_nocheck(bn, t) != GDK_SUCCEED)
goto bunins_failed;
- }
+ }
+ if (bn->tsketch) {
+ if (bat_sketch(&bi, NULL, (uint8_t (*)[CLZ_BUCKETS])
bn->tsketch->base) != GDK_SUCCEED)
+ goto bunins_failed;
+ bn->tsketch->dirty = true;
}
TIMEOUT_CHECK(qry_ctx,
GOTO_LABEL_TIMEOUT_HANDLER(bunins_failed, qry_ctx));
bn->theap->dirty |= bi.count > 0;
@@ -921,6 +980,10 @@ COLcopy2(BAT *b, int tt, bool writable,
memcpy(Tloc(bn, 0), bi.base, bn->theap->free);
bn->theap->dirty |= bi.count > 0;
algo = "memcpy heap";
+ if (bn->tsketch) {
+ memcpy(bn->tsketch->base, bi.skh, SKETCH_SIZE);
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]
