Changeset: fe06a25316c5 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/fe06a25316c5
Modified Files:
monetdb5/modules/mal/tablet.c
monetdb5/modules/mal/tablet.h
sql/backends/monet5/sql.c
sql/backends/monet5/sql_result.c
sql/backends/monet5/sql_result.h
Branch: directappend
Log Message:
Pass loadops to SQLload_file
diffs (273 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
@@ -1540,8 +1540,9 @@ create_rejects_table(Client cntxt)
}
BUN
-SQLload_file(Client cntxt, Tablet *as, bstream *b, stream *out, const char
*csep, const char *rsep, char quote, lng skip, lng maxrow, int best, bool
from_stdin, const char *tabnam, bool escape)
+SQLload_file(Client cntxt, Tablet *as, bstream *b, stream *out, const char
*csep, const char *rsep, char quote, lng skip, lng maxrow, int best, bool
from_stdin, const char *tabnam, bool escape, LoadOps *loadops)
{
+ (void)loadops;
BUN cnt = 0, cntstart = 0, leftover = 0;
int res = 0; /* < 0: error, > 0: success, == 0: continue
processing */
int j;
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
@@ -50,6 +50,7 @@ typedef struct Column_t {
char quote; /* if set use this
character for string quotes */
const void *nildata;
int size;
+ void *appendcol; /* temporary, can use
Columnt_t.extra in the future */
} Column;
/*
@@ -69,7 +70,17 @@ typedef struct Table_t {
BAT *complaints; /* lines that did not match the
required input */
} Tablet;
-mal_export BUN SQLload_file(Client cntxt, Tablet *as, bstream *b, stream *out,
const char *csep, const char *rsep, char quote, lng skip, lng maxrow, int best,
bool from_stdin, const char *tabnam, bool escape);
+
+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 struct LoadOps {
+ void *state;
+ loadfile_claim_fptr claim;
+ loadfile_append_one_fptr append_one;
+} LoadOps;
+
+
+mal_export BUN SQLload_file(Client cntxt, Tablet *as, bstream *b, stream *out,
const char *csep, const char *rsep, char quote, lng skip, lng maxrow, int best,
bool from_stdin, const char *tabnam, bool escape, LoadOps *loadops);
mal_export str TABLETcreate_bats(Tablet *as, BUN est);
mal_export str TABLETcollect(BAT **bats, Tablet *as);
mal_export str TABLETcollect_parts(BAT **bats, Tablet *as, BUN offset);
diff --git a/sql/backends/monet5/sql.c b/sql/backends/monet5/sql.c
--- a/sql/backends/monet5/sql.c
+++ b/sql/backends/monet5/sql.c
@@ -2932,7 +2932,7 @@ mvc_import_table_wrap(Client cntxt, MalB
if (strNil(fname))
fname = NULL;
if (fname == NULL) {
- msg = mvc_import_table(cntxt, &b, be->mvc, be->mvc->scanner.rs,
t, tsep, rsep, ssep, ns, sz, offset, besteffort, true, escape);
+ msg = mvc_import_table(cntxt, &b, be->mvc, be->mvc->scanner.rs,
t, tsep, rsep, ssep, ns, sz, offset, besteffort, true, escape, append_directly);
} else {
if (onclient) {
mnstr_write(be->mvc->scanner.ws, PROMPT3,
sizeof(PROMPT3)-1, 1);
@@ -3018,7 +3018,7 @@ mvc_import_table_wrap(Client cntxt, MalB
close_stream(ss);
throw(MAL, "sql.copy_from", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
}
- msg = mvc_import_table(cntxt, &b, be->mvc, s, t, tsep, rsep,
ssep, ns, sz, offset, besteffort, false, escape);
+ msg = mvc_import_table(cntxt, &b, be->mvc, s, t, tsep, rsep,
ssep, ns, sz, offset, besteffort, false, escape, append_directly);
if (onclient) {
mnstr_write(be->mvc->scanner.ws, PROMPT3,
sizeof(PROMPT3)-1, 1);
mnstr_flush(be->mvc->scanner.ws, MNSTR_FLUSH_DATA);
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
@@ -688,8 +688,93 @@ has_whitespace(const char *s)
return 0;
}
+struct directappend {
+ mvc *mvc;
+ sql_table *t;
+ BAT *all_offsets;
+ BAT *new_offsets;
+ BUN offset;
+};
+
+static void
+directappend_destroy(struct directappend *state)
+{
+ if (state == NULL)
+ return;
+ struct directappend *st = state;
+ if (st->all_offsets)
+ BBPreclaim(st->all_offsets);
+ if (st->new_offsets)
+ BBPreclaim(st->new_offsets);
+}
+
+static str
+directappend_init(struct directappend *state, Client cntxt, sql_table *t)
+{
+ str msg = MAL_SUCCEED;
+ *state = (struct directappend) { 0 };
+ backend *be;
+ mvc *mvc;
+ sql_schema *s;
+
+ msg = checkSQLContext(cntxt);
+ if (msg != MAL_SUCCEED)
+ goto bailout;
+ be = cntxt->sqlcontext;
+ mvc = be->mvc;
+ state->mvc = mvc;
+
+ // temporary: append to table banana2 instead
+ s = t->s;
+ char *tname;
+ tname = "banana2"; // <<=============================================
+ if (NULL == (t = mvc_bind_table(mvc, s, tname))) {
+ msg = createException(SQL, "sql.append_from", SQLSTATE(3F000)
"Table missing: %s.%s", s->base.name, tname);
+ goto bailout;
+ }
+ if (!isTable(t)) {
+ msg = createException(SQL, "sql.append_from", SQLSTATE(42000)
"%s '%s' is not persistent", TABLE_TYPE_DESCRIPTION(t->type, t->properties),
t->base.name);
+ goto bailout;
+ }
+
+ state->t = t;
+
+ state->all_offsets = COLnew(0, TYPE_oid, 0, TRANSIENT);
+ if (state->all_offsets == NULL) {
+ msg = createException(SQL, "sql.append_from", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
+ goto bailout;
+ }
+
+ assert(msg == MAL_SUCCEED);
+ return msg;
+
+bailout:
+ assert(msg != MAL_SUCCEED);
+ directappend_destroy(state);
+ return msg;
+}
+
+static str
+directappend_claim(void *state, size_t nrows, size_t ncols, Column *cols[])
+{
+ (void)state;
+ (void)nrows;
+ (void)ncols;
+ (void)cols;
+ throw(SQL, "direct_append", "not implemented yet");
+}
+
+static str
+directappend_append_one(void *state, void *data, Column *col)
+{
+ (void)state;
+ (void)data;
+ (void)col;
+ throw(SQL, "direct_append", "not implemented yet");
+}
+
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)
+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)
{
int i = 0, j;
node *n;
@@ -697,6 +782,23 @@ mvc_import_table(Client cntxt, BAT ***ba
Column *fmt;
str msg = MAL_SUCCEED;
+
+ struct directappend directappend_state = { 0 };
+ LoadOps our_loadops = {
+ .state = NULL,
+ .claim = directappend_claim,
+ .append_one = directappend_append_one,
+ };
+ LoadOps *loadops = NULL;
+ if (append_directly) {
+ msg = directappend_init(&directappend_state, cntxt, t);
+ if (msg != MAL_SUCCEED)
+ return msg;
+ our_loadops.state = &directappend_state;
+ loadops = &our_loadops;
+
+ }
+
*bats =0; // initialize the receiver
if (!bs)
@@ -706,10 +808,13 @@ mvc_import_table(Client cntxt, BAT ***ba
char *stream_msg = mnstr_error(bs->s);
msg = createException(IO, "sql.copy_from", SQLSTATE(42000)
"Stream not open %s: %s", mnstr_error_kind_name(errnr), stream_msg ? stream_msg
: "unknown error");
free(stream_msg);
+ directappend_destroy(our_loadops.state);
return msg;
}
- if (offset < 0 || offset > (lng) BUN_MAX)
+ if (offset < 0 || offset > (lng) BUN_MAX) {
+ directappend_destroy(our_loadops.state);
throw(IO, "sql.copy_from", SQLSTATE(42000) "Offset out of
range");
+ }
if (offset > 0)
offset--;
@@ -726,14 +831,27 @@ mvc_import_table(Client cntxt, BAT ***ba
.filename = m->scanner.rs == bs ? NULL : "",
};
fmt = GDKzalloc(sizeof(Column) * (as.nr_attrs + 1));
- if (fmt == NULL)
+ if (fmt == NULL) {
+ directappend_destroy(our_loadops.state);
throw(IO, "sql.copy_from", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
+ }
as.format = fmt;
if (!isa_block_stream(bs->s))
out = NULL;
+ // temporary
+ node *n2;
+ n2 = append_directly ?
ol_first_node(directappend_state.t->columns) : NULL;
for (n = ol_first_node(t->columns), i = 0; n; n = n->next, i++)
{
sql_column *col = n->data;
+ // temporary
+ if (n2 != NULL) {
+ sql_column *col2 = n2->data;
+ assert(strcmp(col->base.name, col2->base.name)
== 0);
+ assert(strcmp(col->type.type->base.name,
col2->type.type->base.name) == 0);
+ fmt[i].appendcol = col2;
+ n2 = n2->next;
+ }
fmt[i].name = col->base.name;
fmt[i].sep = (n->next) ? sep : rsep;
@@ -752,6 +870,7 @@ mvc_import_table(Client cntxt, BAT ***ba
BBPunfix(fmt[j].c->batCacheid);
}
GDKfree(fmt[i].data);
+ directappend_destroy(our_loadops.state);
throw(IO, "sql.copy_from", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
}
fmt[i].c = NULL;
@@ -771,11 +890,12 @@ mvc_import_table(Client cntxt, BAT ***ba
fmt[i].size = ATOMsize(fmt[i].adt);
}
if ((msg = TABLETcreate_bats(&as, (BUN) (sz < 0 ? 1000 : sz)))
== MAL_SUCCEED){
- if (!sz || (SQLload_file(cntxt, &as, bs, out, sep,
rsep, ssep ? ssep[0] : 0, offset, sz, best, from_stdin, t->base.name, escape)
!= BUN_NONE &&
+ if (!sz || (SQLload_file(cntxt, &as, bs, out, sep,
rsep, ssep ? ssep[0] : 0, offset, sz, best, from_stdin, t->base.name, escape,
loadops) != BUN_NONE &&
(best || !as.error))) {
*bats = (BAT**) GDKzalloc(sizeof(BAT *) *
as.nr_attrs);
if ( *bats == NULL){
TABLETdestroy_format(&as);
+ directappend_destroy(our_loadops.state);
throw(IO, "sql.copy_from",
SQLSTATE(HY013) MAL_MALLOC_FAIL);
}
msg = TABLETcollect(*bats,&as);
@@ -793,6 +913,7 @@ mvc_import_table(Client cntxt, BAT ***ba
}
TABLETdestroy_format(&as);
}
+ directappend_destroy(our_loadops.state);
return msg;
}
diff --git a/sql/backends/monet5/sql_result.h b/sql/backends/monet5/sql_result.h
--- a/sql/backends/monet5/sql_result.h
+++ b/sql/backends/monet5/sql_result.h
@@ -26,7 +26,7 @@ extern int mvc_export_chunk(backend *b,
extern int mvc_export_prepare(backend *b, stream *s);
-extern str mvc_import_table(Client cntxt, BAT ***bats, mvc *c, bstream *s,
sql_table *t, const char *sep, const char *rsep, const char *ssep, const char
*ns, lng nr, lng offset, int best, bool from_stdin, bool escape);
+extern str mvc_import_table(Client cntxt, BAT ***bats, mvc *c, bstream *s,
sql_table *t, const char *sep, const char *rsep, const char *ssep, const char
*ns, lng nr, lng offset, int best, bool from_stdin, bool escape, bool
append_directly);
sql5_export int mvc_result_table(backend *be, oid query_id, int nr_cols,
mapi_query_t type, BAT *order);
sql5_export int mvc_result_column(backend *be, char *tn, char *name, char
*typename, int digits, int scale, BAT *b);
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
