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 2015/09/11 03:24:43 UTC
[4/5] usergrid git commit: Implements immediate consistency in graph
shard allocation
http://git-wip-us.apache.org/repos/asf/usergrid/blob/43a2b299/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..cc8dceb 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
@@ -20,12 +20,10 @@ package org.apache.usergrid.persistence.graph.serialization.impl.shard;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,10 +43,6 @@ public class ShardEntryGroup {
private List<Shard> shards;
- private final long delta;
-
- private long maxCreatedTime;
-
private Shard compactionTarget;
private Shard rootShard;
@@ -57,15 +51,14 @@ public class ShardEntryGroup {
/**
* The max delta we accept in milliseconds for create time to be considered a member of this group
*/
- public ShardEntryGroup( final long delta ) {
- Preconditions.checkArgument( delta > 0, "delta must be greater than 0" );
- this.delta = delta;
+ public ShardEntryGroup() {
this.shards = new ArrayList<>();
- this.maxCreatedTime = 0;
}
/**
+ * Shard insertion order is assumed to be from Shard.shardIndex == Long.MAX to Shard.shardIndex == Long.MIN
+ *
* Only add a shard if it is within the rules require to meet a group. The rules are outlined below.
*
* Case 1) First shard in the group, always added
@@ -98,8 +91,6 @@ public class ShardEntryGroup {
}
-
-
return false;
}
@@ -110,15 +101,13 @@ public class ShardEntryGroup {
private void addShardInternal( final Shard shard ) {
shards.add( shard );
- maxCreatedTime = Math.max( maxCreatedTime, shard.getCreatedTime() );
-
//we're changing our structure, unset the compaction target
compactionTarget = null;
}
/**
- * Return the minum shard based on time indexes
+ * Return the minimum shard based on time indexes
*/
public Shard getMinShard() {
final int size = shards.size();
@@ -132,6 +121,20 @@ public class ShardEntryGroup {
/**
+ * Get the max shard based on time indexes
+ */
+ public Shard getMaxShard() {
+ final int size = shards.size();
+
+ if ( size < 1 ) {
+ return null;
+ }
+
+ return shards.get( size - 1 );
+ }
+
+
+ /**
* Get the entries that we should read from.
*/
public Collection<Shard> getReadShards() {
@@ -141,46 +144,49 @@ public class ShardEntryGroup {
final Shard compactionTarget = getCompactionTarget();
-
- if(compactionTarget != null){
+ if ( compactionTarget != null ) {
LOG.debug( "Returning shards {} and {} as read shards", compactionTarget, staticShard );
- return Arrays.asList( compactionTarget, staticShard );
+ //if we have a compaction target, we need to read from all shards to ensure we're aggregating data correctly
+ return shards;
}
LOG.debug( "Returning shards {} read shard", staticShard );
- return Collections.singleton( staticShard );
+ return Collections.singleton( staticShard );
}
-
-
/**
- * Get the entries, with the max shard time being first. We write to all shards until they're migrated
+ * Get the entries, with the earliest allocated uncompacted shard being first
*/
- public Collection<Shard> getWriteShards( long currentTime ) {
+ public Collection<Shard> getWriteShards( final long edgeIndex ) {
/**
* The shards in this set can be combined, we should only write to the compaction target to avoid
* adding data to other shards
*/
- if ( !isTooSmallToCompact() && shouldCompact( currentTime ) ) {
+ if ( !isTooSmallToCompact() && shouldCompact() ) {
final Shard compactionTarget = getCompactionTarget();
- LOG.debug( "Returning shard {} as write shard", compactionTarget);
+ LOG.debug( "Returning shard {} as write shard", compactionTarget );
- return Collections.singleton( compactionTarget );
+ //should go into the compaction target, it's a <= the edge value of the compaction
+ if ( compactionTarget.getShardIndex() <= edgeIndex ) {
+ return Collections.singleton( compactionTarget );
+ }
+
+ //otherwise the edge should go into the root shard
}
+
final Shard staticShard = getRootShard();
- LOG.debug( "Returning shard {} as write shard", staticShard);
+ LOG.debug( "Returning shard {} as write shard", staticShard );
return Collections.singleton( staticShard );
-
}
@@ -194,22 +200,22 @@ public class ShardEntryGroup {
/**
* Get the root shard that was created in this group
- * @return
*/
- private Shard getRootShard(){
- if(rootShard != null){
+ private Shard getRootShard() {
+ if ( rootShard != null ) {
return rootShard;
}
- final Shard rootCandidate = shards.get( shards.size() -1 );
+ final Shard rootCandidate = shards.get( shards.size() - 1 );
- if(rootCandidate.isCompacted()){
+ if ( rootCandidate.isCompacted() ) {
rootShard = rootCandidate;
}
return rootShard;
}
+
/**
* Get the shard all compactions should write to. Null indicates we cannot find a shard that could be used as a
* compaction target. Note that this shard may not have surpassed the delta yet You should invoke "shouldCompact"
@@ -238,19 +244,24 @@ public class ShardEntryGroup {
return null;
}
- //Start seeking from the end of our group. The first shard we encounter that is not compacted is our
- // compaction target
+ Shard compactionCandidate = null;
+
+ //Start seeking from the end of our group. The lowest timestamp uncompacted shard is our target
//NOTE: This does not mean we can compact, rather it's just an indication that we have a target set.
- for ( int i = lastIndex - 1; i > -1; i-- ) {
- final Shard compactionCandidate = shards.get( i );
+ for ( int i = 0; i < lastIndex; i++ ) {
+ final Shard currentTargetCompaction = shards.get( i );
- if ( !compactionCandidate.isCompacted() ) {
- compactionTarget = compactionCandidate;
- break;
+ //the shard is not compacted, and we've either never set a candidate
+ //or the candidate has a higher created timestamp than our current shard
+ if ( !currentTargetCompaction.isCompacted() && ( compactionCandidate == null
+ || currentTargetCompaction.getCreatedTime() < compactionCandidate.getCreatedTime() ) ) {
+ compactionCandidate = currentTargetCompaction;
}
}
+ compactionTarget = compactionCandidate;
+
return compactionTarget;
}
@@ -274,24 +285,14 @@ public class ShardEntryGroup {
/**
* Returns true if the newest created shard is path the currentTime - delta
*
- * @param currentTime The current system time in milliseconds
- *
* @return True if these shards can safely be combined into a single shard, false otherwise
*/
- public boolean shouldCompact( final long currentTime ) {
+ public boolean shouldCompact() {
/**
* We don't have enough shards to compact, ignore
*/
- return getCompactionTarget() != null
-
-
- /**
- * If something was created within the delta time frame, not everyone may have seen it due to
- * cache refresh, we can't compact yet.
- */
-
- && currentTime - delta > maxCreatedTime;
+ return getCompactionTarget() != null;
}
@@ -308,17 +309,36 @@ public class ShardEntryGroup {
return !shard.isCompacted() && ( compactionTarget != null && compactionTarget.getShardIndex() != shard
- .getShardIndex() );
+ .getShardIndex() );
+ }
+
+
+ @Override
+ public boolean equals( final Object o ) {
+ if ( this == o ) {
+ return true;
+ }
+ if ( !( o instanceof ShardEntryGroup ) ) {
+ return false;
+ }
+
+ final ShardEntryGroup that = ( ShardEntryGroup ) o;
+
+ return shards.equals( that.shards );
+ }
+
+
+ @Override
+ public int hashCode() {
+ return shards.hashCode();
}
@Override
public String toString() {
return "ShardEntryGroup{" +
- "shards=" + shards +
- ", delta=" + delta +
- ", maxCreatedTime=" + maxCreatedTime +
- ", compactionTarget=" + compactionTarget +
- '}';
+ "shards=" + shards +
+ ", compactionTarget=" + compactionTarget +
+ '}';
}
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/43a2b299/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardGroupCompaction.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardGroupCompaction.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardGroupCompaction.java
index 4fe1a63..f54bd55 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardGroupCompaction.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardGroupCompaction.java
@@ -42,12 +42,11 @@ public interface ShardGroupCompaction {
*
* @return A ListenableFuture with the result. Note that some
*/
- public ListenableFuture<AuditResult> evaluateShardGroup( final ApplicationScope scope,
- final DirectedEdgeMeta edgeMeta,
- final ShardEntryGroup group );
+ ListenableFuture<AuditResult> evaluateShardGroup( final ApplicationScope scope, final DirectedEdgeMeta edgeMeta,
+ final ShardEntryGroup group );
- public enum AuditResult {
+ enum AuditResult {
/**
* We didn't check this shard
*/
http://git-wip-us.apache.org/repos/asf/usergrid/blob/43a2b299/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardedEdgeSerializationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardedEdgeSerializationImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardedEdgeSerializationImpl.java
new file mode 100644
index 0000000..3329ff1
--- /dev/null
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/ShardedEdgeSerializationImpl.java
@@ -0,0 +1,1008 @@
+/*
+ *
+ * * 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 java.util.Collection;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.UUID;
+
+import javax.annotation.Nullable;
+import javax.inject.Inject;
+
+import org.apache.usergrid.persistence.core.astyanax.CassandraConfig;
+import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamily;
+import org.apache.usergrid.persistence.core.astyanax.ScopedRowKey;
+import org.apache.usergrid.persistence.core.consistency.TimeService;
+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.impl.shard.DirectedEdge;
+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.EdgeRowKey;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.EdgeShardStrategy;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.RowKey;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.RowKeyType;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.Shard;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.ShardedEdgeSerialization;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.EdgeSearcher;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.ShardsColumnIterator;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.comparators.DescendingTimestampComparator;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.comparators.OrderedComparator;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.comparators
+ .SourceDirectedEdgeDescendingComparator;
+import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.comparators
+ .TargetDirectedEdgeDescendingComparator;
+import org.apache.usergrid.persistence.graph.serialization.util.GraphValidation;
+import org.apache.usergrid.persistence.model.entity.Id;
+
+import com.google.common.base.Function;
+import com.google.common.base.Optional;
+import com.google.inject.Singleton;
+import com.netflix.astyanax.Keyspace;
+import com.netflix.astyanax.MutationBatch;
+import com.netflix.astyanax.Serializer;
+import com.netflix.astyanax.util.RangeBuilder;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+
+/**
+ * TODO: Rafactor this to use shards only, no shard groups, just collections of shards. The parent caller can aggregate
+ * the results of multiple groups together, this has an impedance mismatch in the API layer.
+ */
+@Singleton
+public class ShardedEdgeSerializationImpl implements ShardedEdgeSerialization {
+
+ protected final Keyspace keyspace;
+ protected final CassandraConfig cassandraConfig;
+ protected final GraphFig graphFig;
+ protected final EdgeShardStrategy writeEdgeShardStrategy;
+ protected final TimeService timeService;
+
+
+ @Inject
+ public ShardedEdgeSerializationImpl( final Keyspace keyspace, final CassandraConfig cassandraConfig,
+ final GraphFig graphFig, final EdgeShardStrategy writeEdgeShardStrategy,
+ final TimeService timeService ) {
+
+
+ checkNotNull( "keyspace required", keyspace );
+ checkNotNull( "cassandraConfig required", cassandraConfig );
+ checkNotNull( "consistencyFig required", graphFig );
+ checkNotNull( "writeEdgeShardStrategy required", writeEdgeShardStrategy );
+ checkNotNull( "timeService required", timeService );
+
+
+ this.keyspace = keyspace;
+ this.cassandraConfig = cassandraConfig;
+ this.graphFig = graphFig;
+ this.writeEdgeShardStrategy = writeEdgeShardStrategy;
+ this.timeService = timeService;
+ }
+
+
+ @Override
+ public MutationBatch writeEdgeFromSource( final EdgeColumnFamilies columnFamilies, final ApplicationScope scope,
+ final MarkedEdge markedEdge, final Collection<Shard> shards,
+ final DirectedEdgeMeta directedEdgeMeta, final UUID timestamp ) {
+ ValidationUtils.validateApplicationScope( scope );
+ GraphValidation.validateEdge( markedEdge );
+ ValidationUtils.verifyTimeUuid( timestamp, "timestamp" );
+
+ return new SourceWriteOp( columnFamilies, markedEdge ) {
+
+ @Override
+ void writeEdge( final MutationBatch batch,
+ final MultiTennantColumnFamily<ScopedRowKey<RowKey>, DirectedEdge> columnFamily,
+ final ApplicationScope scope, final RowKey rowKey, final DirectedEdge edge,
+ final Shard shard, final boolean isDeleted ) {
+
+ batch.withRow( columnFamily, ScopedRowKey.fromKey( scope.getApplication(), rowKey ) ).putColumn( edge, isDeleted );
+
+ if ( !isDeleted ) {
+ writeEdgeShardStrategy.increment( scope, shard, 1, directedEdgeMeta );
+ }
+ }
+ }.createBatch( scope, shards, timestamp );
+ }
+
+
+ @Override
+ public MutationBatch writeEdgeFromSourceWithTargetType( final EdgeColumnFamilies columnFamilies,
+ final ApplicationScope scope, final MarkedEdge markedEdge,
+ final Collection<Shard> shards,
+ final DirectedEdgeMeta directedEdgeMeta,
+ final UUID timestamp ) {
+ ValidationUtils.validateApplicationScope( scope );
+ GraphValidation.validateEdge( markedEdge );
+ ValidationUtils.verifyTimeUuid( timestamp, "timestamp" );
+
+
+ return new SourceTargetTypeWriteOp( columnFamilies, markedEdge ) {
+
+ @Override
+ void writeEdge( final MutationBatch batch,
+ final MultiTennantColumnFamily<ScopedRowKey<RowKeyType>, DirectedEdge> columnFamily,
+ final ApplicationScope scope, final RowKeyType rowKey, final DirectedEdge edge,
+ final Shard shard, final boolean isDeleted ) {
+
+
+ batch.withRow( columnFamily, ScopedRowKey.fromKey( scope.getApplication(), rowKey ) ).putColumn( edge, isDeleted );
+
+
+ if ( !isDeleted ) {
+ writeEdgeShardStrategy.increment( scope, shard, 1, directedEdgeMeta );
+ }
+ }
+ }.createBatch( scope, shards, timestamp );
+ }
+
+
+ @Override
+ public MutationBatch writeEdgeToTarget( final EdgeColumnFamilies columnFamilies, final ApplicationScope scope,
+ final MarkedEdge markedEdge, final Collection<Shard> shards,
+ final DirectedEdgeMeta targetEdgeMeta, final UUID timestamp ) {
+ ValidationUtils.validateApplicationScope( scope );
+ GraphValidation.validateEdge( markedEdge );
+ ValidationUtils.verifyTimeUuid( timestamp, "timestamp" );
+
+
+ return new TargetWriteOp( columnFamilies, markedEdge ) {
+
+ @Override
+ void writeEdge( final MutationBatch batch,
+ final MultiTennantColumnFamily<ScopedRowKey<RowKey>, DirectedEdge> columnFamily,
+ final ApplicationScope scope, final RowKey rowKey, final DirectedEdge edge,
+ final Shard shard, final boolean isDeleted ) {
+
+ batch.withRow( columnFamily, ScopedRowKey.fromKey( scope.getApplication(), rowKey ) ).putColumn( edge, isDeleted );
+
+
+ if ( !isDeleted ) {
+ writeEdgeShardStrategy.increment( scope, shard, 1, targetEdgeMeta );
+ }
+ }
+ }.createBatch( scope, shards, timestamp );
+ }
+
+
+ @Override
+ public MutationBatch writeEdgeToTargetWithSourceType( final EdgeColumnFamilies columnFamilies,
+ final ApplicationScope scope, final MarkedEdge markedEdge,
+ final Collection<Shard> shards,
+ final DirectedEdgeMeta directedEdgeMeta,
+ final UUID timestamp ) {
+ ValidationUtils.validateApplicationScope( scope );
+ GraphValidation.validateEdge( markedEdge );
+ ValidationUtils.verifyTimeUuid( timestamp, "timestamp" );
+
+
+ return new TargetSourceTypeWriteOp( columnFamilies, markedEdge ) {
+
+ @Override
+ void writeEdge( final MutationBatch batch,
+ final MultiTennantColumnFamily<ScopedRowKey<RowKeyType>, DirectedEdge> columnFamily,
+ final ApplicationScope scope, final RowKeyType rowKey, final DirectedEdge edge,
+ final Shard shard, final boolean isDeleted ) {
+
+ batch.withRow( columnFamilies.getTargetNodeSourceTypeCfName(), ScopedRowKey.fromKey( scope.getApplication(), rowKey ) )
+ .putColumn( edge, isDeleted );
+
+
+ if ( !isDeleted ) {
+ writeEdgeShardStrategy.increment( scope, shard, 1, directedEdgeMeta );
+ }
+ }
+ }.createBatch( scope, shards, timestamp );
+ }
+
+
+ @Override
+ public MutationBatch writeEdgeVersions( final EdgeColumnFamilies columnFamilies, final ApplicationScope scope,
+ final MarkedEdge markedEdge, final Collection<Shard> shards,
+ final DirectedEdgeMeta directedEdgeMeta, final UUID timestamp ) {
+
+ ValidationUtils.validateApplicationScope( scope );
+ GraphValidation.validateEdge( markedEdge );
+ ValidationUtils.verifyTimeUuid( timestamp, "timestamp" );
+
+
+ return new EdgeVersions( columnFamilies, markedEdge ) {
+
+ @Override
+ void writeEdge( final MutationBatch batch,
+ final MultiTennantColumnFamily<ScopedRowKey<EdgeRowKey>, Long> columnFamily,
+ final ApplicationScope scope, final EdgeRowKey rowKey, final Long column, final Shard shard,
+ final boolean isDeleted ) {
+ batch.withRow( columnFamilies.getGraphEdgeVersions(), ScopedRowKey.fromKey( scope.getApplication(), rowKey ) )
+ .putColumn( column, isDeleted );
+
+
+ if ( !isDeleted ) {
+ writeEdgeShardStrategy.increment( scope, shard, 1, directedEdgeMeta );
+ }
+ }
+ }.createBatch( scope, shards, timestamp );
+ }
+
+
+ @Override
+ public MutationBatch deleteEdgeFromSource( final EdgeColumnFamilies columnFamilies, final ApplicationScope scope,
+ final MarkedEdge markedEdge, final Collection<Shard> shards,
+ final DirectedEdgeMeta directedEdgeMeta, final UUID timestamp ) {
+
+ return new SourceWriteOp( columnFamilies, markedEdge ) {
+
+ @Override
+ void writeEdge( final MutationBatch batch,
+ final MultiTennantColumnFamily<ScopedRowKey<RowKey>, DirectedEdge> columnFamily,
+ final ApplicationScope scope, final RowKey rowKey, final DirectedEdge edge,
+ final Shard shard, final boolean isDeleted ) {
+
+ batch.withRow( columnFamily, ScopedRowKey.fromKey( scope.getApplication(), rowKey ) ).deleteColumn( edge );
+ writeEdgeShardStrategy.increment( scope, shard, -1, directedEdgeMeta );
+ }
+ }.createBatch( scope, shards, timestamp );
+ }
+
+
+ @Override
+ public MutationBatch deleteEdgeFromSourceWithTargetType( final EdgeColumnFamilies columnFamilies,
+ final ApplicationScope scope, final MarkedEdge markedEdge,
+ final Collection<Shard> shards,
+ final DirectedEdgeMeta directedEdgeMeta,
+ final UUID timestamp ) {
+ return new SourceTargetTypeWriteOp( columnFamilies, markedEdge ) {
+
+ @Override
+ void writeEdge( final MutationBatch batch,
+ final MultiTennantColumnFamily<ScopedRowKey<RowKeyType>, DirectedEdge> columnFamily,
+ final ApplicationScope scope, final RowKeyType rowKey, final DirectedEdge edge,
+ final Shard shard, final boolean isDeleted ) {
+
+
+ batch.withRow( columnFamilies.getSourceNodeTargetTypeCfName(), ScopedRowKey.fromKey( scope.getApplication(), rowKey ) )
+ .deleteColumn( edge );
+ writeEdgeShardStrategy.increment( scope, shard, -1, directedEdgeMeta );
+ }
+ }.createBatch( scope, shards, timestamp );
+ }
+
+
+ @Override
+ public MutationBatch deleteEdgeToTarget( final EdgeColumnFamilies columnFamilies, final ApplicationScope scope,
+ final MarkedEdge markedEdge, final Collection<Shard> shards,
+ final DirectedEdgeMeta directedEdgeMeta, final UUID timestamp ) {
+
+ return new TargetWriteOp( columnFamilies, markedEdge ) {
+
+ @Override
+ void writeEdge( final MutationBatch batch,
+ final MultiTennantColumnFamily<ScopedRowKey<RowKey>, DirectedEdge> columnFamily,
+ final ApplicationScope scope, final RowKey rowKey, final DirectedEdge edge,
+ final Shard shard, final boolean isDeleted ) {
+
+ batch.withRow( columnFamily, ScopedRowKey.fromKey( scope.getApplication(), rowKey ) ).deleteColumn( edge );
+ writeEdgeShardStrategy.increment( scope, shard, -1, directedEdgeMeta );
+ }
+ }.createBatch( scope, shards, timestamp );
+ }
+
+
+ @Override
+ public MutationBatch deleteEdgeToTargetWithSourceType( final EdgeColumnFamilies columnFamilies,
+ final ApplicationScope scope, final MarkedEdge markedEdge,
+ final Collection<Shard> shards,
+ final DirectedEdgeMeta directedEdgeMeta,
+ final UUID timestamp ) {
+
+ return new TargetSourceTypeWriteOp( columnFamilies, markedEdge ) {
+
+ @Override
+ void writeEdge( final MutationBatch batch,
+ final MultiTennantColumnFamily<ScopedRowKey<RowKeyType>, DirectedEdge> columnFamily,
+ final ApplicationScope scope, final RowKeyType rowKey, final DirectedEdge edge,
+ final Shard shard, final boolean isDeleted ) {
+
+ batch.withRow( columnFamilies.getTargetNodeSourceTypeCfName(), ScopedRowKey.fromKey( scope.getApplication(), rowKey ) )
+ .deleteColumn( edge );
+ writeEdgeShardStrategy.increment( scope, shard, -1, directedEdgeMeta );
+ }
+ }.createBatch( scope, shards, timestamp );
+ }
+
+
+ @Override
+ public MutationBatch deleteEdgeVersions( final EdgeColumnFamilies columnFamilies, final ApplicationScope scope,
+ final MarkedEdge markedEdge, final Collection<Shard> shards,
+ final DirectedEdgeMeta directedEdgeMeta, final UUID timestamp ) {
+
+ return new EdgeVersions( columnFamilies, markedEdge ) {
+
+ @Override
+ void writeEdge( final MutationBatch batch,
+ final MultiTennantColumnFamily<ScopedRowKey<EdgeRowKey>, Long> columnFamily,
+ final ApplicationScope scope, final EdgeRowKey rowKey, final Long column, final Shard shard,
+ final boolean isDeleted ) {
+ batch.withRow( columnFamilies.getGraphEdgeVersions(), ScopedRowKey.fromKey( scope.getApplication(), rowKey ) )
+ .deleteColumn( column );
+ writeEdgeShardStrategy.increment( scope, shard, -1, directedEdgeMeta );
+ }
+ }.createBatch( scope, shards, timestamp );
+ }
+
+
+ @Override
+ public Iterator<MarkedEdge> getEdgeVersions( final EdgeColumnFamilies columnFamilies, final ApplicationScope scope,
+ final SearchByEdge search, final Collection<Shard> shards ) {
+ ValidationUtils.validateApplicationScope( scope );
+ GraphValidation.validateSearchByEdge( search );
+
+ final Id targetId = search.targetNode();
+ final Id sourceId = search.sourceNode();
+ final String type = search.getType();
+ final long maxTimestamp = search.getMaxTimestamp();
+ final MultiTennantColumnFamily<ScopedRowKey<EdgeRowKey>, Long> columnFamily =
+ columnFamilies.getGraphEdgeVersions();
+ final Serializer<Long> serializer = columnFamily.getColumnSerializer();
+
+
+ final OrderedComparator<MarkedEdge> comparator = new OrderedComparator<>( DescendingTimestampComparator.INSTANCE, search.getOrder());
+
+
+
+
+ final EdgeSearcher<EdgeRowKey, Long, MarkedEdge> searcher =
+ new EdgeSearcher<EdgeRowKey, Long, MarkedEdge>( scope, shards, search.getOrder(), comparator, maxTimestamp,
+ search.last().transform( TRANSFORM ) ) {
+
+
+ @Override
+ protected Serializer<Long> getSerializer() {
+ return serializer;
+ }
+
+
+ @Override
+ public void buildRange( final RangeBuilder builder ) {
+
+
+ if ( last.isPresent() ) {
+ super.buildRange( 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 );
+ }
+
+
+ @Override
+ protected Long createColumn( final MarkedEdge last ) {
+ return last.getTimestamp();
+ }
+
+
+ @Override
+ protected MarkedEdge createEdge( final Long column, final boolean marked ) {
+ return new SimpleMarkedEdge( sourceId, type, targetId, column.longValue(), marked );
+ }
+
+
+
+ };
+
+ return new ShardsColumnIterator<>( searcher, columnFamily, keyspace, cassandraConfig.getReadCL(),
+ graphFig.getScanPageSize() );
+ }
+
+
+ @Override
+ public Iterator<MarkedEdge> getEdgesFromSource( final EdgeColumnFamilies columnFamilies,
+ final ApplicationScope scope, final SearchByEdgeType search,
+ final Collection<Shard> shards ) {
+
+ ValidationUtils.validateApplicationScope( scope );
+ GraphValidation.validateSearchByEdgeType( search );
+
+ final Id sourceId = search.getNode();
+ final String type = search.getType();
+ final long maxTimestamp = search.getMaxTimestamp();
+ final MultiTennantColumnFamily<ScopedRowKey<RowKey>, DirectedEdge> columnFamily =
+ columnFamilies.getSourceNodeCfName();
+ final Serializer<DirectedEdge> serializer = columnFamily.getColumnSerializer();
+
+
+ final OrderedComparator<MarkedEdge> comparator = new OrderedComparator<>( TargetDirectedEdgeDescendingComparator.INSTANCE, search.getOrder());
+
+
+
+ final EdgeSearcher<RowKey, DirectedEdge, MarkedEdge> searcher =
+ new EdgeSearcher<RowKey, DirectedEdge, MarkedEdge>( scope, shards, search.getOrder(), comparator, maxTimestamp,
+ search.last().transform( TRANSFORM ) ) {
+
+
+ @Override
+ protected Serializer<DirectedEdge> getSerializer() {
+ return serializer;
+ }
+
+
+ @Override
+ protected RowKey generateRowKey( long shard ) {
+ return new RowKey( sourceId, type, shard );
+ }
+
+
+ @Override
+ protected DirectedEdge createColumn( final MarkedEdge last ) {
+ return new DirectedEdge( last.getTargetNode(), last.getTimestamp() );
+ }
+
+
+ @Override
+ protected MarkedEdge createEdge( final DirectedEdge edge, final boolean marked ) {
+ return new SimpleMarkedEdge( sourceId, type, edge.id, edge.timestamp, marked );
+ }
+ };
+
+
+ return new ShardsColumnIterator<>( searcher, columnFamily, keyspace, cassandraConfig.getReadCL(),
+ graphFig.getScanPageSize() );
+ }
+
+
+ @Override
+ public Iterator<MarkedEdge> getEdgesFromSourceByTargetType( final EdgeColumnFamilies columnFamilies,
+ final ApplicationScope scope,
+ final SearchByIdType search,
+ final Collection<Shard> shards ) {
+
+ ValidationUtils.validateApplicationScope( scope );
+ GraphValidation.validateSearchByEdgeType( search );
+
+ final Id targetId = search.getNode();
+ final String type = search.getType();
+ final String targetType = search.getIdType();
+ final long maxTimestamp = search.getMaxTimestamp();
+ final MultiTennantColumnFamily<ScopedRowKey<RowKeyType>, DirectedEdge> columnFamily =
+ columnFamilies.getSourceNodeTargetTypeCfName();
+ final Serializer<DirectedEdge> serializer = columnFamily.getColumnSerializer();
+
+ final OrderedComparator<MarkedEdge> comparator = new OrderedComparator<>( TargetDirectedEdgeDescendingComparator.INSTANCE, search.getOrder());
+
+
+ final EdgeSearcher<RowKeyType, DirectedEdge, MarkedEdge> searcher =
+ new EdgeSearcher<RowKeyType, DirectedEdge, MarkedEdge>( scope, shards, search.getOrder(), comparator, maxTimestamp,
+ search.last().transform( TRANSFORM ) ) {
+
+ @Override
+ protected Serializer<DirectedEdge> getSerializer() {
+ return serializer;
+ }
+
+
+ @Override
+ protected RowKeyType generateRowKey( long shard ) {
+ return new RowKeyType( targetId, type, targetType, shard );
+ }
+
+
+ @Override
+ protected DirectedEdge createColumn( final MarkedEdge last ) {
+ return new DirectedEdge( last.getTargetNode(), last.getTimestamp() );
+ }
+
+
+ @Override
+ protected MarkedEdge createEdge( final DirectedEdge edge, final boolean marked ) {
+ return new SimpleMarkedEdge( targetId, type, edge.id, edge.timestamp, marked );
+ }
+ };
+
+ return new ShardsColumnIterator( searcher, columnFamily, keyspace, cassandraConfig.getReadCL(),
+ graphFig.getScanPageSize() );
+ }
+
+
+ @Override
+ public Iterator<MarkedEdge> getEdgesToTarget( final EdgeColumnFamilies columnFamilies, final ApplicationScope scope,
+ final SearchByEdgeType search, final Collection<Shard> shards ) {
+ ValidationUtils.validateApplicationScope( scope );
+ GraphValidation.validateSearchByEdgeType( search );
+
+ final Id targetId = search.getNode();
+ final String type = search.getType();
+ final long maxTimestamp = search.getMaxTimestamp();
+ final MultiTennantColumnFamily<ScopedRowKey<RowKey>, DirectedEdge> columnFamily =
+ columnFamilies.getTargetNodeCfName();
+ final Serializer<DirectedEdge> serializer = columnFamily.getColumnSerializer();
+
+ final OrderedComparator<MarkedEdge> comparator = new OrderedComparator<>( SourceDirectedEdgeDescendingComparator.INSTANCE, search.getOrder());
+
+ final EdgeSearcher<RowKey, DirectedEdge, MarkedEdge> searcher =
+ new EdgeSearcher<RowKey, DirectedEdge, MarkedEdge>( scope, shards, search.getOrder(),comparator, maxTimestamp,
+ search.last().transform( TRANSFORM ) ) {
+
+ @Override
+ protected Serializer<DirectedEdge> getSerializer() {
+ return serializer;
+ }
+
+
+ @Override
+ protected RowKey generateRowKey( long shard ) {
+ return new RowKey( targetId, type, shard );
+ }
+
+
+ @Override
+ protected DirectedEdge createColumn( final MarkedEdge last ) {
+ return new DirectedEdge( last.getSourceNode(), last.getTimestamp() );
+ }
+
+
+ @Override
+ protected MarkedEdge createEdge( final DirectedEdge edge, final boolean marked ) {
+ return new SimpleMarkedEdge( edge.id, type, targetId, edge.timestamp, marked );
+ }
+ };
+
+
+ return new ShardsColumnIterator<>( searcher, columnFamily, keyspace, cassandraConfig.getReadCL(),
+ graphFig.getScanPageSize() );
+ }
+
+
+ @Override
+ public Iterator<MarkedEdge> getEdgesToTargetBySourceType( final EdgeColumnFamilies columnFamilies,
+ final ApplicationScope scope,
+ final SearchByIdType search,
+ final Collection<Shard> shards ) {
+
+ ValidationUtils.validateApplicationScope( scope );
+ GraphValidation.validateSearchByEdgeType( search );
+
+ final Id targetId = search.getNode();
+ final String sourceType = search.getIdType();
+ final String type = search.getType();
+ final long maxTimestamp = search.getMaxTimestamp();
+ final MultiTennantColumnFamily<ScopedRowKey<RowKeyType>, DirectedEdge> columnFamily =
+ columnFamilies.getTargetNodeSourceTypeCfName();
+ final Serializer<DirectedEdge> serializer = columnFamily.getColumnSerializer();
+
+ final OrderedComparator<MarkedEdge> comparator = new OrderedComparator<>( SourceDirectedEdgeDescendingComparator.INSTANCE, search.getOrder());
+
+
+ final EdgeSearcher<RowKeyType, DirectedEdge, MarkedEdge> searcher =
+ new EdgeSearcher<RowKeyType, DirectedEdge, MarkedEdge>( scope, shards, search.getOrder(), comparator, maxTimestamp,
+ search.last().transform( TRANSFORM ) ) {
+ @Override
+ protected Serializer<DirectedEdge> getSerializer() {
+ return serializer;
+ }
+
+
+ @Override
+ protected RowKeyType generateRowKey( final long shard ) {
+ return new RowKeyType( targetId, type, sourceType, shard );
+ }
+
+
+ @Override
+ protected DirectedEdge createColumn( final MarkedEdge 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 ShardsColumnIterator<>( searcher, columnFamily, keyspace, cassandraConfig.getReadCL(),
+ graphFig.getScanPageSize() );
+ }
+
+
+
+
+
+ /**
+ * Simple callback to perform puts and deletes with a common row setup code
+ *
+ * @param <R> The row key type
+ * @param <C> The column type
+ */
+ private abstract class RowOp<R, C> {
+
+
+ /**
+ * Return the column family used for the write
+ */
+ protected abstract MultiTennantColumnFamily<ScopedRowKey<R>, C> getColumnFamily();
+
+ /**
+ * Get the row key
+ */
+ public abstract R getRowKey( final Shard shard );
+
+ /**
+ * Get the column family value
+ */
+ protected abstract C getDirectedEdge();
+
+ /**
+ * Get the flag on if it's deleted
+ */
+ protected abstract boolean isDeleted();
+
+
+ /**
+ * Write the edge with the given data
+ */
+ abstract void writeEdge( final MutationBatch batch,
+ final MultiTennantColumnFamily<ScopedRowKey<R>, C> columnFamily,
+ final ApplicationScope scope, final R rowKey, final C column, final Shard shard,
+ final boolean isDeleted );
+
+
+ /**
+ * Create a mutation batch
+ */
+ public MutationBatch createBatch( final ApplicationScope scope, final Collection<Shard> shards,
+ final UUID opTimestamp ) {
+
+ final MutationBatch batch =
+ keyspace.prepareMutationBatch().withConsistencyLevel( cassandraConfig.getWriteCL() )
+ .withTimestamp( opTimestamp.timestamp() );
+
+
+ final C column = getDirectedEdge();
+ final MultiTennantColumnFamily<ScopedRowKey<R>, C> columnFamily = getColumnFamily();
+ final boolean isDeleted = isDeleted();
+
+
+ for ( Shard shard : shards ) {
+ final R rowKey = getRowKey( shard );
+ writeEdge( batch, columnFamily, scope, rowKey, column, shard, isDeleted );
+ }
+
+
+ return batch;
+ }
+ }
+
+
+ /**
+ * Perform a write of the source->target
+ */
+ private abstract class SourceWriteOp extends RowOp<RowKey, DirectedEdge> {
+
+ private final MultiTennantColumnFamily<ScopedRowKey<RowKey>, DirectedEdge> columnFamily;
+ private final Id sourceNodeId;
+
+
+ private final String type;
+ private final boolean isDeleted;
+ private final DirectedEdge directedEdge;
+
+
+ /**
+ * Write the source write operation
+ */
+ private SourceWriteOp( final EdgeColumnFamilies edgeColumnFamilies, final MarkedEdge markedEdge ) {
+ this.columnFamily = edgeColumnFamilies.getSourceNodeCfName();
+
+ this.sourceNodeId = markedEdge.getSourceNode();
+
+ this.type = markedEdge.getType();
+ this.isDeleted = markedEdge.isDeleted();
+
+ this.directedEdge = new DirectedEdge( markedEdge.getTargetNode(), markedEdge.getTimestamp() );
+ }
+
+
+ @Override
+ protected MultiTennantColumnFamily<ScopedRowKey<RowKey>, DirectedEdge> getColumnFamily() {
+ return columnFamily;
+ }
+
+
+ @Override
+ public RowKey getRowKey( final Shard shard ) {
+ return new RowKey( sourceNodeId, type, shard.getShardIndex() );
+ }
+
+
+ @Override
+ protected DirectedEdge getDirectedEdge() {
+ return directedEdge;
+ }
+
+
+ @Override
+ protected boolean isDeleted() {
+ return isDeleted;
+ }
+ }
+
+
+ /**
+ * Perform a write of the source->target with target type
+ */
+ private abstract class SourceTargetTypeWriteOp extends RowOp<RowKeyType, DirectedEdge> {
+
+ private final MultiTennantColumnFamily<ScopedRowKey<RowKeyType>, DirectedEdge> columnFamily;
+ private final Id sourceNodeId;
+ private final String type;
+ private Id targetId;
+ private final boolean isDeleted;
+ private final DirectedEdge directedEdge;
+
+
+ /**
+ * Write the source write operation
+ */
+ private SourceTargetTypeWriteOp( final EdgeColumnFamilies edgeColumnFamilies, final MarkedEdge markedEdge ) {
+ this.columnFamily = edgeColumnFamilies.getSourceNodeTargetTypeCfName();
+
+ this.sourceNodeId = markedEdge.getSourceNode();
+
+ this.type = markedEdge.getType();
+ this.targetId = markedEdge.getTargetNode();
+ this.isDeleted = markedEdge.isDeleted();
+
+ this.directedEdge = new DirectedEdge( targetId, markedEdge.getTimestamp() );
+ }
+
+
+ @Override
+ protected MultiTennantColumnFamily<ScopedRowKey<RowKeyType>, DirectedEdge> getColumnFamily() {
+ return columnFamily;
+ }
+
+
+ @Override
+ public RowKeyType getRowKey( final Shard shard ) {
+ return new RowKeyType( sourceNodeId, type, targetId, shard.getShardIndex() );
+ }
+
+
+ @Override
+ protected DirectedEdge getDirectedEdge() {
+ return directedEdge;
+ }
+
+
+ @Override
+ protected boolean isDeleted() {
+ return isDeleted;
+ }
+ }
+
+
+ /**
+ * Perform a write of the target <-- source
+ */
+ private abstract class TargetWriteOp extends RowOp<RowKey, DirectedEdge> {
+
+ private final MultiTennantColumnFamily<ScopedRowKey<RowKey>, DirectedEdge> columnFamily;
+ private final Id targetNode;
+
+
+ private final String type;
+ private final boolean isDeleted;
+ private final DirectedEdge directedEdge;
+
+
+ /**
+ * Write the source write operation
+ */
+ private TargetWriteOp( final EdgeColumnFamilies edgeColumnFamilies, final MarkedEdge markedEdge ) {
+ this.columnFamily = edgeColumnFamilies.getTargetNodeCfName();
+
+ this.targetNode = markedEdge.getTargetNode();
+
+ this.type = markedEdge.getType();
+ this.isDeleted = markedEdge.isDeleted();
+
+ this.directedEdge = new DirectedEdge( markedEdge.getSourceNode(), markedEdge.getTimestamp() );
+ }
+
+
+ @Override
+ protected MultiTennantColumnFamily<ScopedRowKey<RowKey>, DirectedEdge> getColumnFamily() {
+ return columnFamily;
+ }
+
+
+ @Override
+ public RowKey getRowKey( final Shard shard ) {
+ return new RowKey( targetNode, type, shard.getShardIndex() );
+ }
+
+
+ @Override
+ protected DirectedEdge getDirectedEdge() {
+ return directedEdge;
+ }
+
+
+ @Override
+ protected boolean isDeleted() {
+ return isDeleted;
+ }
+ }
+
+
+ /**
+ * Perform a write of the target<--source with source type
+ */
+ private abstract class TargetSourceTypeWriteOp extends RowOp<RowKeyType, DirectedEdge> {
+
+ private final MultiTennantColumnFamily<ScopedRowKey<RowKeyType>, DirectedEdge> columnFamily;
+ private final Id targetNode;
+
+ private final Id sourceNode;
+
+ final String type;
+
+ final boolean isDeleted;
+ final DirectedEdge directedEdge;
+
+
+ /**
+ * Write the source write operation
+ */
+ private TargetSourceTypeWriteOp( final EdgeColumnFamilies edgeColumnFamilies, final MarkedEdge markedEdge ) {
+ this.columnFamily = edgeColumnFamilies.getSourceNodeTargetTypeCfName();
+
+ this.targetNode = markedEdge.getTargetNode();
+ this.sourceNode = markedEdge.getSourceNode();
+
+ this.type = markedEdge.getType();
+ this.isDeleted = markedEdge.isDeleted();
+
+ this.directedEdge = new DirectedEdge( sourceNode, markedEdge.getTimestamp() );
+ }
+
+
+ @Override
+ protected MultiTennantColumnFamily<ScopedRowKey<RowKeyType>, DirectedEdge> getColumnFamily() {
+ return columnFamily;
+ }
+
+
+ @Override
+ public RowKeyType getRowKey( final Shard shard ) {
+ return new RowKeyType( targetNode, type, sourceNode, shard.getShardIndex() );
+ }
+
+
+ @Override
+ protected DirectedEdge getDirectedEdge() {
+ return directedEdge;
+ }
+
+
+ @Override
+ protected boolean isDeleted() {
+ return isDeleted;
+ }
+ }
+
+
+ /**
+ * Perform a write of the edge versions
+ */
+ private abstract class EdgeVersions extends RowOp<EdgeRowKey, Long> {
+
+ private final MultiTennantColumnFamily<ScopedRowKey<EdgeRowKey>, Long> columnFamily;
+ private final Id targetNode;
+
+ private final Id sourceNode;
+
+ final String type;
+
+ final boolean isDeleted;
+ final Long edgeVersion;
+
+
+ /**
+ * Write the source write operation
+ */
+ private EdgeVersions( final EdgeColumnFamilies edgeColumnFamilies, final MarkedEdge markedEdge ) {
+ this.columnFamily = edgeColumnFamilies.getGraphEdgeVersions();
+
+ this.targetNode = markedEdge.getTargetNode();
+ this.sourceNode = markedEdge.getSourceNode();
+
+ this.type = markedEdge.getType();
+ this.isDeleted = markedEdge.isDeleted();
+
+ this.edgeVersion = markedEdge.getTimestamp();
+ }
+
+
+ @Override
+ protected MultiTennantColumnFamily<ScopedRowKey<EdgeRowKey>, Long> getColumnFamily() {
+ return columnFamily;
+ }
+
+
+ @Override
+ public EdgeRowKey getRowKey( final Shard shard ) {
+ return new EdgeRowKey( sourceNode, type, targetNode, shard.getShardIndex() );
+ }
+
+
+ @Override
+ protected Long getDirectedEdge() {
+ return edgeVersion;
+ }
+
+
+ @Override
+ protected boolean isDeleted() {
+ return isDeleted;
+ }
+ }
+
+
+
+
+
+
+ private static final Function<Edge, MarkedEdge> TRANSFORM = new Function<Edge, MarkedEdge>() {
+ @Nullable
+ @Override
+ public MarkedEdge apply( @Nullable final Edge input ) {
+
+ if ( input == null ) {
+ return null;
+ }
+
+ if ( input instanceof MarkedEdge ) {
+ return ( MarkedEdge ) input;
+ }
+
+ return new SimpleMarkedEdge( input.getSourceNode(), input.getType(), input.getTargetNode(),
+ input.getTimestamp(), false );
+ }
+ };
+}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/43a2b299/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/count/NodeShardCounterSerializationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/count/NodeShardCounterSerializationImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/count/NodeShardCounterSerializationImpl.java
index 6934275..fca1949 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/count/NodeShardCounterSerializationImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/count/NodeShardCounterSerializationImpl.java
@@ -43,6 +43,7 @@ import org.apache.usergrid.persistence.graph.serialization.impl.shard.Shard;
import org.apache.usergrid.persistence.graph.serialization.impl.shard.impl.serialize.EdgeShardRowKeySerializer;
import org.apache.usergrid.persistence.model.entity.Id;
+import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.inject.Inject;
import com.google.inject.Singleton;
@@ -138,7 +139,8 @@ public class NodeShardCounterSerializationImpl implements NodeShardCounterSerial
return Collections.singleton(
new MultiTennantColumnFamilyDefinition( EDGE_SHARD_COUNTS, BytesType.class.getSimpleName(),
ColumnTypes.BOOLEAN, CounterColumnType.class.getSimpleName(),
- MultiTennantColumnFamilyDefinition.CacheOption.ALL ) );
+ MultiTennantColumnFamilyDefinition.CacheOption.ALL, Optional
+ .absent() ) );
}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/43a2b299/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeShardSerializationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeShardSerializationImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeShardSerializationImpl.java
index f107307..3ef5f92 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeShardSerializationImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/EdgeShardSerializationImpl.java
@@ -32,8 +32,8 @@ import org.apache.usergrid.persistence.core.astyanax.ColumnParser;
import org.apache.usergrid.persistence.core.astyanax.ColumnTypes;
import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamily;
import org.apache.usergrid.persistence.core.astyanax.MultiTennantColumnFamilyDefinition;
-import org.apache.usergrid.persistence.core.astyanax.ScopedRowKeySerializer;
import org.apache.usergrid.persistence.core.astyanax.ScopedRowKey;
+import org.apache.usergrid.persistence.core.astyanax.ScopedRowKeySerializer;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.core.util.ValidationUtils;
import org.apache.usergrid.persistence.graph.GraphFig;
@@ -62,8 +62,8 @@ public class EdgeShardSerializationImpl implements EdgeShardSerialization {
* Edge shards
*/
private static final MultiTennantColumnFamily<ScopedRowKey<DirectedEdgeMeta>, Long> EDGE_SHARDS =
- new MultiTennantColumnFamily<>( "Edge_Shards",
- new ScopedRowKeySerializer<>( EdgeShardRowKeySerializer.INSTANCE ), LongSerializer.get() );
+ new MultiTennantColumnFamily<>( "Edge_Shards",
+ new ScopedRowKeySerializer<>( EdgeShardRowKeySerializer.INSTANCE ), LongSerializer.get() );
private static final ShardColumnParser COLUMN_PARSER = new ShardColumnParser();
@@ -72,20 +72,22 @@ public class EdgeShardSerializationImpl implements EdgeShardSerialization {
protected final Keyspace keyspace;
protected final CassandraConfig cassandraConfig;
protected final GraphFig graphFig;
+ protected final ShardConsistency shardConsistency;
@Inject
public EdgeShardSerializationImpl( final Keyspace keyspace, final CassandraConfig cassandraConfig,
- final GraphFig graphFig ) {
+ final GraphFig graphFig, final ShardConsistency shardConsistency ) {
this.keyspace = keyspace;
this.cassandraConfig = cassandraConfig;
this.graphFig = graphFig;
+ this.shardConsistency = shardConsistency;
}
@Override
- public MutationBatch writeShardMeta( final ApplicationScope scope,
- final Shard shard, final DirectedEdgeMeta metaData) {
+ public MutationBatch writeShardMeta( final ApplicationScope scope, final Shard shard,
+ final DirectedEdgeMeta metaData ) {
ValidationUtils.validateApplicationScope( scope );
GraphValidation.validateDirectedEdgeMeta( metaData );
@@ -98,7 +100,8 @@ public class EdgeShardSerializationImpl implements EdgeShardSerialization {
final ScopedRowKey rowKey = ScopedRowKey.fromKey( scope.getApplication(), metaData );
- final MutationBatch batch = keyspace.prepareMutationBatch();
+ final MutationBatch batch =
+ keyspace.prepareMutationBatch().withConsistencyLevel( shardConsistency.getShardWriteConsistency() );
batch.withTimestamp( shard.getCreatedTime() ).withRow( EDGE_SHARDS, rowKey )
.putColumn( shard.getShardIndex(), shard.isCompacted() );
@@ -108,8 +111,9 @@ public class EdgeShardSerializationImpl implements EdgeShardSerialization {
@Override
- public Iterator<Shard> getShardMetaData( final ApplicationScope scope,
- final Optional<Shard> start, final DirectedEdgeMeta metaData ) {
+ public Iterator<Shard> getShardMetaDataLocal( final ApplicationScope scope, final Optional<Shard> start,
+ final DirectedEdgeMeta metaData ) {
+
ValidationUtils.validateApplicationScope( scope );
GraphValidation.validateDirectedEdgeMeta( metaData );
@@ -134,27 +138,28 @@ public class EdgeShardSerializationImpl implements EdgeShardSerialization {
final RowQuery<ScopedRowKey<DirectedEdgeMeta>, Long> query =
- keyspace.prepareQuery( EDGE_SHARDS ).setConsistencyLevel( cassandraConfig.getReadCL() ).getKey( rowKey )
- .autoPaginate( true ).withColumnRange( rangeBuilder.build() );
+ keyspace.prepareQuery( EDGE_SHARDS ).setConsistencyLevel( shardConsistency.getShardReadConsistency() ).getKey( rowKey )
+ .autoPaginate( true ).withColumnRange( rangeBuilder.build() );
return new ColumnNameIterator<>( query, COLUMN_PARSER, false );
}
- @Override
- public MutationBatch removeShardMeta( final ApplicationScope scope,
- final Shard shard, final DirectedEdgeMeta metaData) {
+
+ @Override
+ public MutationBatch removeShardMeta( final ApplicationScope scope, final Shard shard,
+ final DirectedEdgeMeta metaData ) {
ValidationUtils.validateApplicationScope( scope );
GraphValidation.valiateShard( shard );
GraphValidation.validateDirectedEdgeMeta( metaData );
-
final ScopedRowKey rowKey = ScopedRowKey.fromKey( scope.getApplication(), metaData );
- final MutationBatch batch = keyspace.prepareMutationBatch();
+ final MutationBatch batch =
+ keyspace.prepareMutationBatch().withConsistencyLevel( shardConsistency.getShardWriteConsistency( ));
batch.withRow( EDGE_SHARDS, rowKey ).deleteColumn( shard.getShardIndex() );
@@ -164,19 +169,14 @@ public class EdgeShardSerializationImpl implements EdgeShardSerialization {
@Override
public Collection<MultiTennantColumnFamilyDefinition> getColumnFamilies() {
-
-
return Collections.singleton(
- new MultiTennantColumnFamilyDefinition( EDGE_SHARDS, BytesType.class.getSimpleName(),
- ColumnTypes.LONG_TYPE_REVERSED, BytesType.class.getSimpleName(),
- MultiTennantColumnFamilyDefinition.CacheOption.KEYS ) );
+ new MultiTennantColumnFamilyDefinition( EDGE_SHARDS, BytesType.class.getSimpleName(),
+ ColumnTypes.LONG_TYPE_REVERSED, BytesType.class.getSimpleName(),
+ MultiTennantColumnFamilyDefinition.CacheOption.KEYS, Optional
+ .absent() ) );
}
-
-
-
-
private static class ShardColumnParser implements ColumnParser<Long, Shard> {
@Override
http://git-wip-us.apache.org/repos/asf/usergrid/blob/43a2b299/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardAllocationImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardAllocationImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardAllocationImpl.java
index 7a7fb3f..085dd97 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardAllocationImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/shard/impl/NodeShardAllocationImpl.java
@@ -32,7 +32,6 @@ import org.apache.usergrid.persistence.core.util.ValidationUtils;
import org.apache.usergrid.persistence.graph.GraphFig;
import org.apache.usergrid.persistence.graph.MarkedEdge;
import org.apache.usergrid.persistence.graph.SearchByEdgeType;
-import org.apache.usergrid.persistence.graph.exception.GraphRuntimeException;
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.EdgeShardSerialization;
@@ -46,6 +45,8 @@ import org.apache.usergrid.persistence.graph.serialization.util.GraphValidation;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
import com.google.inject.Inject;
import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
@@ -58,10 +59,12 @@ import com.netflix.astyanax.util.TimeUUIDUtils;
public class NodeShardAllocationImpl implements NodeShardAllocation {
- private static final Logger LOG = LoggerFactory.getLogger( NodeShardAllocationImpl.class );
+ private static final Logger logger = LoggerFactory.getLogger( NodeShardAllocationImpl.class );
private static final Shard MIN_SHARD = new Shard( 0, 0, true );
+ private static final NoOpCompaction NO_OP_COMPACTION = new NoOpCompaction();
+
private final EdgeShardSerialization edgeShardSerialization;
private final EdgeColumnFamilies edgeColumnFamilies;
private final ShardedEdgeSerialization shardedEdgeSerialization;
@@ -76,7 +79,7 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
final EdgeColumnFamilies edgeColumnFamilies,
final ShardedEdgeSerialization shardedEdgeSerialization,
final NodeShardApproximation nodeShardApproximation, final TimeService timeService,
- final GraphFig graphFig, final ShardGroupCompaction shardGroupCompaction ) {
+ final GraphFig graphFig, final ShardGroupCompaction shardGroupCompaction) {
this.edgeShardSerialization = edgeShardSerialization;
this.edgeColumnFamilies = edgeColumnFamilies;
this.shardedEdgeSerialization = shardedEdgeSerialization;
@@ -88,8 +91,8 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
@Override
- public Iterator<ShardEntryGroup> getShards( final ApplicationScope scope, final Optional<Shard> maxShardId,
- final DirectedEdgeMeta directedEdgeMeta ) {
+ public Iterator<ShardEntryGroup> getShardsLocal( final ApplicationScope scope, final Optional<Shard> maxShardId,
+ final DirectedEdgeMeta directedEdgeMeta ) {
ValidationUtils.validateApplicationScope( scope );
Preconditions.checkNotNull( maxShardId, "maxShardId cannot be null" );
@@ -103,7 +106,7 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
}
else {
- existingShards = edgeShardSerialization.getShardMetaData( scope, maxShardId, directedEdgeMeta );
+ existingShards = edgeShardSerialization.getShardMetaDataLocal( scope, maxShardId, directedEdgeMeta );
}
if ( existingShards == null || !existingShards.hasNext() ) {
@@ -120,71 +123,90 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
existingShards = Collections.singleton( MIN_SHARD ).iterator();
}
- return new ShardEntryGroupIterator( existingShards, graphFig.getShardMinDelta(), shardGroupCompaction, scope,
- directedEdgeMeta );
+ return new ShardEntryGroupIterator( existingShards, shardGroupCompaction, scope, directedEdgeMeta );
}
@Override
- public boolean auditShard( final ApplicationScope scope, final ShardEntryGroup shardEntryGroup,
+ public boolean auditShard( final ApplicationScope scope, final ShardEntryGroup lastLoadedShardEntryGroup,
final DirectedEdgeMeta directedEdgeMeta ) {
ValidationUtils.validateApplicationScope( scope );
- GraphValidation.validateShardEntryGroup( shardEntryGroup );
+ GraphValidation.validateShardEntryGroup( lastLoadedShardEntryGroup );
GraphValidation.validateDirectedEdgeMeta( directedEdgeMeta );
- Preconditions.checkNotNull( shardEntryGroup, "shardEntryGroup cannot be null" );
+ Preconditions.checkNotNull( lastLoadedShardEntryGroup, "lastLoadedShardEntryGroup cannot be null" );
+
+ //we have to read our state from cassandra first to ensure we have an up to date view from other regions
/**
- * Nothing to do, it's been created very recently, we don't create a new one
+ * We need to re-read our state.
+ * Read our shard entry groups to ensure we have a current state
*/
- if ( shardEntryGroup.isCompactionPending() ) {
+ final Iterator<ShardEntryGroup> shardEntryGroupIterator =
+ getCurrentStateIterator( scope, lastLoadedShardEntryGroup, directedEdgeMeta );
+
+
+ if ( !shardEntryGroupIterator.hasNext() ) {
+ logger.warn( "Could not read our shard entries. Our state is unknown, short circuiting" );
return false;
}
- //we can't allocate, we have more than 1 write shard currently. We need to compact first
- if ( shardEntryGroup.entrySize() != 1 ) {
+ final ShardEntryGroup shardEntryGroup = shardEntryGroupIterator.next();
+
+ //someone has already allocated, it's a different group. Move on, nothing to see here
+ if ( !shardEntryGroup.equals( lastLoadedShardEntryGroup ) ) {
+ logger.info( "Stale shard group on audit or group {} found, ignoring", lastLoadedShardEntryGroup );
return false;
}
-
/**
- * Check the min shard in our system
+ * Nothing to do, it's been created very recently, we don't create a new one
*/
- final Shard shard = shardEntryGroup.getMinShard();
-
+ if ( shardEntryGroup.isCompactionPending() ) {
+ logger.info( "Shard group {} is compacting, skipping", lastLoadedShardEntryGroup );
+ return false;
+ }
- if ( shard.getCreatedTime() >= getMinTime() ) {
+ //we can't allocate, we have more than 1 write shard currently. We need to compact first
+ if ( shardEntryGroup.entrySize() != 1 ) {
return false;
}
+ final Shard minShard = lastLoadedShardEntryGroup.getMinShard();
+
+
/**
* Check out if we have a count for our shard allocation
*/
- final long count = nodeShardApproximation.getCount( scope, shard, directedEdgeMeta );
+ final long count = nodeShardApproximation.getCount( scope, minShard, directedEdgeMeta );
final long shardSize = graphFig.getShardSize();
if ( count < shardSize ) {
+ logger.debug( "Our count is less than our shard size, not allocating a new shard" );
return false;
}
- if(LOG.isDebugEnabled()){
- LOG.debug("Count of {} has exceeded shard config of {} will begin compacting", count, shardSize);
- }
+
+ logger.debug( "Count of {} has exceeded shard config of {} will begin compacting", count, shardSize );
+
/**
- * We want to allocate a new shard as close to the max value as possible. This way if we're filling up a shard rapidly, we split it near the head of the values.
- * Further checks to this group will result in more splits, similar to creating a tree type structure and splitting each node.
+ * We want to allocate a new shard as close to the max value as possible. This way if we're filling up a
+ * shard rapidly, we split it near the head of the values.
*
- * This means that the lower shard can be re-split later if it is still too large. We do the division to truncate
- * to a split point < what our current max is that would be approximately be our pivot ultimately if we split from the
- * lower bound and moved forward. Doing this will stop the current shard from expanding and avoid a point where we cannot
- * ultimately compact to the correct shard size.
+ * Further checks to this group will result in more splits, similar to creating a tree type structure and
+ * splitting each node.
+ *
+ * This means that the lower shard can be re-split later if it is still too large. We do the division to
+ * truncate to a split point < what our current max is that would be approximately be our pivot ultimately
+ * if we split from the lower bound and moved forward. Doing this will stop the current shard from expanding
+ * and avoid a point where we cannot ultimately compact to the correct shard size due to excessive tombstones.
*/
@@ -193,13 +215,14 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
*/
final Iterator<MarkedEdge> edges = directedEdgeMeta
- .loadEdges( shardedEdgeSerialization, edgeColumnFamilies, scope, shardEntryGroup.getReadShards(), 0,
- SearchByEdgeType.Order.ASCENDING );
+ .loadEdges( shardedEdgeSerialization, edgeColumnFamilies, scope, shardEntryGroup.getReadShards(), 0,
+ SearchByEdgeType.Order.ASCENDING );
if ( !edges.hasNext() ) {
- LOG.warn( "Tried to allocate a new shard for edge meta data {}, "
- + "but no max value could be found in that row", directedEdgeMeta );
+ logger.warn(
+ "Tried to allocate a new shard for edge meta data {}, " + "but no max value could be found in that row",
+ directedEdgeMeta );
return false;
}
@@ -213,23 +236,26 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
* element will suffice.
*/
+ long i = 1;
- for(long i = 1; edges.hasNext(); i++){
+ for (; edges.hasNext(); i++ ) {
//we hit a pivot shard, set it since it could be the last one we encounter
- if(i% shardSize == 0){
+ if ( i % shardSize == 0 ) {
marked = edges.next();
}
- else{
+ else {
edges.next();
}
}
+ logger.debug( "Iterated {} edges to find the last marked edge", i );
+
/**
* Sanity check in case our counters become severely out of sync with our edge state in cassandra.
*/
- if(marked == null){
- LOG.warn( "Incorrect shard count for shard group {}", shardEntryGroup );
+ if ( marked == null ) {
+ logger.warn( "Incorrect shard count for shard group {}, ignoring", shardEntryGroup );
return false;
}
@@ -237,7 +263,8 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
final Shard newShard = new Shard( marked.getTimestamp(), createTimestamp, false );
- LOG.info( "Allocating new shard {} for edge meta {}", newShard, directedEdgeMeta );
+ logger.info( "Allocating new shard {} for edge meta {} into group {}", newShard, directedEdgeMeta,
+ shardEntryGroup );
final MutationBatch batch = this.edgeShardSerialization.writeShardMeta( scope, newShard, directedEdgeMeta );
@@ -253,24 +280,6 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
}
- @Override
- public long getMinTime() {
-
- final long minimumAllowed = 2 * graphFig.getShardCacheTimeout();
-
- final long minDelta = graphFig.getShardMinDelta();
-
-
- if ( minDelta < minimumAllowed ) {
- throw new GraphRuntimeException( String.format(
- "You must configure the property %s to be >= 2 x %s. Otherwise you risk losing data",
- GraphFig.SHARD_MIN_DELTA, GraphFig.SHARD_CACHE_TIMEOUT ) );
- }
-
- return timeService.getCurrentTime() - minDelta;
- }
-
-
/**
* Return true if the node has been created within our timeout. If this is the case, we dont' need to check
* cassandra, we know it won't exist
@@ -279,9 +288,8 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
//TODO: TN this is broken....
- //The timeout is in milliseconds. Time for a time uuid is 1/10000 of a milli, so we need to get the units correct
- final long timeoutDelta = graphFig.getShardCacheTimeout() ;
-
+ //The timeout is in milliseconds. Time for a time uuid is 1/10000 of a milli, so we need to get the units
+ // correct
final long timeNow = timeService.getCurrentTime();
boolean isNew = true;
@@ -289,18 +297,67 @@ public class NodeShardAllocationImpl implements NodeShardAllocation {
for ( DirectedEdgeMeta.NodeMeta node : directedEdgeMeta.getNodes() ) {
//short circuit
- if(!isNew || node.getId().getUuid().version() > 2){
+ if ( !isNew || node.getId().getUuid().version() > 2 ) {
return false;
}
- final long uuidTime = TimeUUIDUtils.getTimeFromUUID( node.getId().getUuid());
+ final long uuidTime = TimeUUIDUtils.getTimeFromUUID( node.getId().getUuid() );
- final long newExpirationTimeout = uuidTime + timeoutDelta;
+ //take our uuid time and add 10 seconds, if the uuid is within 10 seconds of system time, we can consider
+ // it "new"
+ final long newExpirationTimeout = uuidTime + 1000;
//our expiration is after our current time, treat it as new
- isNew = isNew && newExpirationTimeout > timeNow;
+ isNew = isNew && newExpirationTimeout > timeNow;
}
return isNew;
}
+
+
+ /**
+ * Re-reads our shard groups using our to ensure we get a consistent view of all shards
+ */
+ private Iterator<ShardEntryGroup> getCurrentStateIterator( final ApplicationScope scope,
+ final ShardEntryGroup shardEntryGroup,
+ final DirectedEdgeMeta directedEdgeMeta ) {
+
+ final Shard start = shardEntryGroup.getMaxShard();
+
+ //sanity check
+ if ( start == null ) {
+ logger.warn( "Could not audit shard group {}. Returning.", shardEntryGroup );
+ return Collections.<ShardEntryGroup>emptyList().iterator();
+ }
+
+ logger.debug( "Loading current shard state for shards starting at {}", start );
+
+ final Iterator<Shard> shards = this.edgeShardSerialization
+ .getShardMetaDataLocal( scope, Optional.fromNullable( start ), directedEdgeMeta );
+
+ if ( !shards.hasNext() ) {
+ return Collections.<ShardEntryGroup>emptyList().iterator();
+ }
+
+ return new ShardEntryGroupIterator( shards, NO_OP_COMPACTION, scope, directedEdgeMeta );
+ }
+
+
+ /**
+ * Class that just ignores compaction events, since we're already evaluating the events. A bit of a hack
+ * that shows we need some refactoring
+ */
+ private final static class NoOpCompaction implements ShardGroupCompaction {
+
+ @Override
+ public ListenableFuture<AuditResult> evaluateShardGroup( final ApplicationScope scope,
+ final DirectedEdgeMeta edgeMeta,
+ final ShardEntryGroup group ) {
+
+ //deliberately a no op
+ return Futures.immediateFuture( AuditResult.NOT_CHECKED );
+ }
+ }
+
+
}