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.convert
10 
11 import com.codahale.metrics.Counter
12 import com.typesafe.scalalogging.LazyLogging
13 import org.locationtech.geomesa.convert.EvaluationContext.{EvaluationError, FieldAccessor}
14 import org.locationtech.geomesa.convert2.Field
15 import org.locationtech.geomesa.convert2.metrics.ConverterMetrics
16 
17 import scala.util.control.NonFatal
18 
19 /**
20  * Holds the state associated with a conversion attempt. Evaluation contexts are not thread safe.
21  */
22 trait EvaluationContext {
23 
24   /**
25     * The current line being processed.
26     *
27     * This may be an actual line (e.g. a csv row), or a logical line (e.g. an avro record)
28     *
29     * @return
30     */
31   var line: Long = 0
32 
33   /**
34     * Enrichment caches
35     *
36     * @return
37     */
38   def cache: Map[String, EnrichmentCache]
39 
40   /**
41     * Metrics registry, accessible for tracking any custom values
42     *
43     * @return
44     */
45   def metrics: ConverterMetrics
46 
47   /**
48     * Counter for tracking successes
49     *
50     * @return
51     */
52   def success: com.codahale.metrics.Counter
53 
54   /**
55     * Counter for tracking failures
56     *
57     * @return
58     */
59   def failure: com.codahale.metrics.Counter
60 
61   /**
62    * Access to any errors that have occurred - note that errors will generally only be kept if the converter
63    * error mode is set to `ReturnErrors`
64    *
65    * @return
66    */
67   def errors: java.util.Queue[EvaluationError]
68 
69   /**
70    * Gets a references to a field's value
71    *
72    * @param name field name
73    * @return
74    */
75   def accessor(name: String): FieldAccessor
76 
77   /**
78    * Evaluate all values using the given arguments. The returned array may be mutated on subsequent calls to
79    * `evaluate`, so shouldn't be kept long-term
80    *
81    * @param args single row of input
82    */
83   def evaluate(args: Array[AnyRef]): Either[EvaluationError, Array[AnyRef]]
84 }
85 
86 object EvaluationContext extends LazyLogging {
87 
88   val InputFilePathKey = "inputFilePath"
89 
90   /**
91     * Creates a new, empty evaluation context
92     *
93     * @return
94     */
95   def empty: EvaluationContext = {
96     val metrics = ConverterMetrics.empty
97     val success = metrics.counter("success")
98     val failures = metrics.counter("failure")
99     new StatefulEvaluationContext(Array.empty, Map.empty, Map.empty, metrics, success, failures)
100   }
101 
102   /**
103    * Creates a new evaluation context with the given state
104    *
105    * @param fields converter fields, in topological dependency order
106    * @param globalValues global values
107    * @param caches enrichment caches
108    * @param metrics metrics
109    * @param success success counter
110    * @param failure failure counter
111    * @return
112    */
113   def apply(
114       fields: Seq[Field],
115       globalValues: Map[String, _ <: AnyRef],
116       caches: Map[String, EnrichmentCache],
117       metrics: ConverterMetrics,
118       success: Counter,
119       failure: Counter): EvaluationContext = {
120     new StatefulEvaluationContext(fields.toArray, globalValues, caches, metrics, success, failure)
121   }
122 
123   /**
124     * Gets a global parameter map containing the input file path
125     *
126     * @param file input file path
127     * @return
128     */
129   def inputFileParam(file: String): Map[String, AnyRef] = Map(InputFilePathKey -> file)
130 
131   /**
132    * Trait for reading a field from an evaluation context
133    */
134   sealed trait FieldAccessor {
135     def apply(): AnyRef
136   }
137 
138   case object NullFieldAccessor extends FieldAccessor {
139     override def apply(): AnyRef = null
140   }
141 
142   class FieldValueAccessor(values: Array[AnyRef], i: Int) extends FieldAccessor {
143     override def apply(): AnyRef = values(i)
144   }
145 
146   class GlobalFieldAccessor(value: AnyRef) extends FieldAccessor {
147     override def apply(): AnyRef = value
148   }
149 
150   /**
151    * Marker trait for resources that are dependent on the evaluation context state
152    *
153    * @tparam T type
154    */
155   trait ContextDependent[T <: ContextDependent[T]] {
156 
157     /**
158      * Return a copy of the instance tied to the given evaluation context.
159      *
160      * If the instance does not use the evaluation context, it should return itself instead of a copy
161      *
162      * @param ec evaluation context
163      * @return
164      */
165     def withContext(ec: EvaluationContext): T
166   }
167 
168   /**
169    * Evaluation error
170    *
171    * @param field field name that had an error
172    * @param line line number of the input being evaluated
173    * @param e error
174    */
175   case class EvaluationError(field: String, line: Long, e: Throwable)
176 
177   /**
178     * Evaluation context accessors
179     *
180     * @param ec context
181     */
182   implicit class RichEvaluationContext(val ec: EvaluationContext) extends AnyVal {
183     def getInputFilePath: Option[String] = Option(ec.accessor(InputFilePathKey).apply().asInstanceOf[String])
184   }
185 
186   /**
187     * Evaluation context implementation
188     *
189     * @param fields fields to evaluate, in topological dependency order
190     * @param globalValues global variable name/values
191     * @param cache enrichment caches
192     * @param metrics metrics
193     */
194   class StatefulEvaluationContext(
195       fields: Array[Field],
196       globalValues: Map[String, _ <: AnyRef],
197       val cache: Map[String, EnrichmentCache],
198       val metrics: ConverterMetrics,
199       val success: Counter,
200       val failure: Counter,
201       val errors: java.util.Queue[EvaluationError] = new java.util.ArrayDeque[EvaluationError]()
202     ) extends EvaluationContext {
203 
204     // holder for results from evaluating each row
205     private val values = Array.ofDim[AnyRef](fields.length)
206     // temp array for holding the arguments for a field
207     private val fieldArray = Array.ofDim[AnyRef](1)
208     // copy the transforms and tie them to the context
209     // note: the class isn't fully instantiated yet, but this statement is last in the initialization
210     private val transforms = fields.map(_.transforms.map(_.withContext(this)))
211 
212     override def accessor(name: String): FieldAccessor = {
213       val i = fields.indexWhere(_.name == name)
214       if (i >= 0) { new FieldValueAccessor(values, i) } else {
215         globalValues.get(name) match {
216           case Some(value) => new GlobalFieldAccessor(value)
217           case None => NullFieldAccessor
218         }
219       }
220     }
221 
222     override def evaluate(args: Array[AnyRef]): Either[EvaluationError, Array[AnyRef]] = {
223       var i = 0
224       // note: since fields are in topological order we don't need to clear them
225       while (i < values.length) {
226         values(i) = try {
227           val fieldArgs = fields(i).fieldArg match {
228             case None => args
229             case Some(f) => fieldArray(0) = f.apply(args); fieldArray
230           }
231           transforms(i) match {
232             case Some(t) => t.apply(fieldArgs)
233             case None    => fieldArgs(0)
234           }
235         } catch {
236           case NonFatal(e) => return Left(EvaluationError(fields(i).name, line, e))
237         }
238         i += 1
239       }
240       Right(values)
241     }
242   }
243 }