Merge pull request #12 from elasticio/issue-195-unknown-table-error

define catalog for mysql (getPrimaryKeys method)

Author: Olha Virolainen

README.md

@@ -108,6 +108,8 @@ The format of ``Start Polling From (optional)`` field should be like ``yyyy-mm-d
 - ``ss`` - second
 - ``sss`` - millisecond (optional)
 
+*Please Note: Component Snapshot will not be overwritten in Real-Time flows due to platform behaviour, so we strongly recommend to use Get Rows Polling trigger in Keen Flows only*
+
 ### SELECT trigger (Deprecated)
 This action exists in JDBC component only for backward compatibility. New [**Select trigger**](#select-trigger) is recommended to use.
 
@@ -216,9 +218,6 @@ This action exists in JDBC component only for backward compatibility. [**Upsert
 - ``MSSQL`` - compatible with Microsoft SQL Server 2008 R2 and higher
 3. The current implementation of the action ``Upsert By Primary Key`` doesn't mark non-nullable fields as required fields at a dynamic metadata. In case of updating such fields with an empty value you will get SQL Exception ``Cannot insert the value NULL into...``. You should manually fill in all non-nullable fields with previous data, if you want to update part of columns in a row, even if data in that fields doesn't change. 
 
-## Known issues
-No known issues are there yet.
-
 ## License
 Apache-2.0 © [elastic.io GmbH](https://www.elastic.io "elastic.io GmbH")
 

component.json

@@ -1,5 +1,5 @@
 {
-  "title": "Database MySQL",
+  "title": "Database",
   "description": "Database JDBC connector",
   "credentials": {
     "verifier": "io.elastic.jdbc.JdbcCredentialsVerifier",
@@ -105,8 +105,9 @@
     },
     "select": {
       "main": "io.elastic.jdbc.triggers.SelectTriggerOld",
-      "title": "SELECT (Deprecated)",
-      "description": "This trigger is already deprecated. Please use 'Select' trigger",
+      "title": "SELECT",
+      "deprecated": true,
+      "description": "DEPRECATED: Please use 'Select' Trigger",
       "type": "polling",
       "fields": {
         "tableName": {
@@ -206,7 +207,9 @@
     },
     "createOrUpdateRecord": {
       "main": "io.elastic.jdbc.actions.CreateOrUpdateRecord",
-      "title": "Create or update record (Deprecated)",
+      "title": "Create or update record",
+      "deprecated": true,
+      "description": "DEPRECATED: Please use 'Upsert Row By Primary Key' Action",
       "fields": {
         "tableName": {
           "viewClass": "SelectView",

src/main/java/io/elastic/jdbc/ColumnNamesProvider.java

@@ -55,14 +55,18 @@ public JsonObject getColumns(JsonObject configuration) {
     JsonObjectBuilder properties = Json.createObjectBuilder();
     Connection connection = null;
     ResultSet rs = null;
+    String catalog = null;
     String schemaName = null;
     boolean isEmpty = true;
-    Boolean isOracle = configuration.getString("dbEngine").equals("oracle");
     Boolean isMssql = configuration.getString("dbEngine").equals("mssql");
+    Boolean isMysql = configuration.getString("dbEngine").equals("mysql");
     try {
       connection = Utils.getConnection(configuration);
       DatabaseMetaData dbMetaData = connection.getMetaData();
-      rs = dbMetaData.getPrimaryKeys(null, null, tableName);
+      if (isMysql) {
+        catalog = configuration.getString("databaseName");
+      }
+      rs = dbMetaData.getPrimaryKeys(catalog, null, tableName);
       if (tableName.contains(".")) {
         schemaName = tableName.split("\\.")[0];
         tableName = tableName.split("\\.")[1];

src/main/java/io/elastic/jdbc/ColumnNamesWithPrimaryKeyProvider.java

@@ -57,18 +57,23 @@ public JsonObject getColumns(JsonObject configuration) {
     Connection connection = null;
     ResultSet rs = null;
     ResultSet rsPrimaryKeys = null;
+    String catalog = null;
     String schemaName = null;
     boolean isEmpty = true;
     Boolean isOracle = configuration.getString("dbEngine").equals("oracle");
+    Boolean isMysql = configuration.getString("dbEngine").equals("mysql");
     try {
       connection = Utils.getConnection(configuration);
       DatabaseMetaData dbMetaData = connection.getMetaData();
       if (tableName.contains(".")) {
         schemaName = tableName.split("\\.")[0];
         tableName = tableName.split("\\.")[1];
       }
+      if (isMysql) {
+        catalog = configuration.getString("databaseName");
+      }
       rsPrimaryKeys = dbMetaData
-          .getPrimaryKeys(null, ((isOracle && !schemaName.isEmpty()) ? schemaName : null),
+          .getPrimaryKeys(catalog, ((isOracle && !schemaName.isEmpty()) ? schemaName : null),
               tableName);
       rs = dbMetaData.getColumns(null, schemaName, tableName, "%");
       while (rs.next()) {

src/main/java/io/elastic/jdbc/PrimaryColumnNamesProvider.java

@@ -55,22 +55,27 @@ public JsonObject getPrimaryColumns(JsonObject configuration) {
     JsonObject properties = Json.createObjectBuilder().build();
     Connection connection = null;
     ResultSet rs = null;
+    String catalog = null;
     String schemaName = null;
     Boolean isEmpty = true;
     Boolean isOracle = configuration.getString("dbEngine").equals("oracle");
     Boolean isMssql = configuration.getString("dbEngine").equals("mssql");
+    Boolean isMysql = configuration.getString("dbEngine").equals("mysql");
     List<String> primaryKeys = new ArrayList();
     try {
       connection = Utils.getConnection(configuration);
       DatabaseMetaData dbMetaData = connection.getMetaData();
+      if (isMysql) {
+        catalog = configuration.getString("databaseName");
+      }
       if (tableName.contains(".")) {
         schemaName =
             (isOracle) ? tableName.split("\\.")[0].toUpperCase() : tableName.split("\\.")[0];
         tableName =
             (isOracle) ? tableName.split("\\.")[1].toUpperCase() : tableName.split("\\.")[1];
       }
       rs = dbMetaData
-          .getPrimaryKeys(null, ((isOracle && !schemaName.isEmpty()) ? schemaName : null),
+          .getPrimaryKeys(catalog, ((isOracle && !schemaName.isEmpty()) ? schemaName : null),
               tableName);
       while (rs.next()) {
         primaryKeys.add(rs.getString("COLUMN_NAME"));

src/main/java/io/elastic/jdbc/actions/UpsertRowByPrimaryKey.java

@@ -30,6 +30,7 @@ public void execute(ExecutionParameters parameters) {
     JsonObject resultRow;
     String tableName;
     String dbEngine;
+    String catalog = null;
     String schemaName = "";
     String primaryKey = "";
     int primaryKeysCount = 0;
@@ -54,17 +55,21 @@ public void execute(ExecutionParameters parameters) {
 
     LOGGER.info("Executing lookup primary key");
     boolean isOracle = dbEngine.equals(Engines.ORACLE.name().toLowerCase());
+    Boolean isMysql = configuration.getString("dbEngine").equals("mysql");
 
     try (Connection connection = Utils.getConnection(configuration)) {
       DatabaseMetaData dbMetaData = connection.getMetaData();
+      if (isMysql) {
+        catalog = configuration.getString("databaseName");
+      }
       if (tableName.contains(".")) {
         schemaName =
             (isOracle) ? tableName.split("\\.")[0].toUpperCase() : tableName.split("\\.")[0];
         tableName =
             (isOracle) ? tableName.split("\\.")[1].toUpperCase() : tableName.split("\\.")[1];
       }
       try (ResultSet rs = dbMetaData
-          .getPrimaryKeys(null, ((isOracle && !schemaName.isEmpty()) ? schemaName : null),
+          .getPrimaryKeys(catalog, ((isOracle && !schemaName.isEmpty()) ? schemaName : null),
               tableName)) {
         while (rs.next()) {
           primaryKey = rs.getString("COLUMN_NAME");