Changeset: fe8b41f64d52 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=fe8b41f64d52
Modified Files:
NT/monetdb_config.h.in
clients/Tests/MAL-signatures.stable.out
clients/Tests/MAL-signatures.stable.out.int128
clients/Tests/exports.stable.out
clients/mapiclient/mclient.c
clients/mapilib/mapi.c
clients/mapilib/mapi.h
common/stream/stream.c
configure.ag
monetdb5/extras/mal_optimizer_template/Tests/opt_sql_append.stable.out
monetdb5/mal/mal_client.c
monetdb5/mal/mal_client.h
monetdb5/mal/mal_session.c
sql/ChangeLog
sql/backends/monet5/rel_bin.c
sql/backends/monet5/sql.c
sql/backends/monet5/sql.mal
sql/backends/monet5/sql_result.c
sql/backends/monet5/sql_result.h
sql/backends/monet5/sql_statement.c
sql/backends/monet5/sql_statement.h
sql/backends/monet5/vaults/bam/bam_db_interface.c
sql/benchmarks/tpch/load-sf-0.01-LOCKED.sql
sql/benchmarks/tpch/load-sf-0.01.sql
sql/benchmarks/tpch/load-sf-1.sql
sql/common/sql_types.c
sql/server/rel_updates.c
sql/server/sql_parser.y
sql/server/sql_scan.c
sql/test/BugTracker-2009/Tests/copy_multiple_files.SF-2902320.stable.out
sql/test/Tests/systemfunctions.stable.out
sql/test/Tests/systemfunctions.stable.out.int128
sql/test/Users/Tests/copyinto.stable.err
sql/test/Users/Tests/copyinto.stable.out
sql/test/emptydb-upgrade-chain/Tests/upgrade.stable.out.int128
sql/test/emptydb-upgrade/Tests/upgrade.stable.out.int128
sql/test/emptydb/Tests/check.stable.out
sql/test/emptydb/Tests/check.stable.out.32bit
sql/test/emptydb/Tests/check.stable.out.int128
sql/test/testdb-upgrade-chain/Tests/upgrade.stable.out.int128
sql/test/testdb-upgrade/Tests/upgrade.stable.out.int128
sql/test/testdb-upgrade/Tests/upgrade.stable.out.powerpc64.int128
testing/exportutils.py
Branch: default
Log Message:
Merged branch client-filetrans into default.
diffs (truncated from 3448 to 300 lines):
diff --git a/NT/monetdb_config.h.in b/NT/monetdb_config.h.in
--- a/NT/monetdb_config.h.in
+++ b/NT/monetdb_config.h.in
@@ -910,5 +910,6 @@ typedef __uint128_t uhge;
#define PROMPT1 "\001\001\n" /* prompt: ready for new query
*/
#define PROMPT2 "\001\002\n" /* prompt: more data needed */
+#define PROMPT3 "\001\003\n" /* prompt: get file content */
#endif /* _SEEN_MONETDB_CONFIG_H */
diff --git a/clients/Tests/MAL-signatures.stable.out
b/clients/Tests/MAL-signatures.stable.out
--- a/clients/Tests/MAL-signatures.stable.out
+++ b/clients/Tests/MAL-signatures.stable.out
@@ -10929,7 +10929,7 @@ Ready.
[ "sql", "bind_idxbat", "pattern sql.bind_idxbat(mvc:int, schema:str,
table:str, index:str, access:int, part_nr:int, nr_parts:int):bat[:any_1] ",
"mvc_bind_idxbat_wrap;", "Bind the 'schema.table.index' BAT with
access kind:\n\t0 - base table\n\t1 - inserts\n\t2 - updates" ]
[ "sql", "clear_table", "pattern sql.clear_table(sname:str,
tname:str):lng ", "mvc_clear_table_wrap;", "Clear the table
sname.tname." ]
[ "sql", "commit", "pattern sql.commit():void ", "SQLcommit;",
"Trigger the commit operation for a MAL block" ]
-[ "sql", "copy_from", "pattern sql.copy_from(t:ptr, sep:str,
rsep:str, ssep:str, ns:str, fname:str, nr:lng, offset:lng, locked:int,
best:int, fwf:str):bat[:any]... ", "mvc_import_table_wrap;",
"Import a table from bstream s with the \n\tgiven tuple and seperators
(sep/rsep)" ]
+[ "sql", "copy_from", "pattern sql.copy_from(t:ptr, sep:str,
rsep:str, ssep:str, ns:str, fname:str, nr:lng, offset:lng, locked:int,
best:int, fwf:str, onclient:int):bat[:any]... ", "mvc_import_table_wrap;",
"Import a table from bstream s with the \n\tgiven tuple and seperators
(sep/rsep)" ]
[ "sql", "copy_rejects", "pattern sql.copy_rejects() (rowid:bat[:lng],
fldid:bat[:int], msg:bat[:str], inp:bat[:str]) ", "COPYrejects;", "" ]
[ "sql", "copy_rejects_clear", "pattern sql.copy_rejects_clear():void
", "COPYrejects_clear;", "" ]
[ "sql", "createorderindex", "pattern sql.createorderindex(sch:str,
tbl:str, col:str):void ", "sql_createorderindex;", "Instantiate
the order index on a column" ]
@@ -10970,14 +10970,14 @@ Ready.
[ "sql", "exportHead", "pattern sql.exportHead(s:streams,
res_id:int):void ", "mvc_export_head_wrap;", "Export a result (in
order) to stream s" ]
[ "sql", "exportOperation", "pattern sql.exportOperation():void ",
"mvc_export_operation_wrap;", "Export result of schema/transaction queries"
]
[ "sql", "exportResult", "pattern sql.exportResult(s:streams,
res_id:int):void ", "mvc_export_result_wrap;", "Export a result (in
order) to stream s" ]
-[ "sql", "export_table", "pattern sql.export_table(fname:str, fmt:str,
colsep:str, recsep:str, qout:str, nullrep:str, tbl:bat[:str], attr:bat[:str],
tpe:bat[:str], len:bat[:int], scale:bat[:int], cols:any...):int ",
"mvc_export_row_wrap;", "Prepare a table result set for the COPY INTO stream"
]
-[ "sql", "export_table", "pattern sql.export_table(fname:str, fmt:str,
colsep:str, recsep:str, qout:str, nullrep:str, tbl:bat[:str], attr:bat[:str],
tpe:bat[:str], len:bat[:int], scale:bat[:int], cols:bat[:any]...):int ",
"mvc_export_table_wrap;", "Prepare a table result set for the COPY INTO
stream" ]
+[ "sql", "export_table", "pattern sql.export_table(fname:str, fmt:str,
colsep:str, recsep:str, qout:str, nullrep:str, onclient:int, tbl:bat[:str],
attr:bat[:str], tpe:bat[:str], len:bat[:int], scale:bat[:int], cols:any...):int
", "mvc_export_row_wrap;", "Prepare a table result set for the COPY INTO
stream" ]
+[ "sql", "export_table", "pattern sql.export_table(fname:str, fmt:str,
colsep:str, recsep:str, qout:str, nullrep:str, onclient:int, tbl:bat[:str],
attr:bat[:str], tpe:bat[:str], len:bat[:int], scale:bat[:int],
cols:bat[:any]...):int ", "mvc_export_table_wrap;", "Prepare a table
result set for the COPY INTO stream" ]
[ "sql", "flush_log", "command sql.flush_log():void ",
"SQLflush_log;", "flush the log now" ]
[ "sql", "getVariable", "pattern sql.getVariable(mvc:int,
varname:str):any_1 ", "getVariable;", "Get the value of a session variable" ]
[ "sql", "getVersion", "command sql.getVersion(clientid:int):lng ",
"mvc_getVersion;", "Return the database version identifier for a client."
]
[ "sql", "get_value", "pattern sql.get_value(sname:str,
sequence:str):lng ", "mvc_get_value;", "return the current value of the
sequence" ]
[ "sql", "grow", "pattern sql.grow(tid:bat[:oid], X_0:any_1):int ",
"mvc_grow_wrap;", "Resize the tid column of a declared table." ]
-[ "sql", "importTable", "pattern sql.importTable(sname:str, tname:str,
fname:str...):bat[:any]... ", "mvc_bin_import_table_wrap;", "Import a
table from the files (fname)" ]
+[ "sql", "importTable", "pattern sql.importTable(sname:str, tname:str,
onclient:int, fname:str...):bat[:any]... ", "mvc_bin_import_table_wrap;",
"Import a table from the files (fname)" ]
[ "sql", "include", "pattern sql.include(fname:str):void ",
"SQLinclude;", "Compile and execute a sql statements on the file" ]
[ "sql", "init", "pattern sql.init():void ", "SQLinitEnvironment;",
"Initialize the environment for MAL" ]
[ "sql", "logfile", "pattern sql.logfile(filename:str):void ",
"mvc_logfile;", "Enable/disable saving the sql statement traces" ]
diff --git a/clients/Tests/MAL-signatures.stable.out.int128
b/clients/Tests/MAL-signatures.stable.out.int128
--- a/clients/Tests/MAL-signatures.stable.out.int128
+++ b/clients/Tests/MAL-signatures.stable.out.int128
@@ -15144,7 +15144,7 @@ Ready.
[ "sql", "bind_idxbat", "pattern sql.bind_idxbat(mvc:int, schema:str,
table:str, index:str, access:int, part_nr:int, nr_parts:int):bat[:any_1] ",
"mvc_bind_idxbat_wrap;", "Bind the 'schema.table.index' BAT with
access kind:\n\t0 - base table\n\t1 - inserts\n\t2 - updates" ]
[ "sql", "clear_table", "pattern sql.clear_table(sname:str,
tname:str):lng ", "mvc_clear_table_wrap;", "Clear the table
sname.tname." ]
[ "sql", "commit", "pattern sql.commit():void ", "SQLcommit;",
"Trigger the commit operation for a MAL block" ]
-[ "sql", "copy_from", "pattern sql.copy_from(t:ptr, sep:str,
rsep:str, ssep:str, ns:str, fname:str, nr:lng, offset:lng, locked:int,
best:int, fwf:str):bat[:any]... ", "mvc_import_table_wrap;",
"Import a table from bstream s with the \n\tgiven tuple and seperators
(sep/rsep)" ]
+[ "sql", "copy_from", "pattern sql.copy_from(t:ptr, sep:str,
rsep:str, ssep:str, ns:str, fname:str, nr:lng, offset:lng, locked:int,
best:int, fwf:str, onclient:int):bat[:any]... ", "mvc_import_table_wrap;",
"Import a table from bstream s with the \n\tgiven tuple and seperators
(sep/rsep)" ]
[ "sql", "copy_rejects", "pattern sql.copy_rejects() (rowid:bat[:lng],
fldid:bat[:int], msg:bat[:str], inp:bat[:str]) ", "COPYrejects;", "" ]
[ "sql", "copy_rejects_clear", "pattern sql.copy_rejects_clear():void
", "COPYrejects_clear;", "" ]
[ "sql", "createorderindex", "pattern sql.createorderindex(sch:str,
tbl:str, col:str):void ", "sql_createorderindex;", "Instantiate
the order index on a column" ]
@@ -15186,14 +15186,14 @@ Ready.
[ "sql", "exportHead", "pattern sql.exportHead(s:streams,
res_id:int):void ", "mvc_export_head_wrap;", "Export a result (in
order) to stream s" ]
[ "sql", "exportOperation", "pattern sql.exportOperation():void ",
"mvc_export_operation_wrap;", "Export result of schema/transaction queries"
]
[ "sql", "exportResult", "pattern sql.exportResult(s:streams,
res_id:int):void ", "mvc_export_result_wrap;", "Export a result (in
order) to stream s" ]
-[ "sql", "export_table", "pattern sql.export_table(fname:str, fmt:str,
colsep:str, recsep:str, qout:str, nullrep:str, tbl:bat[:str], attr:bat[:str],
tpe:bat[:str], len:bat[:int], scale:bat[:int], cols:any...):int ",
"mvc_export_row_wrap;", "Prepare a table result set for the COPY INTO stream"
]
-[ "sql", "export_table", "pattern sql.export_table(fname:str, fmt:str,
colsep:str, recsep:str, qout:str, nullrep:str, tbl:bat[:str], attr:bat[:str],
tpe:bat[:str], len:bat[:int], scale:bat[:int], cols:bat[:any]...):int ",
"mvc_export_table_wrap;", "Prepare a table result set for the COPY INTO
stream" ]
+[ "sql", "export_table", "pattern sql.export_table(fname:str, fmt:str,
colsep:str, recsep:str, qout:str, nullrep:str, onclient:int, tbl:bat[:str],
attr:bat[:str], tpe:bat[:str], len:bat[:int], scale:bat[:int], cols:any...):int
", "mvc_export_row_wrap;", "Prepare a table result set for the COPY INTO
stream" ]
+[ "sql", "export_table", "pattern sql.export_table(fname:str, fmt:str,
colsep:str, recsep:str, qout:str, nullrep:str, onclient:int, tbl:bat[:str],
attr:bat[:str], tpe:bat[:str], len:bat[:int], scale:bat[:int],
cols:bat[:any]...):int ", "mvc_export_table_wrap;", "Prepare a table
result set for the COPY INTO stream" ]
[ "sql", "flush_log", "command sql.flush_log():void ",
"SQLflush_log;", "flush the log now" ]
[ "sql", "getVariable", "pattern sql.getVariable(mvc:int,
varname:str):any_1 ", "getVariable;", "Get the value of a session variable" ]
[ "sql", "getVersion", "command sql.getVersion(clientid:int):lng ",
"mvc_getVersion;", "Return the database version identifier for a client."
]
[ "sql", "get_value", "pattern sql.get_value(sname:str,
sequence:str):lng ", "mvc_get_value;", "return the current value of the
sequence" ]
[ "sql", "grow", "pattern sql.grow(tid:bat[:oid], X_0:any_1):int ",
"mvc_grow_wrap;", "Resize the tid column of a declared table." ]
-[ "sql", "importTable", "pattern sql.importTable(sname:str, tname:str,
fname:str...):bat[:any]... ", "mvc_bin_import_table_wrap;", "Import a
table from the files (fname)" ]
+[ "sql", "importTable", "pattern sql.importTable(sname:str, tname:str,
onclient:int, fname:str...):bat[:any]... ", "mvc_bin_import_table_wrap;",
"Import a table from the files (fname)" ]
[ "sql", "include", "pattern sql.include(fname:str):void ",
"SQLinclude;", "Compile and execute a sql statements on the file" ]
[ "sql", "init", "pattern sql.init():void ", "SQLinitEnvironment;",
"Initialize the environment for MAL" ]
[ "sql", "logfile", "pattern sql.logfile(filename:str):void ",
"mvc_logfile;", "Enable/disable saving the sql statement traces" ]
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
@@ -290,7 +290,7 @@ size_t HEAPvmsize(Heap *h);
void IMPSdestroy(BAT *b);
lng IMPSimprintsize(BAT *b);
int MT_check_nr_cores(void);
-int MT_create_thread(MT_Id *t, void( *function)(void *), void *arg, enum
MT_thr_detach d);
+int MT_create_thread(MT_Id *t, void (*function)(void *), void *arg, enum
MT_thr_detach d);
void MT_exiting_thread(void);
MT_Id MT_getpid(void);
size_t MT_getrss(void);
@@ -562,6 +562,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 *(*getfunc)(void *priv, const char
*filename, bool binary, uint64_t offset, size_t *size), char *(*putfunc)(void
*priv, const char *filename, const void *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);
@@ -2717,7 +2718,7 @@ char *buffer_get_buf(buffer *b);
void buffer_init(buffer *restrict b, char *restrict buf, size_t size);
stream *buffer_rastream(buffer *restrict b, const char *restrict name);
stream *buffer_wastream(buffer *restrict b, const char *restrict name);
-stream *callback_stream(void *restrict priv, ssize_t( *read)(void *restrict
priv, void *restrict buf, size_t elmsize, size_t cnt), void( *close)(void
*priv), void( *destroy)(void *priv), const char *restrict name);
+stream *callback_stream(void *restrict priv, ssize_t (*read)(void *restrict
priv, void *restrict buf, size_t elmsize, size_t cnt), void (*close)(void
*priv), void (*destroy)(void *priv), const char *restrict name);
void close_stream(stream *s);
stream *file_rastream(FILE *restrict fp, const char *restrict name);
stream *file_rstream(FILE *restrict fp, const char *restrict name);
@@ -2761,7 +2762,7 @@ int mnstr_readStr(stream *restrict s, ch
ssize_t mnstr_read_block(stream *restrict s, void *restrict buf, size_t
elmsize, size_t cnt);
ssize_t mnstr_readline(stream *restrict s, void *restrict buf, size_t maxcnt);
void mnstr_set_bigendian(stream *s, bool bigendian);
-void mnstr_settimeout(stream *s, unsigned int ms, bool( *func)(void));
+void mnstr_settimeout(stream *s, unsigned int ms, bool (*func)(void));
ssize_t mnstr_write(stream *restrict s, const void *restrict buf, size_t
elmsize, size_t cnt);
int mnstr_writeBte(stream *s, int8_t val);
int mnstr_writeBteArray(stream *restrict s, const int8_t *restrict val, size_t
cnt);
diff --git a/clients/mapiclient/mclient.c b/clients/mapiclient/mclient.c
--- a/clients/mapiclient/mclient.c
+++ b/clients/mapiclient/mclient.c
@@ -3027,6 +3027,123 @@ set_timezone(Mapi mid)
mapi_close_handle(hdl);
}
+struct privdata {
+ stream *f;
+ char *buf;
+};
+
+#define READSIZE (1 << 16)
+//#define READSIZE (1 << 20)
+
+static char *
+getfile(void *data, const char *filename, bool binary,
+ uint64_t offset, size_t *size)
+{
+ stream *f;
+ char *buf;
+ struct privdata *priv = data;
+ ssize_t s;
+
+ if (priv->buf == NULL) {
+ priv->buf = malloc(READSIZE);
+ if (priv->buf == NULL) {
+ *size = 0;
+ return "allocation failed";
+ }
+ }
+ buf = priv->buf;
+ if (filename != NULL) {
+ if (binary) {
+ f = open_rstream(filename);
+ assert(offset <= 1);
+ offset = 0;
+ } else {
+ f = open_rastream(filename);
+#ifdef HAVE_ICONV
+ if (encoding) {
+ stream *tmpf = f;
+ f = iconv_rstream(f, encoding, mnstr_name(f));
+ if (f == NULL)
+ close_stream(tmpf);
+ }
+#endif
+ }
+ if (f == NULL) {
+ *size = 0; /* indicate error */
+ return "cannot open file";
+ }
+ while (offset > 1) {
+ s = mnstr_readline(f, buf, READSIZE);
+ if (s < 0) {
+ close_stream(f);
+ *size = 0;
+ return "error reading file";
+ }
+ if (s == 0) {
+ /* reached EOF withing offset lines */
+ close_stream(f);
+ *size = 0;
+ return NULL;
+ }
+ if (buf[s - 1] == '\n')
+ offset--;
+ }
+ priv->f = f;
+ } else {
+ f = priv->f;
+ if (size == NULL) {
+ /* done reading before reaching EOF */
+ close_stream(f);
+ priv->f = NULL;
+ return NULL;
+ }
+ }
+ s = mnstr_read(f, buf, 1, READSIZE);
+ if (s <= 0) {
+ *size = 0;
+ close_stream(f);
+ priv->f = NULL;
+ return s < 0 ? "error reading file" : NULL;
+ }
+ *size = (size_t) s;
+ return buf;
+}
+
+static char *
+putfile(void *data, const char *filename, const void *buf, size_t bufsize)
+{
+ struct privdata *priv = data;
+
+ if (filename != NULL) {
+ if ((priv->f = open_wastream(filename)) == NULL)
+ return "cannot open file";
+#ifdef HAVE_ICONV
+ if (encoding) {
+ stream *f = priv->f;
+ priv->f = iconv_wstream(f, encoding, mnstr_name(f));
+ if (priv->f == NULL) {
+ close_stream(f);
+ return "cannot open file";
+ }
+ }
+#endif
+ if (buf == NULL || bufsize == 0)
+ return NULL; /* successfully opened file */
+ } else if (buf == NULL) {
+ /* done writing */
+ int flush = mnstr_flush(priv->f);
+ close_stream(priv->f);
+ priv->f = NULL;
+ return flush < 0 ? "error writing output" : 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__));
@@ -3429,6 +3546,10 @@ main(int argc, char **argv)
}
}
+ struct privdata priv;
+ priv = (struct privdata) {0};
+ 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,6 +839,9 @@ struct MapiStruct {
stream *tracelog; /* keep a log for inspection */
stream *from, *to;
uint32_t index; /* to mark the log records */
+ void *filecontentprivate;
+ char *(*getfilecontent)(void *, const char *, bool, uint64_t, size_t *);
+ char *(*putfilecontent)(void *, const char *, const void *, size_t);
};
struct MapiResultSet {
@@ -2679,7 +2682,7 @@ mapi_reconnect(Mapi mid)
/* note: if we make the database field an empty string, it
* means we want the default. However, it *should* be there. */
- if (snprintf(buf, BLOCK, "%s:%s:%s:%s:%s:\n",
+ if (snprintf(buf, BLOCK, "%s:%s:%s:%s:%s:FILETRANS:\n",
#ifdef WORDS_BIGENDIAN
"BIG",
#else
@@ -2972,6 +2975,99 @@ mapi_disconnect(Mapi mid)
return MOK;
}
+/* Set callback function to retrieve or send file content for COPY
+ * INTO queries.
+ *
+ * char *getfile(void *private, const char *filename, bool binary,
+ * uint64_6 offset, size_t *size);
+ * Retrieve data from a file.
+ *
+ * The arguments are:
+ * private - the value of the filecontentprivate argument to
+ * mapi_setfilecallback;
+ * filename - the file to read (the application is free to interpret
+ * this any way it wants, including getting data over the
+ * Internet);
+ * binary - if set, the file is expected to contain binary data and
+ * should therefore be opened in binary mode, otherwise the
+ * file is expected to contain data in the UTF-8 encoding (of
+ * course, the application is free to transparently convert
+ * from the actual encoding to UTF-8);
+ * offset - the line number of the first line to be retrieved (this is
+ * one-based, i.e. the start of the file has line number one;
+ * lines are terminated by '\n');
+ * size - pointer in which to return the size of the chunk that is
+ * being returned.
+ *
+ * The callback function is expected to return data in chunks until it
+ * indicates to the caller that there is no more data or an error has
+ * occurred. Chunks can be any size. The caller does not modify or
+ * free the data returned. The size of the chunk being returned is
_______________________________________________
checkin-list mailing list
checkin-list@monetdb.org
https://www.monetdb.org/mailman/listinfo/checkin-list
