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.process.analytic
10 
11 import com.typesafe.scalalogging.LazyLogging
12 import org.geotools.api.data.{Query, SimpleFeatureSource}
13 import org.geotools.api.feature.Feature
14 import org.geotools.api.feature.`type`.AttributeDescriptor
15 import org.geotools.api.feature.simple.{SimpleFeature, SimpleFeatureType}
16 import org.geotools.api.filter.Filter
17 import org.geotools.api.util.ProgressListener
18 import org.geotools.data.collection.ListFeatureCollection
19 import org.geotools.data.simple.SimpleFeatureCollection
20 import org.geotools.feature.simple.{SimpleFeatureBuilder, SimpleFeatureTypeBuilder}
21 import org.geotools.feature.visitor.{AbstractCalcResult, CalcResult}
22 import org.geotools.process.factory.{DescribeParameter, DescribeProcess, DescribeResult}
23 import org.locationtech.geomesa.filter.factory.FastFilterFactory
24 import org.locationtech.geomesa.index.conf.QueryHints
25 import org.locationtech.geomesa.index.geotools.GeoMesaFeatureCollection
26 import org.locationtech.geomesa.index.index.attribute.AttributeIndex
27 import org.locationtech.geomesa.index.iterators.StatsScan
28 import org.locationtech.geomesa.index.process.GeoMesaProcessVisitor
29 import org.locationtech.geomesa.process.GeoMesaProcess
30 import org.locationtech.geomesa.utils.collection.SelfClosingIterator
31 import org.locationtech.geomesa.utils.geotools.RichAttributeDescriptors.RichAttributeDescriptor
32 import org.locationtech.geomesa.utils.stats.{EnumerationStat, Stat}
33 
34 import scala.collection.JavaConverters._
35 import scala.collection.mutable
36 
37 @DescribeProcess(title = "Geomesa Unique",
38   description = "Finds unique attributes values, optimized for GeoMesa")
39 class UniqueProcess extends GeoMesaProcess with LazyLogging {
40 
41   @DescribeResult(name = "result",
42     description = "Feature collection with an attribute containing the unique values")
43   def execute(
44     @DescribeParameter(name = "features", description = "Input feature collection")
45     features: SimpleFeatureCollection,
46     @DescribeParameter(name = "attribute", description = "Attribute whose unique values are extracted")
47     attribute: String,
48     @DescribeParameter(name = "filter", min = 0, description = "The filter to apply to the feature collection")
49     filter: Filter,
50     @DescribeParameter(name = "histogram", min = 0, description = "Create a histogram of attribute values")
51     histogram: java.lang.Boolean,
52     @DescribeParameter(name = "sort", min = 0, description = "Sort results - allowed to be ASC or DESC")
53     sort: String,
54     @DescribeParameter(name = "sortByCount", min = 0, description = "Sort by histogram counts instead of attribute values")
55     sortByCount: java.lang.Boolean,
56     progressListener: ProgressListener): SimpleFeatureCollection = {
57 
58     val attributeDescriptor = features
59         .getSchema
60         .getAttributeDescriptors
61         .asScala
62         .find(_.getLocalName == attribute)
63         .getOrElse(throw new IllegalArgumentException(s"Attribute $attribute does not exist in feature schema."))
64 
65     val hist = Option(histogram).exists(_.booleanValue)
66     val sortBy = Option(sortByCount).exists(_.booleanValue)
67 
68     val visitor = new AttributeVisitor(features, attributeDescriptor, Option(filter).filter(_ != Filter.INCLUDE), hist)
69     GeoMesaFeatureCollection.visit(features, visitor, progressListener)
70     val uniqueValues = visitor.getResult.attributes
71 
72     val binding = attributeDescriptor.getType.getBinding
73     UniqueProcess.createReturnCollection(uniqueValues, binding, hist, Option(sort), sortBy)
74   }
75 }
76 
77 object UniqueProcess {
78 
79   val SftName = "UniqueValue"
80   val AttributeValue = "value"
81   val AttributeCount = "count"
82 
83   /**
84     * Duplicates output format from geotools UniqueProcess
85     *
86     * @param uniqueValues values
87     * @param binding value binding
88     * @param histogram include counts or just values
89     * @param sort sort
90     * @param sortByCount sort by count or by value
91     * @return
92     */
93   def createReturnCollection(uniqueValues: Map[Any, Long],
94                              binding: Class[_],
95                              histogram: Boolean,
96                              sort: Option[String],
97                              sortByCount: Boolean): SimpleFeatureCollection = {
98 
99     val ft = createUniqueSft(binding, histogram)
100 
101     val sfb = new SimpleFeatureBuilder(ft)
102 
103     val result = new ListFeatureCollection(ft)
104 
105     // if sorting was requested do it here, otherwise return results in iterator order
106     val sorted = sort.map { s =>
107       if (sortByCount) {
108         val ordering = if (s.equalsIgnoreCase("desc")) Ordering[Long].reverse else Ordering[Long]
109         uniqueValues.iterator.toList.sortBy(_._2)(ordering)
110       } else {
111         val ordering = if (s.equalsIgnoreCase("desc")) Ordering[String].reverse else Ordering[String]
112         uniqueValues.iterator.toList.sortBy(_._1.toString)(ordering)
113       }
114     }.getOrElse(uniqueValues.iterator)
115 
116     // histogram includes extra 'count' attribute
117     val addFn = if (histogram) (key: Any, value: Long) => {
118       sfb.add(key)
119       sfb.add(value)
120       result.add(sfb.buildFeature(null))
121     } else (key: Any, _: Long) => {
122       sfb.add(key)
123       result.add(sfb.buildFeature(null))
124     }
125 
126     sorted.foreach { case (key, value) => addFn(key, value) }
127 
128     result
129   }
130 
131   /**
132     * Based on geotools UniqueProcess simple feature type
133     *
134     * @param binding class of attribute
135     * @param histogram return counts or not
136     * @return
137     */
138   def createUniqueSft(binding: Class[_], histogram: Boolean): SimpleFeatureType = {
139     val sftb = new SimpleFeatureTypeBuilder
140     sftb.add(AttributeValue, binding)
141     if (histogram) {
142       // histogram includes extra 'count' attribute
143       sftb.add(AttributeCount, classOf[java.lang.Long])
144     }
145 
146     sftb.setName(SftName)
147     sftb.buildFeatureType
148   }
149 }
150 
151 /**
152  * Visitor that tracks unique attribute values and counts
153  *
154  * @param features features to evaluate
155  * @param attributeDescriptor attribute to evaluate
156  * @param filter optional filter to apply to features before evaluating
157  * @param histogram return counts or not
158  */
159 class AttributeVisitor(val features: SimpleFeatureCollection,
160                        val attributeDescriptor: AttributeDescriptor,
161                        val filter: Option[Filter],
162                        histogram: Boolean) extends GeoMesaProcessVisitor with LazyLogging {
163 
164   import scala.collection.JavaConverters._
165 
166   private val attribute    = attributeDescriptor.getLocalName
167   private val uniqueValues = mutable.Map.empty[Any, Long].withDefaultValue(0)
168 
169   private var attributeIdx: Int = -1
170 
171   // normally handled in our query planner, but we are going to use the filter directly here
172   private lazy val manualFilter = filter.map(FastFilterFactory.optimize(features.getSchema, _))
173 
174   private def getAttribute[T](f: SimpleFeature): T = {
175     if (attributeIdx == -1) {
176       attributeIdx = f.getType.indexOf(attribute)
177     }
178     f.getAttribute(attributeIdx).asInstanceOf[T]
179   }
180 
181   private def addSingularValue(f: SimpleFeature): Unit = {
182     val value = getAttribute[AnyRef](f)
183     if (value != null) {
184       uniqueValues(value) += 1
185     }
186   }
187 
188   private def addMultiValue(f: SimpleFeature): Unit = {
189     val values = getAttribute[java.util.Collection[_]](f)
190     if (values != null) {
191       values.asScala.foreach(uniqueValues(_) += 1)
192     }
193   }
194 
195   private val addValue: SimpleFeature => Unit =
196     if (attributeDescriptor.isList) { addMultiValue } else { addSingularValue }
197 
198   // non-optimized visit
199   override def visit(feature: Feature): Unit = {
200     val f = feature.asInstanceOf[SimpleFeature]
201     if (manualFilter.forall(_.evaluate(f))) {
202       addValue(f)
203     }
204   }
205 
206   override def getResult: AttributeResult = new AttributeResult(uniqueValues.toMap)
207 
208   override def execute(source: SimpleFeatureSource, query: Query): Unit = {
209 
210     import org.locationtech.geomesa.filter.mergeFilters
211 
212     logger.debug(s"Running Geomesa histogram process on source type ${source.getClass.getName}")
213 
214     // combine filters from this process and any input collection
215     filter.foreach(f => query.setFilter(mergeFilters(query.getFilter, f)))
216 
217     val sft = source.getSchema
218 
219     val enumerated = if (attributeDescriptor.isMultiValued) {
220       // stats don't support list types
221       uniqueV5(source, query)
222     } else {
223       // TODO if !histogram, we could write a new unique skipping iterator
224       query.getHints.put(QueryHints.STATS_STRING, Stat.Enumeration(attribute))
225       query.getHints.put(QueryHints.ENCODE_STATS, java.lang.Boolean.TRUE)
226 
227       // execute the query
228       val reader = source.getFeatures(query).features()
229 
230       val enumeration = try {
231         // stats should always return exactly one result, even if there are no features in the table
232         val encoded = reader.next.getAttribute(0).asInstanceOf[String]
233         StatsScan.decodeStat(sft)(encoded).asInstanceOf[EnumerationStat[Any]]
234       } finally {
235         reader.close()
236       }
237 
238       enumeration.frequencies
239     }
240 
241     uniqueValues.clear()
242     enumerated.foreach { case (k, v) => uniqueValues.put(k, v) }
243   }
244 
245   private def uniqueV5(source: SimpleFeatureSource, query: Query): Iterable[(Any, Long)] = {
246     // only return the attribute we are interested in to reduce bandwidth
247     query.setPropertyNames(Seq(attribute).asJava)
248 
249     // if there is no filter, try to force an attribute scan - should be fastest query
250     if (query.getFilter == Filter.INCLUDE && AttributeIndex.indexed(features.getSchema, attribute)) {
251       query.setFilter(AttributeVisitor.getIncludeAttributeFilter(attribute))
252     }
253 
254     // execute the query
255     SelfClosingIterator(source.getFeatures(query).features()).foreach(addValue)
256     uniqueValues.toMap
257   }
258 }
259 
260 object AttributeVisitor {
261 
262   import org.locationtech.geomesa.filter.ff
263 
264   /**
265    * Returns a filter that is equivalent to Filter.INCLUDE, but against the attribute index.
266    *
267    * @param attribute attribute to query
268    * @return
269    */
270   def getIncludeAttributeFilter(attribute: String): Filter =
271     ff.greaterOrEqual(ff.property(attribute), ff.literal(""))
272 }
273 
274 /**
275  * Result class to hold the attribute histogram
276  *
277  * @param attributes result
278  */
279 class AttributeResult(val attributes: Map[Any, Long]) extends AbstractCalcResult {
280 
281   override def getValue: java.util.Map[Any, Long] = attributes.asJava
282 
283   override def isCompatible(targetResults: CalcResult): Boolean =
284     targetResults.isInstanceOf[AttributeResult] || targetResults == CalcResult.NULL_RESULT
285 
286   override def merge(resultsToAdd: CalcResult): CalcResult = {
287     if (!isCompatible(resultsToAdd)) {
288       throw new IllegalArgumentException("Parameter is not a compatible type")
289     } else if (resultsToAdd == CalcResult.NULL_RESULT) {
290       this
291     } else if (resultsToAdd.isInstanceOf[AttributeResult]) {
292       val toAdd = resultsToAdd.getValue.asInstanceOf[Map[Any, Long]]
293       // note ++ on maps will get all keys with second maps values if exists, if not first map values
294       val merged = attributes ++ toAdd.map {
295         case (attr, count) => attr -> (count + attributes.getOrElse(attr, 0L))
296       }
297       new AttributeResult(merged)
298     } else {
299       throw new IllegalArgumentException(
300         "The CalcResults claim to be compatible, but the appropriate merge method has not been implemented.")
301     }
302   }
303 }