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.lambda.data
10 
11 import com.github.benmanes.caffeine.cache.LoadingCache
12 import org.geotools.api.feature.simple.SimpleFeatureType
13 import org.geotools.api.filter.Filter
14 import org.geotools.util.factory.Hints
15 import org.locationtech.geomesa.curve.TimePeriod.TimePeriod
16 import org.locationtech.geomesa.index.stats.GeoMesaStats.GeoMesaStatWriter
17 import org.locationtech.geomesa.index.stats.impl._
18 import org.locationtech.geomesa.index.stats.{GeoMesaStats, Stat}
19 import org.locationtech.geomesa.lambda.stream.TransientStore
20 
21 class LambdaStats(persistent: GeoMesaStats, transients: LoadingCache[String, TransientStore]) extends GeoMesaStats {
22 
23   // note: transient stat methods always return Some for Count and MinMax,
24   // but otherwise return None if not exact
25 
26   override def writer: GeoMesaStatWriter = persistent.writer
27 
28   override def getCount(sft: SimpleFeatureType, filter: Filter, exact: Boolean, queryHints: Hints): Option[Long] = {
29     persistent.getCount(sft, filter, exact, queryHints).map {
30       case -1L => -1L
31       case p =>
32         val t = transient(sft).getCount(sft, filter, exact, queryHints).getOrElse {
33           throw new IllegalStateException("Transient stats returned None")
34         }
35         p + t
36     }
37   }
38 
39   override def getMinMax[T](
40       sft: SimpleFeatureType,
41       attribute: String,
42       filter: Filter,
43       exact: Boolean): Option[MinMax[T]] = {
44     val t = transient(sft).getMinMax[T](sft, attribute, filter, exact).getOrElse {
45       throw new IllegalStateException("Transient stats returned None")
46     }
47     persistent.getMinMax[T](sft, attribute, filter, exact).map(_ + t).orElse(Some(t))
48   }
49 
50   override def getEnumeration[T](
51       sft: SimpleFeatureType,
52       attribute: String,
53       filter: Filter,
54       exact: Boolean): Option[EnumerationStat[T]] = {
55     val per = persistent.getEnumeration[T](sft, attribute, filter, exact)
56     if (!exact) { per } else {
57       per.flatMap(p => transient(sft).getEnumeration[T](sft, attribute, filter, exact).map(_ + p))
58     }
59   }
60 
61   override def getFrequency[T](
62       sft: SimpleFeatureType,
63       attribute: String,
64       precision: Int,
65       filter: Filter,
66       exact: Boolean): Option[Frequency[T]] = {
67     val per = persistent.getFrequency[T](sft, attribute, precision, filter, exact)
68     if (!exact) { per } else {
69       per.flatMap(p => transient(sft).getFrequency[T](sft, attribute, precision, filter, exact).map(_ + p))
70     }
71   }
72 
73   override def getTopK[T](
74       sft: SimpleFeatureType,
75       attribute: String,
76       filter: Filter,
77       exact: Boolean): Option[TopK[T]] = {
78     val per = persistent.getTopK[T](sft, attribute, filter, exact)
79     if (!exact) { per } else {
80       per.flatMap(p => transient(sft).getTopK[T](sft, attribute, filter, exact).map(_ + p))
81     }
82   }
83 
84   override def getHistogram[T](
85       sft: SimpleFeatureType,
86       attribute: String,
87       bins: Int,
88       min: T,
89       max: T,
90       filter: Filter,
91       exact: Boolean): Option[Histogram[T]] = {
92     val per = persistent.getHistogram[T](sft, attribute, bins, min, max, filter, exact)
93     if (!exact) { per } else {
94       per.flatMap(p => transient(sft).getHistogram[T](sft, attribute, bins, min, max, filter, exact).map(_ + p))
95     }
96   }
97 
98   override def getZ3Histogram(
99       sft: SimpleFeatureType,
100       geom: String,
101       dtg: String,
102       period: TimePeriod,
103       bins: Int,
104       filter: Filter,
105       exact: Boolean): Option[Z3Histogram] = {
106     val per = persistent.getZ3Histogram(sft, geom, dtg, period, bins, filter, exact)
107     if (!exact) { per } else {
108       per.flatMap(p => transient(sft).getZ3Histogram(sft, geom, dtg, period, bins, filter, exact).map(_ + p))
109     }
110   }
111 
112   override def getStat[T <: Stat](
113       sft: SimpleFeatureType,
114       query: String,
115       filter: Filter,
116       exact: Boolean): Option[T] = {
117     val per = persistent.getStat[T](sft, query, filter, exact)
118     if (!exact) { per } else {
119       per.flatMap(p => transient(sft).getStat[T](sft, query, filter, exact).map(_ + p)).asInstanceOf[Option[T]]
120     }
121   }
122 
123   // note: closed by closing the wrapping stores in LambdaDataStore.dispose()
124   override def close(): Unit = {}
125 
126   private def transient(sft: SimpleFeatureType): GeoMesaStats = transients.get(sft.getTypeName).stats
127 }