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.fs.storage.common
10 
11 import com.typesafe.config.{Config, ConfigRenderOptions, ConfigValue, ConfigValueFactory}
12 import org.geotools.api.feature.simple.SimpleFeatureType
13 import org.locationtech.geomesa.fs.storage.api.StorageMetadata.{PartitionBounds, PartitionMetadata, StorageFile, StorageFileAction}
14 import org.locationtech.geomesa.fs.storage.common.metadata.PartitionAction.PartitionAction
15 import org.locationtech.geomesa.utils.geotools.SimpleFeatureTypes
16 import org.locationtech.jts.geom.Envelope
17 import pureconfig.ConfigReader.Result
18 import pureconfig._
19 import pureconfig.error.{CannotConvert, ConfigReaderFailures}
20 
21 import java.util.Collections
22 import scala.reflect.ClassTag
23 
24 package object metadata {
25 
26   import scala.collection.JavaConverters._
27 
28   /**
29     * Creates a new simple feature type with the namespace in the simple feature type name
30     *
31     * @param sft simple feature type
32     * @param namespace optional namespace
33     * @return
34     */
35   def namespaced(sft: SimpleFeatureType, namespace: Option[String]): SimpleFeatureType =
36     namespace.map(ns => SimpleFeatureTypes.renameSft(sft, s"$ns:${sft.getTypeName}")).getOrElse(sft)
37 
38   /**
39     * Merge configs for a single partition into a single aggregate config
40     *
41     * @param configs updates for a given partition
42     * @return
43     */
44   def mergePartitionConfigs(configs: Seq[PartitionConfig]): Option[PartitionConfig] = {
45     configs.sortBy(_.timestamp).dropWhile(_.action != PartitionAction.Add).reduceLeftOption { (result, update) =>
46       update.action match {
47         case PartitionAction.Add    => result + update
48         case PartitionAction.Remove => result - update
49       }
50     }
51   }
52 
53   // case classes for serializing to disk
54 
55   case class CompactedConfig(partitions: Seq[PartitionConfig])
56 
57   case class PartitionConfig(
58       name: String,
59       action: PartitionAction,
60       files: Seq[StorageFile],
61       count: Long,
62       envelope: Seq[Double],
63       timestamp: Long) {
64 
65     def +(other: PartitionConfig): PartitionConfig = {
66       require(action == PartitionAction.Add, "Can't aggregate into non-add actions")
67       val ts = math.max(timestamp, other.timestamp)
68       val env = if (envelope.isEmpty) { other.envelope } else if (other.envelope.isEmpty) { envelope } else {
69         val Seq(xmin1, ymin1, xmax1, ymax1) = envelope
70         val Seq(xmin2, ymin2, xmax2, ymax2) = other.envelope
71         Seq(math.min(xmin1, xmin2), math.min(ymin1, ymin2), math.max(xmax1, xmax2), math.max(ymax1, ymax2))
72       }
73       PartitionConfig(name, action, files ++ other.files, count + other.count, env, ts)
74     }
75 
76     def -(other: PartitionConfig): PartitionConfig = {
77       require(action == PartitionAction.Add, "Can't aggregate into non-add actions")
78       val ts = math.max(timestamp, other.timestamp)
79       val names = scala.collection.mutable.Map.empty[String, Int].withDefault(_ => 0)
80       other.files.foreach(f => names(f.name) += 1)
81       // keep oldest files if there are duplicates
82       val fs = files.sortBy(_.timestamp)(Ordering.Long.reverse).filter { f =>
83         names(f.name) match {
84           case 0 => true
85           case i => names.put(f.name, i - 1); false
86         }
87       }
88       PartitionConfig(name, action, fs, math.max(0, count - other.count), envelope, ts)
89     }
90 
91     def toMetadata: PartitionMetadata = {
92       val bounds = envelope match {
93         case Nil => None
94         case Seq(xmin, ymin, xmax, ymax) => Some(PartitionBounds(xmin, ymin, xmax, ymax))
95         case _ => throw new IllegalStateException(s"Invalid envelope: $envelope")
96       }
97       PartitionMetadata(name, files, bounds, count)
98     }
99   }
100 
101   object PartitionAction extends Enumeration {
102     type PartitionAction = Value
103     val Add, Remove, Clear = Value
104   }
105 
106   object EnvelopeConfig {
107     def apply(env: Envelope): Seq[Double] =
108       if (env == null || env.isNull) { Seq.empty } else { Seq(env.getMinX, env.getMinY, env.getMaxX, env.getMaxY) }
109   }
110 
111   // pureconfig converters for our case classes
112 
113   // noinspection ForwardReference
114   sealed trait MetadataConverter {
115 
116     def name: String
117     def suffix: String
118 
119     protected def partitionWriter: ConfigWriter[PartitionConfig]
120     protected def compactedWriter: ConfigWriter[CompactedConfig]
121     protected def options: ConfigRenderOptions
122 
123     private val partitionCt = ClassTag[PartitionConfig](classOf[PartitionConfig])
124     private val compactedCt = ClassTag[CompactedConfig](classOf[CompactedConfig])
125 
126     def renderPartition(partition: PartitionConfig): String =
127       partitionWriter.to(partition).render(options)
128     def parsePartition(config: Config): PartitionConfig =
129       ConfigSource.fromConfig(config).loadOrThrow[PartitionConfig](partitionCt, PartitionReader)
130 
131     def renderCompaction(compaction: Seq[PartitionConfig]): String =
132       compactedWriter.to(CompactedConfig(compaction)).render(options)
133     def parseCompaction(config: Config): Seq[PartitionConfig] =
134       ConfigSource.fromConfig(config).loadOrThrow[CompactedConfig](compactedCt, CompactedReader).partitions
135   }
136 
137   object MetadataConverter {
138 
139     private val options = Seq(RenderCompact, RenderPretty)
140 
141     def apply(name: String): MetadataConverter = {
142       options.find(_.name.equalsIgnoreCase(name)).getOrElse {
143         throw new IllegalArgumentException(
144           s"Render type '$name' does not exist. Available types: ${options.map(_.name).mkString(", ")}")
145       }
146     }
147   }
148 
149   object RenderPretty extends MetadataConverter {
150     override val name: String = FileBasedMetadata.Config.RenderPretty
151     override val suffix: String = ".json"
152     override protected val partitionWriter: ConfigWriter[PartitionConfig] =
153       ConfigWriter.fromFunction(writePartitionConfigVerbose)
154     override protected val compactedWriter: ConfigWriter[CompactedConfig] =
155       ConfigWriter.fromFunction(writeCompactedConfig(partitionWriter))
156     override protected val options: ConfigRenderOptions = ConfigRenderOptions.concise().setFormatted(true)
157   }
158 
159   object RenderCompact extends MetadataConverter {
160     override val name: String = FileBasedMetadata.Config.RenderCompact
161     override val suffix: String = ".conf"
162     override protected val partitionWriter: ConfigWriter[PartitionConfig] =
163       ConfigWriter.fromFunction(writePartitionConfigCompact)
164     override protected val compactedWriter: ConfigWriter[CompactedConfig] =
165       ConfigWriter.fromFunction(writeCompactedConfig(partitionWriter))
166     override protected val options: ConfigRenderOptions = ConfigRenderOptions.concise().setJson(false)
167   }
168 
169   class EnumerationReader[T <: Enumeration](enum: T) extends ConfigReader[T#Value] {
170     override def from(cur: ConfigCursor): Result[T#Value] = {
171       cur.asString.right.flatMap { s =>
172         lazy val reason =
173           CannotConvert(s, enum.getClass.getName, s"value $s is not a valid enum: ${enum.values.mkString(", ")}")
174         enum.values.find(_.toString.startsWith(s)).asInstanceOf[Option[T#Value]].toRight(cur.failed(reason).left.get)
175       }
176     }
177   }
178 
179   private val PartitionReader = ConfigReader.fromCursor(readPartitionConfig)
180   private val CompactedReader = ConfigReader.fromCursor(readCompactedConfig)
181   private val StorageFileCompactReader = ConfigReader.fromCursor(readStorageFileCompact)
182   private val StorageFileVerboseReader = ConfigReader.fromCursor(readStorageFileVerbose)
183   private val BoundReader = ConfigReader.fromCursor(readBound)
184   private val PartitionActionReader = new EnumerationReader(PartitionAction)
185   private val StorageFileActionReader = new EnumerationReader(StorageFileAction)
186 
187   private def writeCompactedConfig(delegate: ConfigWriter[PartitionConfig])(compacted: CompactedConfig): ConfigValue =
188     ConfigValueFactory.fromMap(Collections.singletonMap("partitions", compacted.partitions.map(delegate.to).asJava))
189 
190   private def readCompactedConfig(cur: ConfigCursor): Either[ConfigReaderFailures, CompactedConfig] = {
191     for {
192       obj  <- cur.asObjectCursor.right
193       par  <- obj.atKey("partitions").right.flatMap(_.asListCursor).right
194       list <- convertList[PartitionConfig](par)(PartitionReader).right
195     } yield {
196       CompactedConfig(list)
197     }
198   }
199 
200   private def writePartitionConfigCompact(partition: PartitionConfig): ConfigValue = {
201     val list = new java.util.ArrayList[AnyRef](5)
202     list.add(partition.action.toString.take(1))
203     list.add(Long.box(partition.timestamp))
204     list.add(partition.envelope.asJava)
205     list.add(Long.box(partition.count))
206     list.add(partition.files.map(writeStorageFileCompact).asJava)
207     ConfigValueFactory.fromMap(Collections.singletonMap(partition.name, list))
208   }
209 
210   private def writePartitionConfigVerbose(partition: PartitionConfig): ConfigValue = {
211     val map = new java.util.HashMap[String, AnyRef](6)
212     map.put("name", partition.name)
213     map.put("action", partition.action.toString)
214     map.put("files", partition.files.map(writeStorageFileVerbose).asJava)
215     map.put("count", Long.box(partition.count))
216     map.put("timestamp", Long.box(partition.timestamp))
217     if (partition.envelope.nonEmpty) {
218       val envelope = new java.util.HashMap[String, AnyRef](4)
219       envelope.put("xmin", Double.box(partition.envelope.head))
220       envelope.put("ymin", Double.box(partition.envelope(1)))
221       envelope.put("xmax", Double.box(partition.envelope(2)))
222       envelope.put("ymax", Double.box(partition.envelope(3)))
223       map.put("envelope", envelope)
224     }
225     ConfigValueFactory.fromMap(map)
226   }
227 
228   private def readPartitionConfig(cur: ConfigCursor): Either[ConfigReaderFailures, PartitionConfig] = {
229     val compact = readPartitionConfigCompact(cur)
230     if (compact.isRight) { compact } else { readPartitionConfigVerbose(cur).left.flatMap(_ => compact) }
231   }
232 
233   private def readPartitionConfigCompact(cur: ConfigCursor): Either[ConfigReaderFailures, PartitionConfig] = {
234     cur.asObjectCursor.right.flatMap { obj =>
235       val name = obj.keys.head
236       obj.atKeyOrUndefined(name).asListCursor.right.flatMap { list =>
237         if (list.size != 5) {
238           cur.failed(CannotConvert(cur.valueOpt.map(_.toString).getOrElse(""),
239             classOf[PartitionConfig].getSimpleName, s"value ${cur.valueOpt.orNull} does not have the expected 6 elements"))
240         } else {
241           for {
242             action <- PartitionActionReader.from(list.atIndexOrUndefined(0)).right
243             ts     <- list.atIndexOrUndefined(1).asLong.right
244             env    <- list.atIndexOrUndefined(2).asListCursor.right.flatMap(readEnvelope).right
245             count  <- list.atIndexOrUndefined(3).asLong.right
246             files  <- list.atIndexOrUndefined(4).asListCursor.right.flatMap(convertList(_)(StorageFileCompactReader)).right
247           } yield {
248             PartitionConfig(name, action, files, count, env, ts)
249           }
250         }
251       }
252     }
253   }
254 
255   private def readPartitionConfigVerbose(cur: ConfigCursor): Either[ConfigReaderFailures, PartitionConfig] = {
256     def readEnvelope(obj: ConfigObjectCursor): Either[ConfigReaderFailures, Seq[Double]] = {
257       obj.atKeyOrUndefined("envelope") match {
258         case envelope if envelope.isUndefined => Right(Seq.empty)
259         case envelope =>
260           for {
261             env  <- envelope.asObjectCursor.right
262             xmin <- env.atKey("xmin").right.flatMap(_.asDouble).right
263             ymin <- env.atKey("ymin").right.flatMap(_.asDouble).right
264             xmax <- env.atKey("xmax").right.flatMap(_.asDouble).right
265             ymax <- env.atKey("ymax").right.flatMap(_.asDouble).right
266           } yield {
267             Seq(xmin, ymin, xmax, ymax)
268           }
269       }
270     }
271 
272     for {
273       obj    <- cur.asObjectCursor.right
274       name   <- obj.atKey("name").right.flatMap(_.asString).right
275       action <- obj.atKey("action").right.flatMap(PartitionActionReader.from).right
276       files  <- obj.atKey("files").right.flatMap(_.asListCursor).right.flatMap(convertList(_)(StorageFileVerboseReader)).right
277       count  <- obj.atKey("count").right.flatMap(ConfigReader.longConfigReader.from).right
278       bounds <- readEnvelope(obj).right
279       ts     <- obj.atKey("timestamp").right.flatMap(ConfigReader.longConfigReader.from).right
280     } yield {
281       PartitionConfig(name, action, files, count, bounds, ts)
282     }
283   }
284 
285   private def writeStorageFileCompact(file: StorageFile): ConfigValue = {
286     val list = new java.util.ArrayList[AnyRef](5)
287     list.add(file.name)
288     list.add(file.action.toString.take(1))
289     list.add(Long.box(file.timestamp))
290     if (file.sort.nonEmpty) {
291       list.add(file.sort.asJava)
292     }
293     if (file.bounds.nonEmpty) {
294       list.add(file.bounds.map { case (a, f, t) => Seq(a, f, t).asJava }.asJava)
295     }
296     ConfigValueFactory.fromIterable(list)
297   }
298 
299   private def writeStorageFileVerbose(file: StorageFile): ConfigValue = {
300     val map = new java.util.HashMap[String, AnyRef](5)
301     map.put("name", file.name)
302     map.put("timestamp", Long.box(file.timestamp))
303     map.put("action", file.action.toString)
304     if (file.sort.nonEmpty) {
305       map.put("sort", file.sort.asJava)
306     }
307     if (file.bounds.nonEmpty) {
308       map.put("bounds", file.bounds.map { case (a, f, t) => Seq(a, f, t).asJava }.asJava)
309     }
310     ConfigValueFactory.fromMap(map)
311   }
312 
313   private def readStorageFileCompact(cur: ConfigCursor): Either[ConfigReaderFailures, StorageFile] = {
314     cur.asListCursor.right.flatMap { list =>
315       if (list.size < 3 || list.size > 5) {
316         cur.failed(CannotConvert(cur.valueOpt.map(_.toString).getOrElse(""),
317           classOf[StorageFile].getSimpleName, s"value ${cur.valueOpt.orNull} does not have the expected number of elements"))
318       } else {
319         for {
320           name   <- list.atIndexOrUndefined(0).asString.right
321           action <- StorageFileActionReader.from(list.atIndexOrUndefined(1)).right
322           ts     <- list.atIndexOrUndefined(2).asLong.right
323           sort   <- readSort(list.atIndexOrUndefined(3)).right
324           bounds <- readBounds(list.atIndexOrUndefined(4)).right
325         } yield {
326           StorageFile(name, ts, action, sort, bounds)
327         }
328       }
329     }
330   }
331 
332   private def readStorageFileVerbose(cur: ConfigCursor): Either[ConfigReaderFailures, StorageFile] = {
333     val withAction = for {
334       obj    <- cur.asObjectCursor.right
335       name   <- obj.atKey("name").right.flatMap(_.asString).right
336       ts     <- obj.atKey("timestamp").right.flatMap(ConfigReader.longConfigReader.from).right
337       action <- obj.atKey("action").right.flatMap(StorageFileActionReader.from).right
338       sort   <- readSort(obj.atKeyOrUndefined("sort")).right
339       bounds <- readBounds(obj.atKeyOrUndefined("bounds")).right
340     } yield {
341       StorageFile(name, ts, action, sort, bounds)
342     }
343     if (withAction.isRight) { withAction } else {
344       // note: use 0 for timestamp to sort before any mods
345       val sansAction = for { name <- cur.asString.right } yield { StorageFile(name, 0L) }
346       sansAction.left.flatMap(_ => withAction) // if failure, replace with original error
347     }
348   }
349 
350   private def readEnvelope(env: ConfigListCursor): Either[ConfigReaderFailures, Seq[Double]] = {
351     convertList[Double](env).right.flatMap { coords =>
352       if (coords.isEmpty || coords.length == 4) { Right(coords) } else {
353         env.failed(CannotConvert(env.valueOpt.map(_.toString).getOrElse(""),
354           "Seq[Double]", s"value ${env.valueOpt.orNull} does not have the expected 4 elements"))
355       }
356     }
357   }
358 
359   private def readSort(cur: ConfigCursor): Either[ConfigReaderFailures, Seq[Int]] =
360     if (cur.isUndefined) { Right(Seq.empty) } else { cur.asListCursor.right.flatMap(convertList[Int]) }
361 
362   private def readBounds(cur: ConfigCursor): Either[ConfigReaderFailures, Seq[(Int, String, String)]] =
363     if (cur.isUndefined) { Right(Seq.empty) } else { cur.asListCursor.right.flatMap(convertList(_)(BoundReader)) }
364 
365   private def readBound(cur: ConfigCursor): Either[ConfigReaderFailures, (Int, String, String)] = {
366     cur.asListCursor.right.flatMap { list =>
367       if (list.size != 3) {
368         cur.failed(CannotConvert(cur.valueOpt.map(_.toString).getOrElse(""),
369           "Tuple3[Int, String, String]", s"value ${cur.valueOpt.orNull} does not have 3 elements"))
370       } else {
371         for {
372           attribute <- list.atIndexOrUndefined(0).asInt.right
373           from      <- list.atIndexOrUndefined(1).asString.right
374           to        <- list.atIndexOrUndefined(2).asString.right
375         } yield {
376           (attribute, from, to)
377         }
378       }
379     }
380   }
381 
382   private def convertList[T](
383       list: ConfigListCursor)
384      (implicit reader: ConfigReader[T]): Either[ConfigReaderFailures, Seq[T]] = {
385     val res = Seq.newBuilder[T]
386     var err: ConfigReaderFailures = null
387     var i = 0
388     while (i < list.size) {
389       reader.from(list.atIndexOrUndefined(i)) match {
390         case Right(success) => res += success
391         case Left(failure)  => if (err == null) { err = failure } else { err = err ++ failure }
392       }
393       i += 1
394     }
395     if (err == null) { Right(res.result) } else { Left(err) }
396   }
397 }