valitydev/clickhouse-sink-connector
14
0
Fork 0
mirror of https://github.com/valitydev/clickhouse-sink-connector.git synced 2024-11-06 02:25:18 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Fix to add Nullable to ALTER TABLE ADD COLUMN column_name.

Browse Source
This commit is contained in:
Kanthi Subramanian 2023-04-12 16:04:14 -04:00
parent 3033b8a502
commit 95fba26249
5 changed files with 62 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/Constants.java
View File

@ -3,7 +3,7 @@ package com.altinity.clickhouse.debezium.embedded.ddl.parser;
public class Constants {
public static final String ALTER_TABLE = "ALTER TABLE %s";
public static final String CREATE_TABLE = "CREATE TABLE %s";
public static final String CREATE_TABLE = "CREATE TABLE";
public static final String NULLABLE = "Nullable";
@ -27,6 +27,8 @@ public class Constants {
public static final String NOT_NULL = "notnull";
public static final String NULL = "NULL";
public static final String IF_EXISTS = "if exists ";
public static final String IF_NOT_EXISTS = "if not exists ";
public static final String RENAME_TABLE = "RENAME TABLE %s to %s";
public static final String TRUNCATE_TABLE = "TRUNCATE TABLE %s";
public static final String DROP_TABLE = "DROP TABLE";

22
sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySqlDDLParserListenerImpl.java
View File

@ -289,14 +289,14 @@ public class MySqlDDLParserListenerImpl implements MySqlParserListener {
String modifierWithNull = Constants.ADD_COLUMN_NULLABLE;
String defaultModifier = null;
this.query.append(Constants.CREATE_TABLE).append(" ");
for (ParseTree tree : pt) {
if (tree instanceof TableNameContext) {
this.query.append(String.format(Constants.CREATE_TABLE, tree.getText())).append("(");
}
// HashMap<String, String> colNamesToDataTypeMap = new HashMap<String, String>();
if (tree instanceof MySqlParser.CreateDefinitionsContext) {
this.query.append(tree.getText()).append("(");
}else if(tree instanceof MySqlParser.IfNotExistsContext) {
this.query.append(Constants.IF_NOT_EXISTS);
}else if (tree instanceof MySqlParser.CreateDefinitionsContext) {
for (ParseTree subtree : ((MySqlParser.CreateDefinitionsContext) tree).children) {
if (subtree instanceof TerminalNodeImpl) {
// this.query.append(subtree.getText());
@ -1695,6 +1695,7 @@ public class MySqlDDLParserListenerImpl implements MySqlParserListener {
if (tree instanceof AlterByAddColumnContext) {
modifier = Constants.ADD_COLUMN;
modifierWithNull = Constants.ADD_COLUMN_NULLABLE;
isNullColumn = true;
} else if (tree instanceof MySqlParser.AlterByModifyColumnContext) {
modifier = Constants.MODIFY_COLUMN;
@ -1732,6 +1733,9 @@ public class MySqlDDLParserListenerImpl implements MySqlParserListener {
if (columnDefChild instanceof MySqlParser.NullColumnConstraintContext) {
if (columnDefChild.getText().equalsIgnoreCase(Constants.NULL))
isNullColumn = true;
else if(columnDefChild.getText().equalsIgnoreCase(Constants.NOT_NULL)) {
isNullColumn = false;
}
} else if (columnDefChild instanceof MySqlParser.DefaultColumnConstraintContext) {
if (columnDefChild.getChildCount() >= 2) {
defaultModifier = "DEFAULT " + columnDefChild.getChild(1).getText();
@ -1778,6 +1782,8 @@ public class MySqlDDLParserListenerImpl implements MySqlParserListener {
postProcessModifyColumn(this.tableName, columnName, newColumnName, columnType);
}
String trimmedQuery = this.query.toString().trim();
this.query.delete(0, this.query.toString().length()).append(trimmedQuery);
}
/**
@ -2433,16 +2439,18 @@ public class MySqlDDLParserListenerImpl implements MySqlParserListener {
@Override
public void enterDropTable(MySqlParser.DropTableContext dropTableContext) {
log.debug("DROP TABLE enter");
this.query.append(Constants.DROP_TABLE);
this.query.append(Constants.DROP_TABLE).append(" ");
for (ParseTree child : dropTableContext.children) {
if (child instanceof MySqlParser.TablesContext) {
for (ParseTree tableNameChild : ((MySqlParser.TablesContext) child).children) {
if (tableNameChild instanceof MySqlParser.TableNameContext) {
this.query.append(" ").append(tableNameChild.getText());
this.query.append(tableNameChild.getText());
} else if (tableNameChild instanceof TerminalNodeImpl) {
this.query.append(tableNameChild.getText());
}
}
} else if(child instanceof MySqlParser.IfExistsContext) {
this.query.append(Constants.IF_EXISTS);
}
}
}

6
sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/ClickHouseDebeziumEmbeddedDDLAddColumnIT.java
View File

@ -79,14 +79,14 @@ public class ClickHouseDebeziumEmbeddedDDLAddColumnIT extends ClickHouseDebezium
Map<String, String> addTestColumns = writer.getColumnsDataTypesForTable("add_test");
// Validate all ship_class columns.
Assert.assertTrue(shipClassColumns.get("ship_spec").equalsIgnoreCase("String"));
Assert.assertTrue(shipClassColumns.get("somecol").equalsIgnoreCase("Int32"));
Assert.assertTrue(shipClassColumns.get("ship_spec").equalsIgnoreCase("Nullable(String)"));
Assert.assertTrue(shipClassColumns.get("somecol").equalsIgnoreCase("Nullable(Int32)"));
Assert.assertTrue(shipClassColumns.get("newcol").equalsIgnoreCase("Nullable(Bool)"));
Assert.assertTrue(shipClassColumns.get("customer_address").equalsIgnoreCase("String"));
Assert.assertTrue(shipClassColumns.get("customer_name").equalsIgnoreCase("Nullable(String)"));
// Validate all add_test columns.
Assert.assertTrue(addTestColumns.get("col8").equalsIgnoreCase("String"));
Assert.assertTrue(addTestColumns.get("col8").equalsIgnoreCase("Nullable(String)"));
Assert.assertTrue(addTestColumns.get("col2").equalsIgnoreCase("Nullable(Int32)"));
Assert.assertTrue(addTestColumns.get("col3").equalsIgnoreCase("Nullable(Int32)"));

2
sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/ClickHouseDebeziumEmbeddedDDLModifyColumnIT.java
View File

@ -71,7 +71,7 @@ public class ClickHouseDebeziumEmbeddedDDLModifyColumnIT extends ClickHouseDebez
Map<String, String> addTestColumns = writer.getColumnsDataTypesForTable("add_test");
Assert.assertTrue(shipClassColumns.get("class_name").equalsIgnoreCase("Int32"));
Assert.assertTrue(shipClassColumns.get("tonange").equalsIgnoreCase("Nullable(Decimal(10, 10))"));
Assert.assertTrue(shipClassColumns.get("tonange").equalsIgnoreCase("Decimal(10, 10)"));
Assert.assertTrue(addTestColumns.get("col1").equalsIgnoreCase("Int32"));
Assert.assertTrue(addTestColumns.get("col2").equalsIgnoreCase("Int32"));

50
sink-connector-lightweight/src/test/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySqlDDLParserListenerImplTest.java
View File

@ -13,11 +13,11 @@ public class MySqlDDLParserListenerImplTest {
@Test
public void testCreateTable() {
StringBuffer clickHouseQuery = new StringBuffer();
String createDB = "create table ship_class(id int, class_name varchar(100), tonange decimal(10,2), max_length decimal(10,2), start_build year, end_build year(4), max_guns_size int)";
String createDB = "create table if not exists ship_class(id int, class_name varchar(100), tonange decimal(10,2), max_length decimal(10,2), start_build year, end_build year(4), max_guns_size int)";
MySQLDDLParserService mySQLDDLParserService = new MySQLDDLParserService();
mySQLDDLParserService.parseSql(createDB, "Persons", clickHouseQuery);
Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE ship_class(id Nullable(Int32),class_name Nullable(String),tonange Nullable(Decimal(10,2)),max_length Nullable(Decimal(10,2)),start_build Nullable(Int32),end_build Nullable(Int32),max_guns_size Nullable(Int32),`_sign` Int8,`_version` UInt64) Engine=ReplacingMergeTree(_version) ORDER BY tuple()"));
Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE if not exists ship_class(id Nullable(Int32),class_name Nullable(String),tonange Nullable(Decimal(10,2)),max_length Nullable(Decimal(10,2)),start_build Nullable(Int32),end_build Nullable(Int32),max_guns_size Nullable(Int32),`_sign` Int8,`_version` UInt64) Engine=ReplacingMergeTree(_version) ORDER BY tuple()"));
log.info("Create table " + clickHouseQuery);
}
@ -40,7 +40,7 @@ public class MySqlDDLParserListenerImplTest {
MySQLDDLParserService mySQLDDLParserService = new MySQLDDLParserService();
mySQLDDLParserService.parseSql(createDBQuery, "Persons", clickHouseQuery);
Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE 730b595f_d475_11ed_b64a_398b553542b2(id Nullable(Int32),x Nullable(Int32),`_sign` Int8,`_version` UInt64) Engine=ReplacingMergeTree(_version) ORDER BY (id)"));
Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE if not exists 730b595f_d475_11ed_b64a_398b553542b2(id Nullable(Int32),x Nullable(Int32),`_sign` Int8,`_version` UInt64) Engine=ReplacingMergeTree(_version) ORDER BY (id)"));
log.info("Create table " + clickHouseQuery);
}
@ -59,13 +59,14 @@ public class MySqlDDLParserListenerImplTest {
MySQLDDLParserService mySQLDDLParserService = new MySQLDDLParserService();
mySQLDDLParserService.parseSql(createDB, "Persons", clickHouseQuery);
Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("CREATE TABLE `salaries`(`emp_no` Int32,`salary` Int32,`from_date` Date32,`to_date` Date32,`_sign` Int8,`_version` UInt64) Engine=ReplacingMergeTree(_version) ORDER BY (`emp_no`,`from_date`)"));
log.info("Create table query" + clickHouseQuery.toString());
}
@Test
public void testAlterDatabaseAddColumn() {
String clickhouseExpectedQuery = "ALTER TABLE employees ADD COLUMN ssn_number String";
String clickhouseExpectedQuery = "ALTER TABLE employees ADD COLUMN ssn_number Nullable(String)";
StringBuffer clickHouseQuery = new StringBuffer();
String alterDBAddColumn = "ALTER TABLE employees add column ssn_number varchar(100)";
MySQLDDLParserService mySQLDDLParserService = new MySQLDDLParserService();
@ -94,7 +95,7 @@ public class MySqlDDLParserListenerImplTest {
// Before, After
@Test
public void testAlterDatabaseAddMultipleColumns1() {
String expectedClickHouseQuery = "ALTER TABLE employees ADD COLUMN ship_spec String first, ADD COLUMN somecol Int32 after start_build,";
String expectedClickHouseQuery = "ALTER TABLE employees ADD COLUMN ship_spec Nullable(String) first, ADD COLUMN somecol Nullable(Int32) after start_build,";
StringBuffer clickHouseQuery = new StringBuffer();
String query = "alter table employees add column ship_spec varchar(150) first, add somecol int after start_build, algorithm=instant;";
MySQLDDLParserService mySQLDDLParserService = new MySQLDDLParserService();
@ -108,7 +109,7 @@ public class MySqlDDLParserListenerImplTest {
@Test
public void testAlterDatabaseAddMultipleColumns() {
String expectedClickHouseQuery = "ALTER TABLE employees ADD COLUMN ssn_number String, ADD COLUMN home_address String";
String expectedClickHouseQuery = "ALTER TABLE employees ADD COLUMN ssn_number Nullable(String), ADD COLUMN home_address Nullable(String)";
StringBuffer clickHouseQuery = new StringBuffer();
String alterDBAddColumn = "ALTER TABLE employees add column ssn_number varchar(100), add column home_address varchar(20)";
MySQLDDLParserService mySQLDDLParserService = new MySQLDDLParserService();
@ -138,14 +139,14 @@ public class MySqlDDLParserListenerImplTest {
MySQLDDLParserService mySQLDDLParserService = new MySQLDDLParserService();
mySQLDDLParserService.parseSql(mysqlQuery, "add_test", clickHouseQuery);
String expectedCHQuery = "ALTER TABLE add_test ADD COLUMN customer_address String, ADD COLUMN customer_name Nullable(String) ";
String expectedCHQuery = "ALTER TABLE add_test ADD COLUMN customer_address String, ADD COLUMN customer_name Nullable(String)";
Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase(expectedCHQuery));
log.info("CLICKHOUSE QUERY: " + clickHouseQuery);
}
@Test
public void testAddDefault() {
String expectedClickHouseQuery = "ALTER TABLE add_test ADD COLUMN foo Int32 DEFAULT 2";
String expectedClickHouseQuery = "ALTER TABLE add_test ADD COLUMN foo Nullable(Int32) DEFAULT 2";
String mysqlQuery = "ALTER TABLE add_test ADD COLUMN foo INT DEFAULT 2;";
StringBuffer clickHouseQuery = new StringBuffer();
MySQLDDLParserService mySQLDDLParserService = new MySQLDDLParserService();
@ -156,6 +157,20 @@ public class MySqlDDLParserListenerImplTest {
Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase(expectedClickHouseQuery));
}
@Test
public void testAddColumnWithoutExplicitNull() {
String expectedClickHouseQuery = "ALTER TABLE add_test ADD COLUMN foo Nullable(Int32)";
String mysqlQuery = "ALTER TABLE add_test ADD COLUMN foo INT;";
StringBuffer clickHouseQuery = new StringBuffer();
MySQLDDLParserService mySQLDDLParserService = new MySQLDDLParserService();
mySQLDDLParserService.parseSql(mysqlQuery, "add_test", clickHouseQuery);
log.info("CLICKHOUSE QUERY: " + clickHouseQuery);
Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase(expectedClickHouseQuery));
}
@Test
public void testAlterDatabaseModifyColumns() {
@ -166,7 +181,9 @@ public class MySqlDDLParserListenerImplTest {
mySQLDDLParserService.parseSql(alterDBAddColumn, "contacts", clickHouseQuery);
//Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("ALTER TABLE contacts MODIFY COLUMN last_name Nullable(String)"));
log.info("CLICKHOUSE QUERY" + clickHouseQuery);
//
Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("ALTER TABLE contacts MODIFY COLUMN last_name Nullable(String) \n" +
"ALTER TABLE contacts RENAME COLUMN last_name to new_name"));
//
// StringBuffer clickHouseQuery2 = new StringBuffer();
// String alterDBModifyColumn = "ALTER TABLE contacts\n" +
// " MODIFY last_name varchar(55) NULL\n" +
@ -186,11 +203,13 @@ public class MySqlDDLParserListenerImplTest {
MySQLDDLParserService mySQLDDLParserService2 = new MySQLDDLParserService();
mySQLDDLParserService2.parseSql(sql, "products", clickHouseQuery);
Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("ALTER TABLE products ADD COLUMN stocks Int32"));
StringBuffer clickHouseQuery2 = new StringBuffer();
String defaultSql = "alter table add_test add column stocks bool null default 1;";
MySQLDDLParserService mySQLDDLParserService3 = new MySQLDDLParserService();
mySQLDDLParserService3.parseSql(defaultSql, "add_test", clickHouseQuery2);
Assert.assertTrue(clickHouseQuery2.toString().equalsIgnoreCase("ALTER TABLE add_test ADD COLUMN stocks Nullable(Bool) DEFAULT 1"));
}
@ -316,6 +335,17 @@ public class MySqlDDLParserListenerImplTest {
Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase(sql));
}
@Test
public void dropTableIfExists() {
StringBuffer clickHouseQuery = new StringBuffer();
String sql = "drop table if exists add_test";
MySQLDDLParserService mySQLDDLParserService2 = new MySQLDDLParserService();
mySQLDDLParserService2.parseSql(sql, "table1", clickHouseQuery);
Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase(sql));
}
@Test
public void dropMultipleTables() {
StringBuffer clickHouseQuery = new StringBuffer();
@ -324,7 +354,7 @@ public class MySqlDDLParserListenerImplTest {
MySQLDDLParserService mySQLDDLParserService2 = new MySQLDDLParserService();
mySQLDDLParserService2.parseSql(sql, "table1", clickHouseQuery);
Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase(sql));
Assert.assertTrue(clickHouseQuery.toString().equalsIgnoreCase("drop table add_test,add_test2"));
}
@Test