usergrid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From toddn...@apache.org
Subject [19/19] incubator-usergrid git commit: Fixes issue with ignored compaction operation on edge delete
Date Tue, 02 Jun 2015 21:17:35 GMT
Fixes issue with ignored compaction operation on edge delete


Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/c582d278
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/c582d278
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/c582d278

Branch: refs/heads/USERGRID-685
Commit: c582d2782fab7d035243cd09b80052588623eb38
Parents: dfda3cf
Author: Todd Nine <tnine@apigee.com>
Authored: Tue Jun 2 15:17:07 2015 -0600
Committer: Todd Nine <tnine@apigee.com>
Committed: Tue Jun 2 15:17:07 2015 -0600

----------------------------------------------------------------------
 .../corepersistence/CpRelationManager.java      |  18 +--
 .../asyncevents/EventBuilderImpl.java           |   4 +-
 .../impl/stage/NodeDeleteListenerImpl.java      | 113 ++++++++-----------
 3 files changed, 52 insertions(+), 83 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c582d278/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
index bc3c803..c717c5b 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
@@ -414,7 +414,7 @@ public class CpRelationManager implements RelationManager {
                     });
                 }
             })
-            .toBlocking().lastOrDefault(null);
+            .toBlocking().lastOrDefault( null );
 
         //check if we need to reverse our edges
 
@@ -622,7 +622,7 @@ public class CpRelationManager implements RelationManager {
 
         final IdBuilder pipelineBuilder =
             pipelineBuilderFactory.create( applicationScope ).withCursor( query.getCursor()
)
-                                  .withLimit( query.getLimit() ).fromId(cpHeadEntity.getId());
+                                  .withLimit( query.getLimit() ).fromId( cpHeadEntity.getId()
);
 
 
         final EntityBuilder results;
@@ -791,18 +791,10 @@ public class CpRelationManager implements RelationManager {
 
         final SearchByEdge search = createConnectionSearchByEdge( sourceId, connectionType,
targetEntity.getId() );
 
-        //delete all the edges
-        final Edge lastEdge =
-            gm.loadEdgeVersions( search ).flatMap( returnedEdge -> gm.markEdge( returnedEdge
) ).toBlocking()
+        //delete all the edges and queue their processing
+        gm.loadEdgeVersions( search ).flatMap( returnedEdge -> gm.markEdge( returnedEdge
) ).doOnNext(
+                returnedEdge -> indexService.queueDeleteEdge( applicationScope, returnedEdge
) ).toBlocking()
               .lastOrDefault( null );
-
-        if ( lastEdge != null ) {
-            final ApplicationEntityIndex ei = managerCache.getEntityIndex( applicationScope
);
-            final EntityIndexBatch batch = ei.createBatch();
-
-            SearchEdge indexScope = createSearchEdgeFromSource( lastEdge );
-            batch.deindex( indexScope, targetEntity );
-        }
     }
 
 

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c582d278/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
index 57f30fd..e4cd4b5 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/EventBuilderImpl.java
@@ -108,9 +108,9 @@ public class EventBuilderImpl implements EventBuilder {
         log.debug( "Deleting in app scope {} with edge {} }", applicationScope, edge );
 
         final Observable<IndexOperationMessage> edgeObservable =
-            indexService.deleteIndexEdge( applicationScope, edge ).doOnCompleted( () ->
{
+            indexService.deleteIndexEdge( applicationScope, edge ).flatMap( batch -> {
                 final GraphManager gm = graphManagerFactory.createEdgeManager( applicationScope
);
-                gm.deleteEdge( edge );
+                return gm.deleteEdge( edge ).map( deletedEdge -> batch );
             } );
 
         return edgeObservable;

http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/c582d278/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java
b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java
index 10ee0f6..e4eb5fc 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/impl/stage/NodeDeleteListenerImpl.java
@@ -158,112 +158,89 @@ public class NodeDeleteListenerImpl implements NodeDeleteListener {
         //get all edges pointing to the target node and buffer then into groups for deletion
         Observable<MarkedEdge> targetEdges =
                 getEdgesTypesToTarget( scope, new SimpleSearchEdgeType( node, null, null
) )
-                        .subscribeOn( Schedulers.io() ).flatMap( new Func1<String, Observable<MarkedEdge>>()
{
-                    @Override
-                    public Observable<MarkedEdge> call( final String edgeType ) {
-                        return Observable.create( new ObservableIterator<MarkedEdge>(
"getTargetEdges" ) {
+                        .subscribeOn( Schedulers.io() ).flatMap( edgeType -> Observable.create(
new ObservableIterator<MarkedEdge>( "getTargetEdges" ) {
                             @Override
                             protected Iterator<MarkedEdge> getIterator() {
                                 return storageSerialization.getEdgesToTarget( scope,
                                         new SimpleSearchByEdgeType( node, edgeType, maxVersion,
SearchByEdgeType.Order.DESCENDING,  Optional.<Edge>absent() ) );
                             }
-                        } );
-                    }
-                } );
+                        } ) );
 
 
         //get all edges pointing to the source node and buffer them into groups for deletion
         Observable<MarkedEdge> sourceEdges =
                 getEdgesTypesFromSource( scope, new SimpleSearchEdgeType( node, null, null
) )
-                        .subscribeOn( Schedulers.io() ).flatMap( new Func1<String, Observable<MarkedEdge>>()
{
-                    @Override
-                    public Observable<MarkedEdge> call( final String edgeType ) {
-                        return Observable.create( new ObservableIterator<MarkedEdge>(
"getSourceEdges" ) {
+                        .subscribeOn( Schedulers.io() ).flatMap( edgeType -> Observable.create(
new ObservableIterator<MarkedEdge>( "getSourceEdges" ) {
                             @Override
                             protected Iterator<MarkedEdge> getIterator() {
                                 return storageSerialization.getEdgesFromSource( scope,
                                         new SimpleSearchByEdgeType( node, edgeType, maxVersion,
SearchByEdgeType.Order.DESCENDING,  Optional.<Edge>absent() ) );
                             }
-                        } );
-                    }
-                } );
+                        } ) );
 
         //merge both source and target into 1 observable.  We'll need to check them all regardless
of order
         return Observable.merge( targetEdges, sourceEdges )
 
                 //buffer and delete marked edges in our buffer size so we're making less
trips to cassandra
-                .buffer( graphFig.getScanPageSize() ).flatMap( new Func1<List<MarkedEdge>,
Observable<MarkedEdge>>() {
-                    @Override
-                    public Observable<MarkedEdge> call( final List<MarkedEdge>
markedEdges ) {
+                .buffer( graphFig.getScanPageSize() ).flatMap( markedEdges -> {
 
-                        LOG.debug( "Batching {} edges for node {} for deletion", markedEdges.size(),
node );
+                    LOG.debug( "Batching {} edges for node {} for deletion", markedEdges.size(),
node );
 
-                        final MutationBatch batch = keyspace.prepareMutationBatch();
+                    final MutationBatch batch = keyspace.prepareMutationBatch();
 
-                        Set<TargetPair> sourceNodes = new HashSet<>( markedEdges.size()
);
-                        Set<TargetPair> targetNodes = new HashSet<>( markedEdges.size()
);
+                    Set<TargetPair> sourceNodes = new HashSet<>( markedEdges.size()
);
+                    Set<TargetPair> targetNodes = new HashSet<>( markedEdges.size()
);
 
-                        for ( MarkedEdge edge : markedEdges ) {
+                    for ( MarkedEdge edge : markedEdges ) {
 
-                            //delete the newest edge <= the version on the node delete
+                        //delete the newest edge <= the version on the node delete
 
-                            //we use the version specified on the delete purposefully.  If
these edges are re-written
-                            //at a greater time we want them to exit
-                            batch.mergeShallow( storageSerialization.deleteEdge( scope, edge,
eventTimestamp ) );
+                        //we use the version specified on the delete purposefully.  If these
edges are re-written
+                        //at a greater time we want them to exit
+                        batch.mergeShallow( storageSerialization.deleteEdge( scope, edge,
eventTimestamp ) );
 
-                            sourceNodes.add( new TargetPair( edge.getSourceNode(), edge.getType()
) );
-                            targetNodes.add( new TargetPair( edge.getTargetNode(), edge.getType()
) );
-                        }
+                        sourceNodes.add( new TargetPair( edge.getSourceNode(), edge.getType()
) );
+                        targetNodes.add( new TargetPair( edge.getTargetNode(), edge.getType()
) );
+                    }
 
-                        try {
-                            batch.execute();
-                        }
-                        catch ( ConnectionException e ) {
-                            throw new RuntimeException( "Unable to connect to casandra",
e );
-                        }
+                    try {
+                        batch.execute();
+                    }
+                    catch ( ConnectionException e ) {
+                        throw new RuntimeException( "Unable to connect to casandra", e );
+                    }
 
-                        //now  delete meta data
+                    //now  delete meta data
 
 
-                        //delete both the source and target meta data in parallel for the
edge we deleted in the
-                        // previous step
-                        //if nothing else is using them.  We purposefully do not schedule
them on a new scheduler
-                        //we want them running on the i/o thread from the Observable emitting
all the edges
+                    //delete both the source and target meta data in parallel for the edge
we deleted in the
+                    // previous step
+                    //if nothing else is using them.  We purposefully do not schedule them
on a new scheduler
+                    //we want them running on the i/o thread from the Observable emitting
all the edges
 
-                        //
-                        LOG.debug( "About to audit {} source types", sourceNodes.size() );
+                    //
+                    LOG.debug( "About to audit {} source types", sourceNodes.size() );
 
-                        Observable<Integer> sourceMetaCleanup =
-                                Observable.from( sourceNodes ).flatMap( new Func1<TargetPair,
Observable<Integer>>() {
-                                    @Override
-                                    public Observable<Integer> call( final TargetPair
targetPair ) {
-                                        return edgeMetaRepair
-                                                .repairSources( scope, targetPair.id, targetPair.edgeType,
maxVersion );
-                                    }
-                                } ).last();
+                    Observable<Integer> sourceMetaCleanup =
+                            Observable.from( sourceNodes ).flatMap( targetPair -> edgeMetaRepair
+                                    .repairSources( scope, targetPair.id, targetPair.edgeType,
maxVersion ) ).last();
 
 
-                        LOG.debug( "About to audit {} target types", targetNodes.size() );
+                    LOG.debug( "About to audit {} target types", targetNodes.size() );
 
-                        Observable<Integer> targetMetaCleanup =
-                                Observable.from( targetNodes ).flatMap( new Func1<TargetPair,
Observable<Integer>>() {
-                                    @Override
-                                    public Observable<Integer> call( final TargetPair
targetPair ) {
-                                        return edgeMetaRepair
-                                                .repairTargets( scope, targetPair.id, targetPair.edgeType,
maxVersion );
-                                    }
-                                } ).last();
+                    Observable<Integer> targetMetaCleanup =
+                            Observable.from( targetNodes ).flatMap( targetPair -> edgeMetaRepair
+                                    .repairTargets( scope, targetPair.id, targetPair.edgeType,
maxVersion ) ).last();
 
 
-                        //run both the source/target edge type cleanup, then proceed
-                        return Observable.merge( sourceMetaCleanup, targetMetaCleanup ).lastOrDefault(
null )
-                                         .flatMap( new Func1<Integer, Observable<MarkedEdge>>()
{
-                                             @Override
-                                             public Observable<MarkedEdge> call( final
Integer integer ) {
-                                                 return Observable.from( markedEdges );
-                                             }
-                                         } );
-                    }
+                    //run both the source/target edge type cleanup, then proceed
+                    return Observable.merge( sourceMetaCleanup, targetMetaCleanup ).lastOrDefault(
null )
+                                     .flatMap( new Func1<Integer, Observable<MarkedEdge>>()
{
+                                         @Override
+                                         public Observable<MarkedEdge> call( final
Integer integer ) {
+                                             return Observable.from( markedEdges );
+                                         }
+                                     } );
                 } );
     }
 


Mime
View raw message