usergrid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mru...@apache.org
Subject [10/16] usergrid git commit: Several indexing fixes:
Date Tue, 23 Oct 2018 03:17:48 GMT
Several indexing fixes:

1. remove possible duplicate entities from query results when using collection
   setting indexConsistency=latest
2. deindex all but latest candidate for an entity, even when using
   indexConsistency=latest
3. fix indexing bug when collection settings have been added but fields
   setting does not exist
4. by default deindex deleted entity even when indexing is off for the
   collection (this will clean up old index documents) - this is controlled by
   usergrid.entityManager.deindex_deleted_when_collection_indexing_off
   property.


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

Branch: refs/heads/master
Commit: 1feb3dbab88145c9d5483b05bced03c2f6bafa79
Parents: 35a22ed
Author: Mike Dunker <mdunker@google.com>
Authored: Thu Mar 8 00:32:39 2018 -0800
Committer: Keyur Karnik <keyur.karnik+github@gmail.com>
Committed: Tue Aug 28 16:41:44 2018 -0700

----------------------------------------------------------------------
 .../corepersistence/CpEntityManager.java        |   4 +-
 .../corepersistence/EntityManagerFig.java       |   4 +
 .../corepersistence/index/IndexServiceImpl.java |   5 +
 .../read/search/CandidateEntityFilter.java      | 109 +++++++++++++++----
 .../queue/settings/IndexConsistency.java        |   2 +-
 5 files changed, 99 insertions(+), 25 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/usergrid/blob/1feb3dba/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
index 7a4c781..ed33201 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
@@ -638,7 +638,9 @@ public class CpEntityManager implements EntityManager {
 
         Id entityId = new SimpleId( entityRef.getUuid(), entityRef.getType() );
 
-        if ( !skipIndexingForType( entityId.getType() ) ) {
+        // may still want to delete index entries even if indexing is turned off for new
updates
+        if ( entityManagerFig.deindexDeletedWhenCollectionIndexingOff() ||
+            !skipIndexingForType( entityId.getType() ) ) {
             indexService.queueEntityDelete( applicationScope, entityId );
         }
 

http://git-wip-us.apache.org/repos/asf/usergrid/blob/1feb3dba/stack/core/src/main/java/org/apache/usergrid/corepersistence/EntityManagerFig.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/EntityManagerFig.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/EntityManagerFig.java
index 3c8a53f..23cf1c3 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/EntityManagerFig.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/EntityManagerFig.java
@@ -42,6 +42,10 @@ public interface EntityManagerFig extends GuicyFig {
     @Default( "false" )
     boolean getDeindexOnUpdate();
 
+    @Key( "usergrid.entityManager.deindex_deleted_when_collection_indexing_off")
+    @Default( "true" )
+    boolean deindexDeletedWhenCollectionIndexingOff();
+
     /**
      * Comma-separated list of one or more Amazon regions to use if multiregion
      * is set to true.

http://git-wip-us.apache.org/repos/asf/usergrid/blob/1feb3dba/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java
index 32470f6..b1d493e 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexServiceImpl.java
@@ -199,6 +199,11 @@ public class IndexServiceImpl implements IndexService {
 
             Object fields = jsonMapData.get("fields");
 
+            // if "fields" field doesn't exist, should treat like fields=all
+            if ( fields == null ) {
+                return Optional.absent();
+            }
+
             if ( fields != null && fields instanceof String && "all".equalsIgnoreCase(fields.toString()))
{
                 return Optional.absent();
             }

http://git-wip-us.apache.org/repos/asf/usergrid/blob/1feb3dba/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateEntityFilter.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateEntityFilter.java
b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateEntityFilter.java
index 5dcbd27..f9c93b0 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateEntityFilter.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/search/CandidateEntityFilter.java
@@ -21,6 +21,7 @@ package org.apache.usergrid.corepersistence.pipeline.read.search;
 
 
 import java.util.*;
+import java.util.logging.Filter;
 
 import org.apache.usergrid.corepersistence.index.IndexLocationStrategyFactory;
 import org.apache.usergrid.persistence.index.*;
@@ -234,6 +235,7 @@ public class CandidateEntityFilter extends AbstractFilter<FilterResult<Candidate
         private final List<FilterResult<Candidate>> candidateResults;
         private final IndexProducer indexProducer;
         private final EntitySet entitySet;
+        private List<FilterResult<Candidate>> dedupedCandidateResults = new ArrayList<>();
 
 
         public EntityVerifier( final EntityIndexBatch batch, final EntitySet entitySet,
@@ -252,7 +254,9 @@ public class CandidateEntityFilter extends AbstractFilter<FilterResult<Candidate
          */
         public void merge(boolean keepStaleEntries, String query) {
 
-            for ( final FilterResult<Candidate> candidateResult : candidateResults
) {
+            filterDuplicateCandidates(query);
+
+            for ( final FilterResult<Candidate> candidateResult : dedupedCandidateResults
) {
                 validate( candidateResult , keepStaleEntries, query);
             }
 
@@ -287,6 +291,69 @@ public class CandidateEntityFilter extends AbstractFilter<FilterResult<Candidate
         }
 
 
+        // don't need to worry about whether we are keeping stale entries -- this will remove
candidates that are
+        // older than others in the result set
+        private void filterDuplicateCandidates(String query) {
+
+            Map<Id, UUID> latestEntityVersions = new HashMap<>();
+
+            // walk through candidates and find latest version for each entityID
+            for ( final FilterResult<Candidate> filterResult : candidateResults ) {
+                final Candidate candidate = filterResult.getValue();
+                final CandidateResult candidateResult = candidate.getCandidateResult();
+                final Id candidateId = candidateResult.getId();
+                final UUID candidateVersion = candidateResult.getVersion();
+
+                UUID previousCandidateVersion = latestEntityVersions.get(candidateId);
+                if (previousCandidateVersion != null) {
+                    // replace if newer
+                    if (UUIDComparator.staticCompare(candidateVersion, previousCandidateVersion)
> 0) {
+                        latestEntityVersions.put(candidateId, candidateVersion);
+                    }
+                } else {
+                    latestEntityVersions.put(candidateId, candidateVersion);
+                }
+            }
+
+            // walk through candidates again, saving newest results and deindexing older
+            for ( final FilterResult<Candidate> filterResult : candidateResults ) {
+                final Candidate candidate = filterResult.getValue();
+                final CandidateResult candidateResult = candidate.getCandidateResult();
+                final Id candidateId = candidateResult.getId();
+                final UUID candidateVersion = candidateResult.getVersion();
+
+                final UUID latestCandidateVersion = latestEntityVersions.get(candidateId);
+
+                if (candidateVersion.equals(latestCandidateVersion)) {
+                    // save candidate
+                    dedupedCandidateResults.add(filterResult);
+                } else {
+                    // deindex if not the current version in database
+                    final MvccEntity entity = entitySet.getEntity( candidateId );
+                    final UUID databaseVersion = entity.getVersion();
+
+                    if (!candidateVersion.equals(databaseVersion)) {
+                        Date candidateTimeStamp = UUIDTimeStampToDate(candidateVersion);
+                        Date entityTimeStamp = UUIDTimeStampToDate(databaseVersion);
+
+                        logger.warn( "Found old stale entity on edge {} for entityId {} Entity
version {} ({}).  Candidate version {} ({}). Will not be returned in result set. Query = [{}]",
+                            candidate.getSearchEdge(),
+                            entity.getId().getUuid(),
+                            databaseVersion,
+                            DateUtils.instance.formatIso8601Date(entityTimeStamp),
+                            candidateVersion,
+                            DateUtils.instance.formatIso8601Date(candidateTimeStamp),
+                            query
+                        );
+
+                        final SearchEdge searchEdge = candidate.getSearchEdge();
+                        batch.deindex(searchEdge, entity.getId(), candidateVersion);
+                    }
+                }
+            }
+        }
+
+
         private void validate( final FilterResult<Candidate> filterResult, boolean
keepStaleEntries, String query ) {
 
             final Candidate candidate = filterResult.getValue();
@@ -314,7 +381,7 @@ public class CandidateEntityFilter extends AbstractFilter<FilterResult<Candidate
             }
 
 
-            final UUID entityVersion = entity.getVersion();
+            final UUID databaseVersion = entity.getVersion();
             final Id entityId = entity.getId();
 
             // The entity is marked as deleted
@@ -323,7 +390,7 @@ public class CandidateEntityFilter extends AbstractFilter<FilterResult<Candidate
                 // when updating entities, we don't delete all previous versions from ES
so this action is expected
                 if(logger.isDebugEnabled()){
                     logger.debug( "Deindexing deleted entity on edge {} for entityId {} and
version {}",
-                        searchEdge, entityId, entityVersion);
+                        searchEdge, entityId, databaseVersion);
                 }
 
                 batch.deindex( searchEdge, entityId, candidateVersion );
@@ -331,25 +398,21 @@ public class CandidateEntityFilter extends AbstractFilter<FilterResult<Candidate
             }
 
             // entity exists and is newer than ES version, could be a missed or slow index
event
-            if ( UUIDComparator.staticCompare(entityVersion, candidateVersion) > 0 ) {
+            if ( UUIDComparator.staticCompare(databaseVersion, candidateVersion) > 0 )
{
 
                Date candidateTimeStamp = UUIDTimeStampToDate(candidateVersion);
-               Date entityTimeStamp = UUIDTimeStampToDate(entityVersion);
-
-               Map<String,String> fields = new HashMap<>();
-               for  (Field field : entity.getEntity().get().getFields()) {
-                   fields.put(field.getName(),String.valueOf(field.getValue()));
-               }
-
-               logger.warn( "Found stale entity on edge {} for entityId {} Entity version
date = {}.  Candidate version date = {}. Will be returned in result set = {} Query = [{}]
Entity fields = {}",
-                   searchEdge,
-                   entityId.getUuid(),
-                   DateUtils.instance.formatIso8601Date(entityTimeStamp),
-                   DateUtils.instance.formatIso8601Date(candidateTimeStamp),
-                   keepStaleEntries,
-                   query,
-                   fields
-               );
+               Date entityTimeStamp = UUIDTimeStampToDate(databaseVersion);
+
+                logger.warn( "Found stale entity on edge {} for entityId {} Entity version
{} ({}).  Candidate version {} ({}). Will be returned in result set = {} Query = [{}]",
+                    candidate.getSearchEdge(),
+                    entity.getId().getUuid(),
+                    databaseVersion,
+                    DateUtils.instance.formatIso8601Date(entityTimeStamp),
+                    candidateVersion,
+                    DateUtils.instance.formatIso8601Date(candidateTimeStamp),
+                    keepStaleEntries,
+                    query
+                );
 
                 if (!keepStaleEntries) {
                     batch.deindex(searchEdge, entityId, candidateVersion);
@@ -359,18 +422,18 @@ public class CandidateEntityFilter extends AbstractFilter<FilterResult<Candidate
 
             //ES is newer than cass, it means we haven't repaired the record in Cass, we
don't want to
             //remove the ES record, since the read in cass should cause a read repair, just
ignore
-            if ( UUIDComparator.staticCompare( candidateVersion, entityVersion ) > 0 )
{
+            if ( UUIDComparator.staticCompare( candidateVersion, databaseVersion ) > 0
) {
 
                 logger.warn(
                     "Found a newer version in ES over cassandra for edge {} for entityId
{} and version {}.  Repair should be run",
-                        searchEdge, entityId, entityVersion);
+                        searchEdge, entityId, databaseVersion);
 
                   //TODO trigger an audit after a fail count where we explicitly try to repair
from other regions
 
                 return;
             }
 
-            //they're the same add it
+            // add the result
 
             final Entity returnEntity = entity.getEntity().get();
             if(isGeo){

http://git-wip-us.apache.org/repos/asf/usergrid/blob/1feb3dba/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/settings/IndexConsistency.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/settings/IndexConsistency.java
b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/settings/IndexConsistency.java
index 531716a..fdbd399 100644
--- a/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/settings/IndexConsistency.java
+++ b/stack/corepersistence/queue/src/main/java/org/apache/usergrid/persistence/queue/settings/IndexConsistency.java
@@ -51,7 +51,7 @@ public enum IndexConsistency {
     public static IndexConsistency get(String name) {
         IndexConsistency queueIndexingStrategy =  NAME_MAP.get(name);
         if (queueIndexingStrategy == null) {
-            return LATEST;
+            return STRICT;
         }
         return queueIndexingStrategy;
     }


Mime
View raw message