Merge pull request #555 from Altinity/fix_kafka_multiple_database

Kafka fixes for multiple database.
Altinity · Apr 25, 2024 · 5dd3c72 · 5dd3c72
2 parents 74503dd + e416d80
commit 5dd3c72
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 5 deletions.
diff --git a/.../java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAutoCreateTable.java b/.../java/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAutoCreateTable.java
@@ -49,7 +49,7 @@ public java.lang.String createTableSyntax(ArrayList<String> primaryKey, String t
 
         StringBuilder createTableSyntax = new StringBuilder();
 
-        createTableSyntax.append(CREATE_TABLE).append(" ").append("`").append(databaseName).append(".").append(tableName).append("`").append("(");
+        createTableSyntax.append(CREATE_TABLE).append(" ").append(databaseName).append(".").append("`").append(tableName).append("`").append("(");
 
         for(Field f: fields) {
             String colName = f.name();

diff --git a/...a/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAutoCreateTableTest.java b/...a/com/altinity/clickhouse/sink/connector/db/operations/ClickHouseAutoCreateTableTest.java
@@ -127,7 +127,7 @@ public void testCreateTableSyntax() {
         String query = act.createTableSyntax(primaryKeys, "auto_create_table", "employees",
                 createFields(), this.columnToDataTypesMap, false, false);
         System.out.println("QUERY" + query);
-        Assert.assertTrue(query.equalsIgnoreCase("CREATE TABLE `employees.auto_create_table`(`customerName` String NOT NULL,`occupation` String NOT NULL,`quantity` Int32 NOT NULL,`amount_1` Float32 NOT NULL,`amount` Float64 NOT NULL,`employed` Bool NOT NULL,`blob_storage` String NOT NULL,`blob_storage_scale` Decimal NOT NULL,`json_output` JSON,`max_amount` Float64 NOT NULL,`_sign` Int8,`_version` UInt64) ENGINE = ReplacingMergeTree(_version) PRIMARY KEY(customerName) ORDER BY(customerName)"));
+        Assert.assertTrue(query.equalsIgnoreCase("CREATE TABLE employees.`auto_create_table`(`customerName` String NOT NULL,`occupation` String NOT NULL,`quantity` Int32 NOT NULL,`amount_1` Float32 NOT NULL,`amount` Float64 NOT NULL,`employed` Bool NOT NULL,`blob_storage` String NOT NULL,`blob_storage_scale` Decimal NOT NULL,`json_output` JSON,`max_amount` Float64 NOT NULL,`_sign` Int8,`_version` UInt64) ENGINE = ReplacingMergeTree(_version) PRIMARY KEY(customerName) ORDER BY(customerName)"));
         //Assert.assertTrue(query.equalsIgnoreCase("CREATE TABLE auto_create_table(`customerName` String NOT NULL,`occupation` String NOT NULL,`quantity` Int32 NOT NULL,`amount_1` Float32 NOT NULL,`amount` Float64 NOT NULL,`employed` Bool NOT NULL,`blob_storage` String NOT NULL,`blob_storage_scale` Decimal NOT NULL,`json_output` JSON,`max_amount` Float64 NOT NULL,`_sign` Int8,`_version` UInt64) ENGINE = ReplacingMergeTree(_version) PRIMARY KEY(customerName) ORDER BY (customerName)"));
     }
 
@@ -139,7 +139,7 @@ public void testCreateTableEmptyPrimaryKey() {
         String query = act.createTableSyntax(null, "auto_create_table", "employees", createFields(),
                 this.columnToDataTypesMap, false, false);
 
-        String expectedQuery = "CREATE TABLE `employees.auto_create_table`(`customerName` String NOT NULL,`occupation` String NOT NULL,`quantity` Int32 NOT NULL,`amount_1` Float32 NOT NULL,`amount` Float64 NOT NULL,`employed` Bool NOT NULL,`blob_storage` String NOT NULL,`blob_storage_scale` Decimal NOT NULL,`json_output` JSON,`max_amount` Float64 NOT NULL,`_sign` Int8,`_version` UInt64) ENGINE = ReplacingMergeTree(_version) ORDER BY tuple()";
+        String expectedQuery = "CREATE TABLE employees.`auto_create_table`(`customerName` String NOT NULL,`occupation` String NOT NULL,`quantity` Int32 NOT NULL,`amount_1` Float32 NOT NULL,`amount` Float64 NOT NULL,`employed` Bool NOT NULL,`blob_storage` String NOT NULL,`blob_storage_scale` Decimal NOT NULL,`json_output` JSON,`max_amount` Float64 NOT NULL,`_sign` Int8,`_version` UInt64) ENGINE = ReplacingMergeTree(_version) ORDER BY tuple()";
         Assert.assertTrue(query.equalsIgnoreCase(expectedQuery));
     }
     @Test
@@ -153,7 +153,7 @@ public void testCreateTableMultiplePrimaryKeys() {
         String query = act.createTableSyntax(primaryKeys, "auto_create_table", "customers", createFields(),
                 this.columnToDataTypesMap, false, false);
 
-        String expectedQuery = "CREATE TABLE `customers.auto_create_table`(`customerName` String NOT NULL,`occupation` String NOT NULL,`quantity` Int32 NOT NULL,`amount_1` Float32 NOT NULL,`amount` Float64 NOT NULL,`employed` Bool NOT NULL,`blob_storage` String NOT NULL,`blob_storage_scale` Decimal NOT NULL,`json_output` JSON,`max_amount` Float64 NOT NULL,`_sign` Int8,`_version` UInt64) ENGINE = ReplacingMergeTree(_version) ORDER BY tuple()";
+        String expectedQuery = "CREATE TABLE customers.`auto_create_table`(`customerName` String NOT NULL,`occupation` String NOT NULL,`quantity` Int32 NOT NULL,`amount_1` Float32 NOT NULL,`amount` Float64 NOT NULL,`employed` Bool NOT NULL,`blob_storage` String NOT NULL,`blob_storage_scale` Decimal NOT NULL,`json_output` JSON,`max_amount` Float64 NOT NULL,`_sign` Int8,`_version` UInt64) ENGINE = ReplacingMergeTree(_version) ORDER BY tuple()";
         Assert.assertTrue(query.equalsIgnoreCase(expectedQuery));
         System.out.println(query);
     }
@@ -180,7 +180,7 @@ public void testCreateNewTable() {
         primaryKeys.add("customerName");
 
         try {
-            act.createNewTable(primaryKeys, "auto_create_table", "products", this.createFields(), writer.getConnection(),
+            act.createNewTable(primaryKeys, "auto_create_table", "default", this.createFields(), writer.getConnection(),
                     false, false);
         } catch(SQLException se) {
             Assert.assertTrue(false);