Skip to content

Polling trigger issue #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 9 commits into from
Sep 18, 2018
Merged

Polling trigger issue #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
c63c40d
Trigger Polling fix (does not return updated rows)
shulkaolka Sep 14, 2018
46210df
Ignore test
shulkaolka Sep 17, 2018
88480fb
removed >=
shulkaolka Sep 17, 2018
6633759
we should emit snapshot only if we have some result
shulkaolka Sep 17, 2018
29f25f7
Mssql, delete skipnamber from query
shulkaolka Sep 17, 2018
90c3b88
improved MySql & PostgreSQL
ShkarupaNick Sep 17, 2018
5e5ec67
removed skipNumber from Polling trigger
ShkarupaNick Sep 17, 2018
0b8b838
Merge branch 'PollingTriggerIssue' of https://github.com/elasticio/jd…
shulkaolka Sep 17, 2018
4444712
Refactor executePolling method
shulkaolka Sep 17, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 3 additions & 43 deletions src/main/java/io/elastic/jdbc/QueryBuilders/MSSQL.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,14 +3,10 @@
import io.elastic.jdbc.Utils;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Map;
import javax.json.Json;
import javax.json.JsonObject;
import javax.json.JsonObjectBuilder;
import javax.json.JsonValue;

public class MSSQL extends Query {
Expand All @@ -28,36 +24,8 @@ public ArrayList executePolling(Connection connection) throws SQLException {
" )" +
" SELECT *" +
" FROM Results_CTE" +
" WHERE RowNum > ?" +
" AND RowNum < ?";
try (PreparedStatement stmt = connection.prepareStatement(sql)) {
stmt.setTimestamp(1, pollingValue);
stmt.setInt(2, skipNumber);
stmt.setInt(3, countNumber + skipNumber);
try (ResultSet rs = stmt.executeQuery()) {
ArrayList listResult = new ArrayList();
JsonObjectBuilder row = Json.createObjectBuilder();
ResultSetMetaData metaData = rs.getMetaData();
while (rs.next()) {
for (int i = 1; i <= metaData.getColumnCount(); i++) {
row = Utils.getColumnDataByType(rs, metaData, i, row);
if (metaData.getColumnName(i).toUpperCase().equals(pollingField.toUpperCase())) {
if (maxPollingValue.before(rs.getTimestamp(i))) {
if (rs.getString(metaData.getColumnName(i)).length() > 10) {
maxPollingValue = java.sql.Timestamp
.valueOf(rs.getString(metaData.getColumnName(i)));
} else {
maxPollingValue = java.sql.Timestamp
.valueOf(rs.getString(metaData.getColumnName(i)) + " 00:00:00");
}
}
}
}
listResult.add(row.build());
}
return listResult;
}
}
" WHERE RowNum <= ?";
return getRowsExecutePolling(connection, sql);
}

public JsonObject executeLookup(Connection connection, JsonObject body) throws SQLException {
Expand All @@ -74,7 +42,7 @@ public JsonObject executeLookup(Connection connection, JsonObject body) throws S
" FROM Results_CTE" +
" WHERE RowNum > ?" +
" AND RowNum < ?";
return Utils.getLookupRow(connection, body, sql, skipNumber, countNumber + skipNumber);
return getLookupRow(connection, body, sql, skipNumber, countNumber + skipNumber);
}

public int executeDelete(Connection connection, JsonObject body) throws SQLException {
Expand All @@ -88,14 +56,6 @@ public int executeDelete(Connection connection, JsonObject body) throws SQLExcep
}
}

public boolean executeRecordExists(Connection connection, JsonObject body) throws SQLException {
validateQuery();
String sql = "SELECT COUNT(*)" +
" FROM " + tableName +
" WHERE " + lookupField + " = ?";
return Utils.isRecordExists(connection, body, sql, lookupField);
}

public void executeInsert(Connection connection, String tableName, JsonObject body)
throws SQLException {
validateQuery();
Expand Down
47 changes: 4 additions & 43 deletions src/main/java/io/elastic/jdbc/QueryBuilders/MySQL.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,14 +3,10 @@
import io.elastic.jdbc.Utils;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Map;
import javax.json.Json;
import javax.json.JsonObject;
import javax.json.JsonObjectBuilder;
import javax.json.JsonValue;

public class MySQL extends Query {
Expand All @@ -23,38 +19,11 @@ public ArrayList executePolling(Connection connection) throws SQLException {
sql.append(pollingField);
sql.append(" > ?");
if (orderField != null) {
sql.append(" ORDER BY ").append(orderField);
sql.append(" ORDER BY ").append(orderField).append(" ASC");
}
sql.append(" ASC LIMIT ? OFFSET ?");
sql.append(" LIMIT ?");

try (PreparedStatement stmt = connection.prepareStatement(sql.toString())) {
stmt.setTimestamp(1, pollingValue);
stmt.setInt(2, countNumber);
stmt.setInt(3, skipNumber);
try (ResultSet rs = stmt.executeQuery()) {
ArrayList listResult = new ArrayList();
JsonObjectBuilder row = Json.createObjectBuilder();
ResultSetMetaData metaData = rs.getMetaData();
while (rs.next()) {
for (int i = 1; i <= metaData.getColumnCount(); i++) {
row = Utils.getColumnDataByType(rs, metaData, i, row);
if (metaData.getColumnName(i).toUpperCase().equals(pollingField.toUpperCase())) {
if (maxPollingValue.before(rs.getTimestamp(i))) {
if (rs.getString(metaData.getColumnName(i)).length() > 10) {
maxPollingValue = java.sql.Timestamp
.valueOf(rs.getString(metaData.getColumnName(i)));
} else {
maxPollingValue = java.sql.Timestamp
.valueOf(rs.getString(metaData.getColumnName(i)) + " 00:00:00");
}
}
}
}
listResult.add(row.build());
}
return listResult;
}
}
return getRowsExecutePolling(connection, sql.toString());
}

public JsonObject executeLookup(Connection connection, JsonObject body) throws SQLException {
Expand All @@ -66,7 +35,7 @@ public JsonObject executeLookup(Connection connection, JsonObject body) throws S
sql.append(" = ?");
sql.append(" ORDER BY ").append(lookupField);
sql.append(" ASC LIMIT ? OFFSET ?");
return Utils.getLookupRow(connection, body, sql.toString(), countNumber, skipNumber);
return getLookupRow(connection, body, sql.toString(), countNumber, skipNumber);
}

public int executeDelete(Connection connection, JsonObject body) throws SQLException {
Expand All @@ -79,14 +48,6 @@ public int executeDelete(Connection connection, JsonObject body) throws SQLExcep
}
}

public boolean executeRecordExists(Connection connection, JsonObject body) throws SQLException {
validateQuery();
String sql = "SELECT COUNT(*)" +
" FROM " + tableName +
" WHERE " + lookupField + " = ?";
return Utils.isRecordExists(connection, body, sql, lookupField);
}

public void executeInsert(Connection connection, String tableName, JsonObject body)
throws SQLException {
validateQuery();
Expand Down
56 changes: 9 additions & 47 deletions src/main/java/io/elastic/jdbc/QueryBuilders/Oracle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,54 +3,24 @@
import io.elastic.jdbc.Utils;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Map;
import javax.json.Json;
import javax.json.JsonObject;
import javax.json.JsonObjectBuilder;
import javax.json.JsonValue;

public class Oracle extends Query {

public ArrayList executePolling(Connection connection) throws SQLException {
validateQuery();
String sql = "SELECT * FROM " +
" (SELECT b.*, rank() over (order by " + pollingField + ") as rnk FROM " +
tableName + " b) WHERE " + pollingField + " > ?" +
" AND rnk BETWEEN ? AND ?" +
" ORDER BY " + pollingField;
try (PreparedStatement stmt = connection.prepareStatement(sql)) {
/* data types mapping https://docs.oracle.com/cd/B19306_01/java.102/b14188/datamap.htm */
stmt.setTimestamp(1, pollingValue);
stmt.setInt(2, skipNumber);
stmt.setInt(3, countNumber);
try (ResultSet rs = stmt.executeQuery()) {
ArrayList listResult = new ArrayList();
JsonObjectBuilder row = Json.createObjectBuilder();
ResultSetMetaData metaData = rs.getMetaData();
while (rs.next()) {
for (int i = 1; i <= metaData.getColumnCount(); i++) {
row = Utils.getColumnDataByType(rs, metaData, i, row);
if (metaData.getColumnName(i).toUpperCase().equals(pollingField.toUpperCase())) {
if (maxPollingValue.before(rs.getTimestamp(i))) {
if (rs.getString(metaData.getColumnName(i)).length() > 10) {
maxPollingValue = java.sql.Timestamp
.valueOf(rs.getString(metaData.getColumnName(i)));
} else {
maxPollingValue = java.sql.Timestamp
.valueOf(rs.getString(metaData.getColumnName(i)) + " 00:00:00");
}
}
}
}
listResult.add(row.build());
}
return listResult;
}
}
String sql = String.format("SELECT * FROM ("
+ "SELECT ROW_NUMBER() OVER( ORDER BY %s) as rn, o.* from %s o WHERE %s > ?) "
+ "WHERE rn<=? ORDER BY %s",
pollingField,
tableName,
pollingField,
pollingField);
return getRowsExecutePolling(connection, sql);
}

public JsonObject executeLookup(Connection connection, JsonObject body) throws SQLException {
Expand All @@ -60,7 +30,7 @@ public JsonObject executeLookup(Connection connection, JsonObject body) throws S
tableName + " b) WHERE " + lookupField + " = ? " +
"AND rnk BETWEEN ? AND ? " +
"ORDER BY " + lookupField;
return Utils.getLookupRow(connection, body, sql, skipNumber, countNumber);
return getLookupRow(connection, body, sql, skipNumber, countNumber);
}

public int executeDelete(Connection connection, JsonObject body) throws SQLException {
Expand All @@ -73,14 +43,6 @@ public int executeDelete(Connection connection, JsonObject body) throws SQLExcep
}
}

public boolean executeRecordExists(Connection connection, JsonObject body) throws SQLException {
validateQuery();
String sql = "SELECT COUNT(*)" +
" FROM " + tableName +
" WHERE " + lookupField + " = ?";
return Utils.isRecordExists(connection, body, sql, lookupField);
}

public void executeInsert(Connection connection, String tableName, JsonObject body)
throws SQLException {
validateQuery();
Expand Down
46 changes: 3 additions & 43 deletions src/main/java/io/elastic/jdbc/QueryBuilders/PostgreSQL.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,14 +3,10 @@
import io.elastic.jdbc.Utils;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Map;
import javax.json.Json;
import javax.json.JsonObject;
import javax.json.JsonObjectBuilder;
import javax.json.JsonValue;

public class PostgreSQL extends Query {
Expand All @@ -27,36 +23,8 @@ public ArrayList executePolling(Connection connection) throws SQLException {
" )" +
" SELECT *" +
" FROM results_cte" +
" WHERE rownum > ?" +
" AND rownum < ?";
try (PreparedStatement stmt = connection.prepareStatement(sql)) {
stmt.setTimestamp(1, pollingValue);
stmt.setInt(2, skipNumber);
stmt.setInt(3, countNumber + skipNumber);
try (ResultSet rs = stmt.executeQuery()) {
ArrayList listResult = new ArrayList();
JsonObjectBuilder row = Json.createObjectBuilder();
ResultSetMetaData metaData = rs.getMetaData();
while (rs.next()) {
for (int i = 1; i <= metaData.getColumnCount(); i++) {
row = Utils.getColumnDataByType(rs, metaData, i, row);
if (metaData.getColumnName(i).toUpperCase().equals(pollingField.toUpperCase())) {
if (maxPollingValue.before(rs.getTimestamp(i))) {
if (rs.getString(metaData.getColumnName(i)).length() > 10) {
maxPollingValue = java.sql.Timestamp
.valueOf(rs.getString(metaData.getColumnName(i)));
} else {
maxPollingValue = java.sql.Timestamp
.valueOf(rs.getString(metaData.getColumnName(i)) + " 00:00:00");
}
}
}
}
listResult.add(row.build());
}
return listResult;
}
}
" WHERE rownum <= ?";
return getRowsExecutePolling(connection, sql);
}

public JsonObject executeLookup(Connection connection, JsonObject body) throws SQLException {
Expand All @@ -73,7 +41,7 @@ public JsonObject executeLookup(Connection connection, JsonObject body) throws S
" FROM results_cte" +
" WHERE rownum > ?" +
" AND rownum < ?";
return Utils.getLookupRow(connection, body, sql, skipNumber, countNumber + skipNumber);
return getLookupRow(connection, body, sql, skipNumber, countNumber + skipNumber);
}

public int executeDelete(Connection connection, JsonObject body) throws SQLException {
Expand All @@ -89,14 +57,6 @@ public int executeDelete(Connection connection, JsonObject body) throws SQLExcep
}
}

public boolean executeRecordExists(Connection connection, JsonObject body) throws SQLException {
validateQuery();
String sql = "SELECT COUNT(*)" +
" FROM " + tableName +
" WHERE " + lookupField + " = ?";
return Utils.isRecordExists(connection, body, sql, lookupField);
}

public void executeInsert(Connection connection, String tableName, JsonObject body)
throws SQLException {
validateQuery();
Expand Down
Loading