GEOMESA-3452 Accumulo - Use number of tablets scanned for query planning (#3287)

* `stats` cost evaluation for Accumulo now uses tablets scanned instead
  of cached stat estimates
* Query interceptors moved to a scan-time check instead of plan time

Author: elahrvivaz

docs/user/upgrade.rst

@@ -93,6 +93,12 @@ The following dependencies have been upgraded:
 * scala 2.13 ``2.13.12`` -> ``2.13.16``
 * spark ``3.5.5`` -> ``3.5.7``
 
+StrategyDecider API Update
+--------------------------
+
+The ``org.locationtech.geomesa.index.planning.StrategyDecider`` API has been changed slightly to provide more context for
+implementations to use. The old API method has been deprecated and will be removed in a future version.
+
 Removed Modules
 ---------------
 
@@ -125,6 +131,10 @@ Deprecated Classes
 * ``org.locationtech.geomesa.utils.stats.Cardinality`` - replaced with ``org.locationtech.geomesa.utils.index.Cardinality``
 * ``org.locationtech.geomesa.utils.stats.IndexCoverage`` - replaced with ``org.locationtech.geomesa.utils.index.IndexCoverage``
 
+Deprecated Methods
+------------------
+
+* ``org.locationtech.geomesa.index.planning.StrategyDecider.selectFilterPlan``
 
 Internal API Changes
 --------------------

geomesa-accumulo/geomesa-accumulo-datastore/src/main/scala/org/locationtech/geomesa/accumulo/data/AccumuloDataStore.scala

@@ -25,7 +25,7 @@ import org.locationtech.geomesa.accumulo.data.stats._
 import org.locationtech.geomesa.accumulo.index._
 import org.locationtech.geomesa.accumulo.iterators.{AgeOffIterator, DtgAgeOffIterator, ProjectVersionIterator, VisibilityIterator}
 import org.locationtech.geomesa.filter.FilterHelper
-import org.locationtech.geomesa.index.api.{FilterStrategy, GeoMesaFeatureIndex}
+import org.locationtech.geomesa.index.api.{FilterStrategy, GeoMesaFeatureIndex, QueryStrategy}
 import org.locationtech.geomesa.index.geotools.GeoMesaDataStore
 import org.locationtech.geomesa.index.index.attribute.AttributeIndex
 import org.locationtech.geomesa.index.index.id.IdIndex
@@ -421,7 +421,9 @@ class AccumuloDataStore(val connector: AccumuloClient, override val config: Accu
       val queryPlans = getQueryPlan(query)
 
       if (queryPlans.isEmpty) {
-        EmptyPlan(FilterStrategy(fallbackIndex, None, Some(Filter.EXCLUDE), temporal = false, Float.PositiveInfinity))
+        val filter =
+          FilterStrategy(fallbackIndex, None, Some(Filter.EXCLUDE), temporal = false, Float.PositiveInfinity, query.getHints)
+        EmptyPlan(QueryStrategy(filter, Seq.empty, Seq.empty, Seq.empty, filter.filter, None))
       } else {
         val qps =
           if (queryPlans.lengthCompare(1) == 0) { queryPlans } else {

geomesa-accumulo/geomesa-accumulo-datastore/src/main/scala/org/locationtech/geomesa/accumulo/data/AccumuloIndexAdapter.scala

@@ -8,11 +8,13 @@
 
 package org.locationtech.geomesa.accumulo.data
 
+import com.github.benmanes.caffeine.cache.{CacheLoader, Caffeine}
 import com.typesafe.scalalogging.LazyLogging
 import org.apache.accumulo.core.data.{Key, Range, Value}
 import org.apache.accumulo.core.file.keyfunctor.RowFunctor
 import org.apache.hadoop.io.Text
 import org.geotools.api.feature.simple.{SimpleFeature, SimpleFeatureType}
+import org.locationtech.geomesa.accumulo.AccumuloProperties.TableProperties.TableCacheExpiry
 import org.locationtech.geomesa.accumulo.data.AccumuloIndexAdapter._
 import org.locationtech.geomesa.accumulo.data.AccumuloQueryPlan.{BatchScanPlan, EmptyPlan}
 import org.locationtech.geomesa.accumulo.data.writer.tx.AccumuloAtomicIndexWriter
@@ -35,6 +37,7 @@ import org.locationtech.geomesa.index.index.z2.{Z2Index, Z2IndexValues}
 import org.locationtech.geomesa.index.index.z3.{Z3Index, Z3IndexValues}
 import org.locationtech.geomesa.index.iterators.StatsScan
 import org.locationtech.geomesa.index.planning.LocalQueryRunner.LocalTransformReducer
+import org.locationtech.geomesa.index.utils.Explainer
 import org.locationtech.geomesa.security.SecurityUtils
 import org.locationtech.geomesa.utils.concurrent.CachedThreadPool
 import org.locationtech.geomesa.utils.geotools.SimpleFeatureTypes.{Configs, InternalConfigs}
@@ -58,6 +61,13 @@ class AccumuloIndexAdapter(ds: AccumuloDataStore)
 
   private val tableOps = ds.connector.tableOperations()
 
+  private val tableSizeCache =
+    Caffeine.newBuilder().expireAfterWrite(TableCacheExpiry.toJavaDuration.get).build[String, Integer](
+      new CacheLoader[String, Integer]() {
+        override def load(table: String): Integer = tableOps.listSplits(table).size() + 1
+      }
+    )
+
   // noinspection ScalaDeprecation
   override def createTable(
       index: GeoMesaFeatureIndex[_, _],
@@ -153,21 +163,12 @@ class AccumuloIndexAdapter(ds: AccumuloDataStore)
   override def createQueryPlan(strategy: QueryStrategy): AccumuloQueryPlan = {
     import org.locationtech.geomesa.index.conf.QueryHints.RichHints
 
-    val QueryStrategy(filter, byteRanges, _, _, ecql, hints, _) = strategy
-    val index = filter.index
-    // index api defines empty start/end for open-ended range - in accumulo, it's indicated with null
-    // index api defines start row inclusive, end row exclusive
-    val ranges = byteRanges.map {
-      case BoundedByteRange(start, end) =>
-          val startKey = if (start.length == 0) { null } else { new Key(new Text(start)) }
-          val endKey = if (end.length == 0) { null } else { new Key(new Text(end)) }
-          new Range(startKey, true, endKey, false)
-
-      case SingleRowByteRange(row) =>
-        new Range(new Text(row))
-    }
+    val index = strategy.index
+    val ecql = strategy.ecql
+    val hints = strategy.hints
+    val ranges = strategy.ranges.map(toAccumuloRange)
     val numThreads = if (index.name == IdIndex.name) { ds.config.queries.recordThreads } else { ds.config.queries.threads }
-    val tables = index.getTablesForQuery(filter.filter)
+    val tables = index.getTablesForQuery(strategy.filter.filter)
     val (colFamily, schema) = {
       val (cf, s) = groups.group(index.sft, hints.getTransformDefinition, ecql)
       (Some(new Text(ColumnFamilyMapper(index)(cf))), s)
@@ -180,10 +181,10 @@ class AccumuloIndexAdapter(ds: AccumuloDataStore)
 
     index match {
       case i: AttributeJoinIndex =>
-        AccumuloJoinIndexAdapter.createQueryPlan(ds, i, filter, tables, ranges, colFamily, schema, ecql, hints, numThreads)
+        AccumuloJoinIndexAdapter.createQueryPlan(ds, i, strategy, tables, ranges, colFamily, schema, ecql, hints, numThreads)
 
       case _ =>
-        val (iter, eToF, reduce) = if (strategy.hints.isBinQuery) {
+        val (iter, eToF, reduce) = if (hints.isBinQuery) {
           if (ds.config.remote.bin) {
             val iter = BinAggregatingIterator.configure(schema, index, ecql, hints)
             (Seq(iter), new AccumuloBinResultsToFeatures(), None)
@@ -195,9 +196,9 @@ class AccumuloIndexAdapter(ds: AccumuloDataStore)
             }
             (fti, resultsToFeatures, localReducer)
           }
-        } else if (strategy.hints.isArrowQuery) {
+        } else if (hints.isArrowQuery) {
           if (ds.config.remote.arrow) {
-            val (iter, reduce) = ArrowIterator.configure(schema, index, ds.stats, filter.filter, ecql, hints)
+            val (iter, reduce) = ArrowIterator.configure(schema, index, ds.stats, strategy.filter.filter, ecql, hints)
             (Seq(iter), new AccumuloArrowResultsToFeatures(), Some(reduce))
           } else {
             if (hints.isSkipReduce) {
@@ -207,7 +208,7 @@ class AccumuloIndexAdapter(ds: AccumuloDataStore)
             }
             (fti, resultsToFeatures, localReducer)
           }
-        } else if (strategy.hints.isDensityQuery) {
+        } else if (hints.isDensityQuery) {
           if (ds.config.remote.density) {
             val iter = DensityIterator.configure(schema, index, ecql, hints)
             (Seq(iter), new AccumuloDensityResultsToFeatures(), None)
@@ -219,7 +220,7 @@ class AccumuloIndexAdapter(ds: AccumuloDataStore)
             }
             (fti, resultsToFeatures, localReducer)
           }
-        } else if (strategy.hints.isStatsQuery) {
+        } else if (hints.isStatsQuery) {
           if (ds.config.remote.stats) {
             val iter = StatsIterator.configure(schema, index, ecql, hints)
             val reduce = Some(StatsScan.StatsReducer(schema, hints))
@@ -236,7 +237,7 @@ class AccumuloIndexAdapter(ds: AccumuloDataStore)
           (fti, resultsToFeatures, None)
         }
 
-        if (ranges.isEmpty) { EmptyPlan(strategy.filter, reduce) } else {
+        if (ranges.isEmpty) { EmptyPlan(strategy, reduce) } else {
           // configure additional iterators based on the index
           // TODO pull this out to be SPI loaded so that new indices can be added seamlessly
           val indexIter = if (index.name == Z3Index.name) {
@@ -272,7 +273,7 @@ class AccumuloIndexAdapter(ds: AccumuloDataStore)
           val max = hints.getMaxFeatures
           val project = hints.getProjection
 
-          BatchScanPlan(filter, tables, ranges, iters, colFamily, eToF, reduce, sort, max, project, numThreads)
+          BatchScanPlan(strategy, tables, ranges, iters, colFamily, eToF, reduce, sort, max, project, numThreads)
         }
     }
   }
@@ -290,12 +291,56 @@ class AccumuloIndexAdapter(ds: AccumuloDataStore)
       case (true, true)   => new AccumuloAtomicIndexWriter(ds, sft, indices, wrapper, partition)  with RequiredVisibilityWriter
     }
   }
+
+  override def getStrategyCost(strategy: FilterStrategy, explain: Explainer): Option[Long] = {
+    explain.pushLevel(s"Calculating cost for ${strategy.index.identifier}")
+    val start = System.currentTimeMillis()
+    try {
+      val tables = strategy.index.getTablesForQuery(strategy.filter)
+      if (tables.isEmpty) {
+        return Some(0L)
+      }
+      val ranges = strategy.getQueryStrategy(explain).ranges.map(toAccumuloRange).asJava
+      val cost =
+        tables.foldLeft(0d) { case (sum, table) =>
+          val numTablets = tableSizeCache.get(table)
+          val tabletsScanned = tableOps.locate(table, ranges).groupByTablet().size()
+          explain(s"Strategy hits $tabletsScanned/$numTablets tablets for table $table")
+          val cost = 100 * (tabletsScanned.toDouble / numTablets)
+          sum + cost
+        }
+      Some(cost.toLong)
+    } finally {
+      explain(s"Cost calculations took ${System.currentTimeMillis() - start}ms").popLevel()
+    }
+  }
 }
 
 object AccumuloIndexAdapter {
 
   val ZIterPriority = 23
 
+  /**
+   * Converts a generic index-api range into an Accumulo range
+   *
+   * @param range range
+   * @return
+   */
+  private def toAccumuloRange(range: ByteRange): Range = range match {
+    case BoundedByteRange(lower, upper) =>
+      // index api defines empty start/end for open-ended range - in accumulo, it's indicated with null
+      val start = if (lower.length == 0) { null } else { new Key(new Text(lower)) }
+      val end = if (upper.length == 0) { null } else { new Key(new Text(upper)) }
+      // index api defines start row inclusive, end row exclusive
+      new Range(start, true, end, false)
+
+    case SingleRowByteRange(row) =>
+      new Range(new Text(row))
+
+    case _ =>
+      throw new IllegalArgumentException(s"Unexpected range type $range")
+  }
+
   /**
     * Accumulo entries to features
     *

geomesa-accumulo/geomesa-accumulo-datastore/src/main/scala/org/locationtech/geomesa/accumulo/data/AccumuloJoinIndexAdapter.scala

@@ -62,7 +62,7 @@ object AccumuloJoinIndexAdapter {
   def createQueryPlan(
       ds: AccumuloDataStore,
       index: AttributeJoinIndex,
-      filter: FilterStrategy,
+      strategy: QueryStrategy,
       tables: Seq[String],
       ranges: Seq[org.apache.accumulo.core.data.Range],
       colFamily: Option[Text],
@@ -85,7 +85,7 @@ object AccumuloJoinIndexAdapter {
       val sort = hints.getSortFields
       val max = hints.getMaxFeatures
       val project = hints.getProjection
-      BatchScanPlan(filter, tables, ranges, iterators, colFamily, kvsToFeatures, reduce, sort, max, project, numThreads)
+      BatchScanPlan(strategy, tables, ranges, iterators, colFamily, kvsToFeatures, reduce, sort, max, project, numThreads)
     }
 
     // used when remote processing is disabled
@@ -115,13 +115,13 @@ object AccumuloJoinIndexAdapter {
         }
       } else {
         // have to do a join against the record table
-        createJoinPlan(ds, index, filter, tables, ranges, colFamily, ecql, hints)
+        createJoinPlan(ds, index, strategy, tables, ranges, colFamily, ecql, hints)
       }
     } else if (hints.isArrowQuery) {
       // check to see if we can execute against the index values
       if (index.canUseIndexSchema(ecql, transform)) {
         if (ds.config.remote.arrow) {
-          val (arrowIter, reduce) = ArrowIterator.configure(indexSft, index, ds.stats, filter.filter, ecql, hints)
+          val (arrowIter, reduce) = ArrowIterator.configure(indexSft, index, ds.stats, strategy.filter.filter, ecql, hints)
           plan(Seq(arrowIter), new AccumuloArrowResultsToFeatures(), Some(reduce))
         } else {
           localPlan()
@@ -147,7 +147,7 @@ object AccumuloJoinIndexAdapter {
         }
       } else {
         // have to do a join against the record table
-        createJoinPlan(ds, index, filter, tables, ranges, colFamily, ecql, hints)
+        createJoinPlan(ds, index, strategy, tables, ranges, colFamily, ecql, hints)
       }
     } else if (hints.isDensityQuery) {
       // check to see if we can execute against the index values
@@ -183,7 +183,7 @@ object AccumuloJoinIndexAdapter {
         }
       } else {
         // have to do a join against the record table
-        createJoinPlan(ds, index, filter, tables, ranges, colFamily, ecql, hints)
+        createJoinPlan(ds, index, strategy, tables, ranges, colFamily, ecql, hints)
       }
     } else if (hints.isStatsQuery) {
       // check to see if we can execute against the index values
@@ -197,7 +197,7 @@ object AccumuloJoinIndexAdapter {
         }
       } else {
         // have to do a join against the record table
-        createJoinPlan(ds, index, filter, tables, ranges, colFamily, ecql, hints)
+        createJoinPlan(ds, index, strategy, tables, ranges, colFamily, ecql, hints)
       }
     } else if (index.canUseIndexSchema(ecql, transform)) {
       // we can use the index value
@@ -222,10 +222,10 @@ object AccumuloJoinIndexAdapter {
       plan(iters, toFeatures, None)
     } else {
       // have to do a join against the record table
-      createJoinPlan(ds, index, filter, tables, ranges, colFamily, ecql, hints)
+      createJoinPlan(ds, index, strategy, tables, ranges, colFamily, ecql, hints)
     }
 
-    if (ranges.nonEmpty) { qp } else { EmptyPlan(qp.filter, qp.reducer) }
+    if (ranges.nonEmpty) { qp } else { EmptyPlan(strategy, qp.reducer) }
   }
 
   /**
@@ -235,7 +235,7 @@ object AccumuloJoinIndexAdapter {
   private def createJoinPlan(
       ds: AccumuloDataStore,
       index: AttributeJoinIndex,
-      filter: FilterStrategy,
+      strategy: QueryStrategy,
       tables: Seq[String],
       ranges: Seq[org.apache.accumulo.core.data.Range],
       colFamily: Option[Text],
@@ -280,7 +280,7 @@ object AccumuloJoinIndexAdapter {
       hints.put(QueryHints.Internal.RETURN_SFT, resultSft)
     }
 
-    val recordTables = recordIndex.getTablesForQuery(filter.filter)
+    val recordTables = recordIndex.getTablesForQuery(strategy.filter.filter)
     val recordThreads = ds.config.queries.recordThreads
 
     // function to join the attribute index scan results to the record table
@@ -294,13 +294,13 @@ object AccumuloJoinIndexAdapter {
       }
     }
 
-    val joinQuery = BatchScanPlan(filter, recordTables, Seq.empty, recordIterators, recordColFamily, toFeatures,
+    val joinQuery = BatchScanPlan(strategy, recordTables, Seq.empty, recordIterators, recordColFamily, toFeatures,
       Some(reducer), hints.getSortFields, hints.getMaxFeatures, hints.getProjection, recordThreads)
 
     val attributeIters = visibilityIter(index) ++
         FilterTransformIterator.configure(index.indexSft, index, stFilter, None, hints.getSampling).toSeq
 
-    JoinPlan(filter, tables, ranges, attributeIters, colFamily, recordThreads, joinFunction, joinQuery)
+    JoinPlan(strategy, tables, ranges, attributeIters, colFamily, recordThreads, joinFunction, joinQuery)
   }
 
   private def visibilityIter(index: AttributeJoinIndex): Seq[IteratorSetting] = {

geomesa-accumulo/geomesa-accumulo-datastore/src/main/scala/org/locationtech/geomesa/accumulo/data/AccumuloQueryPlan.scala

@@ -15,7 +15,7 @@ import org.apache.accumulo.core.security.Authorizations
 import org.apache.hadoop.io.Text
 import org.locationtech.geomesa.accumulo.util.BatchMultiScanner
 import org.locationtech.geomesa.index.api.QueryPlan.{FeatureReducer, ResultsToFeatures}
-import org.locationtech.geomesa.index.api.{FilterStrategy, QueryPlan}
+import org.locationtech.geomesa.index.api.{QueryPlan, QueryStrategy}
 import org.locationtech.geomesa.index.utils.Explainer
 import org.locationtech.geomesa.index.utils.Reprojection.QueryReferenceSystems
 import org.locationtech.geomesa.index.utils.ThreadManagement.{LowLevelScanner, ManagedScan, Timeout}
@@ -79,7 +79,7 @@ object AccumuloQueryPlan extends LazyLogging {
     Key.toPrintableString(k.getRow.getBytes, 0, k.getRow.getLength, k.getRow.getLength)
 
   // plan that will not actually scan anything
-  case class EmptyPlan(filter: FilterStrategy, reducer: Option[FeatureReducer] = None) extends AccumuloQueryPlan {
+  case class EmptyPlan(strategy: QueryStrategy, reducer: Option[FeatureReducer] = None) extends AccumuloQueryPlan {
     override val tables: Seq[String] = Seq.empty
     override val iterators: Seq[IteratorSetting] = Seq.empty
     override val ranges: Seq[org.apache.accumulo.core.data.Range] = Seq.empty
@@ -94,7 +94,7 @@ object AccumuloQueryPlan extends LazyLogging {
 
   // batch scan plan
   case class BatchScanPlan(
-      filter: FilterStrategy,
+      strategy: QueryStrategy,
       tables: Seq[String],
       ranges: Seq[org.apache.accumulo.core.data.Range],
       iterators: Seq[IteratorSetting],
@@ -108,6 +108,8 @@ object AccumuloQueryPlan extends LazyLogging {
     ) extends AccumuloQueryPlan {
 
     override def scan(ds: AccumuloDataStore): CloseableIterator[Entry[Key, Value]] = {
+      // query guard hook - also handles full table scan checks
+      strategy.runGuards(ds)
       // convert the relative timeout to an absolute timeout up front
       val timeout = ds.config.queries.timeout.map(Timeout.apply)
       // note: calculate authorizations up front so that multi-threading doesn't mess up auth providers
@@ -123,7 +125,7 @@ object AccumuloQueryPlan extends LazyLogging {
      * @param timeout absolute stop time, as sys time
      * @return
      */
-    def scan(
+    private[accumulo] def scan(
         connector: AccumuloClient,
         auths: Authorizations,
         partitionParallelScans: Boolean,
@@ -162,7 +164,7 @@ object AccumuloQueryPlan extends LazyLogging {
 
   // join on multiple tables - requires multiple scans
   case class JoinPlan(
-      filter: FilterStrategy,
+      strategy: QueryStrategy,
       tables: Seq[String],
       ranges: Seq[org.apache.accumulo.core.data.Range],
       iterators: Seq[IteratorSetting],
@@ -180,6 +182,8 @@ object AccumuloQueryPlan extends LazyLogging {
     override def projection: Option[QueryReferenceSystems] = joinQuery.projection
 
     override def scan(ds: AccumuloDataStore): CloseableIterator[Entry[Key, Value]] = {
+      // query guard hook - also handles full table scan checks
+      strategy.runGuards(ds)
       // convert the relative timeout to an absolute timeout up front
       val timeout = ds.config.queries.timeout.map(Timeout.apply)
       // calculate authorizations up front so that multi-threading doesn't mess up auth providers