hcq/MaxScale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch '2.1-oracle-compat' into develop-new-merge-oracle

Browse Source
This commit is contained in:
Johan Wikman
2017-06-28 22:30:16 +02:00
parent 14bb6cf99b 7f1a949198
commit 6cd6ded3d8
88 changed files with 15138 additions and 745 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
server/core/config.cc
View File

@ -124,6 +124,7 @@ const char CN_SERVICE[] = "service";
const char CN_SESSIONS[] = "sessions";
const char CN_SKIP_PERMISSION_CHECKS[] = "skip_permission_checks";
const char CN_SOCKET[] = "socket";
const char CN_SQL_MODE[] = "sql_mode";
const char CN_STATE[] = "state";
const char CN_STATUS[] = "status";
const char CN_SSL[] = "ssl";
@ -1521,6 +1522,22 @@ handle_global_item(const char *name, const char *value)
{
gateway.qc_args = MXS_STRDUP_A(value);
}
else if (strcmp(name, "sql_mode") == 0)
{
if (strcasecmp(value, "default") == 0)
{
gateway.qc_sql_mode = QC_SQL_MODE_DEFAULT;
}
else if (strcasecmp(value, "oracle") == 0)
{
gateway.qc_sql_mode = QC_SQL_MODE_ORACLE;
}
else
{
MXS_ERROR("'%s' is not a valid value for '%s'. Allowed values are 'DEFAULT' and "
"'ORACLE'. Using 'DEFAULT' as default.", value, name);
}
}
else if (strcmp(name, CN_LOG_THROTTLING) == 0)
{
if (*value == 0)
@ -1882,6 +1899,8 @@ global_defaults()
/* query_classifier */
memset(gateway.qc_name, 0, sizeof(gateway.qc_name));
gateway.qc_args = NULL;
gateway.qc_sql_mode = QC_SQL_MODE_DEFAULT;
}
/**

9
server/core/dcb.cc
View File

@ -2600,6 +2600,15 @@ int dcb_listen(DCB *listener, const char *config, const char *protocol_name)
else if (port > 0)
{
listener_socket = dcb_listen_create_socket_inet(host, port);
if (listener_socket == -1 && strcmp(host, "::") == 0)
{
/** Attempt to bind to the IPv4 if the default IPv6 one is used */
MXS_WARNING("Failed to bind on default IPv6 host '::', attempting "
"to bind on IPv4 version '0.0.0.0'");
strcpy(host, "0.0.0.0");
listener_socket = dcb_listen_create_socket_inet(host, port);
}
}
else
{

2
server/core/gateway.cc
View File

@ -1889,7 +1889,7 @@ int main(int argc, char **argv)
cnf = config_get_global_options();
ss_dassert(cnf);
if (!qc_setup(cnf->qc_name, cnf->qc_args))
if (!qc_setup(cnf->qc_name, cnf->qc_sql_mode, cnf->qc_args))
{
const char* logerr = "Failed to initialise query classifier library.";
print_log_n_stderr(true, true, logerr, logerr, eno);

2
server/core/maxscale/session.h
View File

@ -27,7 +27,7 @@ MXS_BEGIN_DECLS
#define SESSION_INIT {.ses_chk_top = CHK_NUM_SESSION, \
.stats = SESSION_STATS_INIT, .head = MXS_DOWNSTREAM_INIT, .tail = MXS_UPSTREAM_INIT, \
.state = SESSION_STATE_ALLOC, .ses_chk_tail = CHK_NUM_SESSION}
.state = SESSION_STATE_ALLOC, .client_protocol_data = 0, .ses_chk_tail = CHK_NUM_SESSION}
#define SESSION_PROTOCOL(x, type) DCB_PROTOCOL((x)->client_dcb, type)

53
server/core/maxscale/trxboundaryparser.hh
View File

@ -14,7 +14,7 @@
#include <maxscale/cppdefs.hh>
#include <ctype.h>
#include <maxscale/modutil.h>
#include <maxscale/customparser.hh>
#include <maxscale/query_classifier.h>
namespace maxscale
@ -37,8 +37,11 @@ namespace maxscale
* of utmost importance; consequently it is defined in its entirety
* in the header to allow for aggressive inlining.
*/
class TrxBoundaryParser
class TrxBoundaryParser : public maxscale::CustomParser
{
TrxBoundaryParser(const TrxBoundaryParser&);
TrxBoundaryParser& operator = (const TrxBoundaryParser&);
public:
enum token_t
{
@ -88,10 +91,6 @@ public:
* @endcode
*/
TrxBoundaryParser()
: m_pSql(NULL)
, m_len(0)
, m_pI(NULL)
, m_pEnd(NULL)
{
}
@ -546,34 +545,6 @@ private:
return type_mask;
}
inline bool is_next_alpha(char uc, int offset = 1) const
{
ss_dassert(uc >= 'A' && uc <= 'Z');
char lc = uc + ('a' - 'A');
return
((m_pI + offset) < m_pEnd) &&
((*(m_pI + offset) == uc) || (*(m_pI + offset) == lc));
}
bool is_next_char(char c, int offset = 1) const
{
return ((m_pI + offset) < m_pEnd) && (*(m_pI + offset) == c);
}
bool peek_next_char(char* pC) const
{
bool rc = (m_pI + 1 < m_pEnd);
if (rc)
{
*pC = *(m_pI + 1);
}
return rc;
}
// Significantly faster than library version.
static char toupper(char c)
{
@ -730,7 +701,7 @@ private:
}
else if (is_next_alpha('N'))
{
if (is_next_char('L', 2))
if (is_next_alpha('L', 2))
{
token = expect_token(TBP_EXPECT_TOKEN("ONLY"), TK_ONLY);
}
@ -770,7 +741,7 @@ private:
{
token = expect_token(TBP_EXPECT_TOKEN("SNAPSHOT"), TK_SNAPSHOT);
}
else if (is_next_char('T'))
else if (is_next_alpha('T'))
{
token = expect_token(TBP_EXPECT_TOKEN("START"), TK_START);
}
@ -830,16 +801,6 @@ private:
return token;
}
private:
TrxBoundaryParser(const TrxBoundaryParser&);
TrxBoundaryParser& operator = (const TrxBoundaryParser&);
private:
const char* m_pSql;
int m_len;
const char* m_pI;
const char* m_pEnd;
};
}

78
server/core/query_classifier.cc
View File

@ -45,7 +45,7 @@ static QUERY_CLASSIFIER* classifier;
static qc_trx_parse_using_t qc_trx_parse_using = QC_TRX_PARSE_USING_PARSER;
bool qc_setup(const char* plugin_name, const char* plugin_args)
bool qc_setup(const char* plugin_name, qc_sql_mode_t sql_mode, const char* plugin_args)
{
QC_TRACE();
ss_dassert(!classifier);
@ -61,7 +61,7 @@ bool qc_setup(const char* plugin_name, const char* plugin_args)
if (classifier)
{
rv = classifier->qc_setup(plugin_args);
rv = classifier->qc_setup(sql_mode, plugin_args);
if (rv != QC_RESULT_OK)
{
@ -494,42 +494,48 @@ const char* qc_op_to_string(qc_query_op_t op)
case QUERY_OP_UNDEFINED:
return "QUERY_OP_UNDEFINED";
case QUERY_OP_SELECT:
return "QUERY_OP_SELECT";
case QUERY_OP_UPDATE:
return "QUERY_OP_UPDATE";
case QUERY_OP_INSERT:
return "QUERY_OP_INSERT";
case QUERY_OP_DELETE:
return "QUERY_OP_DELETE";
case QUERY_OP_TRUNCATE:
return "QUERY_OP_TRUNCATE";
case QUERY_OP_ALTER:
return "QUERY_OP_ALTER";
case QUERY_OP_CREATE:
return "QUERY_OP_CREATE";
case QUERY_OP_DROP:
return "QUERY_OP_DROP";
case QUERY_OP_CHANGE_DB:
return "QUERY_OP_CHANGE_DB";
case QUERY_OP_LOAD:
return "QUERY_OP_LOAD";
case QUERY_OP_CREATE:
return "QUERY_OP_CREATE";
case QUERY_OP_DELETE:
return "QUERY_OP_DELETE";
case QUERY_OP_DROP:
return "QUERY_OP_DROP";
case QUERY_OP_EXPLAIN:
return "QUERY_OP_EXPLAIN";
case QUERY_OP_GRANT:
return "QUERY_OP_GRANT";
case QUERY_OP_INSERT:
return "QUERY_OP_INSERT";
case QUERY_OP_LOAD:
return "QUERY_OP_LOAD";
case QUERY_OP_REVOKE:
return "QUERY_OP_REVOKE";
case QUERY_OP_SELECT:
return "QUERY_OP_SELECT";
case QUERY_OP_SHOW:
return "QUERY_OP_SHOW";
case QUERY_OP_TRUNCATE:
return "QUERY_OP_TRUNCATE";
case QUERY_OP_UPDATE:
return "QUERY_OP_UPDATE";
default:
return "UNKNOWN_QUERY_OP";
}
@ -921,3 +927,25 @@ uint64_t qc_get_server_version()
return version;
}
qc_sql_mode_t qc_get_sql_mode()
{
QC_TRACE();
ss_dassert(classifier);
qc_sql_mode_t sql_mode;
ss_debug(int32_t rv =) classifier->qc_get_sql_mode(&sql_mode);
ss_dassert(rv == QC_RESULT_OK);
return sql_mode;
}
void qc_set_sql_mode(qc_sql_mode_t sql_mode)
{
QC_TRACE();
ss_dassert(classifier);
ss_debug(int32_t rv =) classifier->qc_set_sql_mode(sql_mode);
ss_dassert(rv == QC_RESULT_OK);
}

6
server/core/session.cc
View File

@ -183,6 +183,12 @@ static MXS_SESSION* session_alloc_body(SERVICE* service, DCB* client_dcb,
session->stats.connect = time(0);
session->stmt.buffer = NULL;
session->stmt.target = NULL;
MXS_CONFIG *config = config_get_global_options();
// If MaxScale is running in Oracle mode, then autocommit needs to
// initially be off.
bool autocommit = (config->qc_sql_mode == QC_SQL_MODE_ORACLE) ? false : true;
session_set_autocommit(session, autocommit);
/*<
* Associate the session to the client DCB and set the reference count on
* the session to indicate that there is a single reference to the

2
server/core/test/testtrxcompare.cc
View File

@ -184,7 +184,7 @@ int main(int argc, char* argv[])
if (mxs_log_init(NULL, ".", MXS_LOG_TARGET_DEFAULT))
{
// We have to setup something in order for the regexes to be compiled.
if (qc_setup("qc_sqlite", NULL) && qc_process_init(QC_INIT_BOTH))
if (qc_setup("qc_sqlite", QC_SQL_MODE_DEFAULT, NULL) && qc_process_init(QC_INIT_BOTH))
{
Tester tester(verbosity);

2
server/core/test/testtrxtracking.cc
View File

@ -421,7 +421,7 @@ int main(int argc, char* argv[])
if (mxs_log_init(NULL, ".", MXS_LOG_TARGET_DEFAULT))
{
// We have to setup something in order for the regexes to be compiled.
if (qc_setup("qc_sqlite", NULL) && qc_process_init(QC_INIT_BOTH))
if (qc_setup("qc_sqlite", QC_SQL_MODE_DEFAULT, NULL) && qc_process_init(QC_INIT_BOTH))
{
rc = EXIT_SUCCESS;

2
server/core/utils.cc
View File

@ -245,8 +245,6 @@ void gw_str_xor(uint8_t *output, const uint8_t *input1, const uint8_t *input2, u
{
*output++ = *input1++ ^ *input2++;
}
*output = '\0';
}
/**********************************************************

2
server/modules/filter/cache/test/testkeygeneration.cc vendored
View File

@ -128,7 +128,7 @@ int main(int argc, char* argv[])
{
if (mxs_log_init(NULL, ".", MXS_LOG_TARGET_DEFAULT))
{
if (qc_setup(NULL, NULL) && qc_process_init(QC_INIT_BOTH))
if (qc_setup(NULL, QC_SQL_MODE_DEFAULT, NULL) && qc_process_init(QC_INIT_BOTH))
{
const char* zModule = argv[1];

2
server/modules/filter/cache/test/testrules.cc vendored
View File

@ -240,7 +240,7 @@ int main()
pConfig->n_threads = 1;
set_libdir(MXS_STRDUP_A("../../../../../query_classifier/qc_sqlite/"));
if (qc_setup("qc_sqlite", "") && qc_process_init(QC_INIT_BOTH))
if (qc_setup("qc_sqlite", QC_SQL_MODE_DEFAULT, "") && qc_process_init(QC_INIT_BOTH))
{
set_libdir(MXS_STRDUP_A("../"));
rc = test();

2
server/modules/filter/cache/test/teststorage.cc vendored
View File

@ -50,7 +50,7 @@ int TestStorage::run(int argc, char** argv)
{
if (mxs_log_init(NULL, ".", MXS_LOG_TARGET_DEFAULT))
{
if (qc_setup(NULL, NULL) && qc_process_init(QC_INIT_BOTH))
if (qc_setup(NULL, QC_SQL_MODE_DEFAULT, NULL) && qc_process_init(QC_INIT_BOTH))
{
const char* zModule = NULL;
size_t threads = m_threads;

106
server/modules/filter/dbfwfilter/dbfwfilter.c
View File

@ -120,6 +120,80 @@ typedef enum
RT_CLAUSE /*< WHERE-clause requirement rule */
} ruletype_t;
/**
* What operator a rule should apply to.
*
* Note that each operator is represented by a unique bit, so that they
* can be combined as a bitmask, while query_op_t enumeration of the query
* classifier consists of a sequence of unique numbers.
*/
typedef enum fw_op
{
FW_OP_UNDEFINED = 0,
// NOTE: If you add something here, check the 'qc_op_to_fw_op' function below.
FW_OP_ALTER = (1 << 0),
FW_OP_CHANGE_DB = (1 << 1),
FW_OP_CREATE = (1 << 2),
FW_OP_DELETE = (1 << 3),
FW_OP_DROP = (1 << 4),
FW_OP_GRANT = (1 << 5),
FW_OP_INSERT = (1 << 6),
FW_OP_LOAD = (1 << 7),
FW_OP_REVOKE = (1 << 8),
FW_OP_SELECT = (1 << 9),
FW_OP_UPDATE = (1 << 10),
} fw_op_t;
/**
* Convert a qc_query_op_t to the equivalent fw_op_t.
*
* @param op A query classifier operator.
*
* @return The corresponding bit value.
*/
static inline fw_op_t qc_op_to_fw_op(qc_query_op_t op)
{
switch (op)
{
case QUERY_OP_ALTER:
return FW_OP_ALTER;
case QUERY_OP_CHANGE_DB:
return FW_OP_CHANGE_DB;
case QUERY_OP_CREATE:
return FW_OP_CREATE;
case QUERY_OP_DELETE:
return FW_OP_DELETE;
case QUERY_OP_DROP:
return FW_OP_DROP;
case QUERY_OP_GRANT:
return FW_OP_GRANT;
case QUERY_OP_INSERT:
return FW_OP_INSERT;
case QUERY_OP_LOAD:
return FW_OP_LOAD;
case QUERY_OP_REVOKE:
return FW_OP_REVOKE;
case QUERY_OP_SELECT:
return FW_OP_SELECT;
case QUERY_OP_UPDATE:
return FW_OP_UPDATE;
default:
return FW_OP_UNDEFINED;
};
}
/**
* Possible actions to take when the query matches a rule
*/
@ -199,7 +273,7 @@ typedef struct rule_t
void* data; /*< Actual implementation of the rule */
char* name; /*< Name of the rule */
ruletype_t type; /*< Type of the rule */
qc_query_op_t on_queries; /*< Types of queries to inspect */
uint32_t on_queries; /*< Types of queries to inspect */
int times_matched; /*< Number of times this rule has been matched */
TIMERANGE* active; /*< List of times when this rule is active */
struct rule_t *next;
@ -543,47 +617,47 @@ bool parse_querytypes(const char* str, RULE* rule)
*dest = '\0';
if (strcmp(buffer, "select") == 0)
{
rule->on_queries |= QUERY_OP_SELECT;
rule->on_queries |= FW_OP_SELECT;
}
else if (strcmp(buffer, "insert") == 0)
{
rule->on_queries |= QUERY_OP_INSERT;
rule->on_queries |= FW_OP_INSERT;
}
else if (strcmp(buffer, "update") == 0)
{
rule->on_queries |= QUERY_OP_UPDATE;
rule->on_queries |= FW_OP_UPDATE;
}
else if (strcmp(buffer, "delete") == 0)
{
rule->on_queries |= QUERY_OP_DELETE;
rule->on_queries |= FW_OP_DELETE;
}
else if (strcmp(buffer, "use") == 0)
{
rule->on_queries |= QUERY_OP_CHANGE_DB;
rule->on_queries |= FW_OP_CHANGE_DB;
}
else if (strcmp(buffer, "grant") == 0)
{
rule->on_queries |= QUERY_OP_GRANT;
rule->on_queries |= FW_OP_GRANT;
}
else if (strcmp(buffer, "revoke") == 0)
{
rule->on_queries |= QUERY_OP_REVOKE;
rule->on_queries |= FW_OP_REVOKE;
}
else if (strcmp(buffer, "drop") == 0)
{
rule->on_queries |= QUERY_OP_DROP;
rule->on_queries |= FW_OP_DROP;
}
else if (strcmp(buffer, "create") == 0)
{
rule->on_queries |= QUERY_OP_CREATE;
rule->on_queries |= FW_OP_CREATE;
}
else if (strcmp(buffer, "alter") == 0)
{
rule->on_queries |= QUERY_OP_ALTER;
rule->on_queries |= FW_OP_ALTER;
}
else if (strcmp(buffer, "load") == 0)
{
rule->on_queries |= QUERY_OP_LOAD;
rule->on_queries |= FW_OP_LOAD;
}
if (done)
@ -1078,7 +1152,7 @@ bool create_rule(void* scanner, const char* name)
if (ruledef && (ruledef->name = MXS_STRDUP(name)))
{
ruledef->type = RT_UNDEFINED;
ruledef->on_queries = QUERY_OP_UNDEFINED;
ruledef->on_queries = FW_OP_UNDEFINED;
ruledef->next = rstack->rule;
ruledef->active = NULL;
ruledef->times_matched = 0;
@ -2113,10 +2187,10 @@ bool rule_matches(FW_INSTANCE* my_instance,
}
}
if (rulebook->rule->on_queries == QUERY_OP_UNDEFINED ||
rulebook->rule->on_queries & optype ||
if (rulebook->rule->on_queries == FW_OP_UNDEFINED ||
rulebook->rule->on_queries & qc_op_to_fw_op(optype) ||
(MYSQL_IS_COM_INIT_DB((uint8_t*)GWBUF_DATA(queue)) &&
rulebook->rule->on_queries & QUERY_OP_CHANGE_DB))
rulebook->rule->on_queries & FW_OP_CHANGE_DB))
{
switch (rulebook->rule->type)
{

6
server/modules/protocol/MySQL/MySQLClient/CMakeLists.txt
View File

@ -1,4 +1,8 @@
add_library(MySQLClient SHARED mysql_client.c)
add_library(MySQLClient SHARED mysql_client.cc)
target_link_libraries(MySQLClient maxscale-common MySQLCommon)
set_target_properties(MySQLClient PROPERTIES VERSION "1.0.0")
install_module(MySQLClient core)
if(BUILD_TESTS)
add_subdirectory(test)
endif()

97
server/modules/protocol/MySQL/MySQLClient/mysql_client.c → server/modules/protocol/MySQL/MySQLClient/mysql_client.cc
View File

@ -14,23 +14,26 @@
#define MXS_MODULE_NAME "MySQLClient"
#include <maxscale/cppdefs.hh>
#include <inttypes.h>
#include <limits.h>
#include <netinet/tcp.h>
#include <sys/stat.h>
#include <maxscale/protocol.h>
#include <netinet/tcp.h>
#include <sys/stat.h>
#include <maxscale/alloc.h>
#include <maxscale/authenticator.h>
#include <maxscale/log_manager.h>
#include <maxscale/protocol/mysql.h>
#include <maxscale/ssl.h>
#include <maxscale/poll.h>
#include <maxscale/modinfo.h>
#include <maxscale/modutil.h>
#include <maxscale/poll.h>
#include <maxscale/protocol/mysql.h>
#include <maxscale/query_classifier.h>
#include <maxscale/authenticator.h>
#include <maxscale/session.h>
#include <maxscale/worker.h>
#include <maxscale/ssl.h>
#include "setsqlmodeparser.hh"
/** Return type of process_special_commands() */
typedef enum spec_com_res_t
@ -86,6 +89,10 @@ static bool parse_kill_query(char *query, uint64_t *thread_id_out, kill_type_t *
*
* @return The module object
*/
extern "C"
{
MXS_MODULE* MXS_CREATE_MODULE()
{
static MXS_PROTOCOL MyObject =
@ -126,6 +133,8 @@ MXS_MODULE* MXS_CREATE_MODULE()
return &info;
}
}
/*lint +e14 */
/**
@ -186,7 +195,7 @@ static void thread_finish(void)
*/
static char *gw_default_auth()
{
return "MySQLAuth";
return (char*)"MySQLAuth";
}
/**
@ -242,7 +251,7 @@ int MySQLSendHandshake(DCB* dcb)
}
else
{
version_string = GW_MYSQL_VERSION;
version_string = (char*)GW_MYSQL_VERSION;
len_version_string = strlen(GW_MYSQL_VERSION);
}
@ -486,7 +495,7 @@ int gw_read_client_event(DCB* dcb)
case MXS_AUTH_STATE_MESSAGE_READ:
/* After this call read_buffer will point to freed data */
if (nbytes_read < 3 || (0 == max_bytes && nbytes_read <
(MYSQL_GET_PAYLOAD_LEN((uint8_t *) GWBUF_DATA(read_buffer)) + 4)) ||
(int)(MYSQL_GET_PAYLOAD_LEN((uint8_t *) GWBUF_DATA(read_buffer)) + 4)) ||
(0 != max_bytes && nbytes_read < max_bytes))
{
@ -671,7 +680,7 @@ gw_read_do_authentication(DCB *dcb, GWBUF *read_buffer, int nbytes_read)
if (dcb->user == NULL)
{
/** User authentication complete, copy the username to the DCB */
MYSQL_session *ses = dcb->data;
MYSQL_session *ses = (MYSQL_session*)dcb->data;
if ((dcb->user = MXS_STRDUP(ses->user)) == NULL)
{
dcb_close(dcb);
@ -691,6 +700,9 @@ gw_read_do_authentication(DCB *dcb, GWBUF *read_buffer, int nbytes_read)
MXS_SESSION *session =
session_alloc_with_id(dcb->service, dcb, protocol->thread_id);
// For the time being only the sql_mode is stored in MXS_SESSION::client_protocol_data.
session->client_protocol_data = QC_SQL_MODE_DEFAULT;
if (session != NULL)
{
CHK_SESSION(session);
@ -848,7 +860,7 @@ static bool process_client_commands(DCB* dcb, int bytes_available, GWBUF** buffe
if (dcb->protocol_packet_length - MYSQL_HEADER_LEN != GW_MYSQL_MAX_PACKET_LEN)
{
/** We're processing the first packet of a command */
proto->current_command = cmd;
proto->current_command = (mysql_server_cmd_t)cmd;
}
dcb->protocol_packet_length = pktlen + MYSQL_HEADER_LEN;
@ -870,6 +882,55 @@ static bool process_client_commands(DCB* dcb, int bytes_available, GWBUF** buffe
return true;
}
/**
* Sets the query classifier mode.
*
* @param session The session for which the query classifier mode is adjusted.
* @param read_buffer Pointer to a buffer, assumed to contain a statement.
* May be reallocated if not contiguous.
*/
void set_qc_mode(MXS_SESSION* session, GWBUF** read_buffer)
{
SetSqlModeParser parser;
SetSqlModeParser::sql_mode_t sql_mode;
switch (parser.get_sql_mode(read_buffer, &sql_mode))
{
case SetSqlModeParser::ERROR:
// In practice only OOM.
break;
case SetSqlModeParser::IS_SET_SQL_MODE:
switch (sql_mode)
{
case SetSqlModeParser::ORACLE:
session_set_autocommit(session, false);
session->client_protocol_data = QC_SQL_MODE_ORACLE;
break;
case SetSqlModeParser::DEFAULT:
session_set_autocommit(session, true);
session->client_protocol_data = QC_SQL_MODE_DEFAULT;
break;
case SetSqlModeParser::SOMETHING:
break;
default:
ss_dassert(!true);
}
break;
case SetSqlModeParser::NOT_SET_SQL_MODE:
break;
default:
ss_dassert(!true);
}
qc_set_sql_mode(static_cast<qc_sql_mode_t>(session->client_protocol_data));
}
/**
* @brief Client read event, process data, client already authenticated
*
@ -921,11 +982,13 @@ gw_read_normal_data(DCB *dcb, GWBUF *read_buffer, int nbytes_read)
if (rcap_type_required(capabilities, RCAP_TYPE_STMT_INPUT))
{
if (nbytes_read < 3 || nbytes_read <
(MYSQL_GET_PAYLOAD_LEN((uint8_t *) GWBUF_DATA(read_buffer)) + 4))
(int)(MYSQL_GET_PAYLOAD_LEN((uint8_t *) GWBUF_DATA(read_buffer)) + 4))
{
dcb->dcb_readqueue = read_buffer;
return 0;
}
set_qc_mode(session, &read_buffer);
}
/** The query classifier classifies according to the service's server that has
@ -1049,7 +1112,7 @@ mysql_client_auth_error_handling(DCB *dcb, int auth_val, int packet_number)
/** Send error 1049 to client */
message_len = 25 + MYSQL_DATABASE_MAXLEN;
fail_str = MXS_CALLOC(1, message_len + 1);
fail_str = (char*)MXS_CALLOC(1, message_len + 1);
MXS_ABORT_IF_NULL(fail_str);
snprintf(fail_str, message_len, "Unknown database '%s'", session->db);
@ -1381,7 +1444,7 @@ static int route_by_statement(MXS_SESSION* session, uint64_t capabilities, GWBUF
{
CHK_GWBUF(packetbuf);
MySQLProtocol* proto = session->client_dcb->protocol;
MySQLProtocol* proto = (MySQLProtocol*)session->client_dcb->protocol;
proto->current_command = (mysql_server_cmd_t)GWBUF_DATA(packetbuf)[4];
/**
@ -1458,9 +1521,9 @@ static int route_by_statement(MXS_SESSION* session, uint64_t capabilities, GWBUF
}
else if ((type & QUERY_TYPE_COMMIT) || (type & QUERY_TYPE_ROLLBACK))
{
mxs_session_trx_state_t trx_state = session_get_trx_state(session);
uint32_t trx_state = session_get_trx_state(session);
trx_state |= SESSION_TRX_ENDING_BIT;
session_set_trx_state(session, trx_state);
session_set_trx_state(session, (mxs_session_trx_state_t)trx_state);
if (type & QUERY_TYPE_ENABLE_AUTOCOMMIT)
{
@ -1523,7 +1586,7 @@ static bool ensure_complete_packet(DCB *dcb, GWBUF **read_buffer, int nbytes_rea
{
uint8_t* data = (uint8_t *) GWBUF_DATA(*read_buffer);
if (nbytes_read < 3 || nbytes_read < MYSQL_GET_PAYLOAD_LEN(data) + 4)
if (nbytes_read < 3 || nbytes_read < (int)MYSQL_GET_PAYLOAD_LEN(data) + 4)
{
dcb->dcb_readqueue = gwbuf_append(dcb->dcb_readqueue, *read_buffer);
return false;
@ -1559,7 +1622,7 @@ static spec_com_res_t process_special_commands(DCB *dcb, GWBUF *read_buffer, int
* The option is stored as a two byte integer with the values 0 for enabling
* multi-statements and 1 for disabling it.
*/
MySQLProtocol *proto = dcb->protocol;
MySQLProtocol *proto = (MySQLProtocol*)dcb->protocol;
uint8_t opt;
if (proto->current_command == MYSQL_COM_SET_OPTION &&

651
server/modules/protocol/MySQL/MySQLClient/setsqlmodeparser.hh Normal file
View File

@ -0,0 +1,651 @@
#pragma once
/*
* Copyright (c) 2016 MariaDB Corporation Ab
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file and at www.mariadb.com/bsl11.
*
* Change Date: 2019-07-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2 or later of the General
* Public License.
*/
#include <maxscale/cppdefs.hh>
#include <maxscale/customparser.hh>
#include <maxscale/protocol/mysql.h>
class SetSqlModeParser : public maxscale::CustomParser
{
public:
enum sql_mode_t
{
DEFAULT, // "set sql_mode=DEFAULT"
ORACLE, // "set sql_mode=ORACLE", "set sql_mode='PIPES_AS_CONCAT,ORACLE', autocommit=false", etc.
SOMETHING // "set sql_mode=PIPES_AS_CONCAT"
};
enum result_t
{
ERROR, // Some fatal error occurred; mem alloc failed, parsing failed, etc.
IS_SET_SQL_MODE, // The COM_QUERY is "set sql_mode=..."
NOT_SET_SQL_MODE // The COM_QUERY is NOT "set sql_mode=..."
};
enum
{
UNUSED_FIRST = 0xFF,
TK_DEFAULT,
TK_GLOBAL,
TK_GLOBAL_VAR,
TK_ORACLE,
TK_SESSION,
TK_SESSION_VAR,
TK_SET,
TK_SQL_MODE,
};
SetSqlModeParser()
{
}
/**
* Return whether the statement is a "SET SQL_MODE=" statement and if so,
* whether the state is ORACLE, DEFAULT or something else.
*
* @param ppBuffer Address of pointer to buffer containing statement.
* The GWBUF must contain a complete statement, but the
* buffer need not be contiguous.
* @param pSql_mode Pointer to variable receiving the sql_mode state, if
* the statement is a "SET SQL_MODE=" statement.
*
* @return ERROR if a fatal error occurred during parsing
* IS_SET_SQL_MODE if the statement is a "SET SQL_MODE=" statement
* NOT_SET_SQL_MODE if the statement is not a "SET SQL_MODE="
* statement
*
* @attention If the result cannot be deduced without parsing the statement,
* then the buffer will be made contiguous and the value of
* @c *ppBuffer will be updated accordingly.
*/
result_t get_sql_mode(GWBUF** ppBuffer, sql_mode_t* pSql_mode)
{
result_t rv = NOT_SET_SQL_MODE;
GWBUF* pBuffer = *ppBuffer;
ss_dassert(gwbuf_length(pBuffer) >= MYSQL_HEADER_LEN);
size_t buf_len = GWBUF_LENGTH(pBuffer);
size_t payload_len;
if (buf_len >= MYSQL_HEADER_LEN)
{
// The first buffer in the chain contains the header so we
// can read the length directly.
payload_len = MYSQL_GET_PAYLOAD_LEN(GWBUF_DATA(pBuffer));
}
else
{
// The first buffer in the chain does not contain the full
// header so we need to copy it first.
uint8_t header[MYSQL_HEADER_LEN];
gwbuf_copy_data(pBuffer, 0, sizeof(header), header);
payload_len = MYSQL_GET_PAYLOAD_LEN(header);
}
if (payload_len >= 20) // sizeof(command byte) + strlen("SET sql_mode=ORACLE"), the minimum needed.
{
// We need 4 bytes from the payload to deduce whether more investigations are needed.
uint8_t payload[4];
uint8_t* pPayload;
if (buf_len >= MYSQL_HEADER_LEN + sizeof(payload))
{
// Enough data in the first buffer of the chain, we can access directly.
pPayload = GWBUF_DATA(pBuffer) + MYSQL_HEADER_LEN;
}
else
{
// Not enough, we copy what we need.
gwbuf_copy_data(pBuffer, MYSQL_HEADER_LEN, sizeof(payload), payload);
pPayload = payload;
}
uint8_t command = pPayload[0];
if (command == MYSQL_COM_QUERY)
{
const uint8_t* pStmt = &pPayload[1];
if (is_alpha(*pStmt))
{
// First character is alphabetic, we can check whether it is "SET".
if (is_set(pStmt))
{
// It is, so we must parse further and must therefore ensure that
// the buffer is contiguous. We get the same buffer back if it
// already is.
pBuffer = gwbuf_make_contiguous(*ppBuffer);
if (pBuffer)
{
*ppBuffer = pBuffer;
initialize(pBuffer);
rv = parse(pSql_mode);
}
else
{
rv = ERROR;
}
}
}
else
{
// If the first character is not an alphabetic character we assume there
// is a comment and make the buffer contiguous to make it possible to
// efficiently bypass the whitespace.
pBuffer = gwbuf_make_contiguous(*ppBuffer);
if (pBuffer)
{
*ppBuffer = pBuffer;
initialize(pBuffer);
bypass_whitespace();
if (is_set(m_pI))
{
rv = parse(pSql_mode);
}
}
else
{
rv = ERROR;
}
}
}
}
return rv;
}
/**
* Returns a @c sql_mode_t as a string.
*
* @param sql_mode An SQL mode.
*
* @return The corresponding string.
*/
static const char* to_string(sql_mode_t sql_mode)
{
switch (sql_mode)
{
case DEFAULT:
return "DEFAULT";
case ORACLE:
return "ORACLE";
case SOMETHING:
return "SOMETHING";
default:
ss_dassert(!true);
return "UNKNOWN";
}
}
/**
* Returns a @c result_t as a string.
*
* @param result_t A result.
*
* @return The corresponding string.
*/
static const char* to_string(result_t result)
{
switch (result)
{
case ERROR:
return "ERROR";
case IS_SET_SQL_MODE:
return "IS_SET_SQL_MODE";
case NOT_SET_SQL_MODE:
return "NOT_SET_SQL_MODE";
default:
ss_dassert(!true);
return "UNKNOWN";
}
}
private:
static bool is_set(const char* pStmt)
{
return
(pStmt[0] == 's' || pStmt[0] == 'S') &&
(pStmt[1] == 'e' || pStmt[1] == 'E') &&
(pStmt[2] == 't' || pStmt[2] == 'T');
}
static bool is_set(const uint8_t* pStmt)
{
return is_set(reinterpret_cast<const char*>(pStmt));
}
static bool is_error(result_t rv)
{
return (rv == ERROR);
}
result_t initialize(GWBUF* pBuffer)
{
ss_dassert(GWBUF_IS_CONTIGUOUS(pBuffer));
result_t rv = ERROR;
char* pSql;
if (modutil_extract_SQL(pBuffer, &pSql, &m_len))
{
m_pSql = pSql;
m_pI = m_pSql;
m_pEnd = m_pI + m_len;
}
return ERROR;
}
bool consume_id()
{
// Consumes "[a-zA-Z]([a-zA-Z0-9_])*
bool rv = false;
if (is_alpha(*m_pI))
{
rv = true;
++m_pI;
while ((m_pI < m_pEnd) && (is_alpha(*m_pI) || is_number(*m_pI) || (*m_pI == '_')))
{
++m_pI;
}
}
return rv;
}
void consume_value()
{
// Consumes everything until a ',' outside of a commented string, or eol is
// encountered.
bool rv = false;
bool consumed = false;
while ((m_pI < m_pEnd) && (*m_pI != ','))
{
switch (*m_pI)
{
case '\'':
case '"':
case '`':
{
char quote = *m_pI;
++m_pI;
while ((m_pI < m_pEnd) && (*m_pI != quote))
{
++m_pI;
}
}
break;
default:
++m_pI;
}
}
}
result_t parse(sql_mode_t* pSql_mode)
{
result_t rv = NOT_SET_SQL_MODE;
token_t token = next_token();
switch (token)
{
case TK_SET:
rv = parse_set(pSql_mode);
break;
case PARSER_EXHAUSTED:
log_exhausted();
break;
case PARSER_UNKNOWN_TOKEN:
default:
log_unexpected();
break;
}
return rv;
}
result_t parse_set(sql_mode_t* pSql_mode)
{
result_t rv = NOT_SET_SQL_MODE;
char c;
do
{
token_t token = next_token();
switch (token)
{
case TK_GLOBAL:
rv = parse_set(pSql_mode);
break;
case TK_SESSION:
rv = parse_set(pSql_mode);
break;
case TK_GLOBAL_VAR:
case TK_SESSION_VAR:
if (next_token() == '.')
{
rv = parse_set(pSql_mode);
}
else
{
rv = ERROR;
}
break;
case TK_SQL_MODE:
if (next_token() == '=')
{
rv = parse_set_sql_mode(pSql_mode);
}
else
{
rv = ERROR;
}
break;
case PARSER_EXHAUSTED:
log_exhausted();
rv = ERROR;
break;
case PARSER_UNKNOWN_TOKEN:
// Might be something like "SET A=B, C=D, SQL_MODE=ORACLE", so we first consume
// the identifier and if it is followed by a "=" we consume the value.
{
char c;
if (consume_id())
{
bypass_whitespace();
if (peek_current_char(&c) && (c == '='))
{
++m_pI;
consume_value();
}
}
else
{
log_unexpected();
rv = ERROR;
}
}
break;
default:
log_unexpected();
rv = ERROR;
break;
}
c = 0;
if (rv != ERROR)
{
bypass_whitespace();
if (peek_current_char(&c))
{
if (c == ',')
{
++m_pI;
}
else
{
c = 0;
}
}
else
{
c = 0;
}
}
}
while (c == ',');
return rv;
}
result_t parse_set_sql_mode(sql_mode_t* pSql_mode)
{
result_t rv = IS_SET_SQL_MODE;
token_t token = next_token();
switch (token)
{
case '\'':
case '"':
case '`':
rv = parse_set_sql_mode_string(token, pSql_mode);
break;
case TK_DEFAULT:
*pSql_mode = DEFAULT;
break;
case TK_ORACLE:
*pSql_mode = ORACLE;
break;
case PARSER_UNKNOWN_TOKEN:
if (consume_id())
{
*pSql_mode = SOMETHING;
}
else
{
rv = ERROR;
}
break;
default:
rv = ERROR;
}
return rv;
}
result_t parse_set_sql_mode_string(char quote, sql_mode_t* pSql_mode)
{
result_t rv = IS_SET_SQL_MODE;
char c;
do
{
rv = parse_set_sql_mode_setting(pSql_mode);
if (!is_error(rv))
{
bypass_whitespace();
if (peek_current_char(&c) && (c == ','))
{
++m_pI;
}
}
}
while (!is_error(rv) && (c == ','));
return rv;
}
result_t parse_set_sql_mode_setting(sql_mode_t* pSql_mode)
{
result_t rv = IS_SET_SQL_MODE;
token_t token = next_token();
switch (token)
{
case TK_ORACLE:
*pSql_mode = ORACLE;
break;
case PARSER_UNKNOWN_TOKEN:
if (consume_id())
{
*pSql_mode = SOMETHING;
}
else
{
rv = ERROR;
}
break;
case PARSER_EXHAUSTED:
log_exhausted();
rv = ERROR;
break;
default:
log_unexpected();
rv = ERROR;
}
return rv;
}
token_t next_token(token_required_t required = TOKEN_NOT_REQUIRED)
{
token_t token = PARSER_UNKNOWN_TOKEN;
bypass_whitespace();
if (m_pI == m_pEnd)
{
token = PARSER_EXHAUSTED;
}
else if (*m_pI == ';')
{
++m_pI;
while ((m_pI != m_pEnd) && isspace(*m_pI))
{
++m_pI;
}
if (m_pI != m_pEnd)
{
MXS_WARNING("Non-space data found after semi-colon: '%.*s'.",
(int)(m_pEnd - m_pI), m_pI);
}
token = PARSER_EXHAUSTED;
}
else
{
switch (*m_pI)
{
case '@':
if (is_next_alpha('S', 2))
{
token = expect_token(MXS_CP_EXPECT_TOKEN("@@SESSION"), TK_SESSION_VAR);
}
else if (is_next_alpha('G', 2))
{
token = expect_token(MXS_CP_EXPECT_TOKEN("@@GLOBAL"), TK_GLOBAL_VAR);
}
else if (is_next_alpha('L', 2))
{
token = expect_token(MXS_CP_EXPECT_TOKEN("@@LOCAL"), TK_SESSION_VAR);
}
break;
case '.':
case '\'':
case '"':
case '`':
case ',':
case '=':
token = *m_pI;
++m_pI;
break;
case 'd':
case 'D':
token = expect_token(MXS_CP_EXPECT_TOKEN("DEFAULT"), TK_DEFAULT);
break;
case 'g':
case 'G':
token = expect_token(MXS_CP_EXPECT_TOKEN("GLOBAL"), TK_GLOBAL);
break;
case 'l':
case 'L':
token = expect_token(MXS_CP_EXPECT_TOKEN("LOCAL"), TK_SESSION);
break;
case 'o':
case 'O':
token = expect_token(MXS_CP_EXPECT_TOKEN("ORACLE"), TK_ORACLE);
break;
case 's':
case 'S':
if (is_next_alpha('E'))
{
if (is_next_alpha('S', 2))
{
token = expect_token(MXS_CP_EXPECT_TOKEN("SESSION"), TK_SESSION);
}
else
{
token = expect_token(MXS_CP_EXPECT_TOKEN("SET"), TK_SET);
}
}
else if (is_next_alpha('Q'))
{
token = expect_token(MXS_CP_EXPECT_TOKEN("SQL_MODE"), TK_SQL_MODE);
}
break;
default:
;
}
}
if ((token == PARSER_EXHAUSTED) && (required == TOKEN_REQUIRED))
{
log_exhausted();
}
return token;
}
};

4
server/modules/protocol/MySQL/MySQLClient/test/CMakeLists.txt Normal file
View File

@ -0,0 +1,4 @@
add_executable(test_setsqlmodeparser test_setsqlmodeparser.cc)
target_link_libraries(test_setsqlmodeparser maxscale-common)
add_test(test_setsqlmodeparser test_setsqlmodeparser)

361
server/modules/protocol/MySQL/MySQLClient/test/test_setsqlmodeparser.cc Normal file
View File

@ -0,0 +1,361 @@
/*
* Copyright (c) 2016 MariaDB Corporation Ab
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file and at www.mariadb.com/bsl11.
*
* Change Date: 2019-07-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2 or later of the General
* Public License.
*/
#include "../setsqlmodeparser.hh"
#include <stdlib.h>
#include <time.h>
#include <iostream>
#include <maxscale/buffer.h>
#include <maxscale/paths.h>
using namespace std;
namespace
{
GWBUF* gwbuf_create_com_query(const char* zStmt)
{
size_t len = strlen(zStmt);
size_t payload_len = len + 1;
size_t gwbuf_len = MYSQL_HEADER_LEN + payload_len;
GWBUF* pBuf = gwbuf_alloc(gwbuf_len);
*((unsigned char*)((char*)GWBUF_DATA(pBuf))) = payload_len;
*((unsigned char*)((char*)GWBUF_DATA(pBuf) + 1)) = (payload_len >> 8);
*((unsigned char*)((char*)GWBUF_DATA(pBuf) + 2)) = (payload_len >> 16);
*((unsigned char*)((char*)GWBUF_DATA(pBuf) + 3)) = 0x00;
*((unsigned char*)((char*)GWBUF_DATA(pBuf) + 4)) = 0x03;
memcpy((char*)GWBUF_DATA(pBuf) + 5, zStmt, len);
return pBuf;
}
}
namespace
{
typedef SetSqlModeParser P;
struct TEST_CASE
{
const char* zStmt;
SetSqlModeParser::result_t result;
SetSqlModeParser::sql_mode_t sql_mode;
} test_cases[] =
{
{
"SET SQL_MODE=DEFAULT",
P::IS_SET_SQL_MODE,
P::DEFAULT
},
{
"SET SQL_MODE=DEFAULT;",
P::IS_SET_SQL_MODE,
P::DEFAULT
},
{
"SET SQL_MODE=DEFAULT; ",
P::IS_SET_SQL_MODE,
P::DEFAULT
},
{
"-- This is a comment\nSET SQL_MODE=DEFAULT",
P::IS_SET_SQL_MODE,
P::DEFAULT
},
{
"#This is a comment\nSET SQL_MODE=DEFAULT",
P::IS_SET_SQL_MODE,
P::DEFAULT
},
{
"/*blah*/ SET /*blah*/ SQL_MODE /*blah*/ = /*blah*/ DEFAULT /*blah*/ ",
P::IS_SET_SQL_MODE,
P::DEFAULT
},
{
"SET SQL_MODE=ORACLE",
P::IS_SET_SQL_MODE,
P::ORACLE
},
{
"SET SQL_MODE=BLAH", // So short that it cannot be DEFAULT|ORACLE
P::NOT_SET_SQL_MODE,
P::ORACLE
},
{
"SET SQL_MODE='BLAH'",
P::IS_SET_SQL_MODE,
P::SOMETHING
},
{
"SET SQL_MODE=BLAHBLAH",
P::IS_SET_SQL_MODE,
P::SOMETHING
},
{
"SET SQL_MODE='ORACLE'",
P::IS_SET_SQL_MODE,
P::ORACLE
},
{
"SET SQL_MODE='BLAH, A, B, ORACLE'",
P::IS_SET_SQL_MODE,
P::ORACLE
},
{
"SET SQL_MODE='BLAH, A, B, XYZ_123'",
P::IS_SET_SQL_MODE,
P::SOMETHING
},
{
"SET VAR1=1234, VAR2=3456, SQL_MODE='A,B, ORACLE'",
P::IS_SET_SQL_MODE,
P::ORACLE
},
{
"SET SQL_MODE=ORACLE, VAR1=3456, VAR2='A=b, c=d', SQL_MODE='A,B, ORACLE'",
P::IS_SET_SQL_MODE,
P::ORACLE
},
{
"SET GLOBAL SQL_MODE=ORACLE",
P::IS_SET_SQL_MODE,
P::ORACLE
},
{
"SET SESSION SQL_MODE=ORACLE",
P::IS_SET_SQL_MODE,
P::ORACLE
},
{
"SET LOCAL SQL_MODE=ORACLE",
P::IS_SET_SQL_MODE,
P::ORACLE
},
{
"SET @@GLOBAL.SQL_MODE=ORACLE",
P::IS_SET_SQL_MODE,
P::ORACLE
},
{
"SET @@SESSION.SQL_MODE=ORACLE",
P::IS_SET_SQL_MODE,
P::ORACLE
},
{
"SET @@LOCAL.SQL_MODE=ORACLE",
P::IS_SET_SQL_MODE,
P::ORACLE
},
{
"SET @@LOCAL . SQL_MODE = ORACLE",
P::IS_SET_SQL_MODE,
P::ORACLE
},
{
"SET @@SESSION.blah = 1234, @@GLOBAL.blahblah = something, sql_mode=ORACLE",
P::IS_SET_SQL_MODE,
P::ORACLE
},
};
const int N_TEST_CASES = sizeof(test_cases)/sizeof(test_cases[0]);
int test(GWBUF** ppStmt,
SetSqlModeParser::sql_mode_t expected_sql_mode,
SetSqlModeParser::result_t expected_result)
{
int rv = EXIT_SUCCESS;
SetSqlModeParser parser;
SetSqlModeParser::sql_mode_t sql_mode;
SetSqlModeParser::result_t result = parser.get_sql_mode(ppStmt, &sql_mode);
if (result == expected_result)
{
if (result == SetSqlModeParser::IS_SET_SQL_MODE)
{
if (sql_mode == expected_sql_mode)
{
cout << "OK";
}
else
{
cout << "ERROR: Expected "
<< "'" << SetSqlModeParser::to_string(expected_sql_mode) << "'"
<< ", got "
<< "'" << SetSqlModeParser::to_string(sql_mode) << "'"
<< ".";
rv = EXIT_FAILURE;
}
}
else
{
cout << "OK";
}
}
else
{
cout << "ERROR: Expected "
<< "'" << SetSqlModeParser::to_string(expected_result) << "'"
<< ", got "
<< "'" << SetSqlModeParser::to_string(result) << "'"
<< ".";
rv = EXIT_FAILURE;
}
cout << endl;
return rv;
}
int test(const TEST_CASE& test_case)
{
int rv = EXIT_SUCCESS;
cout << test_case.zStmt << ": ";
GWBUF* pStmt = gwbuf_create_com_query(test_case.zStmt);
ss_dassert(pStmt);
rv = test(&pStmt, test_case.sql_mode, test_case.result);
gwbuf_free(pStmt);
return rv;
}
int test_contiguous()
{
int rv = EXIT_SUCCESS;
cout << "Test contiguous statements\n"
<< "--------------------------" << endl;
for (int i = 0; i < N_TEST_CASES; ++i)
{
if (test(test_cases[i]) == EXIT_FAILURE)
{
rv = EXIT_FAILURE;
}
}
cout << endl;
return rv;
}
int test_non_contiguous()
{
int rv = EXIT_SUCCESS;
cout << "Test non-contiguous statements\n"
<< "------------------------------" << endl;
for (int i = 0; i < N_TEST_CASES; ++i)
{
TEST_CASE& test_case = test_cases[i];
cout << test_case.zStmt << "(" << strlen(test_case.zStmt) << ": ";
GWBUF* pTail = gwbuf_create_com_query(test_case.zStmt);
ss_dassert(pTail);
GWBUF* pStmt = NULL;
while (pTail)
{
size_t n = MYSQL_HEADER_LEN + rand() % 10; // Between 4 and 13 bytes long chunks.
GWBUF* pHead = gwbuf_split(&pTail, n);
cout << GWBUF_LENGTH(pHead);
pStmt = gwbuf_append(pStmt, pHead);
if (pTail)
{
cout << ", ";
}
}
cout << "): " << flush;
if (test(&pStmt, test_case.sql_mode, test_case.result) == EXIT_FAILURE)
{
rv = EXIT_FAILURE;
}
gwbuf_free(pStmt);
}
cout << endl;
return rv;
}
int test()
{
int rv = EXIT_SUCCESS;
if (test_contiguous() != EXIT_SUCCESS)
{
rv = EXIT_FAILURE;
}
if (test_non_contiguous() != EXIT_SUCCESS)
{
rv = EXIT_FAILURE;
}
if (rv == EXIT_SUCCESS)
{
cout << "OK" << endl;
}
else
{
cout << "ERROR" << endl;
}
return rv;
}
}
int main(int argc, char* argv[])
{
int rv = EXIT_SUCCESS;
srand(time(NULL));
set_datadir(strdup("/tmp"));
set_langdir(strdup("."));
set_process_datadir(strdup("/tmp"));
if (mxs_log_init(NULL, ".", MXS_LOG_TARGET_DEFAULT))
{
rv = test();
mxs_log_finish();
}
else
{
cerr << "error: Could not initialize log." << endl;
}
return rv;
}

2
server/modules/protocol/MySQL/test/CMakeLists.txt
View File

@ -1,4 +1,4 @@
add_executable(test_parse_kill test_parse_kill.c)
add_executable(test_parse_kill test_parse_kill.cc)
target_link_libraries(test_parse_kill maxscale-common MySQLCommon)
add_test(test_parse_kill test_parse_kill)

8
server/modules/protocol/MySQL/test/test_parse_kill.c → server/modules/protocol/MySQL/test/test_parse_kill.cc
View File

@ -1,8 +1,8 @@
#include <maxscale/cdefs.h>
#include "../MySQLClient/mysql_client.c"
#include "../MySQLClient/mysql_client.cc"
int test_one_query(char *query, bool should_succeed, uint64_t expected_tid,
int test_one_query(const char *query, bool should_succeed, uint64_t expected_tid,
kill_type_t expected_kt)
{
char *query_copy = MXS_STRDUP_A(query);
@ -46,7 +46,7 @@ int test_one_query(char *query, bool should_succeed, uint64_t expected_tid,
}
typedef struct test_t
{
char *query;
const char *query;
bool should_succeed;
uint64_t correct_id;
kill_type_t correct_kt;
@ -79,7 +79,7 @@ int main(int argc, char **argv)
int arr_size = sizeof(tests) / sizeof(test_t);
for (int i = 0; i < arr_size; i++)
{
char *query = tests[i].query;
const char *query = tests[i].query;
bool should_succeed = tests[i].should_succeed;
uint64_t expected_tid = tests[i].correct_id;
kill_type_t expected_kt = tests[i].correct_kt;

4
server/modules/protocol/examples/cdc_schema.py
View File

@ -1,4 +1,4 @@
#!/usr/bin/env python3
#!/usr/bin/env python
# Copyright (c) 2016 MariaDB Corporation Ab
#
@ -38,7 +38,7 @@ def parse_field(row):
res["real_type"] = name
if len(parts) > 1:
if len(parts) > 1 and name not in ["enum", "set", "decimal"]:
res["length"] = int(parts[1].split(')')[0])
else:
res["length"] = -1

21
server/modules/routing/avrorouter/avro_file.c
View File

@ -911,6 +911,15 @@ bool is_create_table_statement(AVRO_INSTANCE *router, char* ptr, size_t len)
return rc > 0;
}
bool is_create_like_statement(const char* ptr, size_t len)
{
char sql[len + 1];
memcpy(sql, ptr, len);
sql[len] = '\0';
// This is not pretty but it should work
return strcasestr(sql, " like ") || strcasestr(sql, "(like ");
}
/**
* @brief Detection of table alteration statements
@ -1020,7 +1029,16 @@ void handle_query_event(AVRO_INSTANCE *router, REP_HEADER *hdr, int *pending_tra
if (is_create_table_statement(router, sql, len))
{
TABLE_CREATE *created = table_create_alloc(sql, db);
TABLE_CREATE *created = NULL;
if (is_create_like_statement(sql, len))
{
created = table_create_copy(router, sql, len, db);
}
else
{
created = table_create_alloc(sql, db);
}
if (created && !save_and_replace_table_create(router, created))
{
@ -1053,7 +1071,6 @@ void handle_query_event(AVRO_INSTANCE *router, REP_HEADER *hdr, int *pending_tra
strcat(full_ident, ident);
TABLE_CREATE *created = hashtable_fetch(router->created_tables, full_ident);
ss_dassert(created);
if (created)
{

12
server/modules/routing/avrorouter/avro_rbr.c
View File

@ -131,13 +131,13 @@ bool handle_table_map_event(AVRO_INSTANCE *router, REP_HEADER *hdr, uint8_t *ptr
if (old)
{
router->active_maps[old->id % sizeof(router->active_maps)] = NULL;
router->active_maps[old->id % MAX_MAPPED_TABLES] = NULL;
}
hashtable_delete(router->table_maps, table_ident);
hashtable_add(router->table_maps, (void*) table_ident, map);
hashtable_add(router->open_tables, table_ident, avro_table);
save_avro_schema(router->avrodir, json_schema, map);
router->active_maps[map->id % sizeof(router->active_maps)] = map;
router->active_maps[map->id % MAX_MAPPED_TABLES] = map;
MXS_DEBUG("Table %s mapped to %lu", table_ident, map->id);
rval = true;
@ -164,10 +164,10 @@ bool handle_table_map_event(AVRO_INSTANCE *router, REP_HEADER *hdr, uint8_t *ptr
}
else
{
ss_dassert(router->active_maps[old->id % sizeof(router->active_maps)] == old);
router->active_maps[old->id % sizeof(router->active_maps)] = NULL;
ss_dassert(router->active_maps[old->id % MAX_MAPPED_TABLES] == old);
router->active_maps[old->id % MAX_MAPPED_TABLES] = NULL;
table_map_remap(ptr, ev_len, old);
router->active_maps[old->id % sizeof(router->active_maps)] = old;
router->active_maps[old->id % MAX_MAPPED_TABLES] = old;
MXS_DEBUG("Table %s re-mapped to %lu", table_ident, old->id);
/** No changes in the schema */
rval = true;
@ -294,7 +294,7 @@ bool handle_row_event(AVRO_INSTANCE *router, REP_HEADER *hdr, uint8_t *ptr)
/** There should always be a table map event prior to a row event.
* TODO: Make the active_maps dynamic */
TABLE_MAP *map = router->active_maps[table_id % sizeof(router->active_maps)];
TABLE_MAP *map = router->active_maps[table_id % MAX_MAPPED_TABLES];
if (map)
{

221
server/modules/routing/avrorouter/avro_schema.c
View File

@ -771,6 +771,227 @@ TABLE_CREATE* table_create_alloc(const char* sql, const char* event_db)
return rval;
}
static const char* TOK_CREATE[] =
{
"CREATE",
NULL
};
static const char* TOK_TABLE[] =
{
"TABLE",
NULL
};
static const char* TOK_GROUP_REPLACE[] =
{
"OR",
"REPLACE",
NULL
};
static const char* TOK_GROUP_EXISTS[] =
{
"IF",
"NOT",
"EXISTS",
NULL
};
/**
* Read one token (i.e. SQL keyword)
*/
static const char* get_token(const char* ptr, const char* end, char* dest)
{
while (ptr < end && isspace(*ptr))
{
ptr++;
}
const char* start = ptr;
while (ptr < end && !isspace(*ptr))
{
ptr++;
}
size_t len = ptr - start;
memcpy(dest, start, len);
dest[len] = '\0';
return ptr;
}
/**
* Consume one token
*/
static bool chomp_one_token(const char* expected, const char** ptr, const char* end, char* buf)
{
bool rval = false;
const char* next = get_token(*ptr, end, buf);
if (strcasecmp(buf, expected) == 0)
{
rval = true;
*ptr = next;
}
return rval;
}
/**
* Consume all tokens in a group
*/
static bool chomp_tokens(const char** tokens, const char** ptr, const char* end, char* buf)
{
bool next = true;
bool rval = false;
do
{
next = false;
for (int i = 0; tokens[i]; i++)
{
if (chomp_one_token(tokens[i], ptr, end, buf))
{
rval = true;
next = true;
break;
}
}
}
while (next);
return rval;
}
/**
* Remove any extra characters from a string
*/
static void remove_extras(char* str)
{
char* end = strchr(str, '\0') - 1;
while (end > str && (*end == '`' || *end == ')' || *end == '('))
{
*end-- = '\0';
}
char* start = str;
while (start < end && (*start == '`' || *start == ')' || *start == '('))
{
start++;
}
size_t len = strlen(start);
memmove(str, start, len);
str[len] = '\0';
ss_dassert(strlen(str) == len);
}
/**
* Extract both tables from a `CREATE TABLE t1 LIKE t2` statement
*/
static bool extract_create_like_identifier(const char* sql, size_t len, char* target, char* source)
{
bool rval = false;
char buffer[len + 1];
buffer[0] = '\0';
const char* ptr = sql;
const char* end = ptr + sizeof(buffer);
if (chomp_tokens(TOK_CREATE, &ptr, end, buffer))
{
chomp_tokens(TOK_GROUP_REPLACE, &ptr, end, buffer);
if (chomp_tokens(TOK_TABLE, &ptr, end, buffer))
{
chomp_tokens(TOK_GROUP_EXISTS, &ptr, end, buffer);
// Read the target table name
ptr = get_token(ptr, end, buffer);
strcpy(target, buffer);
// Skip the LIKE token
ptr = get_token(ptr, end, buffer);
// Read the source table name
ptr = get_token(ptr, end, buffer);
remove_extras(buffer);
strcpy(source, buffer);
rval = true;
}
}
return rval;
}
/**
* Create a table from another table
*/
TABLE_CREATE* table_create_copy(AVRO_INSTANCE *router, const char* sql, size_t len, const char* db)
{
TABLE_CREATE* rval = NULL;
char target[MYSQL_TABLE_MAXLEN + 1] = "";
char source[MYSQL_TABLE_MAXLEN + 1] = "";
if (extract_create_like_identifier(sql, len, target, source))
{
char table_ident[MYSQL_TABLE_MAXLEN + MYSQL_DATABASE_MAXLEN + 2] = "";
if (strchr(source, '.') == NULL)
{
strcpy(table_ident, db);
strcat(table_ident, ".");
}
strcat(table_ident, source);
TABLE_CREATE *old = hashtable_fetch(router->created_tables, table_ident);
if (old)
{
int n = old->columns;
char** names = MXS_MALLOC(sizeof(char*) * n);
char** types = MXS_MALLOC(sizeof(char*) * n);
int* lengths = MXS_MALLOC(sizeof(int) * n);
rval = MXS_MALLOC(sizeof(TABLE_CREATE));
MXS_ABORT_IF_FALSE(names && types && lengths && rval);
for (uint64_t i = 0; i < old->columns; i++)
{
names[i] = MXS_STRDUP_A(old->column_names[i]);
types[i] = MXS_STRDUP_A(old->column_types[i]);
lengths[i] = old->column_lengths[i];
}
rval->version = 1;
rval->was_used = false;
rval->column_names = names;
rval->column_lengths = lengths;
rval->column_types = types;
rval->columns = old->columns;
rval->database = MXS_STRDUP_A(db);
char* table = strchr(target, '.');
table = table ? table + 1 : target;
rval->table = MXS_STRDUP_A(table);
}
else
{
MXS_ERROR("Could not find table '%s' that '%s' is being created from: %.*s",
table_ident, target, (int)len, sql);
}
}
return rval;
}
/**
* Free a TABLE_CREATE structure
* @param value Value to free

1
server/modules/routing/avrorouter/avrorouter.h
View File

@ -310,6 +310,7 @@ extern void read_table_info(uint8_t *ptr, uint8_t post_header_len, uint64_t *tab
extern TABLE_MAP *table_map_alloc(uint8_t *ptr, uint8_t hdr_len, TABLE_CREATE* create);
extern void table_map_free(TABLE_MAP *map);
extern TABLE_CREATE* table_create_alloc(const char* sql, const char* db);
extern TABLE_CREATE* table_create_copy(AVRO_INSTANCE *router, const char* sql, size_t len, const char* db);
extern void table_create_free(TABLE_CREATE* value);
extern bool table_create_save(TABLE_CREATE *create, const char *filename);
extern bool table_create_alter(TABLE_CREATE *create, const char *sql, const char *end);