database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2026-10-23 18:49:29 +03:00
Code Activity

MDEV-38144: update Optional_metadata_fields to use MariaDB types

Browse Source 
Currently Optional_metadata_fields has many members that use classes
from the C++ standard library, most notably the use of std::vector and
std::string. This is inconsistent with the coding standards, as MariaDB
defines its own types for arrays/lists/strings. This patch updates
these variables to use existing MariaDB types.

Additionally, the structure of the fields is changed. Instead of having
a separate list to track each metadata field; the structure now exists
at the column level (Column_metadata), and it has individual member
variables to describe its metadata. This allows for more
straightforward memory management, as everything will now be allocated
at once.

There is also a bug fix in this patch. Prior to this refactor, YEAR
columns would not consume their "unsigned" bit flag, resulting in
numeric columns after a YEAR type to have incorrect signage. This is
now fixed.

Reviewed-by: Brandon Nesterenko <brandon.nesterenko@mariadb.com>.
This commit is contained in:
Eric
2026-07-15 19:37:14 -04:00
committed by Georgi Kodinov
parent d0fc15c370
commit 699a09e847
10 changed files with 552 additions and 333 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
mysql-test/suite/binlog/r/binlog_signedness_parse.result Normal file
View File

@@ -0,0 +1,77 @@
ALTER DATABASE test CHARACTER SET latin1 COLLATE latin1_swedish_ci;
RESET MASTER;
SET GLOBAL binlog_row_metadata = FULL;
#
# Test 1: Non-numeric column between two numeric columns does not
# consume a signedness bit (signed INT, CHAR, unsigned INT)
#
CREATE TABLE t1(c_signed INT, c_char CHAR(10), c_unsigned INT UNSIGNED);
INSERT INTO t1 VALUES(-1, 'x', 1);
# Columns(`c_signed` INT,
# `c_char` CHAR(10) CHARSET latin1 COLLATE latin1_swedish_ci,
# `c_unsigned` INT UNSIGNED)
DROP TABLE t1;
RESET MASTER;
#
# Test 2: Byte boundary — 8 signed numerics then a non-numeric then an
# unsigned numeric. The non-numeric must not shift the bit
# cursor into the second byte prematurely.
#
CREATE TABLE t1(
c1 TINYINT, c2 TINYINT, c3 TINYINT, c4 TINYINT,
c5 TINYINT, c6 TINYINT, c7 TINYINT, c8 TINYINT,
c_text TEXT,
c9 TINYINT UNSIGNED);
INSERT INTO t1(c1) VALUES(1);
# Columns(`c1` TINYINT,
# `c2` TINYINT,
# `c3` TINYINT,
# `c4` TINYINT,
# `c5` TINYINT,
# `c6` TINYINT,
# `c7` TINYINT,
# `c8` TINYINT,
# `c_text` TEXT CHARSET latin1 COLLATE latin1_swedish_ci,
# `c9` TINYINT UNSIGNED)
DROP TABLE t1;
RESET MASTER;
#
# Test 3: Multiple non-numeric columns between numeric columns across
# a byte boundary
#
CREATE TABLE t1(
c1 TINYINT UNSIGNED, c2 TINYINT UNSIGNED,
c3 TINYINT UNSIGNED, c4 TINYINT UNSIGNED,
c5 TINYINT UNSIGNED, c6 TINYINT UNSIGNED,
c7 TINYINT UNSIGNED, c8 TINYINT UNSIGNED,
c_blob BLOB, c_varchar VARCHAR(100),
c9 TINYINT, c10 TINYINT UNSIGNED);
INSERT INTO t1(c1) VALUES(1);
# Columns(`c1` TINYINT UNSIGNED,
# `c2` TINYINT UNSIGNED,
# `c3` TINYINT UNSIGNED,
# `c4` TINYINT UNSIGNED,
# `c5` TINYINT UNSIGNED,
# `c6` TINYINT UNSIGNED,
# `c7` TINYINT UNSIGNED,
# `c8` TINYINT UNSIGNED,
# `c_blob` BLOB,
# `c_varchar` VARCHAR(100) CHARSET latin1 COLLATE latin1_swedish_ci,
# `c9` TINYINT,
# `c10` TINYINT UNSIGNED)
DROP TABLE t1;
RESET MASTER;
#
# Test 4: YEAR is treated as a numeric type: it always reads as UNSIGNED
# and consumes a signedness bit, so a numeric column following it
# still reads the correct bit (signed INT, YEAR, unsigned INT).
#
CREATE TABLE t1(c_signed INT, c_year YEAR, c_unsigned INT UNSIGNED);
INSERT INTO t1(c_signed) VALUES(-1);
# Columns(`c_signed` INT,
# `c_year` YEAR UNSIGNED,
# `c_unsigned` INT UNSIGNED)
DROP TABLE t1;
RESET MASTER;
SET GLOBAL binlog_row_metadata = NO_LOG;
ALTER DATABASE test CHARACTER SET utf8mb4 COLLATE utf8mb4_uca1400_ai_ci;

2
mysql-test/suite/binlog/r/binlog_table_map_optional_metadata.result
View File

@@ -9,7 +9,7 @@ c_datetime DATETIME, c_datetime_f DATETIME(3),
c_timestamp TIMESTAMP NOT NULL DEFAULT NOW(),
c_timestamp_f TIMESTAMP(3) DEFAULT "2017-1-1 10:10:10");
INSERT INTO t1(c_year) VALUES(2017);
# Columns(YEAR,
# Columns(YEAR UNSIGNED,
# DATE,
# TIME,
# TIME(3),

83
mysql-test/suite/binlog/t/binlog_signedness_parse.test Normal file
View File

@@ -0,0 +1,83 @@
################################################################################
# Regression tests for parse_signedness() in log_event.cc.
#
# The old implementation (upstream) pushed all bits from every byte into a flat
# vector without regard to column type. The new implementation iterates over
# column metadata and only consumes a bit for columns where is_numeric_type()
# returns true, skipping non-numeric columns.
#
# Tests focus on:
# 1. YEAR is treated as a numeric type and always shown UNSIGNED.
# 2. Non-numeric columns interspersed between numeric columns do not consume
# a signedness bit, so the second numeric column reads the correct bit.
# 3. Correct behavior at byte boundaries when non-numeric columns appear
# between the 8th and 9th numeric column.
################################################################################
--source include/have_binlog_format_row.inc
--source include/test_db_charset_latin1.inc
RESET MASTER;
SET @old_metadata= @@GLOBAL.binlog_row_metadata;
SET GLOBAL binlog_row_metadata = FULL;
--let $MYSQLD_DATADIR= `select @@datadir`
--let $binlog_file= $MYSQLD_DATADIR/master-bin.000001
--echo #
--echo # Test 1: Non-numeric column between two numeric columns does not
--echo # consume a signedness bit (signed INT, CHAR, unsigned INT)
--echo #
CREATE TABLE t1(c_signed INT, c_char CHAR(10), c_unsigned INT UNSIGNED);
INSERT INTO t1 VALUES(-1, 'x', 1);
--source suite/binlog/include/print_optional_metadata.inc
DROP TABLE t1;
RESET MASTER;
--echo #
--echo # Test 2: Byte boundary — 8 signed numerics then a non-numeric then an
--echo # unsigned numeric. The non-numeric must not shift the bit
--echo # cursor into the second byte prematurely.
--echo #
CREATE TABLE t1(
c1 TINYINT, c2 TINYINT, c3 TINYINT, c4 TINYINT,
c5 TINYINT, c6 TINYINT, c7 TINYINT, c8 TINYINT,
c_text TEXT,
c9 TINYINT UNSIGNED);
INSERT INTO t1(c1) VALUES(1);
--source suite/binlog/include/print_optional_metadata.inc
DROP TABLE t1;
RESET MASTER;
--echo #
--echo # Test 3: Multiple non-numeric columns between numeric columns across
--echo # a byte boundary
--echo #
CREATE TABLE t1(
c1 TINYINT UNSIGNED, c2 TINYINT UNSIGNED,
c3 TINYINT UNSIGNED, c4 TINYINT UNSIGNED,
c5 TINYINT UNSIGNED, c6 TINYINT UNSIGNED,
c7 TINYINT UNSIGNED, c8 TINYINT UNSIGNED,
c_blob BLOB, c_varchar VARCHAR(100),
c9 TINYINT, c10 TINYINT UNSIGNED);
INSERT INTO t1(c1) VALUES(1);
--source suite/binlog/include/print_optional_metadata.inc
DROP TABLE t1;
RESET MASTER;
--echo #
--echo # Test 4: YEAR is treated as a numeric type: it always reads as UNSIGNED
--echo # and consumes a signedness bit, so a numeric column following it
--echo # still reads the correct bit (signed INT, YEAR, unsigned INT).
--echo #
CREATE TABLE t1(c_signed INT, c_year YEAR, c_unsigned INT UNSIGNED);
INSERT INTO t1(c_signed) VALUES(-1);
--source suite/binlog/include/print_optional_metadata.inc
DROP TABLE t1;
RESET MASTER;
SET GLOBAL binlog_row_metadata = @old_metadata;
--source include/test_db_charset_restore.inc

290
sql/log_event.cc
View File

@@ -3851,70 +3851,148 @@ Table_map_log_event::~Table_map_log_event()
/**
Parses SIGNEDNESS field.
@param[out] vec stores the signedness flags extracted from field.
@param[out] column_metadata store all column metadata here (including signedness)
@param[in] field SIGNEDNESS field in table_map_event.
@param[in] length length of the field
*/
static void parse_signedness(std::vector<bool> &vec,
unsigned char *field, unsigned int length)
using Optional_column_metadata= Table_map_log_event::Optional_metadata_fields::Column_metadata;
static void parse_signedness(
Dynamic_array<Optional_column_metadata>& column_metadata,
unsigned char *field, unsigned int length)
{
for (unsigned int i= 0; i < length; i++)
unsigned char *p= field;
unsigned char *field_end= p + length;
unsigned char unsigned_bitfield= 0;
unsigned char mask= 0;
for (uint col= 0; col < column_metadata.size(); col++)
{
for (unsigned char c= 0x80; c != 0; c>>= 1)
vec.push_back(field[i] & c);
auto &col_meta= column_metadata.at(col);
if (!is_numeric_type(col_meta.column_type))
continue;
if (mask == 0)
{
if (p >= field_end)
return;
unsigned_bitfield= *p++;
mask= 0x80;
}
col_meta.is_unsigned= unsigned_bitfield & mask;
mask >>= 1;
}
}
/**
Parses DEFAULT_CHARSET field.
@param[out] default_charset stores collation numbers extracted from field.
@param[out] fields store all optional metadata here (including charsets)
@param[in] field DEFAULT_CHARSET field in table_map_event.
@param[in] length length of the field
*/
static void parse_default_charset(Table_map_log_event::Optional_metadata_fields::
Default_charset &default_charset,
unsigned char *field, unsigned int length)
static void parse_default_charset(
Table_map_log_event::Optional_metadata_fields& fields,
unsigned char *field, unsigned int length)
{
unsigned char* p= field;
unsigned char* end= field + length;
default_charset.default_charset= net_field_length(&p);
uint default_cs= net_field_length(&p);
if (unlikely(p > end))
return;
for (uint i= 0; i < fields.m_column_metadata.size(); i++)
if (is_character_type(fields.m_column_metadata.at(i).column_type))
fields.m_column_metadata.at(i).charset= default_cs;
while (p < end)
{
unsigned int col_index= net_field_length(&p);
if (unlikely(p > end))
return;
unsigned int col_charset= net_field_length(&p);
uint charset= net_field_length(&p);
if (unlikely(p > end))
return;
if (col_index < fields.m_column_metadata.size())
fields.m_column_metadata.at(col_index).charset= charset;
}
}
default_charset.charset_pairs.push_back(std::make_pair(col_index,
col_charset));
/**
Parses ENUM_AND_SET_DEFAULT_CHARSET field.
@param[out] fields store all optional metadata here (including charsets)
@param[in] field ENUM_AND_SET_DEFAULT_CHARSET field in table_map_event.
@param[in] length length of the field
*/
static void parse_enum_and_set_default_charset(
Table_map_log_event::Optional_metadata_fields& fields,
unsigned char *field, unsigned int length)
{
unsigned char* p= field;
unsigned char* end= field + length;
uint default_cs= net_field_length(&p);
if (unlikely(p > end))
return;
for (uint i= 0; i < fields.m_column_metadata.size(); i++)
if (is_enum_or_set_type(fields.m_column_metadata.at(i).column_type))
fields.m_column_metadata.at(i).enum_and_set_column_charset= default_cs;
while (p < end)
{
unsigned int col_index= net_field_length(&p);
if (unlikely(p > end))
return;
uint charset= net_field_length(&p);
if (unlikely(p > end))
return;
if (col_index < fields.m_column_metadata.size())
fields.m_column_metadata.at(col_index).enum_and_set_column_charset= charset;
}
}
/**
Parses COLUMN_CHARSET field.
@param[out] vec stores collation numbers extracted from field.
@param[out] column_metadata store all column metadata here (including charsets)
@param[in] field COLUMN_CHARSET field in table_map_event.
@param[in] length length of the field
*/
static void parse_column_charset(std::vector<unsigned int> &vec,
unsigned char *field, unsigned int length)
static void parse_column_charset(
Dynamic_array<Optional_column_metadata>& column_metadata,
unsigned char *field, unsigned int length)
{
unsigned char* p= field;
unsigned char* end= field + length;
while (p < end)
for (uint col= 0; p < end && col < column_metadata.size(); col++)
{
unsigned int charset= net_field_length(&p);
if (!is_character_type(column_metadata.at(col).column_type))
continue;
column_metadata.at(col).charset= net_field_length(&p);
if (unlikely(p > end))
return;
}
}
/**
Parses ENUM_AND_SET_COLUMN_CHARSET field.
@param[out] column_metadata store all column metadata here (including charsets)
@param[in] field ENUM_AND_SET_COLUMN_CHARSET field in table_map_event.
@param[in] length length of the field
*/
static void parse_enum_and_set_column_charset(
Dynamic_array<Optional_column_metadata>& column_metadata,
unsigned char *field, unsigned int length)
{
unsigned char* p= field;
unsigned char* end= field + length;
for (uint col= 0; p < end && col < column_metadata.size(); col++)
{
if (!is_enum_or_set_type(column_metadata.at(col).column_type))
continue;
column_metadata.at(col).enum_and_set_column_charset= net_field_length(&p);
if (unlikely(p > end))
return;
vec.push_back(charset);
}
}
@@ -3922,98 +4000,118 @@ static void parse_column_charset(std::vector<unsigned int> &vec,
Parses COLUMN_NAME field.
@param[in] root Allocate memory here
@param[out] name Store column names extracted from field here
@param[out] column_metadata store all column metadata here (including column names)
@param[in] field COLUMN_NAME field in table_map_event.
@param[in] length length of the field
*/
static bool parse_column_name(MEM_ROOT *root, LEX_CSTRING *name,
unsigned char *field, unsigned int length)
static bool parse_column_name(MEM_ROOT *root,
Dynamic_array<Optional_column_metadata>& column_metadata,
unsigned char *field, unsigned int length)
{
for (uchar *end= field+length; field < end ; name++)
unsigned int col = 0;
for (uchar *end= field+length; field < end && col < column_metadata.size(); col++)
{
LEX_CSTRING& name = column_metadata.at(col).column_name;
uint name_length= net_field_length(&field);
if (unlikely(field + name_length > end))
return 1;
if (!(name->str= strmake_root(root, (char*) field, name_length)))
if (!(name.str= strmake_root(root, (char*) field, name_length)))
return 1;
name->length= name_length;
name.length= name_length;
field+= name_length;
}
name->str= 0; // End marker
return 0;
}
/**
Parses SET_STR_VALUE/ENUM_STR_VALUE field.
@param[out] vec stores SET/ENUM column's string values extracted from
field. Each SET/ENUM column's string values are stored
into a string separate vector. All of them are stored
in 'vec'.
@param[in] field COLUMN_NAME field in table_map_event.
@param[in] length length of the field
@param[out] column_metadata store all column metadata here (including
set/enum strings)
@param[in] field SET_STR_VALUE/ENUM_STR_VALUE field in table_map_event.
@param[in] length length of the field
@param[in] column_type type of the given column (i.e. MYSQL_TYPE_XYZ)
@param[in] var_to_set pointer-to-member selecting which str_vector to fill
@retval false ok
@retval true out of memory while reserving the str_vector
*/
static void parse_set_str_value(std::vector<Table_map_log_event::
Optional_metadata_fields::str_vector> &vec,
unsigned char *field, unsigned int length)
static bool parse_set_str_value(
Dynamic_array<Optional_column_metadata> &column_metadata,
unsigned char *field, unsigned int length,
uchar column_type,
Table_map_log_event::Optional_metadata_fields::str_vector
Optional_column_metadata::*var_to_set)
{
unsigned char* p= field;
unsigned char* end= field + length;
while (p < end)
for (uint col= 0; p < end && col < column_metadata.size(); col++)
{
if (column_metadata.at(col).column_type != column_type)
continue;
unsigned int count= net_field_length(&p);
if (unlikely(p > end))
return;
return 0;
auto* column_strings= &(column_metadata.at(col).*var_to_set);
if (column_strings->reserve(count))
return 1;
vec.push_back(std::vector<std::string>());
for (unsigned int i= 0; i < count; i++)
{
unsigned len1= net_field_length(&p);
if (unlikely(p + len1 > end))
return;
vec.back().push_back(std::string(reinterpret_cast<char *>(p), len1));
return 0;
column_strings->append(LEX_CSTRING{reinterpret_cast<char *>(p), len1});
p+= len1;
}
}
return 0;
}
/**
Parses GEOMETRY_TYPE field.
@param[out] vec stores geometry column's types extracted from field.
@param[out] column_metadata store all column metadata here (including
geometry types)
@param[in] field GEOMETRY_TYPE field in table_map_event.
@param[in] length length of the field
*/
static void parse_geometry_type(std::vector<unsigned int> &vec,
unsigned char *field, unsigned int length)
static void parse_geometry_type(
Dynamic_array<Optional_column_metadata> &column_metadata,
unsigned char *field, unsigned int length)
{
unsigned char* p= field;
unsigned char* end= field + length;
while (p < end)
for (unsigned int col = 0; p < end && col < column_metadata.size(); col++)
{
unsigned int geom_type= net_field_length(&p);
if (unlikely(p > end))
return;
vec.push_back(geom_type);
auto& col_metadata = column_metadata.at(col);
if (col_metadata.column_type == MYSQL_TYPE_GEOMETRY)
{
col_metadata.geometry_type = net_field_length(&p);
if (unlikely(p > end))
return;
}
}
}
/**
Parses SIMPLE_PRIMARY_KEY field.
@param[out] vec stores primary key's column information extracted from
field. Each column has an index and a prefix which are
stored as a unit_pair. prefix is always 0 for
SIMPLE_PRIMARY_KEY field.
@param[out] column_metadata store all column metadata here (including
primary key information). The prefix is always 0 for SIMPLE_PRIMARY_KEY.
@param[in] field SIMPLE_PRIMARY_KEY field in table_map_event.
@param[in] length length of the field
*/
static void parse_simple_pk(std::vector<Table_map_log_event::
Optional_metadata_fields::uint_pair> &vec,
unsigned char *field, unsigned int length)
static void parse_simple_pk(
Dynamic_array<Optional_column_metadata> &column_metadata,
unsigned char *field, unsigned int length)
{
unsigned char* p= field;
unsigned char* end= field + length;
@@ -4023,23 +4121,22 @@ static void parse_simple_pk(std::vector<Table_map_log_event::
unsigned int col_index= net_field_length(&p);
if (unlikely(p > end))
return;
vec.push_back(std::make_pair(col_index, (unsigned int) 0));
if (col_index < column_metadata.size())
column_metadata.at(col_index).primary_key = 0;
}
}
/**
Parses PRIMARY_KEY_WITH_PREFIX field.
@param[out] vec stores primary key's column information extracted from
field. Each column has an index and a prefix which are
stored as a unit_pair.
@param[out] column_metadata store all column metadata here (including
primary key lengths)
@param[in] field PRIMARY_KEY_WITH_PREFIX field in table_map_event.
@param[in] length length of the field
*/
static void parse_pk_with_prefix(std::vector<Table_map_log_event::
Optional_metadata_fields::uint_pair> &vec,
unsigned char *field, unsigned int length)
static void parse_pk_with_prefix(
Dynamic_array<Optional_column_metadata> &column_metadata,
unsigned char *field, unsigned int length)
{
unsigned char* p= field;
unsigned char* end= field + length;
@@ -4052,21 +4149,42 @@ static void parse_pk_with_prefix(std::vector<Table_map_log_event::
unsigned int col_prefix= net_field_length(&p);
if (unlikely(p > end))
return;
vec.push_back(std::make_pair(col_index, col_prefix));
if (col_index < column_metadata.size())
column_metadata.at(col_index).primary_key = col_prefix;
}
}
Table_map_log_event::Optional_metadata_fields::
Optional_metadata_fields(MEM_ROOT *root, uint master_columns,
const uchar* column_types,
const uchar* field_metadata,
uchar* optional_metadata,
size_t optional_metadata_len,
bool only_column_names)
: m_column_metadata(PSI_INSTRUMENT_MEM, master_columns)
{
unsigned int len;
uchar *metadata_end;
allocation_error= 0;
m_column_name= 0;
// initializing the elements of m_column_metadata and their types
for (uint i= 0; i < master_columns; i++)
{
Column_metadata col{};
col.column_type = column_types[i];
if (field_metadata)
{
if (col.column_type == MYSQL_TYPE_STRING &&
(*field_metadata == MYSQL_TYPE_ENUM ||
*field_metadata == MYSQL_TYPE_SET))
col.column_type= *field_metadata;
advance_field_metadata_ptr(col.column_type, &field_metadata);
}
if (m_column_metadata.append(col))
goto error;
}
if (optional_metadata == NULL)
return;
@@ -4084,42 +4202,42 @@ Optional_metadata_fields(MEM_ROOT *root, uint master_columns,
switch(type) {
case SIGNEDNESS:
parse_signedness(m_signedness, field, len);
parse_signedness(m_column_metadata, field, len);
break;
case DEFAULT_CHARSET:
parse_default_charset(m_default_charset, field, len);
parse_default_charset(*this, field, len);
break;
case COLUMN_CHARSET:
parse_column_charset(m_column_charset, field, len);
parse_column_charset(m_column_metadata, field, len);
break;
case COLUMN_NAME:
if (!(m_column_name= (LEX_CSTRING*) alloc_root(root,
sizeof(LEX_CSTRING) *
(master_columns +1 ))))
goto error;
if (parse_column_name(root, m_column_name, field, len))
if (parse_column_name(root, m_column_metadata, field, len))
goto error;
break;
case SET_STR_VALUE:
parse_set_str_value(m_set_str_value, field, len);
if (parse_set_str_value(m_column_metadata, field, len, MYSQL_TYPE_SET,
&Optional_column_metadata::set_str_values))
goto error;
break;
case ENUM_STR_VALUE:
parse_set_str_value(m_enum_str_value, field, len);
if (parse_set_str_value(m_column_metadata, field, len, MYSQL_TYPE_ENUM,
&Optional_column_metadata::enum_str_values))
goto error;
break;
case GEOMETRY_TYPE:
parse_geometry_type(m_geometry_type, field, len);
parse_geometry_type(m_column_metadata, field, len);
break;
case SIMPLE_PRIMARY_KEY:
parse_simple_pk(m_primary_key, field, len);
parse_simple_pk(m_column_metadata, field, len);
break;
case PRIMARY_KEY_WITH_PREFIX:
parse_pk_with_prefix(m_primary_key, field, len);
parse_pk_with_prefix(m_column_metadata, field, len);
break;
case ENUM_AND_SET_DEFAULT_CHARSET:
parse_default_charset(m_enum_and_set_default_charset, field, len);
parse_enum_and_set_default_charset(*this, field, len);
break;
case ENUM_AND_SET_COLUMN_CHARSET:
parse_column_charset(m_enum_and_set_column_charset, field, len);
parse_enum_and_set_column_charset(m_column_metadata, field, len);
break;
default:
DBUG_ASSERT(0);
@@ -4130,12 +4248,12 @@ Optional_metadata_fields(MEM_ROOT *root, uint master_columns,
error:
allocation_error= 1;
/*
m_column_name may be partially filled (remaining entries uninitialized)
if parsing failed midway. Reset it so callers checking "!m_column_name"
fall back to positional column mapping instead of reading uninitialized
LEX_CSTRING entries.
The column names may be partially filled if parsing failed midway. Clear
all of them so callers checking the first column's name fall back to
positional column mapping instead of using a half-parsed name list.
*/
m_column_name= 0;
for (size_t i= 0; i < m_column_metadata.size(); i++)
m_column_metadata.at(i).column_name= LEX_CSTRING{nullptr, 0};
return;
}

156
sql/log_event.h
View File

@@ -30,14 +30,53 @@
#define _log_event_h
#include <my_bitmap.h>
#include "mysql/psi/psi_base.h"
#include "rpl_constants.h"
#include <vector>
#include "sql_array.h"
#include <string>
#include <functional>
#include <memory>
#include <map>
#include <optional>
#include <lex_charset.h>
static inline bool is_numeric_type(uint type)
{
switch (type)
{
case MYSQL_TYPE_TINY:
case MYSQL_TYPE_SHORT:
case MYSQL_TYPE_INT24:
case MYSQL_TYPE_LONG:
case MYSQL_TYPE_LONGLONG:
case MYSQL_TYPE_NEWDECIMAL:
case MYSQL_TYPE_FLOAT:
case MYSQL_TYPE_DOUBLE:
case MYSQL_TYPE_YEAR:
return true;
default:
return false;
}
}
static inline bool is_character_type(uint type)
{
switch (type)
{
case MYSQL_TYPE_STRING:
case MYSQL_TYPE_VAR_STRING:
case MYSQL_TYPE_VARCHAR:
case MYSQL_TYPE_BLOB:
// Base class is blob for geom type
case MYSQL_TYPE_GEOMETRY:
return true;
default:
return false;
}
}
static inline bool is_enum_or_set_type(uint type)
{
return type == MYSQL_TYPE_ENUM || type == MYSQL_TYPE_SET;
}
#ifdef MYSQL_CLIENT
#include "sql_const.h"
#include "rpl_utility.h"
@@ -4434,6 +4473,42 @@ class table_def;
</tr>
</table>
*/
/**
Advances a field metadata pointer past the metadata for one column.
The advancement size depends on the column type, mirroring the layout
written by Field::save_field_metadata().
@param[in] type The resolved column type (e.g. MYSQL_TYPE_ENUM,
not MYSQL_TYPE_STRING, for enum columns)
@param[in,out] meta_ptr Pointer into the field metadata blob; advanced
in-place by the number of bytes consumed.
*/
inline void advance_field_metadata_ptr(uint type, const uchar** meta_ptr)
{
switch (type) {
case MYSQL_TYPE_FLOAT:
case MYSQL_TYPE_DOUBLE:
case MYSQL_TYPE_TIMESTAMP2:
case MYSQL_TYPE_DATETIME2:
case MYSQL_TYPE_TIME2:
case MYSQL_TYPE_BLOB:
case MYSQL_TYPE_GEOMETRY:
(*meta_ptr)++;
break;
case MYSQL_TYPE_NEWDECIMAL:
case MYSQL_TYPE_BIT:
case MYSQL_TYPE_ENUM:
case MYSQL_TYPE_SET:
case MYSQL_TYPE_VARCHAR:
case MYSQL_TYPE_VAR_STRING:
case MYSQL_TYPE_STRING:
(*meta_ptr) += 2;
break;
default:
break;
}
}
class Table_map_log_event : public Log_event
{
public:
@@ -4517,42 +4592,32 @@ public:
};
struct Optional_metadata_fields
{
typedef std::pair<unsigned int, unsigned int> uint_pair;
typedef std::vector<std::string> str_vector;
typedef Dynamic_array<LEX_CSTRING> str_vector;
bool allocation_error; /* Set if allocation of data structures fails */
struct Default_charset
struct Column_metadata
{
Default_charset() : default_charset(0) {}
bool empty() const { return default_charset == 0; }
LEX_CSTRING column_name{ nullptr, 0 };
uchar column_type{};
bool is_unsigned{ false };
// Default charset for the columns which are not in charset_pairs.
unsigned int default_charset;
// each str_vector stores values for the enum/set column
str_vector enum_str_values{PSI_INSTRUMENT_MEM, 0};
str_vector set_str_values{PSI_INSTRUMENT_MEM, 0};
/* The uint_pair means <column index, column charset number>. */
std::vector<uint_pair> charset_pairs;
uint geometry_type{ 0 };
// Prefix length of primary key (if applicable), where a value of 0
// indicates to use the entire column as a primary key
// (SIMPLE_PRIMARY_KEY).
std::optional<uint> primary_key{};
// 0 means no charset information present
uint charset{ 0 };
uint enum_and_set_column_charset{ 0 };
};
// Contents of DEFAULT_CHARSET field is converted into Default_charset.
Default_charset m_default_charset;
// Contents of ENUM_AND_SET_DEFAULT_CHARSET are converted into
// Default_charset.
Default_charset m_enum_and_set_default_charset;
std::vector<bool> m_signedness;
// Character set number of every string column
std::vector<unsigned int> m_column_charset;
// Character set number of every ENUM or SET column.
std::vector<unsigned int> m_enum_and_set_column_charset;
LEX_CSTRING *m_column_name;
// each str_vector stores values of one enum/set column
std::vector<str_vector> m_enum_str_value;
std::vector<str_vector> m_set_str_value;
std::vector<unsigned int> m_geometry_type;
/*
The uint_pair means <column index, prefix length>. Prefix length is 0 if
whole column value is used.
*/
std::vector<uint_pair> m_primary_key;
Dynamic_array<Column_metadata> m_column_metadata;
/*
It parses m_optional_metadata and populates into above variables.
@@ -4565,9 +4630,28 @@ public:
@param[in] only_column_names Only read column names
*/
Optional_metadata_fields(MEM_ROOT *root, uint master_cols,
const uchar* column_types,
const uchar* field_metadata,
uchar* optional_metadata,
size_t optional_metadata_len,
bool only_column_names);
~Optional_metadata_fields()
{
/*
Dynamic_array frees its own buffer in its destructor, but it does so
via delete_dynamic(), which only releases the flat element buffer and
does not run the element destructors. Each Column_metadata owns
heap-allocated str_vectors (enum_str_values / set_str_values), so
destroy every element explicitly to avoid leaking those buffers.
*/
for (size_t i= 0; i < m_column_metadata.size(); i++)
m_column_metadata.at(i).~Column_metadata();
}
/* Owns heap allocations; copying would lead to double frees. */
Optional_metadata_fields(const Optional_metadata_fields&)= delete;
Optional_metadata_fields& operator=(const Optional_metadata_fields&)= delete;
};
/**
@@ -4707,11 +4791,7 @@ private:
bool init_primary_key_field();
#endif
#ifdef MYSQL_CLIENT
class Charset_iterator;
class Default_charset_iterator;
class Column_charset_iterator;
#endif
char const *m_dbnam;
size_t m_dblen;
char const *m_tblnam;

264
sql/log_event_client.cc
View File

@@ -286,45 +286,6 @@ err:
return 1;
}
static inline bool is_numeric_type(uint type)
{
switch (type)
{
case MYSQL_TYPE_TINY:
case MYSQL_TYPE_SHORT:
case MYSQL_TYPE_INT24:
case MYSQL_TYPE_LONG:
case MYSQL_TYPE_LONGLONG:
case MYSQL_TYPE_NEWDECIMAL:
case MYSQL_TYPE_FLOAT:
case MYSQL_TYPE_DOUBLE:
return true;
default:
return false;
}
return false;
}
static inline bool is_character_type(uint type)
{
switch (type)
{
case MYSQL_TYPE_STRING:
case MYSQL_TYPE_VAR_STRING:
case MYSQL_TYPE_VARCHAR:
case MYSQL_TYPE_BLOB:
// Base class is blob for geom type
case MYSQL_TYPE_GEOMETRY:
return true;
default:
return false;
}
}
static inline bool is_enum_or_set_type(uint type) {
return type == MYSQL_TYPE_ENUM || type == MYSQL_TYPE_SET;
}
/*
Log_event::print_header()
@@ -3227,7 +3188,10 @@ bool Table_map_log_event::print(FILE *file, PRINT_EVENT_INFO *print_event_info)
{
MEM_ROOT root;
init_alloc_root(0, &root, 4096, 0, 0);
Optional_metadata_fields fields(&root, m_colcnt,
Optional_metadata_fields fields(&root,
m_colcnt,
m_coltype,
m_field_metadata,
m_optional_metadata,
m_optional_metadata_len,
0);
@@ -3268,93 +3232,7 @@ bool Table_map_log_event::print_body(PRINT_EVENT_INFO *print_event_info)
return 0;
}
/**
Interface for iterator over charset columns.
*/
class Table_map_log_event::Charset_iterator
{
public:
typedef Table_map_log_event::Optional_metadata_fields::Default_charset
Default_charset;
virtual const CHARSET_INFO *next()= 0;
virtual ~Charset_iterator(){};
/**
Factory method to create an instance of the appropriate subclass.
*/
static std::unique_ptr<Charset_iterator> create_charset_iterator(
const Default_charset &default_charset,
const std::vector<uint> &column_charset);
};
/**
Implementation of charset iterator for the DEFAULT_CHARSET type.
*/
class Table_map_log_event::Default_charset_iterator : public Charset_iterator
{
public:
Default_charset_iterator(const Default_charset &default_charset)
: m_iterator(default_charset.charset_pairs.begin()),
m_end(default_charset.charset_pairs.end()),
m_column_index(0),
m_default_charset_info(
get_charset(default_charset.default_charset, 0)) {}
const CHARSET_INFO *next() override {
const CHARSET_INFO *ret;
if (m_iterator != m_end && m_iterator->first == m_column_index) {
ret = get_charset(m_iterator->second, 0);
m_iterator++;
} else
ret = m_default_charset_info;
m_column_index++;
return ret;
}
~Default_charset_iterator(){};
private:
std::vector<Optional_metadata_fields::uint_pair>::const_iterator m_iterator,
m_end;
uint m_column_index;
const CHARSET_INFO *m_default_charset_info;
};
//Table_map_log_event::Default_charset_iterator::~Default_charset_iterator(){int a=8;a++; a--;};
/**
Implementation of charset iterator for the COLUMNT_CHARSET type.
*/
class Table_map_log_event::Column_charset_iterator : public Charset_iterator
{
public:
Column_charset_iterator(const std::vector<uint> &column_charset)
: m_iterator(column_charset.begin()), m_end(column_charset.end()) {}
const CHARSET_INFO *next() override {
const CHARSET_INFO *ret = nullptr;
if (m_iterator != m_end) {
ret = get_charset(*m_iterator, 0);
m_iterator++;
}
return ret;
}
~Column_charset_iterator(){};
private:
std::vector<uint>::const_iterator m_iterator;
std::vector<uint>::const_iterator m_end;
};
//Table_map_log_event::Column_charset_iterator::~Column_charset_iterator(){int a=8;a++; a--;};
std::unique_ptr<Table_map_log_event::Charset_iterator>
Table_map_log_event::Charset_iterator::create_charset_iterator(
const Default_charset &default_charset,
const std::vector<uint> &column_charset)
{
if (!default_charset.empty())
return std::unique_ptr<Charset_iterator>(
new Default_charset_iterator(default_charset));
else
return std::unique_ptr<Charset_iterator>(
new Column_charset_iterator(column_charset));
}
/**
return the string name of a type.
@@ -3389,43 +3267,36 @@ static void get_type_name(uint type, unsigned char** meta_ptr,
my_snprintf(typestr, typestr_length, "BIGINT");
break;
case MYSQL_TYPE_NEWDECIMAL:
my_snprintf(typestr, typestr_length, "DECIMAL(%d,%d)",
(*meta_ptr)[0], (*meta_ptr)[1]);
(*meta_ptr)+= 2;
break;
my_snprintf(typestr, typestr_length, "DECIMAL(%d,%d)",
(*meta_ptr)[0], (*meta_ptr)[1]);
break;
case MYSQL_TYPE_FLOAT:
my_snprintf(typestr, typestr_length, "FLOAT");
(*meta_ptr)++;
break;
case MYSQL_TYPE_DOUBLE:
my_snprintf(typestr, typestr_length, "DOUBLE");
(*meta_ptr)++;
break;
case MYSQL_TYPE_BIT:
my_snprintf(typestr, typestr_length, "BIT(%d)",
(((*meta_ptr)[0])) + (*meta_ptr)[1]*8);
(*meta_ptr)+= 2;
break;
case MYSQL_TYPE_TIMESTAMP2:
if (**meta_ptr != 0)
my_snprintf(typestr, typestr_length, "TIMESTAMP(%d)", **meta_ptr);
else
my_snprintf(typestr, typestr_length, "TIMESTAMP");
(*meta_ptr)++;
break;
case MYSQL_TYPE_DATETIME2:
if (**meta_ptr != 0)
my_snprintf(typestr, typestr_length, "DATETIME(%d)", **meta_ptr);
else
my_snprintf(typestr, typestr_length, "DATETIME");
(*meta_ptr)++;
break;
case MYSQL_TYPE_TIME2:
if (**meta_ptr != 0)
my_snprintf(typestr, typestr_length, "TIME(%d)", **meta_ptr);
else
my_snprintf(typestr, typestr_length, "TIME");
(*meta_ptr)++;
break;
case MYSQL_TYPE_NEWDATE:
case MYSQL_TYPE_DATE:
@@ -3436,11 +3307,9 @@ static void get_type_name(uint type, unsigned char** meta_ptr,
break;
case MYSQL_TYPE_ENUM:
my_snprintf(typestr, typestr_length, "ENUM");
(*meta_ptr)+= 2;
break;
case MYSQL_TYPE_SET:
my_snprintf(typestr, typestr_length, "SET");
(*meta_ptr)+= 2;
break;
case MYSQL_TYPE_BLOB:
{
@@ -3459,8 +3328,6 @@ static void get_type_name(uint type, unsigned char** meta_ptr,
my_snprintf(typestr, typestr_length, "INVALID_%s(%d)", type_name, size);
else
my_snprintf(typestr, typestr_length, "%s%s", names[size], type_name);
(*meta_ptr)++;
}
break;
case MYSQL_TYPE_VARCHAR:
@@ -3471,8 +3338,6 @@ static void get_type_name(uint type, unsigned char** meta_ptr,
else
my_snprintf(typestr, typestr_length, "VARBINARY(%d)",
uint2korr(*meta_ptr));
(*meta_ptr)+= 2;
break;
case MYSQL_TYPE_STRING:
{
@@ -3484,8 +3349,6 @@ static void get_type_name(uint type, unsigned char** meta_ptr,
my_snprintf(typestr, typestr_length, "CHAR(%d)", len/cs->mbmaxlen);
else
my_snprintf(typestr, typestr_length, "BINARY(%d)", len);
(*meta_ptr)+= 2;
}
break;
case MYSQL_TYPE_GEOMETRY:
@@ -3499,43 +3362,26 @@ static void get_type_name(uint type, unsigned char** meta_ptr,
else
my_snprintf(typestr, typestr_length, "INVALID_GEOMETRY_TYPE(%u)",
geometry_type);
(*meta_ptr)++;
}
break;
default:
*typestr= 0;
break;
}
advance_field_metadata_ptr(type, (const uchar**) meta_ptr);
}
void Table_map_log_event::print_columns(IO_CACHE *file,
const Optional_metadata_fields &fields)
{
unsigned char* field_metadata_ptr= m_field_metadata;
std::vector<bool>::const_iterator signedness_it= fields.m_signedness.begin();
std::unique_ptr<Charset_iterator> charset_it =
Charset_iterator::create_charset_iterator(fields.m_default_charset,
fields.m_column_charset);
std::unique_ptr<Charset_iterator> enum_and_set_charset_it =
Charset_iterator::create_charset_iterator(
fields.m_enum_and_set_default_charset,
fields.m_enum_and_set_column_charset);
std::vector<Optional_metadata_fields::str_vector>::const_iterator
set_str_values_it= fields.m_set_str_value.begin();
std::vector<Optional_metadata_fields::str_vector>::const_iterator
enum_str_values_it= fields.m_enum_str_value.begin();
std::vector<unsigned int>::const_iterator geometry_type_it=
fields.m_geometry_type.begin();
LEX_CSTRING *col_names= fields.m_column_name;
uint geometry_type= 0;
my_b_printf(file, "# Columns(");
for (unsigned long i= 0; i < m_colcnt; i++)
{
uint real_type = m_coltype[i];
const Optional_metadata_fields::Column_metadata& column_metadata= fields.m_column_metadata.at(i);
uint real_type= m_coltype[i];
if (real_type == MYSQL_TYPE_STRING &&
(*field_metadata_ptr == MYSQL_TYPE_ENUM ||
*field_metadata_ptr == MYSQL_TYPE_SET))
@@ -3543,32 +3389,31 @@ void Table_map_log_event::print_columns(IO_CACHE *file,
// Get current column's collation id if it is a character, enum,
// or set column
const CHARSET_INFO *cs = NULL;
const CHARSET_INFO *cs= NULL;
if (is_character_type(real_type))
cs = charset_it->next();
else if (is_enum_or_set_type(real_type))
cs = enum_and_set_charset_it->next();
// Print column name
if (col_names && col_names->str)
{
pretty_print_identifier(file, col_names->str, col_names->length);
my_b_printf(file, " ");
col_names++;
if (column_metadata.charset)
cs= get_charset(column_metadata.charset, 0);
}
else if (is_enum_or_set_type(real_type))
{
if (column_metadata.enum_and_set_column_charset)
cs= get_charset(column_metadata.enum_and_set_column_charset, 0);
}
// update geometry_type for geometry columns
if (real_type == MYSQL_TYPE_GEOMETRY)
// Print column name
const LEX_CSTRING& col_name= column_metadata.column_name;
if (col_name.str)
{
geometry_type= (geometry_type_it != fields.m_geometry_type.end()) ?
*geometry_type_it++ : 0;
pretty_print_identifier(file, col_name.str, col_name.length);
my_b_printf(file, " ");
}
// print column type
const uint TYPE_NAME_LEN = 100;
char type_name[TYPE_NAME_LEN];
get_type_name(real_type, &field_metadata_ptr, cs, type_name,
TYPE_NAME_LEN, geometry_type);
TYPE_NAME_LEN, column_metadata.geometry_type);
if (type_name[0] == '\0')
{
@@ -3578,12 +3423,10 @@ void Table_map_log_event::print_columns(IO_CACHE *file,
my_b_printf(file, "%s", type_name);
// Print UNSIGNED for numeric column
if (is_numeric_type(real_type) &&
signedness_it != fields.m_signedness.end())
if (is_numeric_type(real_type))
{
if (*signedness_it == true)
if (column_metadata.is_unsigned == true)
my_b_printf(file, " UNSIGNED");
signedness_it++;
}
// if the column is not marked as 'null', print 'not null'
@@ -3592,27 +3435,23 @@ void Table_map_log_event::print_columns(IO_CACHE *file,
// Print string values of SET and ENUM column
const Optional_metadata_fields::str_vector *str_values= NULL;
if (real_type == MYSQL_TYPE_ENUM &&
enum_str_values_it != fields.m_enum_str_value.end())
if (real_type == MYSQL_TYPE_ENUM && !column_metadata.enum_str_values.is_empty())
{
str_values= &(*enum_str_values_it);
enum_str_values_it++;
str_values= &column_metadata.enum_str_values;
}
else if (real_type == MYSQL_TYPE_SET &&
set_str_values_it != fields.m_set_str_value.end())
else if (real_type == MYSQL_TYPE_SET && !column_metadata.set_str_values.is_empty())
{
str_values= &(*set_str_values_it);
set_str_values_it++;
str_values= &column_metadata.set_str_values;
}
if (str_values != NULL)
{
const char *separator= "(";
for (Optional_metadata_fields::str_vector::const_iterator it=
str_values->begin(); it != str_values->end(); it++)
for (const LEX_CSTRING *it= str_values->front(); it != str_values->end();
it++)
{
my_b_printf(file, "%s", separator);
pretty_print_str(file, it->c_str(), it->size());
pretty_print_str(file, it->str, it->length);
separator= ",";
}
my_b_printf(file, ")");
@@ -3631,27 +3470,40 @@ void Table_map_log_event::print_columns(IO_CACHE *file,
void Table_map_log_event::print_primary_key
(IO_CACHE *file,const Optional_metadata_fields &fields)
{
if (!fields.m_primary_key.empty())
auto& col_metadata= fields.m_column_metadata;
bool has_pk= false;
for (uint i = 0; i < col_metadata.size(); i++) {
if (col_metadata.at(i).primary_key.has_value()) {
has_pk= true;
break;
}
}
if (has_pk)
{
my_b_printf(file, "# Primary Key(");
std::vector<Optional_metadata_fields::uint_pair>::const_iterator it=
fields.m_primary_key.begin();
for (; it != fields.m_primary_key.end(); it++)
const char *delimiter = "";
for (uint col_idx= 0; col_idx < col_metadata.size(); col_idx++)
{
if (it != fields.m_primary_key.begin())
my_b_printf(file, ", ");
const auto& col= col_metadata.at(col_idx);
if (!col.primary_key.has_value())
continue;
my_b_printf(file, delimiter);
// Print column name or column index
if (!fields.m_column_name)
my_b_printf(file, "%u", it->first);
const LEX_CSTRING& column_name= col.column_name;
if (column_name.str)
my_b_printf(file, "%s", column_name.str);
else
my_b_printf(file, "%s", fields.m_column_name[it->first].str);
my_b_printf(file, "%u", col_idx);
// Print prefix length
if (it->second != 0)
my_b_printf(file, "(%u)", it->second);
if (col.primary_key.value() != 0)
my_b_printf(file, "(%u)", col.primary_key.value());
delimiter= ", ";
}
my_b_printf(file, ")\n");

1
sql/log_event_server.cc
View File

@@ -61,6 +61,7 @@
#include "rpl_constants.h"
#include "sql_digest.h"
#include "zlib.h"
#include <map>
#define log_cs &my_charset_latin1

3
sql/rpl_utility.h
View File

@@ -100,6 +100,9 @@ public:
in a Table_map event.
*/
static uint field_metadata_length(uint binlog_type);
const uchar *field_types() const { return m_type; }
/*
Return a representation of the type data for one field.

7
sql/rpl_utility_server.cc
View File

@@ -1359,10 +1359,13 @@ default_column_mapping:
Table_map_log_event::Optional_metadata_fields
opt_metadata(rgi->thd->mem_root, master_cols,
m_tabledef.field_types(),
nullptr,
(uchar*) m_tabledef.optional_metadata.str,
m_tabledef.optional_metadata.length, 1);
if (!opt_metadata.m_column_name)
if (!opt_metadata.m_column_metadata.is_empty() &&
!opt_metadata.m_column_metadata.at(0).column_name.str)
{
/*
If there are no column names provided in the optional metadata
@@ -1374,7 +1377,7 @@ default_column_mapping:
for (uint col= 0; col < master_cols; col++)
{
const LEX_CSTRING *field_name= &opt_metadata.m_column_name[col];
const LEX_CSTRING *field_name= &opt_metadata.m_column_metadata.at(col).column_name;
Field *field= table->find_field_by_name(field_name);
if (unlikely(!field))
{

2
sql/sql_array.h
View File

@@ -197,6 +197,8 @@ public:
size_t size() const { return array.elements; }
bool is_empty() const { return array.elements == 0; }
const Elem *end() const
{
return back() + 1;