Changeset: 646b89c671af for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/646b89c671af
Modified Files:
monetdb5/modules/mal/tablet.c
monetdb5/modules/mal/tablet.h
sql/backends/monet5/sql_result.c
Branch: directappend
Log Message:
Append simple types in batch
diffs (195 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
@@ -806,7 +806,7 @@ SQLload_error(READERtask *task, lng idx,
* either case an entry is added to the error table.
*/
static inline int
-SQLinsert_val(READERtask *task, int col, int idx)
+SQLinsert_val(READERtask *task, int col, int idx, bool one_by_one)
{
Column *fmt = task->as->format + col;
const void *adt;
@@ -896,6 +896,10 @@ SQLinsert_val(READERtask *task, int col,
fmt->c->tnonil = false;
}
if (task->loadops) {
+ if (!one_by_one)
+ return ret;
+ // Simple fixed size types will be appended in bulk.
+ // Here we deal with the more messy ones
str msg = task->loadops->append_one(task->loadops->state, idx,
adt, fmt->appendcol);
if (msg == MAL_SUCCEED)
return ret;
@@ -923,6 +927,65 @@ SQLinsert_val(READERtask *task, int col,
}
static int
+SQLworker_directappend_column(READERtask *task, int col)
+{
+ Column *c = &task->as->format[col];
+ int count = task->top[task->cur];
+
+ int type = c->adt;
+ size_t width = ATOMsize(type);
+ bool batch_mode = !ATOMvarsized(type);
+
+ if (!batch_mode) {
+ for (int i = 0; i < count; i++) {
+ if (SQLinsert_val(task, col, i, true) < 0)
+ return -1;
+ }
+ return 0;
+ }
+
+ size_t allocation_size = count * width;
+ GDKfree(c->data);
+ c->data = GDKmalloc(allocation_size);
+ if (c->data) {
+ c->len = allocation_size;
+ } else {
+ tablet_error(task, lng_nil, lng_nil, int_nil, "cannot allocate
memory", "");
+ return -1;
+ }
+
+ char * const allocation = c->data;
+ char *cursor = c->data;
+ for (int i = 0; i < count; i++) {
+ // We have to be careful here, c->data is not pointing at the
beginning
+ // of a malloc'ed area, but into the middle. If SQLinsert_val
tries to
+ // reallocate it we're screwed. However, c->len is sufficient
so it
+ // shouldn't try to reallocate.
+ c->data = cursor;
+ c->len = width;
+ if (SQLinsert_val(task, col, i, false) < 0) {
+ // We don't free our allocation here, it can probably
be reused for the next block.
+ c->data = allocation;
+ c->len = allocation_size;
+ return -1;
+ }
+ cursor += width;
+ }
+ // We don't free our allocation here, it can probably be reused for the
next block.
+ c->data = allocation;
+ c->len = allocation_size;
+
+ // Now insert it.
+ str msg = task->loadops->append_batch(task->loadops->state, c->data,
count, width, c->appendcol);
+ if (msg != MAL_SUCCEED) {
+ tablet_error(task, lng_nil, lng_nil, col, "bulk insert failed",
msg);
+ return -1;
+ }
+
+ return 0;
+}
+
+static int
SQLworker_column(READERtask *task, int col)
{
int i;
@@ -932,11 +995,7 @@ SQLworker_column(READERtask *task, int c
return 0;
if (task->loadops) {
- for (i = 0; i < task->top[task->cur]; i++) {
- if (SQLinsert_val(task, col, i) < 0)
- return -1;
- }
- return 0;
+ return SQLworker_directappend_column(task, col);
}
/* watch out for concurrent threads */
@@ -951,7 +1010,7 @@ SQLworker_column(READERtask *task, int c
MT_lock_unset(&mal_copyLock);
for (i = 0; i < task->top[task->cur]; i++) {
- if (!fmt[col].skip && SQLinsert_val(task, col, i) < 0) {
+ if (!fmt[col].skip && SQLinsert_val(task, col, i, true) < 0) {
BATsetcount(fmt[col].c, BATcount(fmt[col].c));
return -1;
}
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
@@ -74,11 +74,13 @@ typedef struct Table_t {
// SQLload_file doesn't know how to manipulate the sql transaction
bookkeeping, caller does.
typedef str (*loadfile_claim_fptr)(void *state, size_t nrows, size_t ncols,
Column *cols[]);
typedef str (*loadfile_append_one_fptr)(void *state, size_t idx, const void
*data, void *col);
+typedef str (*loadfile_append_batch_fptr)(void *state, const void *data, BUN
count, int width, 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_append_batch_fptr append_batch;
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
@@ -876,8 +876,6 @@ directappend_get_offsets_bat(void *state
return state->all_offsets;
}
-// Currently we're appending the values one by one but we need to switch to
-// a bulk interface.
static str
directappend_append_one(void *state_, size_t idx, const void *const_data, void
*col)
{
@@ -907,6 +905,46 @@ directappend_append_one(void *state_, si
return MAL_SUCCEED;
}
+static str
+directappend_append_batch(void *state_, const void *const_data, BUN count, int
width, void *col)
+{
+ struct directappend *state = state_;
+ sqlstore *store = state->mvc->session->tr->store;
+ sql_column *c = col;
+ int tpe = c->type.type->localtype;
+
+ (void)width;
+ assert(width== ATOMsize(tpe));
+
+ BUN scattered_count = state->new_offsets ? BATcount(state->new_offsets)
: 0;
+
+ int ret = LOG_OK;
+
+ if (scattered_count > 0) {
+ BUN dummy_offset = GDK_oid_max;
+ ret = store->storage_api.append_col(
+ state->mvc->session->tr, c,
+ dummy_offset, state->new_offsets,
+ (void*)const_data, scattered_count, tpe
+ );
+ }
+
+ if (ret == LOG_OK && count > scattered_count) {
+ char *remaining_data = (char*)const_data + scattered_count *
width;
+ BUN remaining_count = count - scattered_count;
+ ret = store->storage_api.append_col(
+ state->mvc->session->tr, c,
+ state->offset, NULL,
+ remaining_data, remaining_count, 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
mvc_import_table(Client cntxt, BAT ***bats, mvc *m, bstream *bs, sql_table *t,
const char *sep, const char *rsep, const char *ssep, const char *ns, lng sz,
lng offset, int best, bool from_stdin, bool escape, bool append_directly)
{
@@ -922,6 +960,7 @@ mvc_import_table(Client cntxt, BAT ***ba
.state = NULL,
.claim = directappend_claim,
.append_one = directappend_append_one,
+ .append_batch = directappend_append_batch,
.get_offsets = directappend_get_offsets_bat,
};
LoadOps *loadops = NULL;
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
