Update Readme.md and Use one transaction for UpsertAction

Author: shulkaolka

README.md

@@ -22,7 +22,7 @@ Following actions are inside:
 ### Requirements
 Before you can deploy any code into elastic.io **you must be a registered elastic.io platform user**. Please see our home page at [http://www.elastic.io](http://www.elastic.io) to learn how. 
 #### Environment variables
-For unit-testing is needed specify following environment variables:
+For unit-testing is needed to specify following environment variables:
 1. Connection to MSSQL:
  - ``CONN_USER_MSSQL`` - user login
  - ``CONN_PASSWORD_MSSQL`` - user password
@@ -192,15 +192,15 @@ Checkbox ``Don't throw Error on an Empty Result`` allows to emit an empty respon
 As an input metadata you will get a Primary Key field to provide the data inside as a clause value.
 
 ## Current limitations
-1. Only tables with one [PRIMARY KEY](https://en.wikipedia.org/wiki/Primary_key "PRIMARY KEY") is supported. Message ``Table has not Primary Key. Should be one Primary Key
-`` should be displayed if the table without any primary key will be selected and message ``Composite Primary Key is not supported
-`` should be displayed if the table with composite primary key will be selected.
-2. Only following versions of DataBase are supported:
+1. Only tables with one [PRIMARY KEY](https://en.wikipedia.org/wiki/Primary_key "PRIMARY KEY") is supported. You will see the message ``Table has not Primary Key. Should be one Primary Key
+``, if the selected table doesn't have a primary key. Also, you will see the message ``Composite Primary Key is not supported
+``, if the selected table has composite primary key.
+2. Only following versions of database types are supported:
 - ``MySQL`` - compatible with MySQL Server 5.5, 5.6, 5.7 and 8.0.
 - ``PostgreSQL`` - compatible with PostgreSQL 8.2 and higher
 - ``Oracle`` - compatible with Oracle Database 8.1.7 - 12.1.0.2
 - ``MSSQL`` - compatible with Microsoft SQL Server 2008 R2 and higher
-3. For Action "Upsert by primary key" metadata analysis ignores nullable/non-nullable fields, it will be throw SQL Server exception ``Cannot insert the value NULL into...`` if non-nullable fields don't fill for Insert operation. 
+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.

src/main/java/io/elastic/jdbc/QueryBuilders/MSSQL.java

@@ -148,51 +148,4 @@ public void executeUpdate(Connection connection, String tableName, String idColu
     }
   }
 
-  public void executeUpsert(Connection connection, String idColumn, JsonObject body)
-      throws SQLException {
-    validateQuery();
-
-    StringBuilder keys = new StringBuilder();
-    StringBuilder values = new StringBuilder();
-    StringBuilder setString = new StringBuilder();
-    for (Map.Entry<String, JsonValue> entry : body.entrySet()) {
-      if (setString.length() > 0) {
-        setString.append(",");
-      }
-      setString.append(entry.getKey()).append(" = ?");
-      if (keys.length() > 0) {
-        keys.append(",");
-      }
-      keys.append(entry.getKey());
-      if (values.length() > 0) {
-        values.append(",");
-      }
-      values.append("?");
-    }
-    String sql = "BEGIN TRANSACTION;" +
-        " IF EXISTS (SELECT * FROM " + tableName +
-        " WHERE " + idColumn + "= ?)" +
-        " UPDATE " + tableName +
-        " SET " + setString.toString() +
-        " WHERE " + idColumn + " = ?" +
-        " ELSE INSERT INTO " + tableName +
-        " (" + keys.toString() + ")" +
-        " VALUES (" + values.toString() + ")" +
-        " COMMIT;";
-    try (PreparedStatement stmt = connection.prepareStatement(sql)) {
-      Utils.setStatementParam(stmt, 1, idColumn, body);
-      int i = 2;
-      for (Map.Entry<String, JsonValue> entry : body.entrySet()) {
-        Utils.setStatementParam(stmt, i, entry.getKey(), body);
-        i++;
-      }
-      Utils.setStatementParam(stmt, i++, idColumn, body);
-      for (Map.Entry<String, JsonValue> entry : body.entrySet()) {
-        Utils.setStatementParam(stmt, i, entry.getKey(), body);
-        i++;
-      }
-      stmt.execute();
-    }
-  }
-
 }

src/main/java/io/elastic/jdbc/QueryBuilders/MySQL.java

@@ -139,40 +139,4 @@ public void executeUpdate(Connection connection, String tableName, String idColu
     }
   }
 
-  public void executeUpsert(Connection connection, String idColumn, JsonObject body)
-      throws SQLException {
-    validateQuery();
-    StringBuilder keys = new StringBuilder();
-    StringBuilder values = new StringBuilder();
-    StringBuilder setString = new StringBuilder();
-    for (Map.Entry<String, JsonValue> entry : body.entrySet()) {
-      if (setString.length() > 0) {
-        setString.append(",");
-      }
-      setString.append(entry.getKey()).append(" = ?");
-      if (keys.length() > 0) {
-        keys.append(",");
-      }
-      keys.append(entry.getKey());
-      if (values.length() > 0) {
-        values.append(",");
-      }
-      values.append("?");
-    }
-    String sql = "INSERT INTO " + tableName +
-        " (" + keys.toString() + ")" +
-        " VALUES (" + values.toString() + ")" +
-        " ON DUPLICATE KEY UPDATE " + setString + ";";
-    try (PreparedStatement stmt = connection.prepareStatement(sql)) {
-      int i = 1;
-      int countBodyEntry = body.size();
-      for (Map.Entry<String, JsonValue> entry : body.entrySet()) {
-        Utils.setStatementParam(stmt, i, entry.getKey(), body);
-        Utils.setStatementParam(stmt, i + countBodyEntry, entry.getKey(), body);
-        i++;
-      }
-      stmt.execute();
-    }
-  }
-
 }

src/main/java/io/elastic/jdbc/QueryBuilders/Oracle.java

@@ -133,47 +133,4 @@ public void executeUpdate(Connection connection, String tableName, String idColu
     }
   }
 
-  public void executeUpsert(Connection connection, String idColumn, JsonObject body)
-      throws SQLException {
-    validateQuery();
-    StringBuilder keys = new StringBuilder();
-    StringBuilder values = new StringBuilder();
-    StringBuilder setString = new StringBuilder();
-    for (Map.Entry<String, JsonValue> entry : body.entrySet()) {
-      if (setString.length() > 0) {
-        setString.append(",");
-      }
-      setString.append(entry.getKey()).append(" = ?");
-      if (keys.length() > 0) {
-        keys.append(",");
-      }
-      keys.append(entry.getKey());
-      if (values.length() > 0) {
-        values.append(",");
-      }
-      values.append("?");
-    }
-    String sql = "BEGIN " +
-        " INSERT INTO " + tableName +
-        " (" + keys.toString() + ")" +
-        " VALUES (" + values.toString() + ");" +
-        " EXCEPTION" +
-        " WHEN DUP_VAL_ON_INDEX THEN" +
-        " UPDATE " + tableName +
-        " SET " + setString.toString() +
-        " WHERE " + idColumn + " = ?;" +
-        " END;";
-    try (PreparedStatement stmt = connection.prepareStatement(sql)) {
-      //set Statement parameters for Insert (i) and Update operation (i + countBodyEntry)
-      int i = 1;
-      int countBodyEntry = body.size();
-      for (Map.Entry<String, JsonValue> entry : body.entrySet()) {
-        Utils.setStatementParam(stmt, i, entry.getKey(), body);
-        Utils.setStatementParam(stmt, i + countBodyEntry, entry.getKey(), body);
-        i++;
-      }
-      Utils.setStatementParam(stmt, i + countBodyEntry, idColumn, body);
-      stmt.execute();
-    }
-  }
 }

src/main/java/io/elastic/jdbc/QueryBuilders/PostgreSQL.java

@@ -149,42 +149,4 @@ public void executeUpdate(Connection connection, String tableName, String idColu
     }
   }
 
-  public void executeUpsert(Connection connection, String idColumn, JsonObject body)
-      throws SQLException {
-    validateQuery();
-    StringBuilder keys = new StringBuilder();
-    StringBuilder values = new StringBuilder();
-    StringBuilder setString = new StringBuilder();
-    for (Map.Entry<String, JsonValue> entry : body.entrySet()) {
-      if (setString.length() > 0) {
-        setString.append(",");
-      }
-      setString.append(entry.getKey()).append(" = ?");
-      if (keys.length() > 0) {
-        keys.append(",");
-      }
-      keys.append(entry.getKey());
-      if (values.length() > 0) {
-        values.append(",");
-      }
-      values.append("?");
-    }
-    String sql = "INSERT INTO " + tableName +
-        " (" + keys.toString() + ")" +
-        " VALUES (" + values.toString() + ")" +
-        " ON CONFLICT (" + idColumn + ")" +
-        " DO UPDATE " +
-        " SET " + setString.toString() + ";";
-    try (PreparedStatement stmt = connection.prepareStatement(sql)) {
-      //set Statement parameters for Insert (i) and Update operation (i + countBodyEntry)
-      int i = 1;
-      int countBodyEntry = body.size();
-      for (Map.Entry<String, JsonValue> entry : body.entrySet()) {
-        Utils.setStatementParam(stmt, i, entry.getKey(), body);
-        Utils.setStatementParam(stmt, i + countBodyEntry, entry.getKey(), body);
-        i++;
-      }
-      stmt.execute();
-    }
-  }
 }

src/main/java/io/elastic/jdbc/QueryBuilders/Query.java

@@ -8,6 +8,7 @@
 import java.sql.SQLException;
 import java.sql.Timestamp;
 import java.util.ArrayList;
+import java.util.Map;
 import java.util.Map.Entry;
 import javax.json.Json;
 import javax.json.JsonObject;
@@ -95,9 +96,6 @@ abstract public void executeInsert(Connection connection, String tableName, Json
   abstract public void executeUpdate(Connection connection, String tableName, String idColumn,
       String idValue, JsonObject body) throws SQLException;
 
-  abstract public void executeUpsert(Connection connection, String idColumn,
-      JsonObject body) throws SQLException;
-
   public ArrayList executeSelectTrigger(Connection connection, String sqlQuery)
       throws SQLException {
     try (PreparedStatement stmt = connection.prepareStatement(sqlQuery)) {
@@ -142,6 +140,119 @@ public ArrayList executeSelectQuery(Connection connection, String sqlQuery, Json
     }
   }
 
+  public JsonObject executeUpsert(Connection connection, String idColumn,
+      JsonObject body) throws SQLException {
+    validateQuery();
+    JsonObject foundRow;
+    JsonObjectBuilder row = Json.createObjectBuilder();
+    int rowsCount = 0;
+    int i;
+    ResultSet rs;
+    ResultSetMetaData metaData;
+
+    StringBuilder keys = new StringBuilder();
+    StringBuilder values = new StringBuilder();
+    StringBuilder setString = new StringBuilder();
+    for (Map.Entry<String, JsonValue> entry : body.entrySet()) {
+      if (keys.length() > 0) {
+        keys.append(",");
+      }
+      keys.append(entry.getKey());
+      if (values.length() > 0) {
+        values.append(",");
+      }
+      values.append("?");
+      if (!entry.getKey().equals(idColumn)) {
+        if (setString.length() > 0) {
+          setString.append(",");
+        }
+        setString.append(entry.getKey()).append(" = ?");
+      }
+    }
+
+    String sqlSELECT =
+        "    SELECT" +
+            "        *" +
+            "    FROM " + tableName +
+            "    WHERE " + idColumn + " = ?";
+    String sqlInsert = "INSERT INTO " + tableName +
+        " (" + keys.toString() + ")" +
+        " VALUES (" + values.toString() + ")";
+    String sqlUpdate = "UPDATE " + tableName +
+        " SET " + setString.toString() +
+        " WHERE " + idColumn + " = ?";
+
+    PreparedStatement stmtSelect = null;
+    PreparedStatement stmtInsert = null;
+    PreparedStatement stmtUpdate = null;
+
+    try {
+      connection.setAutoCommit(false);
+
+      stmtSelect = connection.prepareStatement(sqlSELECT);
+      Utils.setStatementParam(stmtSelect, 1, idColumn, body);
+      rs = stmtSelect.executeQuery();
+      metaData = rs.getMetaData();
+      while (rs.next()) {
+        for (i = 1; i <= metaData.getColumnCount(); i++) {
+          row = Utils.getColumnDataByType(rs, metaData, i, row);
+        }
+        rowsCount++;
+        if (rowsCount > 1) {
+          throw new RuntimeException("Error: the number of matching rows is not exactly one");
+        }
+      }
+      foundRow = row.build();
+
+      i = 1;
+      if (foundRow.size() == 0) {
+        stmtInsert = connection.prepareStatement(sqlInsert);
+        for (Map.Entry<String, JsonValue> entry : body.entrySet()) {
+          Utils.setStatementParam(stmtInsert, i, entry.getKey(), body);
+          i++;
+        }
+        stmtInsert.execute();
+      } else {
+        stmtUpdate = connection.prepareStatement(sqlUpdate);
+        for (Map.Entry<String, JsonValue> entry : body.entrySet()) {
+          if (!entry.getKey().equals(idColumn)) {
+            Utils.setStatementParam(stmtUpdate, i, entry.getKey(), body);
+            i++;
+          }
+        }
+        Utils.setStatementParam(stmtUpdate, i, idColumn, body);
+        stmtUpdate.execute();
+      }
+
+      rs = stmtSelect.executeQuery();
+      metaData = rs.getMetaData();
+      rowsCount = 0;
+      while (rs.next()) {
+        for (i = 1; i <= metaData.getColumnCount(); i++) {
+          row = Utils.getColumnDataByType(rs, metaData, i, row);
+        }
+        rowsCount++;
+        if (rowsCount > 1) {
+          throw new RuntimeException("Error: the number of matching rows is not exactly one");
+        }
+      }
+      connection.commit();
+
+    } finally {
+      if (stmtSelect != null) {
+        stmtSelect.close();
+      }
+      if (stmtInsert != null) {
+        stmtInsert.close();
+      }
+      if (stmtUpdate != null) {
+        stmtUpdate.close();
+      }
+      connection.setAutoCommit(true);
+    }
+    return row.build();
+  }
+
   public void validateQuery() {
     if (tableName == null) {
       throw new RuntimeException("Table name is required field");