Changeset: 14e7b2fe4998 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/14e7b2fe4998
Modified Files:
common/stream/mapi_stream.c
common/stream/stream.h
common/stream/stream_internal.h
monetdb5/modules/mal/mal_mapi.c
sql/backends/monet5/sql_bincopy.c
sql/backends/monet5/sql_bincopyconvert.c
sql/backends/monet5/sql_bincopyconvert.h
sql/backends/monet5/sql_result.c
sql/backends/monet5/sql_result.h
sql/backends/monet5/sql_scenario.c
Branch: binresultset
Log Message:
Implement Xexportbin
diffs (truncated from 432 to 300 lines):
diff --git a/common/stream/mapi_stream.c b/common/stream/mapi_stream.c
--- a/common/stream/mapi_stream.c
+++ b/common/stream/mapi_stream.c
@@ -11,6 +11,30 @@
#include "stream_internal.h"
#include "mapi_prompt.h"
+static ssize_t byte_counting_write(stream *restrict s, const void *restrict
buf, size_t elmsize, size_t cnt)
+{
+ uint64_t *counter = (uint64_t*) s->stream_data.p;
+ ssize_t nwritten = s->inner->write(s->inner, buf, elmsize, cnt);
+ if (nwritten >= 0) {
+ *counter += elmsize * nwritten;
+ }
+ return nwritten;
+}
+
+
+stream *
+byte_counting_stream(stream *wrapped, uint64_t *counter)
+{
+ stream *s = create_wrapper_stream(NULL, wrapped);
+ if (!s)
+ return NULL;
+ s->stream_data.p = counter;
+ s->write = &byte_counting_write;
+ s->destroy = &destroy_stream;
+ return s;
+}
+
+
static void
discard(stream *s)
diff --git a/common/stream/stream.h b/common/stream/stream.h
--- a/common/stream/stream.h
+++ b/common/stream/stream.h
@@ -267,4 +267,8 @@ stream_export stream *create_text_stream
stream_export stream *mapi_request_upload(const char *filename, bool binary,
bstream *rs, stream *ws);
stream_export stream *mapi_request_download(const char *filename, bool binary,
bstream *rs, stream *ws);
+// write-only
+stream_export stream *byte_counting_stream(stream *wrapped, uint64_t *counter);
+
+
#endif /*_STREAM_H_*/
diff --git a/common/stream/stream_internal.h b/common/stream/stream_internal.h
--- a/common/stream/stream_internal.h
+++ b/common/stream/stream_internal.h
@@ -269,8 +269,8 @@ typedef struct bs bs;
struct bs {
unsigned nr; /* how far we got in buf */
unsigned itotal; /* amount available in current read block */
- size_t blks; /* read/writen blocks (possibly partial) */
- size_t bytes; /* read/writen bytes */
+ int64_t blks; /* read/writen blocks (possibly partial) */
+ int64_t bytes; /* read/writen bytes */
char buf[BLOCK]; /* the buffered data (minus the size of
* size-short */
};
diff --git a/monetdb5/modules/mal/mal_mapi.c b/monetdb5/modules/mal/mal_mapi.c
--- a/monetdb5/modules/mal/mal_mapi.c
+++ b/monetdb5/modules/mal/mal_mapi.c
@@ -185,7 +185,7 @@ doChallenge(void *data)
}
// Send the challenge over the block stream
- mnstr_printf(fdout, "%s:mserver:9:%s:%s:%s:sql=%d:",
+ mnstr_printf(fdout, "%s:mserver:9:%s:%s:%s:sql=%d:BINARY=1:",
challenge,
mcrypt_getHashAlgorithms(),
#ifdef WORDS_BIGENDIAN
diff --git a/sql/backends/monet5/sql_bincopy.c
b/sql/backends/monet5/sql_bincopy.c
--- a/sql/backends/monet5/sql_bincopy.c
+++ b/sql/backends/monet5/sql_bincopy.c
@@ -21,6 +21,12 @@
#include "copybinary_support.h"
+#define bailout(...) do { \
+ msg = createException(MAL, mal_operator, SQLSTATE(42000)
__VA_ARGS__); \
+ goto end; \
+ } while (0)
+
+
static str
load_trivial(BAT *bat, stream *s, BUN rows_estimate, int *eof_seen)
{
@@ -325,36 +331,40 @@ end:
}
static str
-dump_trivial(BAT *b, stream *s)
+dump_trivial(BAT *b, stream *s, BUN start, BUN length)
{
assert(!ATOMvarsized(BATttype(b)));
-
- return write_out(Tloc(b, 0), Tloc(b, BATcount(b)), s);
+ BUN end = start + length;
+ assert(end <= BATcount(b));
+ return write_out(Tloc(b, start), Tloc(b, end), s);
}
static str
-dump_fixed_width(BAT *b, stream *s, bool byteswap, bincopy_encoder_t encoder,
size_t record_size)
+dump_fixed_width(BAT *b, stream *s, BUN start, BUN length, bool byteswap,
bincopy_encoder_t encoder, size_t record_size)
{
const char *mal_operator = "sql.export_bin_column";
str msg = MAL_SUCCEED;
char *buffer = NULL;
+ BUN end = start + length;
+ assert(end <= BATcount(b));
+
if (record_size == 0) {
int tt = BATttype(b);
record_size = (size_t) ATOMsize(tt);
}
size_t buffer_size = 1024 * 1024;
BUN batch_size = buffer_size / record_size;
- if (batch_size > BATcount(b))
- batch_size = BATcount(b);
+ if (batch_size > length)
+ batch_size = length;
buffer_size = batch_size * record_size;
buffer = GDKmalloc(buffer_size);
if (buffer == NULL)
bailout(MAL_MALLOC_FAIL);
BUN n;
- for (BUN pos = 0; pos < BATcount(b); pos += n) {
- n = BATcount(b) - pos;
+ for (BUN pos = start; pos < end; pos += n) {
+ n = end - pos;
if (n > batch_size)
n = batch_size;
msg = encoder(buffer, Tloc(b, pos), n, 0, byteswap);
@@ -370,8 +380,8 @@ end:
return msg;
}
-static str
-dump_column(const struct type_record_t *rec, BAT *b, bool byteswap, stream *s)
+str
+dump_binary_column(const struct type_record_t *rec, BAT *b, BUN start, BUN
length, bool byteswap, stream *s)
{
str msg = MAL_SUCCEED;
@@ -386,11 +396,11 @@ dump_column(const struct type_record_t *
encoder = NULL;
if (dumper) {
- msg = rec->dumper(b, s, byteswap);
+ msg = rec->dumper(b, s, start, length, byteswap);
} else if (encoder) {
- msg = dump_fixed_width(b, s, byteswap, rec->encoder,
rec->record_size);
+ msg = dump_fixed_width(b, s, start, length, byteswap,
rec->encoder, rec->record_size);
} else {
- msg = dump_trivial(b, s);
+ msg = dump_trivial(b, s, start, length);
}
return msg;
@@ -421,7 +431,7 @@ export_column(backend *be, BAT *b, bool
bailout("%s", mnstr_peek_error(NULL));
}
- msg = dump_column(rec, b, byteswap, s);
+ msg = dump_binary_column(rec, b, 0, BATcount(b), byteswap, s);
if (s && msg == MAL_SUCCEED) {
if (mnstr_flush(s, MNSTR_FLUSH_DATA) != 0) {
diff --git a/sql/backends/monet5/sql_bincopyconvert.c
b/sql/backends/monet5/sql_bincopyconvert.c
--- a/sql/backends/monet5/sql_bincopyconvert.c
+++ b/sql/backends/monet5/sql_bincopyconvert.c
@@ -17,6 +17,13 @@
#include "mal_interpreter.h"
#include "mstring.h"
+
+#define bailout(...) do { \
+ msg = createException(MAL, mal_operator, SQLSTATE(42000)
__VA_ARGS__); \
+ goto end; \
+ } while (0)
+
+
static str
validate_bit(void *dst_, void *src_, size_t count, int width, bool byteswap)
{
@@ -361,7 +368,7 @@ end:
}
static str
-dump_zero_terminated_text(BAT *bat, stream *s, bool byteswap)
+dump_zero_terminated_text(BAT *bat, stream *s, BUN start, BUN length, bool
byteswap)
{
(void)byteswap;
const char *mal_operator = "sql.export_bin_column";
@@ -370,9 +377,11 @@ dump_zero_terminated_text(BAT *bat, stre
assert(ATOMstorage(tpe) == TYPE_str); (void)tpe;
assert(mnstr_isbinary(s));
- BUN end = BATcount(bat);
+
+ BUN end = start + length;
+ assert(end <= BATcount(bat));
BATiter bi = bat_iterator(bat);
- for (BUN p = 0; p < end; p++) {
+ for (BUN p = start; p < end; p++) {
const char *v = BUNtvar(bi, p);
if (mnstr_writeStr(s, v) < 0 || mnstr_writeBte(s, 0) < 0) {
bailout("%s", mnstr_peek_error(s));
@@ -425,3 +434,9 @@ find_type_rec(const char *name)
return t;
return NULL;
}
+
+bool
+can_dump_binary_column(const type_record_t *rec)
+{
+ return rec->encoder_trivial || rec->dumper || rec->encoder;
+}
diff --git a/sql/backends/monet5/sql_bincopyconvert.h
b/sql/backends/monet5/sql_bincopyconvert.h
--- a/sql/backends/monet5/sql_bincopyconvert.h
+++ b/sql/backends/monet5/sql_bincopyconvert.h
@@ -39,7 +39,7 @@ typedef str (*bincopy_decoder_t)(void *d
typedef str (*bincopy_loader_t)(BAT *bat, stream *s, int *eof_reached, int
width, bool byteswap);
typedef str (*bincopy_encoder_t)(void *dst, void *src, size_t count, int
width, bool byteswap);
-typedef str (*bincopy_dumper_t)(BAT *bat, stream *s, bool byteswap);
+typedef str (*bincopy_dumper_t)(BAT *bat, stream *s, BUN start, BUN length,
bool byteswap);
struct type_record_t {
char *method;
@@ -59,11 +59,9 @@ typedef const struct type_record_t type_
extern type_record_t *find_type_rec(const char *name);
+extern bool can_dump_binary_column(const type_record_t *rec);
-#define bailout(...) do { \
- msg = createException(MAL, mal_operator, SQLSTATE(42000)
__VA_ARGS__); \
- goto end; \
- } while (0)
+extern str dump_binary_column(const type_record_t *rec, BAT *b, BUN start, BUN
length, bool byteswap, stream *s);
#endif
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
@@ -18,6 +18,7 @@
#include "bat/res_table.h"
#include "bat/bat_storage.h"
#include "rel_exp.h"
+#include "sql_bincopyconvert.h"
#ifndef HAVE_LLABS
#define llabs(x) ((x) < 0 ? -(x) : (x))
@@ -1883,3 +1884,114 @@ mvc_export_error(backend *be, stream *s,
return "Unknown internal error";
}
}
+
+static ssize_t
+align_dump(stream *s, uint64_t *pos, unsigned int alignment)
+{
+ uint64_t a = (uint64_t)alignment;
+ // must be a power of two
+ assert(a > 0);
+ assert((a & (a-1)) == 0);
+
+ static char zeroes[32] = { 0 };
+ uint64_t gap = (-*pos) % a;
+ return mnstr_write(s, zeroes, 1, gap);
+}
+
+
+struct bindump_record {
+ BAT *bat;
+ type_record_t *type_rec;
+ int64_t start;
+ int64_t length;
+};
+
+int
+mvc_export_bin_chunk(backend *b, stream *s, int res_id, BUN offset, BUN nr)
+{
+ int ret = -42;
+ struct bindump_record *colinfo;
+ stream *countstream = NULL;
+ uint64_t byte_count = 0;
+ BUN end = offset + nr;
+
+ res_table *res = res_tables_find(b->results, res_id);
+ if (res == NULL)
+ return 0;
+
+ colinfo = GDKzalloc(res->nr_cols * sizeof(*colinfo));
+ if (!colinfo) {
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]
