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.metadata
10 
11 import com.github.benmanes.caffeine.cache.{Cache, CacheLoader, Caffeine, LoadingCache}
12 import com.typesafe.scalalogging.LazyLogging
13 import org.locationtech.geomesa.utils.collection.CloseableIterator
14 import org.locationtech.geomesa.utils.conf.GeoMesaSystemProperties.SystemProperty
15 import org.locationtech.geomesa.utils.io.WithClose
16 import org.locationtech.geomesa.utils.text.DateParsing
17 
18 import java.time.format.DateTimeFormatterBuilder
19 import java.time.temporal.ChronoField
20 import java.time.{Instant, ZoneOffset}
21 import java.util.Locale
22 import java.util.concurrent.TimeUnit
23 import java.util.concurrent.atomic.AtomicBoolean
24 
25 /**
26   * Metadata persisted in a database table. The underlying table will be lazily created when required.
27   * Metadata values are cached with a configurable timeout to save repeated database reads.
28   *
29   * @tparam T type param
30   */
31 trait TableBasedMetadata[T] extends GeoMesaMetadata[T] with LazyLogging {
32 
33   import scala.collection.JavaConverters._
34 
35   /**
36     * Serializer
37     *
38     * @return
39     */
40   def serializer: MetadataSerializer[T]
41 
42   /**
43     * Checks if the underlying table exists
44     *
45     * @return
46     */
47   protected def checkIfTableExists: Boolean
48 
49   /**
50     * Creates the underlying table
51     */
52   protected def createTable(): Unit
53 
54   /**
55     * Create an instance to use for backup
56     *
57     * @param timestamp formatted timestamp for the current time
58     * @return
59     */
60   protected def createEmptyBackup(timestamp: String): TableBasedMetadata[T]
61 
62   /**
63     * Writes key/value pairs
64     *
65     * @param rows keys/values
66     */
67   protected def write(typeName: String, rows: Seq[(String, Array[Byte])]): Unit
68 
69   /**
70     * Deletes multiple rows
71     *
72     * @param typeName simple feature type name
73     * @param keys keys
74     */
75   protected def delete(typeName: String, keys: Seq[String])
76 
77   /**
78     * Reads a value from the underlying table
79     *
80     * @param typeName simple feature type name
81     * @param key key
82     * @return value, if it exists
83     */
84   protected def scanValue(typeName: String, key: String): Option[Array[Byte]]
85 
86   /**
87     * Reads row keys from the underlying table
88     *
89     * @param typeName simple feature type name
90     * @param prefix scan prefix, or empty string for all values
91     * @return matching tuples of (key, value)
92     */
93   protected def scanValues(typeName: String, prefix: String = ""): CloseableIterator[(String, Array[Byte])]
94 
95   /**
96     * Reads all row keys from the underlying table
97     *
98     * @return matching tuples of (typeName, key)
99     */
100   protected def scanKeys(): CloseableIterator[(String, String)]
101 
102   // only synchronize if table doesn't exist - otherwise it's ready only and we can avoid synchronization
103   private val tableExists: AtomicBoolean = new AtomicBoolean(checkIfTableExists)
104 
105   private val expiry = TableBasedMetadata.Expiry.toDuration.get.toMillis
106 
107   // cache for our metadata - invalidate every 10 minutes so we keep things current
108   private val metaDataCache: LoadingCache[(String, String), Option[T]] =
109     Caffeine.newBuilder().expireAfterWrite(expiry, TimeUnit.MILLISECONDS).build(
110       new CacheLoader[(String, String), Option[T]] {
111         override def load(typeNameAndKey: (String, String)): Option[T] = {
112           if (!tableExists.get) { None } else {
113             val (typeName, key) = typeNameAndKey
114             scanValue(typeName, key).map(serializer.deserialize(typeName, _))
115           }
116         }
117       }
118     )
119 
120   // keep a separate cache for scan queries vs point lookups, so that the point lookups don't cache
121   // partial values for a scan result
122   private val metaDataScanCache: LoadingCache[(String, String), Seq[(String, T)]] =
123     Caffeine.newBuilder().expireAfterWrite(expiry, TimeUnit.MILLISECONDS).build(
124       new CacheLoader[(String, String), Seq[(String, T)]] {
125         override def load(typeNameAndPrefix: (String, String)): Seq[(String, T)] = {
126           if (!tableExists.get) { Seq.empty } else {
127             val (typeName, prefix) = typeNameAndPrefix
128             WithClose(scanValues(typeName, prefix)) { iter =>
129               val builder = Seq.newBuilder[(String, T)]
130               iter.foreach { case (k, v) => builder += k -> serializer.deserialize(typeName, v) }
131               builder.result()
132             }
133           }
134         }
135       }
136     )
137 
138   private lazy val formatter =
139     new DateTimeFormatterBuilder()
140       .parseCaseInsensitive()
141       .appendValue(ChronoField.YEAR, 4)
142       .appendValue(ChronoField.MONTH_OF_YEAR, 2)
143       .appendValue(ChronoField.DAY_OF_MONTH, 2)
144       .appendLiteral('T')
145       .appendValue(ChronoField.HOUR_OF_DAY, 2)
146       .appendValue(ChronoField.MINUTE_OF_HOUR, 2)
147       .appendValue(ChronoField.SECOND_OF_MINUTE, 2)
148       .toFormatter(Locale.US)
149       .withZone(ZoneOffset.UTC)
150 
151   override def getFeatureTypes: Array[String] = {
152     if (!tableExists.get) { Array.empty } else {
153       WithClose(scanKeys()) { keys =>
154         keys.collect { case (typeName, key) if key == GeoMesaMetadata.AttributesKey => typeName }.toArray
155       }
156     }
157   }
158 
159   override def read(typeName: String, key: String, cache: Boolean): Option[T] = {
160     if (!cache) {
161       metaDataCache.invalidate((typeName, key))
162     }
163     metaDataCache.get((typeName, key))
164   }
165 
166   override def scan(typeName: String, prefix: String, cache: Boolean): Seq[(String, T)] = {
167     if (!cache) {
168       metaDataScanCache.invalidate((typeName, prefix))
169     }
170     metaDataScanCache.get((typeName, prefix))
171   }
172 
173   override def insert(typeName: String, key: String, value: T): Unit = insert(typeName, Map(key -> value))
174 
175   override def insert(typeName: String, kvPairs: Map[String, T]): Unit = {
176     ensureTableExists()
177     val strings = kvPairs.map { case (k, v) =>
178       metaDataCache.put((typeName, k), Option(v)) // note: side effect in map
179       (k, serializer.serialize(typeName, v))
180     }
181     write(typeName, strings.toSeq)
182     invalidate(metaDataScanCache, typeName)
183   }
184 
185   override def invalidateCache(typeName: String, key: String): Unit = {
186     metaDataCache.invalidate((typeName, key))
187     invalidate(metaDataScanCache, typeName)
188   }
189 
190   override def remove(typeName: String, key: String): Unit = remove(typeName, Seq(key))
191 
192   override def remove(typeName: String, keys: Seq[String]): Unit = {
193     if (tableExists.get) {
194       delete(typeName, keys)
195       // also remove from the cache
196       keys.foreach(k => metaDataCache.invalidate((typeName, k)))
197       invalidate(metaDataScanCache, typeName)
198     } else {
199       logger.debug(s"Trying to delete '$typeName: ${keys.mkString(", ")}' but table does not exist")
200     }
201   }
202 
203   override def delete(typeName: String): Unit = {
204     if (tableExists.get) {
205       WithClose(scanValues(typeName)) { rows =>
206         if (rows.nonEmpty) {
207           delete(typeName, rows.map(_._1).toSeq)
208         }
209       }
210     } else {
211       logger.debug(s"Trying to delete type '$typeName' but table does not exist")
212     }
213     Seq(metaDataCache, metaDataScanCache).foreach(invalidate(_, typeName))
214   }
215 
216   override def backup(typeName: String): Unit = {
217     if (tableExists.get) {
218       WithClose(scanValues(typeName)) { rows =>
219         if (rows.nonEmpty) {
220           WithClose(createEmptyBackup(DateParsing.formatInstant(Instant.now, formatter))) { metadata =>
221             metadata.ensureTableExists()
222             metadata.write(typeName, rows.toSeq)
223           }
224         }
225       }
226     } else {
227       logger.debug(s"Trying to back up type '$typeName' but table does not exist")
228     }
229   }
230 
231   // checks that the table is already created, and creates it if not
232   def ensureTableExists(): Unit = {
233     if (tableExists.compareAndSet(false, true)) {
234       createTable()
235     }
236   }
237 
238   override def resetCache(): Unit={
239     tableExists.set(checkIfTableExists)
240     metaDataCache.invalidateAll()
241     metaDataScanCache.invalidateAll()
242   }
243 
244   /**
245     * Invalidate all keys for the given feature type
246     *
247     * @param cache cache to invalidate
248     * @param typeName feature type name
249     */
250   private def invalidate(cache: Cache[(String, String), _], typeName: String): Unit = {
251     cache.asMap.asScala.keys.foreach { k =>
252       if (k._1 == typeName) {
253         cache.invalidate(k)
254       }
255     }
256   }
257 }
258 
259 object TableBasedMetadata {
260   val Expiry = SystemProperty("geomesa.metadata.expiry", "10 minutes")
261 }