Sprint 26 master (#45)

* Add Rebound

* Add new TimeStamp Provider

* Create Execute Custom Query action

Author: denyshld

CHANGELOG.md

@@ -1,4 +1,11 @@
 # JDBC Component
+## 2.3.2 (October 21, 2019)
+
+* Add rebound mechanism in case of deadlocks for actions: Insert, UpsertByPK, DeleteByPK
+
+## 2.4.0 (october 17, 2019)
+
+* Add `Custom Query` action
 
 ## 2.3.1 (September 30, 2019)
 

README.md

@@ -4,7 +4,7 @@
 This is an open source component for working with object-relational database management systems on [elastic.io platform](http://www.elastic.io "elastic.io platform").
 
 ### Completeness Matrix
-![JDBC Component Completeness Matrix](https://user-images.githubusercontent.com/16806832/65326373-f6f20680-dbb9-11e9-9ad2-0900f68fd6e3.png)
+![JDBC Component Completeness Matrix](https://user-images.githubusercontent.com/22715422/67289390-38dad900-f4e7-11e9-9a45-1c7775c9c7d5.png)
 
 [JDBC Component Completeness Matrix](https://docs.google.com/spreadsheets/d/1sZr9ydJbMK8v-TguctmFDiqgjRKcrpbdj4CeFuZEkQU/edit?usp=sharing)
 
@@ -117,10 +117,56 @@ The format of ``Start Polling From (optional)`` field should be like ``yyyy-mm-d
 
 *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*
 
+#### Input fields description
+![image](https://user-images.githubusercontent.com/16806832/67293348-f5836900-f4ec-11e9-8e6a-e91b9417ff9d.png)
+
+##### Tables List
+
+Dropdown list with available table names, required field
+
+##### Timestamp (or similar) Column
+
+Dropdown list with available Column names, that have a type like `java.sql.Date` or `java.sql.Timestamp`, required field
+
+##### Start Polling From (optional)
+
+Optional field, indicates the beginning time to start polling from (defaults to the current time)
+
+
 ### SELECT trigger (Deprecated)
 This action exists in JDBC component only for backward compatibility. New [**Select trigger**](#select-trigger) is recommended to use.
 
 ## Actions
+### Execute custom query action
+Action to execute custom SQL query from provided request string.
+
+**Note:** SQL request will be executed according to chosen database JDBC specification.
+
+Execution result returns as array of objects. If request contains multiple sql statements - them will execute inside one transaction.
+If one of statements fails, transaction will be rollbacked.
+
+#### Input fields description
+
+As input metadata, you will get one field named `query` to provide request string
+
+#### Query Samples:
+
+Select:
+```sql
+SELECT name, size FROM stars
+```
+
+Update: 
+```sql
+INSERT INTO stars values (1,'Taurus', '2015-02-19 10:10:10.0', 123, 5, 'true', '2015-02-19')
+```
+
+Posgresql batch multiple statements request:
+```sql
+DELETE FROM stars WHERE id = 1;
+UPDATE stars SET radius = 5 WHERE id = 2;
+```
+
 ### Select action
 ![image](https://user-images.githubusercontent.com/40201204/43592439-39ec5738-967e-11e8-8632-3655b08982d3.png)
 The action will execute an [SQL](https://en.wikipedia.org/wiki/SQL "SQL") query that can return multiple results, it has limitations on the query and suited only for SELECT type of queries.
@@ -175,7 +221,7 @@ As an input metadata you will get a Primary Key field to provide the data inside
 The action will execute ``INSERT`` command into the table from ``Table`` dropdown list the values specified in the body.
 
 #### List of Expected Config fields
-
+   * `Enable Rebound` if `Yes` in case of deadlocks rebound message using Sailor rebound mechanism, number of rebound can be specified via environment variable: `ELASTICIO_REBOUND_LIMIT` recommended value 3
 #### Input fields description
 ##### Table
 
@@ -199,6 +245,7 @@ As output metadata, you will get execution insert result like:
 ![image](https://user-images.githubusercontent.com/40201204/43592505-5b6bbfe8-967e-11e8-845e-2ce8ac707357.png)
 The action will execute delete query from a ``Table`` dropdown field, as criteria can be used only [PRIMARY KEY](https://en.wikipedia.org/wiki/Primary_key "PRIMARY KEY"). The action returns count of affected rows.
 Checkbox ``Don't throw Error on an Empty Result`` allows to emit an empty response, otherwise you will get an error on empty response.
+`Enable Rebound` if `Yes` in case of deadlocks rebound message using Sailor rebound mechanism, number of rebound can be specified via environment variable: `ELASTICIO_REBOUND_LIMIT` recommended value 3
 #### Input fields description
 ![image](https://user-images.githubusercontent.com/40201204/43644579-f593d1c8-9737-11e8-9b97-ee9e575a19f7.png)
 As an input metadata you will get a Primary Key field to provide the data inside as a clause value.
@@ -260,19 +307,23 @@ The action will execute ``SELECT`` command from a ``Tables`` dropdown field, as
 5. Specify input data (field with red asterisk is Primary key), and click "Continue"
 ![image](https://user-images.githubusercontent.com/16806832/44981854-83fcfa00-af7c-11e8-9ef2-8c06e77fed1e.png)
 
-6. Retrieving sample
+6. Enable rebound mechanism if needed
+![image](https://user-images.githubusercontent.com/18464641/67211608-b76e4280-f423-11e9-85f6-f7ec58cc24f1.png)
+
+7. Retrieving sample
 ![image](https://user-images.githubusercontent.com/16806832/44983059-86f9e980-af80-11e8-8178-77e463488c7a.png)
 
-7. Retrieve sample result
+8. Retrieve sample result
 ![image](https://user-images.githubusercontent.com/16806832/44982952-2ec2e780-af80-11e8-98b1-58c3adbc15b9.png)
 
-8. Click "Continue"
+9. Click "Continue"
 ![image](https://user-images.githubusercontent.com/16806832/44983101-b0b31080-af80-11e8-82d8-0e70e4b4ff97.png)
 
-9. Finish component configuration
+10. Finish component configuration
 ![image](https://user-images.githubusercontent.com/16806832/44983365-90378600-af81-11e8-9be4-4dbb39af0fdc.png)
 
 #### Input fields description
+* `Enable Rebound` if `Yes` in case of deadlocks rebound message using Sailor rebound mechanism, number of rebound can be specified via environment variable: `ELASTICIO_REBOUND_LIMIT` recommended value 3
 As an input metadata you will get all fields of selected table. [PRIMARY KEY](https://en.wikipedia.org/wiki/Primary_key "PRIMARY KEY") is required field (will mark as asterisk) and other input fields are optional.
 ![image](https://user-images.githubusercontent.com/16806832/44397461-1a76f780-a549-11e8-8247-9a6f9aa3f3b4.png)
 
@@ -294,6 +345,11 @@ Please use [**Upsert row by primary key**](#upsert-row-by-primary-key-action) in
 4. The current implementation of the action ``Execute stored procedure`` doesn't support ResultSet MSSQL output.
 5. The current implementation of the action ``Execute stored procedure`` doesn't support any array types parameters.
 (MySQL does not have schemas by definition)
+6. Rebound mechanism only works for this SQL State: 
+ - ``MySQL``: 40001, XA102
+ - ``Oracle``: 61000
+ - ``MSSQL``: 40001
+ - ``PostgreSQL``:  40P01
 
 ## License
 Apache-2.0 © [elastic.io GmbH](https://www.elastic.io "elastic.io GmbH")

build.gradle

@@ -1,5 +1,5 @@
 group = 'io.elastic'
-version = '2.3.1'
+version = '2.4.0'
 apply plugin: 'java'
 apply plugin: 'idea'
 apply plugin: 'eclipse'

component.json

@@ -1,6 +1,7 @@
 {
   "title": "Database",
   "description": "Database JDBC connector",
+  "buildType": "docker",
   "credentials": {
     "verifier": "io.elastic.jdbc.JdbcCredentialsVerifier",
     "fields": {
@@ -72,7 +73,7 @@
           "viewClass": "SelectView",
           "prompt": "Select a Timestamp (or similar) Column",
           "label": "Timestamp (or similar) Column",
-          "model": "io.elastic.jdbc.providers.ColumnNamesProvider",
+          "model": "io.elastic.jdbc.providers.TimeStampColumnNamesProvider",
           "required": true,
           "require": [
             "tableName"
@@ -138,6 +139,37 @@
     }
   },
   "actions": {
+    "customQuery": {
+      "main": "io.elastic.jdbc.actions.CustomQuery",
+      "title": "Execute custom query",
+      "description": "Executes provided sql query string as is",
+      "metadata": {
+        "in": {
+          "type": "object",
+          "required": true,
+          "properties": {
+            "query": {
+              "type": "string",
+              "required": true
+            }
+          }
+        },
+        "out":{
+          "type": "object",
+          "required": false,
+          "properties": {
+            "result": {
+              "required": false,
+              "type": "array"
+            },
+            "updated": {
+              "required": false,
+              "type": "number"
+            }
+          }
+        }
+      }
+    },
     "lookupRowByPrimaryKey": {
       "main": "io.elastic.jdbc.actions.LookupRowByPrimaryKey",
       "title": "Lookup Row By Primary Key",
@@ -158,7 +190,7 @@
       "dynamicMetadata": "io.elastic.jdbc.providers.PrimaryColumnNamesProvider"
     },
     "upsertRowByPrimaryKey": {
-      "main": "io.elastic.jdbc.actions.providers.UpsertRowByPrimaryKey",
+      "main": "io.elastic.jdbc.actions.UpsertRowByPrimaryKey",
       "title": "Upsert Row By Primary Key",
       "description": "Executes upsert by primary key",
       "fields": {
@@ -168,6 +200,16 @@
           "label": "Table",
           "required": true,
           "model": "io.elastic.jdbc.providers.TableNameProvider"
+        },
+        "reboundEnabled" : {
+          "viewClass": "SelectView",
+          "prompt": "Default is No",
+          "label": "Enable Rebound",
+          "required": false,
+          "model" : {
+            "Yes" : "Yes",
+            "No" : "No"
+          }
         }
       },
       "dynamicMetadata": "io.elastic.jdbc.providers.ColumnNamesWithPrimaryKeyProvider"
@@ -183,6 +225,16 @@
           "label": "Table",
           "required": true,
           "model": "io.elastic.jdbc.providers.TableNameProvider"
+        },
+        "reboundEnabled" : {
+          "viewClass": "SelectView",
+          "prompt": "Default is No",
+          "label": "Enable Rebound",
+          "required": false,
+          "model" : {
+            "Yes" : "Yes",
+            "No" : "No"
+          }
         }
       },
       "dynamicMetadata": "io.elastic.jdbc.providers.ColumnNamesForInsertProvider"
@@ -202,6 +254,16 @@
         "nullableResult": {
           "label": "Don`t throw Error on an Empty Result",
           "viewClass": "CheckBoxView"
+        },
+        "reboundEnabled" : {
+          "viewClass": "SelectView",
+          "prompt": "Default is No",
+          "label": "Enable Rebound",
+          "required": false,
+          "model" : {
+            "Yes" : "Yes",
+            "No" : "No"
+          }
         }
       },
       "dynamicMetadata": "io.elastic.jdbc.providers.PrimaryColumnNamesProvider"

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

@@ -0,0 +1,112 @@
+package io.elastic.jdbc.actions;
+
+import io.elastic.api.ExecutionParameters;
+import io.elastic.api.Message;
+import io.elastic.api.Module;
+import io.elastic.jdbc.utils.Utils;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import javax.json.Json;
+import javax.json.JsonArray;
+import javax.json.JsonArrayBuilder;
+import javax.json.JsonObject;
+import javax.json.JsonObjectBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CustomQuery implements Module {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(CustomQuery.class);
+
+  private static final String JSON_RESULT_ARRAY_NAME = "result";
+  private static final String JSON_RESULT_COUNT_NAME = "updated";
+
+  @Override
+  public void execute(ExecutionParameters parameters) {
+    LOGGER.info("Starting execute custom query action");
+    final JsonObject configuration = parameters.getConfiguration();
+    final JsonObject body = parameters.getMessage().getBody();
+    final String dbEngine = Utils.getDbEngine(configuration);
+    final String queryString = body.getString("query");
+    LOGGER.info("Found dbEngine: '{}' and query: '{}'", dbEngine, queryString);
+
+    List<Message> messages = new ArrayList<>();
+    try (Connection connection = Utils.getConnection(configuration)) {
+      connection.setAutoCommit(false);
+      try (Statement statement = connection.createStatement()) {
+        boolean status = statement.execute(queryString);
+        if (status) {
+          ResultSet resultSet = statement.getResultSet();
+          messages.add(this.processResultSetToMessage(resultSet));
+        } else {
+          messages.add(this.processUpdateCountToMessage(statement.getUpdateCount()));
+        }
+
+        while (statement.getMoreResults() || statement.getUpdateCount() != -1) {
+          if (statement.getUpdateCount() != -1) {
+            messages.add(this.processUpdateCountToMessage(statement.getUpdateCount()));
+          } else {
+            messages.add(this.processResultSetToMessage(statement.getResultSet()));
+          }
+        }
+
+        connection.commit();
+      } catch (Exception e) {
+        connection.rollback();
+        connection.setAutoCommit(true);
+        throw e;
+      }
+    } catch (SQLException e) {
+      throw new RuntimeException(e);
+    }
+
+    messages.forEach(message -> {
+      LOGGER.trace("Emit data= {}", message.getBody());
+      parameters.getEventEmitter().emitData(message);
+    });
+
+    LOGGER.info("Custom query action is successfully executed");
+  }
+
+  private Message processResultSetToMessage(ResultSet resultSet) throws SQLException {
+    JsonArray result = customResultSetToJsonArray(resultSet);
+    return new Message.Builder()
+        .body(Json.createObjectBuilder()
+            .add(JSON_RESULT_ARRAY_NAME, result)
+            .build()
+        ).build();
+  }
+
+  private Message processUpdateCountToMessage(int updateCount) {
+    return new Message.Builder()
+        .body(Json.createObjectBuilder()
+            .add(JSON_RESULT_COUNT_NAME, updateCount)
+            .build()
+        ).build();
+  }
+
+  public static JsonArray customResultSetToJsonArray(ResultSet resultSet) throws SQLException {
+    JsonArrayBuilder jsonBuilder = Json.createArrayBuilder();
+
+    if (resultSet == null) {
+      return jsonBuilder.build();
+    }
+
+    ResultSetMetaData metaData = resultSet.getMetaData();
+
+    while (resultSet.next()) {
+      JsonObjectBuilder entry = Json.createObjectBuilder();
+      for (int i = 1; i <= metaData.getColumnCount(); i++) {
+        Utils.getColumnDataByType(resultSet, metaData, i, entry);
+      }
+      jsonBuilder.add(entry.build());
+    }
+
+    return jsonBuilder.build();
+  }
+}