/
AccumuloLayerDeleter.scala
65 lines (55 loc) · 2.48 KB
/
AccumuloLayerDeleter.scala
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
/*
* Copyright 2016 Azavea
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package geotrellis.spark.io.accumulo
import geotrellis.spark.LayerId
import geotrellis.spark.io._
import geotrellis.util.LazyLogging
import org.apache.accumulo.core.client.{BatchWriterConfig, Connector}
import org.apache.accumulo.core.security.Authorizations
import org.apache.accumulo.core.data.{Range => AccumuloRange}
import spray.json.JsonFormat
import spray.json.DefaultJsonProtocol._
import scala.collection.JavaConversions._
class AccumuloLayerDeleter(val attributeStore: AttributeStore, connector: Connector) extends LazyLogging with LayerDeleter[LayerId] {
def delete(id: LayerId): Unit = {
try {
val header = attributeStore.readHeader[AccumuloLayerHeader](id)
val numThreads = 1
val config = new BatchWriterConfig()
config.setMaxWriteThreads(numThreads)
val deleter = connector.createBatchDeleter(header.tileTable, new Authorizations(), numThreads, config)
deleter.fetchColumnFamily(columnFamily(id))
deleter.setRanges(new AccumuloRange() :: Nil)
deleter.delete()
} catch {
case e: AttributeNotFoundError =>
logger.info(s"Metadata for $id was not found. Any associated layer data (if any) will require manual deletion")
throw new LayerDeleteError(id).initCause(e)
} finally {
attributeStore.delete(id)
}
}
}
object AccumuloLayerDeleter {
def apply(attributeStore: AttributeStore, connector: Connector): AccumuloLayerDeleter =
new AccumuloLayerDeleter(attributeStore, connector)
def apply(attributeStore: AttributeStore, instance: AccumuloInstance): AccumuloLayerDeleter =
new AccumuloLayerDeleter(attributeStore, instance.connector)
def apply(attributeStore: AccumuloAttributeStore): AccumuloLayerDeleter =
new AccumuloLayerDeleter(attributeStore, attributeStore.connector)
def apply(instance: AccumuloInstance): AccumuloLayerDeleter =
apply(AccumuloAttributeStore(instance.connector), instance.connector)
}