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.converter
10 
11 import com.typesafe.scalalogging.LazyLogging
12 import org.locationtech.geomesa.fs.storage.api._
13 import org.locationtech.geomesa.fs.storage.converter.ConverterStorageFactory._
14 import org.locationtech.geomesa.utils.geotools.{SftArgResolver, SftArgs}
15 
16 import java.util.regex.Pattern
17 
18 class ConverterMetadataFactory extends StorageMetadataFactory with LazyLogging {
19 
20   import scala.collection.JavaConverters._
21 
22   override def name: String = ConverterStorage.Encoding
23 
24   override def load(context: FileSystemContext): Option[StorageMetadata] = {
25     if (!Option(context.conf.get(ConverterPathParam)).contains(context.root.getName)) { None } else {
26       try {
27         val sft = {
28           val sftArg = Option(context.conf.get(SftConfigParam))
29               .orElse(Option(context.conf.get(SftNameParam)))
30               .getOrElse(throw new IllegalArgumentException(s"Must provide either simple feature type config or name"))
31           SftArgResolver.getArg(SftArgs(sftArg, null)) match {
32             case Left(e) => throw new IllegalArgumentException("Could not load SimpleFeatureType with provided parameters", e)
33             case Right(schema) => schema
34           }
35         }
36 
37         val partitionSchemeOpts =
38           context.conf.getValByRegex(Pattern.quote(PartitionOptsPrefix) + ".*").asScala.map {
39             case (k, v) => k.substring(PartitionOptsPrefix.length) -> v
40           }
41 
42         val scheme = {
43           val partitionSchemeName =
44             Option(context.conf.get(PartitionSchemeParam))
45                 .getOrElse(throw new IllegalArgumentException(s"Must provide partition scheme name"))
46           PartitionSchemeFactory.load(sft, NamedOptions(partitionSchemeName, partitionSchemeOpts.toMap))
47         }
48 
49         val leafStorage = Option(context.conf.get(LeafStorageParam)).map(_.toBoolean).getOrElse {
50           val deprecated = partitionSchemeOpts.get("leaf-storage").map { s =>
51             logger.warn("Using deprecated leaf-storage partition-scheme option. Please define leaf-storage using " +
52                 s"'$LeafStorageParam'")
53             s.toBoolean
54           }
55           deprecated.getOrElse(true)
56         }
57 
58         Some(new ConverterMetadata(context, sft, scheme, leafStorage))
59       } catch {
60         case e: IllegalArgumentException => logger.warn(s"Couldn't create converter storage metadata: $e", e); None
61       }
62     }
63   }
64 
65   override def create(context: FileSystemContext, config: Map[String, String], meta: Metadata): StorageMetadata =
66     throw new NotImplementedError("Converter storage is read only")
67 }