1 /***********************************************************************
2  * Copyright (c) 2013-2025 General Atomics Integrated Intelligence, 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  * https://www.apache.org/licenses/LICENSE-2.0
7  ***********************************************************************/
8 
9 package org.locationtech.geomesa.kafka.index
10 
11 import com.typesafe.scalalogging.LazyLogging
12 import io.micrometer.core.instrument.Tags
13 import org.geotools.api.data.{FeatureListener, SimpleFeatureSource}
14 import org.geotools.api.feature.simple.{SimpleFeature, SimpleFeatureType}
15 import org.geotools.api.filter.Filter
16 import org.geotools.api.filter.expression.Expression
17 import org.locationtech.geomesa.filter.factory.FastFilterFactory
18 import org.locationtech.geomesa.kafka.data.KafkaDataStore._
19 
20 import java.io.Closeable
21 import java.util.concurrent._
22 
23 trait KafkaFeatureCache extends KafkaListeners with Closeable {
24   def put(feature: SimpleFeature): Unit
25   def remove(id: String): Unit
26   def clear(): Unit
27   def size(): Int
28   def size(filter: Filter): Int
29   def query(id: String): Option[SimpleFeature]
30   def query(filter: Filter): Iterator[SimpleFeature]
31   def views: Seq[KafkaFeatureCacheView]
32 }
33 
34 object KafkaFeatureCache extends LazyLogging {
35 
36   /**
37    * Create a standard feature cache
38    *
39    * @param sft simple feature type
40    * @param config cache config
41    * @param views layer view config
42    * @return
43    */
44   def apply(sft: SimpleFeatureType, config: IndexConfig, views: Seq[LayerView] = Seq.empty, tags: Tags = Tags.empty()): KafkaFeatureCache = {
45     if (config.expiry == ImmediatelyExpireConfig) {
46       new NoOpFeatureCache(views.map(v => KafkaFeatureCacheView.empty(v.viewSft)))
47     } else {
48       new KafkaFeatureCacheImpl(sft, config, views, tags)
49     }
50   }
51 
52   /**
53     * No-op cache
54     *
55     * @return
56     */
57   def empty(views: Seq[LayerView] = Seq.empty): KafkaFeatureCache =
58     new EmptyFeatureCache(views.map(v => KafkaFeatureCacheView.empty(v.viewSft)))
59 
60   /**
61     * Cache that won't spatially index the features
62     *
63     * @param sft simple feature type
64     * @param ordering feature ordering
65     * @return
66     */
67   def nonIndexing(sft: SimpleFeatureType, ordering: ExpiryTimeConfig = NeverExpireConfig): KafkaFeatureCache = {
68     val event: PartialFunction[ExpiryTimeConfig, String] = {
69       case EventTimeConfig(_, exp, true) => exp
70     }
71 
72     val ord = ordering match {
73       case o if event.isDefinedAt(o) => Some(event.apply(o))
74       // all filters use the same event time ordering
75       case FilteredExpiryConfig(filters) if event.isDefinedAt(filters.head._2) => Some(event.apply(filters.head._2))
76       case _ => None
77     }
78 
79     ord.map(FastFilterFactory.toExpression(sft, _)) match {
80       case None => new NonIndexingFeatureCache()
81       case Some(exp) => new NonIndexingEventTimeFeatureCache(exp)
82     }
83   }
84 
85   /**
86     * Non-indexing feature cache that just tracks the most recent feature
87     */
88   private class NonIndexingFeatureCache extends KafkaFeatureCache {
89 
90     private val state = new ConcurrentHashMap[String, SimpleFeature]
91 
92     override def put(feature: SimpleFeature): Unit = state.put(feature.getID, feature)
93 
94     override def remove(id: String): Unit = state.remove(id)
95 
96     override def clear(): Unit = state.clear()
97 
98     override def close(): Unit = {}
99 
100     override def size(): Int = state.size()
101 
102     override def size(filter: Filter): Int = query(filter).length
103 
104     override def query(id: String): Option[SimpleFeature] = Option(state.get(id))
105 
106     override def query(filter: Filter): Iterator[SimpleFeature] = {
107       import scala.collection.JavaConverters._
108       val features = state.asScala.valuesIterator
109       if (filter == Filter.INCLUDE) { features } else {
110         features.filter(filter.evaluate)
111       }
112     }
113 
114     override def views: Seq[KafkaFeatureCacheView] = Seq.empty
115   }
116 
117   /**
118     * Non-indexing feature cache that just tracks the most recent feature, based on event time
119     *
120     * @param time event time expression
121     */
122   private class NonIndexingEventTimeFeatureCache(time: Expression) extends KafkaFeatureCache {
123 
124     private val state = new ConcurrentHashMap[String, (SimpleFeature, Long)]
125 
126     /**
127       * Note: this method is not thread-safe. The `state` and can fail to replace the correct values
128       * if the same feature is updated simultaneously from two different threads
129       *
130       * In our usage, this isn't a problem, as a given feature ID is always operated on by a single thread
131       * due to kafka consumer partitioning
132       */
133     override def put(feature: SimpleFeature): Unit = {
134       val tuple = (feature, FeatureStateFactory.time(time, feature))
135       val old = state.put(feature.getID, tuple)
136       if (old != null && old._2 > tuple._2) {
137         state.replace(feature.getID, tuple, old)
138       }
139     }
140 
141     override def remove(id: String): Unit = state.remove(id)
142 
143     override def clear(): Unit = state.clear()
144 
145     override def close(): Unit = {}
146 
147     override def size(): Int = state.size()
148 
149     override def size(filter: Filter): Int = query(filter).length
150 
151     override def query(id: String): Option[SimpleFeature] = Option(state.get(id)).map(_._1)
152 
153     override def query(filter: Filter): Iterator[SimpleFeature] = {
154       import scala.collection.JavaConverters._
155       val features = state.asScala.valuesIterator.map(_._1)
156       if (filter == Filter.INCLUDE) { features } else {
157         features.filter(filter.evaluate)
158       }
159     }
160 
161     override def views: Seq[KafkaFeatureCacheView] = Seq.empty
162   }
163 
164   class EmptyFeatureCache(val views: Seq[KafkaFeatureCacheView]) extends KafkaFeatureCache {
165     override def put(feature: SimpleFeature): Unit = throw new UnsupportedOperationException("Empty feature cache")
166     override def remove(id: String): Unit = throw new UnsupportedOperationException("Empty feature cache")
167     override def clear(): Unit = throw new UnsupportedOperationException("Empty feature cache")
168     override def size(): Int = 0
169     override def size(filter: Filter): Int = 0
170     override def query(id: String): Option[SimpleFeature] = None
171     override def query(filter: Filter): Iterator[SimpleFeature] = Iterator.empty
172     override def close(): Unit = {}
173     override def addListener(source: SimpleFeatureSource, listener: FeatureListener): Unit = {}
174     override def removeListener(source: SimpleFeatureSource, listener: FeatureListener): Unit = {}
175   }
176 
177   private class NoOpFeatureCache(val views: Seq[KafkaFeatureCacheView]) extends KafkaFeatureCache {
178     override def put(feature: SimpleFeature): Unit = {}
179     override def remove(id: String): Unit = {}
180     override def clear(): Unit = {}
181     override def size(): Int = 0
182     override def size(filter: Filter): Int = 0
183     override def query(id: String): Option[SimpleFeature] = None
184     override def query(filter: Filter): Iterator[SimpleFeature] = Iterator.empty
185     override def close(): Unit = {}
186   }
187 }