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.partitions
10 
11 import org.geotools.api.feature.simple.{SimpleFeature, SimpleFeatureType}
12 import org.geotools.api.filter.Filter
13 import org.locationtech.geomesa.filter.FilterHelper
14 import org.locationtech.geomesa.fs.storage.api.PartitionScheme.SimplifiedFilter
15 import org.locationtech.geomesa.fs.storage.api.{NamedOptions, PartitionScheme, PartitionSchemeFactory}
16 import org.locationtech.geomesa.utils.date.DateUtils.toInstant
17 import org.locationtech.geomesa.utils.text.DateParsing
18 
19 import java.time.format.{DateTimeFormatter, DateTimeFormatterBuilder}
20 import java.time.temporal.{ChronoField, ChronoUnit, TemporalAdjusters, WeekFields}
21 import java.time.{ZoneOffset, ZonedDateTime}
22 import java.util.Date
23 import scala.annotation.tailrec
24 import scala.collection.mutable.ListBuffer
25 import scala.util.control.NonFatal
26 
27 case class DateTimeScheme(
28     formatter: DateTimeFormatter,
29     pattern: String,
30     stepUnit: ChronoUnit,
31     step: Int,
32     dtg: String,
33     dtgIndex: Int
34   ) extends PartitionScheme {
35 
36   import FilterHelper.ff
37   import org.locationtech.geomesa.filter.{andOption, isTemporalFilter, partitionSubFilters}
38 
39   import ChronoUnit._
40 
41   // note: `truncatedTo` is only valid up to DAYS, other units require additional steps
42   private val truncate: ZonedDateTime => ZonedDateTime = stepUnit match {
43     case NANOS | MICROS | MILLIS | SECONDS | MINUTES | HOURS | DAYS =>
44       dt => dt.truncatedTo(stepUnit)
45 
46     case MONTHS =>
47       val adjuster = TemporalAdjusters.firstDayOfMonth()
48       dt => dt.`with`(adjuster).truncatedTo(DAYS)
49 
50     case YEARS =>
51       val adjuster = TemporalAdjusters.firstDayOfYear()
52       dt => dt.`with`(adjuster).truncatedTo(DAYS)
53 
54     case _ =>
55       dt => DateParsing.parse(formatter.format(dt), formatter) // fall back to format and re-parse
56   }
57 
58   // TODO This may not be the best way to calculate max depth...
59   // especially if we are going to use other separators
60   override val depth: Int = pattern.count(_ == '/') + 1
61 
62   override def getPartitionName(feature: SimpleFeature): String =
63     formatter.format(toInstant(feature.getAttribute(dtgIndex).asInstanceOf[Date]).atZone(ZoneOffset.UTC))
64 
65   override def getSimplifiedFilters(filter: Filter, partition: Option[String]): Option[Seq[SimplifiedFilter]] = {
66     getCoveringPartitions(filter).map { case (covered, intersecting) =>
67       val result = Seq.newBuilder[SimplifiedFilter]
68 
69       if (covered.nonEmpty) {
70         // remove the temporal filter that we've already accounted for in our covered partitions
71         val coveredFilter = andOption(partitionSubFilters(filter, isTemporalFilter(_, dtg))._2)
72         result += SimplifiedFilter(coveredFilter.getOrElse(Filter.INCLUDE), covered, partial = false)
73       }
74       if (intersecting.nonEmpty) {
75         result += SimplifiedFilter(filter, intersecting.distinct, partial = false)
76       }
77 
78       partition match {
79         case None => result.result
80         case Some(p) =>
81           val matched = result.result.find(_.partitions.contains(p))
82           matched.map(_.copy(partitions = Seq(p))).toSeq
83       }
84     }
85   }
86 
87   override def getIntersectingPartitions(filter: Filter): Option[Seq[String]] =
88     getCoveringPartitions(filter).map { case (covered, intersecting) => covered ++ intersecting }
89 
90   override def getCoveringFilter(partition: String): Filter = {
91     val zdt = DateParsing.parse(partition, formatter)
92     val start = DateParsing.format(zdt)
93     val end = DateParsing.format(zdt.plus(1, stepUnit))
94     ff.and(ff.greaterOrEqual(ff.property(dtg), ff.literal(start)), ff.less(ff.property(dtg), ff.literal(end)))
95   }
96 
97   private def getCoveringPartitions(filter: Filter): Option[(Seq[String], Seq[String])] = {
98     val bounds = FilterHelper.extractIntervals(filter, dtg, handleExclusiveBounds = false)
99     if (bounds.disjoint) {
100       Some((Seq.empty, Seq.empty))
101     } else if (bounds.isEmpty || !bounds.forall(_.isBoundedBothSides)) {
102       None
103     } else {
104       // there should be no duplicates in covered partitions, as our bounds will not overlap,
105       // but there may be multiple partial intersects with a given partition
106       val covered = ListBuffer.empty[String]
107       val intersecting = ListBuffer.empty[String]
108 
109       bounds.values.foreach { bound =>
110         // note: we verified both sides are bounded above
111         val lower = bound.lower.value.get
112         val upper = bound.upper.value.get
113         val start = truncate(lower)
114         // `stepUnit.between` claims to be upper endpoint exclusive, but doesn't seem to be...
115         // if exclusive end, subtract one milli so we don't search the whole next partition if it's on the endpoint
116         // note: we only support millisecond resolution
117         val end = truncate(if (bound.upper.inclusive) { upper } else { upper.minus(1, MILLIS) })
118         val steps = stepUnit.between(start, end).toInt
119 
120         // do our endpoints match the partition boundary, or do we need to apply a filter to the first/last partition?
121         val coveringFirst = bound.lower.inclusive && lower == start
122         val coveringLast =
123           (bound.upper.exclusive && upper == end.plus(1, stepUnit)) ||
124               (bound.upper.inclusive && !upper.isBefore(end.plus(1, stepUnit).minus(1, MILLIS)))
125 
126         if (steps == 0) {
127           if (coveringFirst && coveringLast) {
128             covered += formatter.format(start)
129           } else {
130             intersecting += formatter.format(start)
131           }
132         } else {
133           // add the first partition
134           if (coveringFirst) {
135             covered += formatter.format(start)
136           } else {
137             intersecting += formatter.format(start)
138           }
139 
140           @tailrec
141           def addSteps(step: Int, current: ZonedDateTime): Unit = {
142             if (step == steps) {
143               // last partition
144               if (coveringLast) {
145                 covered += formatter.format(current)
146               } else {
147                 intersecting += formatter.format(current)
148               }
149             } else {
150               covered += formatter.format(current)
151               addSteps(step + 1, current.plus(1, stepUnit))
152             }
153           }
154 
155           addSteps(1, start.plus(1, stepUnit))
156         }
157       }
158 
159       Some((covered.toSeq, intersecting.toSeq))
160     }
161   }
162 }
163 
164 object DateTimeScheme {
165 
166   val Name = "datetime"
167 
168   def apply(format: String, stepUnit: ChronoUnit, step: Int, dtg: String, dtgIndex: Int): DateTimeScheme =
169     DateTimeScheme(DateTimeFormatter.ofPattern(format), format, stepUnit, step, dtg, dtgIndex)
170 
171   @deprecated("Pattern is not correct when using this constructor")
172   def apply(formatter: DateTimeFormatter, stepUnit: ChronoUnit, step: Int, dtg: String, dtgIndex: Int): DateTimeScheme =
173     DateTimeScheme(formatter, formatter.toString, stepUnit, step, dtg, dtgIndex)
174 
175   object Config {
176     val DateTimeFormatOpt: String = "datetime-format"
177     val StepUnitOpt      : String = "step-unit"
178     val StepOpt          : String = "step"
179     val DtgAttribute     : String = "dtg-attribute"
180   }
181 
182   object Formats {
183 
184     def apply(name: String): Option[Format] = all.find(_.name.equalsIgnoreCase(name))
185 
186     case class Format private[Formats](name: String, formatter: DateTimeFormatter, pattern: String, unit: ChronoUnit)
187 
188     private[Formats] object Format {
189       def apply(name: String, format: String, unit: ChronoUnit): Format =
190         Format(name, DateTimeFormatter.ofPattern(format), format, unit)
191     }
192 
193     val Minute       : Format = Format("minute",        "yyyy/MM/dd/HH/mm", ChronoUnit.MINUTES)
194     val Hourly       : Format = Format("hourly",        "yyyy/MM/dd/HH",    ChronoUnit.HOURS  )
195     val Daily        : Format = Format("daily",         "yyyy/MM/dd",       ChronoUnit.DAYS   )
196     val Monthly      : Format = Format("monthly",       "yyyy/MM",          ChronoUnit.MONTHS )
197     val JulianMinute : Format = Format("julian-minute", "yyyy/DDD/HH/mm",   ChronoUnit.MINUTES)
198     val JulianHourly : Format = Format("julian-hourly", "yyyy/DDD/HH",      ChronoUnit.HOURS  )
199     val JulianDaily  : Format = Format("julian-daily",  "yyyy/DDD",         ChronoUnit.DAYS   )
200 
201     // java.time doesn't seem to have a way to parse back out a year/week format...
202     // to get around that we have to define the default day of week in the formatter
203     val Weekly: Format = {
204       val formatter =
205         new DateTimeFormatterBuilder()
206           .appendValue(WeekFields.ISO.weekBasedYear(), 4)
207           .appendLiteral("/W")
208           .appendValue(WeekFields.ISO.weekOfWeekBasedYear(), 2)
209           .parseDefaulting(ChronoField.DAY_OF_WEEK, 1)
210           .toFormatter()
211       Format("weekly", formatter, "YYYY/'W'ww", ChronoUnit.WEEKS)
212     }
213 
214     private val all = Seq(Minute, Hourly, Daily, Weekly, Monthly, JulianMinute, JulianHourly, JulianDaily)
215   }
216 
217   class DateTimePartitionSchemeFactory extends PartitionSchemeFactory {
218     override def load(sft: SimpleFeatureType, config: NamedOptions): Option[PartitionScheme] = {
219       import org.locationtech.geomesa.utils.geotools.RichSimpleFeatureType.RichSimpleFeatureType
220 
221       lazy val step = config.options.get(Config.StepOpt).map(_.toInt).getOrElse(1)
222       lazy val dtg = config.options.get(Config.DtgAttribute).orElse(sft.getDtgField).getOrElse {
223         throw new IllegalArgumentException(s"DateTime scheme requires valid attribute '${Config.DtgAttribute}'")
224       }
225       lazy val dtgIndex = Some(sft.indexOf(dtg)).filter(_ != -1).getOrElse {
226         throw new IllegalArgumentException(s"Attribute '$dtg' does not exist in feature type ${sft.getTypeName}")
227       }
228 
229       if (config.name == Name) {
230         val unit = config.options.get(Config.StepUnitOpt).map(c => ChronoUnit.valueOf(c.toUpperCase)).getOrElse {
231           throw new IllegalArgumentException(s"DateTime scheme requires valid unit '${Config.StepUnitOpt}'")
232         }
233         val format = config.options.getOrElse(Config.DateTimeFormatOpt,
234           throw new IllegalArgumentException(s"DateTime scheme requires valid format '${Config.DateTimeFormatOpt}'"))
235         require(!format.endsWith("/"), "Format cannot end with a slash")
236         val formatter = try { DateTimeFormatter.ofPattern(format) } catch {
237           case NonFatal(e) => throw new IllegalArgumentException(s"Invalid date format '$format':", e)
238         }
239         Some(DateTimeScheme(formatter, format, unit, step, dtg, dtgIndex))
240       } else {
241         Formats(config.name).map(f => DateTimeScheme(f.formatter, f.pattern, f.unit, step, dtg, dtgIndex))
242       }
243     }
244   }
245 }