changes.

contrib/src/main/java/com/datatorrent/contrib/memsql/MemsqlOutputOperator.java

diff --git a/contrib/src/main/java/com/datatorrent/contrib/memsql/MemsqlOutputOperator.java b/contrib/src/main/java/com/datatorrent/contrib/memsql/MemsqlOutputOperator.java
index ce83aa7..795cd53 100644
--- a/contrib/src/main/java/com/datatorrent/contrib/memsql/MemsqlOutputOperator.java
+++ b/contrib/src/main/java/com/datatorrent/contrib/memsql/MemsqlOutputOperator.java
@@ -192,23 +192,23 @@
       String getterExpression = expression.get(i);
       if (type == Types.CHAR) {
         GetterChar getChar = PojoUtils.createGetterChar(fqcn, getterExpression);
-        getters.add(getChar.get(tuple));
+        getters.add(getChar);
       }
       else if (type == Types.VARCHAR) {
         GetterString getVarchar = PojoUtils.createGetterString(fqcn, getterExpression);
-        getters.add(getVarchar.get(tuple));
+        getters.add(getVarchar);
       }
       else if (type == Types.BOOLEAN || type == Types.TINYINT) {
         GetterBoolean getBoolean = PojoUtils.createGetterBoolean(fqcn, getterExpression);
-        getters.add(getBoolean.get(tuple));
+        getters.add(getBoolean);
       }
       else if (type == Types.SMALLINT) {
         GetterShort getShort = PojoUtils.createGetterShort(fqcn, getterExpression);
-        getters.add(getShort.get(tuple));
+        getters.add(getShort);
       }
       else if (type == Types.INTEGER) {
         GetterInt getInt = PojoUtils.createGetterInt(fqcn, getterExpression);
-        getters.add(getInt.get(tuple));
+        getters.add(getInt);
       }
       else if (type == Types.BIGINT) {
         GetterLong getLong = PojoUtils.createExpressionGetterLong(fqcn, getterExpression);
@@ -216,31 +216,31 @@
       }
       else if (type == Types.DECIMAL) {
         GetterObject getObject = PojoUtils.createGetterObject(fqcn, getterExpression);
-        getters.add((Number)getObject.get(tuple));
+        getters.add(getObject);
       }
       else if (type == Types.FLOAT) {
         GetterFloat getFloat = PojoUtils.createGetterFloat(fqcn, getterExpression);
-        getters.add(getFloat.get(tuple));
+        getters.add(getFloat);
       }
       else if (type == Types.DOUBLE) {
         GetterDouble getDouble = PojoUtils.createGetterDouble(fqcn, getterExpression);
-        getters.add(getDouble.get(tuple));
+        getters.add(getDouble);
       }
       else if (type == Types.DATE) {
         GetterObject getObject = PojoUtils.createGetterObject(fqcn, getterExpression);
-        getters.add((Date)getObject.get(tuple));
+        getters.add(getObject);
       }
       else if (type == Types.TIME) {
         GetterObject getObject = PojoUtils.createGetterObject(fqcn, getterExpression);
-        getters.add((Time)getObject.get(tuple));
+        getters.add(getObject);
       }
       else if (type == Types.ARRAY) {
         GetterObject getObject = PojoUtils.createGetterObject(fqcn, getterExpression);
-        getters.add((Array)getObject.get(tuple));
+        getters.add(getObject);
       }
       else if (type == Types.OTHER) {
         GetterObject getObject = PojoUtils.createGetterObject(fqcn, getterExpression);
-        getters.add(getObject.get(tuple));
+        getters.add(getObject);
       }
 
     }
@@ -256,9 +256,55 @@
   @Override
   protected void setStatementParameters(PreparedStatement statement, Object tuple) throws SQLException
   {
-    int size = dataColumns.size();
+    int size = columnDataTypes.size();
+    Object getter;
     for (int i = 0; i < size; i++) {
-      statement.setObject(i + 1, getters.get(i));
+      int type = columnDataTypes.get(i);
+      switch (type) {
+        case (Types.CHAR):
+          getter = ((GetterChar)getters.get(i)).get(tuple);
+          break;
+        case (Types.VARCHAR):
+          getter = ((GetterString)getters.get(i)).get(tuple);
+          break;
+        case (Types.BOOLEAN):
+          getter = ((GetterString)getters.get(i)).get(tuple);
+          break;
+        case (Types.SMALLINT):
+          getter = ((GetterString)getters.get(i)).get(tuple);
+          break;
+        case (Types.INTEGER):
+          getter = ((GetterString)getters.get(i)).get(tuple);
+          break;
+        case (Types.BIGINT):
+          getter = ((GetterString)getters.get(i)).get(tuple);
+          break;
+        case (Types.DECIMAL):
+          getter = ((GetterString)getters.get(i)).get(tuple);
+          break;
+        case (Types.FLOAT):
+          getter = ((GetterString)getters.get(i)).get(tuple);
+          break;
+        case (Types.DOUBLE):
+          getter = ((GetterString)getters.get(i)).get(tuple);
+          break;
+        case (Types.DATE):
+          getter = ((GetterString)getters.get(i)).get(tuple);
+          break;
+        case (Types.TIME):
+          getter = ((GetterString)getters.get(i)).get(tuple);
+          break;
+        case (Types.ARRAY):
+          getter = ((GetterString)getters.get(i)).get(tuple);
+          break;
+        case (Types.OTHER):
+          getter = ((GetterString)getters.get(i)).get(tuple);
+          break;
+        default:
+          getter = ((GetterString)getters.get(i)).get(tuple);
+          break;
+      }
+       statement.setObject(i + 1, getter);
     }
   }
 

contrib/src/main/java/com/datatorrent/contrib/splunk/SplunkInputOperator.java

diff --git a/contrib/src/main/java/com/datatorrent/contrib/splunk/SplunkInputOperator.java b/contrib/src/main/java/com/datatorrent/contrib/splunk/SplunkInputOperator.java
new file mode 100644
index 0000000..0aa510a
--- /dev/null
+++ b/contrib/src/main/java/com/datatorrent/contrib/splunk/SplunkInputOperator.java
@@ -0,0 +1,38 @@
+/*
+ *  Copyright (c) 2012-2015 Malhar, Inc.
+ *  All Rights Reserved.
+ */
+package com.datatorrent.contrib.splunk;
+
+import javax.validation.constraints.NotNull;
+
+public class SplunkInputOperator extends AbstractSplunkInputOperator<String>
+{
+  @NotNull
+  private String query = "search * | head 100";
+
+  @Override
+  public String getTuple(String value)
+  {
+    return value;
+  }
+
+  @Override
+  public String queryToRetrieveData()
+  {
+    return query;
+  }
+
+  /*
+   * Query to retrieve data from Splunk.
+   */
+  public String getQuery()
+  {
+    return query;
+  }
+
+  public void setQuery(String query)
+  {
+    this.query = query;
+  }
+}