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.hbase.data
10 
11 import com.typesafe.scalalogging.{LazyLogging, StrictLogging}
12 import org.apache.hadoop.fs.Path
13 import org.apache.hadoop.hbase.TableName
14 import org.apache.hadoop.hbase.client._
15 import org.apache.hadoop.hbase.filter.MultiRowRangeFilter.RowRange
16 import org.apache.hadoop.hbase.filter.{FilterList, KeyOnlyFilter, MultiRowRangeFilter, Filter => HFilter}
17 import org.apache.hadoop.hbase.io.compress.Compression
18 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding
19 import org.apache.hadoop.hbase.regionserver.BloomType
20 import org.apache.hadoop.hbase.security.visibility.CellVisibility
21 import org.locationtech.geomesa.hbase.HBaseSystemProperties
22 import org.locationtech.geomesa.index.api.IndexAdapter.RequiredVisibilityWriter
23 import org.locationtech.geomesa.utils.concurrent.CachedThreadPool
24 import org.locationtech.geomesa.utils.io.{IsCloseable, IsFlushableImplicits}
25 
26 import java.nio.charset.StandardCharsets
27 import java.util.regex.Pattern
28 import java.util.{Collections, Locale, UUID}
29 import scala.util.Try
30 // noinspection ScalaDeprecation
31 import org.geotools.api.feature.simple.{SimpleFeature, SimpleFeatureType}
32 import org.locationtech.geomesa.hbase.HBaseSystemProperties.{CoprocessorPath, CoprocessorUrl, TableAvailabilityTimeout}
33 import org.locationtech.geomesa.hbase.aggregators.HBaseArrowAggregator.HBaseArrowResultsToFeatures
34 import org.locationtech.geomesa.hbase.aggregators.HBaseBinAggregator.HBaseBinResultsToFeatures
35 import org.locationtech.geomesa.hbase.aggregators.HBaseDensityAggregator.HBaseDensityResultsToFeatures
36 import org.locationtech.geomesa.hbase.aggregators.HBaseStatsAggregator.HBaseStatsResultsToFeatures
37 import org.locationtech.geomesa.hbase.aggregators.{HBaseArrowAggregator, HBaseBinAggregator, HBaseDensityAggregator, HBaseStatsAggregator}
38 import org.locationtech.geomesa.hbase.data.HBaseQueryPlan.{CoprocessorPlan, EmptyPlan, ScanPlan, TableScan}
39 import org.locationtech.geomesa.hbase.rpc.coprocessor.GeoMesaCoprocessor
40 import org.locationtech.geomesa.hbase.rpc.filter._
41 import org.locationtech.geomesa.hbase.utils.HBaseVersions
42 import org.locationtech.geomesa.index.api.IndexAdapter.BaseIndexWriter
43 import org.locationtech.geomesa.index.api.QueryPlan.{FeatureReducer, IndexResultsToFeatures}
44 import org.locationtech.geomesa.index.api.WritableFeature.FeatureWrapper
45 import org.locationtech.geomesa.index.api._
46 import org.locationtech.geomesa.index.conf.QueryHints
47 import org.locationtech.geomesa.index.filters.{S2Filter, S3Filter, Z2Filter, Z3Filter}
48 import org.locationtech.geomesa.index.index.id.IdIndex
49 import org.locationtech.geomesa.index.index.s2.{S2Index, S2IndexValues}
50 import org.locationtech.geomesa.index.index.s3.{S3Index, S3IndexValues}
51 import org.locationtech.geomesa.index.index.z2.{Z2Index, Z2IndexValues}
52 import org.locationtech.geomesa.index.index.z3.{Z3Index, Z3IndexValues}
53 import org.locationtech.geomesa.index.iterators.StatsScan
54 import org.locationtech.geomesa.index.planning.LocalQueryRunner.LocalTransformReducer
55 import org.locationtech.geomesa.utils.index.ByteArrays
56 import org.locationtech.geomesa.utils.io.{CloseWithLogging, FlushWithLogging, WithClose}
57 import org.locationtech.geomesa.utils.text.StringSerialization
58 
59 import java.util.concurrent.TimeUnit
60 import scala.util.Random
61 import scala.util.control.NonFatal
62 
63 class HBaseIndexAdapter(ds: HBaseDataStore) extends IndexAdapter[HBaseDataStore] with StrictLogging {
64 
65   import HBaseIndexAdapter._
66   import org.locationtech.geomesa.utils.geotools.RichSimpleFeatureType.RichSimpleFeatureType
67 
68   import scala.collection.JavaConverters._
69 
70   override def createTable(
71       index: GeoMesaFeatureIndex[_, _],
72       partition: Option[String],
73       splits: => Seq[Array[Byte]]): Unit = {
74     // write table name to metadata
75     val name = TableName.valueOf(index.configureTableName(partition, tableNameLimit))
76 
77     WithClose(ds.connection.getAdmin) { admin =>
78       if (!admin.tableExists(name)) {
79         logger.debug(s"Creating table $name")
80 
81         val conf = admin.getConfiguration
82 
83         val compression = index.sft.getCompression.map { alg =>
84           logger.debug(s"Setting compression '$alg' on table $name for feature ${index.sft.getTypeName}")
85           // note: all compression types in HBase are case-sensitive and lower-cased
86           Compression.getCompressionAlgorithmByName(alg.toLowerCase(Locale.US))
87         }
88 
89         val cols = groups.apply(index.sft).map(_._1)
90         val bloom = Some(BloomType.NONE)
91         val encoding = if (index.name == IdIndex.name) { None } else { Some(DataBlockEncoding.FAST_DIFF) }
92 
93         // noinspection ScalaDeprecation
94         val coprocessor = if (!ds.config.remoteFilter) { None } else {
95           def urlFromSysProp: Option[Path] = CoprocessorUrl.option.orElse(CoprocessorPath.option).map(new Path(_))
96           lazy val coprocessorUrl = ds.config.coprocessors.url.orElse(urlFromSysProp).orElse {
97             try {
98               // the jar should be under hbase.dynamic.jars.dir to enable filters, so look there
99               val dir = new Path(conf.get("hbase.dynamic.jars.dir"))
100               WithClose(dir.getFileSystem(conf)) { fs =>
101                 if (!fs.isDirectory(dir)) { None } else {
102                   fs.listStatus(dir).collectFirst {
103                     case s if distributedJarNamePattern.matcher(s.getPath.getName).matches() => s.getPath
104                   }
105                 }
106               }
107             } catch {
108               case NonFatal(e) => logger.warn("Error checking dynamic jar path:", e); None
109             }
110           }
111           // if the coprocessors are installed site-wide don't register them in the table descriptor.
112           // this key is CoprocessorHost.USER_REGION_COPROCESSOR_CONF_KEY - but don't want to pull in
113           // a dependency on hbase-server just for this constant
114           val installed = Option(conf.get("hbase.coprocessor.user.region.classes"))
115           val names = installed.map(_.split(":").toSet).getOrElse(Set.empty[String])
116           if (names.contains(CoprocessorClass)) { None } else {
117             logger.debug(s"Using coprocessor path ${coprocessorUrl.orNull}")
118             // TODO: Warn if the path given is different from paths registered in other coprocessors
119             // if so, other tables would need updating
120             Some(CoprocessorClass -> coprocessorUrl)
121           }
122         }
123 
124         try {
125           HBaseVersions.createTableAsync(admin, name, cols, bloom, compression, encoding, None, coprocessor, splits)
126         } catch {
127           case _: org.apache.hadoop.hbase.TableExistsException => // ignore, another thread created it for us
128         }
129       }
130 
131       waitForTable(admin, name)
132     }
133   }
134 
135   override def renameTable(from: String, to: String): Unit = {
136     WithClose(ds.connection.getAdmin) { admin =>
137       val existing = TableName.valueOf(from)
138       val renamed = TableName.valueOf(to)
139       if (admin.tableExists(existing)) {
140         // renaming in hbase requires creating a snapshot and using that to create the new table
141         val snapshot = StringSerialization.alphaNumericSafeString(UUID.randomUUID().toString)
142         admin.disableTable(existing)
143         admin.snapshot(snapshot, existing)
144         admin.cloneSnapshot(snapshot, renamed)
145         admin.deleteSnapshot(snapshot)
146         admin.deleteTable(existing)
147         waitForTable(admin, renamed)
148       }
149     }
150   }
151 
152   override def deleteTables(tables: Seq[String]): Unit = {
153     WithClose(ds.connection.getAdmin) { admin =>
154       def deleteOne(name: String): Unit = {
155         val table = TableName.valueOf(name)
156         if (admin.tableExists(table)) {
157           HBaseVersions.disableTableAsync(admin, table)
158           val timeout = TableAvailabilityTimeout.toUnboundedDuration.filter(_.isFinite)
159           logger.debug(s"Waiting for table '$table' to be disabled with " +
160               s"${timeout.map(t => s"a timeout of $t").getOrElse("no timeout")}")
161           val stop = timeout.map(t => System.currentTimeMillis() + t.toMillis)
162           while (!admin.isTableDisabled(table) && stop.forall(_ > System.currentTimeMillis())) {
163             Thread.sleep(1000)
164           }
165           // no async operation, but timeout can be controlled through hbase-site.xml "hbase.client.sync.wait.timeout.msec"
166           admin.deleteTable(table)
167         }
168       }
169       tables.toList.map(t => CachedThreadPool.submit(() => deleteOne(t))).foreach(_.get)
170     }
171   }
172 
173   override def clearTables(tables: Seq[String], prefix: Option[Array[Byte]]): Unit = {
174     def clearOne(name: String): Unit = {
175       val tableName = TableName.valueOf(name)
176       WithClose(ds.connection.getTable(tableName)) { table =>
177         val scan = new Scan().setFilter(new KeyOnlyFilter)
178         prefix.foreach(scan.setRowPrefixFilter)
179         ds.applySecurity(scan)
180         val mutateParams = new BufferedMutatorParams(tableName)
181         WithClose(table.getScanner(scan), ds.connection.getBufferedMutator(mutateParams)) { case (scanner, mutator) =>
182           scanner.iterator.asScala.grouped(10000).foreach { result =>
183             // TODO GEOMESA-2546 set delete visibilities
184             val deletes = result.map(r => new Delete(r.getRow))
185             mutator.mutate(deletes.asJava)
186           }
187         }
188       }
189     }
190     tables.toList.map(t => CachedThreadPool.submit(() => clearOne(t))).foreach(_.get)
191   }
192 
193   override def createQueryPlan(strategy: QueryStrategy): HBaseQueryPlan = {
194 
195     import org.locationtech.geomesa.index.conf.QueryHints.RichHints
196 
197     val QueryStrategy(filter, byteRanges, _, _, ecql, hints, _) = strategy
198     val index = filter.index
199 
200     // index api defines empty start/end for open-ended range
201     // index api defines start row inclusive, end row exclusive
202     // both these conventions match the conventions for hbase scan objects
203     val ranges = byteRanges.map {
204       case BoundedByteRange(start, stop) => new RowRange(start, true, stop, false)
205       case SingleRowByteRange(row)       => new RowRange(row, true, ByteArrays.rowFollowingRow(row), false)
206     }
207     val small = byteRanges.headOption.exists(_.isInstanceOf[SingleRowByteRange])
208 
209     val tables = index.getTablesForQuery(filter.filter).map(TableName.valueOf)
210     val (colFamily, schema) = groups.group(index.sft, hints.getTransformDefinition, ecql)
211 
212     val transform: Option[(String, SimpleFeatureType)] = hints.getTransform
213 
214     // check for an empty query plan, if there are no tables or ranges to scan
215     def empty(reducer: Option[FeatureReducer]): Option[HBaseQueryPlan] =
216       if (tables.isEmpty || ranges.isEmpty) { Some(EmptyPlan(filter, reducer)) } else { None }
217 
218     if (!ds.config.remoteFilter) {
219       // everything is done client side
220       // note: we assume visibility filtering is still done server-side as it's part of core hbase
221       // note: we use the full filter here, since we can't use the z3 server-side filter
222       // for some attribute queries we wouldn't need the full filter...
223       val reducer = Some(new LocalTransformReducer(schema, filter.filter, None, transform, hints))
224       empty(reducer).getOrElse {
225         val scans = configureScans(tables, ranges, small, colFamily, Seq.empty, coprocessor = false)
226         val resultsToFeatures = new HBaseResultsToFeatures(index, schema)
227         val sort = hints.getSortFields
228         val max = hints.getMaxFeatures
229         val project = hints.getProjection
230         ScanPlan(filter, ranges, scans, resultsToFeatures, reducer, sort, max, project)
231       }
232     } else {
233       // TODO pull this out to be SPI loaded so that new indices can be added seamlessly
234       val indexFilter = strategy.index match {
235         case _: Z3Index =>
236           strategy.values.map { case v: Z3IndexValues =>
237             (Z3HBaseFilter.Priority, Z3HBaseFilter(Z3Filter(v), index.keySpace.sharding.length))
238           }
239 
240         case _: Z2Index =>
241           strategy.values.map { case v: Z2IndexValues =>
242             (Z2HBaseFilter.Priority, Z2HBaseFilter(Z2Filter(v), index.keySpace.sharding.length))
243           }
244 
245         case _: S2Index =>
246           strategy.values.map { case v: S2IndexValues =>
247             (S2HBaseFilter.Priority, S2HBaseFilter(S2Filter(v), index.keySpace.sharding.length))
248           }
249 
250         case _: S3Index =>
251           strategy.values.map { case v: S3IndexValues =>
252             (S3HBaseFilter.Priority, S3HBaseFilter(S3Filter(v), index.keySpace.sharding.length))
253           }
254         // TODO GEOMESA-1807 deal with non-points in a pushdown XZ filter
255 
256         case _ => None
257       }
258 
259       val max = hints.getMaxFeatures
260       val projection = hints.getProjection
261       lazy val returnSchema = transform.map(_._2).getOrElse(schema)
262       lazy val filters = {
263         val cqlFilter = if (ecql.isEmpty && transform.isEmpty && hints.getSampling.isEmpty) { Seq.empty } else {
264           Seq((CqlTransformFilter.Priority, CqlTransformFilter(schema, strategy.index, ecql, transform, hints)))
265         }
266         (cqlFilter ++ indexFilter).sortBy(_._1).map(_._2)
267       }
268       lazy val coprocessorOptions =
269         Map(GeoMesaCoprocessor.YieldOpt -> String.valueOf(ds.config.coprocessors.yieldPartialResults))
270       lazy val scans = configureScans(tables, ranges, small, colFamily, filters, coprocessor = false)
271       lazy val coprocessorScans =
272         configureScans(tables, ranges, small, colFamily, indexFilter.toSeq.map(_._2), coprocessor = true)
273       lazy val resultsToFeatures = new HBaseResultsToFeatures(index, returnSchema)
274       lazy val localReducer = Some(new LocalTransformReducer(returnSchema, None, None, None, hints))
275 
276       if (hints.isDensityQuery) {
277         empty(None).getOrElse {
278           if (ds.config.coprocessors.enabled.density) {
279             val options = HBaseDensityAggregator.configure(schema, index, ecql, hints) ++ coprocessorOptions
280             val results = new HBaseDensityResultsToFeatures()
281             CoprocessorPlan(filter, ranges, coprocessorScans, options, results, None, max, projection)
282           } else {
283             if (hints.isSkipReduce) {
284               // override the return sft to reflect what we're actually returning,
285               // since the density sft is only created in the local reduce step
286               hints.hints.put(QueryHints.Internal.RETURN_SFT, returnSchema)
287             }
288             ScanPlan(filter, ranges, scans, resultsToFeatures, localReducer, None, max, projection)
289           }
290         }
291       } else if (hints.isArrowQuery) {
292         val config = HBaseArrowAggregator.configure(schema, index, ds.stats, filter.filter, ecql, hints)
293         val reducer = Some(config.reduce)
294         empty(reducer).getOrElse {
295           if (ds.config.coprocessors.enabled.arrow) {
296             val options = config.config ++ coprocessorOptions
297             val results = new HBaseArrowResultsToFeatures()
298             CoprocessorPlan(filter, ranges, coprocessorScans, options, results, reducer, max, projection)
299           } else {
300             if (hints.isSkipReduce) {
301               // override the return sft to reflect what we're actually returning,
302               // since the arrow sft is only created in the local reduce step
303               hints.hints.put(QueryHints.Internal.RETURN_SFT, returnSchema)
304             }
305             ScanPlan(filter, ranges, scans, resultsToFeatures, localReducer, None, max, projection)
306           }
307         }
308       } else if (hints.isStatsQuery) {
309         val reducer = Some(StatsScan.StatsReducer(returnSchema, hints))
310         empty(reducer).getOrElse {
311           if (ds.config.coprocessors.enabled.stats) {
312             val options = HBaseStatsAggregator.configure(schema, index, ecql, hints) ++ coprocessorOptions
313             val results = new HBaseStatsResultsToFeatures()
314             CoprocessorPlan(filter, ranges, coprocessorScans, options, results, reducer, max, projection)
315           } else {
316             if (hints.isSkipReduce) {
317               // override the return sft to reflect what we're actually returning,
318               // since the stats sft is only created in the local reduce step
319               hints.hints.put(QueryHints.Internal.RETURN_SFT, returnSchema)
320             }
321             ScanPlan(filter, ranges, scans, resultsToFeatures, localReducer, None, max, projection)
322           }
323         }
324       } else if (hints.isBinQuery) {
325         empty(None).getOrElse {
326           if (ds.config.coprocessors.enabled.bin) {
327             val options = HBaseBinAggregator.configure(schema, index, ecql, hints) ++ coprocessorOptions
328             val results = new HBaseBinResultsToFeatures()
329             CoprocessorPlan(filter, ranges, coprocessorScans, options , results, None, max, projection)
330           } else {
331             if (hints.isSkipReduce) {
332               // override the return sft to reflect what we're actually returning,
333               // since the bin sft is only created in the local reduce step
334               hints.hints.put(QueryHints.Internal.RETURN_SFT, returnSchema)
335             }
336             ScanPlan(filter, ranges, scans, resultsToFeatures, localReducer, None, max, projection)
337           }
338         }
339       } else {
340         empty(None).getOrElse {
341           ScanPlan(filter, ranges, scans, resultsToFeatures, None, hints.getSortFields, max, projection)
342         }
343       }
344     }
345   }
346 
347   override def createWriter(
348       sft: SimpleFeatureType,
349       indices: Seq[GeoMesaFeatureIndex[_, _]],
350       partition: Option[String],
351       atomic: Boolean): HBaseIndexWriter = {
352     require(!atomic, "HBase data store does not currently support atomic writes")
353     val wrapper = WritableFeature.wrapper(sft, groups)
354     if (sft.isVisibilityRequired) {
355       new HBaseIndexWriter(ds, indices, wrapper, partition) with RequiredVisibilityWriter
356     } else {
357       new HBaseIndexWriter(ds, indices, wrapper, partition)
358     }
359   }
360 
361   /**
362    * Configure the hbase scan
363    *
364    * @param tables tables being scanned, used for region location information
365    * @param ranges ranges to scan, non-empty. needs to be mutable as we will sort it in place
366    * @param small whether 'small' ranges (i.e. gets)
367    * @param colFamily col family to scan
368    * @param filters scan filters
369    * @param coprocessor is this a coprocessor scan or not
370    * @return
371    */
372   protected def configureScans(
373       tables: Seq[TableName],
374       ranges: Seq[RowRange],
375       small: Boolean,
376       colFamily: Array[Byte],
377       filters: Seq[HFilter],
378       coprocessor: Boolean): Seq[TableScan] = {
379     val cacheBlocks = HBaseSystemProperties.ScannerBlockCaching.toBoolean.get // has a default value so .get is safe
380     val cacheSize = HBaseSystemProperties.ScannerCaching.toInt
381 
382     logger.debug(s"HBase client scanner: block caching: $cacheBlocks, caching: $cacheSize")
383 
384     if (small && !coprocessor) {
385       val filter = filters match {
386         case Nil    => None
387         case Seq(f) => Some(f)
388         case f      => Some(new FilterList(f: _*))
389       }
390       // note: we have to copy the ranges for each table scan
391       tables.map { table =>
392         val scans = ranges.map { r =>
393           val scan = new Scan(r.getStartRow, r.getStopRow)
394           scan.addFamily(colFamily).setCacheBlocks(cacheBlocks).setSmall(true)
395           filter.foreach(scan.setFilter)
396           cacheSize.foreach(scan.setCaching)
397           ds.applySecurity(scan)
398           scan
399         }
400         TableScan(table, scans)
401       }
402     } else {
403       // split and group ranges by region server
404       // note: we have to copy the ranges for each table scan anyway
405       val rangesPerTable: Seq[(TableName, collection.Map[String, java.util.List[RowRange]])] =
406         tables.map(t => t -> groupRangesByRegion(t, ranges))
407 
408       def createGroup(group: java.util.List[RowRange]): Scan = {
409         val scan = new Scan(group.get(0).getStartRow, group.get(group.size() - 1).getStopRow)
410         val mrrf = if (group.size() < 2) { filters } else {
411           // TODO GEOMESA-1806
412           // currently, the MultiRowRangeFilter constructor will call sortAndMerge a second time
413           // this is unnecessary as we have already sorted and merged
414           // note: mrrf first priority
415           filters.+:(new MultiRowRangeFilter(group))
416         }
417         scan.setFilter(if (mrrf.lengthCompare(1) > 0) { new FilterList(mrrf: _*) } else { mrrf.headOption.orNull })
418         scan.addFamily(colFamily).setCacheBlocks(cacheBlocks)
419         cacheSize.foreach(scan.setCaching)
420 
421         // apply visibilities
422         ds.applySecurity(scan)
423 
424         scan
425       }
426 
427       rangesPerTable.map { case (table, rangesPerRegion) =>
428         val maxRangesPerGroup = {
429           def calcMax(maxPerGroup: Int, threads: Int): Int = {
430             val totalRanges = rangesPerRegion.values.map(_.size).sum
431             math.min(maxPerGroup, math.max(1, math.ceil(totalRanges.toDouble / threads).toInt))
432           }
433           if (coprocessor) {
434             calcMax(ds.config.coprocessors.maxRangesPerExtendedScan, ds.config.coprocessors.threads)
435           } else {
436             calcMax(ds.config.queries.maxRangesPerExtendedScan, ds.config.queries.threads)
437           }
438         }
439 
440         val groupedScans = Seq.newBuilder[Scan]
441 
442         rangesPerRegion.foreach { case (_, list) =>
443           // our ranges are non-overlapping, so just sort them but don't bother merging them
444           Collections.sort(list)
445 
446           var i = 0
447           while (i < list.size()) {
448             val groupSize = math.min(maxRangesPerGroup, list.size() - i)
449             groupedScans += createGroup(list.subList(i, i + groupSize))
450             i += groupSize
451           }
452         }
453 
454         // shuffle the ranges, otherwise our threads will tend to all hit the same region server at once
455         TableScan(table, Random.shuffle(groupedScans.result))
456       }
457     }
458   }
459 
460   /**
461    * Split and group ranges by region server
462    *
463    * @param table table being scanned
464    * @param ranges ranges to group
465    * @return
466    */
467   private def groupRangesByRegion(
468       table: TableName,
469       ranges: Seq[RowRange]): scala.collection.Map[String, java.util.List[RowRange]] = {
470     val rangesPerRegion = scala.collection.mutable.Map.empty[String, java.util.List[RowRange]]
471     WithClose(ds.connection.getRegionLocator(table)) { locator =>
472       ranges.foreach(groupRange(locator, _, rangesPerRegion))
473     }
474     rangesPerRegion
475   }
476 
477   /**
478    * Group the range based on the region server hosting it. Splits ranges as needed if they span
479    * more than one region
480    *
481    * @param locator region locator
482    * @param range range to group
483    * @param result collected results
484    */
485   @scala.annotation.tailrec
486   private def groupRange(
487       locator: RegionLocator,
488       range: RowRange,
489       result: scala.collection.mutable.Map[String, java.util.List[RowRange]]): Unit = {
490     var encodedName: String = null
491     var split: Array[Byte] = null
492     try {
493       val regionInfo = locator.getRegionLocation(range.getStartRow).getRegionInfo
494       encodedName = regionInfo.getEncodedName
495       val regionEndKey = regionInfo.getEndKey // note: this is exclusive
496       if (regionEndKey.nonEmpty &&
497           (range.getStopRow.isEmpty || ByteArrays.ByteOrdering.compare(regionEndKey, range.getStopRow) <= 0)) {
498         if (ByteArrays.ByteOrdering.compare(range.getStartRow, regionEndKey) < 0) {
499           split = regionEndKey
500         } else {
501           logger.warn(s"HBase region location does not correspond to requested range:\n" +
502               s"  requested row: ${ByteArrays.toHex(range.getStartRow)}\n" +
503               s"  region: $encodedName ${ByteArrays.toHex(regionInfo.getStartKey)} :: ${ByteArrays.toHex(regionEndKey)}")
504         }
505       }
506     } catch {
507       case NonFatal(e) => logger.warn(s"Error checking range location for '$range''", e)
508     }
509     val buffer = result.getOrElseUpdate(encodedName, new java.util.ArrayList())
510     if (split == null) {
511       buffer.add(range)
512     } else {
513       // split the range based on the current region
514       buffer.add(new RowRange(range.getStartRow, true, split, false))
515       groupRange(locator, new RowRange(split, true, range.getStopRow, false), result)
516     }
517   }
518 }
519 
520 object HBaseIndexAdapter extends LazyLogging {
521 
522   private val distributedJarNamePattern = Pattern.compile("^geomesa-hbase-distributed-runtime.*\\.jar$")
523 
524   // these are in the geomesa-hbase-server module, so not accessible directly
525   val CoprocessorClass = "org.locationtech.geomesa.hbase.server.coprocessor.GeoMesaCoprocessor"
526   val AggregatorPackage = "org.locationtech.geomesa.hbase.server.common"
527 
528   val durability: Durability = HBaseSystemProperties.WalDurability.option match {
529     case Some(value) =>
530       Durability.values.find(_.toString.equalsIgnoreCase(value)).getOrElse {
531         logger.error(s"Invalid HBase WAL durability setting: $value. Falling back to default durability")
532         Durability.USE_DEFAULT
533       }
534     case None => Durability.USE_DEFAULT
535   }
536 
537   /**
538    * Waits for a table to come online after being created
539    *
540    * @param admin hbase admin
541    * @param table table name
542    */
543   def waitForTable(admin: Admin, table: TableName): Unit = {
544     if (!admin.isTableAvailable(table)) {
545       val timeout = TableAvailabilityTimeout.toUnboundedDuration.filter(_.isFinite)
546       logger.debug(s"Waiting for table '$table' to become available with " +
547           s"${timeout.map(t => s"a timeout of $t").getOrElse("no timeout")}")
548       val stop = timeout.map(t => System.currentTimeMillis() + t.toMillis)
549       while (!admin.isTableAvailable(table) && stop.forall(_ > System.currentTimeMillis())) {
550         Thread.sleep(1000)
551       }
552     }
553   }
554 
555   /**
556     * Deserializes row bytes into simple features
557     *
558     * @param _index index
559     * @param _sft sft
560     */
561   class HBaseResultsToFeatures(_index: GeoMesaFeatureIndex[_, _], _sft: SimpleFeatureType) extends
562     IndexResultsToFeatures[Result](_index, _sft) {
563 
564     def this() = this(null, null) // no-arg constructor required for serialization
565 
566     override def apply(result: Result): SimpleFeature = {
567       val cell = result.rawCells()(0)
568       val id = index.getIdFromRow(cell.getRowArray, cell.getRowOffset, cell.getRowLength, null)
569       serializer.deserialize(id, cell.getValueArray, cell.getValueOffset, cell.getValueLength)
570     }
571   }
572 
573   /**
574     * Writer for hbase
575     *
576     * @param ds datastore
577     * @param indices indices to write to
578     * @param partition partition to write to
579     */
580   class HBaseIndexWriter(
581       ds: HBaseDataStore,
582       indices: Seq[GeoMesaFeatureIndex[_, _]],
583       wrapper: FeatureWrapper[WritableFeature],
584       partition: Option[String]
585     ) extends BaseIndexWriter(indices, wrapper) {
586 
587     import org.locationtech.geomesa.utils.geotools.RichSimpleFeatureType.RichSimpleFeatureType
588 
589     private val batchSize = HBaseSystemProperties.WriteBatchSize.toLong
590     private val flushTimeout = HBaseSystemProperties.WriteFlushTimeout.toLong
591     private val deleteVis = HBaseSystemProperties.DeleteVis.option.map(new CellVisibility(_))
592 
593     private val pools = {
594       // mimic config from default hbase connection
595       val maxThreads = math.max(1, ds.connection.getConfiguration.getInt("hbase.htable.threads.max", Int.MaxValue))
596       Array.fill(indices.length)(new CachedThreadPool(maxThreads))
597     }
598 
599     private val mutators = indices.toArray.map { index =>
600       // should always be writing to a single table here
601       val table = index.getTableName(partition)
602       val params = new BufferedMutatorParams(TableName.valueOf(table))
603       batchSize.foreach(params.writeBufferSize)
604       flushTimeout.foreach(params.setWriteBufferPeriodicFlushTimeoutMs)
605 
606       // We have to pass a pool explicitly and close it after manually,
607       // cause of HBase issue where pools got leaked and never closed
608       // (in case of long running Spark jobs 24+ hours the workers go out of memory without custom pool)
609       params.pool(pools(indices.indexOf(index)))
610       ds.connection.getBufferedMutator(params)
611     }
612 
613     private val expiration = indices.headOption.flatMap(_.sft.getFeatureExpiration).orNull
614 
615     private var i = 0
616 
617     override protected def append(feature: WritableFeature, values: Array[RowKeyValue[_]]): Unit = {
618       val ttl = if (expiration != null) {
619         val t = expiration.expires(feature.feature) - System.currentTimeMillis
620         if (t > 0) {
621           t
622         }
623         else {
624           logger.warn("Feature is already past its TTL; not added to database")
625           return
626         }
627       } else {
628         0L
629       }
630 
631       i = 0
632       while (i < values.length) {
633         val mutator = mutators(i)
634         values(i) match {
635           case kv: SingleRowKeyValue[_] =>
636             kv.values.foreach { value =>
637               val put = new Put(kv.row)
638               put.addImmutable(value.cf, value.cq, value.value)
639               if (!value.vis.isEmpty) {
640                 put.setCellVisibility(new CellVisibility(new String(value.vis, StandardCharsets.UTF_8)))
641               }
642               put.setDurability(durability)
643               if (ttl > 0) put.setTTL(ttl)
644               mutator.mutate(put)
645             }
646 
647           case mkv: MultiRowKeyValue[_] =>
648             mkv.rows.foreach { row =>
649               mkv.values.foreach { value =>
650                 val put = new Put(row)
651                 put.addImmutable(value.cf, value.cq, value.value)
652                 if (!value.vis.isEmpty) {
653                   put.setCellVisibility(new CellVisibility(new String(value.vis, StandardCharsets.UTF_8)))
654                 }
655                 put.setDurability(durability)
656                 if (ttl > 0) put.setTTL(ttl)
657                 mutator.mutate(put)
658               }
659             }
660         }
661         i += 1
662       }
663     }
664 
665     override protected def update(
666         feature: WritableFeature,
667         values: Array[RowKeyValue[_]],
668         previous: WritableFeature,
669         previousValues: Array[RowKeyValue[_]]): Unit = {
670       delete(previous, previousValues)
671       // for updates, ensure that our timestamps don't clobber each other
672       flush()
673       Thread.sleep(1)
674       append(feature, values)
675     }
676 
677     override protected def delete(feature: WritableFeature, values: Array[RowKeyValue[_]]): Unit = {
678       i = 0
679       while (i < values.length) {
680         val mutator = mutators(i)
681         values(i) match {
682           case kv: SingleRowKeyValue[_] =>
683             kv.values.foreach { value =>
684               val del = new Delete(kv.row)
685               del.addFamily(value.cf) // note: passing in the column qualifier seems to keep deletes from working
686               if (!value.vis.isEmpty) {
687                 del.setCellVisibility(new CellVisibility(new String(value.vis, StandardCharsets.UTF_8)))
688               } else {
689                 deleteVis.foreach(del.setCellVisibility)
690               }
691               mutator.mutate(del)
692             }
693 
694           case mkv: MultiRowKeyValue[_] =>
695             mkv.rows.foreach { row =>
696               mkv.values.foreach { value =>
697                 val del = new Delete(row)
698                 del.addFamily(value.cf) // note: passing in the column qualifier seems to keep deletes from working
699                 if (!value.vis.isEmpty) {
700                   del.setCellVisibility(new CellVisibility(new String(value.vis, StandardCharsets.UTF_8)))
701                 } else {
702                   deleteVis.foreach(del.setCellVisibility)
703                 }
704                 mutator.mutate(del)
705               }
706             }
707         }
708         i += 1
709       }
710     }
711 
712     override def flush(): Unit = FlushWithLogging.raise(mutators)(BufferedMutatorIsFlushable.arrayIsFlushable)
713 
714     override def close(): Unit = {
715       try { CloseWithLogging.raise(mutators) } finally {
716         pools.foreach(CloseWithLogging(_)(IsCloseable.executorServiceIsCloseable))
717       }
718       if (!pools.foldLeft(true) { case (terminated, pool) => terminated && pool.awaitTermination(60, TimeUnit.SECONDS) }) {
719         logger.warn("Failed to terminate thread pool after 60 seconds")
720       }
721     }
722   }
723 
724   object BufferedMutatorIsFlushable extends IsFlushableImplicits[BufferedMutator] {
725     override protected def flush(f: BufferedMutator): Try[Unit] = Try(f.flush())
726   }
727 
728 }