GEOMESA-3578 FSDS - Standardize target file size (#3545)

* Target file size must be set explicitly
* Avoids updating metadata for one-off commands
* Store configs in separate rows in JdbcMetadata to reduce update cost
* Fix potential conflict between feature types in JdbcMetadata

Author: elahrvivaz

docs/user/filesystem/commandline.rst

@@ -44,7 +44,6 @@ Argument                 Description
 ``-p, --path *``         The filesystem root path used to store data
 ``-f, --feature-name *`` The name of the schema
 ``--partitions``         Partitions to compact (omit to compact all partitions)
-``--target-file-size``   Target size for data files (e.g. 500MB or 1GB)
 ``--mode``               One of ``local`` or ``distributed`` (to use map/reduce)
 ``--temp-path``          Path to a temp directory used for working files
 ======================== =========================================================
@@ -113,7 +112,6 @@ Argument                 Description
 ``-p, --path *``         The filesystem root path used to store data
 ``--partition-scheme``   Partition schemes
 ``--num-reducers``       Number of reducers to use (required for distributed ingest)
-``--target-file-size``   Target size for data files (e.g. 500MB or 1GB)
 ``--temp-path``          Path to a temp directory used for working files
 ``--storage-opt``        Additional storage options to set as SimpleFeatureType user data, in the form ``key=value``
 ======================== =============================================================
@@ -137,6 +135,7 @@ sub-commands:
 
 * ``register`` - create a new metadata entry for an existing data file
 * ``unregister`` - remove a metadata entry for an existing data file
+* ``configure`` - set or unset metadata configuration values
 * ``migrate`` - migrate metadata from one type to another
 * ``check-consistency`` - check consistency between the metadata and data files
 
@@ -176,6 +175,17 @@ Argument                 Description
 ``<file> *``             The path of the file to unregister, relative to the storage root path
 ======================== =====================================================================================================
 
+``configure``
+~~~~~~~~~~~~~
+
+The ``configure`` sub-command lets you set storage-level configuration options.
+
+======================== =====================================================================================================
+Argument                 Description
+======================== =====================================================================================================
+``--set *``              The configuration to set, in the form ``key=value``. To remove a key, use an empty value
+======================== =====================================================================================================
+
 ``migrate``
 ~~~~~~~~~~~~~~
 

docs/user/filesystem/index_config.rst

@@ -75,9 +75,9 @@ the user data key ``geomesa.fs.file-size``:
         // or set directly in the user data as a string
         sft.getUserData.put("geomesa.fs.file-size", "1GB")
 
-Note that target file size can also be specified in some operations, which will override any default configured
-in the feature type. See :ref:`fsds_compact_command` and :ref:`fsds_ingest_command` for details. See
-:ref:`fsds_size_threshold_prop` for controlling the file size error margin.
+Once the schema has been created, the file size can be configured through the storage metadata key ``target-file-size``. See
+:ref:`fsds_manage_metadata_command` for setting metadata keys, and see :ref:`fsds_size_threshold_prop` for controlling the file
+size error margin.
 
 Configuring Visibility Persistence
 ----------------------------------

geomesa-features/geomesa-feature-exporters/src/main/scala/org/locationtech/geomesa/features/exporters/GeoParquetExporter.scala

@@ -54,7 +54,7 @@ class GeoParquetExporter(path: String) extends FeatureExporter with LazyLogging
     Some(i)
   }
 
-  override def bytes: Long = if (writer == null) { 0 } else { writer.getDataSize }
+  override def bytes: Long = if (writer == null) { 0 } else { writer.size }
 
   override def close(): Unit = CloseWithLogging(Seq(writer, fs).filter(_ != null))
 }

geomesa-fs/geomesa-fs-storage/geomesa-fs-storage-convert/src/main/scala/org/locationtech/geomesa/fs/storage/converter/ConverterStorage.scala

@@ -44,7 +44,7 @@ class ConverterStorage(
     new ConverterFileSystemReader(fs, context.root, converter, filter, transform, pathFiltering)
   }
 
-  override def compact(partition: Partition, fileSize: Option[Long], threads: Int): Unit =
+  override def compact(partition: Partition, threads: Int): Unit =
     throw new UnsupportedOperationException("Converter storage does not support compactions")
 
   override def register(file: URI): StorageFile =

geomesa-fs/geomesa-fs-storage/geomesa-fs-storage-core/src/main/scala/org/locationtech/geomesa/fs/storage/core/FileSystemStorage.scala

@@ -171,13 +171,11 @@ abstract class FileSystemStorage(val context: FileSystemContext, val metadata: S
    * multiple threads or storage instances attempt to compact the same partition simultaneously.
    *
    * @param partition partition to compact, or all partitions
-   * @param fileSize approximate target size of files, in bytes
    * @param threads suggested threads to use for file system operations
    */
-  def compact(partition: Partition, fileSize: Option[Long] = None, threads: Int = 1): Unit = {
-    val target = fileSize.orElse(this.sizer.targetSize)
+  def compact(partition: Partition, threads: Int = 1): Unit = {
     val files = metadata.getFiles(partition)
-    val toCompact = target match {
+    val toCompact = sizer.targetSize match {
       case None => files
       case Some(t) =>
         files.filter { f =>
@@ -201,7 +199,7 @@ abstract class FileSystemStorage(val context: FileSystemContext, val metadata: S
       // tracks newly added files so we can register them atomically
       val fileTracker = new FileTracker(metadata.sft, metadata.schemes)
 
-      WithClose(createWriter(partition, StorageFileAction.Append, FileType.Compacted, target, fileTracker)) { writer =>
+      WithClose(createWriter(partition, StorageFileAction.Append, FileType.Compacted, fileTracker)) { writer =>
         WithClose(FileSystemThreadedReader(reader, toCompact, threads)) { reader =>
           while (reader.hasNext) {
             val feature = reader.next()
@@ -265,7 +263,7 @@ abstract class FileSystemStorage(val context: FileSystemContext, val metadata: S
       case StorageFileAction.Modify => FileType.Modified
       case StorageFileAction.Delete => FileType.Deleted
     }
-    createWriter(partition, action, fileType, sizer.targetSize, metadata)
+    createWriter(partition, action, fileType, metadata)
   }
 
   /**
@@ -274,30 +272,28 @@ abstract class FileSystemStorage(val context: FileSystemContext, val metadata: S
    * @param partition partition being written to
    * @param action write type
    * @param fileType file type
-   * @param targetFileSize target file size
    * @param metadata metata to track added files
    * @return
    */
   private def createWriter(
       partition: Partition,
       action: StorageFileAction,
       fileType: FileType,
-      targetFileSize: Option[Long],
       metadata: StorageMetadata): FileSystemWriter = {
 
-    def pathAndWriter: (URI, FileSystemWriter) = {
+    def newWriter(): FileSystemWriter = {
       val file = FileSystemStorage.newFilePath(metadata.sft.getTypeName, fileType, encoding)
       val path = context.root.resolve(file)
       val updateObserver = new MetadataObserver(metadata, file, partition, action)
       val observer = if (observers.isEmpty) { updateObserver } else {
         new CompositeObserver(observers.map(_.apply(path)).+:(updateObserver))
       }
-      (path, createWriter(path, partition, observer))
+      createWriter(path, partition, observer)
     }
 
-    targetFileSize match {
-      case None => pathAndWriter._2
-      case Some(s) => new ChunkedFileSystemWriter(fs, Iterator.continually(pathAndWriter), sizer.estimator(s))
+    sizer.targetSize match {
+      case None => newWriter()
+      case Some(s) => new ChunkedFileSystemWriter(Iterator.continually(newWriter()), sizer.estimator(s))
     }
   }
 }
@@ -345,47 +341,54 @@ object FileSystemStorage {
       * @param feature feature
       */
     def write(feature: SimpleFeature): Unit
+
+    /**
+     * Gets the size of the data written so far, in bytes. May not be accurate until the writer is
+     * closed, due to buffering, etc
+     *
+     * @return
+     */
+    def size: Long
   }
 
   /**
    * Writes files up to a given size, then starts a new file
    *
-   * @param fs file system
    * @param writers iterator of files to write
    * @param estimator target file size estimator
    */
-  private class ChunkedFileSystemWriter(
-      fs: ObjectStore,
-      writers: Iterator[(URI, FileSystemWriter)],
-      estimator: UpdatingFileSizeEstimator
-    ) extends FileSystemWriter {
-
-    private var count = 0L // number of features written
-    private var total = 0L // sum size of all finished chunks
-    private var remaining = estimator.estimate(0L)
+  class ChunkedFileSystemWriter(writers: Iterator[FileSystemWriter], estimator: UpdatingFileSizeEstimator)
+      extends FileSystemWriter {
 
-    private var path: URI = _
+    private var totalCount = 0L // total number of features written across all chunks
+    private var totalBytes = 0L // sum size of all finished chunks
+    private var remaining = estimator.estimate(0L)
     private var writer: FileSystemWriter = _
 
     override def write(feature: SimpleFeature): Unit = {
       if (writer == null) {
-        val (path, writer) = writers.next()
-        this.path = path
-        this.writer = writer
+        writer = writers.next()
       }
       writer.write(feature)
-      count += 1
+      totalCount += 1
       remaining -= 1
       if (remaining == 0) {
-        writer.close()
-        writer = null
-        // adjust our estimate to account for the actual bytes written
-        total += fs.size(path)
-        estimator.update(total, count)
-        remaining = estimator.estimate(0L)
+        val dataSize = writer.size
+        if (estimator.done(dataSize)) {
+          writer.close()
+          totalBytes += writer.size // re-calculate now that writer is closed, so we get the final, accurate size
+          writer = null
+          // adjust our estimate to account for the actual bytes written
+          estimator.update(totalBytes, totalCount)
+          remaining = estimator.estimate(0L)
+        } else {
+          remaining = math.max(100L, estimator.estimate(dataSize))
+        }
       }
     }
 
+    override def size: Long = totalBytes + Option(writer).fold(0L)(_.size)
+
     override def flush(): Unit = if (writer != null) { writer.flush() }
 
     override def close(): Unit = {

geomesa-fs/geomesa-fs-storage/geomesa-fs-storage-core/src/main/scala/org/locationtech/geomesa/fs/storage/core/StorageMetadata.scala

@@ -98,7 +98,7 @@ trait StorageMetadata extends Closeable {
    * Set a key-value pair
    *
    * @param key key
-   * @param value value
+   * @param value value - may be null
    */
   def set(key: String, value: String): Unit = throw new UnsupportedOperationException()
 }

geomesa-fs/geomesa-fs-storage/geomesa-fs-storage-core/src/main/scala/org/locationtech/geomesa/fs/storage/core/metadata/FileBasedMetadata.scala

@@ -62,7 +62,11 @@ class FileBasedMetadata(fs: ObjectStore, meta: Metadata, directory: URI)
 
   // note: this isn't synchronized across jvms
   override def set(key: String, value: String): Unit = FileBasedMetadata.synchronized {
-    kvs.put(key, value)
+    if (value == null) {
+      kvs.remove(key)
+    } else {
+      kvs.put(key, value)
+    }
     WithClose(fs.overwrite(metadataFilePath)) { out =>
       MetadataSerialization.serialize(out, meta.copy(config = kvs.asScala.toMap))
     }