GEOMESA-424 cql 'OR' filter query does not deduplicate

* moved deduplication code into QueryPlanner.getIterator method and used it for both duplicatable data and OR queries
* refactoring to be a bit more selective
* added inclusion-exclusion test to AccumuloDataStoreTest

Author: John Armstrong

geomesa-core/src/main/scala/org/locationtech/geomesa/core/index/QueryPlanner.scala

@@ -85,22 +85,40 @@ case class QueryPlanner(schema: String,
                   output: ExplainerOutputType = log): CloseableIterator[Entry[Key,Value]] = {
 
     output(s"Running ${ExplainerOutputType.toString(query)}")
-
     val ff = CommonFactoryFinder.getFilterFactory2
     val isDensity = query.getHints.containsKey(BBOX_KEY)
-    val queries: Iterator[Query] =
-      if(isDensity) {
-        val env = query.getHints.get(BBOX_KEY).asInstanceOf[ReferencedEnvelope]
-        val q1 = new Query(featureType.getTypeName, ff.bbox(ff.property(featureType.getGeometryDescriptor.getLocalName), env))
-        Iterator(DataUtilities.mixQueries(q1, query, "geomesa.mixed.query"))
+    val duplicatableData = IndexSchema.mayContainDuplicates(featureType)
+
+    def flatten(queries: Seq[Query]): CloseableIterator[Entry[Key, Value]] =
+      queries.toIterator.ciFlatMap(configureScanners(acc, sft, _, isDensity, output))
+
+    // in some cases, where duplicates may appear in overlapping queries or the data itself, remove them
+    def deduplicate(queries: Seq[Query]): CloseableIterator[Entry[Key, Value]] = {
+      val flatQueries = flatten(queries)
+      val decoder = SimpleFeatureDecoder(getReturnSFT(query), featureEncoding)
+      new DeDuplicatingIterator(flatQueries, (key: Key, value: Value) => decoder.extractFeatureId(value))
+    }
+
+    if(isDensity) {
+      val env = query.getHints.get(BBOX_KEY).asInstanceOf[ReferencedEnvelope]
+      val q1 = new Query(featureType.getTypeName, ff.bbox(ff.property(featureType.getGeometryDescriptor.getLocalName), env))
+      val mixedQuery = DataUtilities.mixQueries(q1, query, "geomesa.mixed.query")
+      if (duplicatableData) {
+        deduplicate(Seq(mixedQuery))
       } else {
-        splitQueryOnOrs(query, output)
+        flatten(Seq(mixedQuery))
       }
-
-    queries.ciFlatMap(configureScanners(acc, sft, _, isDensity, output))
+    } else {
+      val rawQueries = splitQueryOnOrs(query, output)
+      if (rawQueries.length > 1 || duplicatableData) {
+        deduplicate(rawQueries)
+      } else {
+        flatten(rawQueries)
+      }
+    }
   }
 
-  def splitQueryOnOrs(query: Query, output: ExplainerOutputType): Iterator[Query] = {
+  def splitQueryOnOrs(query: Query, output: ExplainerOutputType): Seq[Query] = {
     val originalFilter = query.getFilter
     output(s"Original filter: $originalFilter")
 
@@ -117,7 +135,7 @@ case class QueryPlanner(schema: String,
       val q = new Query(query)
       q.setFilter(filter)
       q
-    }.toIterator
+    }
   }
 
   /**
@@ -155,20 +173,14 @@ case class QueryPlanner(schema: String,
     val returnSFT = getReturnSFT(query)
     val decoder = SimpleFeatureDecoder(returnSFT, featureEncoding)
 
-    // the final iterator may need duplicates removed
-    val uniqKVIter: CloseableIterator[Entry[Key,Value]] =
-      if (IndexSchema.mayContainDuplicates(featureType))
-        new DeDuplicatingIterator(accumuloIterator, (key: Key, value: Value) => decoder.extractFeatureId(value))
-      else accumuloIterator
-
     // Decode according to the SFT return type.
     // if this is a density query, expand the map
     if (query.getHints.containsKey(DENSITY_KEY)) {
-      uniqKVIter.flatMap { kv: Entry[Key, Value] =>
+      accumuloIterator.flatMap { kv: Entry[Key, Value] =>
         DensityIterator.expandFeature(decoder.decode(kv.getValue))
       }
     } else {
-      uniqKVIter.map { kv => decoder.decode(kv.getValue) }
+      accumuloIterator.map { kv => decoder.decode(kv.getValue) }
     }
   }
 

geomesa-core/src/test/scala/org/locationtech/geomesa/core/data/AccumuloDataStoreTest.scala

@@ -39,7 +39,7 @@ import org.geotools.referencing.CRS
 import org.geotools.referencing.crs.DefaultGeographicCRS
 import org.joda.time.DateTime
 import org.junit.runner.RunWith
-import org.locationtech.geomesa.core.index.{IndexSchemaBuilder, _}
+import org.locationtech.geomesa.core.index._
 import org.locationtech.geomesa.core.iterators.TestData
 import org.locationtech.geomesa.core.security.{AuthorizationsProvider, DefaultAuthorizationsProvider, FilteringAuthorizationsProvider}
 import org.locationtech.geomesa.core.util.{SelfClosingIterator, CloseableIterator}
@@ -53,6 +53,7 @@ import org.specs2.mutable.Specification
 import org.specs2.runner.JUnitRunner
 
 import scala.collection.JavaConversions._
+import scala.util.Random
 
 @RunWith(classOf[JUnitRunner])
 class AccumuloDataStoreTest extends Specification {
@@ -175,6 +176,7 @@ class AccumuloDataStoreTest extends Specification {
           "and there are no results" >> { features.hasNext should be equalTo false }
         }
       }
+
       "process a DWithin query correctly" in {
         // create the data store
         val sftName = "dwithintest"
@@ -210,6 +212,55 @@ class AccumuloDataStoreTest extends Specification {
         "and no more results" >> { features.hasNext must beFalse }
       }
 
+      "process an OR query correctly" in {
+        val sftName = "ortest"
+        val sft = SimpleFeatureTypes.createType(sftName, s"NAME:String,dtg:Date,*geom:Point:srid=4326")
+        sft.getUserData.put(SF_PROPERTY_START_TIME, "dtg")
+        ds.createSchema(sft)
+
+        val fs = ds.getFeatureSource(sftName).asInstanceOf[AccumuloFeatureStore]
+
+        {
+          val randVal: (Double, Double) => Double = {
+            val r = new Random(System.nanoTime())
+            (low, high) => {
+              (r.nextDouble() * (high - low)) + low
+            }
+          }
+          val fc = new DefaultFeatureCollection(sftName, sft)
+          for (i <- 0 until 1000) {
+            val lat = randVal(-0.001, 0.001)
+            val lon = randVal(-0.001, 0.001)
+            val geom = WKTUtils.read(s"POINT($lat $lon)")
+            val builder = new SimpleFeatureBuilder(sft, featureFactory)
+            builder.addAll(List("testType", null, geom))
+            val feature = builder.buildFeature(s"fid-$i")
+            feature.getUserData.put(Hints.USE_PROVIDED_FID, java.lang.Boolean.TRUE)
+            fc.add(feature)
+          }
+          fs.addFeatures(fc)
+        }
+
+        val geomFactory = JTSFactoryFinder.getGeometryFactory
+        val urq = ff.dwithin(ff.property("geom"), ff.literal(geomFactory.createPoint(new Coordinate( 0.0005,  0.0005))), 150.0, "meters")
+        val llq = ff.dwithin(ff.property("geom"), ff.literal(geomFactory.createPoint(new Coordinate(-0.0005, -0.0005))), 150.0, "meters")
+        val orq = ff.or(urq, llq)
+        val andq = ff.and(urq, llq)
+        val urQuery  = new Query(sftName,  urq)
+        val llQuery  = new Query(sftName,  llq)
+        val orQuery  = new Query(sftName,  orq)
+        val andQuery = new Query(sftName, andq)
+
+        val urNum  = fs.getFeatures( urQuery).features.length
+        val llNum  = fs.getFeatures( llQuery).features.length
+        val orNum  = fs.getFeatures( orQuery).features.length
+        val andNum = fs.getFeatures(andQuery).features.length
+
+        "obeying inclusion-exclusion principle" >> {
+          (urNum + llNum) mustEqual (orNum + andNum)
+        }
+      }
+
       "handle transformations" in {
         val sftName = "transformtest1"
         val sft = SimpleFeatureTypes.createType(sftName, s"name:String,dtg:Date,*geom:Point:srid=4326")