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

Merge branch '2.1' into develop

Browse Source
This commit is contained in:
Johan Wikman
2017-03-03 13:37:14 +02:00
parent c926577a3c 76e56a2688
commit b3e8ea9b5a
56 changed files with 2355 additions and 630 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
server/modules/routing/avrorouter/avro.c
View File

@ -72,14 +72,14 @@ static const char* alter_table_regex =
/* The router entry points */
static MXS_ROUTER *createInstance(SERVICE *service, char **options);
static void *newSession(MXS_ROUTER *instance, MXS_SESSION *session);
static void closeSession(MXS_ROUTER *instance, void *router_session);
static void freeSession(MXS_ROUTER *instance, void *router_session);
static int routeQuery(MXS_ROUTER *instance, void *router_session, GWBUF *queue);
static MXS_ROUTER_SESSION *newSession(MXS_ROUTER *instance, MXS_SESSION *session);
static void closeSession(MXS_ROUTER *instance, MXS_ROUTER_SESSION *router_session);
static void freeSession(MXS_ROUTER *instance, MXS_ROUTER_SESSION *router_session);
static int routeQuery(MXS_ROUTER *instance, MXS_ROUTER_SESSION *router_session, GWBUF *queue);
static void diagnostics(MXS_ROUTER *instance, DCB *dcb);
static void clientReply(MXS_ROUTER *instance, void *router_session, GWBUF *queue,
static void clientReply(MXS_ROUTER *instance, MXS_ROUTER_SESSION *router_session, GWBUF *queue,
DCB *backend_dcb);
static void errorReply(MXS_ROUTER *instance, void *router_session, GWBUF *message,
static void errorReply(MXS_ROUTER *instance, MXS_ROUTER_SESSION *router_session, GWBUF *message,
DCB *backend_dcb, mxs_error_action_t action, bool *succp);
static uint64_t getCapabilities(MXS_ROUTER* instance);
extern int MaxScaleUptime();
@ -630,7 +630,7 @@ createInstance(SERVICE *service, char **options)
* @param session The session itself
* @return Session specific data for this session
*/
static void *
static MXS_ROUTER_SESSION *
newSession(MXS_ROUTER *instance, MXS_SESSION *session)
{
AVRO_INSTANCE *inst = (AVRO_INSTANCE *) instance;
@ -703,7 +703,7 @@ newSession(MXS_ROUTER *instance, MXS_SESSION *session)
* @param router_cli_ses The particular session to free
*
*/
static void freeSession(MXS_ROUTER* router_instance, void* router_client_ses)
static void freeSession(MXS_ROUTER* router_instance, MXS_ROUTER_SESSION* router_client_ses)
{
AVRO_INSTANCE *router = (AVRO_INSTANCE *) router_instance;
AVRO_CLIENT *client = (AVRO_CLIENT *) router_client_ses;
@ -752,7 +752,7 @@ static void freeSession(MXS_ROUTER* router_instance, void* router_client_ses)
* @param instance The router instance data
* @param router_session The session being closed
*/
static void closeSession(MXS_ROUTER *instance, void *router_session)
static void closeSession(MXS_ROUTER *instance, MXS_ROUTER_SESSION *router_session)
{
AVRO_INSTANCE *router = (AVRO_INSTANCE *) instance;
AVRO_CLIENT *client = (AVRO_CLIENT *) router_session;
@ -784,7 +784,7 @@ static void closeSession(MXS_ROUTER *instance, void *router_session)
* @return 1 on success, 0 on error
*/
static int
routeQuery(MXS_ROUTER *instance, void *router_session, GWBUF *queue)
routeQuery(MXS_ROUTER *instance, MXS_ROUTER_SESSION *router_session, GWBUF *queue)
{
AVRO_INSTANCE *router = (AVRO_INSTANCE *) instance;
AVRO_CLIENT *client = (AVRO_CLIENT *) router_session;
@ -960,7 +960,7 @@ diagnostics(MXS_ROUTER *router, DCB *dcb)
* @param queue The GWBUF with reply data
*/
static void
clientReply(MXS_ROUTER *instance, void *router_session, GWBUF *queue, DCB *backend_dcb)
clientReply(MXS_ROUTER *instance, MXS_ROUTER_SESSION *router_session, GWBUF *queue, DCB *backend_dcb)
{
/** We should never end up here */
ss_dassert(false);
@ -1002,7 +1002,7 @@ extract_message(GWBUF *errpkt)
*
*/
static void
errorReply(MXS_ROUTER *instance, void *router_session, GWBUF *message, DCB *backend_dcb,
errorReply(MXS_ROUTER *instance, MXS_ROUTER_SESSION *router_session, GWBUF *message, DCB *backend_dcb,
mxs_error_action_t action,
bool *succp)
{

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

@ -30,7 +30,7 @@ static bool warn_large_enumset = false; /**< Remove when support for ENUM/SET va
uint8_t* process_row_event_data(TABLE_MAP *map, TABLE_CREATE *create,
avro_value_t *record, uint8_t *ptr,
uint8_t *columns_present);
uint8_t *columns_present, uint8_t *end);
void notify_all_clients(AVRO_INSTANCE *router);
void add_used_table(AVRO_INSTANCE* router, const char* table);
@ -309,9 +309,10 @@ bool handle_row_event(AVRO_INSTANCE *router, REP_HEADER *hdr, uint8_t *ptr)
while (ptr - start < hdr->event_size - BINLOG_EVENT_HDR_LEN)
{
/** Add the current GTID and timestamp */
uint8_t *end = ptr + hdr->event_size;
int event_type = get_event_type(hdr->event_type);
prepare_record(router, hdr, event_type, &record);
ptr = process_row_event_data(map, create, &record, ptr, col_present);
ptr = process_row_event_data(map, create, &record, ptr, col_present, end);
avro_file_writer_append_value(table->avro_file, &record);
/** Update rows events have the before and after images of the
@ -320,7 +321,7 @@ bool handle_row_event(AVRO_INSTANCE *router, REP_HEADER *hdr, uint8_t *ptr)
if (event_type == UPDATE_EVENT)
{
prepare_record(router, hdr, UPDATE_EVENT_AFTER, &record);
ptr = process_row_event_data(map, create, &record, ptr, col_present);
ptr = process_row_event_data(map, create, &record, ptr, col_present, end);
avro_file_writer_append_value(table->avro_file, &record);
}
@ -497,7 +498,7 @@ int get_metadata_len(uint8_t type)
* @return Pointer to the first byte after the current row event
*/
uint8_t* process_row_event_data(TABLE_MAP *map, TABLE_CREATE *create, avro_value_t *record,
uint8_t *ptr, uint8_t *columns_present)
uint8_t *ptr, uint8_t *columns_present, uint8_t *end)
{
int npresent = 0;
avro_value_t field;
@ -507,10 +508,12 @@ uint8_t* process_row_event_data(TABLE_MAP *map, TABLE_CREATE *create, avro_value
/** BIT type values use the extra bits in the row event header */
int extra_bits = (((ncolumns + 7) / 8) * 8) - ncolumns;
ss_dassert(ptr < end);
/** Store the null value bitmap */
uint8_t *null_bitmap = ptr;
ptr += (ncolumns + 7) / 8;
ss_dassert(ptr < end);
for (long i = 0; i < map->columns && npresent < ncolumns; i++)
{
@ -544,6 +547,7 @@ uint8_t* process_row_event_data(TABLE_MAP *map, TABLE_CREATE *create, avro_value
}
avro_value_set_string(&field, strval);
ptr += bytes;
ss_dassert(ptr < end);
}
else
{
@ -553,6 +557,7 @@ uint8_t* process_row_event_data(TABLE_MAP *map, TABLE_CREATE *create, avro_value
str[bytes] = '\0';
avro_value_set_string(&field, str);
ptr += bytes + 1;
ss_dassert(ptr < end);
}
}
else if (column_is_bit(map->column_types[i]))
@ -572,21 +577,36 @@ uint8_t* process_row_event_data(TABLE_MAP *map, TABLE_CREATE *create, avro_value
}
avro_value_set_int(&field, value);
ptr += bytes;
ss_dassert(ptr < end);
}
else if (column_is_decimal(map->column_types[i]))
{
double f_value = 0.0;
ptr += unpack_decimal_field(ptr, metadata + metadata_offset, &f_value);
avro_value_set_double(&field, f_value);
ss_dassert(ptr < end);
}
else if (column_is_variable_string(map->column_types[i]))
{
size_t sz;
char *str = mxs_lestr_consume(&ptr, &sz);
int bytes = metadata[metadata_offset] | metadata[metadata_offset + 1] << 8;
if (bytes > 255)
{
sz = gw_mysql_get_byte2(ptr);
ptr += 2;
}
else
{
sz = *ptr;
ptr++;
}
char buf[sz + 1];
memcpy(buf, str, sz);
memcpy(buf, ptr, sz);
buf[sz] = '\0';
ptr += sz;
avro_value_set_string(&field, buf);
ss_dassert(ptr < end);
}
else if (column_is_blob(map->column_types[i]))
{
@ -596,6 +616,7 @@ uint8_t* process_row_event_data(TABLE_MAP *map, TABLE_CREATE *create, avro_value
ptr += bytes;
avro_value_set_bytes(&field, ptr, len);
ptr += len;
ss_dassert(ptr < end);
}
else if (column_is_temporal(map->column_types[i]))
{
@ -604,6 +625,7 @@ uint8_t* process_row_event_data(TABLE_MAP *map, TABLE_CREATE *create, avro_value
ptr += unpack_temporal_value(map->column_types[i], ptr, &metadata[metadata_offset], &tm);
format_temporal_value(buf, sizeof(buf), map->column_types[i], &tm);
avro_value_set_string(&field, buf);
ss_dassert(ptr < end);
}
/** All numeric types (INT, LONG, FLOAT etc.) */
else
@ -613,6 +635,7 @@ uint8_t* process_row_event_data(TABLE_MAP *map, TABLE_CREATE *create, avro_value
ptr += unpack_numeric_field(ptr, map->column_types[i],
&metadata[metadata_offset], lval);
set_numeric_field_value(&field, map->column_types[i], &metadata[metadata_offset], lval);
ss_dassert(ptr < end);
}
ss_dassert(metadata_offset <= map->column_metadata_size);
metadata_offset += get_metadata_len(map->column_types[i]);

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

@ -449,6 +449,10 @@ static const char *extract_field_name(const char* ptr, char* dest, size_t size)
while (*ptr && (isspace(*ptr) || (bt = *ptr == '`')))
{
ptr++;
if (bt)
{
break;
}
}
if (strncasecmp(ptr, "constraint", 10) == 0 || strncasecmp(ptr, "index", 5) == 0 ||
@ -481,11 +485,6 @@ static const char *extract_field_name(const char* ptr, char* dest, size_t size)
/** Valid identifier */
size_t bytes = ptr - start;
if (bt)
{
bytes--;
}
memcpy(dest, start, bytes);
dest[bytes] = '\0';