thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-21 23:21:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Materialize my sql support enum data type

Browse Source 
* Implemented Enum for MaterializeMySQL
This commit is contained in:
Storozhuk Kostiantyn 2021-06-24 15:05:47 +03:00 committed by Kostiantyn Storozhuk
parent 4fe722d1a4
commit 4a3145f586
16 changed files with 182 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
programs/odbc-bridge/ODBCBlockInputStream.cpp
View File

@ -115,6 +115,8 @@ void ODBCBlockInputStream::insertValue(
assert_cast<ColumnFloat64 &>(column).insertValue(row.get<double>(idx));
break;
case ValueType::vtFixedString:[[fallthrough]];
case ValueType::vtEnum8:
case ValueType::vtEnum16:
case ValueType::vtString:
assert_cast<ColumnString &>(column).insert(row.get<std::string>(idx));
break;

4
src/Core/ExternalResultDescription.cpp
View File

@ -67,9 +67,9 @@ void ExternalResultDescription::init(const Block & sample_block_)
else if (which.isUUID())
types.emplace_back(ValueType::vtUUID, is_nullable);
else if (which.isEnum8())
types.emplace_back(ValueType::vtString, is_nullable);
types.emplace_back(ValueType::vtEnum8, is_nullable);
else if (which.isEnum16())
types.emplace_back(ValueType::vtString, is_nullable);
types.emplace_back(ValueType::vtEnum16, is_nullable);
else if (which.isDateTime64())
types.emplace_back(ValueType::vtDateTime64, is_nullable);
else if (which.isDecimal32())

2
src/Core/ExternalResultDescription.h
View File

@ -22,6 +22,8 @@ struct ExternalResultDescription
vtInt64,
vtFloat32,
vtFloat64,
vtEnum8,
vtEnum16,
vtString,
vtDate,
vtDateTime,

17
src/Core/MySQL/MySQLReplication.cpp
View File

@ -298,7 +298,6 @@ namespace MySQLReplication
}
/// Types that do not used in the binlog event:
/// MYSQL_TYPE_ENUM
/// MYSQL_TYPE_SET
/// MYSQL_TYPE_TINY_BLOB
/// MYSQL_TYPE_MEDIUM_BLOB
@ -562,6 +561,22 @@ namespace MySQLReplication
row.push_back(dispatch((meta >> 8) & 0xFF, meta & 0xFF, read_decimal));
break;
}
case MYSQL_TYPE_ENUM:
{
if((meta & 0xFF) == 1)
{
UInt8 val = 0;
payload.readStrict(reinterpret_cast<char *>(&val), 1);
row.push_back(Field{UInt8{val}});
}
else
{
UInt16 val = 0;
payload.readStrict(reinterpret_cast<char *>(&val), 2);
row.push_back(Field{UInt16{val}});
}
break;
}
case MYSQL_TYPE_VARCHAR:
case MYSQL_TYPE_VAR_STRING:
{

2
src/DataStreams/MongoDBBlockInputStream.cpp
View File

@ -243,6 +243,8 @@ namespace
insertNumber<Float64>(column, value, name);
break;
case ValueType::vtEnum8:
case ValueType::vtEnum16:
case ValueType::vtString:
{
if (value.type() == Poco::MongoDB::ElementTraits<ObjectId::Ptr>::TypeId)

2
src/DataStreams/PostgreSQLBlockInputStream.cpp
View File

@ -157,6 +157,8 @@ void PostgreSQLBlockInputStream::insertValue(IColumn & column, std::string_view
assert_cast<ColumnFloat64 &>(column).insertValue(pqxx::from_string<double>(value));
break;
case ValueType::vtFixedString:[[fallthrough]];
case ValueType::vtEnum8:
case ValueType::vtEnum16:
case ValueType::vtString:
assert_cast<ColumnString &>(column).insertData(value.data(), value.size());
break;

3
src/DataTypes/DataTypeEnum.cpp
View File

@ -262,6 +262,9 @@ void registerDataTypeEnum(DataTypeFactory & factory)
factory.registerDataType("Enum8", createExact<DataTypeEnum<Int8>>);
factory.registerDataType("Enum16", createExact<DataTypeEnum<Int16>>);
factory.registerDataType("Enum", create);
/// MySQL
factory.registerAlias("ENUM", "Enum", DataTypeFactory::CaseInsensitive);
}
}

2
src/Dictionaries/CassandraBlockInputStream.cpp
View File

@ -110,6 +110,8 @@ void CassandraBlockInputStream::insertValue(IColumn & column, ValueType type, co
assert_cast<ColumnFloat64 &>(column).insertValue(value);
break;
}
case ValueType::vtEnum8:
case ValueType::vtEnum16:
case ValueType::vtString:
{
const char * value = nullptr;

2
src/Dictionaries/RedisBlockInputStream.cpp
View File

@ -91,6 +91,8 @@ namespace DB
case ValueType::vtFloat64:
insert<Float64>(column, string_value);
break;
case ValueType::vtEnum8:
case ValueType::vtEnum16:
case ValueType::vtString:
assert_cast<ColumnString &>(column).insert(parse<String>(string_value));
break;

9
src/Formats/MySQLBlockInputStream.cpp
View File

@ -10,6 +10,7 @@
#include <Columns/ColumnDecimal.h>
#include <Columns/ColumnFixedString.h>
#include <DataTypes/IDataType.h>
#include <DataTypes/DataTypeEnum.h>
#include <DataTypes/DataTypeNullable.h>
#include <IO/ReadBufferFromString.h>
#include <IO/ReadHelpers.h>
@ -157,6 +158,14 @@ namespace
assert_cast<ColumnFloat64 &>(column).insertValue(value.getDouble());
read_bytes_size += 8;
break;
case ValueType::vtEnum8:
assert_cast<ColumnInt8 &>(column).insertValue(static_cast<const DataTypeEnum<Int8> &>(data_type).castToValue(value.data()).get<Int64>());
read_bytes_size += assert_cast<ColumnInt8 &>(column).byteSize();
break;
case ValueType::vtEnum16:
assert_cast<ColumnInt16 &>(column).insertValue(static_cast<const DataTypeEnum<Int16> &>(data_type).castToValue(value.data()).get<Int64>());
read_bytes_size += assert_cast<ColumnInt16 &>(column).byteSize();
break;
case ValueType::vtString:
assert_cast<ColumnString &>(column).insertData(value.data(), value.size());
read_bytes_size += assert_cast<ColumnString &>(column).byteSize();

32
src/Interpreters/MySQL/InterpretersMySQLDDLQuery.cpp
View File

@ -92,22 +92,40 @@ static NamesAndTypesList getColumnsList(const ASTExpressionList * columns_defini
}
ASTPtr data_type = declare_column->data_type;
auto * data_type_function = data_type->as<ASTFunction>();
if (is_unsigned)
if (data_type_function)
{
auto * data_type_function = data_type->as<ASTFunction>();
String type_name_upper = Poco::toUpper(data_type_function->name);
if (data_type_function)
if (is_unsigned)
{
String type_name_upper = Poco::toUpper(data_type_function->name);
/// For example(in MySQL): CREATE TABLE test(column_name INT NOT NULL ... UNSIGNED)
if (type_name_upper.find("INT") != std::string::npos && !endsWith(type_name_upper, "SIGNED")
if (type_name_upper.find("INT") != String::npos && !endsWith(type_name_upper, "SIGNED")
&& !endsWith(type_name_upper, "UNSIGNED"))
data_type_function->name = type_name_upper + " UNSIGNED";
}
}
/// Transforms MySQL ENUM's list of strings to ClickHouse string-integer pairs
/// For example ENUM('a', 'b', 'c') -> ENUM('a'=1, 'b'=2, 'c'=3)
/// Elements on a position further than 32767 are assigned negative values, starting with -32768.
/// Note: Enum would be transfomed to Enum8 if number of ellements is less then 128, otherwise it would be transformed to Enum16.
if (type_name_upper.find("ENUM") != String::npos)
{
UInt16 i = 0;
for (ASTPtr & child : data_type_function->arguments->children)
{
auto newChild = std::make_shared<ASTFunction>();
newChild->name = "equals";
auto * literal = child->as<ASTLiteral>();
newChild->arguments = std::make_shared<ASTExpressionList>();
newChild->arguments->children.push_back(std::make_shared<ASTLiteral>(literal->value.get<String>()));
newChild->arguments->children.push_back(std::make_shared<ASTLiteral>(Int16(++i)));
child = newChild;
}
}
}
if (is_nullable)
data_type = makeASTFunction("Nullable", data_type);

22
src/Interpreters/MySQL/tests/gtest_create_rewritten.cpp
View File

@ -235,3 +235,25 @@ TEST(MySQLCreateRewritten, QueryWithColumnComments)
std::string(MATERIALIZEMYSQL_TABLE_COLUMNS) +
") ENGINE = ReplacingMergeTree(_version) PARTITION BY intDiv(key, 4294967) ORDER BY tuple(key)");
}
TEST(MySQLCreateRewritten, QueryWithEnum)
{
tryRegisterFunctions();
const auto & context_holder = getContext();
EXPECT_EQ(queryToString(tryRewrittenCreateQuery(
"CREATE TABLE `test_database`.`test_table_1`(`key` INT NOT NULL PRIMARY KEY, `test` ENUM('a','b','c'))", context_holder.context)),
"CREATE TABLE test_database.test_table_1 (`key` Int32, `test` Nullable(Enum8('a' = 1, 'b' = 2, 'c' = 3))" +
std::string(MATERIALIZEMYSQL_TABLE_COLUMNS) +
") ENGINE = ReplacingMergeTree(_version) PARTITION BY intDiv(key, 4294967) ORDER BY tuple(key)");
EXPECT_EQ(queryToString(tryRewrittenCreateQuery(
"CREATE TABLE `test_database`.`test_table_1`(`key` INT NOT NULL PRIMARY KEY, `test` ENUM('a','b','c') NOT NULL)", context_holder.context)),
"CREATE TABLE test_database.test_table_1 (`key` Int32, `test` Enum8('a' = 1, 'b' = 2, 'c' = 3)" +
std::string(MATERIALIZEMYSQL_TABLE_COLUMNS) +
") ENGINE = ReplacingMergeTree(_version) PARTITION BY intDiv(key, 4294967) ORDER BY tuple(key)");
EXPECT_EQ(queryToString(tryRewrittenCreateQuery(
"CREATE TABLE `test_database`.`test_table_1`(`key` INT NOT NULL PRIMARY KEY, `test` ENUM('a','b','c') COMMENT 'test_comment')", context_holder.context)),
"CREATE TABLE test_database.test_table_1 (`key` Int32, `test` Nullable(Enum8('a' = 1, 'b' = 2, 'c' = 3)) COMMENT 'test_comment'" +
std::string(MATERIALIZEMYSQL_TABLE_COLUMNS) +
") ENGINE = ReplacingMergeTree(_version) PARTITION BY intDiv(key, 4294967) ORDER BY tuple(key)");
}

2
tests/integration/parallel.json
View File

@ -177,6 +177,8 @@
"test_materialize_mysql_database/test.py::test_system_tables_table[clickhouse_node1]",
"test_materialize_mysql_database/test.py::test_materialize_with_column_comments[clickhouse_node0]",
"test_materialize_mysql_database/test.py::test_materialize_with_column_comments[clickhouse_node1]",
"test_materialize_mysql_database/test.py::test_materialize_with_enum[clickhouse_node0]",
"test_materialize_mysql_database/test.py::test_materialize_with_enum[clickhouse_node1]",
"test_materialize_mysql_database/test.py::test_utf8mb4[clickhouse_node0]",
"test_materialize_mysql_database/test.py::test_utf8mb4[clickhouse_node1]",
"test_parts_delete_zookeeper/test.py::test_merge_doesnt_work_without_zookeeper",

2
tests/integration/parallel_skip.json
View File

@ -181,6 +181,8 @@
"test_materialize_mysql_database/test.py::test_system_tables_table[clickhouse_node1]",
"test_materialize_mysql_database/test.py::test_materialize_with_column_comments[clickhouse_node0]",
"test_materialize_mysql_database/test.py::test_materialize_with_column_comments[clickhouse_node1]",
"test_materialize_mysql_database/test.py::test_materialize_with_enum[clickhouse_node0]",
"test_materialize_mysql_database/test.py::test_materialize_with_enum[clickhouse_node1]",
"test_materialize_mysql_database/test.py::test_utf8mb4[clickhouse_node0]",
"test_materialize_mysql_database/test.py::test_utf8mb4[clickhouse_node1]",
"test_parts_delete_zookeeper/test.py::test_merge_doesnt_work_without_zookeeper",

80
tests/integration/test_materialize_mysql_database/materialize_with_ddl.py
View File

@ -853,6 +853,86 @@ def materialize_with_column_comments_test(clickhouse_node, mysql_node, service_n
clickhouse_node.query("DROP DATABASE materialize_with_column_comments_test")
mysql_node.query("DROP DATABASE materialize_with_column_comments_test")
def materialize_with_enum8_test(clickhouse_node, mysql_node, service_name):
mysql_node.query("DROP DATABASE IF EXISTS materialize_with_enum8_test")
clickhouse_node.query("DROP DATABASE IF EXISTS materialize_with_enum8_test")
mysql_node.query("CREATE DATABASE materialize_with_enum8_test")
enum8_values_count = 127
enum8_values = ""
enum8_values_with_backslash = ""
for i in range(1, enum8_values_count):
enum8_values += '\'' + str(i) + "\', "
enum8_values_with_backslash += "\\\'" + str(i) +"\\\' = " + str(i) + ", "
enum8_values += '\'' + str(enum8_values_count) + '\''
enum8_values_with_backslash += "\\\'" + str(enum8_values_count) +"\\\' = " + str(enum8_values_count)
mysql_node.query("CREATE TABLE materialize_with_enum8_test.test (id int NOT NULL PRIMARY KEY, value ENUM(" + enum8_values + ")) ENGINE=InnoDB")
mysql_node.query("INSERT INTO materialize_with_enum8_test.test (id, value) VALUES (1, '1'),(2, '2')")
clickhouse_node.query("CREATE DATABASE materialize_with_enum8_test ENGINE = MaterializeMySQL('{}:3306', 'materialize_with_enum8_test', 'root', 'clickhouse')".format(service_name))
check_query(clickhouse_node, "SELECT value FROM materialize_with_enum8_test.test ORDER BY id", "1\n2\n")
mysql_node.query("INSERT INTO materialize_with_enum8_test.test (id, value) VALUES (3, '127')")
check_query(clickhouse_node, "SELECT value FROM materialize_with_enum8_test.test ORDER BY id", "1\n2\n127\n")
check_query(clickhouse_node, "DESCRIBE TABLE materialize_with_enum8_test.test", "id\tInt32\t\t\t\t\t\nvalue\tNullable(Enum8(" + enum8_values_with_backslash + "))\t\t\t\t\t\n_sign\tInt8\tMATERIALIZED\t1\t\t\t\n_version\tUInt64\tMATERIALIZED\t1\t\t\t\n")
clickhouse_node.query("DROP DATABASE materialize_with_enum8_test")
mysql_node.query("DROP DATABASE materialize_with_enum8_test")
def materialize_with_enum16_test(clickhouse_node, mysql_node, service_name):
mysql_node.query("DROP DATABASE IF EXISTS materialize_with_enum16_test")
clickhouse_node.query("DROP DATABASE IF EXISTS materialize_with_enum16_test")
mysql_node.query("CREATE DATABASE materialize_with_enum16_test")
enum16_values_count = 600
enum16_values = ""
enum16_values_with_backslash = ""
for i in range(1, enum16_values_count):
enum16_values += '\'' + str(i) + "\', "
enum16_values_with_backslash += "\\\'" + str(i) +"\\\' = " + str(i) + ", "
enum16_values += '\'' + str(enum16_values_count) + '\''
enum16_values_with_backslash += "\\\'" + str(enum16_values_count) +"\\\' = " + str(enum16_values_count)
mysql_node.query("CREATE TABLE materialize_with_enum16_test.test (id int NOT NULL PRIMARY KEY, value ENUM(" + enum16_values + ")) ENGINE=InnoDB")
mysql_node.query("INSERT INTO materialize_with_enum16_test.test (id, value) VALUES (1, '1'),(2, '2')")
clickhouse_node.query("CREATE DATABASE materialize_with_enum16_test ENGINE = MaterializeMySQL('{}:3306', 'materialize_with_enum16_test', 'root', 'clickhouse')".format(service_name))
check_query(clickhouse_node, "SELECT value FROM materialize_with_enum16_test.test ORDER BY id", "1\n2\n")
mysql_node.query("INSERT INTO materialize_with_enum16_test.test (id, value) VALUES (3, '500')")
check_query(clickhouse_node, "SELECT value FROM materialize_with_enum16_test.test ORDER BY id", "1\n2\n500\n")
check_query(clickhouse_node, "DESCRIBE TABLE materialize_with_enum16_test.test", "id\tInt32\t\t\t\t\t\nvalue\tNullable(Enum16(" + enum16_values_with_backslash + "))\t\t\t\t\t\n_sign\tInt8\tMATERIALIZED\t1\t\t\t\n_version\tUInt64\tMATERIALIZED\t1\t\t\t\n")
clickhouse_node.query("DROP DATABASE materialize_with_enum16_test")
mysql_node.query("DROP DATABASE materialize_with_enum16_test")
def alter_enum8_to_enum16_test(clickhouse_node, mysql_node, service_name):
mysql_node.query("DROP DATABASE IF EXISTS alter_enum8_to_enum16_test")
clickhouse_node.query("DROP DATABASE IF EXISTS alter_enum8_to_enum16_test")
mysql_node.query("CREATE DATABASE alter_enum8_to_enum16_test")
enum8_values_count = 100
enum8_values = ""
enum8_values_with_backslash = ""
for i in range(1, enum8_values_count):
enum8_values += '\'' + str(i) + "\', "
enum8_values_with_backslash += "\\\'" + str(i) +"\\\' = " + str(i) + ", "
enum8_values += '\'' + str(enum8_values_count) + '\''
enum8_values_with_backslash += "\\\'" + str(enum8_values_count) +"\\\' = " + str(enum8_values_count)
mysql_node.query("CREATE TABLE alter_enum8_to_enum16_test.test (id int NOT NULL PRIMARY KEY, value ENUM(" + enum8_values + ")) ENGINE=InnoDB")
mysql_node.query("INSERT INTO alter_enum8_to_enum16_test.test (id, value) VALUES (1, '1'),(2, '2')")
clickhouse_node.query("CREATE DATABASE alter_enum8_to_enum16_test ENGINE = MaterializeMySQL('{}:3306', 'alter_enum8_to_enum16_test', 'root', 'clickhouse')".format(service_name))
mysql_node.query("INSERT INTO alter_enum8_to_enum16_test.test (id, value) VALUES (3, '75')")
check_query(clickhouse_node, "SELECT value FROM alter_enum8_to_enum16_test.test ORDER BY id", "1\n2\n75\n")
check_query(clickhouse_node, "DESCRIBE TABLE alter_enum8_to_enum16_test.test", "id\tInt32\t\t\t\t\t\nvalue\tNullable(Enum8(" + enum8_values_with_backslash + "))\t\t\t\t\t\n_sign\tInt8\tMATERIALIZED\t1\t\t\t\n_version\tUInt64\tMATERIALIZED\t1\t\t\t\n")
enum16_values_count = 600
enum16_values = ""
enum16_values_with_backslash = ""
for i in range(1, enum16_values_count):
enum16_values += '\'' + str(i) + "\', "
enum16_values_with_backslash += "\\\'" + str(i) +"\\\' = " + str(i) + ", "
enum16_values += '\'' + str(enum16_values_count) + '\''
enum16_values_with_backslash += "\\\'" + str(enum16_values_count) +"\\\' = " + str(enum16_values_count)
mysql_node.query("ALTER TABLE alter_enum8_to_enum16_test.test MODIFY COLUMN value ENUM(" + enum16_values + ")")
check_query(clickhouse_node, "DESCRIBE TABLE alter_enum8_to_enum16_test.test", "id\tInt32\t\t\t\t\t\nvalue\tNullable(Enum16(" + enum16_values_with_backslash + "))\t\t\t\t\t\n_sign\tInt8\tMATERIALIZED\t1\t\t\t\n_version\tUInt64\tMATERIALIZED\t1\t\t\t\n")
mysql_node.query("INSERT INTO alter_enum8_to_enum16_test.test (id, value) VALUES (4, '500')")
check_query(clickhouse_node, "SELECT value FROM alter_enum8_to_enum16_test.test ORDER BY id", "1\n2\n75\n500\n")
clickhouse_node.query("DROP DATABASE alter_enum8_to_enum16_test")
mysql_node.query("DROP DATABASE alter_enum8_to_enum16_test")
def move_to_prewhere_and_column_filtering(clickhouse_node, mysql_node, service_name):
clickhouse_node.query("DROP DATABASE IF EXISTS cond_on_key_col")
mysql_node.query("DROP DATABASE IF EXISTS cond_on_key_col")

9
tests/integration/test_materialize_mysql_database/test.py
View File

@ -223,6 +223,15 @@ def test_materialize_with_column_comments(started_cluster, started_mysql_8_0, st
materialize_with_ddl.materialize_with_column_comments_test(clickhouse_node, started_mysql_5_7, "mysql57")
materialize_with_ddl.materialize_with_column_comments_test(clickhouse_node, started_mysql_8_0, "mysql80")
@pytest.mark.parametrize(('clickhouse_node'), [node_db_ordinary, node_db_ordinary])
def test_materialize_with_enum(started_cluster, started_mysql_8_0, started_mysql_5_7, clickhouse_node):
materialize_with_ddl.materialize_with_enum8_test(clickhouse_node, started_mysql_5_7, "mysql57")
materialize_with_ddl.materialize_with_enum16_test(clickhouse_node, started_mysql_5_7, "mysql57")
materialize_with_ddl.alter_enum8_to_enum16_test(clickhouse_node, started_mysql_5_7, "mysql57")
materialize_with_ddl.materialize_with_enum8_test(clickhouse_node, started_mysql_8_0, "mysql80")
materialize_with_ddl.materialize_with_enum16_test(clickhouse_node, started_mysql_8_0, "mysql80")
materialize_with_ddl.alter_enum8_to_enum16_test(clickhouse_node, started_mysql_8_0, "mysql80")
@pytest.mark.parametrize(('clickhouse_node'), [node_disable_bytes_settings, node_disable_rows_settings])
def test_mysql_settings(started_cluster, started_mysql_8_0, started_mysql_5_7, clickhouse_node):