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.convert2
10 
11 import com.codahale.metrics.Counter
12 import com.typesafe.scalalogging.LazyLogging
13 import org.geotools.api.feature.simple.{SimpleFeature, SimpleFeatureType}
14 import org.locationtech.geomesa.convert.EvaluationContext.{EvaluationError, FieldAccessor}
15 import org.locationtech.geomesa.convert.Modes.ErrorMode
16 import org.locationtech.geomesa.convert.{EnrichmentCache, EvaluationContext}
17 import org.locationtech.geomesa.convert2.AbstractCompositeConverter.{CompositeEvaluationContext, PredicateContext}
18 import org.locationtech.geomesa.convert2.metrics.ConverterMetrics
19 import org.locationtech.geomesa.convert2.transforms.Predicate
20 import org.locationtech.geomesa.utils.collection.CloseableIterator
21 import org.locationtech.geomesa.utils.io.CloseWithLogging
22 
23 import java.io.InputStream
24 import scala.util.control.NonFatal
25 
26 abstract class AbstractCompositeConverter[T <: AnyRef](
27     sft: SimpleFeatureType,
28     errorMode: ErrorMode,
29     delegates: Seq[(Predicate, ParsingConverter[T])]
30   ) extends SimpleFeatureConverter with LazyLogging {
31 
32   protected def parse(is: InputStream, ec: EvaluationContext): CloseableIterator[T]
33 
34   override def targetSft: SimpleFeatureType = sft
35 
36   override def createEvaluationContext(globalParams: Map[String, Any]): EvaluationContext =
37     CompositeEvaluationContext(delegates.map(_._2), globalParams)
38 
39   override def createEvaluationContext(
40       globalParams: Map[String, Any],
41       success: Counter,
42       failure: Counter): EvaluationContext = {
43     CompositeEvaluationContext(delegates.map(_._2), globalParams, success, failure)
44   }
45 
46   override def process(is: InputStream, ec: EvaluationContext): CloseableIterator[SimpleFeature] = {
47     val contexts = ec match {
48       case cec: CompositeEvaluationContext => cec.contexts.iterator
49       case _ => Iterator.continually(ec)
50     }
51     val predicates = delegates.map { case (predicate, converter) =>
52       if (!contexts.hasNext) {
53         throw new IllegalArgumentException(s"Invalid evaluation context doesn't match this converter: $ec")
54       }
55       val context = contexts.next()
56       PredicateContext(predicate.withContext(context), converter, context)
57     }
58 
59     val args = Array.ofDim[AnyRef](1)
60 
61     def matches(p: Predicate): Boolean = {
62       try { p.apply(args) } catch {
63         case NonFatal(e) => logger.warn(s"Error evaluating predicate $p: ", e); false
64       }
65     }
66 
67     val routing = predicates.head.context.metrics.histogram("predicate.eval.nanos")
68 
69     def eval(element: T): CloseableIterator[SimpleFeature] = {
70       val start = System.nanoTime()
71       args(0) = element
72       val converted = predicates.collectFirst { case p if matches(p.predicate) =>
73         routing.update(System.nanoTime() - start) // note: invoke before executing the conversion
74         p.context.line = ec.line // update the line in the sub context
75         p.converter.convert(CloseableIterator.single(element), p.context)
76       }
77       converted.getOrElse {
78         routing.update(System.nanoTime() - start)
79         ec.failure.inc()
80         CloseableIterator.empty
81       }
82     }
83 
84     val hist = predicates.head.context.metrics.histogram("parse.nanos")
85     new ErrorHandlingIterator(parse(is, ec), errorMode, ec, hist).flatMap(eval)
86   }
87 
88   override def close(): Unit = CloseWithLogging(delegates.map(_._2))
89 }
90 
91 object AbstractCompositeConverter {
92 
93   case class CompositeEvaluationContext(contexts: Seq[EvaluationContext], success: Counter, failure: Counter)
94       extends EvaluationContext {
95     override def cache: Map[String, EnrichmentCache] = throw new NotImplementedError()
96     override def metrics: ConverterMetrics = throw new NotImplementedError()
97     override def accessor(name: String): FieldAccessor = throw new NotImplementedError()
98     override def evaluate(args: Array[AnyRef]): Either[EvaluationError, Array[AnyRef]] =
99       throw new NotImplementedError()
100     override def errors: java.util.Queue[EvaluationError] = throw new NotImplementedError()
101   }
102 
103   object CompositeEvaluationContext {
104 
105     def apply(
106         converters: Seq[_ <: SimpleFeatureConverter],
107         globalParams: Map[String, Any]): CompositeEvaluationContext = {
108       // create a single set of shared success/failure counters
109       val head = converters.head.createEvaluationContext(globalParams)
110       val tail = converters.tail.map(_.createEvaluationContext(globalParams, head.success, head.failure))
111       CompositeEvaluationContext(head +: tail, head.success, head.failure)
112     }
113 
114     def apply(
115         converters: Seq[_ <: SimpleFeatureConverter],
116         globalParams: Map[String, Any],
117         success: Counter,
118         failure: Counter): CompositeEvaluationContext = {
119       val contexts = converters.map(_.createEvaluationContext(globalParams, success, failure))
120       CompositeEvaluationContext(contexts, success, failure)
121     }
122   }
123 
124   case class PredicateContext[T <: SimpleFeatureConverter](
125       predicate: Predicate,
126       converter: T,
127       context: EvaluationContext
128     )
129 }