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.accumulo.tools.data
10 
11 import com.beust.jcommander.{Parameter, ParameterException, Parameters}
12 import org.apache.accumulo.core.client.admin.TableOperations
13 import org.apache.accumulo.core.data.Key
14 import org.apache.hadoop.io.Text
15 import org.locationtech.geomesa.accumulo.tools.data.AccumuloCompactCommand.{CompactParams, RangeCompaction}
16 import org.locationtech.geomesa.accumulo.tools.{AccumuloDataStoreCommand, AccumuloDataStoreParams}
17 import org.locationtech.geomesa.curve.BinnedTime
18 import org.locationtech.geomesa.index.api.GeoMesaFeatureIndex
19 import org.locationtech.geomesa.index.index.id.IdIndex
20 import org.locationtech.geomesa.index.index.z3.Z3Index
21 import org.locationtech.geomesa.tools.utils.ParameterConverters.DurationConverter
22 import org.locationtech.geomesa.tools.{Command, RequiredTypeNameParam}
23 import org.locationtech.geomesa.utils.index.ByteArrays
24 import org.locationtech.geomesa.utils.text.TextTools
25 import org.locationtech.geomesa.utils.uuid.Z3UuidGenerator
26 
27 import java.nio.charset.StandardCharsets
28 import java.time.{Instant, ZoneOffset, ZonedDateTime}
29 import java.util.concurrent.{Executors, TimeUnit}
30 import java.util.{Date, UUID}
31 import scala.concurrent.duration.Duration
32 import scala.util.control.NonFatal
33 
34 class AccumuloCompactCommand extends AccumuloDataStoreCommand {
35 
36   import org.locationtech.geomesa.filter.ff
37   import org.locationtech.geomesa.utils.geotools.RichSimpleFeatureType.RichSimpleFeatureType
38 
39   import scala.collection.JavaConverters._
40 
41   override val name = "compact"
42   override val params = new CompactParams
43 
44   override def execute(): Unit = withDataStore { ds =>
45     val sft = ds.getSchema(params.featureName)
46     if (sft == null) {
47       throw new ParameterException(s"Schema '${params.featureName}' does not exist in the data store")
48     }
49 
50     val start = System.currentTimeMillis()
51 
52     val executor = Executors.newFixedThreadPool(params.threads)
53 
54     val ops = ds.client.tableOperations()
55 
56     val msg = new StringBuilder(s"Starting incremental compaction using ${params.threads} simultaneous threads")
57 
58     val interval = Option(params.from).map { from =>
59       val now = System.currentTimeMillis()
60       val start = now - from.toMillis
61       val end = Option(params.duration).map(d => start + d.toMillis).getOrElse(now)
62       def toString(millis: Long) = ZonedDateTime.ofInstant(Instant.ofEpochMilli(millis), ZoneOffset.UTC).toString
63       msg.append(s" from ${toString(start)}/${toString(end)}")
64       (start, end)
65     }
66 
67     val z3Bins = interval.map { case (s, e) =>
68       val toBin = BinnedTime.timeToBinnedTime(sft.getZ3Interval)
69       (toBin(s).bin, toBin(e).bin)
70     }
71 
72     val filter = interval.flatMap { case (s, e) =>
73       sft.getDtgField.map { dtg =>
74         ff.between(ff.property(dtg), ff.literal(new Date(s)), ff.literal(new Date(e)))
75       }
76     }
77 
78     Command.user.info(msg.toString)
79 
80     def filterSplits(index: GeoMesaFeatureIndex[_, _]): Iterator[Seq[Text]] => Iterator[Seq[Text]] = {
81       z3Bins match {
82         case Some((min, max)) if index.name == Z3Index.name =>
83           val offset = index.keySpace.sharding.length + index.keySpace.sharing.length
84 
85           def compareStart(s: Text): Boolean =
86             s == null || s.getLength < offset + 2 || ByteArrays.readShort(s.getBytes, offset) <= max
87           def compareEnd(e: Text): Boolean =
88             e == null || e.getLength < offset + 2 || ByteArrays.readShort(e.getBytes, offset) >= min
89 
90           iter => iter.filter { case Seq(s, e) => compareStart(s) && compareEnd(e) }
91 
92         case Some((min, max)) if params.z3Ids && index.name == IdIndex.name =>
93           val offset = if (sft.isTableSharing) { 1 } else { 0 }
94           if (sft.isUuidEncoded) {
95             // uuid is already stored in correct binary format
96             def compareStart(s: Text): Boolean =
97               s == null || s.getLength < offset + 3 || Z3UuidGenerator.timeBin(s.getBytes, offset) <= max
98             def compareEnd(e: Text): Boolean =
99               e == null || e.getLength < offset + 3 || Z3UuidGenerator.timeBin(e.getBytes, offset) >= min
100 
101             iter => iter.filter { case Seq(s, e) => compareStart(s) && compareEnd(e) }
102           } else {
103             // uuid is stored as a string, must be parsed into a uuid and converted to bytes
104             def compareStart(s: Text): Boolean = {
105               if (s == null) { true } else {
106                 try {
107                   val uuidString = new String(s.getBytes, offset, s.getLength - offset, StandardCharsets.UTF_8)
108                   val uuid = UUID.fromString(uuidString)
109                   Z3UuidGenerator.timeBin(ByteArrays.toBytes(uuid.getMostSignificantBits)) <= max
110                 } catch {
111                   case NonFatal(_) => true // split doesn't contain a whole row key
112                 }
113               }
114             }
115             def compareEnd(e: Text): Boolean = {
116               if (e == null) { true } else {
117                 try {
118                   val uuidString = new String(e.getBytes, offset, e.getLength - offset, StandardCharsets.UTF_8)
119                   val uuid = UUID.fromString(uuidString)
120                   Z3UuidGenerator.timeBin(ByteArrays.toBytes(uuid.getMostSignificantBits)) >= min
121                 } catch {
122                   case NonFatal(_) => true // split doesn't contain a whole row key
123                 }
124               }
125             }
126 
127             iter => iter.filter { case Seq(s, e) => compareStart(s) && compareEnd(e) }
128           }
129 
130         case _ =>
131           iter => iter
132       }
133     }
134 
135     ds.manager.indices(sft).foreach { index =>
136       val filtering = filterSplits(index)
137 
138       index.getTablesForQuery(filter).foreach { table =>
139         val tableSplits = ops.listSplits(table).asScala.toList
140 
141         var count = 0
142 
143         if (tableSplits.isEmpty) {
144           executor.submit(new RangeCompaction(ops, table, null, null))
145           count += 1
146         } else {
147           val head = Iterator.single(Seq(null, tableSplits.head))
148           val last = Iterator.single(Seq(tableSplits.last, null))
149           val middle = if (tableSplits.lengthCompare(1) == 0) { Iterator.empty } else { tableSplits.sliding(2) }
150 
151           // filter out ranges by table sharing, if possible
152           val splits = if (sft.isTableSharing) {
153             val Array(prefix) = sft.getTableSharingBytes // should be one byte
154             (head ++ middle ++ last).filter { case Seq(s, e) =>
155               (s == null || ByteArrays.UnsignedByteOrdering.compare(s.getBytes.apply(0), prefix) <= 0) &&
156                   (e == null || ByteArrays.UnsignedByteOrdering.compare(e.getBytes.apply(0), prefix) >= 0)
157             }
158           } else {
159             head ++ middle ++ last
160           }
161 
162           // filter out ranges based on our time interval, if possible
163           filtering(splits).foreach { case Seq(s, e)  =>
164             executor.submit(new RangeCompaction(ops, table, s, e))
165             count += 1
166           }
167         }
168         Command.user.info(s"Found $count splits for table $table")
169       }
170     }
171 
172     executor.shutdown()
173     executor.awaitTermination(Long.MaxValue, TimeUnit.MILLISECONDS)
174 
175     Command.user.info(s"Compactions complete in ${TextTools.getTime(start)}")
176   }
177 }
178 
179 object AccumuloCompactCommand {
180 
181   @Parameters(commandDescription = "Incrementally compact tables for a GeoMesa feature type")
182   class CompactParams extends RequiredTypeNameParam with AccumuloDataStoreParams {
183 
184     @Parameter(names = Array("--threads"), description = "Number of ranges to compact simultaneously")
185     var threads: Integer = Int.box(4)
186 
187     @Parameter(names = Array("--from"), description = "How long ago to compact data, based on the default date attribute, relative to current time. E.g. '1 day', '2 weeks and 1 hour', etc", converter = classOf[DurationConverter])
188     var from: Duration = _
189 
190     @Parameter(names = Array("--duration"), description = "Amount of time to compact data, based on the default date attribute, relative to '--from'. E.g. '1 day', '2 weeks and 1 hour', etc", converter = classOf[DurationConverter])
191     var duration: Duration = _
192 
193     @Parameter(names = Array("--z3-feature-ids"), description = "Will only compact ID records that correspond with the time period, based on features being written with the Z3FeatureIdGenerator")
194     var z3Ids: Boolean = false
195   }
196 
197   class RangeCompaction(ops: TableOperations, table: String, start: Text, end: Text) extends Runnable {
198     override def run(): Unit = {
199       Command.user.info(s"Starting compaction of $table [ ${rowToString(start)} :: ${rowToString(end)} ]")
200       ops.compact(table, start, end, false, true)
201     }
202   }
203 
204   private def rowToString(row: Text): String = {
205     if (row == null) { "null" } else { Key.toPrintableString(row.getBytes, 0, row.getLength, row.getLength) }
206   }
207 }