You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by to...@apache.org on 2014/08/31 03:03:26 UTC
[2/3] working out multiget
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/94a92c39/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java
index 2be6c55..7b40400 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java
@@ -19,6 +19,7 @@
package org.apache.usergrid.persistence.graph.impl.stage;
+import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@@ -29,6 +30,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.usergrid.persistence.core.hystrix.HystrixCassandra;
+import org.apache.usergrid.persistence.core.javadriver.BatchStatementUtils;
import org.apache.usergrid.persistence.core.rx.ObservableIterator;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.graph.GraphFig;
@@ -42,6 +44,9 @@ import org.apache.usergrid.persistence.graph.serialization.EdgeSerialization;
import org.apache.usergrid.persistence.graph.serialization.NodeSerialization;
import org.apache.usergrid.persistence.model.entity.Id;
+import com.datastax.driver.core.BatchStatement;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.Statement;
import com.google.common.base.Optional;
import com.google.inject.Inject;
import com.netflix.astyanax.Keyspace;
@@ -66,7 +71,7 @@ public class NodeDeleteListenerImpl implements NodeDeleteListener {
private final EdgeMetadataSerialization edgeMetadataSerialization;
private final EdgeMetaRepair edgeMetaRepair;
private final GraphFig graphFig;
- protected final Keyspace keyspace;
+ protected final Session session;
/**
@@ -77,7 +82,7 @@ public class NodeDeleteListenerImpl implements NodeDeleteListener {
final EdgeMetadataSerialization edgeMetadataSerialization,
final EdgeMetaRepair edgeMetaRepair, final GraphFig graphFig,
final EdgeSerialization storageSerialization,
- final Keyspace keyspace ) {
+ final Session session) {
this.nodeSerialization = nodeSerialization;
@@ -85,7 +90,7 @@ public class NodeDeleteListenerImpl implements NodeDeleteListener {
this.edgeMetadataSerialization = edgeMetadataSerialization;
this.edgeMetaRepair = edgeMetaRepair;
this.graphFig = graphFig;
- this.keyspace = keyspace;
+ this.session = session;
}
@@ -128,7 +133,9 @@ public class NodeDeleteListenerImpl implements NodeDeleteListener {
.doOnCompleted( new Action0() {
@Override
public void call() {
- HystrixCassandra.async(nodeSerialization.delete( scope, node, maxVersion.get() ));
+ final Collection<? extends Statement>
+ statements = nodeSerialization.delete( scope, node, maxVersion.get() );
+ BatchStatementUtils.runBatches(session, statements );
}
} );
}
@@ -151,7 +158,7 @@ public class NodeDeleteListenerImpl implements NodeDeleteListener {
//get all edges pointing to the target node and buffer then into groups for deletion
Observable<MarkedEdge> targetEdges =
- getEdgesTypesToTarget( scope, new SimpleSearchEdgeType( node, null, null ) )
+ getEdgesTypesToTarget( scope, new SimpleSearchEdgeType( node, null,null ) )
.subscribeOn( Schedulers.io() ).flatMap( new Func1<String, Observable<MarkedEdge>>() {
@Override
public Observable<MarkedEdge> call( final String edgeType ) {
@@ -192,7 +199,7 @@ public class NodeDeleteListenerImpl implements NodeDeleteListener {
LOG.debug( "Batching {} edges for node {} for deletion", markedEdges.size(), node );
- final MutationBatch batch = keyspace.prepareMutationBatch();
+ final BatchStatement batch = BatchStatementUtils.fastStatement();
Set<TargetPair> sourceNodes = new HashSet<>( markedEdges.size() );
Set<TargetPair> targetNodes = new HashSet<>( markedEdges.size() );
@@ -203,13 +210,13 @@ public class NodeDeleteListenerImpl implements NodeDeleteListener {
//we use the version specified on the delete purposefully. If these edges are re-written
//at a greater time we want them to exit
- batch.mergeShallow( storageSerialization.deleteEdge( scope, edge, eventTimestamp ) );
+ batch.addAll( storageSerialization.deleteEdge( scope, edge, eventTimestamp ) );
sourceNodes.add( new TargetPair( edge.getSourceNode(), edge.getType() ) );
targetNodes.add( new TargetPair( edge.getTargetNode(), edge.getType() ) );
}
- HystrixCassandra.async( batch );
+ session.execute( batch );
//now delete meta data
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/94a92c39/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeMetadataSerialization.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeMetadataSerialization.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeMetadataSerialization.java
index 9e0beb5..d98c129 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeMetadataSerialization.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeMetadataSerialization.java
@@ -20,8 +20,10 @@
package org.apache.usergrid.persistence.graph.serialization;
+import java.util.Collection;
import java.util.Iterator;
+import org.apache.usergrid.persistence.core.migration.DatastaxMigration;
import org.apache.usergrid.persistence.core.migration.Migration;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.graph.Edge;
@@ -29,19 +31,20 @@ import org.apache.usergrid.persistence.graph.SearchEdgeType;
import org.apache.usergrid.persistence.graph.SearchIdType;
import org.apache.usergrid.persistence.model.entity.Id;
+import com.datastax.driver.core.Statement;
import com.netflix.astyanax.MutationBatch;
/**
* Simple interface for serializing an edge meta data
*/
-public interface EdgeMetadataSerialization extends Migration {
+public interface EdgeMetadataSerialization extends DatastaxMigration {
/**
* EdgeWrite both the source--->Target edge and the target <----- source edge into the mutation
*/
- MutationBatch writeEdge( ApplicationScope scope, Edge edge );
+ Collection<? extends Statement> writeEdge( ApplicationScope scope, Edge edge );
/**
* Remove all meta data from the source to the target type. The caller must ensure that this is the last edge with
@@ -52,7 +55,7 @@ public interface EdgeMetadataSerialization extends Migration {
*
* @return a mutation batch with the delete operations
*/
- MutationBatch removeEdgeTypeFromSource( ApplicationScope scope, Edge edge );
+ Collection<? extends Statement> removeEdgeTypeFromSource( ApplicationScope scope, Edge edge );
/**
@@ -65,7 +68,7 @@ public interface EdgeMetadataSerialization extends Migration {
*
* @return A mutation batch to use on issuing the delelete
*/
- MutationBatch removeEdgeTypeFromSource( ApplicationScope scope, Id sourceNode, String type, long timestamp );
+ Collection<? extends Statement> removeEdgeTypeFromSource( ApplicationScope scope, Id sourceNode, String type, long timestamp );
/**
* Remove all meta data from the source to the target type. The caller must ensure that this is the last edge with
@@ -76,7 +79,7 @@ public interface EdgeMetadataSerialization extends Migration {
*
* @return a mutation batch with the delete operations
*/
- MutationBatch removeIdTypeFromSource( ApplicationScope scope, Edge edge );
+ Collection<? extends Statement> removeIdTypeFromSource( ApplicationScope scope, Edge edge );
/**
@@ -91,7 +94,7 @@ public interface EdgeMetadataSerialization extends Migration {
*
* @return a mutation batch with the delete operations
*/
- MutationBatch removeIdTypeFromSource( ApplicationScope scope, Id sourceNode, String type, String idType,
+ Collection<? extends Statement> removeIdTypeFromSource( ApplicationScope scope, Id sourceNode, String type, String idType,
long timestamp );
/**
@@ -103,7 +106,7 @@ public interface EdgeMetadataSerialization extends Migration {
*
* @return a mutation batch with the delete operations
*/
- MutationBatch removeEdgeTypeToTarget( ApplicationScope scope, Edge edge );
+ Collection<? extends Statement> removeEdgeTypeToTarget( ApplicationScope scope, Edge edge );
/**
@@ -117,7 +120,7 @@ public interface EdgeMetadataSerialization extends Migration {
*
* @return A mutation batch to use on issuing the delelete
*/
- MutationBatch removeEdgeTypeToTarget( ApplicationScope scope, Id targetNode, String type, long timestamp );
+ Collection<? extends Statement> removeEdgeTypeToTarget( ApplicationScope scope, Id targetNode, String type, long timestamp );
/**
@@ -129,7 +132,7 @@ public interface EdgeMetadataSerialization extends Migration {
*
* @return a mutation batch with the delete operations
*/
- MutationBatch removeIdTypeToTarget( ApplicationScope scope, Edge edge );
+ Collection<? extends Statement> removeIdTypeToTarget( ApplicationScope scope, Edge edge );
/**
@@ -144,7 +147,7 @@ public interface EdgeMetadataSerialization extends Migration {
*
* @return a mutation batch with the delete operations
*/
- MutationBatch removeIdTypeToTarget( ApplicationScope scope, Id targetNode, String type, String idType,
+ Collection<? extends Statement> removeIdTypeToTarget( ApplicationScope scope, Id targetNode, String type, String idType,
long timestamp );
/**
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/94a92c39/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerialization.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerialization.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerialization.java
index 8d28304..86516a6 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerialization.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgeSerialization.java
@@ -20,6 +20,7 @@
package org.apache.usergrid.persistence.graph.serialization;
+import java.util.Collection;
import java.util.Iterator;
import java.util.UUID;
@@ -29,6 +30,7 @@ import org.apache.usergrid.persistence.graph.SearchByEdge;
import org.apache.usergrid.persistence.graph.SearchByEdgeType;
import org.apache.usergrid.persistence.graph.SearchByIdType;
+import com.datastax.driver.core.Statement;
import com.netflix.astyanax.MutationBatch;
@@ -44,7 +46,7 @@ public interface EdgeSerialization {
* @param scope The org scope of the graph
* @param edge The edge to write
*/
- MutationBatch writeEdge( ApplicationScope scope, MarkedEdge edge, UUID timestamp );
+ Collection<? extends Statement> writeEdge( ApplicationScope scope, MarkedEdge edge, UUID timestamp );
/**
* EdgeWrite both the source -->target edge and the target<--- source edge into the mutation
@@ -52,7 +54,7 @@ public interface EdgeSerialization {
* @param scope The org scope of the graph
* @param edge The edge to write
*/
- MutationBatch deleteEdge( ApplicationScope scope, MarkedEdge edge, UUID timestamp );
+ Collection<? extends Statement> deleteEdge( ApplicationScope scope, MarkedEdge edge, UUID timestamp );
/**
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/94a92c39/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/NodeSerialization.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/NodeSerialization.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/NodeSerialization.java
index 77e53c8..ae49dee 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/NodeSerialization.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/NodeSerialization.java
@@ -23,11 +23,13 @@ package org.apache.usergrid.persistence.graph.serialization;
import java.util.Collection;
import java.util.Map;
+import org.apache.usergrid.persistence.core.migration.DatastaxMigration;
import org.apache.usergrid.persistence.core.migration.Migration;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.graph.Edge;
import org.apache.usergrid.persistence.model.entity.Id;
+import com.datastax.driver.core.Statement;
import com.google.common.base.Optional;
import com.netflix.astyanax.MutationBatch;
@@ -35,7 +37,7 @@ import com.netflix.astyanax.MutationBatch;
/**
* Simple interface for serializing node information for mark/sweep
*/
-public interface NodeSerialization extends Migration {
+public interface NodeSerialization extends DatastaxMigration {
/**
@@ -45,7 +47,7 @@ public interface NodeSerialization extends Migration {
* @param node The node to mark
* @param timestamp The timestamp to mark for deletion. Anything <= this time is considered deleted from the graph
*/
- MutationBatch mark( ApplicationScope scope, Id node, long timestamp );
+ Collection<? extends Statement> mark( ApplicationScope scope, Id node, long timestamp );
/**
@@ -54,7 +56,7 @@ public interface NodeSerialization extends Migration {
* @param node
* @return
*/
- MutationBatch delete( ApplicationScope scope, Id node, long timestamp );
+ Collection<? extends Statement> delete( ApplicationScope scope, Id node, long timestamp );
/**
* Get the maximum timestamp of a node marked for deletion. If the node has no mark
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/94a92c39/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeMetadataSerializationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeMetadataSerializationImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeMetadataSerializationImpl.java
index 924eddb..930e12d 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeMetadataSerializationImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeMetadataSerializationImpl.java
@@ -22,21 +22,12 @@ package org.apache.usergrid.persistence.graph.serialization.impl;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.Iterator;
+import java.util.UUID;
-import org.apache.cassandra.db.marshal.BytesType;
-import org.apache.cassandra.db.marshal.UTF8Type;
-
-import org.apache.usergrid.persistence.core.astyanax.CassandraConfig;
-import org.apache.usergrid.persistence.core.astyanax.ColumnNameIterator;
-import org.apache.usergrid.persistence.core.astyanax.CompositeFieldSerializer;
-import org.apache.usergrid.persistence.core.astyanax.IdRowCompositeSerializer;
-import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamily;
-import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamilyDefinition;
-import org.apache.usergrid.persistence.core.astyanax.OrganizationScopedRowKeySerializer;
-import org.apache.usergrid.persistence.core.astyanax.ScopedRowKey;
-import org.apache.usergrid.persistence.core.astyanax.StringColumnParser;
-import org.apache.usergrid.persistence.core.migration.Migration;
+import org.apache.usergrid.persistence.core.javadriver.RowIterator;
+import org.apache.usergrid.persistence.core.javadriver.RowParser;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.core.util.ValidationUtils;
import org.apache.usergrid.persistence.graph.Edge;
@@ -47,396 +38,464 @@ import org.apache.usergrid.persistence.graph.serialization.EdgeMetadataSerializa
import org.apache.usergrid.persistence.graph.serialization.util.GraphValidation;
import org.apache.usergrid.persistence.model.entity.Id;
+import com.datastax.driver.core.BoundStatement;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.Statement;
+import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.inject.Inject;
import com.google.inject.Singleton;
-import com.netflix.astyanax.Keyspace;
-import com.netflix.astyanax.MutationBatch;
-import com.netflix.astyanax.model.CompositeBuilder;
-import com.netflix.astyanax.model.CompositeParser;
-import com.netflix.astyanax.query.RowQuery;
-import com.netflix.astyanax.serializers.StringSerializer;
-import com.netflix.astyanax.util.RangeBuilder;
/**
* Class to perform all edge metadata I/O
*/
@Singleton
-public class EdgeMetadataSerializationImpl implements EdgeMetadataSerialization, Migration {
-
- private static final byte[] HOLDER = new byte[] { 0 };
-
-
- //row key serializers
- private static final IdRowCompositeSerializer ID_SER = IdRowCompositeSerializer.get();
- private static final OrganizationScopedRowKeySerializer<Id> ROW_KEY_SER =
- new OrganizationScopedRowKeySerializer<Id>( ID_SER );
-
- private static final StringSerializer STRING_SERIALIZER = StringSerializer.get();
-
- private static final EdgeTypeRowCompositeSerializer EDGE_SER = new EdgeTypeRowCompositeSerializer();
- private static final OrganizationScopedRowKeySerializer<EdgeIdTypeKey> EDGE_TYPE_ROW_KEY =
- new OrganizationScopedRowKeySerializer<EdgeIdTypeKey>( EDGE_SER );
-
- private static final StringColumnParser PARSER = StringColumnParser.get();
-
+public class EdgeMetadataSerializationImpl implements EdgeMetadataSerialization {
/**
* CFs where the row key contains the source node id
*/
- private static final MultiTennantColumnFamily<ApplicationScope, Id, String> CF_SOURCE_EDGE_TYPES =
- new MultiTennantColumnFamily<ApplicationScope, Id, String>( "Graph_Source_Edge_Types", ROW_KEY_SER,
- STRING_SERIALIZER );
+ private static final String CF_SOURCE_EDGE_TYPES = "Graph_Source_Edge_Types";
+
+ private static final String CF_SOURCE_EDGE_TYPES_CREATE = "CREATE TABLE IF NOT EXISTS " + CF_SOURCE_EDGE_TYPES
+ + "( scopeId uuid, scopeType varchar, sourceNodeId uuid, sourceNodeType varchar, " +
+ "edgeType varchar, PRIMARY KEY ((scopeId, scopeType, sourceNodeId, sourceNodeType), edgeType) )";
//all target id types for source edge type
- private static final MultiTennantColumnFamily<ApplicationScope, EdgeIdTypeKey, String> CF_SOURCE_EDGE_ID_TYPES =
- new MultiTennantColumnFamily<ApplicationScope, EdgeIdTypeKey, String>( "Graph_Source_Edge_Id_Types",
- EDGE_TYPE_ROW_KEY, STRING_SERIALIZER );
+ private static final String CF_SOURCE_EDGE_ID_TYPES = "Graph_Source_Edge_Id_Types";
+
+
+ private static final String CF_SOURCE_EDGE_ID_TYPES_CREATE = "CREATE TABLE IF NOT EXISTS " + CF_SOURCE_EDGE_ID_TYPES
+ + "( scopeId uuid, scopeType varchar, sourceNodeId uuid, sourceNodeType varchar, edgeType varchar, " +
+ "targetNodeType varchar, PRIMARY KEY ((scopeId, scopeType, sourceNodeId, sourceNodeType, " +
+ "edgeType), targetNodeType) )";
+
/**
* CFs where the row key is the target node id
*/
- private static final MultiTennantColumnFamily<ApplicationScope, Id, String> CF_TARGET_EDGE_TYPES =
- new MultiTennantColumnFamily<ApplicationScope, Id, String>( "Graph_Target_Edge_Types", ROW_KEY_SER,
- STRING_SERIALIZER );
+ private static final String CF_TARGET_EDGE_TYPES = "Graph_Target_Edge_Types";
+
+ private static final String CF_TARGET_EDGE_TYPES_CREATE = "CREATE TABLE IF NOT EXISTS " + CF_TARGET_EDGE_TYPES
+ + "( scopeId uuid, scopeType varchar, targetNodeId uuid, targetNodeType varchar, edgeType varchar, " +
+ "PRIMARY KEY ((scopeId, scopeType, "
+ +
+ "targetNodeId, targetNodeType), edgeType ))";
//all source id types for target edge type
- private static final MultiTennantColumnFamily<ApplicationScope, EdgeIdTypeKey, String> CF_TARGET_EDGE_ID_TYPES =
- new MultiTennantColumnFamily<ApplicationScope, EdgeIdTypeKey, String>( "Graph_Target_Edge_Id_Types",
- EDGE_TYPE_ROW_KEY, STRING_SERIALIZER );
+ private static final String CF_TARGET_EDGE_ID_TYPES = "Graph_Target_Edge_Id_Types";
+ private static final String CF_TARGET_EDGE_ID_TYPES_CREATE = "CREATE TABLE IF NOT EXISTS " + CF_TARGET_EDGE_ID_TYPES
+ + " ( scopeId uuid, scopeType varchar, targetNodeId uuid, targetNodeType varchar, edgeType varchar, "
+ + "sourceNodeType varchar, PRIMARY KEY ((scopeId, scopeType, targetNodeId, targetNodeType, " +
+ "edgeType), sourceNodeType ) )";
- protected final Keyspace keyspace;
- private final CassandraConfig cassandraConfig;
+
+ protected final Session session;
private final GraphFig graphFig;
+ private PreparedStatement sourceEdgeTypesWrite;
+ private PreparedStatement sourceEdgeTypesDelete;
+ private PreparedStatement sourceEdgeTypesSelect;
- @Inject
- public EdgeMetadataSerializationImpl( final Keyspace keyspace, final CassandraConfig cassandraConfig,
- final GraphFig graphFig ) {
- Preconditions.checkNotNull( "cassandraConfig is required", cassandraConfig );
+ private PreparedStatement targetEdgeTypesWrite;
+ private PreparedStatement targetEdgeTypesDelete;
+ private PreparedStatement targetEdgeTypesSelect;
+
+
+
+ private PreparedStatement sourceEdgeTargetTypesWrite;
+ private PreparedStatement sourceEdgeTargetTypesSelect;
+ private PreparedStatement sourceEdgeTargetTypesDelete;
+
+
+ private PreparedStatement targetEdgeSourceTypesWrite;
+ private PreparedStatement targetEdgeSourceTypesDelete;
+ private PreparedStatement targetEdgeSourceTypesSelect;
+
+
+ /**
+ * TODO. The version logic in this doesn't work correctly if the highest edge is deleted. We should remove this,
+ * and just use a supplied timestamp. This way the caller can determine what max timestamp to use during a delete
+ */
+
+ /*
+ *
+ * @param session
+ * @param cassandraConfig
+ * @param graphFig
+ */
+ @Inject
+ public EdgeMetadataSerializationImpl( final Session session, final GraphFig graphFig ) {
Preconditions.checkNotNull( "consistencyFig is required", graphFig );
- Preconditions.checkNotNull( "keyspace is required", keyspace );
+ Preconditions.checkNotNull( "session is required", session );
- this.keyspace = keyspace;
- this.cassandraConfig = cassandraConfig;
+ this.session = session;
this.graphFig = graphFig;
}
@Override
- public MutationBatch writeEdge( final ApplicationScope scope, final Edge edge ) {
+ public Collection<? extends Statement> writeEdge( final ApplicationScope scope, final Edge edge ) {
ValidationUtils.validateApplicationScope( scope );
GraphValidation.validateEdge( edge );
- final Id source = edge.getSourceNode();
- final Id target = edge.getTargetNode();
- final String edgeType = edge.getType();
- final long timestamp = edge.getTimestamp();
-
- final MutationBatch batch = keyspace.prepareMutationBatch().withConsistencyLevel( cassandraConfig.getWriteCL() )
- .withTimestamp( timestamp );
-
-
- //add source->target edge type to meta data
- final ScopedRowKey<ApplicationScope, Id> sourceKey = new ScopedRowKey<ApplicationScope, Id>( scope, source );
-
- batch.withRow( CF_SOURCE_EDGE_TYPES, sourceKey ).putColumn( edgeType, HOLDER );
-
+ final Id applicationId = scope.getApplication();
+ final UUID applicationUuid = applicationId.getUuid();
+ final String applicationType = applicationId.getType();
- //write source->target edge type and id type to meta data
- EdgeIdTypeKey tk = new EdgeIdTypeKey( source, edgeType );
- final ScopedRowKey<ApplicationScope, EdgeIdTypeKey> sourceTypeKey =
- new ScopedRowKey<ApplicationScope, EdgeIdTypeKey>( scope, tk );
+ final Id source = edge.getSourceNode();
+ final UUID sourceUuid = source.getUuid();
+ final String sourceType = source.getType();
- batch.withRow( CF_SOURCE_EDGE_ID_TYPES, sourceTypeKey ).putColumn( target.getType(), HOLDER );
+ final Id target = edge.getTargetNode();
+ final UUID targetUuid = target.getUuid();
+ final String targetType = target.getType();
- //write target<--source edge type meta data
- final ScopedRowKey<ApplicationScope, Id> targetKey = new ScopedRowKey<ApplicationScope, Id>( scope, target );
+ final String edgeType = edge.getType();
+ final long timestamp = edge.getTimestamp();
- batch.withRow( CF_TARGET_EDGE_TYPES, targetKey ).putColumn( edgeType, HOLDER );
+ final BoundStatement sourceStatement = this.sourceEdgeTypesWrite
+ .bind( applicationUuid, applicationType, sourceUuid, sourceType, edgeType, timestamp );
- //write target<--source edge type and id type to meta data
- final ScopedRowKey<ApplicationScope, EdgeIdTypeKey> targetTypeKey =
- new ScopedRowKey<ApplicationScope, EdgeIdTypeKey>( scope, new EdgeIdTypeKey( target, edgeType ) );
+ final BoundStatement sourceTargetTypeStatement = this.sourceEdgeTargetTypesWrite
+ .bind( applicationUuid, applicationType, sourceUuid, sourceType, edgeType, targetType, timestamp );
+ final BoundStatement targetStatement = this.targetEdgeTypesWrite
+ .bind( applicationUuid, applicationType, targetUuid, targetType, edgeType, timestamp );
- batch.withRow( CF_TARGET_EDGE_ID_TYPES, targetTypeKey ).putColumn( source.getType(), HOLDER );
+ final BoundStatement targetSourceTypeStatement = this.targetEdgeSourceTypesWrite
+ .bind( applicationUuid, applicationType, targetUuid, targetType, edgeType, sourceType, timestamp );
- return batch;
+ return Arrays.asList( sourceStatement, sourceTargetTypeStatement, targetStatement, targetSourceTypeStatement );
}
@Override
- public MutationBatch removeEdgeTypeFromSource( final ApplicationScope scope, final Edge edge ) {
+ public Collection<? extends Statement> removeEdgeTypeFromSource( final ApplicationScope scope, final Edge edge ) {
return removeEdgeTypeFromSource( scope, edge.getSourceNode(), edge.getType(), edge.getTimestamp() );
}
@Override
- public MutationBatch removeEdgeTypeFromSource( final ApplicationScope scope, final Id sourceNode, final String type,
- final long version ) {
- return removeEdgeType( scope, sourceNode, type, version, CF_SOURCE_EDGE_TYPES );
+ public Collection<? extends Statement> removeEdgeTypeFromSource( final ApplicationScope scope, final Id sourceNode,
+ final String type, final long version ) {
+
+ final Id applicationId = scope.getApplication();
+ final UUID applicationUuid = applicationId.getUuid();
+ final String applicationType = applicationId.getType();
+ final UUID sourceId = sourceNode.getUuid();
+ final String sourceType = sourceNode.getType();
+
+
+ final BoundStatement removeStatement = this.sourceEdgeTypesDelete
+ .bind( version, applicationUuid, applicationType, sourceId, sourceType, type );
+
+ return Collections.singleton( removeStatement );
}
@Override
- public MutationBatch removeIdTypeFromSource( final ApplicationScope scope, final Edge edge ) {
+ public Collection<? extends Statement> removeIdTypeFromSource( final ApplicationScope scope, final Edge edge ) {
return removeIdTypeFromSource( scope, edge.getSourceNode(), edge.getType(), edge.getTargetNode().getType(),
edge.getTimestamp() );
}
@Override
- public MutationBatch removeIdTypeFromSource( final ApplicationScope scope, final Id sourceNode, final String type,
- final String idType, final long version ) {
- return removeIdType( scope, sourceNode, idType, type, version, CF_SOURCE_EDGE_ID_TYPES );
+ public Collection<? extends Statement> removeIdTypeFromSource( final ApplicationScope scope, final Id sourceNode,
+ final String type, final String idType,
+ final long version ) {
+ final Id applicationId = scope.getApplication();
+ final UUID applicationUuid = applicationId.getUuid();
+ final String applicationType = applicationId.getType();
+
+ final UUID sourceId = sourceNode.getUuid();
+ final String sourceType = sourceNode.getType();
+
+
+ final BoundStatement removeStatement = this.sourceEdgeTargetTypesDelete
+ .bind( version, applicationUuid, applicationType, sourceId, sourceType, type, idType );
+
+ return Collections.singleton( removeStatement );
}
@Override
- public MutationBatch removeEdgeTypeToTarget( final ApplicationScope scope, final Edge edge ) {
+ public Collection<? extends Statement> removeEdgeTypeToTarget( final ApplicationScope scope, final Edge edge ) {
+
return removeEdgeTypeToTarget( scope, edge.getTargetNode(), edge.getType(), edge.getTimestamp() );
}
@Override
- public MutationBatch removeEdgeTypeToTarget( final ApplicationScope scope, final Id targetNode, final String type,
- final long version ) {
- return removeEdgeType( scope, targetNode, type, version, CF_TARGET_EDGE_TYPES );
+ public Collection<? extends Statement> removeEdgeTypeToTarget( final ApplicationScope scope, final Id targetNode,
+ final String type, final long version ) {
+ final Id applicationId = scope.getApplication();
+ final UUID applicationUuid = applicationId.getUuid();
+ final String applicationType = applicationId.getType();
+ final UUID targetUuid = targetNode.getUuid();
+ final String targetType = targetNode.getType();
+
+
+ final BoundStatement removeStatement = this.targetEdgeTypesDelete
+ .bind( version, applicationUuid, applicationType, targetUuid, type, targetType );
+
+ return Collections.singleton( removeStatement );
}
@Override
- public MutationBatch removeIdTypeToTarget( final ApplicationScope scope, final Edge edge ) {
+ public Collection<? extends Statement> removeIdTypeToTarget( final ApplicationScope scope, final Edge edge ) {
return removeIdTypeToTarget( scope, edge.getTargetNode(), edge.getType(), edge.getSourceNode().getType(),
edge.getTimestamp() );
}
@Override
- public MutationBatch removeIdTypeToTarget( final ApplicationScope scope, final Id targetNode, final String type,
- final String idType, final long version ) {
- return removeIdType( scope, targetNode, idType, type, version, CF_TARGET_EDGE_ID_TYPES );
- }
-
+ public Collection<? extends Statement> removeIdTypeToTarget( final ApplicationScope scope, final Id targetNode,
+ final String type, final String idType,
+ final long version ) {
- /**
- * Remove the edge
- *
- * @param scope The scope
- * @param rowKeyId The id to use in the row key
- * @param edgeType The edge type
- * @param version The version of the edge
- * @param cf The column family
- */
- private MutationBatch removeEdgeType( final ApplicationScope scope, final Id rowKeyId, final String edgeType,
- final long version,
- final MultiTennantColumnFamily<ApplicationScope, Id, String> cf ) {
+ final Id applicationId = scope.getApplication();
+ final UUID applicationUuid = applicationId.getUuid();
+ final String applicationType = applicationId.getType();
- //write target<--source edge type meta data
- final ScopedRowKey<ApplicationScope, Id> rowKey = new ScopedRowKey<ApplicationScope, Id>( scope, rowKeyId );
+ final UUID targetUuid = targetNode.getUuid();
+ final String targetType = targetNode.getType();
- final MutationBatch batch = keyspace.prepareMutationBatch().withTimestamp( version );
- batch.withRow( cf, rowKey ).deleteColumn( edgeType );
+ final BoundStatement removeStatement = this.targetEdgeSourceTypesDelete
+ .bind( version, applicationUuid, applicationType, targetUuid, targetType, type, idType );
- return batch;
+ return Collections.singleton( removeStatement );
}
- /**
- * Remove the id type
- *
- * @param scope The scope to use
- * @param rowId The id to use in the row key
- * @param idType The id type to use in the column
- * @param edgeType The edge type to use in the column
- * @param version The version to use on the column
- * @param cf The column family to use
- *
- * @return A populated mutation with the remove operations
- */
- private MutationBatch removeIdType( final ApplicationScope scope, final Id rowId, final String idType,
- final String edgeType, final long version,
- final MultiTennantColumnFamily<ApplicationScope, EdgeIdTypeKey, String> cf ) {
-
+ @Override
+ public Iterator<String> getEdgeTypesFromSource( final ApplicationScope scope, final SearchEdgeType search ) {
- final MutationBatch batch = keyspace.prepareMutationBatch().withTimestamp( version );
+ final Id applicationId = scope.getApplication();
+ final UUID applicationUuid = applicationId.getUuid();
+ final String applicationType = applicationId.getType();
+ final Id searchNode = search.getNode();
+ final UUID sourceId = searchNode.getUuid();
+ final String sourceType = searchNode.getType();
+ final String start = getStart( search );
+ final int limit = graphFig.getScanPageSize();
- //write target<--source edge type and id type to meta data
- final ScopedRowKey<ApplicationScope, EdgeIdTypeKey> rowKey =
- new ScopedRowKey<ApplicationScope, EdgeIdTypeKey>( scope, new EdgeIdTypeKey( rowId, edgeType ) );
+ final BoundStatement statement =
+ this.sourceEdgeTypesSelect.bind( applicationUuid, applicationType, sourceId, sourceType, start );
- batch.withRow( cf, rowKey ).deleteColumn( idType );
+ statement.setFetchSize( limit );
- return batch;
+ return new RowIterator<>( session, statement, EDGE_PARSER );
}
@Override
- public Iterator<String> getEdgeTypesFromSource( final ApplicationScope scope, final SearchEdgeType search ) {
- return getEdgeTypes( scope, search, CF_SOURCE_EDGE_TYPES );
- }
+ public Iterator<String> getIdTypesFromSource( final ApplicationScope scope, final SearchIdType search ) {
+ final Id applicationId = scope.getApplication();
+ final UUID applicationUuid = applicationId.getUuid();
+ final String applicationType = applicationId.getType();
+ final Id searchNode = search.getNode();
+ final UUID sourceId = searchNode.getUuid();
+ final String sourceType = searchNode.getType();
+ final String edgeType = search.getEdgeType();
+ final String start = getStart( search );
- @Override
- public Iterator<String> getIdTypesFromSource( final ApplicationScope scope, final SearchIdType search ) {
- return getIdTypes( scope, search, CF_SOURCE_EDGE_ID_TYPES );
+ final int limit = graphFig.getScanPageSize();
+
+
+ final BoundStatement statement = this.sourceEdgeTargetTypesSelect
+ .bind( applicationUuid, applicationType, sourceId, sourceType, edgeType, start );
+
+ statement.setFetchSize( limit );
+
+ return new RowIterator<>( session, statement, TARGET_ID_TYPE_PARSER );
}
@Override
public Iterator<String> getEdgeTypesToTarget( final ApplicationScope scope, final SearchEdgeType search ) {
- return getEdgeTypes( scope, search, CF_TARGET_EDGE_TYPES );
- }
+ final Id applicationId = scope.getApplication();
+ final UUID applicationUuid = applicationId.getUuid();
+ final String applicationType = applicationId.getType();
+ final Id searchNode = search.getNode();
+ final UUID targetId = searchNode.getUuid();
+ final String targetType = searchNode.getType();
+ final String start = getStart( search );
+ final int limit = graphFig.getScanPageSize();
- /**
- * Get the edge types from the search criteria.
- *
- * @param scope The org scope
- * @param search The edge type search info
- * @param cf The column family to execute on
- */
- private Iterator<String> getEdgeTypes( final ApplicationScope scope, final SearchEdgeType search,
- final MultiTennantColumnFamily<ApplicationScope, Id, String> cf ) {
- ValidationUtils.validateApplicationScope( scope );
- GraphValidation.validateSearchEdgeType( search );
+ final BoundStatement statement =
+ this.targetEdgeTypesSelect.bind( applicationUuid, applicationType, targetId, targetType, start );
+
+ statement.setFetchSize( limit );
+
+ return new RowIterator<>( session, statement, EDGE_PARSER );
+ }
- final ScopedRowKey<ApplicationScope, Id> sourceKey = new ScopedRowKey<>( scope, search.getNode() );
+ @Override
+ public Iterator<String> getIdTypesToTarget( final ApplicationScope scope, final SearchIdType search ) {
+ final Id applicationId = scope.getApplication();
+ final UUID applicationUuid = applicationId.getUuid();
+ final String applicationType = applicationId.getType();
+ final Id searchNode = search.getNode();
+ final UUID targetId = searchNode.getUuid();
+ final String targetType = searchNode.getType();
+ final String edgeType = search.getEdgeType();
- //resume from the last if specified. Also set the range
+ final String start = getStart( search );
+ final int limit = graphFig.getScanPageSize();
- final RangeBuilder rangeBuilder = createRange( search );
- RowQuery<ScopedRowKey<ApplicationScope, Id>, String> query =
- keyspace.prepareQuery( cf ).getKey( sourceKey ).autoPaginate( true )
- .withColumnRange( rangeBuilder.build() );
+ final BoundStatement statement = this.targetEdgeSourceTypesSelect
+ .bind( applicationUuid, applicationType, targetId, targetType, edgeType, start );
- return new ColumnNameIterator<>( query, PARSER, search.getLast().isPresent() );
+ statement.setFetchSize( limit );
+
+ return new RowIterator<>( session, statement, SOURCE_ID_TYPE_PARSER );
}
@Override
- public Iterator<String> getIdTypesToTarget( final ApplicationScope scope, final SearchIdType search ) {
- return getIdTypes( scope, search, CF_TARGET_EDGE_ID_TYPES );
+ public Collection<String> getColumnFamilies() {
+ return Arrays.asList( CF_SOURCE_EDGE_TYPES_CREATE, CF_SOURCE_EDGE_ID_TYPES_CREATE, CF_TARGET_EDGE_TYPES_CREATE,
+ CF_TARGET_EDGE_ID_TYPES_CREATE );
}
- /**
- * Get the id types from the specified column family
- *
- * @param scope The organization scope to use
- * @param search The search criteria
- * @param cf The column family to search
- */
- public Iterator<String> getIdTypes( final ApplicationScope scope, final SearchIdType search,
- final MultiTennantColumnFamily<ApplicationScope, EdgeIdTypeKey, String> cf ) {
- ValidationUtils.validateApplicationScope( scope );
- GraphValidation.validateSearchEdgeIdType( search );
-
+ @Override
+ public void prepareStatements() {
+ /**
+ * Source with edge types
+ */
+ this.sourceEdgeTypesWrite = session.prepare( "INSERT INTO " + CF_SOURCE_EDGE_TYPES
+ + " (scopeId, scopeType, sourceNodeId, sourceNodeType, edgeType) VALUES (?, ?, ?, ?, " +
+ "?) USING TIMESTAMP ?" );
- final ScopedRowKey<ApplicationScope, EdgeIdTypeKey> sourceTypeKey =
- new ScopedRowKey<>( scope, new EdgeIdTypeKey( search.getNode(), search.getEdgeType() ) );
+ this.sourceEdgeTypesDelete = session.prepare( "DELETE FROM " + CF_SOURCE_EDGE_TYPES
+ + " USING TIMESTAMP ? WHERE scopeId = ? AND scopeType = ? AND sourceNodeId = ? AND sourceNodeType = ? "
+ +
+ "AND edgeType = ? " );
- final RangeBuilder rangeBuilder = createRange( search );
+ this.sourceEdgeTypesSelect = session.prepare( "SELECT edgeType FROM " + CF_SOURCE_EDGE_TYPES
+ + " WHERE scopeId = ? AND scopeType = ? AND sourceNodeId = ? AND sourceNodeType = ? AND edgeType > ?" );
+ /**
+ * Source with target types
+ */
- RowQuery<ScopedRowKey<ApplicationScope, EdgeIdTypeKey>, String> query =
- keyspace.prepareQuery( cf ).getKey( sourceTypeKey ).autoPaginate( true )
- .withColumnRange( rangeBuilder.build() );
+ this.sourceEdgeTargetTypesWrite = session.prepare( "INSERT INTO " + CF_SOURCE_EDGE_ID_TYPES
+ + " (scopeId, scopeType, sourceNodeId, sourceNodeType, edgeType, targetNodeType) VALUES (?, ?, ?, ?, " +
+ "?, ?) USING TIMESTAMP ?" );
+ this.sourceEdgeTargetTypesDelete = session.prepare( "DELETE FROM " + CF_SOURCE_EDGE_ID_TYPES
+ + " USING TIMESTAMP ? WHERE scopeId = ? AND scopeType = ? AND sourceNodeId = ? AND sourceNodeType = ?" +
+ " AND "
+ +
+ "edgeType = ? AND targetNodeType = ? " );
- return new ColumnNameIterator<>( query, PARSER, search.getLast().isPresent() );
- }
+ this.sourceEdgeTargetTypesSelect = session.prepare( "SELECT targetNodeType FROM " + CF_SOURCE_EDGE_ID_TYPES
+ + " WHERE scopeId = ? AND scopeType = ? AND sourceNodeId = ? AND sourceNodeType = ? AND " +
+ "edgeType = ? AND targetNodeType > ?" );
- @Override
- public Collection<MultiTennantColumnFamilyDefinition> getColumnFamilies() {
- return Arrays.asList( graphCf( CF_SOURCE_EDGE_TYPES ), graphCf( CF_TARGET_EDGE_TYPES ),
- graphCf( CF_SOURCE_EDGE_ID_TYPES ), graphCf( CF_TARGET_EDGE_ID_TYPES ) );
- }
+ /**
+ * Target with edge types
+ */
+ this.targetEdgeTypesWrite = session.prepare( "INSERT INTO " + CF_TARGET_EDGE_TYPES
+ + " (scopeId, scopeType, targetNodeId, targetNodeType, edgeType) VALUES (?, ?, ?, ?, " +
+ "?) USING TIMESTAMP ?" );
- /**
- * Helper to generate an edge definition by the type
- */
- private MultiTennantColumnFamilyDefinition graphCf( MultiTennantColumnFamily cf ) {
- return new MultiTennantColumnFamilyDefinition( cf, BytesType.class.getSimpleName(),
- UTF8Type.class.getSimpleName(), BytesType.class.getSimpleName(),
- MultiTennantColumnFamilyDefinition.CacheOption.KEYS );
- }
+ this.targetEdgeTypesDelete = session.prepare( "DELETE FROM " + CF_TARGET_EDGE_TYPES
+ + " USING TIMESTAMP ? WHERE scopeId = ? AND scopeType = ? AND targetNodeId = ? AND edgeType = ? AND targetNodeType = ?" );
- /**
- * Inner class to serialize and edgeIdTypeKey
- */
- private static class EdgeTypeRowCompositeSerializer implements CompositeFieldSerializer<EdgeIdTypeKey> {
+ this.targetEdgeTypesSelect = session.prepare( "SELECT edgeType FROM " + CF_TARGET_EDGE_TYPES
+ + " WHERE scopeId = ? AND scopeType = ? AND targetNodeId = ? AND targetNodeType = ? AND edgeType > ?" );
- private static final IdRowCompositeSerializer ID_SER = IdRowCompositeSerializer.get();
+ /**
+ * Target with source edge types
+ */
+ this.targetEdgeSourceTypesWrite = session.prepare( "INSERT INTO " + CF_TARGET_EDGE_ID_TYPES
+ + " (scopeId, scopeType, targetNodeId, targetNodeType, edgeType, sourceNodeType) VALUES (?, ?, ?, ?, " +
+ "?, ?) USING TIMESTAMP ?" );
- @Override
- public void toComposite( final CompositeBuilder builder, final EdgeIdTypeKey value ) {
- ID_SER.toComposite( builder, value.node );
+ this.targetEdgeSourceTypesDelete = session.prepare( "DELETE FROM " + CF_TARGET_EDGE_ID_TYPES
+ + " USING TIMESTAMP ? WHERE scopeId = ? AND scopeType = ? AND targetNodeId = ? AND targetNodeType = ? AND edgeType = ?" +
+ " AND sourceNodeType = ? " );
- builder.addString( value.edgeType );
- }
+ this.targetEdgeSourceTypesSelect = session.prepare( "SELECT sourceNodeType FROM " + CF_TARGET_EDGE_ID_TYPES
+ + " WHERE scopeId = ? AND scopeType = ? AND targetNodeId = ? AND targetNodeType = ? AND edgeType = ?" +
+ " AND sourceNodeType > ?" );
+ }
- @Override
- public EdgeIdTypeKey fromComposite( final CompositeParser composite ) {
- final Id id = ID_SER.fromComposite( composite );
- final String edgeType = composite.readString();
+ private String getStart( final SearchEdgeType search ) {
+ final Optional<String> last = search.getLast();
- return new EdgeIdTypeKey( id, edgeType );
+ if ( last.isPresent() ) {
+ return last.get();
}
- }
- private RangeBuilder createRange( final SearchEdgeType search ) {
- final RangeBuilder builder = new RangeBuilder().setLimit( graphFig.getScanPageSize() );
+ final Optional<String> prefix = search.prefix();
-
- //we have a last, it's where we need to start seeking from
- if ( search.getLast().isPresent() ) {
- builder.setStart( search.getLast().get() );
+ if ( prefix.isPresent() ) {
+ return prefix + "\uffff";
}
- //no last was set, but we have a prefix, set it
- else if ( search.prefix().isPresent() ) {
- builder.setStart( search.prefix().get() );
- }
+ return "";
+ }
- //we have a prefix, so make sure we only seek to prefix + max UTF value
- if ( search.prefix().isPresent() ) {
- builder.setEnd( search.prefix().get() + "\uffff" );
+ private final RowParser<String> EDGE_PARSER = new RowParser<String>() {
+ @Override
+ public String parseRow( final Row row ) {
+ return row.getString( "edgeType" );
}
+ };
- return builder;
- }
+ private final RowParser<String> SOURCE_ID_TYPE_PARSER = new RowParser<String>() {
+ @Override
+ public String parseRow( final Row row ) {
+ return row.getString( "sourceNodeType" );
+ }
+ };
+
+ private final RowParser<String> TARGET_ID_TYPE_PARSER = new RowParser<String>() {
+ @Override
+ public String parseRow( final Row row ) {
+ return row.getString( "targetNodeType" );
+ }
+ };
// private void setStart( final SearchEdgeType search, final RangeBuilder builder ) {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/94a92c39/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeSerializationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeSerializationImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeSerializationImpl.java
index 4c1ae79..3eab817 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeSerializationImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgeSerializationImpl.java
@@ -46,6 +46,7 @@ import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.Shard
import org.apache.usergrid.persistence.graph.serialization.util.GraphValidation;
import org.apache.usergrid.persistence.model.entity.Id;
+import com.datastax.driver.core.Statement;
import com.google.inject.Singleton;
import com.netflix.astyanax.Keyspace;
import com.netflix.astyanax.MutationBatch;
@@ -96,7 +97,7 @@ public class EdgeSerializationImpl implements EdgeSerialization {
@Override
- public MutationBatch writeEdge( final ApplicationScope scope, final MarkedEdge markedEdge, final UUID timestamp ) {
+ public Collection<? extends Statement> writeEdge( final ApplicationScope scope, final MarkedEdge markedEdge, final UUID timestamp ) {
ValidationUtils.validateApplicationScope( scope );
GraphValidation.validateEdge( markedEdge );
@@ -181,12 +182,12 @@ public class EdgeSerializationImpl implements EdgeSerialization {
edgeVersionsMeta, timestamp ) );
- return batch;
+ return null;
}
@Override
- public MutationBatch deleteEdge( final ApplicationScope scope, final MarkedEdge markedEdge, final UUID timestamp ) {
+ public Collection<? extends Statement> deleteEdge( final ApplicationScope scope, final MarkedEdge markedEdge, final UUID timestamp ) {
ValidationUtils.validateApplicationScope( scope );
GraphValidation.validateEdge( markedEdge );
ValidationUtils.verifyTimeUuid( timestamp, "timestamp" );
@@ -270,7 +271,7 @@ public class EdgeSerializationImpl implements EdgeSerialization {
edgeVersionsMeta, timestamp ) );
- return batch;
+ return null;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/94a92c39/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/NodeSerializationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/NodeSerializationImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/NodeSerializationImpl.java
index b14fad6..c6b3652 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/NodeSerializationImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/NodeSerializationImpl.java
@@ -24,42 +24,32 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import javax.annotation.PostConstruct;
import javax.inject.Inject;
-import org.apache.cassandra.db.marshal.BooleanType;
-import org.apache.cassandra.db.marshal.BytesType;
-
-import org.apache.usergrid.persistence.core.astyanax.CassandraConfig;
-import org.apache.usergrid.persistence.core.astyanax.IdRowCompositeSerializer;
-import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamily;
-import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamilyDefinition;
-import org.apache.usergrid.persistence.core.astyanax.OrganizationScopedRowKeySerializer;
-import org.apache.usergrid.persistence.core.astyanax.ScopedRowKey;
-import org.apache.usergrid.persistence.core.hystrix.HystrixCassandra;
-import org.apache.usergrid.persistence.core.migration.Migration;
+import org.apache.usergrid.persistence.core.javadriver.BatchStatementUtils;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.core.util.ValidationUtils;
import org.apache.usergrid.persistence.graph.Edge;
-import org.apache.usergrid.persistence.graph.exception.GraphRuntimeException;
import org.apache.usergrid.persistence.graph.serialization.NodeSerialization;
import org.apache.usergrid.persistence.graph.serialization.util.GraphValidation;
import org.apache.usergrid.persistence.model.entity.Id;
+import org.apache.usergrid.persistence.model.entity.SimpleId;
+import com.datastax.driver.core.BoundStatement;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.Statement;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.inject.Singleton;
-import com.netflix.astyanax.Keyspace;
-import com.netflix.astyanax.MutationBatch;
-import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
-import com.netflix.astyanax.connectionpool.exceptions.NotFoundException;
-import com.netflix.astyanax.model.Column;
-import com.netflix.astyanax.model.Row;
-import com.netflix.astyanax.model.Rows;
-import com.netflix.astyanax.query.ColumnFamilyQuery;
-import com.netflix.astyanax.serializers.BooleanSerializer;
/**
@@ -67,79 +57,99 @@ import com.netflix.astyanax.serializers.BooleanSerializer;
*
*/
@Singleton
-public class NodeSerializationImpl implements NodeSerialization, Migration {
+public class NodeSerializationImpl implements NodeSerialization {
- //Row key by node id.
- private static final IdRowCompositeSerializer ROW_SERIALIZER = IdRowCompositeSerializer.get();
- private static final BooleanSerializer BOOLEAN_SERIALIZER = BooleanSerializer.get();
/**
- * Column name is always just "true"
+ * CFs where the row key contains the source node id
*/
- private static final boolean COLUMN_NAME = true;
+ private static final String GRAPH_MARKED_EDGES = "Graph_Marked_Nodes";
-
- /**
- * Columns are always a byte, and the entire value is contained within a row key. This is intentional This allows
- * us to make heavy use of Cassandra's bloom filters, as well as key caches. Since most nodes will only exist for a
- * short amount of time in this CF, we'll most likely have them in the key cache, and we'll also bounce from the
- * BloomFilter on read. This means our performance will be no worse than checking a distributed cache in RAM for
- * the existence of a marked node.
- */
- private static final MultiTennantColumnFamily<ApplicationScope, Id, Boolean> GRAPH_DELETE =
- new MultiTennantColumnFamily<ApplicationScope, Id, Boolean>( "Graph_Marked_Nodes",
- new OrganizationScopedRowKeySerializer<Id>( ROW_SERIALIZER ), BOOLEAN_SERIALIZER );
+ private static final String GRAPH_MARKED_EDGES_CREATE = "CREATE TABLE IF NOT EXISTS " + GRAPH_MARKED_EDGES
+ + " ( scopeId uuid, scopeType varchar, nodeId uuid, nodeType varchar, timestamp bigint, PRIMARY KEY ((scopeId , scopeType, nodeId, nodeType)) ) WITH caching = 'all';";
- protected final Keyspace keyspace;
- protected final CassandraConfig fig;
+ protected final Session session;
+ private PreparedStatement insert;
+ private PreparedStatement delete;
+ private PreparedStatement select;
@Inject
- public NodeSerializationImpl( final Keyspace keyspace, final CassandraConfig fig ) {
- this.keyspace = keyspace;
- this.fig = fig;
+ public NodeSerializationImpl( final Session session ) {
+ this.session = session;
+ }
+
+
+ @PostConstruct
+ public void setup() {
+
}
@Override
- public Collection<MultiTennantColumnFamilyDefinition> getColumnFamilies() {
- return Collections.singleton(
- new MultiTennantColumnFamilyDefinition( GRAPH_DELETE, BytesType.class.getSimpleName(),
- BooleanType.class.getSimpleName(), BytesType.class.getSimpleName(),
- MultiTennantColumnFamilyDefinition.CacheOption.ALL ) );
+ public Collection<String> getColumnFamilies() {
+ return Collections.singleton( GRAPH_MARKED_EDGES_CREATE );
}
@Override
- public MutationBatch mark( final ApplicationScope scope, final Id node, final long timestamp ) {
+ public void prepareStatements() {
+ this.insert = this.session.prepare( "INSERT INTO " + GRAPH_MARKED_EDGES
+ + " (scopeId, scopeType , nodeId, nodeType, timestamp) VALUES (?, ?, ?, ?, " +
+ "? ) USING TIMESTAMP ?" );
+
+ this.delete = this.session.prepare( "DELETE FROM " + GRAPH_MARKED_EDGES
+ + " USING TIMESTAMP ? WHERE scopeId = ? AND scopeType = ? AND nodeId = ? AND nodeType = ? " );
+
+
+ this.select = this.session.prepare( "SELECT timestamp FROM " + GRAPH_MARKED_EDGES
+ + " WHERE scopeId = ? AND scopeType = ? AND nodeId = ? AND nodeType = ? " );
+ }
+
+
+ @Override
+ public Collection<? extends Statement> mark( final ApplicationScope scope, final Id node, final long timestamp ) {
ValidationUtils.validateApplicationScope( scope );
ValidationUtils.verifyIdentity( node );
GraphValidation.validateTimestamp( timestamp, "timestamp" );
- MutationBatch batch = keyspace.prepareMutationBatch().withConsistencyLevel( fig.getWriteCL() );
+ final Id applicationId = scope.getApplication();
+ final UUID applicationUuid = applicationId.getUuid();
+ final String applicationType = applicationId.getType();
+
+ final UUID nodeUuid = node.getUuid();
+ final String nodeType = node.getType();
- batch.withRow( GRAPH_DELETE, ScopedRowKey.fromKey( scope, node ) ).setTimestamp( timestamp )
- .putColumn( COLUMN_NAME, timestamp );
+ final BoundStatement statement =
+ this.insert.bind( applicationUuid, applicationType, nodeUuid, nodeType, timestamp, timestamp );
- return batch;
+
+ return Collections.singleton( statement );
}
@Override
- public MutationBatch delete( final ApplicationScope scope, final Id node, final long timestamp ) {
+ public Collection<? extends Statement> delete( final ApplicationScope scope, final Id node, final long timestamp ) {
ValidationUtils.validateApplicationScope( scope );
ValidationUtils.verifyIdentity( node );
GraphValidation.validateTimestamp( timestamp, "timestamp" );
- MutationBatch batch = keyspace.prepareMutationBatch().withConsistencyLevel( fig.getWriteCL() );
+ final Id applicationId = scope.getApplication();
+ final UUID applicationUuid = applicationId.getUuid();
+ final String applicationType = applicationId.getType();
+
+ final UUID nodeUuid = node.getUuid();
+ final String nodeType = node.getType();
+
+
+ final BoundStatement statement =
+ this.delete.bind(timestamp, applicationUuid, applicationType, nodeUuid, nodeType );
- batch.withRow( GRAPH_DELETE, ScopedRowKey.fromKey( scope, node ) ).setTimestamp( timestamp )
- .deleteColumn( COLUMN_NAME );
- return batch;
+ return Collections.singleton( statement );
}
@@ -148,26 +158,27 @@ public class NodeSerializationImpl implements NodeSerialization, Migration {
ValidationUtils.validateApplicationScope( scope );
ValidationUtils.verifyIdentity( node );
- ColumnFamilyQuery<ScopedRowKey<ApplicationScope, Id>, Boolean> query =
- keyspace.prepareQuery( GRAPH_DELETE ).setConsistencyLevel( fig.getReadCL() );
+ final Id applicationId = scope.getApplication();
+ final UUID applicationUuid = applicationId.getUuid();
+ final String applicationType = applicationId.getType();
+ final UUID nodeUuid = node.getUuid();
+ final String nodeType = node.getType();
- try {
- Column<Boolean> result = HystrixCassandra
- .user( query.getKey( ScopedRowKey.fromKey( scope, node ) ).getColumn( COLUMN_NAME ) )
- .getResult();
- return Optional.of( result.getLongValue() );
- }
- catch (RuntimeException re ) {
- if(re.getCause().getCause() instanceof NotFoundException) {
- //swallow, there's just no column
- return Optional.absent();
- }
+ final BoundStatement statement =
+ this.select.bind( applicationUuid, applicationType, nodeUuid, nodeType );
+
+
+ final Row row = session.execute( statement ).one();
- throw re;
+ if ( row == null ) {
+ return Optional.absent();
}
+ final Long timestamp = row.getLong( "timestamp" );
+
+ return Optional.of( timestamp );
}
@@ -176,33 +187,67 @@ public class NodeSerializationImpl implements NodeSerialization, Migration {
ValidationUtils.validateApplicationScope( scope );
Preconditions.checkNotNull( edges, "edges cannot be null" );
+ final Id applicationId = scope.getApplication();
+ final UUID applicationScopeId = applicationId.getUuid();
+ final String applicationScopeType = applicationId.getType();
- final ColumnFamilyQuery<ScopedRowKey<ApplicationScope, Id>, Boolean> query =
- keyspace.prepareQuery( GRAPH_DELETE ).setConsistencyLevel( fig.getReadCL() );
-
-
- final List<ScopedRowKey<ApplicationScope, Id>> keys =
- new ArrayList<ScopedRowKey<ApplicationScope, Id>>( edges.size() );
+ final Set<Id> uniqueIds = new HashSet<>( edges.size() );
//worst case all are marked
final Map<Id, Long> versions = new HashMap<>( edges.size() );
for ( final Edge edge : edges ) {
- keys.add( ScopedRowKey.fromKey( scope, edge.getSourceNode() ) );
- keys.add( ScopedRowKey.fromKey( scope, edge.getTargetNode() ) );
+ uniqueIds.add( edge.getSourceNode() );
+ uniqueIds.add( edge.getTargetNode() );
+ }
+//
+// List<Statement> statements = new ArrayList<>(uniqueIds.size());
+//
+// for(Id uniqueId : uniqueIds){
+// final BoundStatement bound = select.bind( applicationScopeId, applicationScopeType, uniqueId.getUuid(), uniqueId.getType() );
+//
+// statements.add( bound );
+// }
+//
+// final Prep
+//
+
+ final StringBuilder nodeIdInClause = new StringBuilder( );
+ final StringBuilder nodeTypeInClause = new StringBuilder( );
+
+ for ( Id uniqueId : uniqueIds ) {
+
+
+ nodeIdInClause.append( uniqueId.getUuid() ).append( " ," );
+ nodeTypeInClause.append("'").append( uniqueId.getType() ) .append( "' ," );
}
+ nodeIdInClause.deleteCharAt( nodeIdInClause.length() - 1);
+ nodeTypeInClause.deleteCharAt( nodeTypeInClause.length() - 1);
+
+ final StringBuilder builder = new StringBuilder( uniqueIds.size() * 50 );
+
+ builder.append( "select timestamp from " ).append( GRAPH_MARKED_EDGES )
+ .append( " WHERE scopeId = ").append(applicationScopeId).append(" AND scopeType = '").append(applicationScopeType ).append("' AND nodeId IN ( " ).append(nodeIdInClause).append(" ) AND nodeType IN ( ").append(nodeTypeInClause ).append( " )" );
+
+
+
+
+
+ final List<Row> results = session.execute(builder.toString()).all();
+
+
+ for ( Row row : results ) {
+
+ final long timestamp = row.getLong( "timestamp" );
+ final UUID nodeUuid = row.getUUID( "nodeId" );
+ final String nodeType = row.getString( "nodeType" );
+
- final Rows<ScopedRowKey<ApplicationScope, Id>, Boolean> results = HystrixCassandra
- .user( query.getRowSlice( keys ).withColumnSlice( Collections.singletonList( COLUMN_NAME ) ) )
- .getResult();
+ final Id nodeId = new SimpleId( nodeUuid, nodeType );
- for ( Row<ScopedRowKey<ApplicationScope, Id>, Boolean> row : results ) {
- Column<Boolean> column = row.getColumns().getColumnByName( COLUMN_NAME );
- if ( column != null ) {
- versions.put( row.getKey().getKey(), column.getLongValue() );
- }
+ versions.put( nodeId, timestamp );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/94a92c39/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroup.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroup.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroup.java
index 11bf7a4..1aea576 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroup.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardEntryGroup.java
@@ -35,9 +35,9 @@ import com.google.common.base.Preconditions;
/**
* There are cases where we need to read or write to more than 1 shard. This object encapsulates a set of shards that
* should be written to and read from. All reads should combine the data sets from all shards in the group, and writes
- * should be written to each shard. Once the shard can safely be compacted a background process should be triggered to
+ * should be written to a target shard. Once the shard can safely be compacted a background process should be triggered to
* remove additional shards and make seeks faster. This multiread/write should only occur during the time period of the
- * delta (in milliseconds), after which the next read will asynchronously compact the shards into a single shard.
+ * user configured delta (in milliseconds), after which the next read will asynchronously compact the shards into a single shard.
*/
public class ShardEntryGroup {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/94a92c39/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerIT.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerIT.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerIT.java
index 61bc7e7..7cdcc8a 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerIT.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/GraphManagerIT.java
@@ -74,7 +74,7 @@ public abstract class GraphManagerIT {
@Before
public void mockApp() {
- this.scope = new ApplicationScopeImpl(createId("application") );
+ this.scope = new ApplicationScopeImpl( createId( "application" ) );
}
@@ -1283,8 +1283,8 @@ public abstract class GraphManagerIT {
final long maxVersion = System.currentTimeMillis();
Iterator<Edge> results =
- gm.loadEdgesFromSource( createSearchByEdge( sourceId, edge1.getType(), maxVersion, null ) )
- .toBlocking().getIterator();
+ gm.loadEdgesFromSource( createSearchByEdge( sourceId, edge1.getType(), maxVersion, null ) ).toBlocking()
+ .getIterator();
assertEquals( "Edge found", edge2, results.next() );
@@ -1320,8 +1320,9 @@ public abstract class GraphManagerIT {
//now re-read, nothing should be there since they're marked
- results = gm.loadEdgesFromSource( createSearchByEdge( sourceId, edge1.getType(), maxVersion, null ) )
- .toBlocking().getIterator();
+ results =
+ gm.loadEdgesFromSource( createSearchByEdge( sourceId, edge1.getType(), maxVersion, null ) ).toBlocking()
+ .getIterator();
assertFalse( "No more edges", results.hasNext() );
@@ -1365,8 +1366,8 @@ public abstract class GraphManagerIT {
final long maxVersion = System.currentTimeMillis();
Iterator<Edge> results =
- gm.loadEdgesToTarget( createSearchByEdge( targetId, edge1.getType(), maxVersion, null ) )
- .toBlocking().getIterator();
+ gm.loadEdgesToTarget( createSearchByEdge( targetId, edge1.getType(), maxVersion, null ) ).toBlocking()
+ .getIterator();
assertEquals( "Edge found", edge2, results.next() );
@@ -1402,8 +1403,8 @@ public abstract class GraphManagerIT {
//now re-read, nothing should be there since they're marked
- results = gm.loadEdgesToTarget( createSearchByEdge( targetId, edge1.getType(), maxVersion, null ) )
- .toBlocking().getIterator();
+ results = gm.loadEdgesToTarget( createSearchByEdge( targetId, edge1.getType(), maxVersion, null ) ).toBlocking()
+ .getIterator();
assertFalse( "No more edges", results.hasNext() );
@@ -1476,7 +1477,7 @@ public abstract class GraphManagerIT {
@Test
- public void testSourceSubTypes() {
+ public void testSourceSubTypesPrefix() {
//now test sub edges
final GraphManager gm = getHelper( emf.createEdgeManager( scope ) );
@@ -1624,7 +1625,7 @@ public abstract class GraphManagerIT {
}
- @Test( expected = NullPointerException.class )
+ @Test(expected = NullPointerException.class)
public void invalidEdgeTypesWrite( @All Edge edge ) {
final GraphManager em = emf.createEdgeManager( scope );
@@ -1632,7 +1633,7 @@ public abstract class GraphManagerIT {
}
- @Test( expected = NullPointerException.class )
+ @Test(expected = NullPointerException.class)
public void invalidEdgeTypesDelete( @All Edge edge ) {
final GraphManager em = emf.createEdgeManager( scope );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/94a92c39/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/EdgeDeleteListenerTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/EdgeDeleteListenerTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/EdgeDeleteListenerTest.java
index 1f6df4c..ce618a4 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/EdgeDeleteListenerTest.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/EdgeDeleteListenerTest.java
@@ -34,6 +34,7 @@ import org.slf4j.LoggerFactory;
import org.apache.usergrid.persistence.collection.guice.MigrationManagerRule;
import org.apache.usergrid.persistence.core.cassandra.ITRunner;
+import org.apache.usergrid.persistence.core.javadriver.BatchStatementUtils;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.graph.GraphFig;
import org.apache.usergrid.persistence.graph.GraphManagerFactory;
@@ -45,6 +46,7 @@ import org.apache.usergrid.persistence.graph.serialization.EdgeSerialization;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.util.UUIDGenerator;
+import com.datastax.driver.core.Session;
import com.google.inject.Inject;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
@@ -83,7 +85,7 @@ public class EdgeDeleteListenerTest {
@Inject
- protected GraphManagerFactory emf;
+ protected Session session;
@Inject
@@ -123,9 +125,9 @@ public class EdgeDeleteListenerTest {
MarkedEdge edgeV3 = createMarkedEdge( sourceId, edgeType, targetId, edgeTimestamp + 2 );
- storageEdgeSerialization.writeEdge( scope, edgeV1, UUIDGenerator.newTimeUUID() ).execute();
- storageEdgeSerialization.writeEdge( scope, edgeV2, UUIDGenerator.newTimeUUID() ).execute();
- storageEdgeSerialization.writeEdge( scope, edgeV3, UUIDGenerator.newTimeUUID() ).execute();
+ BatchStatementUtils.runBatches( session, storageEdgeSerialization.writeEdge( scope, edgeV1, UUIDGenerator.newTimeUUID() ));
+ BatchStatementUtils.runBatches( session, storageEdgeSerialization.writeEdge( scope, edgeV2, UUIDGenerator.newTimeUUID() ));
+ BatchStatementUtils.runBatches( session, storageEdgeSerialization.writeEdge( scope, edgeV3, UUIDGenerator.newTimeUUID() ));
@@ -218,13 +220,13 @@ public class EdgeDeleteListenerTest {
final UUID foobar = UUIDGenerator.newTimeUUID();
- storageEdgeSerialization.writeEdge( scope, edgeV1, foobar ).execute();
- storageEdgeSerialization.writeEdge( scope, edgeV2, foobar ).execute();
- storageEdgeSerialization.writeEdge( scope, edgeV3, foobar ).execute();
+ BatchStatementUtils.runBatches( session, storageEdgeSerialization.writeEdge( scope, edgeV1, foobar ));
+ BatchStatementUtils.runBatches( session, storageEdgeSerialization.writeEdge( scope, edgeV2, foobar ));
+ BatchStatementUtils.runBatches( session, storageEdgeSerialization.writeEdge( scope, edgeV3, foobar ));
- edgeMetadataSerialization.writeEdge( scope, edgeV1 ).execute();
- edgeMetadataSerialization.writeEdge( scope, edgeV2 ).execute();
- edgeMetadataSerialization.writeEdge( scope, edgeV3 ).execute();
+ BatchStatementUtils.runBatches( session, edgeMetadataSerialization.writeEdge( scope, edgeV1 ));
+ BatchStatementUtils.runBatches( session, edgeMetadataSerialization.writeEdge( scope, edgeV2 ));
+ BatchStatementUtils.runBatches( session, edgeMetadataSerialization.writeEdge( scope, edgeV3 ));
//now perform the listener execution, should only clean up to edge v2
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/94a92c39/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListenerTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListenerTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListenerTest.java
index 3b66a31..414db10 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListenerTest.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/NodeDeleteListenerTest.java
@@ -33,6 +33,7 @@ import org.slf4j.LoggerFactory;
import org.apache.usergrid.persistence.collection.guice.MigrationManagerRule;
import org.apache.usergrid.persistence.core.cassandra.ITRunner;
+import org.apache.usergrid.persistence.core.javadriver.BatchStatementUtils;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.graph.Edge;
import org.apache.usergrid.persistence.graph.GraphFig;
@@ -47,6 +48,7 @@ import org.apache.usergrid.persistence.graph.serialization.NodeSerialization;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.util.UUIDGenerator;
+import com.datastax.driver.core.Session;
import com.google.inject.Inject;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
@@ -96,6 +98,9 @@ public class NodeDeleteListenerTest {
@Inject
protected GraphFig graphFig;
+ @Inject
+ protected Session session;
+
protected ApplicationScope scope;
@@ -169,7 +174,7 @@ public class NodeDeleteListenerTest {
UUID deleteEventTimestamp = UUIDGenerator.newTimeUUID();
long timestamp = System.currentTimeMillis();
- nodeSerialization.mark( scope, sourceNode, timestamp ).execute();
+ BatchStatementUtils.runBatches( session, nodeSerialization.mark( scope, sourceNode, timestamp ));
int count = deleteListener.receive( scope, sourceNode, deleteEventTimestamp ).toBlocking().last();
@@ -254,7 +259,7 @@ public class NodeDeleteListenerTest {
//mark the node so
final long deleteBefore = System.currentTimeMillis();
- nodeSerialization.mark( scope, targetNode, deleteBefore ).execute();
+ BatchStatementUtils.runBatches( session, nodeSerialization.mark( scope, targetNode, deleteBefore ));
int count = deleteListener.receive( scope, targetNode, UUIDGenerator.newTimeUUID() ).toBlocking().last();
@@ -364,7 +369,7 @@ public class NodeDeleteListenerTest {
long deleteVersion = Long.MAX_VALUE;
- nodeSerialization.mark( scope, toDelete, deleteVersion ).execute();
+ BatchStatementUtils.runBatches( session, nodeSerialization.mark( scope, toDelete, deleteVersion ));
int count = deleteListener.receive( scope, toDelete, UUIDGenerator.newTimeUUID() ).toBlocking().last();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/94a92c39/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeDeleteRepairTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeDeleteRepairTest.java b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeDeleteRepairTest.java
index e8af91c..c779809 100644
--- a/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeDeleteRepairTest.java
+++ b/stack/corepersistence/graph/src/test/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeDeleteRepairTest.java
@@ -32,6 +32,7 @@ import org.slf4j.LoggerFactory;
import org.apache.usergrid.persistence.collection.guice.MigrationManagerRule;
import org.apache.usergrid.persistence.core.cassandra.ITRunner;
+import org.apache.usergrid.persistence.core.javadriver.BatchStatementUtils;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.graph.MarkedEdge;
import org.apache.usergrid.persistence.graph.SearchByEdgeType;
@@ -41,6 +42,7 @@ import org.apache.usergrid.persistence.graph.serialization.EdgeSerialization;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.util.UUIDGenerator;
+import com.datastax.driver.core.Session;
import com.google.inject.Inject;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
@@ -76,6 +78,9 @@ public class EdgeDeleteRepairTest {
@Inject
protected EdgeDeleteRepair edgeDeleteRepair;
+ @Inject
+ protected Session session;
+
protected ApplicationScope scope;
@@ -122,11 +127,12 @@ public class EdgeDeleteRepairTest {
//write it as non deleted to storage
- storageEdgeSerialization.writeEdge( scope, edge1, UUIDGenerator.newTimeUUID() ).execute();
+ BatchStatementUtils.runBatches( session,
+ storageEdgeSerialization.writeEdge( scope, edge1, UUIDGenerator.newTimeUUID() ));
final MarkedEdge edge2 = createEdge( sourceId, edgeType, targetId );
- storageEdgeSerialization.writeEdge( scope, edge2, UUIDGenerator.newTimeUUID() ).execute();
+ BatchStatementUtils.runBatches( session, storageEdgeSerialization.writeEdge( scope, edge2, UUIDGenerator.newTimeUUID() ));
//now repair delete the first edge