From 92eb864c96dc3096c28f6d3a2e13402d7bc0a292 Mon Sep 17 00:00:00 2001 From: Peter Wicks Date: Mon, 15 Aug 2016 10:49:37 -0600 Subject: [PATCH 1/4] NIFI-2356 --- .../processors/standard/ConvertJSONToSQL.java | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertJSONToSQL.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertJSONToSQL.java index cbc301ab4c84..194aa1953a45 100755 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertJSONToSQL.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertJSONToSQL.java @@ -23,6 +23,7 @@ import java.nio.charset.StandardCharsets; import java.sql.Connection; import java.sql.ResultSet; +import java.sql.ResultSetMetaData; import java.sql.SQLException; import java.util.ArrayList; import java.util.HashMap; @@ -668,6 +669,13 @@ public boolean isRequired() { } public static ColumnDescription from(final ResultSet resultSet) throws SQLException { + final ResultSetMetaData md = resultSet.getMetaData(); + List columns = new ArrayList<>(); + + for (int i = 1; i < md.getColumnCount() + 1; i++) { + columns.add(md.getColumnName(i)); + } + final String columnName = resultSet.getString("COLUMN_NAME"); final int dataType = resultSet.getInt("DATA_TYPE"); final int colSize = resultSet.getInt("COLUMN_SIZE"); @@ -675,7 +683,12 @@ public static ColumnDescription from(final ResultSet resultSet) throws SQLExcept final String nullableValue = resultSet.getString("IS_NULLABLE"); final boolean isNullable = "YES".equalsIgnoreCase(nullableValue) || nullableValue.isEmpty(); final String defaultValue = resultSet.getString("COLUMN_DEF"); - final String autoIncrementValue = resultSet.getString("IS_AUTOINCREMENT"); + String autoIncrementValue = "NO"; + + if(columns.contains("IS_AUTOINCREMENT")){ + autoIncrementValue = resultSet.getString("IS_AUTOINCREMENT"); + } + final boolean isAutoIncrement = "YES".equalsIgnoreCase(autoIncrementValue); final boolean required = !isNullable && !isAutoIncrement && defaultValue == null; From ab0568341c5639998f1a292311a58069ead21e7a Mon Sep 17 00:00:00 2001 From: Peter Wicks Date: Mon, 15 Aug 2016 15:50:47 -0600 Subject: [PATCH 2/4] NIFI-2576 --- .../nifi/processors/standard/PutSQL.java | 13 +++++- .../nifi/processors/standard/TestPutSQL.java | 43 +++++++++++++++++++ 2 files changed, 55 insertions(+), 1 deletion(-) diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java index 5b1a048163b9..dcf9430b22f4 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java @@ -54,6 +54,8 @@ import java.sql.Time; import java.sql.Timestamp; import java.sql.Types; +import java.text.ParseException; +import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.BitSet; import java.util.Collections; @@ -768,7 +770,16 @@ private void setParameter(final PreparedStatement stmt, final String attrName, f stmt.setTime(parameterIndex, new Time(Long.parseLong(parameterValue))); break; case Types.TIMESTAMP: - stmt.setTimestamp(parameterIndex, new Timestamp(Long.parseLong(parameterValue))); + long lTimestamp=0L; + + try { + SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss.SSS"); + java.util.Date parsedDate = dateFormat.parse(parameterValue); + stmt.setTimestamp(parameterIndex, new Timestamp(parsedDate.getTime())); + } catch(ParseException e){ + stmt.setTimestamp(parameterIndex, new Timestamp(Long.parseLong(parameterValue))); + } + break; case Types.CHAR: case Types.VARCHAR: diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutSQL.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutSQL.java index b5ebca52c8ea..592d329767b6 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutSQL.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutSQL.java @@ -28,6 +28,8 @@ import java.sql.SQLException; import java.sql.Statement; import java.sql.Types; +import java.text.ParseException; +import java.text.SimpleDateFormat; import java.util.HashMap; import java.util.Map; @@ -261,6 +263,47 @@ public void testUsingSqlDataTypesWithNegativeValues() throws InitializationExcep } } + @Test + public void testUsingTimestampValuesEpochAndString() throws InitializationException, ProcessException, SQLException, IOException, ParseException { + final TestRunner runner = TestRunners.newTestRunner(PutSQL.class); + try (final Connection conn = service.getConnection()) { + try (final Statement stmt = conn.createStatement()) { + stmt.executeUpdate("CREATE TABLE TIMESTAMPTESTS (id integer primary key, ts1 timestamp, ts2 timestamp)"); + } + } + + runner.addControllerService("dbcp", service); + runner.enableControllerService(service); + runner.setProperty(PutSQL.CONNECTION_POOL, "dbcp"); + + final String arg2TS = "2001-01-01 23:01:01.001"; + final String art3TS = "2002-02-02 22:02:02.002"; + SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss.SSS"); + java.util.Date parsedDate = dateFormat.parse(arg2TS); + + final Map attributes = new HashMap<>(); + attributes.put("sql.args.1.type", String.valueOf(Types.TIMESTAMP)); + attributes.put("sql.args.1.value", Long.toString(parsedDate.getTime())); + attributes.put("sql.args.2.type", String.valueOf(Types.TIMESTAMP)); + attributes.put("sql.args.2.value", art3TS); + + runner.enqueue("INSERT INTO TIMESTAMPTESTS (ID, ts1, ts2) VALUES (1, ?, ?)".getBytes(), attributes); + runner.run(); + + runner.assertAllFlowFilesTransferred(PutSQL.REL_SUCCESS, 1); + + try (final Connection conn = service.getConnection()) { + try (final Statement stmt = conn.createStatement()) { + final ResultSet rs = stmt.executeQuery("SELECT * FROM TIMESTAMPTESTS"); + assertTrue(rs.next()); + assertEquals(1, rs.getInt(1)); + assertEquals(arg2TS, rs.getString(2)); + assertEquals(art3TS, rs.getString(3)); + assertFalse(rs.next()); + } + } + } + @Test public void testStatementsWithPreparedParameters() throws InitializationException, ProcessException, SQLException, IOException { final TestRunner runner = TestRunners.newTestRunner(PutSQL.class); From 4ce80fa69ac8524756818dd6d530954985b430c5 Mon Sep 17 00:00:00 2001 From: Peter Wicks Date: Mon, 15 Aug 2016 16:00:31 -0600 Subject: [PATCH 3/4] NIFI-2576 --- .../java/org/apache/nifi/processors/standard/PutSQL.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java index dcf9430b22f4..a6b171c36d89 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java @@ -775,11 +775,13 @@ private void setParameter(final PreparedStatement stmt, final String attrName, f try { SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss.SSS"); java.util.Date parsedDate = dateFormat.parse(parameterValue); - stmt.setTimestamp(parameterIndex, new Timestamp(parsedDate.getTime())); + lTimestamp = parsedDate.getTime(); } catch(ParseException e){ - stmt.setTimestamp(parameterIndex, new Timestamp(Long.parseLong(parameterValue))); + lTimestamp = Long.parseLong(parameterValue); } + stmt.setTimestamp(parameterIndex, new Timestamp(lTimestamp)); + break; case Types.CHAR: case Types.VARCHAR: From c680ff7f17cf67a778e5afd31f7283f5ea7f8a73 Mon Sep 17 00:00:00 2001 From: Peter Wicks Date: Mon, 15 Aug 2016 21:03:30 -0600 Subject: [PATCH 4/4] NIFI-2576 --- .../org/apache/nifi/processors/standard/PutSQL.java | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java index a6b171c36d89..55d3855e9abd 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java @@ -156,6 +156,8 @@ public class PutSQL extends AbstractProcessor { private static final String FRAGMENT_INDEX_ATTR = "fragment.index"; private static final String FRAGMENT_COUNT_ATTR = "fragment.count"; + private static final Pattern LONG_PATTERN = Pattern.compile("^\\d{1,19}$"); + @Override protected List getSupportedPropertyDescriptors() { final List properties = new ArrayList<>(); @@ -613,6 +615,8 @@ private void setParameters(final PreparedStatement stmt, final Map flowFiles, final Long tr * @param jdbcType the JDBC Type of the SQL parameter to set * @throws SQLException if the PreparedStatement throws a SQLException when calling the appropriate setter */ - private void setParameter(final PreparedStatement stmt, final String attrName, final int parameterIndex, final String parameterValue, final int jdbcType) throws SQLException { + private void setParameter(final PreparedStatement stmt, final String attrName, final int parameterIndex, final String parameterValue, final int jdbcType) throws SQLException, ParseException { if (parameterValue == null) { stmt.setNull(parameterIndex, jdbcType); } else { @@ -772,12 +776,12 @@ private void setParameter(final PreparedStatement stmt, final String attrName, f case Types.TIMESTAMP: long lTimestamp=0L; - try { + if(LONG_PATTERN.matcher(parameterValue).matches()){ + lTimestamp = Long.parseLong(parameterValue); + }else { SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss.SSS"); java.util.Date parsedDate = dateFormat.parse(parameterValue); lTimestamp = parsedDate.getTime(); - } catch(ParseException e){ - lTimestamp = Long.parseLong(parameterValue); } stmt.setTimestamp(parameterIndex, new Timestamp(lTimestamp));