Added logic to support multiple databases

doc/img/mysql_ddl_execution

@@ -0,0 +1,53 @@
+<mxfile host="app.diagrams.net" modified="2024-03-05T11:52:46.132Z" agent="Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" etag="6-FdwkM_er-bKnZmaXG9" version="23.0.2" type="device">
+  <diagram name="Page-1" id="4kWa008UeLW88EDx8_1b">
+    <mxGraphModel dx="1360" dy="759" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="850" pageHeight="1100" math="0" shadow="0">
+      <root>
+        <mxCell id="0" />
+        <mxCell id="1" parent="0" />
+        <mxCell id="Kt27N8K72PIJ8C6K6JpF-4" value="" style="whiteSpace=wrap;html=1;aspect=fixed;" vertex="1" parent="1">
+          <mxGeometry x="160" y="40" width="470" height="470" as="geometry" />
+        </mxCell>
+        <mxCell id="Kt27N8K72PIJ8C6K6JpF-7" style="edgeStyle=orthogonalEdgeStyle;rounded=0;orthogonalLoop=1;jettySize=auto;html=1;entryX=0;entryY=0.5;entryDx=0;entryDy=0;" edge="1" parent="1" source="Kt27N8K72PIJ8C6K6JpF-2" target="Kt27N8K72PIJ8C6K6JpF-6">
+          <mxGeometry relative="1" as="geometry" />
+        </mxCell>
+        <mxCell id="Kt27N8K72PIJ8C6K6JpF-2" value="" style="outlineConnect=0;dashed=0;verticalLabelPosition=bottom;verticalAlign=top;align=center;html=1;shape=mxgraph.aws3.mysql_db_instance_2;fillColor=#2E73B8;gradientColor=none;" vertex="1" parent="1">
+          <mxGeometry x="30" y="170" width="60" height="80" as="geometry" />
+        </mxCell>
+        <mxCell id="Kt27N8K72PIJ8C6K6JpF-5" value="ClickHouse" style="shape=cylinder3;whiteSpace=wrap;html=1;boundedLbl=1;backgroundOutline=1;size=15;fillColor=#fff2cc;strokeColor=#d6b656;" vertex="1" parent="1">
+          <mxGeometry x="730" y="240" width="60" height="80" as="geometry" />
+        </mxCell>
+        <mxCell id="Kt27N8K72PIJ8C6K6JpF-14" style="edgeStyle=orthogonalEdgeStyle;rounded=0;orthogonalLoop=1;jettySize=auto;html=1;" edge="1" parent="1" source="Kt27N8K72PIJ8C6K6JpF-6" target="Kt27N8K72PIJ8C6K6JpF-10">
+          <mxGeometry relative="1" as="geometry" />
+        </mxCell>
+        <mxCell id="Kt27N8K72PIJ8C6K6JpF-6" value="Main&lt;br&gt;Thread&lt;br&gt;(DDL)" style="rounded=0;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+          <mxGeometry x="200" y="200" width="120" height="60" as="geometry" />
+        </mxCell>
+        <mxCell id="Kt27N8K72PIJ8C6K6JpF-10" value="&lt;b&gt;Thread&amp;nbsp;&lt;br&gt;Pool&lt;/b&gt;" style="rounded=0;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+          <mxGeometry x="430" y="80" width="180" height="370" as="geometry" />
+        </mxCell>
+        <mxCell id="Kt27N8K72PIJ8C6K6JpF-8" value="Thread-1" style="rounded=0;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+          <mxGeometry x="460" y="100" width="120" height="60" as="geometry" />
+        </mxCell>
+        <mxCell id="Kt27N8K72PIJ8C6K6JpF-11" value="Thread-2" style="rounded=0;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+          <mxGeometry x="460" y="180" width="120" height="60" as="geometry" />
+        </mxCell>
+        <mxCell id="Kt27N8K72PIJ8C6K6JpF-12" value="Thread-3" style="rounded=0;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+          <mxGeometry x="460" y="300" width="120" height="60" as="geometry" />
+        </mxCell>
+        <mxCell id="Kt27N8K72PIJ8C6K6JpF-13" value="Thread-4" style="rounded=0;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+          <mxGeometry x="460" y="380" width="120" height="60" as="geometry" />
+        </mxCell>
+        <mxCell id="Kt27N8K72PIJ8C6K6JpF-15" value="&lt;b&gt;Sink Connector Lighweight&lt;/b&gt;" style="text;html=1;strokeColor=none;fillColor=none;align=center;verticalAlign=middle;whiteSpace=wrap;rounded=0;" vertex="1" parent="1">
+          <mxGeometry x="330" y="520" width="150" height="30" as="geometry" />
+        </mxCell>
+        <mxCell id="Kt27N8K72PIJ8C6K6JpF-16" value="" style="endArrow=classic;html=1;rounded=0;entryX=0;entryY=0.5;entryDx=0;entryDy=0;entryPerimeter=0;exitX=0.996;exitY=0.516;exitDx=0;exitDy=0;exitPerimeter=0;" edge="1" parent="1" source="Kt27N8K72PIJ8C6K6JpF-4" target="Kt27N8K72PIJ8C6K6JpF-5">
+          <mxGeometry width="50" height="50" relative="1" as="geometry">
+            <mxPoint x="400" y="420" as="sourcePoint" />
+            <mxPoint x="450" y="370" as="targetPoint" />
+            <Array as="points" />
+          </mxGeometry>
+        </mxCell>
+      </root>
+    </mxGraphModel>
+  </diagram>
+</mxfile>

sink-connector-lightweight/pom.xml

@@ -23,7 +23,6 @@
         <quarkus.platform.group-id>io.quarkus.platform</quarkus.platform.group-id>
         <quarkus.platform.version>2.14.0.Final</quarkus.platform.version>
         <surefire-plugin.version>3.0.0-M7</surefire-plugin.version>
-        <apache.httpclient.version>5.2.1</apache.httpclient.version>
         <sink-connector-library-version>0.0.8</sink-connector-library-version>
     </properties>
     <dependencyManagement>
@@ -187,11 +186,11 @@
             <groupId>org.apache.maven</groupId>
             <artifactId>maven-artifact</artifactId>
         </dependency>
-        <!-- <dependency>
+        <dependency>
             <groupId>org.apache.logging.log4j</groupId>
             <artifactId>log4j-api</artifactId>
             <version>2.17.1</version>
-        </dependency> -->
+        </dependency>
         <dependency>
             <groupId>org.antlr</groupId>
             <artifactId>antlr4-runtime</artifactId>

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumChangeEventCapture.java

@@ -83,31 +83,59 @@ public DebeziumChangeEventCapture() {
         singleThreadDebeziumEventExecutor = Executors.newFixedThreadPool(1);
     }
 
-    private void performDDLOperation(String DDL, Properties props, SourceRecord sr, ClickHouseSinkConnectorConfig config) {
-
-        DBCredentials dbCredentials = parseDBConfiguration(config);
-        if (writer == null) {
+    // Function to store the DBwriter instance for a given database.
+    private Map<String, BaseDbWriter> dbWriterMap = new HashMap<>();
+
+    // Function to retrieve the DBWriter from dbWriterMap for a given database.
+    // if it exists or create a new one.
+    private BaseDbWriter getDbWriter(String databaseName, ClickHouseSinkConnectorConfig config) {
+        if(dbWriterMap.containsKey(databaseName)) {
+            return dbWriterMap.get(databaseName);
+        } else {
+            DBCredentials dbCredentials = parseDBConfiguration(config);
             String jdbcUrl = BaseDbWriter.getConnectionString(dbCredentials.getHostName(), dbCredentials.getPort(),
-                        dbCredentials.getDatabase());
-            conn = BaseDbWriter.createConnection(jdbcUrl, "Client_1",
+                    databaseName);
+            ClickHouseConnection conn = BaseDbWriter.createConnection(jdbcUrl, "Client_1",
                     dbCredentials.getUserName(), dbCredentials.getPassword(), config);
+            BaseDbWriter writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
+                    databaseName, dbCredentials.getUserName(),
+                    dbCredentials.getPassword(), config, conn);
+            dbWriterMap.put(databaseName, writer);
+            return writer;
+        }
+    }
 
-            writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
-                    dbCredentials.getDatabase(), dbCredentials.getUserName(),
-                    dbCredentials.getPassword(), config, this.conn);
+    private void performDDLOperation(String DDL, Properties props, SourceRecord sr, ClickHouseSinkConnectorConfig config) {
 
+        String databaseName = "system";
+        if(sr != null && sr.key() != null) {
+            if(sr.key() instanceof Struct) {
+                Struct keyStruct = (Struct) sr.key();
+                //System.out.println("Do something");
+                String recordDbName = (String) keyStruct.get("databaseName");
+                if(recordDbName != null && recordDbName.isEmpty() == false) {
+                    databaseName = recordDbName;
+                }
+            }
         }
 
-        try {
-            String clickHouseVersion = writer.getClickHouseVersion();
-            isNewReplacingMergeTreeEngine = new com.altinity.clickhouse.sink.connector.db.DBMetadata()
-                    .checkIfNewReplacingMergeTree(clickHouseVersion);
-        } catch (Exception e) {
-            log.error("Error retrieving version");
+        if(writer == null) {
+
+            DBCredentials dbCredentials = parseDBConfiguration(config);
+            String jdbcUrl = BaseDbWriter.getConnectionString(dbCredentials.getHostName(), dbCredentials.getPort(),
+                    databaseName);
+            ClickHouseConnection conn = BaseDbWriter.createConnection(jdbcUrl, "Client_1",
+                    dbCredentials.getUserName(), dbCredentials.getPassword(), config);
+            writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
+                    databaseName, dbCredentials.getUserName(),
+                    dbCredentials.getPassword(), config, conn);
         }
+        //BaseDbWriter writer = getDbWriter(databaseName, config);
+
+
         StringBuffer clickHouseQuery = new StringBuffer();
         AtomicBoolean isDropOrTruncate = new AtomicBoolean(false);
-        MySQLDDLParserService mySQLDDLParserService = new MySQLDDLParserService(config);
+        MySQLDDLParserService mySQLDDLParserService = new MySQLDDLParserService(config, databaseName);
         mySQLDDLParserService.parseSql(DDL, "", clickHouseQuery, isDropOrTruncate);
         ClickHouseAlterTable cat = new ClickHouseAlterTable();
 
@@ -118,6 +146,7 @@ private void performDDLOperation(String DDL, Properties props, SourceRecord sr,
             log.info("Executed Source DB DDL: " + DDL + " Snapshot:" + isSnapshotDDL(sr));
         }
 
+
         long currentTime = System.currentTimeMillis();
         boolean ddlProcessingResult = true;
         Metrics.updateDdlMetrics(DDL, currentTime, 0, ddlProcessingResult);
@@ -144,6 +173,15 @@ private void performDDLOperation(String DDL, Properties props, SourceRecord sr,
             ddlProcessingResult = false;
             //throw new RuntimeException(e);
         }
+
+        try {
+            String clickHouseVersion = writer.getClickHouseVersion();
+            isNewReplacingMergeTreeEngine = new com.altinity.clickhouse.sink.connector.db.DBMetadata()
+                    .checkIfNewReplacingMergeTree(clickHouseVersion);
+        } catch (Exception e) {
+            log.error("Error retrieving version");
+        }
+
         long elapsedTime = System.currentTimeMillis() - currentTime;
         Metrics.updateDdlMetrics(DDL, currentTime, elapsedTime, ddlProcessingResult);
     }
@@ -189,15 +227,13 @@ private ClickHouseStruct processEveryChangeRecord(Properties props, ChangeEvent<
 
 
                 if (DDL != null && DDL.isEmpty() == false)
-                //&& ((DDL.toUpperCase().contains("ALTER TABLE") || DDL.toUpperCase().contains("RENAME TABLE"))))
                 {
                     log.info("***** DDL received, Flush all existing records");
                     this.executor.shutdown();
                     this.executor.awaitTermination(60, TimeUnit.SECONDS);
 
-
                     performDDLOperation(DDL, props, sr, config);
-                    setupProcessingThread(config, new MySQLDDLParserService(config));
+                    setupProcessingThread(config);
                 }
 
             } else {
@@ -283,14 +319,14 @@ private boolean checkIfDDLNeedsToBeIgnored(Properties props, SourceRecord sr, At
     private void createDatabaseForDebeziumStorage(ClickHouseSinkConnectorConfig config, Properties props) {
         try {
             DBCredentials dbCredentials = parseDBConfiguration(config);
-            if (writer == null) {
+            //if (writer == null) {
                 String jdbcUrl = BaseDbWriter.getConnectionString(dbCredentials.getHostName(), dbCredentials.getPort(),
                         dbCredentials.getDatabase());
-                conn = BaseDbWriter.createConnection(jdbcUrl, "Client_1",dbCredentials.getUserName(), dbCredentials.getPassword(), config);
-                writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
+                ClickHouseConnection conn = BaseDbWriter.createConnection(jdbcUrl, "Client_1",dbCredentials.getUserName(), dbCredentials.getPassword(), config);
+                BaseDbWriter writer = new BaseDbWriter(dbCredentials.getHostName(), dbCredentials.getPort(),
                         dbCredentials.getDatabase(), dbCredentials.getUserName(),
                         dbCredentials.getPassword(), config, conn);
-            }
+            //}
 
             String tableName = props.getProperty(JdbcOffsetBackingStoreConfig.OFFSET_STORAGE_PREFIX +
                     JdbcOffsetBackingStoreConfig.PROP_TABLE_NAME.name());
@@ -592,7 +628,7 @@ public void setup(Properties props, DebeziumRecordParserService debeziumRecordPa
 
         // Start debezium event loop if its requested from REST API.
         if(!config.getBoolean(ClickHouseSinkConnectorConfigVariables.SKIP_REPLICA_START.toString()) || forceStart) {
-            this.setupProcessingThread(config, ddlParserService);
+            this.setupProcessingThread(config);
             setupDebeziumEventCapture(props, debeziumRecordParserService, config);
         } else {
             log.info(ClickHouseSinkConnectorConfigVariables.SKIP_REPLICA_START.toString() + " variable set to true, Replication is skipped, use sink-connector-client to start replication");
@@ -669,7 +705,7 @@ DBCredentials parseDBConfiguration(ClickHouseSinkConnectorConfig config) {
      *
      * @param config
      */
-    private void setupProcessingThread(ClickHouseSinkConnectorConfig config, DDLParserService ddlParserService) {
+    private void setupProcessingThread(ClickHouseSinkConnectorConfig config) {
 
         // Setup separate thread to read messages from shared buffer.
         // this.records = new ConcurrentLinkedQueue<>();
@@ -689,6 +725,7 @@ private void appendToRecords(List<ClickHouseStruct> convertedRecords) {
             this.records.add(convertedRecords);
         }
 
+
     }
 
     /**

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/DDLParserService.java

@@ -3,8 +3,8 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 public interface DDLParserService {
-    String parseSql(String sql, String tableName, StringBuffer parsedQuery);
+    String parseSql(String sql, String tableName,  StringBuffer parsedQuery);
 
-    String parseSql(String sql, String tableName, StringBuffer parsedQuery, AtomicBoolean isDropOrTruncate);
+    String parseSql(String sql, String tableName,StringBuffer parsedQuery, AtomicBoolean isDropOrTruncate);
 
 }

sink-connector-lightweight/src/main/java/com/altinity/clickhouse/debezium/embedded/ddl/parser/MySQLDDLParserService.java

@@ -26,14 +26,17 @@
 @Singleton
 public class MySQLDDLParserService implements DDLParserService {
 
+    private String databaseName;
+
     private ClickHouseSinkConnectorConfig config;
 
     @Inject
     public MySQLDDLParserService() {
 
     }
-    public MySQLDDLParserService(ClickHouseSinkConnectorConfig config) {
+    public MySQLDDLParserService(ClickHouseSinkConnectorConfig config, String databaseName) {
         this.config = config;
+        this.databaseName = databaseName;
     }
 
     @Override
@@ -50,15 +53,15 @@ public String parseSql(String sql, String tableName, StringBuffer parsedQuery) {
         parser.addErrorListener(errorListener);
         lexer.addErrorListener(errorListener);
 
-        MySqlDDLParserListenerImpl listener = new MySqlDDLParserListenerImpl(parsedQuery, tableName, config);
+        MySqlDDLParserListenerImpl listener = new MySqlDDLParserListenerImpl(parsedQuery, tableName, databaseName, config);
         ParseTreeWalker walker = new ParseTreeWalker();
         walker.walk(listener, parser.root());
 
         return clickHouseResult;
     }
 
     @Override
-    public String parseSql(String sql, String tableName, StringBuffer parsedQuery, AtomicBoolean isDropOrTruncate) {
+    public String parseSql(String sql, String tableName,  StringBuffer parsedQuery, AtomicBoolean isDropOrTruncate) {
         String clickHouseResult = null;
 
         MySqlLexer lexer = new MySqlLexer(new CaseChangingCharStream(CharStreams.fromString(sql), true));
@@ -70,7 +73,7 @@ public String parseSql(String sql, String tableName, StringBuffer parsedQuery, A
         parser.addErrorListener(errorListener);
         lexer.addErrorListener(errorListener);
 
-        MySqlDDLParserListenerImpl listener = new MySqlDDLParserListenerImpl(parsedQuery, tableName, this.config);
+        MySqlDDLParserListenerImpl listener = new MySqlDDLParserListenerImpl(parsedQuery, tableName, databaseName, this.config);
         ParseTreeWalker walker = new ParseTreeWalker();
         walker.walk(listener, parser.root());