1 /***********************************************************************
2  * Copyright (c) 2017-2024 IBM
3  * Copyright (c) 2013-2024 Commonwealth Computer Research, Inc.
4  * All rights reserved. This program and the accompanying materials
5  * are made available under the terms of the Apache License, Version 2.0
6  * which accompanies this distribution and is available at
7  * http://www.opensource.org/licenses/apache2.0.php.
8  ***********************************************************************/
9 
10 package org.locationtech.geomesa.cassandra.data
11 
12 import com.datastax.driver.core.{Row, Statement}
13 import org.geotools.api.filter.Filter
14 import org.locationtech.geomesa.cassandra.utils.CassandraBatchScan
15 import org.locationtech.geomesa.index.api.QueryPlan.{FeatureReducer, ResultsToFeatures}
16 import org.locationtech.geomesa.index.api.{FilterStrategy, QueryPlan}
17 import org.locationtech.geomesa.index.utils.Explainer
18 import org.locationtech.geomesa.index.utils.Reprojection.QueryReferenceSystems
19 import org.locationtech.geomesa.index.utils.ThreadManagement.Timeout
20 import org.locationtech.geomesa.utils.collection.CloseableIterator
21 
22 sealed trait CassandraQueryPlan extends QueryPlan[CassandraDataStore] {
23 
24   override type Results = Row
25 
26   def tables: Seq[String]
27   def ranges: Seq[Statement]
28   def numThreads: Int
29 
30   /**
31     * Note: filter is applied in entriesToFeatures, this is just for explain logging
32     * @return
33     */
34   def clientSideFilter: Option[Filter]
35 
36   override def explain(explainer: Explainer, prefix: String): Unit =
37     CassandraQueryPlan.explain(this, explainer, prefix)
38 }
39 
40 object CassandraQueryPlan {
41   def explain(plan: CassandraQueryPlan, explainer: Explainer, prefix: String): Unit = {
42     import org.locationtech.geomesa.filter.filterToString
43     explainer.pushLevel(s"${prefix}Plan: ${plan.getClass.getName}")
44     explainer(s"Tables: ${plan.tables.mkString(", ")}")
45     explainer(s"Ranges (${plan.ranges.size}): ${plan.ranges.take(5).map(_.toString).mkString(", ")}")
46     explainer(s"Client-side filter: ${plan.clientSideFilter.map(filterToString).getOrElse("none")}")
47     explainer(s"Reduce: ${plan.reducer.getOrElse("none")}")
48     explainer.popLevel()
49   }
50 }
51 
52 // plan that will not actually scan anything
53 case class EmptyPlan(filter: FilterStrategy, reducer: Option[FeatureReducer] = None) extends CassandraQueryPlan {
54   override val tables: Seq[String] = Seq.empty
55   override val ranges: Seq[Statement] = Seq.empty
56   override val numThreads: Int = 0
57   override val clientSideFilter: Option[Filter] = None
58   override val resultsToFeatures: ResultsToFeatures[Row] = ResultsToFeatures.empty
59   override val sort: Option[Seq[(String, Boolean)]] = None
60   override val maxFeatures: Option[Int] = None
61   override val projection: Option[QueryReferenceSystems] = None
62   override def scan(ds: CassandraDataStore): CloseableIterator[Row] = CloseableIterator.empty
63 }
64 
65 case class StatementPlan(
66     filter: FilterStrategy,
67     tables: Seq[String],
68     ranges: Seq[Statement],
69     numThreads: Int,
70     // note: filter is applied in entriesToFeatures, this is just for explain logging
71     clientSideFilter: Option[Filter],
72     resultsToFeatures: ResultsToFeatures[Row],
73     reducer: Option[FeatureReducer],
74     sort: Option[Seq[(String, Boolean)]],
75     maxFeatures: Option[Int],
76     projection: Option[QueryReferenceSystems]
77   ) extends CassandraQueryPlan {
78 
79   override def scan(ds: CassandraDataStore): CloseableIterator[Row] =
80     CassandraBatchScan(this, ds.session, ranges, numThreads, ds.config.queries.timeout.map(Timeout.apply))
81 }