ATLAS-2044: In-memory filtering for correctness after index query

repository/src/main/java/org/apache/atlas/discovery/ClassificationSearchProcessor.java

@@ -87,6 +87,8 @@ public class ClassificationSearchProcessor extends SearchProcessor {
             indexQueryString = STRAY_ELIPSIS_PATTERN.matcher(indexQueryString).replaceAll("");
 
             this.indexQuery = graph.indexQuery(Constants.VERTEX_INDEX, indexQueryString);
+
+            constructInMemoryPredicate(classificationType, filterCriteria, indexAttributes);
         } else {
             indexQuery = null;
         }
@@ -184,6 +186,9 @@ public class ClassificationSearchProcessor extends SearchProcessor {
                     getVertices(queryResult, classificationVertices);
                 }
 
+                // Do in-memory filtering before the graph query
+                CollectionUtils.filter(classificationVertices, inMemoryPredicate);
+
                 for (AtlasVertex classificationVertex : classificationVertices) {
                     Iterable<AtlasEdge> edges = classificationVertex.getEdges(AtlasEdgeDirection.IN);
 
@@ -208,19 +213,7 @@ public class ClassificationSearchProcessor extends SearchProcessor {
 
                 super.filter(entityVertices);
 
-                for (AtlasVertex entityVertex : entityVertices) {
-                    resultIdx++;
-
-                    if (resultIdx <= startIdx) {
-                        continue;
-                    }
-
-                    ret.add(entityVertex);
-
-                    if (ret.size() == limit) {
-                        break;
-                    }
-                }
+                resultIdx = collectResultVertices(ret, startIdx, limit, resultIdx, entityVertices);
             }
         } finally {
             AtlasPerfTracer.log(perf);

repository/src/main/java/org/apache/atlas/discovery/EntitySearchProcessor.java

@@ -71,6 +71,8 @@ public class EntitySearchProcessor extends SearchProcessor {
 
         if (attrSearchByIndex) {
             constructFilterQuery(indexQuery, entityType, filterCriteria, indexAttributes);
+
+            constructInMemoryPredicate(entityType, filterCriteria, indexAttributes);
         } else {
             graphAttributes.addAll(indexAttributes);
         }
@@ -165,11 +167,10 @@ public class EntitySearchProcessor extends SearchProcessor {
                         break;
                     }
 
-                    while (idxQueryResult.hasNext()) {
-                        AtlasVertex vertex = idxQueryResult.next().getVertex();
+                    getVerticesFromIndexQueryResult(idxQueryResult, entityVertices);
 
-                        entityVertices.add(vertex);
-                    }
+                    // Do in-memory filtering before the graph query
+                    CollectionUtils.filter(entityVertices, inMemoryPredicate);
 
                     if (graphQuery != null) {
                         AtlasGraphQuery guidQuery = context.getGraph().query().in(Constants.GUID_PROPERTY_KEY, getGuids(entityVertices));
@@ -191,19 +192,7 @@ public class EntitySearchProcessor extends SearchProcessor {
 
                 super.filter(entityVertices);
 
-                for (AtlasVertex entityVertex : entityVertices) {
-                    resultIdx++;
-
-                    if (resultIdx <= startIdx) {
-                        continue;
-                    }
-
-                    ret.add(entityVertex);
-
-                    if (ret.size() == limit) {
-                        break;
-                    }
-                }
+                resultIdx = collectResultVertices(ret, startIdx, limit, resultIdx, entityVertices);
             }
         } finally {
             AtlasPerfTracer.log(perf);

repository/src/main/java/org/apache/atlas/discovery/FullTextSearchProcessor.java

@@ -141,19 +141,7 @@ public class FullTextSearchProcessor extends SearchProcessor {
 
                 super.filter(entityVertices);
 
-                for (AtlasVertex entityVertex : entityVertices) {
-                    resultIdx++;
-
-                    if (resultIdx <= startIdx) {
-                        continue;
-                    }
-
-                    ret.add(entityVertex);
-
-                    if (ret.size() == limit) {
-                        break;
-                    }
-                }
+                resultIdx = collectResultVertices(ret, startIdx, limit, resultIdx, entityVertices);
             }
         } finally {
             AtlasPerfTracer.log(perf);