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/14 17:20:25 UTC
[6/6] git commit: Implemented New rolling shard algorithm. This
should allow eventual shard consistency between all clients without the need
for an external locking allocation system
Implemented New rolling shard algorithm. This should allow eventual shard consistency between all clients without the need for an external locking allocation system
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/e9d652dd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/e9d652dd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/e9d652dd
Branch: refs/heads/USERGRID-188
Commit: e9d652dd3ec0862e032145b97522d1a73f6f4f53
Parents: 71cb9c2
Author: Todd Nine <tn...@apigee.com>
Authored: Wed Jul 2 09:35:36 2014 -0600
Committer: Todd Nine <to...@apache.org>
Committed: Thu Aug 14 09:19:15 2014 -0600
----------------------------------------------------------------------
.../UniqueValueSerializationStrategyImpl.java | 2 +-
.../persistence/core/astyanax/ColumnTypes.java | 3 +
.../astyanax/MultiKeyColumnNameIterator.java | 217 ++++
.../core/astyanax/MultiTennantColumnFamily.java | 13 +-
.../persistence/core/rx/OrderedMerge.java | 3 +
.../core/scope/ApplicationScopeImpl.java | 3 +-
.../usergrid/persistence/graph/GraphFig.java | 30 +-
.../persistence/graph/guice/GraphModule.java | 37 +-
.../impl/stage/EdgeDeleteListenerImpl.java | 1 -
.../graph/serialization/EdgeSerialization.java | 2 +-
.../impl/EdgeSerializationImpl.java | 1019 +++---------------
.../impl/NodeSerializationImpl.java | 10 +-
.../serialization/impl/shard/DirectedEdge.java | 41 +
.../impl/shard/DirectedEdgeMeta.java | 405 +++++++
.../impl/shard/EdgeColumnFamilies.java | 60 ++
.../serialization/impl/shard/EdgeRowKey.java | 44 +
.../impl/shard/EdgeShardSerialization.java | 21 +-
.../impl/shard/EdgeShardStrategy.java | 44 +-
.../impl/shard/NodeShardAllocation.java | 20 +-
.../impl/shard/NodeShardApproximation.java | 30 +-
.../impl/shard/NodeShardCache.java | 20 +-
.../serialization/impl/shard/NodeType.java | 66 ++
.../graph/serialization/impl/shard/RowKey.java | 54 +
.../serialization/impl/shard/RowKeyType.java | 60 ++
.../graph/serialization/impl/shard/Shard.java | 142 +++
.../impl/shard/ShardEntryGroup.java | 270 +++++
.../impl/shard/ShardGroupCompaction.java | 43 +
.../serialization/impl/shard/ShardOperator.java | 43 +
.../impl/shard/ShardedEdgeSerialization.java | 119 ++
.../serialization/impl/shard/count/Counter.java | 8 +-
.../shard/count/NodeShardApproximationImpl.java | 174 ++-
.../count/NodeShardCounterSerialization.java | 2 +-
.../NodeShardCounterSerializationImpl.java | 80 +-
.../impl/shard/count/ShardKey.java | 51 +-
.../impl/shard/impl/EdgeRowKey.java | 46 -
.../impl/shard/impl/EdgeRowKeySerializer.java | 17 +-
.../impl/shard/impl/EdgeSearcher.java | 143 +++
.../impl/shard/impl/EdgeSerializer.java | 77 ++
.../shard/impl/EdgeShardRowKeySerializer.java | 103 ++
.../shard/impl/EdgeShardSerializationImpl.java | 93 +-
.../shard/impl/NodeShardAllocationImpl.java | 193 ++--
.../impl/shard/impl/NodeShardCacheImpl.java | 256 +++--
.../impl/shard/impl/RowSerializer.java | 64 ++
.../impl/shard/impl/RowTypeSerializer.java | 60 ++
.../shard/impl/ShardEntryGroupIterator.java | 100 ++
.../shard/impl/ShardGroupCompactionImpl.java | 75 ++
.../impl/shard/impl/ShardRowIterator.java | 134 +++
.../impl/ShardedEdgeSerializationImpl.java | 656 +++++++++++
.../shard/impl/SizebasedEdgeColumnFamilies.java | 150 +++
.../shard/impl/SizebasedEdgeShardStrategy.java | 52 +-
.../impl/shard/impl/SourceEdgeSearcher.java | 22 +
.../graph/ComittedGraphManagerIT.java | 138 ---
.../graph/CommittedGraphManagerIT.java | 138 +++
.../graph/GraphManagerShardingIT.java | 43 +-
.../impl/shard/EdgeShardSerializationTest.java | 108 +-
.../impl/shard/NodeShardAllocationTest.java | 461 ++++++--
.../impl/shard/NodeShardCacheTest.java | 228 ++--
.../impl/shard/ShardEntryGroupTest.java | 422 ++++++++
.../shard/count/NodeShardApproximationTest.java | 107 +-
.../NodeShardCounterSerializationTest.java | 9 +-
.../shard/impl/ShardEntryGroupIteratorTest.java | 232 ++++
stack/corepersistence/pom.xml | 2 +-
62 files changed, 5429 insertions(+), 1837 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e9d652dd/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/UniqueValueSerializationStrategyImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/UniqueValueSerializationStrategyImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/UniqueValueSerializationStrategyImpl.java
index 73c038e..b7e113e 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/UniqueValueSerializationStrategyImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/write/UniqueValueSerializationStrategyImpl.java
@@ -56,7 +56,7 @@ public class UniqueValueSerializationStrategyImpl
private static final EntityVersionSerializer ENTITY_VERSION_SER = new EntityVersionSerializer();
- private static final MultiTennantColumnFamily<CollectionScope, Field, EntityVersion>
+ private static final MultiTennantColumnFamily<CollectionScope, Field, EntityVersion>
CF_UNIQUE_VALUES =
new MultiTennantColumnFamily<CollectionScope, Field, EntityVersion>( "Unique_Values",
ROW_KEY_SER,
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e9d652dd/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/ColumnTypes.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/ColumnTypes.java b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/ColumnTypes.java
index b5e41ff..a055ca7 100644
--- a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/ColumnTypes.java
+++ b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/ColumnTypes.java
@@ -1,6 +1,7 @@
package org.apache.usergrid.persistence.core.astyanax;
+import org.apache.cassandra.db.marshal.BooleanType;
import org.apache.cassandra.db.marshal.DynamicCompositeType;
@@ -19,6 +20,8 @@ public class ColumnTypes {
public static final String UUID_TYPE_REVERSED = "UUIDType(reversed=true)";
+ public static final String BOOLEAN = "BooleanType";
+
/**
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e9d652dd/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/MultiKeyColumnNameIterator.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/MultiKeyColumnNameIterator.java b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/MultiKeyColumnNameIterator.java
new file mode 100644
index 0000000..c5a8c95
--- /dev/null
+++ b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/MultiKeyColumnNameIterator.java
@@ -0,0 +1,217 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one
+ * * or more contributor license agreements. See the NOTICE file
+ * * distributed with this work for additional information
+ * * regarding copyright ownership. The ASF licenses this file
+ * * to you under the Apache License, Version 2.0 (the
+ * * "License"); you may not use this file except in compliance
+ * * with the License. You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing,
+ * * software distributed under the License is distributed on an
+ * * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * * KIND, either express or implied. See the License for the
+ * * specific language governing permissions and limitations
+ * * under the License.
+ *
+ */
+package org.apache.usergrid.persistence.core.astyanax;
+
+
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.NoSuchElementException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Exchanger;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.usergrid.persistence.core.rx.OrderedMerge;
+
+import com.amazonaws.services.redshift.model.UnsupportedOptionException;
+import com.google.common.base.Preconditions;
+
+import rx.Observable;
+import rx.Subscriber;
+import rx.schedulers.Schedulers;
+
+
+/**
+ * Simple iterator that wraps a collection of ColumnNameIterators. We do this because we can't page with a
+ * multiRangeScan correctly for multiple round trips. As a result, we do this since only 1 iterator with minimum values
+ * could potentially feed the entire result set.
+ *
+ * Compares the parsed values and puts them in order. If more than one row key emits the same value the first value is
+ * selected, and ignored from subsequent iterators.
+ */
+public class MultiKeyColumnNameIterator<C, T> implements Iterable<T>, Iterator<T> {
+
+
+ private InnerIterator<T> iterator;
+
+
+ public MultiKeyColumnNameIterator( final Collection<ColumnNameIterator<C, T>> columnNameIterators,
+ final Comparator<T> comparator, final int bufferSize ) {
+
+
+ Observable<T>[] observables = new Observable[columnNameIterators.size()];
+
+ int i = 0;
+
+ for ( ColumnNameIterator<C, T> columnNameIterator : columnNameIterators ) {
+
+ observables[i] = Observable.from( columnNameIterator, Schedulers.io() );
+
+ i++;
+ }
+
+
+ //merge them into 1 observable, and remove duplicates from the stream
+ Observable<T> merged = OrderedMerge.orderedMerge( comparator, bufferSize, observables ).distinctUntilChanged();
+
+
+ iterator = new InnerIterator(bufferSize);
+
+ merged.subscribe( iterator );
+ }
+
+
+ @Override
+ public Iterator<T> iterator() {
+ return this;
+ }
+
+
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+
+ @Override
+ public T next() {
+ return iterator.next();
+ }
+
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException( "You cannot remove elements from a merged iterator, it is read only" );
+ }
+
+
+ /**
+ * Internal iterator that will put next elements into a blocking queue until it reaches capacity. At this point it
+ * will block then emitting thread until more elements are taken. Assumed the Observable is run on a I/O thread,
+ * NOT the current thread.
+ */
+ private final class InnerIterator<T> extends Subscriber<T> implements Iterator<T> {
+
+ private CountDownLatch startLatch = new CountDownLatch( 1 );
+
+ private final LinkedBlockingQueue<T> queue;
+
+
+ private Throwable error;
+ private boolean done = false;
+
+ private T next;
+
+
+ private InnerIterator( int maxSize ) {
+ queue = new LinkedBlockingQueue<>( maxSize );
+ }
+
+
+ @Override
+ public boolean hasNext() {
+
+
+ //we're done
+ if ( next != null ) {
+ return true;
+ }
+
+
+ try {
+ startLatch.await();
+ }
+ catch ( InterruptedException e ) {
+ throw new RuntimeException( "Unable to wait for start of submission" );
+ }
+
+
+
+ //this is almost a busy wait, and is intentional, if we have nothing to poll, we want to get it as soon
+ //as it's available. We generally only hit this once
+ do {
+ next = queue.poll();
+ }
+ while ( next == null && !done );
+
+
+ return next != null;
+ }
+
+
+ @Override
+ public T next() {
+
+ if ( error != null ) {
+ throw new RuntimeException( "An error occurred when populating the iterator", error );
+ }
+
+ if ( !hasNext() ) {
+ throw new NoSuchElementException( "No more elements are present" );
+ }
+
+
+ T toReturn = next;
+ next = null;
+ return toReturn;
+ }
+
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOptionException( "Remove is unsupported" );
+ }
+
+
+ @Override
+ public void onCompleted() {
+ done = true;
+ startLatch.countDown();
+ }
+
+
+ @Override
+ public void onError( final Throwable e ) {
+ error = e;
+ done = true;
+ startLatch.countDown();
+ }
+
+
+ @Override
+ public void onNext( final T t ) {
+
+ //may block if we get full, that's expected behavior
+ try {
+ queue.put( t );
+ }
+ catch ( InterruptedException e ) {
+ throw new RuntimeException( "Unable to take from queue" );
+ }
+
+ startLatch.countDown();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e9d652dd/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/MultiTennantColumnFamily.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/MultiTennantColumnFamily.java b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/MultiTennantColumnFamily.java
index 19e5aae..6234184 100644
--- a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/MultiTennantColumnFamily.java
+++ b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/astyanax/MultiTennantColumnFamily.java
@@ -33,20 +33,15 @@ import com.netflix.astyanax.model.ColumnFamily;
public class MultiTennantColumnFamily<S extends ApplicationScope, K, V>
extends ColumnFamily<ScopedRowKey<S, K>, V> {
- public MultiTennantColumnFamily(
- final String columnFamilyName,
- final Serializer<ScopedRowKey<S, K>> keySerializer,
- final Serializer<V> columnSerializer ) {
+ public MultiTennantColumnFamily( final String columnFamilyName, final Serializer<ScopedRowKey<S, K>> keySerializer,
+ final Serializer<V> columnSerializer ) {
super( columnFamilyName, keySerializer, columnSerializer );
}
- public MultiTennantColumnFamily(
- final String columnFamilyName,
- final Serializer<ScopedRowKey<S, K>> keySerializer,
- final Serializer<V> columnSerializer,
- final Serializer<?> defaultValueSerializer ) {
+ public MultiTennantColumnFamily( final String columnFamilyName, final Serializer<ScopedRowKey<S, K>> keySerializer,
+ final Serializer<V> columnSerializer, final Serializer<?> defaultValueSerializer ) {
super( columnFamilyName, keySerializer, columnSerializer, defaultValueSerializer );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e9d652dd/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/rx/OrderedMerge.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/rx/OrderedMerge.java b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/rx/OrderedMerge.java
index 5ba3fbb..4032176 100644
--- a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/rx/OrderedMerge.java
+++ b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/rx/OrderedMerge.java
@@ -181,6 +181,9 @@ public final class OrderedMerge<T> implements Observable.OnSubscribe<T> {
InnerObserver<T> maxObserver = null;
T max = null;
+ /**
+ * TODO T.N. change this to be an 0(1) for min and O(log n) to update after pop rather than O(n*inner)
+ */
for ( InnerObserver<T> inner : innerSubscribers ) {
//nothing to do, this inner
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e9d652dd/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/scope/ApplicationScopeImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/scope/ApplicationScopeImpl.java b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/scope/ApplicationScopeImpl.java
index 692ba49..4e067c2 100644
--- a/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/scope/ApplicationScopeImpl.java
+++ b/stack/corepersistence/common/src/main/java/org/apache/usergrid/persistence/core/scope/ApplicationScopeImpl.java
@@ -33,7 +33,8 @@ public class ApplicationScopeImpl implements ApplicationScope {
public ApplicationScopeImpl( final Id application ) {
- this.application = application;}
+ this.application = application;
+ }
@Override
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e9d652dd/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/GraphFig.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/GraphFig.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/GraphFig.java
index e0ce45c..45064c9 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/GraphFig.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/GraphFig.java
@@ -39,11 +39,26 @@ public interface GraphFig extends GuicyFig {
public static final String SHARD_CACHE_SIZE = "usergrid.graph.shard.cache.size";
+
+ /**
+ * Get the cache timeout. The local cache will exist for this amount of time max (in millis).
+ */
public static final String SHARD_CACHE_TIMEOUT = "usergrid.graph.shard.cache.timeout";
- public static final String COUNTER_WRITE_FLUSH_COUNT = "usergrid.graph.shard.counter.flush.count";
+ /**
+ * The minimum amount of time than can occur (in millis) between shard allocation. Must be at least 2x the cache timeout.
+ *
+ * Note that you should also pad this for node clock drift. A good value for this would be 2x the shard cache timeout + 30 seconds,
+ * assuming you have NTP and allow a max drift of 30 seconds
+ */
+ public static final String SHARD_MIN_DELTA = "usergrid.graph.shard.min.delta";
+
+
+ public static final String COUNTER_WRITE_FLUSH_COUNT = "usergrid.graph.shard.counter.beginFlush.count";
- public static final String COUNTER_WRITE_FLUSH_INTERVAL = "usergrid.graph.shard.counter.flush.interval";
+ public static final String COUNTER_WRITE_FLUSH_INTERVAL = "usergrid.graph.shard.counter.beginFlush.interval";
+
+ public static final String COUNTER_WRITE_FLUSH_QUEUE_SIZE = "usergrid.graph.shard.counter.queue.size";
@@ -69,11 +84,18 @@ public interface GraphFig extends GuicyFig {
@Key(SHARD_CACHE_TIMEOUT)
long getShardCacheTimeout();
+ @Default("60000")
+ @Key( SHARD_MIN_DELTA )
+ long getShardMinDelta();
+
+
@Default( "250000" )
@Key( SHARD_CACHE_SIZE )
long getShardCacheSize();
+
+
@Default( "10000" )
@Key( COUNTER_WRITE_FLUSH_COUNT )
long getCounterFlushCount();
@@ -82,5 +104,9 @@ public interface GraphFig extends GuicyFig {
@Default( "30000" )
@Key( COUNTER_WRITE_FLUSH_INTERVAL )
long getCounterFlushInterval();
+
+ @Default( "1000" )
+ @Key(COUNTER_WRITE_FLUSH_QUEUE_SIZE )
+ int getCounterFlushQueueSize();
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e9d652dd/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/guice/GraphModule.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/guice/GraphModule.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/guice/GraphModule.java
index aca8d00..ca7c270 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/guice/GraphModule.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/guice/GraphModule.java
@@ -43,17 +43,21 @@ import org.apache.usergrid.persistence.graph.serialization.NodeSerialization;
import org.apache.usergrid.persistence.graph.serialization.impl.EdgeMetadataSerializationImpl;
import org.apache.usergrid.persistence.graph.serialization.impl.EdgeSerializationImpl;
import org.apache.usergrid.persistence.graph.serialization.impl.NodeSerializationImpl;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.EdgeColumnFamilies;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.EdgeShardSerialization;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.EdgeShardStrategy;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.NodeShardAllocation;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.NodeShardApproximation;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.NodeShardCache;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.ShardedEdgeSerialization;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.count.NodeShardApproximationImpl;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.count.NodeShardCounterSerialization;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.count.NodeShardCounterSerializationImpl;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.EdgeShardSerializationImpl;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.NodeShardAllocationImpl;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.NodeShardCacheImpl;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.ShardedEdgeSerializationImpl;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.SizebasedEdgeColumnFamilies;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.SizebasedEdgeShardStrategy;
import com.google.inject.AbstractModule;
@@ -101,7 +105,6 @@ public class GraphModule extends AbstractModule {
bind( EdgeShardSerialization.class ).to( EdgeShardSerializationImpl.class );
-
//Repair/cleanup classes.
bind( EdgeMetaRepair.class ).to( EdgeMetaRepairImpl.class );
bind( EdgeDeleteRepair.class ).to( EdgeDeleteRepairImpl.class );
@@ -111,7 +114,9 @@ public class GraphModule extends AbstractModule {
* Add our listeners
*/
bind( NodeDeleteListener.class ).to( NodeDeleteListenerImpl.class );
- bind( EdgeDeleteListener.class).to( EdgeDeleteListenerImpl.class );
+ bind( EdgeDeleteListener.class ).to( EdgeDeleteListenerImpl.class );
+
+
/**
@@ -129,7 +134,7 @@ public class GraphModule extends AbstractModule {
migrationBinding.addBinding().to( Key.get( EdgeMetadataSerialization.class ) );
//bind each singleton to the multi set. Otherwise we won't migrate properly
- migrationBinding.addBinding().to( Key.get( EdgeSerialization.class, StorageEdgeSerialization.class ) );
+ migrationBinding.addBinding().to( Key.get( EdgeColumnFamilies.class, StorageEdgeSerialization.class ) );
migrationBinding.addBinding().to( Key.get( EdgeShardSerialization.class ) );
migrationBinding.addBinding().to( Key.get( NodeShardCounterSerialization.class ) );
@@ -143,22 +148,36 @@ public class GraphModule extends AbstractModule {
@Singleton
@Inject
@StorageEdgeSerialization
- public EdgeSerialization permanentStorageSerialization( final NodeShardCache cache, final Keyspace keyspace,
- final CassandraConfig cassandraConfig,
- final GraphFig graphFig,
- final NodeShardApproximation shardApproximation) {
+ public EdgeSerialization storageSerialization( final NodeShardCache cache, final Keyspace keyspace,
+ final CassandraConfig cassandraConfig, final GraphFig graphFig,
+ final NodeShardApproximation shardApproximation,
+ final TimeService timeService,
+ @StorageEdgeSerialization
+ final EdgeColumnFamilies edgeColumnFamilies ) {
final EdgeShardStrategy sizeBasedStrategy = new SizebasedEdgeShardStrategy( cache, shardApproximation );
+
+ final ShardedEdgeSerialization serialization = new ShardedEdgeSerializationImpl(keyspace, cassandraConfig, graphFig, sizeBasedStrategy,
+ timeService );
+
+
final EdgeSerializationImpl edgeSerialization =
- new EdgeSerializationImpl( keyspace, cassandraConfig, graphFig, sizeBasedStrategy );
+ new EdgeSerializationImpl( keyspace, cassandraConfig, graphFig, sizeBasedStrategy, edgeColumnFamilies,
+ serialization );
return edgeSerialization;
}
-
+ @Provides
+ @Singleton
+ @Inject
+ @StorageEdgeSerialization
+ public EdgeColumnFamilies storageSerializationColumnFamilies() {
+ return new SizebasedEdgeColumnFamilies();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e9d652dd/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeDeleteListenerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeDeleteListenerImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeDeleteListenerImpl.java
index 55ab25b..6cfe6cc 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeDeleteListenerImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/EdgeDeleteListenerImpl.java
@@ -48,7 +48,6 @@ public class EdgeDeleteListenerImpl implements EdgeDeleteListener {
@Inject
public EdgeDeleteListenerImpl(
final EdgeDeleteRepair edgeDeleteRepair, final EdgeMetaRepair edgeMetaRepair ) {
-
this.edgeDeleteRepair = edgeDeleteRepair;
this.edgeMetaRepair = edgeMetaRepair;
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e9d652dd/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 b80b7ab..0caffc3 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
@@ -36,7 +36,7 @@ import com.netflix.astyanax.MutationBatch;
/**
* Simple interface for serializing ONLY an edge
*/
-public interface EdgeSerialization extends Migration {
+public interface EdgeSerialization {
/**
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e9d652dd/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 10d1048..18facc0 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
@@ -20,333 +20,82 @@
package org.apache.usergrid.persistence.graph.serialization.impl;
-import java.nio.ByteBuffer;
-import java.util.Arrays;
-import java.util.Collection;
import java.util.Iterator;
-import java.util.NoSuchElementException;
import java.util.UUID;
import javax.inject.Inject;
-import org.apache.cassandra.db.marshal.BytesType;
-import org.apache.cassandra.db.marshal.DynamicCompositeType;
-
import org.apache.usergrid.persistence.core.astyanax.CassandraConfig;
-import org.apache.usergrid.persistence.core.astyanax.ColumnNameIterator;
-import org.apache.usergrid.persistence.core.astyanax.ColumnParser;
-import org.apache.usergrid.persistence.core.astyanax.ColumnTypes;
-import org.apache.usergrid.persistence.core.astyanax.CompositeFieldSerializer;
-import org.apache.usergrid.persistence.core.astyanax.IdColDynamicCompositeSerializer;
-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.migration.Migration;
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.GraphFig;
import org.apache.usergrid.persistence.graph.MarkedEdge;
import org.apache.usergrid.persistence.graph.SearchByEdge;
import org.apache.usergrid.persistence.graph.SearchByEdgeType;
import org.apache.usergrid.persistence.graph.SearchByIdType;
-import org.apache.usergrid.persistence.graph.impl.SimpleMarkedEdge;
import org.apache.usergrid.persistence.graph.serialization.EdgeSerialization;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.DirectedEdgeMeta;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.EdgeColumnFamilies;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.EdgeShardStrategy;
-import org.apache.usergrid.persistence.graph.serialization.util.EdgeHasher;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.Shard;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.ShardEntryGroup;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.ShardOperator;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.ShardedEdgeSerialization;
import org.apache.usergrid.persistence.graph.serialization.util.EdgeUtils;
import org.apache.usergrid.persistence.model.entity.Id;
-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.Serializer;
-import com.netflix.astyanax.model.AbstractComposite;
-import com.netflix.astyanax.model.Column;
-import com.netflix.astyanax.model.CompositeBuilder;
-import com.netflix.astyanax.model.CompositeParser;
-import com.netflix.astyanax.model.DynamicComposite;
-import com.netflix.astyanax.query.RowQuery;
-import com.netflix.astyanax.serializers.AbstractSerializer;
-import com.netflix.astyanax.serializers.LongSerializer;
-import com.netflix.astyanax.serializers.StringSerializer;
-import com.netflix.astyanax.serializers.UUIDSerializer;
-import com.netflix.astyanax.util.RangeBuilder;
import static com.google.common.base.Preconditions.checkNotNull;
-import static org.apache.usergrid.persistence.core.astyanax.ColumnTypes.LONG_TYPE_REVERSED;
-import static org.apache.usergrid.persistence.core.astyanax.ColumnTypes.UUID_TYPE_REVERSED;
/**
* Serialization for edges. Delegates partitioning to the sharding strategy.
*/
@Singleton
-public class EdgeSerializationImpl implements EdgeSerialization, Migration {
-
-
- //Row key with no type
- private static final RowSerializer ROW_SERIALIZER = new RowSerializer();
-
- //row key with target id type
- private static final RowTypeSerializer ROW_TYPE_SERIALIZER = new RowTypeSerializer();
-
- private static final EdgeRowKeySerializer EDGE_ROW_KEY_SERIALIZER = new EdgeRowKeySerializer();
-
- //Edge serializers
- private static final EdgeSerializer EDGE_SERIALIZER = new EdgeSerializer();
-
- private static final LongSerializer LONG_SERIALIZER = LongSerializer.get();
-
-
- /**
- * Constant for the dynamic composite comparator type we'll need
- */
- public static final String EDGE_DYNAMIC_COMPOSITE_TYPE =
- //we purposefully associate lower case "l" and "u" with reversed types. This way we can use
- //the default serialization in Astayanax, but get reverse order in cassandra
- DynamicCompositeType.class.getSimpleName() + "(s=>UTF8Type,l=>" + LONG_TYPE_REVERSED + ",u=>"
- + UUID_TYPE_REVERSED + ")";
-
+public class EdgeSerializationImpl implements EdgeSerialization {
- /**
- * Get all graph edge versions
- */
- private final MultiTennantColumnFamily<ApplicationScope, EdgeRowKey, Long> graphEdgeVersionsCf;
-
-
- // column families
- /**
- * Edges that are from the source node. The row key is the source node
- */
- private final MultiTennantColumnFamily<ApplicationScope, RowKey, DirectedEdge> sourceNodeEdgesCf;
-
-
- /**
- * Edges that are incoming to the target node. The target node is the row key
- */
- private final MultiTennantColumnFamily<ApplicationScope, RowKey, DirectedEdge> targetNodeEdgesCf;
-
- /**
- * The edges that are from the source node with target type. The source node is the row key.
- */
- private final MultiTennantColumnFamily<ApplicationScope, RowKeyType, DirectedEdge> sourceNodeTargetTypeCf;
-
-
- /**
- * The edges that are to the target node with the source type. The target node is the row key
- */
- private final MultiTennantColumnFamily<ApplicationScope, RowKeyType, DirectedEdge> targetNodeSourceTypeCf;
protected final Keyspace keyspace;
protected final CassandraConfig cassandraConfig;
protected final GraphFig graphFig;
protected final EdgeShardStrategy edgeShardStrategy;
+ protected final EdgeColumnFamilies edgeColumnFamilies;
+ protected final ShardedEdgeSerialization shardedEdgeSerialization;
@Inject
public EdgeSerializationImpl( final Keyspace keyspace, final CassandraConfig cassandraConfig,
- final GraphFig graphFig, final EdgeShardStrategy edgeShardStrategy ) {
+ final GraphFig graphFig, final EdgeShardStrategy edgeShardStrategy,
+ final EdgeColumnFamilies edgeColumnFamilies,
+ final ShardedEdgeSerialization shardedEdgeSerialization ) {
+
checkNotNull( "keyspace required", keyspace );
checkNotNull( "cassandraConfig required", cassandraConfig );
checkNotNull( "consistencyFig required", graphFig );
- checkNotNull( "sourceNodeCfName required", edgeShardStrategy.getSourceNodeCfName() );
- checkNotNull( "targetNodeCfName required", edgeShardStrategy.getTargetNodeCfName() );
- checkNotNull( "sourceNodeTargetTypeCfName required", edgeShardStrategy.getSourceNodeTargetTypeCfName() );
- checkNotNull( "targetNodeSourceTypeCfName required", edgeShardStrategy.getTargetNodeSourceTypeCfName() );
this.keyspace = keyspace;
this.cassandraConfig = cassandraConfig;
this.graphFig = graphFig;
this.edgeShardStrategy = edgeShardStrategy;
-
- //initialize the CF's from our implementation
- sourceNodeEdgesCf = new MultiTennantColumnFamily<>( edgeShardStrategy.getSourceNodeCfName(),
- new OrganizationScopedRowKeySerializer<>( ROW_SERIALIZER ), EDGE_SERIALIZER );
-
-
- targetNodeEdgesCf = new MultiTennantColumnFamily<>( edgeShardStrategy.getTargetNodeCfName(),
- new OrganizationScopedRowKeySerializer<>( ROW_SERIALIZER ), EDGE_SERIALIZER );
-
-
- sourceNodeTargetTypeCf = new MultiTennantColumnFamily<>( edgeShardStrategy.getSourceNodeTargetTypeCfName(),
- new OrganizationScopedRowKeySerializer<>( ROW_TYPE_SERIALIZER ), EDGE_SERIALIZER );
-
-
- /**
- * The edges that are to the target node with the source type. The target node is the row key
- */
- targetNodeSourceTypeCf = new MultiTennantColumnFamily<>( edgeShardStrategy.getTargetNodeSourceTypeCfName(),
- new OrganizationScopedRowKeySerializer<>( ROW_TYPE_SERIALIZER ), EDGE_SERIALIZER );
-
- graphEdgeVersionsCf = new MultiTennantColumnFamily<>( edgeShardStrategy.getGraphEdgeVersions(),
- new OrganizationScopedRowKeySerializer<>( EDGE_ROW_KEY_SERIALIZER ), LONG_SERIALIZER );
+ this.edgeColumnFamilies = edgeColumnFamilies;
+ this.shardedEdgeSerialization = shardedEdgeSerialization;
}
@Override
public MutationBatch writeEdge( final ApplicationScope scope, final MarkedEdge markedEdge, final UUID timestamp ) {
- ValidationUtils.validateApplicationScope( scope );
- EdgeUtils.validateEdge( markedEdge );
- ValidationUtils.verifyTimeUuid( timestamp, "timestamp" );
-
-
- final MutationBatch batch = keyspace.prepareMutationBatch().withConsistencyLevel( cassandraConfig.getWriteCL() )
- .withTimestamp( timestamp.timestamp() );
-
- final boolean isDeleted = markedEdge.isDeleted();
-
-
- doWrite( scope, markedEdge, new RowOp<RowKey>() {
- @Override
- public void writeEdge( final MultiTennantColumnFamily<ApplicationScope, RowKey, DirectedEdge> columnFamily,
- final RowKey rowKey, final DirectedEdge edge ) {
- batch.withRow( columnFamily, ScopedRowKey.fromKey( scope, rowKey ) ).putColumn( edge, isDeleted );
- }
-
-
- @Override
- public void countEdge( final Id rowId, final long shardId, final String... types ) {
- if(!isDeleted){
- edgeShardStrategy.increment( scope, rowId, shardId, 1l, types );
- }
- }
-
-
- @Override
- public void writeVersion( final MultiTennantColumnFamily<ApplicationScope, EdgeRowKey, Long> columnFamily,
- final EdgeRowKey rowKey, final long timestamp ) {
- batch.withRow( columnFamily, ScopedRowKey.fromKey( scope, rowKey ) ).putColumn( timestamp, isDeleted );
- }
- } );
-
-
- return batch;
+ return shardedEdgeSerialization.writeEdge( edgeColumnFamilies, scope, markedEdge, timestamp );
}
@Override
public MutationBatch deleteEdge( final ApplicationScope scope, final MarkedEdge markedEdge, final UUID timestamp ) {
- ValidationUtils.validateApplicationScope( scope );
- EdgeUtils.validateEdge( markedEdge );
- ValidationUtils.verifyTimeUuid( timestamp, "timestamp" );
-
-
- final MutationBatch batch = keyspace.prepareMutationBatch().withConsistencyLevel( cassandraConfig.getWriteCL() )
- .withTimestamp( timestamp.timestamp() );
-
-
- doWrite( scope, markedEdge, new RowOp<RowKey>() {
- @Override
- public void writeEdge( final MultiTennantColumnFamily<ApplicationScope, RowKey, DirectedEdge> columnFamily,
- final RowKey rowKey, final DirectedEdge edge ) {
- batch.withRow( columnFamily, ScopedRowKey.fromKey( scope, rowKey ) ).deleteColumn( edge );
- }
-
-
- @Override
- public void countEdge( final Id rowId, final long shardId, final String... types ) {
- edgeShardStrategy.increment( scope, rowId, shardId, -1, types );
- }
-
-
- @Override
- public void writeVersion( final MultiTennantColumnFamily<ApplicationScope, EdgeRowKey, Long> columnFamily,
- final EdgeRowKey rowKey, final long timestamp ) {
- batch.withRow( columnFamily, ScopedRowKey.fromKey( scope, rowKey ) ).deleteColumn( timestamp );
- }
- } );
-
-
- return batch;
- }
-
-
- /**
- * EdgeWrite the edges internally
- *
- * @param scope The scope to encapsulate
- * @param edge The edge to write
- * @param op The row operation to invoke
- */
- private void doWrite( final ApplicationScope scope, final MarkedEdge edge, final RowOp op ) {
- ValidationUtils.validateApplicationScope( scope );
- EdgeUtils.validateEdge( edge );
-
- final Id sourceNodeId = edge.getSourceNode();
- final String souceNodeType = sourceNodeId.getType();
- final Id targetNodeId = edge.getTargetNode();
- final String targetNodeType = targetNodeId.getType();
- final long timestamp = edge.getTimestamp();
- final String type = edge.getType();
-
-
- /**
- * Key in the serializers based on the edge
- */
-
- final long sourceRowKeyShard = edgeShardStrategy.getWriteShard( scope, sourceNodeId, timestamp, type );
- final RowKey sourceRowKey = new RowKey( sourceNodeId, type, sourceRowKeyShard);
-
-
-
- final long sourceWithTypeRowKeyShard = edgeShardStrategy.getWriteShard( scope, sourceNodeId, timestamp, type, targetNodeType );
-
- final RowKeyType sourceRowKeyType = new RowKeyType( sourceNodeId, type, targetNodeId, sourceWithTypeRowKeyShard );
-
- final DirectedEdge sourceEdge = new DirectedEdge( targetNodeId, timestamp );
-
-
-
- final long targetRowKeyShard = edgeShardStrategy.getWriteShard( scope, targetNodeId, timestamp, type );
- final RowKey targetRowKey = new RowKey( targetNodeId, type, targetRowKeyShard);
-
- final long targetWithTypeRowKeyShard = edgeShardStrategy.getWriteShard( scope, targetNodeId, timestamp, type, souceNodeType );
- final RowKeyType targetRowKeyType = new RowKeyType( targetNodeId, type, sourceNodeId, targetWithTypeRowKeyShard );
-
- final DirectedEdge targetEdge = new DirectedEdge( sourceNodeId, timestamp );
-
-
- final EdgeRowKey edgeRowKey = new EdgeRowKey( sourceNodeId, type, targetNodeId, edgeShardStrategy
- .getWriteShard( scope, sourceNodeId, timestamp, type, targetNodeId.getUuid().toString(),
- targetNodeId.getType() ) );
-
-
- /**
- * write edges from source->target
- */
-
- op.writeEdge( sourceNodeEdgesCf, sourceRowKey, sourceEdge );
- op.countEdge( sourceNodeId, sourceRowKeyShard, type );
-
- op.writeEdge( sourceNodeTargetTypeCf, sourceRowKeyType, sourceEdge );
- op.countEdge( sourceNodeId, sourceWithTypeRowKeyShard, type, targetNodeType );
-
-
-
-
- /**
- * write edges from target<-source
- */
- op.writeEdge( targetNodeEdgesCf, targetRowKey, targetEdge );
- op.countEdge( targetNodeId, targetRowKeyShard, type );
-
- op.writeEdge( targetNodeSourceTypeCf, targetRowKeyType, targetEdge );
- op.countEdge( targetNodeId, targetWithTypeRowKeyShard, type, souceNodeType );
-
-
- /**
- * Write this in the timestamp log for this edge of source->target
- */
- op.writeVersion( graphEdgeVersionsCf, edgeRowKey, timestamp );
-
-
+ return shardedEdgeSerialization.deleteEdge( edgeColumnFamilies, scope, markedEdge, timestamp );
}
@@ -360,55 +109,42 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
final String type = search.getType();
final long maxTimestamp = search.getMaxTimestamp();
- final EdgeSearcher<EdgeRowKey, Long, MarkedEdge> searcher =
- new EdgeSearcher<EdgeRowKey, Long, MarkedEdge>( scope, maxTimestamp, search.last(),
- edgeShardStrategy.getReadShards( scope, sourceId, maxTimestamp, type ) ) {
-
- @Override
- protected Serializer<Long> getSerializer() {
- return LONG_SERIALIZER;
- }
-
-
- @Override
- public void setRange( final RangeBuilder builder ) {
-
-
- if ( last.isPresent() ) {
- super.setRange( builder );
- return;
- }
-
- //start seeking at a value < our max version
- builder.setStart( maxTimestamp );
- }
-
-
- @Override
- protected EdgeRowKey generateRowKey( long shard ) {
- return new EdgeRowKey( sourceId, type, targetId, shard );
- }
+ //Create our operator to perform seeks on the shard
+// final ShardOperator operator = new Edge(sourceId, NodeType.SOURCE, type) {
+//
+// @Override
+// public Iterator<MarkedEdge> getEdges( final ApplicationScope scope, final Shard shard, final long maxValue ) {
+//
+// final SearchByEdgeType search = new SimpleSearchByEdgeType( sourceId, type, maxTimestamp, null);
+//
+// return shardedEdgeSerialization.getEdgesFromSource( edgeColumnFamilies, scope, search,
+// createGroup( shard ) );
+// }
+// };
+//
+ final ShardOperator shardOperator = new ShardOperator() {
+ @Override
+ public Iterator<MarkedEdge> getEdges( final ApplicationScope scope, final Shard shard,
+ final long maxValue ) {
+ return null;
+ }
+ };
- @Override
- protected Long getStartColumn( final Edge last ) {
- return last.getTimestamp();
- }
+ final DirectedEdgeMeta versionMetaData = DirectedEdgeMeta.fromEdge( sourceId, targetId, type );
- @Override
- protected MarkedEdge createEdge( final Long column, final boolean marked ) {
- return new SimpleMarkedEdge( sourceId, type, targetId, column.longValue(), marked );
- }
- };
+ final Iterator<ShardEntryGroup> readShards =
+ edgeShardStrategy.getReadShards(scope, maxTimestamp, versionMetaData );
- return new ShardRowIterator<>( searcher, graphEdgeVersionsCf );
+ return shardedEdgeSerialization.getEdgeVersions( edgeColumnFamilies, scope, search, readShards );
}
@Override
public Iterator<MarkedEdge> getEdgesFromSource( final ApplicationScope scope, final SearchByEdgeType edgeType ) {
+
ValidationUtils.validateApplicationScope( scope );
EdgeUtils.validateSearchByEdgeType( edgeType );
@@ -416,36 +152,21 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
final String type = edgeType.getType();
final long maxTimestamp = edgeType.getMaxTimestamp();
- final EdgeSearcher<RowKey, DirectedEdge, MarkedEdge> searcher =
- new EdgeSearcher<RowKey, DirectedEdge, MarkedEdge>( scope, maxTimestamp, edgeType.last(),
- edgeShardStrategy.getReadShards( scope, sourceId, maxTimestamp, type ) ) {
-
-
- @Override
- protected Serializer<DirectedEdge> getSerializer() {
- return EDGE_SERIALIZER;
- }
-
-
- @Override
- protected RowKey generateRowKey( long shard ) {
- return new RowKey( sourceId, type, shard );
- }
-
+ final ShardOperator shardOperator = new ShardOperator() {
+ @Override
+ public Iterator<MarkedEdge> getEdges( final ApplicationScope scope, final Shard shard,
+ final long maxValue ) {
+ return null;
+ }
+ };
- @Override
- protected DirectedEdge getStartColumn( final Edge last ) {
- return new DirectedEdge( last.getTargetNode(), last.getTimestamp() );
- }
+ final DirectedEdgeMeta sourceMeta = DirectedEdgeMeta.fromSourceNode( sourceId, type );
- @Override
- protected MarkedEdge createEdge( final DirectedEdge edge, final boolean marked ) {
- return new SimpleMarkedEdge( sourceId, type, edge.id, edge.timestamp, marked );
- }
- };
+ final Iterator<ShardEntryGroup> readShards =
+ edgeShardStrategy.getReadShards(scope, maxTimestamp, sourceMeta );
- return new ShardRowIterator<>( searcher, sourceNodeEdgesCf );
+ return shardedEdgeSerialization.getEdgesFromSource( edgeColumnFamilies, scope, edgeType, readShards );
}
@@ -456,40 +177,42 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
ValidationUtils.validateApplicationScope( scope );
EdgeUtils.validateSearchByEdgeType( edgeType );
- final Id targetId = edgeType.getNode();
+ final Id sourceId = edgeType.getNode();
final String type = edgeType.getType();
final String targetType = edgeType.getIdType();
final long maxTimestamp = edgeType.getMaxTimestamp();
- final EdgeSearcher<RowKeyType, DirectedEdge, MarkedEdge> searcher =
- new EdgeSearcher<RowKeyType, DirectedEdge, MarkedEdge>( scope, maxTimestamp, edgeType.last(),
- edgeShardStrategy.getReadShards( scope, targetId, maxTimestamp, type, targetType ) ) {
-
- @Override
- protected Serializer<DirectedEdge> getSerializer() {
- return EDGE_SERIALIZER;
- }
+// //Create our operator to perform seeks on the shard
+// final ShardOperator operator = new EdgeByNodeTypeShardOperator(sourceId, NodeType.SOURCE, type, targetType) {
+//
+// @Override
+// public Iterator<MarkedEdge> getEdges( final ApplicationScope scope, final Shard shard, final long maxValue ) {
+//
+// final SearchByIdType search = new SimpleSearchByIdType( sourceId, type, maxTimestamp, targetType, null);
+//
+// return shardedEdgeSerialization.getEdgesFromSourceByTargetType( edgeColumnFamilies, scope, search, createGroup(shard));
+// }
+// };
+//
+//
+// final Iterator<ShardEntryGroup> readShards = edgeShardStrategy.getReadShards( scope, maxTimestamp, operator );
+ final ShardOperator shardOperator = new ShardOperator() {
+ @Override
+ public Iterator<MarkedEdge> getEdges( final ApplicationScope scope, final Shard shard,
+ final long maxValue ) {
+ return null;
+ }
+ };
- @Override
- protected RowKeyType generateRowKey( long shard ) {
- return new RowKeyType( targetId, type, targetType, shard );
- }
-
+ final DirectedEdgeMeta sourceMeta = DirectedEdgeMeta.fromSourceNodeTargetType( sourceId, type, targetType );
- @Override
- protected DirectedEdge getStartColumn( final Edge last ) {
- return new DirectedEdge( last.getTargetNode(), last.getTimestamp() );
- }
+ final Iterator<ShardEntryGroup> readShards =
+ edgeShardStrategy.getReadShards(scope, maxTimestamp, sourceMeta );
- @Override
- protected MarkedEdge createEdge( final DirectedEdge edge, final boolean marked ) {
- return new SimpleMarkedEdge( targetId, type, edge.id, edge.timestamp, marked );
- }
- };
- return new ShardRowIterator( searcher, sourceNodeTargetTypeCf );
+ return shardedEdgeSerialization.getEdgesFromSourceByTargetType( edgeColumnFamilies, scope, edgeType, readShards );
}
@@ -503,36 +226,39 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
final String type = edgeType.getType();
final long maxTimestamp = edgeType.getMaxTimestamp();
- final EdgeSearcher<RowKey, DirectedEdge, MarkedEdge> searcher =
- new EdgeSearcher<RowKey, DirectedEdge, MarkedEdge>( scope, maxTimestamp, edgeType.last(),
- edgeShardStrategy.getReadShards( scope, targetId, maxTimestamp, type ) ) {
-
- @Override
- protected Serializer<DirectedEdge> getSerializer() {
- return EDGE_SERIALIZER;
- }
+// //Create our operator to perform seeks on the shard
+// final ShardOperator operator = new EdgeByTypeShardOperator(targetId, NodeType.TARGET, type) {
+//
+// @Override
+// public Iterator<MarkedEdge> getEdges( final ApplicationScope scope, final Shard shard, final long maxValue ) {
+//
+// final SearchByEdgeType search = new SimpleSearchByEdgeType( targetId, type, maxTimestamp, null);
+//
+// return shardedEdgeSerialization.getEdgesToTarget( edgeColumnFamilies, scope, search,
+// createGroup( shard ) );
+// }
+// };
+//
+//
+// final Iterator<ShardEntryGroup> readShards = edgeShardStrategy.getReadShards( scope, maxTimestamp, operator);
- @Override
- protected RowKey generateRowKey( long shard ) {
- return new RowKey( targetId, type, shard );
- }
-
+ final ShardOperator shardOperator = new ShardOperator() {
+ @Override
+ public Iterator<MarkedEdge> getEdges( final ApplicationScope scope, final Shard shard,
+ final long maxValue ) {
+ return null;
+ }
+ };
- @Override
- protected DirectedEdge getStartColumn( final Edge last ) {
- return new DirectedEdge( last.getSourceNode(), last.getTimestamp() );
- }
+ final DirectedEdgeMeta sourceMeta = DirectedEdgeMeta.fromTargetNode( targetId, type );
- @Override
- protected MarkedEdge createEdge( final DirectedEdge edge, final boolean marked ) {
- return new SimpleMarkedEdge( edge.id, type, targetId, edge.timestamp, marked );
- }
- };
+ final Iterator<ShardEntryGroup> readShards =
+ edgeShardStrategy.getReadShards(scope, maxTimestamp, sourceMeta );
- return new ShardRowIterator<>( searcher, targetNodeEdgesCf );
+ return shardedEdgeSerialization.getEdgesToTarget( edgeColumnFamilies, scope, edgeType, readShards );
}
@@ -548,523 +274,38 @@ public class EdgeSerializationImpl implements EdgeSerialization, Migration {
final String type = edgeType.getType();
final long maxTimestamp = edgeType.getMaxTimestamp();
-
- final EdgeSearcher<RowKeyType, DirectedEdge, MarkedEdge> searcher =
- new EdgeSearcher<RowKeyType, DirectedEdge, MarkedEdge>( scope, maxTimestamp, edgeType.last(),
- edgeShardStrategy.getReadShards( scope, targetId, maxTimestamp, type, sourceType ) ) {
- @Override
- protected Serializer<DirectedEdge> getSerializer() {
- return EDGE_SERIALIZER;
- }
-
-
- @Override
- protected RowKeyType generateRowKey( final long shard ) {
- return new RowKeyType( targetId, type, sourceType, shard );
- }
-
-
- @Override
- protected DirectedEdge getStartColumn( final Edge last ) {
- return new DirectedEdge( last.getTargetNode(), last.getTimestamp() );
- }
-
-
- @Override
- protected MarkedEdge createEdge( final DirectedEdge edge, final boolean marked ) {
- return new SimpleMarkedEdge( edge.id, type, targetId, edge.timestamp, marked );
- }
- };
-
- return new ShardRowIterator<>( searcher, targetNodeSourceTypeCf );
- }
-
-
- @Override
- public Collection<MultiTennantColumnFamilyDefinition> getColumnFamilies() {
- return Arrays
- .asList( graphCf( sourceNodeEdgesCf ), graphCf( targetNodeEdgesCf ), graphCf( sourceNodeTargetTypeCf ),
- graphCf( targetNodeSourceTypeCf ),
- new MultiTennantColumnFamilyDefinition( graphEdgeVersionsCf, BytesType.class.getSimpleName(),
- ColumnTypes.LONG_TYPE_REVERSED, BytesType.class.getSimpleName(),
- MultiTennantColumnFamilyDefinition.CacheOption.KEYS ) );
- }
-
-
- /**
- * Helper to generate an edge definition by the type
- */
- private MultiTennantColumnFamilyDefinition graphCf( MultiTennantColumnFamily cf ) {
- return new MultiTennantColumnFamilyDefinition( cf, BytesType.class.getSimpleName(), EDGE_DYNAMIC_COMPOSITE_TYPE,
- BytesType.class.getSimpleName(), MultiTennantColumnFamilyDefinition.CacheOption.KEYS );
- }
-
-
- /**
- * Internal class to represent edge data for serialization
- */
- private static class DirectedEdge {
-
- public final long timestamp;
- public final Id id;
-
-
- private DirectedEdge( final Id id, final long timestamp ) {
- this.timestamp = timestamp;
- this.id = id;
- }
- }
-
-
- /**
- * Serializes to a source->target edge Note that we cannot set the edge type on de-serialization. Only the target
- * Id and version.
- */
- private static class EdgeSerializer extends AbstractSerializer<DirectedEdge> {
-
- private static final IdColDynamicCompositeSerializer ID_COL_SERIALIZER = IdColDynamicCompositeSerializer.get();
-// private static final UUIDSerializer UUID_SERIALIZER = UUIDSerializer.get();
-// private static final StringSerializer STRING_SERIALIZER = StringSerializer.get().getString(;
-// )
-
-
- @Override
- public ByteBuffer toByteBuffer( final DirectedEdge edge ) {
-
- DynamicComposite composite = new DynamicComposite();
-
-// //add our edge
-// composite.addComponent( edge.timestamp, LONG_SERIALIZER, LONG_TYPE_REVERSED,
-// AbstractComposite.ComponentEquality.EQUAL );
-
-// //we do this explicity instead of re-using the id composite serializer b/c we want high order
-// //time uuids first, not second. In this column family, there is no sort
-// composite.addComponent( edge.id.getUuid(), UUID_SERIALIZER, UUID_TYPE_REVERSED,
-// AbstractComposite.ComponentEquality.EQUAL );
+// //Create our operator to perform seeks on the shard
+// final ShardOperator operator = new EdgeByNodeTypeShardOperator(targetId, NodeType.TARGET, type, sourceType) {
//
-// composite.addComponent( edge.id.getType(), STRING_SERIALIZER );
-
- composite.addComponent( edge.timestamp, LONG_SERIALIZER );
-
- ID_COL_SERIALIZER.toComposite( composite, edge.id);
-
- return composite.serialize();
- }
-
-
- @Override
- public DirectedEdge fromByteBuffer( final ByteBuffer byteBuffer ) {
- DynamicComposite composite = DynamicComposite.fromByteBuffer( byteBuffer );
-
- Preconditions.checkArgument( composite.size() == 3, "Composite should have 3 elements" );
-
-
- //return the version
- final long timestamp = composite.get( 0, LONG_SERIALIZER );
-
-
- //parse our id
- final Id id = ID_COL_SERIALIZER.fromComposite( composite, 1 );
-
-
- return new DirectedEdge( id, timestamp );
- }
- }
-
-
- /**
- * Class that represents an edge row key
- */
- private static class RowKey {
- public final Id nodeId;
- public final long[] hash;
- public final long shard;
-
-
- /**
- * Create a row key with the node and the edgeType
- */
- public RowKey( Id nodeId, String edgeType, final long shard ) {
- this( nodeId, EdgeHasher.createEdgeHash( edgeType ), shard );
- }
-
-
- /**
- * Create a new row key with the hash, should only be used in deserialization or internal callers.
- */
- protected RowKey( Id nodeId, long[] hash, final long shard ) {
- this.nodeId = nodeId;
- this.hash = hash;
- this.shard = shard;
- }
- }
-
-
- /**
- * The row key with the additional type
- */
- private static class RowKeyType extends RowKey {
-
- /**
- * Create a row key with the node id in the row key, the edge type, and the type from the typeid
- *
- * @param nodeId The node id in the row key
- * @param edgeType The type of the edge
- * @param typeId The type of hte id
- */
- public RowKeyType( final Id nodeId, final String edgeType, final Id typeId, final long shard ) {
- this( nodeId, edgeType, typeId.getType(), shard );
- }
-
-
- /**
- * Create a row key with the node id in the row key, the edge type, adn the target type from the id
- */
- public RowKeyType( final Id nodeId, final String edgeType, final String targetType, final long shard ) {
- super( nodeId, EdgeHasher.createEdgeHash( edgeType, targetType ), shard );
- }
-
-
- /**
- * Internal use in de-serializing. Should only be used in this case or by internal callers
- */
- private RowKeyType( final Id nodeId, final long[] hash, final long shard ) {
- super( nodeId, hash, shard );
- }
- }
-
-
- /**
- * Used to store row keys by sourceId, targetId and edgeType
- */
- private static class EdgeRowKey {
- public final Id sourceId;
- public final Id targetId;
- public final String edgeType;
- public final long shardId;
-
-
- private EdgeRowKey( final Id sourceId, final String edgeType, final Id targetId, final long shardId ) {
- this.sourceId = sourceId;
- this.targetId = targetId;
- this.edgeType = edgeType;
- this.shardId = shardId;
- }
- }
-
-
- /**
- * Searcher to be used when performing the search. Performs I/O transformation as well as parsing for the iterator.
- * If there are more row keys available to seek, the iterator will return true
- *
- * @param <R> The row type
- * @param <C> The column type
- * @param <T> The parsed return type
- */
- private static abstract class EdgeSearcher<R, C, T>
- implements ColumnParser<C, T>, Iterator<ScopedRowKey<ApplicationScope, R>> {
-
- protected final Optional<Edge> last;
- protected final long maxTimestamp;
- protected final ApplicationScope scope;
- protected final Iterator<Long> shards;
-
-
- protected EdgeSearcher( final ApplicationScope scope, final long maxTimestamp, final Optional<Edge> last,
- final Iterator<Long> shards ) {
- this.scope = scope;
- this.maxTimestamp = maxTimestamp;
- this.last = last;
- this.shards = shards;
- }
-
-
- @Override
- public boolean hasNext() {
- return shards.hasNext();
- }
-
-
- @Override
- public ScopedRowKey<ApplicationScope, R> next() {
- return ScopedRowKey.fromKey( scope, generateRowKey( shards.next() ) );
- }
-
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException( "Remove is unsupported" );
- }
-
-
- /**
- * Set the range on a search
- */
- public void setRange( final RangeBuilder builder ) {
-
- //set our start range since it was supplied to us
- if ( last.isPresent() ) {
- C sourceEdge = getStartColumn( last.get() );
-
-
- builder.setStart( sourceEdge, getSerializer() );
- }
- else {
-
-
+// @Override
+// public Iterator<MarkedEdge> getEdges( final ApplicationScope scope, final Shard shard, final long maxValue ) {
+//
+// final SearchByIdType search = new SimpleSearchByIdType( targetId, type, maxTimestamp, sourceType, null);
+//
+// return shardedEdgeSerialization.getEdgesFromSourceByTargetType( edgeColumnFamilies, scope, search, createGroup(shard));
+// }
+// };
+//
+//
+//
+// Iterator<ShardEntryGroup> readShards = edgeShardStrategy
+// .getReadShards( scope, maxTimestamp, operator );
+ final ShardOperator shardOperator = new ShardOperator() {
+ @Override
+ public Iterator<MarkedEdge> getEdges( final ApplicationScope scope, final Shard shard,
+ final long maxValue ) {
+ return null;
}
- }
-
-
- public boolean hasPage() {
- return last.isPresent();
- }
-
+ };
- @Override
- public T parseColumn( final Column<C> column ) {
- final C edge = column.getName();
+ final DirectedEdgeMeta sourceMeta = DirectedEdgeMeta.fromSourceNodeTargetType( targetId, type, sourceType );
- return createEdge( edge, column.getBooleanValue() );
- }
+ final Iterator<ShardEntryGroup> readShards =
+ edgeShardStrategy.getReadShards(scope, maxTimestamp, sourceMeta );
- /**
- * Get the column's serializer
- */
- protected abstract Serializer<C> getSerializer();
-
- /**
- * Create a row key for this search to use
- *
- * @param shard The shard to use in the row key
- */
- protected abstract R generateRowKey( final long shard );
-
-
- /**
- * Set the start column to begin searching from. The last is provided
- */
- protected abstract C getStartColumn( final Edge last );
-
-
- /**
- * Create an edge to return to the user based on the directed edge provided
- *
- * @param column The column name
- * @param marked The marked flag in the column value
- */
- protected abstract T createEdge( final C column, final boolean marked );
- }
-
-
- /**
- * Class to perform serialization for row keys from edges
- */
- private static class RowSerializer implements CompositeFieldSerializer<RowKey> {
-
- private static final IdRowCompositeSerializer ID_SER = IdRowCompositeSerializer.get();
-
-
- @Override
- public void toComposite( final CompositeBuilder builder, final RowKey key ) {
-
- //add the row id to the composite
- ID_SER.toComposite( builder, key.nodeId );
-
- builder.addLong( key.hash[0] );
- builder.addLong( key.hash[1] );
- builder.addLong( key.shard );
- }
-
-
- @Override
- public RowKey fromComposite( final CompositeParser composite ) {
-
- final Id id = ID_SER.fromComposite( composite );
- final long[] hash = new long[] { composite.readLong(), composite.readLong() };
- final long shard = composite.readLong();
-
-
- return new RowKey( id, hash, shard );
- }
- }
-
-
- /**
- * Class to perform serialization for row keys from edges
- */
- private static class RowTypeSerializer implements CompositeFieldSerializer<RowKeyType> {
-
- private static final IdRowCompositeSerializer ID_SER = IdRowCompositeSerializer.get();
-
-
- @Override
- public void toComposite( final CompositeBuilder builder, final RowKeyType keyType ) {
-
- //add the row id to the composite
- ID_SER.toComposite( builder, keyType.nodeId );
-
- builder.addLong( keyType.hash[0] );
- builder.addLong( keyType.hash[1] );
- builder.addLong( keyType.shard );
- }
-
-
- @Override
- public RowKeyType fromComposite( final CompositeParser composite ) {
-
- final Id id = ID_SER.fromComposite( composite );
- final long[] hash = new long[] { composite.readLong(), composite.readLong() };
- final long shard = composite.readLong();
-
- return new RowKeyType( id, hash, shard );
- }
- }
-
-
- /**
- * Class to perform serialization for row keys from edges
- */
- private static class EdgeRowKeySerializer implements CompositeFieldSerializer<EdgeRowKey> {
-
- private static final IdRowCompositeSerializer ID_SER = IdRowCompositeSerializer.get();
-
-
- @Override
- public void toComposite( final CompositeBuilder builder, final EdgeRowKey key ) {
-
- //add the row id to the composite
- ID_SER.toComposite( builder, key.sourceId );
- builder.addString( key.edgeType );
- ID_SER.toComposite( builder, key.targetId );
- builder.addLong( key.shardId );
- }
-
-
- @Override
- public EdgeRowKey fromComposite( final CompositeParser composite ) {
-
- final Id sourceId = ID_SER.fromComposite( composite );
- final String edgeType = composite.readString();
- final Id targetId = ID_SER.fromComposite( composite );
- final long shard = composite.readLong();
-
- return new EdgeRowKey( sourceId, edgeType, targetId, shard );
- }
- }
-
-
- /**
- * Simple callback to perform puts and deletes with a common row setup code
- *
- * @param <R> The row key type
- */
- private static interface RowOp<R> {
-
- /**
- * Write the edge with the given data
- */
- void writeEdge( final MultiTennantColumnFamily<ApplicationScope, R, DirectedEdge> columnFamily, R rowKey,
- DirectedEdge edge );
-
- /**
- * Perform the count on the edge
- */
- void countEdge( final Id rowId, long shardId, String... types );
-
- /**
- * Write the edge into the version cf
- */
- void writeVersion( final MultiTennantColumnFamily<ApplicationScope, EdgeRowKey, Long> columnFamily,
- EdgeRowKey rowKey, long timestamp );
+ return shardedEdgeSerialization.getEdgesToTargetBySourceType( edgeColumnFamilies, scope, edgeType, readShards );
}
-
- /**
- * Internal iterator to iterate over multiple row keys
- *
- * @param <R> The row type
- * @param <C> The column type
- * @param <T> The parsed return type
- */
- private class ShardRowIterator<R, C, T> implements Iterator<T> {
-
- private final EdgeSearcher<R, C, T> searcher;
-
- private final MultiTennantColumnFamily<ApplicationScope, R, C> cf;
-
- private Iterator<T> currentColumnIterator;
-
-
- private ShardRowIterator( final EdgeSearcher<R, C, T> searcher,
- final MultiTennantColumnFamily<ApplicationScope, R, C> cf ) {
- this.searcher = searcher;
- this.cf = cf;
- }
-
-
- @Override
- public boolean hasNext() {
- //we have more columns to return
- if ( currentColumnIterator != null && currentColumnIterator.hasNext() ) {
- return true;
- }
-
- /**
- * We have another row key, advance to it and re-check
- */
- if ( searcher.hasNext() ) {
- advanceRow();
- return hasNext();
- }
-
- //we have no more columns, and no more row keys, we're done
- return false;
- }
-
-
- @Override
- public T next() {
- if ( !hasNext() ) {
- throw new NoSuchElementException( "There are no more rows or columns left to advance" );
- }
-
- return currentColumnIterator.next();
- }
-
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException( "Remove is unsupported" );
- }
-
-
- /**
- * Advance our iterator to the next row (assumes the check for row keys is elsewhere)
- */
- private void advanceRow() {
-
- /**
- * If the edge is present, we need to being seeking from this
- */
-
- final RangeBuilder rangeBuilder = new RangeBuilder().setLimit( graphFig.getScanPageSize() );
-
-
- //set the range into the search
- searcher.setRange( rangeBuilder );
-
- final ScopedRowKey<ApplicationScope, R> rowKey = searcher.next();
-
-
- RowQuery<ScopedRowKey<ApplicationScope, R>, C> query =
- keyspace.prepareQuery( cf ).setConsistencyLevel( cassandraConfig.getReadCL() ).getKey( rowKey )
- .autoPaginate( true ).withColumnRange( rangeBuilder.build() );
-
-
- currentColumnIterator = new ColumnNameIterator<C, T>( query, searcher, searcher.hasPage() );
- }
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e9d652dd/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 5be988b..a5ca964 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
@@ -168,20 +168,20 @@ public class NodeSerializationImpl implements NodeSerialization, Migration {
@Override
- public Map<Id, Long> getMaxVersions( final ApplicationScope scope, final Collection<? extends Edge> nodeIds ) {
+ public Map<Id, Long> getMaxVersions( final ApplicationScope scope, final Collection<? extends Edge> edges ) {
ValidationUtils.validateApplicationScope( scope );
- Preconditions.checkNotNull( nodeIds, "nodeIds cannot be null" );
+ Preconditions.checkNotNull( edges, "edges cannot be null" );
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>>(nodeIds.size());
+ final List<ScopedRowKey<ApplicationScope, Id>> keys = new ArrayList<ScopedRowKey<ApplicationScope, Id>>(edges.size());
//worst case all are marked
- final Map<Id, Long> versions = new HashMap<>(nodeIds.size());
+ final Map<Id, Long> versions = new HashMap<>(edges.size());
- for(final Edge edge: nodeIds){
+ for(final Edge edge: edges){
keys.add( ScopedRowKey.fromKey( scope, edge.getSourceNode() ) );
keys.add( ScopedRowKey.fromKey( scope, edge.getTargetNode() ) );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e9d652dd/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/DirectedEdge.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/DirectedEdge.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/DirectedEdge.java
new file mode 100644
index 0000000..720c948
--- /dev/null
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/DirectedEdge.java
@@ -0,0 +1,41 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one
+ * * or more contributor license agreements. See the NOTICE file
+ * * distributed with this work for additional information
+ * * regarding copyright ownership. The ASF licenses this file
+ * * to you under the Apache License, Version 2.0 (the
+ * * "License"); you may not use this file except in compliance
+ * * with the License. You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing,
+ * * software distributed under the License is distributed on an
+ * * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * * KIND, either express or implied. See the License for the
+ * * specific language governing permissions and limitations
+ * * under the License.
+ *
+ */
+
+package org.apache.usergrid.persistence.graph.serialization.impl.shard;
+
+
+import org.apache.usergrid.persistence.model.entity.Id;
+
+
+/**
+ * Internal class to represent edge data for serialization
+ */
+public class DirectedEdge {
+
+ public final long timestamp;
+ public final Id id;
+
+
+ public DirectedEdge( final Id id, final long timestamp ) {
+ this.timestamp = timestamp;
+ this.id = id;
+ }
+}