You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sf...@apache.org on 2014/10/10 17:50:47 UTC
[40/43] git commit: Updated tests and added loader debugging
Updated tests and added loader debugging
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/38d5885a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/38d5885a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/38d5885a
Branch: refs/heads/two-dot-o-events
Commit: 38d5885a9d6243d36e968f88cce32aeea711bdf1
Parents: c0f79d1
Author: Todd Nine <to...@apache.org>
Authored: Thu Oct 9 23:38:39 2014 -0600
Committer: Todd Nine <to...@apache.org>
Committed: Thu Oct 9 23:38:39 2014 -0600
----------------------------------------------------------------------
.../corepersistence/CpRelationManager.java | 7 +-
.../usergrid/corepersistence/CpWalker.java | 2 +
.../corepersistence/results/EntityVerifier.java | 19 ++--
.../results/FilteringLoader.java | 103 ++++++++++---------
.../corepersistence/StaleIndexCleanupTest.java | 34 +++++-
.../index/query/CandidateResult.java | 30 ++++++
6 files changed, 129 insertions(+), 66 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/38d5885a/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 3d5445c..00c48c3 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
@@ -913,9 +913,11 @@ public class CpRelationManager implements RelationManager {
int maxQueries = 10; // max re-queries to satisfy query limit
+ final int originalLimit = query.getLimit();
+
Results results = null;
int queryCount = 0;
- int originalLimit = query.getLimit();
+
boolean satisfied = false;
while ( !satisfied && queryCount++ < maxQueries ) {
@@ -933,7 +935,7 @@ public class CpRelationManager implements RelationManager {
if ( crs.isEmpty() || !crs.hasCursor() ) { // no results, no cursor, can't get more
satisfied = true;
- } else if ( results.size() == query.getLimit() ) { // got what we need
+ } else if ( results.size() == originalLimit ) { // got what we need
satisfied = true;
} else if ( crs.hasCursor() ) {
@@ -1487,6 +1489,7 @@ public class CpRelationManager implements RelationManager {
if ( query.getSortPredicates().isEmpty() ) {
+ //TODO, should this be descending?
Query.SortPredicate asc = new Query.SortPredicate(
PROPERTY_CREATED, Query.SortDirection.ASCENDING );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/38d5885a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
index 584f9a8..e780b77 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
@@ -17,6 +17,8 @@ package org.apache.usergrid.corepersistence;
import java.util.Stack;
import java.util.logging.Level;
+
+import org.apache.usergrid.corepersistence.util.CpNamingUtils;
import org.apache.usergrid.persistence.EntityRef;
import static org.apache.usergrid.persistence.Schema.TYPE_APPLICATION;
import org.apache.usergrid.persistence.collection.CollectionScope;
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/38d5885a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/EntityVerifier.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/EntityVerifier.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/EntityVerifier.java
index 958cebe..13bd91b 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/EntityVerifier.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/EntityVerifier.java
@@ -89,30 +89,31 @@ public class EntityVerifier implements ResultsVerifier {
public boolean isValid( final CandidateResult candidateResult ) {
final Id entityId = candidateResult.getId();
- final MvccEntity version = ids.getEntity( entityId );
+ final MvccEntity savedEntity = ids.getEntity( entityId );
- //version wasn't found ,deindex
- if ( version == null ) {
+ //version wasn't found deindex
+ if ( savedEntity == null ) {
logger.warn( "Version for Entity {}:{} not found", entityId.getUuid(), entityId.getUuid() );
-
-
return false;
}
- final UUID savedVersion = version.getVersion();
+ final UUID candidateVersion = candidateResult.getVersion();
+ final UUID savedVersion = savedEntity.getVersion();
- if ( UUIDComparator.staticCompare( savedVersion, candidateResult.getVersion() ) > 0 ) {
+ if ( UUIDComparator.staticCompare( savedVersion, candidateVersion ) > 0 ) {
logger.debug( "Stale version of Entity uuid:{} type:{}, stale v:{}, latest v:{}", new Object[] {
- entityId.getUuid(), entityId.getType(), savedVersion, candidateResult.getVersion()
+ entityId.getUuid(), entityId.getType(), candidateVersion, savedEntity
} );
return false;
}
- final Optional<org.apache.usergrid.persistence.model.entity.Entity> entity = version.getEntity();
+ final Optional<org.apache.usergrid.persistence.model.entity.Entity> entity = savedEntity.getEntity();
if ( !entity.isPresent() ) {
+ logger.warn( "Entity uuid:{} version v:{} is deleted but indexed, this is a bug ", entityId.getUuid(),
+ savedEntity.getEntity() );
return false;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/38d5885a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/FilteringLoader.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/FilteringLoader.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/FilteringLoader.java
index 40ff10d..a7eb6ea 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/FilteringLoader.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/FilteringLoader.java
@@ -84,13 +84,13 @@ public class FilteringLoader implements ResultsLoader {
private final ApplicationScope applicationScope;
- protected FilteringLoader( final CpManagerCache managerCache, final ResultsVerifier resultsLoader,
- final EntityRef ownerId, final ApplicationScope applicationScope ) {
- this.managerCache = managerCache;
- this.resultsLoader = resultsLoader;
- this.ownerId = new SimpleId( ownerId.getUuid(), ownerId.getType());
- this.applicationScope = applicationScope;
- }
+ protected FilteringLoader( final CpManagerCache managerCache, final ResultsVerifier resultsLoader,
+ final EntityRef ownerId, final ApplicationScope applicationScope ) {
+ this.managerCache = managerCache;
+ this.resultsLoader = resultsLoader;
+ this.ownerId = new SimpleId( ownerId.getUuid(), ownerId.getType() );
+ this.applicationScope = applicationScope;
+ }
@Override
@@ -101,10 +101,21 @@ public class FilteringLoader implements ResultsLoader {
final EntityIndexBatch indexBatch = index.createBatch();
+ /**
+ * For each entity, holds the index it appears in our candidates for keeping ordering correct
+ */
final Map<Id, Integer> orderIndex = new HashMap<>( crs.size() );
- final Map<Id, CandidateResult> idResultMapping = new HashMap<>( crs.size() );
+ /**
+ * Maps the entity ids to our candidates
+ */
+ final Map<Id, CandidateResult> maxCandidateMapping = new HashMap<>( crs.size() );
+ /**
+ * Groups all candidate results by types. When search connections there will be multiple types,
+ * so we want to batch
+ * fetch them more efficiently
+ */
final HashMultimap<String, CandidateResult> groupedByScopes = HashMultimap.create( crs.size(), crs.size() );
final Iterator<CandidateResult> iter = crs.iterator();
@@ -115,48 +126,48 @@ public class FilteringLoader implements ResultsLoader {
*/
for ( int i = 0; iter.hasNext(); i++ ) {
- final CandidateResult cr = iter.next();
-
- final String collectionType = CpNamingUtils.getCollectionScopeNameFromEntityType( cr.getId().getType() );
+ final CandidateResult currentCandidate = iter.next();
- final Id entityId = cr.getId();
+ final String collectionType = CpNamingUtils.getCollectionScopeNameFromEntityType( currentCandidate.getId().getType() );
- //if we've already seen this one, put which ever is greater
+ final Id entityId = currentCandidate.getId();
- final CandidateResult previousMax = idResultMapping.get( entityId );
+ //check if we've seen this candidate by id
+ final CandidateResult previousMax = maxCandidateMapping.get( entityId );
+ //its not been seen, save it
if ( previousMax == null ) {
- idResultMapping.put( entityId, cr );
+ maxCandidateMapping.put( entityId, currentCandidate );
orderIndex.put( entityId, i );
- groupedByScopes.put( collectionType, cr );
+ groupedByScopes.put( collectionType, currentCandidate );
+ continue;
}
//we have seen it, compare them
- else {
- final UUID previousMaxVersion = previousMax.getVersion();
+ final UUID previousMaxVersion = previousMax.getVersion();
- final UUID currentVersion = cr.getVersion();
+ final UUID currentVersion = currentCandidate.getVersion();
- //this is a newer version, we know we already have a stale entity, add it to be cleaned up
- if ( UUIDComparator.staticCompare( currentVersion, previousMaxVersion ) > 0 ) {
+ //this is a newer version, we know we already have a stale entity, add it to be cleaned up
+ if ( UUIDComparator.staticCompare( currentVersion, previousMaxVersion ) > 0 ) {
- //de-index it
- logger.debug( "Stale version of Entity uuid:{} type:{}, stale v:{}, latest v:{}", new Object[] {
- entityId.getUuid(), entityId.getType(), previousMax, currentVersion
- } );
+ //de-index it
+ logger.debug( "Stale version of Entity uuid:{} type:{}, stale v:{}, latest v:{}", new Object[] {
+ entityId.getUuid(), entityId.getType(), previousMaxVersion, currentVersion
+ } );
- //deindex
- deIndex( indexBatch, ownerId, previousMax );
+ //deindex this document, and remove the previous maxVersion
+ deIndex( indexBatch, ownerId, previousMax );
+ groupedByScopes.remove( collectionType, previousMax );
- //TODO, fire the entity repair cleanup task here instead of de-indexing
+ //TODO, fire the entity repair cleanup task here instead of de-indexing
- //replace the value with a more current version
- idResultMapping.put( entityId, cr );
- orderIndex.put( entityId, i );
- groupedByScopes.put( collectionType, cr );
- }
+ //replace the value with a more current version
+ maxCandidateMapping.put( entityId, currentCandidate );
+ orderIndex.put( entityId, i );
+ groupedByScopes.put( collectionType, currentCandidate );
}
}
@@ -166,7 +177,7 @@ public class FilteringLoader implements ResultsLoader {
final TreeMap<Integer, Id> sortedResults = new TreeMap<>();
- for ( final String scopeName : groupedByScopes.keys() ) {
+ for ( final String scopeName : groupedByScopes.keySet() ) {
final Set<CandidateResult> candidateResults = groupedByScopes.get( scopeName );
@@ -198,30 +209,25 @@ public class FilteringLoader implements ResultsLoader {
final EntityCollectionManager ecm = managerCache.getEntityCollectionManager( collScope );
- //load the results into the loader for this ech
+ //load the results into the loader for this scope for validation
resultsLoader.loadResults( idsToLoad, ecm );
-
-
-
-
- //now compare them
-
+ //now let the loader validate each candidate. For instance, the "max" in this candidate
+ //could still be a stale result, so it needs validated
for ( final Id requestedId : idsToLoad ) {
- final CandidateResult cr = idResultMapping.get( requestedId );
+ final CandidateResult cr = maxCandidateMapping.get( requestedId );
- //ask the loader if this is valid, if not discard it
- if(!resultsLoader.isValid( cr )){
+ //ask the loader if this is valid, if not discard it and de-index it
+ if ( !resultsLoader.isValid( cr ) ) {
deIndex( indexBatch, ownerId, cr );
continue;
}
//if we get here we're good, we need to add this to our results
- final int candidateIndex = orderIndex.get( requestedId );
+ final int candidateIndex = orderIndex.get( requestedId );
sortedResults.put( candidateIndex, requestedId );
-
}
}
@@ -229,13 +235,10 @@ public class FilteringLoader implements ResultsLoader {
//execute the cleanup
indexBatch.execute();
- return resultsLoader.getResults(sortedResults.values());
+ return resultsLoader.getResults( sortedResults.values() );
}
-
-
-
protected void deIndex( final EntityIndexBatch batch, final Id ownerId, final CandidateResult candidateResult ) {
IndexScope indexScope = new IndexScopeImpl( ownerId,
CpNamingUtils.getCollectionScopeNameFromEntityType( candidateResult.getId().getType() ) );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/38d5885a/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
index 757d590..1645ddf 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/StaleIndexCleanupTest.java
@@ -111,7 +111,7 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
final int numUpdates = 5;
// create lots of entities
- final List<Entity> things = new ArrayList<Entity>();
+ final List<Entity> things = new ArrayList<Entity>(numEntities);
for ( int i=0; i<numEntities; i++) {
final String thingName = "thing" + i;
things.add( em.create("thing", new HashMap<String, Object>() {{
@@ -126,12 +126,17 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
// update each one a bunch of times
int count = 0;
+
+ List<Entity> maxVersions = new ArrayList<>(numEntities);
+
for ( Entity thing : things ) {
+ Entity toUpdate = null;
+
for ( int j=0; j<numUpdates; j++) {
- Entity toUpdate = em.get( thing.getUuid() );
- thing.setProperty( "property" + j, RandomStringUtils.randomAlphanumeric(10));
+ toUpdate = em.get( thing.getUuid() );
+ toUpdate.setProperty( "property" + j, RandomStringUtils.randomAlphanumeric(10));
em.update(toUpdate);
Thread.sleep( writeDelayMs );
@@ -141,6 +146,8 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
logger.info("Updated {} of {} times", count, numEntities * numUpdates);
}
}
+
+ maxVersions.add( toUpdate );
}
em.refreshIndex();
@@ -150,19 +157,36 @@ public class StaleIndexCleanupTest extends AbstractCoreIT {
// query EntityManager for results and page through them
// should return numEntities becuase it filters out the stale entities
+ final int limit = 8;
Query q = Query.fromQL("select *");
- q.setLimit( 8 );
+ q.setLimit( limit );
int thingCount = 0;
String cursor = null;
+
+
+
+ int index = 0;
+
do {
Results results = em.searchCollection( em.getApplicationRef(), "things", q);
cursor = results.getCursor();
if ( cursor != null ) {
- assertEquals( 8, results.size() );
+ assertEquals( limit, results.size() );
}
thingCount += results.size();
+ for(int i = 0; i < results.size(); i ++, index++){
+ final Entity returned = results.getEntities().get( i);
+ //last entities appear first
+ final Entity expected = maxVersions.get( index );
+
+ assertEquals("correct entity returned", expected, returned);
+
+ }
+
+
} while ( cursor != null );
+
assertEquals( "Expect no stale candidates", numEntities, thingCount );
// EntityManager should have kicked off a batch cleanup of those stale indexes
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/38d5885a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/CandidateResult.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/CandidateResult.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/CandidateResult.java
index 34c53d3..f49cf2d 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/CandidateResult.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/CandidateResult.java
@@ -41,4 +41,34 @@ public class CandidateResult implements EntityVersion {
public Id getId() {
return entityId;
}
+
+
+ @Override
+ public boolean equals( final Object o ) {
+ if ( this == o ) {
+ return true;
+ }
+ if ( !( o instanceof CandidateResult ) ) {
+ return false;
+ }
+
+ final CandidateResult that = ( CandidateResult ) o;
+
+ if ( !entityId.equals( that.entityId ) ) {
+ return false;
+ }
+ if ( !entityVersion.equals( that.entityVersion ) ) {
+ return false;
+ }
+
+ return true;
+ }
+
+
+ @Override
+ public int hashCode() {
+ int result = entityId.hashCode();
+ result = 31 * result + entityVersion.hashCode();
+ return result;
+ }
}