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.accumulo.iterators
10 
11 import com.typesafe.scalalogging.LazyLogging
12 import org.apache.accumulo.core.client.IteratorSetting
13 import org.apache.accumulo.core.client.admin.TableOperations
14 import org.apache.accumulo.core.data.{Key, Value}
15 import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope
16 import org.apache.accumulo.core.iterators.{IteratorEnvironment, SortedKeyValueIterator}
17 import org.geotools.api.feature.simple.SimpleFeatureType
18 import org.locationtech.geomesa.index.api.GeoMesaFeatureIndex
19 import org.locationtech.geomesa.index.filters.{AgeOffFilter, DtgAgeOffFilter}
20 import org.locationtech.geomesa.utils.conf.FeatureExpiration
21 import org.locationtech.geomesa.utils.conf.FeatureExpiration.FeatureTimeExpiration
22 
23 import java.util.concurrent.TimeUnit
24 import scala.concurrent.duration.Duration
25 import scala.util.control.NonFatal
26 
27 /**
28   * Age off data based on the dtg value stored in the SimpleFeature
29   */
30 class DtgAgeOffIterator extends AgeOffIterator with DtgAgeOffFilter {
31 
32   override def init(source: SortedKeyValueIterator[Key, Value],
33                     options: java.util.Map[String, String],
34                     env: IteratorEnvironment): Unit = {
35     import org.locationtech.geomesa.utils.geotools.RichSimpleFeatureType.RichSimpleFeatureType
36 
37     import scala.collection.JavaConverters._
38 
39     super.init(source, options, env)
40     try {
41       super.init(options.asScala.toMap)
42     } catch {
43       case _: NoSuchElementException => dtgIndex = sft.getDtgIndex.get // fallback for old configuration
44     }
45   }
46 
47   override def accept(k: Key, v: Value): Boolean = {
48     val value = v.get()
49     accept(null, -1, -1, value, 0, value.length, -1)
50   }
51 
52   override def deepCopy(env: IteratorEnvironment): SortedKeyValueIterator[Key, Value] = {
53     val copy = super.deepCopy(env).asInstanceOf[DtgAgeOffIterator]
54     copy.sft = sft
55     copy.index = index
56     copy.reusableSf = reusableSf
57     copy.dtgIndex = dtgIndex
58     copy
59   }
60 }
61 
62 object DtgAgeOffIterator extends LazyLogging {
63 
64   val Name = "dtg-age-off"
65 
66   def configure(sft: SimpleFeatureType,
67                 index: GeoMesaFeatureIndex[_, _],
68                 expiry: Duration,
69                 dtgField: Option[String],
70                 priority: Int = 5): IteratorSetting = {
71     val is = new IteratorSetting(priority, Name, classOf[DtgAgeOffIterator])
72     DtgAgeOffFilter.configure(sft, index, expiry, dtgField).foreach { case (k, v) => is.addOption(k, v) }
73     is
74   }
75 
76   def expiry(tableOps: TableOperations, sft: SimpleFeatureType, table: String): Option[FeatureExpiration] = {
77     try {
78       list(tableOps, table).map { is =>
79         val attribute = sft.getDescriptor(is.getOptions.get(DtgAgeOffFilter.Configuration.DtgOpt).toInt).getLocalName
80         val expiry = java.time.Duration.parse(is.getOptions.get(AgeOffFilter.Configuration.ExpiryOpt)).toMillis
81         FeatureTimeExpiration(attribute, sft.indexOf(attribute), Duration(expiry, TimeUnit.MILLISECONDS))
82       }
83     } catch {
84       case NonFatal(e) => logger.error("Error converting iterator settings to FeatureExpiration:", e); None
85     }
86   }
87 
88   def list(tableOps: TableOperations, table: String): Option[IteratorSetting] = {
89     import org.locationtech.geomesa.utils.conversions.ScalaImplicits.RichIterator
90     IteratorScope.values.iterator.flatMap(scope => Option(tableOps.getIteratorSetting(table, Name, scope))).headOption
91   }
92 
93   def set(
94       tableOps: TableOperations,
95       sft: SimpleFeatureType,
96       index: GeoMesaFeatureIndex[_, _],
97       expiry: Duration,
98       dtg: String): Unit = {
99     index.getTableNames().foreach { table =>
100       tableOps.attachIterator(table, configure(sft, index, expiry, Option(dtg))) // all scopes
101     }
102   }
103 
104   def clear(tableOps: TableOperations, table: String): Unit = {
105     if (IteratorScope.values.exists(scope => tableOps.getIteratorSetting(table, Name, scope) != null)) {
106       tableOps.removeIterator(table, Name, java.util.EnumSet.allOf(classOf[IteratorScope]))
107     }
108   }
109 }