Add catalog support (#297)

This adds support for multiple catalogs ("databases").
The client application can now set the catalog at connection level -
with SQLSetConnectAttr - to a pattern matching a local or remote cluster
name and ES/SQL will execute the query on the respective cluster/s.

A default catalog can also be set through the connection string, through
the "Catalog" DSN attribute.

In case the catalog is not set, or set to an empty or null string, the
local catalog will be chosen for execution.

The driver will simply add a "catalog" request attribute containing the
value received from the application, or skip adding the attribute if
this is empty.

Author: bpintea

driver/catalogue.c

@@ -87,35 +87,52 @@ SQLRETURN EsSQLStatisticsW(
 	return fake_answer(hstmt, &statistics);
 }
 
-BOOL TEST_API set_current_catalog(esodbc_dbc_st *dbc, wstr_st *catalog)
+void free_current_catalog(esodbc_dbc_st *dbc)
 {
-	if (dbc->catalog.cnt) {
-		DBGH(dbc, "catalog already set to `" LWPDL "`.", LWSTR(&dbc->catalog));
-		if (! EQ_WSTR(&dbc->catalog, catalog)) {
-			/* this should never happen, as cluster's name is not updateable
-			 * on the fly. */
-			ERRH(dbc, "overwriting previously set catalog value!");
-			free(dbc->catalog.str);
-			dbc->catalog.str = NULL;
-			dbc->catalog.cnt = 0;
-		} else {
-			return FALSE;
+	if (dbc->catalog.w.str) {
+		free(dbc->catalog.w.str);
+		dbc->catalog.w.str = NULL;
+		dbc->catalog.w.cnt = 0;
+	}
+	if (dbc->catalog.c.str) {
+		free(dbc->catalog.c.str);
+		dbc->catalog.c.str = NULL;
+		dbc->catalog.c.cnt = 0;
+	}
+}
+
+SQLRETURN set_current_catalog(esodbc_dbc_st *dbc, wstr_st *catalog)
+{
+	if (dbc->catalog.w.cnt) {
+		DBGH(dbc, "catalog previously set to `" LWPDL "`.",
+			LWSTR(&dbc->catalog.w));
+		if (! EQ_WSTR(&dbc->catalog.w, catalog)) {
+			free_current_catalog(dbc);
 		}
 	}
-	if (! catalog->cnt) {
-		WARNH(dbc, "attempting to set catalog name to empty value.");
-		return FALSE;
+	if (! catalog->cnt || ! catalog->str) {
+		WARNH(dbc, "catalog name set to empty value.");
+		return SQL_SUCCESS;
 	}
-	if (! (dbc->catalog.str = malloc((catalog->cnt + 1) * sizeof(SQLWCHAR)))) {
+	dbc->catalog.w.str = malloc((catalog->cnt + 1) * sizeof(SQLWCHAR));
+	if (! dbc->catalog.w.str) {
 		ERRNH(dbc, "OOM for %zu wchars.", catalog->cnt + 1);
-		return FALSE;
+		RET_HDIAGS(dbc, SQL_STATE_HY001);
 	}
-	wmemcpy(dbc->catalog.str, catalog->str, catalog->cnt);
-	dbc->catalog.str[catalog->cnt] = L'\0';
-	dbc->catalog.cnt = catalog->cnt;
-	INFOH(dbc, "current catalog name: `" LWPDL "`.", LWSTR(&dbc->catalog));
+	wmemcpy(dbc->catalog.w.str, catalog->str, catalog->cnt);
+	dbc->catalog.w.str[catalog->cnt] = L'\0';
+	dbc->catalog.w.cnt = catalog->cnt;
 
-	return TRUE;
+	if (! wstr_to_utf8(catalog, &dbc->catalog.c)) {
+		goto err;
+	}
+
+	INFOH(dbc, "current catalog name: `" LWPDL "`.", LWSTR(&dbc->catalog.w));
+	return SQL_SUCCESS;
+
+err:
+	free_current_catalog(dbc);
+	RET_HDIAG(dbc, SQL_STATE_HY000, "Saving current catalog failed", 0);
 }
 
 /* writes into 'dest', of size 'room', the current requested attr. of 'dbc'.
@@ -198,9 +215,6 @@ SQLSMALLINT fetch_server_attr(esodbc_dbc_st *dbc, SQLINTEGER attr_id,
 			/* 0-term room left out when binding */
 			buff[attr_val.cnt] = L'\0'; /* write_wstr() expects the 0-term */
 		}
-		if (attr_id == SQL_ATTR_CURRENT_CATALOG) {
-			set_current_catalog(dbc, &attr_val);
-		}
 	}
 	DBGH(dbc, "attribute %ld value: `" LWPDL "`.", attr_id, LWSTR(&attr_val));
 

driver/catalogue.h

@@ -35,7 +35,8 @@ enum {
 
 SQLSMALLINT fetch_server_attr(esodbc_dbc_st *dbc, SQLINTEGER attr_id,
 	SQLWCHAR *dest, SQLSMALLINT room);
-BOOL TEST_API set_current_catalog(esodbc_dbc_st *dbc, wstr_st *catalog);
+SQLRETURN set_current_catalog(esodbc_dbc_st *dbc, wstr_st *catalog);
+void free_current_catalog(esodbc_dbc_st *dbc);
 SQLRETURN TEST_API update_varchar_defs(esodbc_stmt_st *stmt);
 
 

driver/connect.c

@@ -1501,6 +1501,11 @@ SQLRETURN config_dbc(esodbc_dbc_st *dbc, esodbc_dsn_attrs_st *attrs)
 	/* early execution */
 	dbc->early_exec = wstr2bool(&attrs->early_exec);
 	INFOH(dbc, "early execution: %s.", dbc->early_exec ? "true" : "false");
+	/* default current catalog */
+	if (attrs->catalog.cnt &&
+			(! SQL_SUCCEEDED(set_current_catalog(dbc, &attrs->catalog)))) {
+		goto err;
+	}
 
 	/* how to print the floats? */
 	assert(1 <= attrs->sci_floats.cnt); /* default should apply */
@@ -1657,13 +1662,9 @@ void cleanup_dbc(esodbc_dbc_st *dbc)
 		dbc->srv_ver.str = NULL;
 		dbc->srv_ver.cnt = 0;
 	}
-	if (dbc->catalog.str) {
-		free(dbc->catalog.str);
-		dbc->catalog.str = NULL;
-		dbc->catalog.cnt = 0;
-	} else {
-		assert(dbc->catalog.cnt == 0);
-	}
+	free_current_catalog(dbc);
+	assert(dbc->catalog.w.cnt == 0);
+	assert(dbc->catalog.c.cnt == 0);
 	if (dbc->varchar_limit_str.str) {
 		free(dbc->varchar_limit_str.str);
 		dbc->varchar_limit_str.str = NULL;
@@ -3266,38 +3267,6 @@ SQLRETURN EsSQLDisconnect(SQLHDBC ConnectionHandle)
 	return SQL_SUCCESS;
 }
 
-/* ES/SQL doesn't support catalogs (yet). This function checks that a
- * previously retrieved (and cached) catalog value is the same with what the
- * app currently tries to set it to.
- * Ideally, the app provided value would be cached here too (as per the spec:
- * "SQL_ATTR_CURRENT_CATALOG can be set before or after connecting"), in case
- * there's no connection "established" yet and checked at "establishment"
- * time. But there's no client reported yet setting a catalog value before
- * connecting. */
-static SQLRETURN check_catalog_name(esodbc_dbc_st *dbc, SQLWCHAR *name,
-	SQLINTEGER len)
-{
-	wstr_st catalog;
-	catalog.str = name;
-	if (len < 0) {
-		catalog.cnt = wcslen(name);
-	} else {
-		catalog.cnt = ((size_t)len)/sizeof(SQLWCHAR);
-	}
-	if (! EQ_WSTR(&dbc->catalog, &catalog)) {
-		if (! dbc->catalog.cnt) {
-			/* this will happen if the app tries to set a value that it
-			 * discovered over a different connection.
-			 * TODO on a first reported issue. */
-			WARNH(dbc, "connection's current catalog not yet set!");
-		}
-		ERRH(dbc, "setting catalog name not supported.");
-		RET_HDIAGS(dbc, SQL_STATE_HYC00);
-	}
-	WARNH(dbc, "ignoring attempt to set the current catalog.");
-	return SQL_SUCCESS;
-}
-
 /*
  * https://docs.microsoft.com/en-us/sql/odbc/reference/develop-app/unicode-drivers :
  * """
@@ -3325,6 +3294,7 @@ SQLRETURN EsSQLSetConnectAttrW(
 	_In_reads_bytes_opt_(StringLength) SQLPOINTER Value,
 	SQLINTEGER StringLength)
 {
+	wstr_st catalog;
 	esodbc_dbc_st *dbc = DBCH(ConnectionHandle);
 
 	switch(Attribute) {
@@ -3434,11 +3404,13 @@ SQLRETURN EsSQLSetConnectAttrW(
 			RET_HDIAGS(dbc, SQL_STATE_IM009);
 
 		case SQL_ATTR_CURRENT_CATALOG:
-			INFOH(dbc, "setting current catalog to: `" LWPDL "`.",
-				/* string should be 0-term'd */
-				0 <= StringLength ? StringLength/sizeof(SQLWCHAR) : SHRT_MAX,
-				(SQLWCHAR *)Value);
-			return check_catalog_name(dbc, (SQLWCHAR *)Value, StringLength);
+			if (StringLength < 0) {
+				ERRH(dbc, "invalid catalog name lenght: %ld", StringLength);
+				RET_HDIAGS(dbc, SQL_STATE_HY090);
+			}
+			catalog.str = (SQLWCHAR *)Value;
+			catalog.cnt = StringLength/sizeof(SQLWCHAR);
+			return set_current_catalog(dbc, &catalog);
 
 		case SQL_ATTR_TRACE:
 		case SQL_ATTR_TRACEFILE: /* DM-only */
@@ -3504,7 +3476,16 @@ SQLRETURN EsSQLGetConnectAttrW(
 				ERRH(dbc, "no connection active.");
 				RET_HDIAGS(dbc, SQL_STATE_08003);
 			}
-			if ((used = fetch_server_attr(dbc, SQL_ATTR_CURRENT_CATALOG,
+			if (dbc->catalog.w.cnt) {
+				if (! SQL_SUCCEEDED(write_wstr(dbc, (SQLWCHAR *)ValuePtr,
+								&dbc->catalog.w, (SQLSMALLINT)BufferLength,
+								&used))) {
+					ERRH(dbc, "failed to copy current catalog out.");
+					RET_STATE(dbc->hdr.diag.state);
+				}
+				used = SHRT_MAX <= dbc->catalog.w.cnt ? SHRT_MAX :
+					(SQLSMALLINT)dbc->catalog.w.cnt;
+			} else if ((used = fetch_server_attr(dbc, SQL_ATTR_CURRENT_CATALOG,
 							(SQLWCHAR *)ValuePtr,
 							(SQLSMALLINT)BufferLength)) < 0) {
 				ERRH(dbc, "failed to get current catalog.");
@@ -3583,14 +3564,14 @@ SQLRETURN EsSQLGetConnectAttrW(
 		/* MS Access/Jet proprietary info type */
 		case 30002:
 			ERRH(dbc, "unsupported info type.");
-			RET_HDIAGS(DBCH(ConnectionHandle), SQL_STATE_HY092);
+			RET_HDIAGS(dbc, SQL_STATE_HY092);
 #endif
 
 		default:
 			ERRH(dbc, "unknown Attribute type %ld.", Attribute);
 			// FIXME: add the other attributes
 			FIXME;
-			RET_HDIAGS(DBCH(ConnectionHandle), SQL_STATE_HY092);
+			RET_HDIAGS(dbc, SQL_STATE_HY092);
 	}
 
 	return SQL_SUCCESS;

driver/handles.h

@@ -131,7 +131,6 @@ typedef struct struct_dbc {
 
 	wstr_st dsn; /* data source name SQLGetInfo(SQL_DATA_SOURCE_NAME) */
 	wstr_st server; /* ~ name; requested with SQLGetInfo(SQL_SERVER_NAME) */
-	wstr_st catalog; /* cached value; checked against if app setting it */
 	wstr_st srv_ver; /* server version: SQLGetInfo(SQL_DBMS_VER) */
 
 	cstr_st proxy_url;
@@ -166,6 +165,10 @@ typedef struct struct_dbc {
 		ESODBC_CMPSS_AUTO,
 	} compression;
 	BOOL apply_tz; /* should the times be converted from UTC to local TZ? */
+	struct {
+		wstr_st w; /* NB: w.str and c.str are co-allocated */
+		cstr_st c;
+	} catalog; /* current ~  */
 	BOOL early_exec; /* should prepared, non-param queries be exec'd early? */
 	enum {
 		ESODBC_FLTS_DEFAULT = 0,
@@ -559,7 +562,7 @@ SQLRETURN EsSQLSetDescRec(
 /* return the code associated with the given state (and debug-log) */
 #define RET_STATE(_s)	\
 	do { \
-		assert(_s < SQL_STATE_MAX); \
+		assert(SQL_STATE_00000 <= _s && _s < SQL_STATE_MAX); \
 		return esodbc_errors[_s].retcode; \
 	} while (0)
 

driver/queries.c

@@ -3052,28 +3052,40 @@ static SQLRETURN statement_len_cbor(esodbc_stmt_st *stmt, size_t *enc_len,
 		/* "field_multi_value_leniency": true/false */
 		bodylen += cbor_str_obj_len(sizeof(REQ_KEY_MULTIVAL) - 1);
 		bodylen += CBOR_OBJ_BOOL_LEN;
+		(*keys) ++;
 		/* "index_include_frozen": true/false */
 		bodylen += cbor_str_obj_len(sizeof(REQ_KEY_IDX_FROZEN) - 1);
 		bodylen += CBOR_OBJ_BOOL_LEN;
+		(*keys) ++;
 		/* "time_zone": "-05:45" */
 		bodylen += cbor_str_obj_len(sizeof(REQ_KEY_TIMEZONE) - 1);
 		bodylen += cbor_str_obj_len(tz_param.cnt); /* lax len */
+		(*keys) ++;
+		/* "catalog": "my_cluster" */
+		if (dbc->catalog.c.cnt) {
+			bodylen += cbor_str_obj_len(sizeof(REQ_KEY_CATALOG) - 1);
+			bodylen += cbor_str_obj_len(dbc->catalog.c.cnt);
+			(*keys) ++;
+		}
 		bodylen += cbor_str_obj_len(sizeof(REQ_KEY_VERSION) - 1);
 		bodylen += cbor_str_obj_len(version.cnt);
-		*keys += 4; /* field_m._val., idx._inc._frozen, time_zone, version */
+		(*keys) ++;
 	}
 	/* mode */
 	bodylen += cbor_str_obj_len(sizeof(REQ_KEY_MODE) - 1);
 	bodylen += cbor_str_obj_len(sizeof(REQ_VAL_MODE) - 1);
+	(*keys) ++;
 	/* client_id */
 	bodylen += cbor_str_obj_len(sizeof(REQ_KEY_CLT_ID) - 1);
 	bodylen += cbor_str_obj_len(sizeof(REQ_VAL_CLT_ID) - 1);
+	(*keys) ++;
 	/* binary_format */
 	bodylen += cbor_str_obj_len(sizeof(REQ_KEY_BINARY_FMT) - 1);
 	bodylen += CBOR_OBJ_BOOL_LEN;
-	*keys += 3; /* mode, client_id, binary_format */
+	(*keys) ++;
 	/* TODO: request_/page_timeout */
 
+	assert(*keys <= REST_REQ_KEY_COUNT);
 	*enc_len = bodylen;
 	return SQL_SUCCESS;
 }
@@ -3125,6 +3137,12 @@ static SQLRETURN statement_len_json(esodbc_stmt_st *stmt, size_t *outlen)
 		/* "time_zone": "-05:45" */
 		bodylen += sizeof(JSON_KEY_TIMEZONE) - 1;
 		bodylen += tz_param.cnt;
+		/* "catalog": "my_cluster" */
+		if (dbc->catalog.c.cnt) {
+			bodylen += sizeof(JSON_KEY_CATALOG) - 1;
+			bodylen += dbc->catalog.c.cnt;
+			bodylen += /* 2x `"` */2;
+		}
 		/* "version": */
 		bodylen += sizeof(JSON_KEY_VERSION) - 1;
 		bodylen += version.cnt + /* 2x`"` */2;
@@ -3407,6 +3425,14 @@ static SQLRETURN serialize_to_cbor(esodbc_stmt_st *stmt, cstr_st *dest,
 		}
 		err = cbor_encode_text_string(&map, tz.str, tz.cnt);
 		FAIL_ON_CBOR_ERR(stmt, err);
+		if (dbc->catalog.c.cnt) {
+			err = cbor_encode_text_string(&map, REQ_KEY_CATALOG,
+					sizeof(REQ_KEY_CATALOG) - 1);
+			FAIL_ON_CBOR_ERR(stmt, err);
+			err = cbor_encode_text_string(&map, dbc->catalog.c.str,
+					dbc->catalog.c.cnt);
+			FAIL_ON_CBOR_ERR(stmt, err);
+		}
 		/* version */
 		err = cbor_encode_text_string(&map, REQ_KEY_VERSION,
 				sizeof(REQ_KEY_VERSION) - 1);
@@ -3535,6 +3561,15 @@ static SQLRETURN serialize_to_json(esodbc_stmt_st *stmt, cstr_st *dest)
 				sizeof(JSON_VAL_TIMEZONE_Z) - 1);
 			pos += sizeof(JSON_VAL_TIMEZONE_Z) - 1;
 		}
+		if (dbc->catalog.c.cnt) {
+			/* "catalog": "my_cluster" */
+			memcpy(body + pos, JSON_KEY_CATALOG, sizeof(JSON_KEY_CATALOG) - 1);
+			pos += sizeof(JSON_KEY_CATALOG) - 1;
+			body[pos ++] = '"';
+			memcpy(body + pos, dbc->catalog.c.str, dbc->catalog.c.cnt);
+			pos += dbc->catalog.c.cnt;
+			body[pos ++] = '"';
+		}
 		/* "version": ... */
 		memcpy(body + pos, JSON_KEY_VERSION, sizeof(JSON_KEY_VERSION) - 1);
 		pos += sizeof(JSON_KEY_VERSION) - 1;

driver/queries.h

@@ -142,12 +142,15 @@ SQLRETURN EsSQLRowCount(_In_ SQLHSTMT StatementHandle, _Out_ SQLLEN *RowCount);
 #define REQ_KEY_MULTIVAL		"field_multi_value_leniency"
 #define REQ_KEY_IDX_FROZEN		"index_include_frozen"
 #define REQ_KEY_TIMEZONE		"time_zone"
+#define REQ_KEY_CATALOG			"catalog"
 #define REQ_KEY_BINARY_FMT		"binary_format"
+
+#define REST_REQ_KEY_COUNT		13 /* "query" / "cursor" count as one */
+
 /* keys for the "params" argument */
 #define REQ_KEY_PARAM_TYPE		"type"
 #define REQ_KEY_PARAM_VAL		"value"
 
-#define REST_REQ_KEY_COUNT		11 /* "query" or "cursor" */
 
 #ifdef _WIN64
 #	define REQ_VAL_CLT_ID		"odbc64"
@@ -172,6 +175,7 @@ SQLRETURN EsSQLRowCount(_In_ SQLHSTMT StatementHandle, _Out_ SQLLEN *RowCount);
 #define JSON_KEY_MULTIVAL		", \"" REQ_KEY_MULTIVAL "\": " /* n-th */
 #define JSON_KEY_IDX_FROZEN		", \"" REQ_KEY_IDX_FROZEN "\": " /* n-th */
 #define JSON_KEY_TIMEZONE		", \"" REQ_KEY_TIMEZONE "\": " /* n-th key */
+#define JSON_KEY_CATALOG		", \"" REQ_KEY_CATALOG "\": " /* n-th key */
 #define JSON_KEY_BINARY_FMT		", \"" REQ_KEY_BINARY_FMT "\": " /* n-th key */
 
 #define JSON_VAL_TIMEZONE_Z		"\"" REQ_VAL_TIMEZONE_Z "\""