Changeset: c340a28726db for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/c340a28726db
Modified Files:
monetdb5/modules/mal/tablet.c
monetdb5/modules/mal/tablet.h
sql/backends/monet5/sql_result.c
Branch: directappend
Log Message:
New-style append to table banana2, if it exists
diffs (truncated from 324 to 300 lines):
diff --git a/monetdb5/modules/mal/tablet.c b/monetdb5/modules/mal/tablet.c
--- a/monetdb5/modules/mal/tablet.c
+++ b/monetdb5/modules/mal/tablet.c
@@ -636,6 +636,7 @@ typedef struct {
int besteffort;
bte *rowerror;
int errorcnt;
+ LoadOps *loadops;
} READERtask;
static void
@@ -892,9 +893,16 @@ SQLinsert_val(READERtask *task, int col,
adt = fmt->nildata;
fmt->c->tnonil = false;
}
+ if (task->loadops) {
+ str msg = task->loadops->append_one(task->loadops->state, idx,
adt, fmt->appendcol);
+ if (msg != MAL_SUCCEED)
+ goto failure;
+ }
+
if (bunfastapp(fmt->c, adt) == GDK_SUCCEED)
return ret;
+failure:
/* failure */
if (task->rowerror) {
lng row = BATcount(fmt->c);
@@ -1546,7 +1554,8 @@ SQLload_file(Client cntxt, Tablet *as, b
BUN cnt = 0, cntstart = 0, leftover = 0;
int res = 0; /* < 0: error, > 0: success, == 0: continue
processing */
int j;
- BUN firstcol;
+ BAT *countbat;
+ // BUN firstcol;
BUN i, attr;
READERtask task;
READERtask ptask[MAXWORKERS];
@@ -1564,6 +1573,8 @@ SQLload_file(Client cntxt, Tablet *as, b
.from_stdin = from_stdin,
.as = as,
.escape = escape, /* TODO: implement feature!!! */
+ // .loadops = loadops,
+ .loadops = loadops,
};
/* create the reject tables */
@@ -1712,13 +1723,23 @@ SQLload_file(Client cntxt, Tablet *as, b
#ifdef MLOCK_TST
mlock(task.b->buf, task.b->size);
#endif
- for (firstcol = 0; firstcol < task.as->nr_attrs; firstcol++)
- if (task.as->format[firstcol].c != NULL)
- break;
+ countbat = NULL;
+ if (loadops) {
+ countbat = loadops->get_offsets(loadops->state);
+ } else {
+ for (BUN i = 0; i < task.as->nr_attrs; i++) {
+ if (task.as->format[i].c != NULL) {
+ countbat = task.as->format[i].c;
+ break;
+ }
+ }
+ }
+ assert(countbat != NULL);
+
while (res == 0 && cnt < task.maxrow) {
// track how many elements are in the aggregated BATs
- cntstart = BATcount(task.as->format[firstcol].c);
+ cntstart = BATcount(countbat);
/* block until the producer has data available */
MT_sema_down(&task.consumer);
cnt += task.top[task.cur];
@@ -1745,6 +1766,19 @@ SQLload_file(Client cntxt, Tablet *as, b
MT_sema_up(&ptask[j].sema);
}
}
+
+ if (task.top[task.cur] && loadops) {
+ /* while the worker threads are working, allocate rows
*/
+ str msg = loadops->claim(loadops->state,
task.top[task.cur], 0, NULL);
+ if (msg != MAL_SUCCEED) {
+ tablet_error(&task, BATcount(countbat),
lng_nil, lng_nil, msg, "SQLload_file");
+ res = -1;
+ for (j = 0; j < threads; j++)
+ MT_sema_down(&ptask[j].reply);
+ break;
+ }
+ }
+
if (task.top[task.cur]) {
/* await completion of row break phase */
for (j = 0; j < threads; j++) {
@@ -1807,7 +1841,7 @@ SQLload_file(Client cntxt, Tablet *as, b
/* TRC_DEBUG(MAL_SERVER, "Trim bbest '%d' table size " BUNFMT " -
rows found so far " BUNFMT "\n",
best,
BATcount(as->format[firstcol].c), task.cnt); */
- if (best && BATcount(as->format[firstcol].c)) {
+ if (best && BATcount(countbat)) {
BUN limit;
int width;
@@ -1872,7 +1906,7 @@ SQLload_file(Client cntxt, Tablet *as, b
/* TRC_DEBUG(MAL_SERVER, "End of block stream eof=%d - res=%d\n",
task.ateof, res);*/
- cnt = BATcount(task.as->format[firstcol].c);
+ cnt = BATcount(countbat);
task.ateof = true;
task.state = ENDOFCOPY;
diff --git a/monetdb5/modules/mal/tablet.h b/monetdb5/modules/mal/tablet.h
--- a/monetdb5/modules/mal/tablet.h
+++ b/monetdb5/modules/mal/tablet.h
@@ -72,11 +72,13 @@ typedef struct Table_t {
typedef str (*loadfile_claim_fptr)(void *state, size_t nrows, size_t ncols,
Column *cols[]);
-typedef str (*loadfile_append_one_fptr)(void *state, void *data, Column *col);
+typedef str (*loadfile_append_one_fptr)(void *state, size_t idx, const void
*data, void *col);
+typedef BAT *(*loadfile_get_offsets_bat_fptr)(void *state);
typedef struct LoadOps {
void *state;
loadfile_claim_fptr claim;
loadfile_append_one_fptr append_one;
+ loadfile_get_offsets_bat_fptr get_offsets;
} LoadOps;
diff --git a/sql/backends/monet5/sql_result.c b/sql/backends/monet5/sql_result.c
--- a/sql/backends/monet5/sql_result.c
+++ b/sql/backends/monet5/sql_result.c
@@ -755,22 +755,156 @@ bailout:
}
static str
-directappend_claim(void *state, size_t nrows, size_t ncols, Column *cols[])
+directappend_claim(void *state_, size_t nrows, size_t ncols, Column *cols[])
{
- (void)state;
- (void)nrows;
+ str msg = MAL_SUCCEED;
+
(void)ncols;
(void)cols;
- throw(SQL, "direct_append", "not implemented yet");
+
+ assert(state_ != NULL);
+ struct directappend *state = state_;
+
+ if (state->new_offsets != NULL) {
+ // leftover from previous round. logic below counts on it not
being present.
+ // we can change that but have to do so carefully.
+ // for now just drop it
+ BBPreclaim(state->new_offsets);
+ state->new_offsets = NULL;
+ }
+
+ // Allocate room for this batch
+ BUN dummy_offset = 424242424242;
+ state->offset = dummy_offset;
+ sql_trans *tr = state->mvc->session->tr;
+ sqlstore *store = tr->store;
+ int ret = store->storage_api.claim_tab(tr, state->t, nrows,
&state->offset, &state->new_offsets);
+ // int ret = mvc_claim_slots(state->mvc->session->tr, state->t, nrows,
&state->offset, &state->new_offsets);
+ if (ret != LOG_OK) {
+ msg = createException(SQL, "sql.append_from", SQLSTATE(3F000)
"Could not claim slots");
+ goto bailout;
+ }
+
+ // Append the batch to all_offsets
+ if (state->new_offsets != NULL) {
+ if (BATappend(state->all_offsets, state->new_offsets, NULL,
false) != GDK_SUCCEED) {
+ msg = createException(SQL, "sql.append_from",
SQLSTATE(3F000) "BATappend failed");
+ goto bailout;
+ }
+ } else {
+ // is there a BATfunction for this?
+ BUN oldcount = BATcount(state->all_offsets);
+ BUN newcount = oldcount + nrows;
+ if (BATcapacity(state->all_offsets) < newcount) {
+ if (BATextend(state->all_offsets, newcount) !=
GDK_SUCCEED) {
+ msg = createException(SQL, "sql.append_from",
SQLSTATE(HY013) MAL_MALLOC_FAIL);
+ goto bailout;
+ }
+ }
+ oid * oo = Tloc(state->all_offsets, oldcount);
+ for (BUN i = 0; i < nrows; i++)
+ *oo++ = state->offset + i;
+ BATsetcount(state->all_offsets, newcount);
+ }
+
+ // The protocol for mvc_claim is that it returns either a consecutive
block,
+ // by setting *offset, or a BAT of positions by setting *offsets.
However,
+ // it is possible and even likely that only the first few items of the
BAT
+ // are actually scattered positions, while the rest is still a
consecutive
+ // block at the end. Appending at the end is much cheaper so we peel the
+ // consecutive elements off the back of the BAT and treat them
separately.
+ //
+ // In the remainder of the function, 'state->newoffsets' holds
'front_count'
+ // positions if it exists, while another 'back_count' positions start at
+ // 'back_offset'.
+ size_t front_count;
+ size_t back_count;
+ BUN back_offset;
+ if (state->new_offsets != NULL) {
+ if (state->new_offsets->tsorted) {
+ assert(BATcount(state->new_offsets) >= 1);
+ BUN start = BATcount(state->new_offsets) - 1;
+ oid at_start = *(oid*)Tloc(state->new_offsets, start);
+ while (start > 0) {
+ oid below_start =
*(oid*)Tloc(state->new_offsets, start - 1);
+ if (at_start != below_start + 1)
+ break;
+ start = start - 1;
+ at_start = below_start;
+ }
+ front_count = start;
+ back_count = nrows - start;
+ back_offset = at_start;
+ BATsetcount(state->new_offsets, start);
+ } else {
+ front_count = nrows;
+ back_count = 0;
+ back_offset = dummy_offset;
+ }
+ } else {
+ front_count = 0;
+ back_count = nrows;
+ back_offset = state->offset;
+ }
+
+ // debugging
+ if (front_count > 0) {
+ for (size_t j = 0; j < front_count; j++) {
+ BUN pos = (BUN)*(oid*)Tloc(state->new_offsets, j);
+ fprintf(stderr, "scattered offset[%zu] = " BUNFMT "\n",
j, pos);
+ }
+ }
+ if (back_count > 0) {
+ BUN start = back_offset;
+ BUN end = start + (BUN)back_count - 1;
+ fprintf(stderr, "consecutive offsets: " BUNFMT " .. "
BUNFMT"\n", start, end);
+ }
+
+ state->offset = back_offset;
+
+ assert(msg == MAL_SUCCEED);
+ return msg;
+
+bailout:
+ assert(msg != MAL_SUCCEED);
+ return msg;
+}
+
+static BAT*
+directappend_get_offsets_bat(void *state_)
+{
+ assert(state_);
+ struct directappend *state = state_;
+ return state->all_offsets;
}
static str
-directappend_append_one(void *state, void *data, Column *col)
+directappend_append_one(void *state_, size_t idx, const void *const_data, void
*col)
{
- (void)state;
- (void)data;
- (void)col;
- throw(SQL, "direct_append", "not implemented yet");
+ struct directappend *state = state_;
+ BAT *scattered_offsets = state->new_offsets;
+ BUN scattered_count = scattered_offsets ? BATcount(scattered_offsets) :
0;
+ BUN off;
+ if (idx < scattered_count) {
+ off = *(oid*)Tloc(scattered_offsets, idx);
+ fprintf(stderr, "Took offset " BUNFMT " from position %zu of
the offsets BAT\n", off, idx);
+ } else {
+ off = state->offset + (idx - scattered_count);
+ fprintf(stderr, "Took offset " BUNFMT " as %zu plus base "
BUNFMT "\n", off, idx, state->offset);
+ }
+
+ sql_column *c = col;
+ int tpe = c->type.type->localtype;
+ sqlstore *store = state->mvc->session->tr->store;
+
+ // unfortunately, append_col_fptr doesn't take const void*.
+ void *data = (void*)const_data;
+ int ret = store->storage_api.append_col(state->mvc->session->tr, c,
off, NULL, data, 1, tpe);
+ if (ret != LOG_OK) {
+ throw(SQL, "sql.append", SQLSTATE(42000) "Append failed%s", ret
== LOG_CONFLICT ? " due to conflict with another transaction" : "");
+ }
+
+ return MAL_SUCCEED;
}
str
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
