Merge pull request #889 from Altinity/885-remove-clickhouse-calls-to-get-column-data-type-for-every-batch-insert

Remove additional call to get column data types from clickhouse

Author: subkanthi

doc/configuration.md

@@ -2,6 +2,7 @@
 ### Configuration Reference
  Configuration                                             | Description                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
 |-----------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| topic.prefix                                              | Unique name for connector.                                                                                     
 | database.hostname                                         | Source Database HostName                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             |
 | database.port                                             | Source Database Port number                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
 | database.user                                             | Source Database Username(user needs to have replication permission, Refer https://debezium.io/documentation/reference/stable/connectors/mysql.html)                                                                                                          GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED BY 'password';                                                                                                                                                  |

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/batch/GroupInsertQueryWithBatchRecords.java

@@ -64,7 +64,7 @@ public boolean groupQueryWithRecords(List<ClickHouseStruct> records,
                         log.error("**** ERROR ALTER TABLE: " + tableName, e);
                     }
                 }
-                columnNameToDataTypeMap = new DBMetadata().getColumnsDataTypesForTable(tableName, connection, databaseName, config );
+                //columnNameToDataTypeMap = new DBMetadata().getColumnsDataTypesForTable(tableName, connection, databaseName, config );
                 result = updateQueryToRecordsMap(record, record.getAfterModifiedFields(), queryToRecordsMap, tableName, config, columnNameToDataTypeMap);
             } else if(CdcRecordState.CDC_RECORD_STATE_BOTH == getCdcSectionBasedOnOperation(record.getCdcOperation()))  {
                 if(record.getBeforeModifiedFields() != null) {