From 3189a13da1f231cdad8c2e3b9734a1c7d5205ad9 Mon Sep 17 00:00:00 2001
From: Bryan Bende <bbende@apache.org>
Date: Wed, 23 Dec 2015 10:02:21 -0500
Subject: [PATCH] NIFI-1328 Explicitly setting the parameters based on type in
 PutSQL, falling back to setObject()

This closes #146
---
 .../nifi/processors/standard/PutSQL.java      | 107 +++++++++++++-----
 1 file changed, 81 insertions(+), 26 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 91daecd7a2..5b1a048163 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
@@ -16,31 +16,6 @@
  */
 package org.apache.nifi.processors.standard;
 
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.charset.StandardCharsets;
-import java.sql.BatchUpdateException;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.SQLNonTransientException;
-import java.sql.Statement;
-import java.util.ArrayList;
-import java.util.BitSet;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.ListIterator;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
 import org.apache.nifi.annotation.behavior.InputRequirement;
 import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
 import org.apache.nifi.annotation.behavior.ReadsAttribute;
@@ -64,6 +39,36 @@ import org.apache.nifi.processor.io.InputStreamCallback;
 import org.apache.nifi.processor.util.StandardValidators;
 import org.apache.nifi.stream.io.StreamUtils;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.sql.BatchUpdateException;
+import java.sql.Connection;
+import java.sql.Date;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLNonTransientException;
+import java.sql.Statement;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.sql.Types;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
 @SupportsBatching
 @SeeAlso(ConvertJSONToSQL.class)
 @InputRequirement(Requirement.INPUT_REQUIRED)
@@ -725,7 +730,57 @@ public class PutSQL extends AbstractProcessor {
      * @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 {
-        stmt.setObject(parameterIndex, parameterValue, jdbcType);
+        if (parameterValue == null) {
+            stmt.setNull(parameterIndex, jdbcType);
+        } else {
+            switch (jdbcType) {
+                case Types.BIT:
+                case Types.BOOLEAN:
+                    stmt.setBoolean(parameterIndex, Boolean.parseBoolean(parameterValue));
+                    break;
+                case Types.TINYINT:
+                    stmt.setByte(parameterIndex, Byte.parseByte(parameterValue));
+                    break;
+                case Types.SMALLINT:
+                    stmt.setShort(parameterIndex, Short.parseShort(parameterValue));
+                    break;
+                case Types.INTEGER:
+                    stmt.setInt(parameterIndex, Integer.parseInt(parameterValue));
+                    break;
+                case Types.BIGINT:
+                    stmt.setLong(parameterIndex, Long.parseLong(parameterValue));
+                    break;
+                case Types.REAL:
+                    stmt.setFloat(parameterIndex, Float.parseFloat(parameterValue));
+                    break;
+                case Types.FLOAT:
+                case Types.DOUBLE:
+                    stmt.setDouble(parameterIndex, Double.parseDouble(parameterValue));
+                    break;
+                case Types.DECIMAL:
+                case Types.NUMERIC:
+                    stmt.setBigDecimal(parameterIndex, new BigDecimal(parameterValue));
+                    break;
+                case Types.DATE:
+                    stmt.setDate(parameterIndex, new Date(Long.parseLong(parameterValue)));
+                    break;
+                case Types.TIME:
+                    stmt.setTime(parameterIndex, new Time(Long.parseLong(parameterValue)));
+                    break;
+                case Types.TIMESTAMP:
+                    stmt.setTimestamp(parameterIndex, new Timestamp(Long.parseLong(parameterValue)));
+                    break;
+                case Types.CHAR:
+                case Types.VARCHAR:
+                case Types.LONGNVARCHAR:
+                case Types.LONGVARCHAR:
+                    stmt.setString(parameterIndex, parameterValue);
+                    break;
+                default:
+                    stmt.setObject(parameterIndex, parameterValue, jdbcType);
+                    break;
+            }
+        }
     }