You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@apex.apache.org by bh...@apache.org on 2016/04/22 15:44:22 UTC
[1/2] incubator-apex-malhar git commit: APEXMALHAR-1985: Using
startRow to fetch startRowToken
Repository: incubator-apex-malhar
Updated Branches:
refs/heads/master bdd93488e -> 26237a2fb
APEXMALHAR-1985: Using startRow to fetch startRowToken
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/df22b468
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/df22b468
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/df22b468
Branch: refs/heads/master
Commit: df22b468d670d655bfb9b40eafaf07ee2223c0af
Parents: 0d49ea3
Author: Priyanka Gugale <pr...@datatorrent.com>
Authored: Wed Jan 27 17:55:57 2016 +0530
Committer: Priyanka Gugale <pr...@datatorrent.com>
Committed: Thu Apr 21 16:15:27 2016 +0530
----------------------------------------------------------------------
.../cassandra/CassandraPOJOInputOperator.java | 74 ++++++++++++++------
.../cassandra/CassandraOperatorTest.java | 8 +--
2 files changed, 58 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/df22b468/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraPOJOInputOperator.java
----------------------------------------------------------------------
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 9f2a08c..3d87711 100644
--- a/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraPOJOInputOperator.java
+++ b/contrib/src/main/java/com/datatorrent/contrib/cassandra/CassandraPOJOInputOperator.java
@@ -28,16 +28,17 @@ import org.apache.hadoop.classification.InterfaceStability.Evolving;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
-
import com.datatorrent.api.Context;
import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.api.DefaultOutputPort;
import com.datatorrent.api.Operator;
import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
-
import com.datatorrent.lib.util.FieldInfo;
import com.datatorrent.lib.util.PojoUtils;
import com.datatorrent.lib.util.PojoUtils.*;
@@ -62,7 +63,8 @@ public class CassandraPOJOInputOperator extends AbstractCassandraInputOperator<O
{
@NotNull
private List<FieldInfo> fieldInfos;
- private Number startRow = 0;
+ private Number startRow;
+ private Long startRowToken = Long.MIN_VALUE;
@NotNull
private String tablename;
@NotNull
@@ -73,8 +75,10 @@ public class CassandraPOJOInputOperator extends AbstractCassandraInputOperator<O
@Min(1)
private int limit = 10;
+ private String TOKEN_QUERY;
private transient DataType primaryKeyColumnType;
private transient Row lastRowInBatch;
+ private transient BoundStatement fetchKeyStatement;
protected final transient List<Object> setters;
protected final transient List<DataType> columnDataTypes;
@@ -189,6 +193,19 @@ public class CassandraPOJOInputOperator extends AbstractCassandraInputOperator<O
}
@Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ Long keyToken;
+ TOKEN_QUERY = "select token(" + primaryKeyColumn + ") from " + store.keyspace + "." + tablename + " where " + primaryKeyColumn + " = ?";
+ PreparedStatement statement = store.getSession().prepare(TOKEN_QUERY);
+ fetchKeyStatement = new BoundStatement(statement);
+ if (startRow != null && (keyToken = fetchKeyTokenFromDB(startRow)) != null) {
+ startRowToken = keyToken;
+ }
+ }
+
+ @Override
public void activate(OperatorContext context)
{
com.datastax.driver.core.ResultSet rs = store.getSession().execute("select * from " + store.keyspace + "." + tablename + " LIMIT " + 1);
@@ -348,7 +365,7 @@ public class CassandraPOJOInputOperator extends AbstractCassandraInputOperator<O
public String queryToRetrieveData()
{
if (query.contains("%v")) {
- return query.replace("%v", startRow + "");
+ return query.replace("%v", startRowToken + "");
}
return query;
}
@@ -362,24 +379,41 @@ public class CassandraPOJOInputOperator extends AbstractCassandraInputOperator<O
{
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());
- }
+ startRowToken = getPrimaryKeyToken(primaryKeyColumnType.getName());
}
+ }
+ private Long getPrimaryKeyToken(DataType.Name primaryKeyDataType)
+ {
+ Object keyValue;
+ switch (primaryKeyDataType) {
+ case UUID:
+ keyValue = lastRowInBatch.getUUID(primaryKeyColumn);
+ break;
+ case INT:
+ keyValue = lastRowInBatch.getInt(primaryKeyColumn);
+ break;
+ case COUNTER:
+ keyValue = lastRowInBatch.getLong(primaryKeyColumn);
+ break;
+ case FLOAT:
+ keyValue = lastRowInBatch.getFloat(primaryKeyColumn);
+ break;
+ case DOUBLE:
+ keyValue = lastRowInBatch.getDouble(primaryKeyColumn);
+ break;
+ default:
+ throw new RuntimeException("unsupported data type " + primaryKeyColumnType.getName());
+ }
+ return fetchKeyTokenFromDB(keyValue);
+ }
+
+ private Long fetchKeyTokenFromDB(Object keyValue)
+ {
+ fetchKeyStatement.bind(keyValue);
+ ResultSet rs = store.getSession().execute(fetchKeyStatement);
+ Long keyTokenValue = rs.one().getLong(0);
+ return keyTokenValue;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/df22b468/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java
----------------------------------------------------------------------
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 56d9857..f4aa29a 100644
--- a/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java
+++ b/contrib/src/test/java/com/datatorrent/contrib/cassandra/CassandraOperatorTest.java
@@ -358,27 +358,27 @@ public class CassandraOperatorTest
String query2 = "SELECT * FROM " + KEYSPACE + "." + "%t where token(%p) > %v;";
inputOperator.setQuery(query2);
+ inputOperator.setStartRow(10);
inputOperator.setup(context);
inputOperator.outputPort.setup(tpc);
inputOperator.activate(context);
- inputOperator.setStartRow(10);
inputOperator.beginWindow(1);
inputOperator.emitTuples();
inputOperator.endWindow();
- Assert.assertEquals("rows from db", 14, sink.collectedTuples.size());
+ Assert.assertEquals("rows from db", 26, sink.collectedTuples.size());
sink.clear();
inputOperator.columnDataTypes.clear();
String query3 = "SELECT * FROM " + KEYSPACE + "." + "%t where token(%p) > %v LIMIT %l;";
inputOperator.setQuery(query3);
+ inputOperator.setStartRow(1);
+ inputOperator.setLimit(10);
inputOperator.setup(context);
inputOperator.outputPort.setup(tpc);
inputOperator.activate(context);
- inputOperator.setStartRow(1);
- inputOperator.setLimit(10);
inputOperator.beginWindow(2);
inputOperator.emitTuples();
inputOperator.endWindow();
[2/2] incubator-apex-malhar git commit: Merge branch
'APEXMALHAR-1985_cassandra_scan' of
https://github.com/DT-Priyanka/incubator-apex-malhar
Posted by bh...@apache.org.
Merge branch 'APEXMALHAR-1985_cassandra_scan' of https://github.com/DT-Priyanka/incubator-apex-malhar
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/26237a2f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/26237a2f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/26237a2f
Branch: refs/heads/master
Commit: 26237a2fb720605767e66677255172e806cd5073
Parents: bdd9348 df22b46
Author: bhupesh <bh...@gmail.com>
Authored: Fri Apr 22 18:28:26 2016 +0530
Committer: bhupesh <bh...@gmail.com>
Committed: Fri Apr 22 18:28:26 2016 +0530
----------------------------------------------------------------------
.../cassandra/CassandraPOJOInputOperator.java | 74 ++++++++++++++------
.../cassandra/CassandraOperatorTest.java | 8 +--
2 files changed, 58 insertions(+), 24 deletions(-)
----------------------------------------------------------------------