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.index.s2
10 
11 import org.geotools.api.feature.simple.SimpleFeatureType
12 import org.geotools.api.filter.Filter
13 import org.geotools.util.factory.Hints
14 import org.locationtech.geomesa.curve.S2SFC
15 import org.locationtech.geomesa.filter.{FilterHelper, FilterValues}
16 import org.locationtech.geomesa.index.api
17 import org.locationtech.geomesa.index.api.IndexKeySpace.IndexKeySpaceFactory
18 import org.locationtech.geomesa.index.api.ShardStrategy.{NoShardStrategy, Z2ShardStrategy}
19 import org.locationtech.geomesa.index.api._
20 import org.locationtech.geomesa.index.conf.QueryHints.LOOSE_BBOX
21 import org.locationtech.geomesa.index.conf.QueryProperties
22 import org.locationtech.geomesa.index.geotools.GeoMesaDataStoreFactory
23 import org.locationtech.geomesa.index.utils.Explainer
24 import org.locationtech.geomesa.utils.geotools.{GeometryUtils, WholeWorldPolygon}
25 import org.locationtech.geomesa.utils.index.{ByteArrays, VisibilityLevel}
26 import org.locationtech.jts.geom.{Geometry, Point}
27 
28 import scala.util.control.NonFatal
29 
30 /**
31   * @author sunyabo 2019年07月24日 14:42
32   * @version V1.0
33   */
34 class S2IndexKeySpace(val sft: SimpleFeatureType, val sharding: ShardStrategy, geomField: String)
35   extends IndexKeySpace[S2IndexValues, Long] {
36 
37   import org.locationtech.geomesa.utils.geotools.RichSimpleFeatureType.RichSimpleFeatureType
38 
39   require(classOf[Point].isAssignableFrom(sft.getDescriptor(geomField).getType.getBinding),
40     s"Expected field $geomField to have a point binding, but instead it has: " +
41       sft.getDescriptor(geomField).getType.getBinding.getSimpleName)
42 
43   private val sfc: S2SFC =
44     S2SFC(
45       QueryProperties.S2MinLevel,
46       QueryProperties.S2MaxLevel,
47       QueryProperties.S2LevelMod,
48       QueryProperties.S2MaxCells
49     )
50 
51   private val geomIndex: Int = sft.indexOf(geomField)
52 
53   /**
54     * The attributes used to create the index keys
55     *
56     * @return
57     */
58   override val attributes: Seq[String] = Seq(geomField)
59 
60   /**
61     * Length of an index key. If static (general case), will return a Right with the length. If dynamic,
62     * will return Left with a function to determine the length from a given (row, offset, length)
63     *
64     * @return
65     */
66   override val indexKeyByteLength: Right[(Array[Byte], Int, Int) => Int, Int] = Right(8 + sharding.length)
67 
68   /**
69     * Table sharing
70     *
71     * @return
72     */
73   override val sharing: Array[Byte] = Array.empty
74 
75   /**
76     * Index key from the attributes of a simple feature
77     *
78     * @param writable simple feature with cached values
79     * @param tier    tier bytes
80     * @param id      feature id bytes
81     * @param lenient if input values should be strictly checked, default false
82     * @return
83     */
84   override def toIndexKey(writable: WritableFeature,
85                           tier: Array[Byte],
86                           id: Array[Byte],
87                           lenient: Boolean): RowKeyValue[Long] = {
88     val geom = writable.getAttribute[Point](geomIndex)
89     if (geom == null) {
90       throw new IllegalArgumentException(s"Null geometry in feature ${writable.feature.getID}")
91     }
92     val s = try { sfc.index(geom.getX, geom.getY, lenient) } catch {
93       case NonFatal(e) => throw new IllegalArgumentException(s"Invalid s value from geometry: $geom", e)
94     }
95     val shard = sharding(writable)
96 
97     // create the byte array - allocate a single array up front to contain everything
98     // ignore tier, not used here
99     val bytes = Array.ofDim[Byte](shard.length + 8 + id.length)
100 
101     if (shard.isEmpty) {
102       ByteArrays.writeLong(s, bytes)
103       System.arraycopy(id, 0, bytes, 8, id.length)
104     } else {
105       bytes(0) = shard.head // shard is only a single byte
106       ByteArrays.writeLong(s, bytes, 1)
107       System.arraycopy(id, 0, bytes, 9, id.length)
108     }
109 
110     SingleRowKeyValue(bytes, sharing, shard, s, tier, id, writable.values)
111   }
112 
113   /**
114     * Extracts values out of the filter used for range and push-down predicate creation
115     *
116     * @param filter  query filter
117     * @param explain explainer
118     * @return
119     */
120   override def getIndexValues(filter: Filter, explain: Explainer): S2IndexValues = {
121 
122     val geometries: FilterValues[Geometry] = {
123       val extracted = FilterHelper.extractGeometries(filter, geomField) // intersect since we have points
124       if (extracted.nonEmpty) { extracted } else { FilterValues(Seq(WholeWorldPolygon)) }
125     }
126 
127     explain(s"Geometries: $geometries")
128 
129     if (geometries.disjoint) {
130       explain("Non-intersecting geometries extracted, short-circuiting to empty query")
131       return S2IndexValues(sfc, geometries, Seq.empty)
132     }
133 
134     // compute our ranges based on the coarse bounds for our query
135     val xy: Seq[(Double, Double, Double, Double)] = {
136       val multiplier = QueryProperties.PolygonDecompMultiplier.toInt.get
137       val bits = QueryProperties.PolygonDecompBits.toInt.get
138       geometries.values.flatMap(GeometryUtils.bounds(_, multiplier, bits))
139     }
140 
141     S2IndexValues(sfc, geometries, xy)
142   }
143 
144   /**
145     * Creates ranges over the index keys
146     *
147     * @param values     index values @see getIndexValues
148     * @param multiplier hint for how many times the ranges will be multiplied. can be used to
149     *                   inform the number of ranges generated
150     * @return
151     */
152   override def getRanges(values: S2IndexValues, multiplier: Int): Iterator[ScanRange[Long]] = {
153     val S2IndexValues(_, geoms, xy) = values
154     if (geoms.disjoint) {
155       Iterator.empty
156     } else {
157       // note: `target` will always be Some, as ScanRangesTarget has a default value
158       val target = QueryProperties.ScanRangesTarget.option.map(t => math.max(1, t.toInt / multiplier))
159       sfc.ranges(xy, -1, target).iterator.map(r => BoundedRange(r.lower, r.upper))
160     }
161   }
162 
163   /**
164     * Creates bytes from ranges
165     *
166     * @param ranges typed scan ranges. @see `getRanges`
167     * @param tier   will the ranges have tiered ranges appended, or not
168     * @return
169     */
170   override def getRangeBytes(ranges: Iterator[ScanRange[Long]], tier: Boolean): Iterator[api.ByteRange] = {
171     if (sharding.length == 0) {
172       ranges.map {
173         case BoundedRange(lo, hi) => BoundedByteRange(ByteArrays.toBytes(lo), ByteArrays.toBytesFollowingPrefix(hi))
174         case r => throw new IllegalArgumentException(s"Unexpected range type $r")
175       }
176     } else {
177       ranges.flatMap {
178         case BoundedRange(lo, hi) =>
179           val lower = ByteArrays.toBytes(lo)
180           val upper = ByteArrays.toBytesFollowingPrefix(hi)
181           sharding.shards.map(p => BoundedByteRange(ByteArrays.concat(p, lower), ByteArrays.concat(p, upper)))
182 
183         case r => throw new IllegalArgumentException(s"Unexpected range type $r")
184       }
185     }
186   }
187 
188   /**
189     * Determines if the ranges generated by `getRanges` are sufficient to fulfill the query,
190     * or if additional filtering needs to be done
191     *
192     * @param config data store config
193     * @param values index values @see getIndexValues
194     * @param hints  query hints
195     * @return
196     */
197   override def useFullFilter(values: Option[S2IndexValues],
198                              config: Option[GeoMesaDataStoreFactory.GeoMesaDataStoreConfig],
199                              hints: Hints): Boolean = {
200     // if the user has requested strict bounding boxes, we apply the full filter
201     !Option(hints.get(LOOSE_BBOX)).map(Boolean.unbox).getOrElse(config.forall(_.queries.looseBBox)) ||
202       // if the spatial predicate is rectangular (e.g. a bbox), the index is fine enough that we
203       // don't need to apply the filter on top of it. this may cause some minor errors at extremely
204       // fine resolutions, but the performance is worth it
205       // if we have a complicated geometry predicate, we need to pass it through to be evaluated
206       values.exists(_.geometries.values.exists(g => !GeometryUtils.isRectangular(g))) ||
207       // for attribute-level vis, we need to re-evaluate the filter to account for visibility of the row key
208       (sft.getVisibilityLevel == VisibilityLevel.Attribute && values.exists(_.geometries.nonEmpty))
209   }
210 
211 }
212 
213 object S2IndexKeySpace extends IndexKeySpaceFactory[S2IndexValues, Long] {
214 
215   override def supports(sft: SimpleFeatureType, attributes: Seq[String]): Boolean =
216     attributes.lengthCompare(1) == 0 && sft.indexOf(attributes.head) != -1 &&
217       classOf[Point].isAssignableFrom(sft.getDescriptor(attributes.head).getType.getBinding)
218 
219   override def apply(sft: SimpleFeatureType, attributes: Seq[String], tier: Boolean): S2IndexKeySpace = {
220     val shards = if (tier) { NoShardStrategy } else { Z2ShardStrategy(sft) }
221     new S2IndexKeySpace(sft, shards, attributes.head)
222   }
223 }