1 /***********************************************************************
2  * Copyright (c) 2013-2024 Commonwealth Computer Research, Inc.
3  * All rights reserved. This program and the accompanying materials
4  * are made available under the terms of the Apache License, Version 2.0
5  * which accompanies this distribution and is available at
6  * http://www.opensource.org/licenses/apache2.0.php.
7  ***********************************************************************/
8 
9 package org.locationtech.geomesa.index.geotools
10 
11 import com.typesafe.scalalogging.LazyLogging
12 import org.geotools.api.data._
13 import org.geotools.api.feature.FeatureVisitor
14 import org.geotools.api.feature.simple.{SimpleFeature, SimpleFeatureType}
15 import org.geotools.api.filter.Filter
16 import org.geotools.api.filter.expression.{Expression, PropertyName}
17 import org.geotools.api.filter.sort.SortBy
18 import org.geotools.api.util.ProgressListener
19 import org.geotools.data.simple.SimpleFeatureCollection
20 import org.geotools.data.store.DataFeatureCollection
21 import org.geotools.data.util.NullProgressListener
22 import org.geotools.feature.FeatureCollection
23 import org.geotools.feature.collection.{DecoratingFeatureCollection, DecoratingSimpleFeatureCollection}
24 import org.geotools.feature.visitor.GroupByVisitor.GroupByRawResult
25 import org.geotools.feature.visitor._
26 import org.geotools.geometry.jts.ReferencedEnvelope
27 import org.locationtech.geomesa.filter.FilterHelper
28 import org.locationtech.geomesa.index.geotools.GeoMesaFeatureCollection.GeoMesaFeatureVisitingCollection
29 import org.locationtech.geomesa.index.process.GeoMesaProcessVisitor
30 import org.locationtech.geomesa.index.stats.GeoMesaStats
31 import org.locationtech.geomesa.utils.collection.CloseableIterator
32 import org.locationtech.geomesa.utils.io.WithClose
33 import org.locationtech.geomesa.utils.stats._
34 
35 import java.util.Collections
36 import java.util.concurrent.atomic.AtomicLong
37 import scala.annotation.tailrec
38 
39 /**
40   * Feature collection implementation
41   */
42 class GeoMesaFeatureCollection(source: GeoMesaFeatureSource, query: Query)
43     extends GeoMesaFeatureVisitingCollection(source, source.ds.stats, query) {
44 
45   private val transaction = source match {
46     case s: SimpleFeatureStore => s.getTransaction
47     case _ => Transaction.AUTO_COMMIT
48   }
49 
50   private lazy val featureReader = source.ds.getFeatureReader(source.sft, transaction, query)
51 
52   override def getSchema: SimpleFeatureType = featureReader.schema
53 
54   override def reader(): FeatureReader[SimpleFeatureType, SimpleFeature] = featureReader.reader()
55 
56   override def subCollection(filter: Filter): SimpleFeatureCollection = {
57     val merged = new Query(query)
58     val filters = Seq(merged.getFilter, filter).filter(_ != Filter.INCLUDE)
59     FilterHelper.filterListAsAnd(filters).foreach(merged.setFilter)
60     new GeoMesaFeatureCollection(source, merged)
61   }
62 
63   override def sort(order: SortBy): SimpleFeatureCollection = {
64     val merged = new Query(query)
65     if (merged.getSortBy == null) {
66       merged.setSortBy(order)
67     } else {
68       merged.setSortBy(merged.getSortBy :+ order: _*)
69     }
70     new GeoMesaFeatureCollection(source, merged)
71   }
72 
73   override def getBounds: ReferencedEnvelope = source.getBounds(query)
74 
75   override def getCount: Int = source.getCount(query)
76 
77   // note: this shouldn't return -1 (as opposed to FeatureSource.getCount), but we still don't return a valid
78   // size unless exact counts are enabled
79   override def size: Int = {
80     val count = getCount
81     if (count < 0) { 0 } else { count }
82   }
83 
84   override def contains(o: Any): Boolean = {
85     o match {
86       case f: SimpleFeature =>
87         val sub = subCollection(FilterHelper.ff.id(f.getIdentifier))
88         WithClose(CloseableIterator(sub.features()))(_.nonEmpty)
89 
90       case _ => false
91     }
92   }
93 
94   override def containsAll(collection: java.util.Collection[_]): Boolean = {
95     val size = collection.size()
96     if (size == 0) { true } else {
97       val filters = Seq.newBuilder[Filter]
98       filters.sizeHint(size)
99 
100       val features = collection.iterator()
101       while (features.hasNext) {
102         features.next match {
103           case f: SimpleFeature => filters += FilterHelper.ff.id(f.getIdentifier)
104           case _ => return false
105         }
106       }
107 
108       val sub = subCollection(org.locationtech.geomesa.filter.orFilters(filters.result))
109       WithClose(CloseableIterator(sub.features()))(_.length) == size
110     }
111   }
112 }
113 
114 object GeoMesaFeatureCollection extends LazyLogging {
115 
116   import scala.collection.JavaConverters._
117 
118   private val oneUp = new AtomicLong(0)
119 
120   def nextId: String = s"GeoMesaFeatureCollection-${oneUp.getAndIncrement()}"
121 
122   /**
123     * Attempts to visit the feature collection in an optimized manner. This will unwrap any decorating
124     * feature collections that may interfere with the `accepts` method.
125     *
126     * Note that generally this may not be a good idea - collections are presumably wrapped for a reason.
127     * However, our visitation functionality keeps being broken by changes in GeoServer, so we're being
128     * defensive here.
129     *
130     * @param collection feature collection
131     * @param visitor visitor
132     * @param progress progress monitor
133     */
134   def visit(
135       collection: FeatureCollection[SimpleFeatureType, SimpleFeature],
136       visitor: FeatureVisitor,
137       progress: ProgressListener = new NullProgressListener): Unit = {
138     val unwrapped = if (collection.isInstanceOf[GeoMesaFeatureVisitingCollection]) { collection } else {
139       try { unwrap(collection) } catch {
140         case e: Throwable => logger.debug("Error trying to unwrap feature collection:", e); collection
141       }
142     }
143     unwrapped.accepts(visitor, progress)
144   }
145 
146   /**
147     * Attempts to remove any decorating feature collections
148     *
149     * @param collection collection
150     * @param level level of collections that have been removed, used to detect potentially infinite looping
151     * @return
152     */
153   @tailrec
154   private def unwrap(
155       collection: FeatureCollection[SimpleFeatureType, SimpleFeature],
156       level: Int = 1): FeatureCollection[SimpleFeatureType, SimpleFeature] = {
157 
158     // noinspection TypeCheckCanBeMatch
159     val unwrapped = if (collection.isInstanceOf[DecoratingSimpleFeatureCollection]) {
160       getDelegate(collection, classOf[DecoratingSimpleFeatureCollection])
161     } else if (collection.isInstanceOf[DecoratingFeatureCollection[SimpleFeatureType, SimpleFeature]]) {
162       getDelegate(collection, classOf[DecoratingFeatureCollection[SimpleFeatureType, SimpleFeature]])
163     } else {
164       logger.debug(s"Unable to unwrap feature collection $collection of class ${collection.getClass.getName}")
165       return collection
166     }
167 
168     logger.debug(s"Unwrapped feature collection $collection of class ${collection.getClass.getName} to " +
169         s"$unwrapped of class ${unwrapped.getClass.getName}")
170 
171     if (unwrapped.isInstanceOf[GeoMesaFeatureVisitingCollection]) {
172       unwrapped
173     } else if (level > 9) {
174       logger.debug("Aborting feature collection unwrapping after 10 iterations")
175       unwrapped
176     } else {
177       unwrap(unwrapped, level + 1)
178     }
179   }
180 
181   /**
182     * Uses reflection to access the protected field 'delegate' in decorating feature collection classes
183     *
184     * @param collection decorating feature collection
185     * @param clas decorating feature base class
186     * @return
187     */
188   private def getDelegate(
189       collection: FeatureCollection[SimpleFeatureType, SimpleFeature],
190       clas: Class[_]): FeatureCollection[SimpleFeatureType, SimpleFeature] = {
191     val m = clas.getDeclaredField("delegate")
192     m.setAccessible(true)
193     m.get(collection).asInstanceOf[FeatureCollection[SimpleFeatureType, SimpleFeature]]
194   }
195 
196   /**
197     * Base class for handling feature visitors
198     *
199     * @param source feature source
200     * @param stats geomesa stat hook
201     * @param query query
202     */
203   abstract class GeoMesaFeatureVisitingCollection(source: SimpleFeatureSource, stats: GeoMesaStats, query: Query)
204       extends DataFeatureCollection(nextId) with LazyLogging {
205 
206     private def unoptimized(visitor: FeatureVisitor, progress: ProgressListener): Unit = {
207       lazy val warning = s"Using unoptimized method for visiting '${visitor.getClass.getName}'"
208       logger.warn(warning)
209       if (progress != null) {
210         progress.warningOccurred(getClass.getName, "accepts()", warning)
211       }
212       super.accepts(visitor, progress)
213     }
214 
215     override def accepts(visitor: FeatureVisitor, progress: ProgressListener): Unit = {
216       visitor match {
217         case v: GeoMesaProcessVisitor =>
218           v.execute(source, query)
219 
220         case v: AverageVisitor if v.getExpression.isInstanceOf[PropertyName] =>
221           val attribute = v.getExpression.asInstanceOf[PropertyName].getPropertyName
222           val stat = Stat.DescriptiveStats(Seq(attribute))
223           stats.getStat[DescriptiveStats](source.getSchema, stat, query.getFilter, exact = true) match {
224             case Some(s) if s.count <= Int.MaxValue.toLong => v.setValue(s.count.toInt, s.sum(0))
225             case Some(s) => v.setValue(s.mean(0))
226             case None    => unoptimized(visitor, progress)
227           }
228 
229         case v: BoundsVisitor =>
230           v.getBounds.expandToInclude(stats.getBounds(source.getSchema, query.getFilter))
231 
232         case v: CountVisitor =>
233           v.setValue(source.getCount(query))
234 
235         case v: MaxVisitor if v.getExpression.isInstanceOf[PropertyName] =>
236           val attribute = v.getExpression.asInstanceOf[PropertyName].getPropertyName
237           minMax(attribute, exact = false).orElse(minMax(attribute, exact = true)) match {
238             case Some((_, max)) => v.setValue(max)
239             case None           => unoptimized(visitor, progress)
240           }
241 
242         case v: GroupByVisitor if v.getExpression.isInstanceOf[PropertyName] =>
243           val attribute = v.getExpression.asInstanceOf[PropertyName].getPropertyName
244           groupBy(attribute, v.getGroupByAttributes.asScala.toSeq, v.getAggregateVisitor) match {
245             case Some(result) => v.setValue(result)
246             case None         => unoptimized(visitor, progress)
247           }
248 
249         case v: MinVisitor if v.getExpression.isInstanceOf[PropertyName] =>
250           val attribute = v.getExpression.asInstanceOf[PropertyName].getPropertyName
251           minMax(attribute, exact = false).orElse(minMax(attribute, exact = true)) match {
252             case Some((min, _)) => v.setValue(min)
253             case None           => unoptimized(visitor, progress)
254           }
255 
256         case v: SumVisitor if v.getExpression.isInstanceOf[PropertyName] =>
257           val attribute = v.getExpression.asInstanceOf[PropertyName].getPropertyName
258           val stat = Stat.DescriptiveStats(Seq(attribute))
259           stats.getStat[DescriptiveStats](source.getSchema, stat, query.getFilter, exact = true) match {
260             case Some(s) => v.setValue(s.sum(0))
261             case None    => unoptimized(visitor, progress)
262           }
263 
264         case v: UniqueVisitor if v.getExpression.isInstanceOf[PropertyName] =>
265           val attribute = v.getExpression.asInstanceOf[PropertyName].getPropertyName
266           val stat = Stat.Enumeration(attribute)
267           stats.getStat[EnumerationStat[Any]](source.getSchema, stat, query.getFilter, exact = true) match {
268             case Some(s) => v.setValue(s.values.toList.asJava)
269             case None    => unoptimized(visitor, progress)
270           }
271 
272         case _ =>
273           unoptimized(visitor, progress)
274       }
275     }
276 
277     private def minMax(attribute: String, exact: Boolean): Option[(Any, Any)] =
278       stats.getMinMax[Any](source.getSchema, attribute, query.getFilter, exact).map(_.bounds)
279 
280     private def groupBy(
281         attribute: String,
282         groupByExpression: Seq[Expression],
283         aggregate: FeatureVisitor): Option[java.util.List[GroupByRawResult]] = {
284       if (groupByExpression.lengthCompare(1) != 0
285           || groupByExpression.exists(e => !e.isInstanceOf[PropertyName])) { None } else {
286         val groupBy = groupByExpression.map(_.asInstanceOf[PropertyName].getPropertyName).head
287         val op: Option[(String, Stat => Any)] = aggregate match {
288           case _: CountVisitor =>
289             Some(Stat.Count() -> { (s: Stat) => math.min( s.asInstanceOf[CountStat].count, Int.MaxValue.toLong).toInt })
290 
291           case _: MaxVisitor =>
292             Some(Stat.MinMax(attribute) -> { (s: Stat) => s.asInstanceOf[MinMax[Any]].max })
293 
294           case _: MinVisitor =>
295             Some(Stat.MinMax(attribute) -> { (s: Stat) => s.asInstanceOf[MinMax[Any]].min })
296 
297           case _ =>
298             None
299         }
300         op.flatMap { case (nested, unwrap) =>
301           val stat = Stat.GroupBy(groupBy, nested)
302           stats.getStat[GroupBy[AnyRef]](source.getSchema, stat, query.getFilter, exact = true).map { grouped =>
303             val result = new java.util.ArrayList[GroupByRawResult]
304             grouped.iterator.foreach { case (group, stat) =>
305               result.add(new GroupByRawResult(Collections.singletonList(group), unwrap(stat)))
306             }
307             result
308           }
309         }
310       }
311     }
312   }
313 }