mapi.c

这个是内存数据库的客户端

		if (mid->blk.lim - mid->blk.end < BLOCK) {
			int len;

			len = mid->blk.lim;
			if (mid->blk.nxt <= BLOCK) {
				/* extend space */
				len += BLOCK;
			}
			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 = len;
		}

		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 = stream_read(mid->from, mid->blk.buf + mid->blk.end, 1, BLOCK);
		check_stream(mid, mid->from, "Connection terminated", "read_line", (mid->blk.eos = 1, (char *) 0));
		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) { /* add prompt */
			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;
			if (!nl)
				nl = mid->blk.buf + mid->blk.end + 1;
		}
		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;
}

/* set or unset the autocommit flag in the server */
MapiMsg
mapi_setAutocommit(Mapi mid, int autocommit)
{
	if (mid->auto_commit == autocommit)
		return MOK;
	if (mid->languageId != LANG_SQL) {
		mapi_setError(mid, "autocommit only supported in SQL", "mapi_setAutocommit", MERROR);
		return MERROR;
	}
	mid->auto_commit = autocommit;
	if (autocommit)
		return mapi_Xcommand(mid, "auto_commit", "1");
	else
		return mapi_Xcommand(mid, "auto_commit", "0");
}

MapiMsg
mapi_output(Mapi mid, char * output)
{
	mapi_clrError(mid);
	if (mid->languageId == LANG_XQUERY && strcmp(output,"dm")!=0)
		return mapi_Xcommand(mid, "output", output);
	return MOK;
}

MapiMsg
mapi_stream_into(Mapi mid, char *docname, char *colname)
{
	mapi_clrError(mid);
	if (mid->languageId == LANG_XQUERY) {
		char buf[BUFSIZ];
		int i;

		/* HACK alert, switch to mil too lose the extra 'S' */
		mid->languageId = LANG_MIL; 
		if (!colname)
			return mapi_Xcommand(mid, "copy", docname);
		i = snprintf(buf, BUFSIZ, "%s %s", docname, colname);
		if (i < 0)
			return MERROR;
		return mapi_Xcommand(mid, "copy", buf);
	}
	return MOK;
}

MapiMsg
mapi_profile(Mapi mid, int flag)
{
	mapi_clrError(mid);
	mid->profile = flag;
	if (mid->profile && mid->languageId == LANG_XQUERY)
		return mapi_Xcommand(mid, "profile", "");
	return MOK;
}

MapiMsg
mapi_trace(Mapi mid, int flag)
{
	mapi_clrError(mid);
	mid->trace = flag;
	return MOK;
}


static int
slice_row(const char *reply, char *null, char ***anchorsp, int length, int endchar)
{
	/* This function does the actual work for splicing a real,
	   multi-column row into columns.  It skips over the first
	   character and ends at the end of the string or at endchar,
	   whichever comes first. */
	char *start;
	char **anchors;
	int i;

	reply++;		/* skip over initial char (usually '[') */
	i = 0;
	anchors = length == 0 ? NULL : malloc(length * sizeof(*anchors));
	do {
		if (i >= length) {
			length = i + 1;
			REALLOC(anchors, length);
		}
		if (!unquote(reply, &start, &reply, endchar) && null && strcmp(start, null) == 0) {
			/* indicate NULL/nil with NULL pointer */
			free(start);
			start = NULL;
		}
		anchors[i++] = start;
		while (reply && *reply && isspace((int) (unsigned char) *reply))
			reply++;
	} while (reply && *reply && *reply != endchar);
	*anchorsp = anchors;
	return i;
}

static MapiMsg
mapi_cache_freeup_internal(struct MapiResultSet *result, int k)
{
	int i;			/* just a counter */
	int n = 0;		/* # of tuples being deleted from front */

	result->cache.tuplecount = 0;
	for (i = 0; i < result->cache.writer - k; i++) {
		if (result->cache.line[i].rows) {
			if (result->cache.line[i].rows[0] == '[' || result->cache.line[i].rows[0] == '=')
				n++;
			free(result->cache.line[i].rows);
		}
		result->cache.line[i].rows = result->cache.line[i + k].rows;
		result->cache.line[i + k].rows = 0;
		if (result->cache.line[i].anchors)
			free(result->cache.line[i].anchors);
		result->cache.line[i].anchors = result->cache.line[i + k].anchors;
		result->cache.line[i + k].anchors = 0;
		result->cache.line[i].fldcnt = result->cache.line[i + k].fldcnt;
		if (result->cache.line[i].rows &&
		    (result->cache.line[i].rows[0] == '[' ||
		     result->cache.line[i].rows[0] == '=')) {
			result->cache.line[i].tuplerev = result->cache.tuplecount;
			result->cache.line[result->cache.tuplecount++].tupleindex = i;
		}
	}
	/* after the previous loop, i == result->cache.writer - k, and
	   the last (result->cache.writer - k) cache entries have been
	   cleared already , so we don't need to go the Full Monty
	   here */
	for ( /*i = result->cache.writer - k */ ; i < k /*result->cache.writer */ ; i++) {
		if (result->cache.line[i].rows) {
			if (result->cache.line[i].rows[0] == '[' ||
			    result->cache.line[i].rows[0] == '=')
				n++;
			free(result->cache.line[i].rows);
		}
		result->cache.line[i].rows = 0;
		if (result->cache.line[i].anchors)
			free(result->cache.line[i].anchors);
		result->cache.line[i].anchors = 0;
		result->cache.line[i].fldcnt = 0;
	}
	result->cache.reader -= k;
	if (result->cache.reader < 0)
		result->cache.reader = -1;
	result->cache.writer -= k;
	if (result->cache.writer < 0)	/* "cannot happen" */
		result->cache.writer = 0;
	result->cache.first += n;

	return MOK;
}

static void
mapi_extend_cache(struct MapiResultSet *result)
{
	int incr, newsize, oldsize = result->cache.limit, i;

	/* extend row cache */
	if (oldsize == 0)
		incr = 100;
	else
		incr = oldsize * 2;
	if (incr > 200000)
		incr = 20000;
	newsize = oldsize + incr;
	if (result->cache.rowlimit > 0 && newsize > result->cache.rowlimit) {
		newsize = result->cache.rowlimit;
		incr = newsize - oldsize;
		if (incr <= 0) {
			if (result->cache.reader >= 0) {
				/* flush all read entries */
				mapi_cache_freeup_internal(result, result->cache.reader + 1);
				/* since we've made space, we can return */
				return;
			}
			/* not enough space, so increase limit */
			result->cache.rowlimit += 100;
			mapi_extend_cache(result);	/* try again */
			return;
		}
	}

	REALLOC(result->cache.line, newsize + 1);
	assert(result->cache.line);
	for (i = oldsize; i <= newsize; i++) {
		result->cache.line[i].fldcnt = 0;
		result->cache.line[i].rows = NULL;
		result->cache.line[i].tupleindex = -1;
		result->cache.line[i].tuplerev = -1;
		result->cache.line[i].anchors = NULL;
	}
	result->cache.limit = newsize;
}

/* store a line in the cache */
static void
add_cache(struct MapiResultSet *result, char *line)
{
	/* manage the row cache space first */
	if (result->cache.writer >= result->cache.limit)
		mapi_extend_cache(result);

	result->cache.line[result->cache.writer].rows = line;
	result->cache.line[result->cache.writer].tuplerev = result->cache.tuplecount;
	result->cache.line[result->cache.writer + 1].tuplerev = result->cache.tuplecount + 1;
	if (*line == '[' || *line == '=') {
		result->cache.line[result->cache.tuplecount++].tupleindex = result->cache.writer;
		if (result->row_count < result->cache.first + result->cache.tuplecount)
			result->row_count = result->cache.first + result->cache.tuplecount;
	}
	result->cache.writer++;
}

static struct MapiResultSet *
parse_header_line(MapiHdl hdl, char *line, struct MapiResultSet *result)
{
	char *tag, *etag;
	int i, n;
	char **anchors;

	if (line[0] == '&') {
		char *nline = line;
		int qt;

		/* handle fields &qt */

		nline++;	/* query type */
		qt = strtol(nline, &nline, 0);

		if (qt != Q_BLOCK || result == NULL)
			result = new_result(hdl);
		result->querytype = qt;

		nline++;		/* skip space */
		switch (qt) {
		case Q_TRANS:
			if (*nline == 'f')
				hdl->mid->auto_commit = 0;
			else
				hdl->mid->auto_commit = 1;
			break;
		case Q_UPDATE:
			result->row_count = strtol(nline, &nline, 0);
			break;
		case Q_TABLE:
		case Q_PREPARE: {
			int ntuples;	/* not used */

			sscanf(nline, "%d %d %d %d", &result->tableid, &result->row_count, &result->fieldcnt, &ntuples);
			break;
		}
		case Q_BLOCK:
			/* Mapi ignores the Q_BLOCK header, so spoof the querytype
			 * back to a Q_TABLE to let it go unnoticed */
			result->querytype = Q_TABLE;
			break;
		}


		if (result->fieldcnt > result->maxfields) {
			REALLOC(result->fields, result->fieldcnt);
			memset(result->fields + result->maxfields, 0, (result->fieldcnt - result->maxfields) * sizeof(*result->fields));
			result->maxfields = result->fieldcnt;
		}

		/* start of new SQL result */
		return result;
	}
	if (result == NULL)
		result = new_result(hdl);

	line = strdup(line);	/* make copy we can play with */

	etag = strrchr(line, '#');
	if (etag == 0 || etag == line) {
		/* not a useful header line */
		free(line);
		return result;
	}

	n = slice_row(line, NULL, &anchors, 10, '#');

	tag = etag + 1;
	while (*tag && isspace((int) (unsigned char) *tag))
		tag++;

	if (n > result->fieldcnt) {
		result->fieldcnt = n;
		if (n > result->maxfields) {
			REALLOC(result->fields, n);
			memset(result->fields + result->maxfields, 0, (n - result->maxfields) * sizeof(*result->fields));
			result->maxfields = n;
		}
	}

	if (strcmp(tag, "name") == 0) {
		for (i = 0; i < n; i++) {
			if (anchors[i]) {
				if (result->fields[i].columnname)
					free(result->fields[i].columnname);
				result->fields[i].columnname = anchors[i];
				anchors[i] = NULL;
			}
		}
	} else if (strcmp(tag, "type") == 0) {
		for (i = 0; i < n; i++) {
			if (anchors[i]) {
				if (result->fields[i].columntype)
					free(result->fields[i].columntype);
				result->fields[i].columntype = anchors[i];
				anchors[i] = NULL;
			}
		}
	} else if (strcmp(tag, "length") == 0) {
		for (i = 0; i < n; i++) {
			if (anchors[i])
				result->fields[i].columnlength = atoi(anchors[i]);
		}
	} else if (strcmp(tag, "table_name") == 0) {
		for (i = 0; i < n; i++) {
			if (anchors[i]) {
				if (result->fields[i].tablename)
					free(result->fields[i].tablename);
				result->fields[i].tablename = anchors[i];
				anchors[i] = NULL;
			}
		}
	}

	/* clean up */
	free(line);
	for (i = 0; i < n; i++)
		if (anchors[i])
			free(anchors[i]);
	free(anchors);

	return result;
}

/* Read ahead and cache data read.  Depending on the second argument,
   reading may stop at the first non-header and non-error line, or at
   a prompt.
   This function is called either after a command has been sent to the
   server (in which case the second argument is 1), when the
   application asks for a result tuple that hadn't been cached yet (in
   which case the second argument is also 1), or whenever all pending
   data needs to be read in order to send a new command to the server
   (in which case the second argument is 0).
   Header lines result tuples are stored in the cache.  Certain header
   lines may cause a new result set to be created in which case all
   subsequent lines are added to that result set.
*/
static MapiMsg
read_into_cache(MapiHdl hdl, int lookahead)
{
	char *line, *copy;
	Mapi mid;
	struct MapiResultSet *result;

	mid = hdl->mid;
	assert(mid->active == hdl);
	if (hdl->needmore) {
		hdl->needmore = 0;
		stream_flush(mid->to);
		check_stream(mid, mid->to, "write error on stream", "read_into_cache", mid->error);
	}
	if ((result = hdl->active) == NULL)
		result = hdl->result;	/* may also be NULL */
	for (;;) {
		line = read_line(mid);
		if (line == NULL)
			return mid->error;
		switch (*line) {
		case PROMPTBEG:
			mid->active = NULL;
			hdl->active = NULL;
			/* set needmore flag if line equals PROMPT2 up
			   to newline */
			copy = PROMPT2;
			while (*line) {
				if (*line != *copy)
					return mid->error;
				line++;
				copy++;
			}
			if (*copy == '\n' || *copy == 0) {
				/* skip end of block */
				mid->active = hdl;
				read_line(mid);
				hdl->needmore = 1;
				mid->active = hdl;
			}
			return mid->error;
		case '!':
			/* start a new result set if we don't have one
			   yet (obviously), or if we've already seen
			   normal output for the current one */
			if (result == NULL || result->cache.writer > 0) {
				result = new_result(hdl);
				hdl->active = result;
			}
			add_error(result, line);
			if (!mid->error)
				mid->error = MSERVER;
			break;
		case '%':
		case '#':
		case '&':
			if (lookahead < 0)
				lookahead = 1;
			result = parse_header_line(hdl, line, result);
			hdl->active = result;
			if (result && *line != '&')
				add_cache(result, strdup(line));
			break;
		default:
			if (result == NULL) {
				result = new_result(hdl);
				hdl->active = result;
			}
			add_cache(result, strdup(line));
			if (lookahead > 0 && (result->querytype == -1 /* unknown (not SQL) */  ||
					      result->querytype == Q_TABLE ||
					      result->querytype == Q_UPDATE))
				return mid->error;
			break;
		}
	}
}

MapiMsg
mapi_virtual_result(MapiHdl hdl, int columns, const char **columnnames, const char **columntypes, const int *columnlengths, int tuplecount, const char ***tuples)
{
	Mapi mid;
	struct MapiResultSet *result;
	int i, n;
	const char **tuple;
	char **anchors;

	if (columns <= 0)
		return MERROR;
	mid = hdl->mid;
	if (mid->active && read_into_cache(mid->active, 0) != MOK)
		return MERROR;
	assert(mid->active == NULL);
	finish_handle(hdl);
	assert(hdl->result == NULL);
	assert(hdl->active == NULL);
	hdl->active = result = new_result(hdl);
	result->fieldcnt = result->maxfields = columns;
	REALLOC(result->fields, columns);