Changeset: b5c70672c18f for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=b5c70672c18f
Modified Files:
clients/Tests/exports.stable.out
clients/mapiclient/mclient.c
clients/mapilib/mapi.c
clients/mapilib/mapi.h
sql/backends/monet5/sql.c
sql/backends/monet5/sql_result.c
sql/backends/monet5/sql_result.h
Branch: client-filetrans
Log Message:
Initial work on COPY INTO 'file'.
diffs (truncated from 431 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
@@ -559,7 +559,7 @@ MapiMsg mapi_seek_row(MapiHdl hdl, int64
MapiHdl mapi_send(Mapi mid, const char *cmd);
MapiMsg mapi_setAutocommit(Mapi mid, bool autocommit);
MapiMsg mapi_set_size_header(Mapi mid, int value);
-void mapi_setfilecallback(Mapi mid, char *( *func)(void *priv, const char
*filename, bool binary, uint64_t offset, uint64_t *size), void *priv);
+void mapi_setfilecallback(Mapi mid, char *( *getfunc)(void *priv, const char
*filename, bool binary, uint64_t offset, size_t *size), char *( *putfunc)(void
*priv, const char *filename, const void *restrict data, size_t size), void
*priv);
int mapi_split_line(MapiHdl hdl);
MapiMsg mapi_start_talking(Mapi mid);
MapiMsg mapi_store_field(MapiHdl hdl, int fnr, int outtype, void *outparam);
diff --git a/clients/mapiclient/mclient.c b/clients/mapiclient/mclient.c
--- a/clients/mapiclient/mclient.c
+++ b/clients/mapiclient/mclient.c
@@ -3036,7 +3036,7 @@ struct privdata {
#define READSIZE (1 << 20)
static char *
-getfile(void *data, const char *filename, bool binary, uint64_t offset,
uint64_t *size)
+getfile(void *data, const char *filename, bool binary, uint64_t offset, size_t
*size)
{
stream *f;
char *buf;
@@ -3086,10 +3086,33 @@ getfile(void *data, const char *filename
priv->f = NULL;
return NULL;
}
- *size = (uint64_t) s;
+ *size = (size_t) s;
return buf;
}
+static char *
+putfile(void *data, const char *filename, const void *restrict buf, size_t
bufsize)
+{
+ struct privdata *priv = data;
+
+ if (filename != NULL) {
+ if ((priv->f = open_wastream(filename)) == NULL)
+ return "cannot open file";
+ if (buf == NULL || bufsize == 0)
+ return NULL;
+ } else if (buf == NULL) {
+ close_stream(priv->f);
+ priv->f = NULL;
+ return NULL;
+ }
+ if (mnstr_write(priv->f, buf, 1, bufsize) < (ssize_t) bufsize) {
+ close_stream(priv->f);
+ priv->f = NULL;
+ return "error writing output";
+ }
+ return NULL; /* success */
+}
+
__declspec(noreturn) static void usage(const char *prog, int xit)
__attribute__((__noreturn__));
@@ -3494,7 +3517,7 @@ main(int argc, char **argv)
struct privdata priv;
priv = (struct privdata) {0};
- mapi_setfilecallback(mid, getfile, &priv);
+ mapi_setfilecallback(mid, getfile, putfile, &priv);
if (!autocommit)
mapi_setAutocommit(mid, autocommit);
diff --git a/clients/mapilib/mapi.c b/clients/mapilib/mapi.c
--- a/clients/mapilib/mapi.c
+++ b/clients/mapilib/mapi.c
@@ -839,8 +839,9 @@ struct MapiStruct {
stream *tracelog; /* keep a log for inspection */
stream *from, *to;
uint32_t index; /* to mark the log records */
- void *getfilecontentprivate;
- char *(*getfilecontent)(void *, const char *, bool, uint64_t, uint64_t
*);
+ void *filecontentprivate;
+ char *(*getfilecontent)(void *, const char *, bool, uint64_t, size_t *);
+ char *(*putfilecontent)(void *, const char *, const void *restrict,
size_t);
};
struct MapiResultSet {
@@ -2985,10 +2986,11 @@ mapi_disconnect(Mapi mid)
* callback function the opportunity to free any resources.
*/
void
-mapi_setfilecallback(Mapi mid, char *(*getfilecontent)(void *, const char *,
bool, uint64_t, uint64_t *), void *getfilecontentprivate)
+mapi_setfilecallback(Mapi mid, char *(*getfilecontent)(void *, const char *,
bool, uint64_t, size_t *), char *(*putfilecontent)(void *, const char *, const
void *restrict, size_t), void *filecontentprivate)
{
mid->getfilecontent = getfilecontent;
- mid->getfilecontentprivate = getfilecontentprivate;
+ mid->putfilecontent = putfilecontent;
+ mid->filecontentprivate = filecontentprivate;
}
#define testBinding(hdl,fnr,funcname) \
@@ -3882,10 +3884,55 @@ parse_header_line(MapiHdl hdl, char *lin
}
static void
-handle_file(MapiHdl hdl, uint64_t off, char *filename, bool binary)
+write_file(MapiHdl hdl, char *filename)
{
Mapi mid = hdl->mid;
- uint64_t size = 0, flushsize = 0;
+ char *line;
+ char data[BLOCK];
+ ssize_t len;
+
+ (void) read_line(mid); /* read flush marker */
+ if (filename == NULL) {
+ /* malloc failure */
+ mnstr_printf(mid->to, "!HY001!allocation failure\n");
+ mnstr_flush(mid->to);
+ return;
+ }
+ if (mid->putfilecontent == NULL) {
+ free(filename);
+ mnstr_printf(mid->to, "!HY000!cannot send files\n");
+ mnstr_flush(mid->to);
+ return;
+ }
+ line = mid->putfilecontent(mid->filecontentprivate, filename, NULL, 0);
+ free(filename);
+ if (line != NULL) {
+ if (strchr(line, '\n'))
+ line = "incorrect response from application";
+ mnstr_printf(mid->to, "!HY000!%.64s\n", line);
+ mnstr_flush(mid->to);
+ return;
+ }
+ mnstr_flush(mid->to);
+ while ((len = mnstr_read(mid->from, data, 1, sizeof(data))) > 0) {
+ if (line == NULL)
+ line = mid->putfilecontent(mid->filecontentprivate,
+ NULL, data, len);
+ }
+ if (line == NULL)
+ line = mid->putfilecontent(mid->filecontentprivate,
+ NULL, NULL, 0);
+ if (line && strchr(line, '\n'))
+ line = "incorrect response from application";
+ mnstr_printf(mid->to, "%s\n", line ? line : "");
+ mnstr_flush(mid->to);
+}
+
+static void
+read_file(MapiHdl hdl, uint64_t off, char *filename, bool binary)
+{
+ Mapi mid = hdl->mid;
+ size_t size = 0, flushsize = 0;
char *data, *line;
(void) read_line(mid); /* read flush marker */
@@ -3901,7 +3948,7 @@ handle_file(MapiHdl hdl, uint64_t off, c
mnstr_flush(mid->to);
return;
}
- data = mid->getfilecontent(mid->getfilecontentprivate, filename,
binary, off, &size);
+ data = mid->getfilecontent(mid->filecontentprivate, filename, binary,
off, &size);
free(filename);
if (data != NULL && size == 0) {
if (strchr(data, '\n'))
@@ -3917,24 +3964,24 @@ handle_file(MapiHdl hdl, uint64_t off, c
line = read_line(mid);
if (line == NULL) {
/* error */
- (void)
mid->getfilecontent(mid->getfilecontentprivate, NULL, false, 0, NULL);
+ (void)
mid->getfilecontent(mid->filecontentprivate, NULL, false, 0, NULL);
return;
}
assert(line[0] == PROMPTBEG);
if (line[0] != PROMPTBEG) {
/* error */
- (void)
mid->getfilecontent(mid->getfilecontentprivate, NULL, false, 0, NULL);
+ (void)
mid->getfilecontent(mid->filecontentprivate, NULL, false, 0, NULL);
return;
}
if (line[1] == PROMPT3[1]) {
- (void)
mid->getfilecontent(mid->getfilecontentprivate, NULL, false, 0, NULL);
+ (void)
mid->getfilecontent(mid->filecontentprivate, NULL, false, 0, NULL);
(void) read_line(mid);
return;
}
assert(line[1] == PROMPT2[1]);
if (line[1] != PROMPT2[1]) {
/* error */
- (void)
mid->getfilecontent(mid->getfilecontentprivate, NULL, false, 0, NULL);
+ (void)
mid->getfilecontent(mid->filecontentprivate, NULL, false, 0, NULL);
return;
}
(void) read_line(mid);
@@ -3945,7 +3992,7 @@ handle_file(MapiHdl hdl, uint64_t off, c
return;
}
flushsize += size;
- data = mid->getfilecontent(mid->getfilecontentprivate, NULL,
false, 0, &size);
+ data = mid->getfilecontent(mid->filecontentprivate, NULL,
false, 0, &size);
}
mnstr_flush(mid->to);
line = read_line(mid);
@@ -4035,12 +4082,13 @@ read_into_cache(MapiHdl hdl, int lookahe
}
assert(*line == ' ');
line++; /* skip one space */
- handle_file(hdl, off, strdup(line),
binary);
+ read_file(hdl, off, strdup(line),
binary);
break;
}
case 'w':
- assert(0);
- /* not yet implemented */
+ line++; /* skip one space */
+ write_file(hdl, strdup(line));
+ break;
}
continue;
}
diff --git a/clients/mapilib/mapi.h b/clients/mapilib/mapi.h
--- a/clients/mapilib/mapi.h
+++ b/clients/mapilib/mapi.h
@@ -125,7 +125,13 @@ mapi_export char **mapi_resolve(const ch
mapi_export MapiMsg mapi_disconnect(Mapi mid);
mapi_export MapiMsg mapi_reconnect(Mapi mid);
mapi_export MapiMsg mapi_ping(Mapi mid);
-mapi_export void mapi_setfilecallback(Mapi mid, char *(*func)(void *priv,
const char *filename, bool binary, uint64_t offset, uint64_t *size), void
*priv);
+mapi_export void mapi_setfilecallback(
+ Mapi mid,
+ char *(*getfunc)(void *priv, const char *filename,
+ bool binary, uint64_t offset, size_t *size),
+ char *(*putfunc)(void *priv, const char *filename,
+ const void *restrict data, size_t size),
+ void *priv);
mapi_export MapiMsg mapi_error(Mapi mid);
mapi_export const char *mapi_error_str(Mapi mid);
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
@@ -192,14 +192,11 @@ getSQLContext(Client cntxt, MalBlkPtr mb
{
backend *be;
(void) mb;
-
- if (cntxt == NULL)
- throw(SQL, "mvc", SQLSTATE(42005) "No client record");
- if (cntxt->sqlcontext == NULL)
- throw(SQL, "mvc", SQLSTATE(42006) "SQL module not initialized");
+ str msg;
+
+ if ((msg = checkSQLContext(cntxt)) != MAL_SUCCEED)
+ return msg;
be = (backend *) cntxt->sqlcontext;
- if (be->mvc == NULL)
- throw(SQL, "mvc", SQLSTATE(42006) "SQL module not initialized,
mvc struct missing");
if (c)
*c = be->mvc;
if (b)
@@ -2031,7 +2028,7 @@ mvc_result_set_wrap( Client cntxt, MalBl
BBPunfix(bid);
}
/* now send it to the channel cntxt->fdout */
- if (mvc_export_result(cntxt->sqlcontext, cntxt->fdout, res,
mb->starttime, mb->optimize))
+ if (mvc_export_result(cntxt->sqlcontext, cntxt->fdout, res, true,
mb->starttime, mb->optimize))
msg = createException(SQL, "sql.resultset", SQLSTATE(45000)
"Result set construction failed");
mb->starttime = 0;
mb->optimize = 0;
@@ -2079,13 +2076,12 @@ mvc_export_table_wrap( Client cntxt, Mal
mvc *m = NULL;
BAT *order = NULL, *b = NULL, *tbl = NULL, *atr = NULL, *tpe =
NULL,*len = NULL,*scale = NULL;
res_table *t = NULL;
+ bool tostdout;
(void) format;
if ((msg = getSQLContext(cntxt, mb, &m, NULL)) != NULL)
return msg;
- if ((msg = checkSQLContext(cntxt)) != NULL)
- return msg;
bid = *getArgReference_bat(stk,pci,12);
order = BATdescriptor(bid);
@@ -2165,22 +2161,38 @@ mvc_export_table_wrap( Client cntxt, Mal
goto wrapup_result_set1;
/* now select the file channel */
- if ( strcmp(filename,"stdout") == 0 )
+ if ( strcmp(filename,"stdout") == 0 ) {
+ tostdout = true;
s= cntxt->fdout;
- else if ( (s = open_wastream(filename)) == NULL || mnstr_errnr(s)) {
- int errnr = mnstr_errnr(s);
- if (s)
- close_stream(s);
- msg= createException(IO, "streams.open", SQLSTATE(42000)
"could not open file '%s': %s",
- filename?filename:"stdout",
strerror(errnr));
- goto wrapup_result_set1;
+ } else {
+ tostdout = false;
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
