Changeset: 8155970591cd for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=8155970591cd
Modified Files:
clients/mapilib/mapi.c
common/stream/stream.c
common/stream/stream.h
sql/backends/monet5/sql_result.c
sql/include/sql_query.h
Branch: protocol
Log Message:
new result set first part and mapi read_line cleanup
diffs (truncated from 430 to 300 lines):
diff --git a/clients/mapilib/mapi.c b/clients/mapilib/mapi.c
--- a/clients/mapilib/mapi.c
+++ b/clients/mapilib/mapi.c
@@ -876,9 +876,7 @@ struct MapiRowBuf {
struct BlockCache {
char *buf;
int lim;
- int nxt;
int end;
- int eos; /* end of sequence */
};
@@ -1894,18 +1892,12 @@ mapi_new(void)
mid->redircnt = 0;
mid->redirmax = 10;
mid->tracelog = NULL;
- mid->blk.eos = 0;
- mid->blk.buf = malloc(BLOCK + 1);
+ mid->blk.buf = malloc(BLOCK);
if (mid->blk.buf == NULL) {
mapi_destroy(mid);
return NULL;
}
- mid->blk.buf[BLOCK] = 0;
- mid->blk.buf[0] = 0;
- mid->blk.nxt = 0;
- mid->blk.end = 0;
mid->blk.lim = BLOCK;
-
mid->first = NULL;
return mid;
@@ -2558,7 +2550,7 @@ mapi_reconnect(Mapi mid)
/* consume server challenge */
len = mnstr_read_block(mid->from, buf, 1, BLOCK);
- check_stream(mid, mid->from, "Connection terminated while starting",
"mapi_reconnect", (mid->blk.eos = 1, mid->error));
+ check_stream(mid, mid->from, "Connection terminated while starting",
"mapi_reconnect", mid->error);
assert(len < BLOCK);
buf[len] = 0;
@@ -3491,82 +3483,32 @@ mapi_param_store(MapiHdl hdl)
static char *
read_line(Mapi mid)
{
- char *reply;
- char *nl;
- char *s; /* from where to search for newline */
-
+ int ret = 0;
if (mid->active == NULL)
return 0;
-
- /* check if we need to read more blocks to get a new line */
- mid->blk.eos = 0;
- s = mid->blk.buf + mid->blk.nxt;
- while ((nl = strchr(s, '\n')) == NULL && !mid->blk.eos) {
- ssize_t len;
-
- if (mid->blk.lim - mid->blk.end < BLOCK) {
- int len;
-
- len = mid->blk.lim;
- if (mid->blk.nxt <= BLOCK) {
- /* extend space */
- len += BLOCK;
+ mid->blk.end = 0;
+ do {
+ if ((mid->blk.end + 1) == mid->blk.lim) {
+ REALLOC(mid->blk.buf, mid->blk.lim + BLOCK);
+ if (!mid->blk.buf) {
+ return 0;
}
- REALLOC(mid->blk.buf, len + 1);
- if (mid->blk.nxt > 0) {
- memmove(mid->blk.buf, mid->blk.buf +
mid->blk.nxt, mid->blk.end - mid->blk.nxt + 1);
- mid->blk.end -= mid->blk.nxt;
- mid->blk.nxt = 0;
+ mid->blk.lim += BLOCK;
+ }
+ /* mid->from is **always** buffered, so no point in rolling an
additional cache on top */
+ if ((ret = mnstr_readChr(mid->from, mid->blk.buf +
mid->blk.end)) != 1) {
+ if (ret == 0) {
+ mid->blk.buf[0] = PROMPTBEG;
+ mid->blk.buf[1] = '\n';
+ mid->blk.buf[2] = 0;
+ return mid->blk.buf;
}
- mid->blk.lim = len;
+ return 0;
}
-
- s = mid->blk.buf + mid->blk.end;
-
- /* fetch one more block */
- if (mid->trace == MAPI_TRACE)
- printf("fetch next block: start at:%d\n", mid->blk.end);
- len = mnstr_read(mid->from, mid->blk.buf + mid->blk.end, 1,
BLOCK);
- check_stream(mid, mid->from, "Connection terminated during read
line", "read_line", (mid->blk.eos = 1, (char *) 0));
- if (mid->tracelog) {
- mapi_log_header(mid, "R");
- mnstr_write(mid->tracelog, mid->blk.buf + mid->blk.end,
1, len);
- mnstr_flush(mid->tracelog);
- }
- mid->blk.buf[mid->blk.end + len] = 0;
- if (mid->trace == MAPI_TRACE) {
- printf("got next block: length:" SSZFMT "\n", len);
- printf("text:%s\n", mid->blk.buf + mid->blk.end);
- }
- if (len == 0) { /* add prompt */
- if (mid->blk.end > mid->blk.nxt) {
- /* add fake newline since newline was
- * missing from server */
- nl = mid->blk.buf + mid->blk.end;
- *nl = '\n';
- mid->blk.end++;
- }
- len = 2;
- mid->blk.buf[mid->blk.end] = PROMPTBEG;
- mid->blk.buf[mid->blk.end + 1] = '\n';
- mid->blk.buf[mid->blk.end + 2] = 0;
- }
- mid->blk.end += (int) len;
- }
- if (mid->trace == MAPI_TRACE) {
- printf("got complete block: \n");
- printf("text:%s\n", mid->blk.buf + mid->blk.nxt);
- }
-
- /* we have a complete line in the buffer */
- assert(nl);
- *nl++ = 0;
- reply = mid->blk.buf + mid->blk.nxt;
- mid->blk.nxt = (int) (nl - mid->blk.buf);
-
- if (mid->trace == MAPI_TRACE)
- printf("read_line:%s\n", reply);
- return reply;
+ } while (mid->blk.buf[mid->blk.end++] != '\n');
+
+ mid->blk.buf[mid->blk.end-1] = 0;
+ return mid->blk.buf;
}
/* set or unset the autocommit flag in the server */
@@ -3976,6 +3918,58 @@ read_into_cache(MapiHdl hdl, int lookahe
if (line == NULL)
return mid->error;
switch (*line) {
+ case 42: {
+ int result_set_id;
+ lng nr_rows;
+ lng nr_cols;
+ lng i;
+ if (!mnstr_readInt(mid->from, &result_set_id) ||
+ !mnstr_readLng(mid->from, &nr_rows) ||
+ !mnstr_readLng(mid->from, &nr_cols)) {
+ return mid->error;
+ }
+ fprintf(stderr, "result_set_id=%d, nr_rows=%llu,
nr_cols=%lld\n", result_set_id, nr_rows, nr_cols);
+
+
+ for (i = 0; i < nr_cols; i++) {
+ lng col_info_length;
+ char *table_name, *col_name, *type_sql_name;
+ int typelen;
+
+ if (!mnstr_readLng(mid->from,
&col_info_length)) {
+ return mid->error;
+ }
+ // possible improvement, set col_info_length to
max length of the three strings
+ table_name = malloc(col_info_length);
+ col_name = malloc(col_info_length);
+ type_sql_name = malloc(col_info_length);
+ if (!table_name || !col_name || !type_sql_name)
{
+ return mid->error;
+ }
+
+ if (!mnstr_readStr(mid->from, table_name) ||
+ !mnstr_readStr(mid->from,
col_name) ||
+ !mnstr_readStr(mid->from,
type_sql_name) ||
+ !mnstr_readInt(mid->from,
&typelen)) {
+ return mid->error;
+ }
+ fprintf(stderr, "%lld col_info_length=%lld,
table_name=%s, col_name=%s, type_sql_name=%s, type_len=%d\n", i,
col_info_length, table_name, col_name, type_sql_name, typelen);
+ }
+
+ {
+ lng nrows = 0;
+ char dummy;
+ // we flush on the other side so this read will
always fail
+ mnstr_readChr(mid->from, &dummy);
+ if (!mnstr_readLng(mid->from, &nrows)) {
+ return mid->error;
+ }
+ fprintf(stderr, "nrows=%llu\n", nrows);
+ }
+
+
+ return 0;
+ }
case PROMPTBEG: /* \001 */
mid->active = NULL;
hdl->active = NULL;
diff --git a/common/stream/stream.c b/common/stream/stream.c
--- a/common/stream/stream.c
+++ b/common/stream/stream.c
@@ -4419,6 +4419,15 @@ mnstr_read_block(stream *s, void *buf, s
return len;
}
+
+int
+mnstr_readChr(stream *s, char *val)
+{
+ if (s == NULL || val == NULL)
+ return -1;
+ return (int) s->read(s, (void *) val, sizeof(*val), 1);
+}
+
int
mnstr_readBte(stream *s, signed char *val)
{
@@ -4487,6 +4496,29 @@ mnstr_writeInt(stream *s, int val)
}
int
+mnstr_writeStr(stream *s, const char* val)
+{
+ if (s == NULL || s->errnr)
+ return 0;
+ return s->write(s, (void *) val, strlen(val), (size_t) 1) == 1;
+}
+
+int
+mnstr_readStr(stream *s, char* val)
+{
+ if (s == NULL || s->errnr)
+ return 0;
+ do {
+ if (mnstr_readChr(s, val) != 1) {
+ return -1;
+ }
+ val++;
+ } while (*(val - 1) != '\0');
+ return 1;
+}
+
+
+int
mnstr_readLng(stream *s, lng *val)
{
if (s == NULL || val == NULL)
diff --git a/common/stream/stream.h b/common/stream/stream.h
--- a/common/stream/stream.h
+++ b/common/stream/stream.h
@@ -29,6 +29,7 @@
#include <signal.h>
#include <limits.h>
+#define STREAM_DEBUG 1
/* avoid using "#ifdef WIN32" so that this file does not need our config.h */
#if defined(_MSC_VER) || defined(__CYGWIN__) || defined(__MINGW32__)
# ifndef LIBSTREAM
@@ -103,6 +104,7 @@ stream_export int mnstr_init(void);
* !0 on success
*/
stream_export int mnstr_readBte(stream *s, signed char *val);
+stream_export int mnstr_readChr(stream *s, char *val);
stream_export int mnstr_writeBte(stream *s, signed char val);
stream_export int mnstr_readSht(stream *s, short *val);
stream_export int mnstr_writeSht(stream *s, short val);
@@ -118,6 +120,9 @@ stream_export int mnstr_writeHge(stream
stream_export int mnstr_readBteArray(stream *s, signed char *val, size_t cnt);
stream_export int mnstr_writeBteArray(stream *s, const signed char *val,
size_t cnt);
+stream_export int mnstr_writeStr(stream *s, const char *val);
+stream_export int mnstr_readStr(stream *s, char* val);
+
stream_export int mnstr_readShtArray(stream *s, short *val, size_t cnt);
stream_export int mnstr_writeShtArray(stream *s, const short *val, size_t cnt);
stream_export int mnstr_readIntArray(stream *s, int *val, size_t cnt);
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
@@ -1858,12 +1858,105 @@ mvc_export_file(backend *b, stream *s, r
return res;
}
-//static int mvc_export_resultset_prot10(res_table* table, stream*
outputstream, size_t blocksize) {
-// (void) table;
-// (void) outputstream;
-// (void) blocksize;
-// return 42;
-//}
+static int write_str_term(stream* s, str val) {
+ return mnstr_writeStr(s, val) && mnstr_writeBte(s, 0);
+}
+
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
