diff --git a/contrib/src/main/java/com/datatorrent/contrib/cassandra/AbstractCassandraInputOperator.java b/contrib/src/main/java/com/datatorrent/contrib/cassandra/AbstractCassandraInputOperator.java
index eb48846..7e67398 100644
--- a/contrib/src/main/java/com/datatorrent/contrib/cassandra/AbstractCassandraInputOperator.java
+++ b/contrib/src/main/java/com/datatorrent/contrib/cassandra/AbstractCassandraInputOperator.java
@@ -34,7 +34,6 @@
  * <p>
  * This is an abstract class. Sub-classes need to implement {@link #queryToRetrieveData()} and {@link #getTuple(Row)}.
  * </p>
- *
  * @param <T>
  * @displayName Abstract Cassandra Input
  * @category Store
@@ -75,27 +74,17 @@
   {
     String query = queryToRetrieveData();
     logger.debug(String.format("select statement: %s", query));
-    if (!"".equals(query)) {
-      try {
-        ResultSet result = store.getSession().execute(query);
-        processResult(result);
-      }
-      catch (Exception ex) {
-        store.disconnect();
-        DTThrowable.rethrow(ex);
+
+    try {
+      ResultSet result = store.getSession().execute(query);
+      for(Row row: result) {
+        T tuple = getTuple(row);
+        outputPort.emit(tuple);
       }
     }
-  }
-
-  /*
-   * ProcessResult can be overridden in classes providing implementation of  AbstractCassandraInputOperator.
-   * Providing a basic implementation here which iterates through the rows in result,gets the particular tuple from row and emits it.
-   */
-  protected  void processResult(ResultSet result)
-  {
-    for (Row row: result) {
-          T tuple = getTuple(row);
-          outputPort.emit(tuple);
+    catch (Exception ex) {
+      store.disconnect();
+      DTThrowable.rethrow(ex);
     }
   }
 }
diff --git a/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraPOJOInputOperator.java b/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraPOJOInputOperator.java
index b1caa66..edd84a6 100644
--- a/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraPOJOInputOperator.java
+++ b/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraPOJOInputOperator.java
@@ -17,7 +17,6 @@
 
 import com.datastax.driver.core.ColumnDefinitions;
 import com.datastax.driver.core.DataType;
-import com.datastax.driver.core.ResultSet;
 import com.datastax.driver.core.Row;
 import com.datatorrent.api.Context.OperatorContext;
 import com.datatorrent.lib.util.PojoUtils;
@@ -36,8 +35,9 @@
 
 /**
  * <p>
- * CassandraInputOperator</p>
+ * CassandraPOJOInputOperator</p>
  * A Generic implementation of AbstractCassandraInputOperator which gets field values from Cassandra database columns and sets in a POJO.
+ * User can give a parameterized query with parameters %t for table name, %p for primary key, %s for start value and %l for limit.
  *
  * @displayName Cassandra POJO Input Operator
  * @category Input
@@ -54,13 +54,14 @@
   private String tablename;
   private final transient List<Object> setters;
   @NotNull
-  private String retrieveQuery;
+  private String query;
+
   private transient Class<?> objectClass = null;
   private boolean useAllColumns;
-  protected Number lastRowIdInBatch = 0;
   @NotNull
   protected String primaryKeyColumn;
   protected transient DataType primaryKeyColumnType;
+  private transient Row lastRowInBatch;
 
   @Min(1)
   private int limit = 10;
@@ -139,16 +140,18 @@
   }
 
   /*
-   * Query input by user: Example: select * from keyspace.tablename;
+   * Parameterized query with parameters such as %t for table name , %p for primary key, %s for start value and %l for limit.
+   * Example of retrieveQuery:
+   * select * from %t where token(%p) > %s limit %l;
    */
-  public String getRetrieveQuery()
+  public String getQuery()
   {
-    return retrieveQuery;
+    return query;
   }
 
-  public void setRetrieveQuery(String retrieveQuery)
+  public void setQuery(String query)
   {
-    this.retrieveQuery = retrieveQuery;
+    this.query = query.replace("%t", tablename);
   }
 
   /*
@@ -224,6 +227,17 @@
       }
 
       primaryKeyColumnType = rsMetaData.getType(primaryKeyColumn);
+       if(query.contains("%p"))
+       {
+          query = query.replace("%p", primaryKeyColumn);
+       }
+       if(query.contains("%l"))
+       {
+         query = query.replace("%l", limit+"");
+       }
+
+
+       logger.debug("query is {}",query);
 
       for (int i = 0; i < numberOfColumns; i++) {
         // Get the designated column's data type.
@@ -290,6 +304,7 @@
   @SuppressWarnings("unchecked")
   public Object getTuple(Row row)
   {
+    lastRowInBatch = row;
     Object obj = null;
     final int size = columnDataTypes.size();
 
@@ -366,76 +381,53 @@
     return obj;
   }
 
+  /*
+   * This method replaces the parameters in Query with actual values given by user.
+   * Example of retrieveQuery:
+   * select * from %t where token(%p) > %v limit %l;
+   */
   @Override
   public String queryToRetrieveData()
   {
-    boolean flag = false;
-
-    switch (primaryKeyColumnType.getName()) {
-      case INT:
-        if (startRow.intValue() > lastRowIdInBatch.intValue()) {
-          flag = true;
-        }
-        break;
-      case COUNTER:
-        if (startRow.longValue() > lastRowIdInBatch.longValue()) {
-          flag = true;
-        }
-        break;
-      case FLOAT:
-        if (startRow.floatValue() > lastRowIdInBatch.floatValue()) {
-          flag = true;
-        }
-        break;
-      case DOUBLE:
-        if (startRow.doubleValue() > lastRowIdInBatch.doubleValue()) {
-          flag = true;
-        }
-        break;
+    String parameterizedQuery;
+    if(query.contains("%v"))
+    {
+      parameterizedQuery = query.replace("%v", startRow+"");
     }
-
-    if (flag) {
-      return "";
+    else
+    {
+      parameterizedQuery = query;
     }
-
-    startRow = lastRowIdInBatch.intValue() + 1;
-    StringBuilder sb = new StringBuilder();
-    sb.append(retrieveQuery).append(" where ").append("token(").append(primaryKeyColumn).append(")").append(">=").append(startRow).append(" LIMIT ").append(limit);
-    logger.debug("retrievequery is {}", sb.toString());
-
-    return sb.toString();
+    return parameterizedQuery;
   }
 
+
   /*
-   * Overriding processResult to save primarykey column value from last row in batch.
+   * Overriding emitTupes to save primarykey column value from last row in batch.
    */
   @Override
-  protected void processResult(ResultSet result)
+  public void emitTuples()
   {
-      Row lastRowInBatch = null;
-        for (Row row: result) {
-          Object tuple = getTuple(row);
-          outputPort.emit(tuple);
-          lastRowInBatch = row;
-        }
-           if (lastRowInBatch != null) {
-          switch (primaryKeyColumnType.getName()) {
-            case INT:
-              lastRowIdInBatch = lastRowInBatch.getInt(0);
-              break;
-            case COUNTER:
-              lastRowIdInBatch = lastRowInBatch.getLong(0);
-              break;
-            case FLOAT:
-              lastRowIdInBatch = lastRowInBatch.getFloat(0);
-              break;
-            case DOUBLE:
-              lastRowIdInBatch = lastRowInBatch.getDouble(0);
-              break;
-            default:
-              throw new RuntimeException("unsupported data type " + primaryKeyColumnType.getName());
-          }
-        }
+    super.emitTuples();
+    if (lastRowInBatch != null) {
+      switch (primaryKeyColumnType.getName()) {
+        case INT:
+          startRow = lastRowInBatch.getInt(primaryKeyColumn);
+          break;
+        case COUNTER:
+          startRow = lastRowInBatch.getLong(primaryKeyColumn);
+          break;
+        case FLOAT:
+          startRow = lastRowInBatch.getFloat(primaryKeyColumn);
+          break;
+        case DOUBLE:
+          startRow = lastRowInBatch.getDouble(primaryKeyColumn);
+          break;
+        default:
+          throw new RuntimeException("unsupported data type " + primaryKeyColumnType.getName());
+      }
+    }
+
   }
 
   private static final Logger logger = LoggerFactory.getLogger(CassandraPOJOInputOperator.class);
diff --git a/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java b/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java
index a6d0d21..0658d19 100644
--- a/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java
+++ b/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java
@@ -292,7 +292,7 @@
   @Test
   public void TestCassandraInputOperator()
   {
-    String retrieveQuery = "SELECT * FROM " + KEYSPACE + "." + TABLE_NAME_INPUT;
+    String query1 = "SELECT * FROM " + KEYSPACE + "." + "%t;";
     CassandraStore store = new CassandraStore();
     store.setNode(NODE);
     store.setKeyspace(KEYSPACE);
@@ -305,7 +305,7 @@
     inputOperator.setStore(store);
     inputOperator.setOutputClass("com.datatorrent.contrib.cassandra.TestInputPojo");
     inputOperator.setTablename(TABLE_NAME_INPUT);
-    inputOperator.setRetrieveQuery(retrieveQuery);
+    inputOperator.setQuery(query1);
     ArrayList<String> columns = new ArrayList<String>();
     columns.add("id");
     columns.add("age");
@@ -324,14 +324,9 @@
     inputOperator.outputPort.setSink(sink);
     inputOperator.setup(context);
     inputOperator.beginWindow(0);
-    inputOperator.insertEventsInTable(10);
     inputOperator.emitTuples();
     inputOperator.endWindow();
-    Assert.assertEquals("rows from db", 10, sink.collectedTuples.size());
-    inputOperator.beginWindow(1);
-    inputOperator.emitTuples();
-    inputOperator.endWindow();
-    Assert.assertEquals("rows from db", 20, sink.collectedTuples.size());
+    Assert.assertEquals("rows from db", 30, sink.collectedTuples.size());
     ArrayList<Integer> listOfIDs = inputOperator.getIds();
     // Rows are not stored in the same order in cassandra table in which they are inserted.
     for (int i = 0; i < 10; i++) {
@@ -341,6 +336,25 @@
       Assert.assertEquals("age set in testpojo", inputOperator.getAge().get(object.getId()).intValue(), object.getAge());
     }
 
+    sink.clear();
+    String query2 = "SELECT * FROM " + KEYSPACE + "." + "%t where token(%p) > %v;";
+    inputOperator.setQuery(query2);
+    inputOperator.setStartRow(10);
+    inputOperator.beginWindow(1);
+    inputOperator.emitTuples();
+    inputOperator.endWindow();
+    Assert.assertEquals("rows from db", 14, sink.collectedTuples.size());
+
+    sink.clear();
+    String query3 = "SELECT * FROM " + KEYSPACE + "." + "%t where token(%p) > %v LIMIT %l;";
+    inputOperator.setQuery(query3);
+    inputOperator.setStartRow(1);
+    inputOperator.setLimit(10);
+    inputOperator.beginWindow(2);
+    inputOperator.emitTuples();
+    inputOperator.endWindow();
+    Assert.assertEquals("rows from db", 10, sink.collectedTuples.size());
+
   }
 
   public static class TestPojo