Added docs and clarified code

blootsvoets · blootsvoets · commit f6dad313d782 · 2022-08-29T11:58:08.000+02:00
diff --git a/README.md b/README.md
@@ -151,6 +151,52 @@ Secrets can be provided as environment variables as well:
 
 Replace `SOURCE` with `TARGET` in the variables above to configure the target storage.
 
+### Path format
+
+The output path at the target storage is determined by the path format. The class that handles path
+output by default is the `org.radarbase.output.path.FormattedPathFactory`. The default format is
+```
+${projectId}/${userId}/${topic}/${filename}
+```
+Each format parameter is enclosed by a dollar sign with curly brackets.
+
+The full set of parameters is listed here:
+```yaml
+paths:
+  # Input directories in source storage
+  inputs:
+    - /testIn
+  # Temporary directory for local file processing.
+  temp: ./output/+tmp
+  # Output directory in target storage
+  output: /output
+  # Output path construction factory
+  factory: org.radarbase.output.path.FormattedPathFactory
+  # Additional properties
+  # properties:
+  #   format: ${projectId}/${userId}/${topic}/${time:mm}/${time:YYYYmmDD_HH'00'}${attempt}${extension}
+  #   plugins: fixed time key value org.example.plugin.MyPathPlugin
+```
+
+The FormattedPathFactory can use multiple plugins to format paths based on a given record.
+The `fixed` plugin has a number of fixed parameters that can be used:
+
+| Parameter | Description                                                             |
+|-----------|-------------------------------------------------------------------------|
+| projectId | record project ID                                                       |
+| userId    | record user ID                                                          |
+| sourceId  | record source ID                                                        |
+| topic     | Kafka topic                                                             | 
+| filename  | default time binning with attempt suffix and file extension             |
+| attempt   | attempt suffix for if a file with an incompatible format already exists |
+| extension | file extension                                                          |
+
+At least `filename` should be used, or a combination of `attempt` and `extension`.
+
+Then there are also plugins that take their own format. The `time` plugin formats a parameter according to the record time. It takes parameters with format `time:<date format>` where `<date format>` should be replaced by a [Java date format](https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/time/format/DateTimeFormatter.html), such as `YYYY-mm-dd`. The plugin tries to use the following time fields, in this order: a double `time` in the value struct, `timeStart` double or `start` long in the key struct, `dateTime` string in the value struct, `date` string in the value struct, `timeReceived` double in the value struct or `timeCompleted` double in the value struct. The first valid value used. If no valid time values are found, `unknown-date` is returned.
+
+The `key` and `value` plugins read values from the key or value structs of a given record. For example, parameter `value:color.red` will attempt to read the value struct, finding first the `color` field and then the enclosed `red` field. If no such value exists, `unknown-value` will be used in the format.
+
 ### Cleaner
 
 Source files can be automatically be removed by a cleaner process. This checks whether the file has already been extracted and is older than a configured age. This feature is not enabled by default. It can be configured in the `cleaner` configuration section:
@@ -199,3 +245,5 @@ To implement alternative storage paths, storage drivers or storage formats, put
 | `compression: factory: ...` | `org.radarbase.output.compression.CompressionFactory` | Factory class to use for data compression. | CompressionFactory   |
 
 The respective `<type>: properties: {}` configuration parameters can be used to provide custom configuration of the factory. This configuration will be passed to the `Plugin#init(Map<String, String>)` method.
+
+By adding additional path format plugins to the classpath, the path format of FormattedPathFactory may be expanded with different parameters or lookup engines.
diff --git a/src/main/java/org/radarbase/output/path/FixedPathFormatterPlugin.kt b/src/main/java/org/radarbase/output/path/FixedPathFormatterPlugin.kt
@@ -3,21 +3,30 @@ package org.radarbase.output.path
 import org.radarbase.output.path.RecordPathFactory.Companion.sanitizeId
 
 class FixedPathFormatterPlugin : PathFormatterPlugin() {
-    override val allowedFormats: String = lookupTable.keys.joinToString(separator = ", ")
+    override val allowedFormats: String = allowedParamNames.joinToString(separator = ", ")
 
-    override fun createLookupTable(
-        parameterNames: Set<String>,
-    ): Map<String, PathFormatParameters.() -> String> = lookupTable.filterKeys { it in parameterNames }
+    override fun lookup(parameterContents: String): PathFormatParameters.() -> String = when (parameterContents) {
+        "projectId" -> ({ sanitizeId(key.get("projectId"), "unknown-project") })
+        "userId" -> ({ sanitizeId(key.get("userId"), "unknown-user") })
+        "sourceId" -> ({ sanitizeId(key.get("sourceId"), "unknown-source") })
+        "topic" -> ({ topic })
+        "filename" -> ({ timeBin + attempt.toAttemptSuffix() + extension })
+        "attempt" -> ({ attempt.toAttemptSuffix() })
+        "extension" -> ({ extension })
+        else -> throw IllegalArgumentException("Unknown path parameter $parameterContents")
+    }
+
+    override fun extractParamContents(paramName: String): String? = paramName.takeIf { it in allowedParamNames }
 
     companion object {
-        val lookupTable = mapOf<String, PathFormatParameters.() -> String>(
-            "projectId" to { sanitizeId(key.get("projectId"), "unknown-project") },
-            "userId" to { sanitizeId(key.get("userId"), "unknown-user") },
-            "sourceId" to { sanitizeId(key.get("sourceId"), "unknown-source") },
-            "topic" to { topic },
-            "filename" to { timeBin + attempt.toAttemptSuffix() + extension },
-            "attempt" to { attempt.toAttemptSuffix() },
-            "extension" to { extension },
+        val allowedParamNames = setOf(
+            "projectId",
+            "userId",
+            "sourceId",
+            "topic",
+            "filename",
+            "attempt",
+            "extension",
         )
 
         private fun Int.toAttemptSuffix() = if (this == 0) "" else "_$this"
diff --git a/src/main/java/org/radarbase/output/path/FormattedPathFactory.kt b/src/main/java/org/radarbase/output/path/FormattedPathFactory.kt
@@ -35,12 +35,12 @@ open class FormattedPathFactory : RecordPathFactory() {
 
         format = properties["format"]
             ?: run {
-                logger.warn("Path format not provided, using {} instead", DEFAULT_FORMAT)
+                logger.warn("Path format not provided, using '{}' instead", DEFAULT_FORMAT)
                 DEFAULT_FORMAT
             }
         val pluginClassNames = properties["plugins"]
             ?: run {
-                logger.warn("Path format plugins not provided, using {} instead", DEFAULT_FORMAT_PLUGINS)
+                logger.warn("Path format plugins not provided, using '{}' instead", DEFAULT_FORMAT_PLUGINS)
                 DEFAULT_FORMAT_PLUGINS
             }
 
diff --git a/src/main/java/org/radarbase/output/path/KeyPathFormatterPlugin.kt b/src/main/java/org/radarbase/output/path/KeyPathFormatterPlugin.kt
@@ -1,18 +1,18 @@
 package org.radarbase.output.path
 
+import org.radarbase.output.path.RecordPathFactory.Companion.sanitizeId
 import org.radarbase.output.path.ValuePathFormatterPlugin.Companion.lookup
 
 class KeyPathFormatterPlugin : PathFormatterPlugin() {
+    override val prefix: String = "key"
+
     override val allowedFormats: String = "key:my.key.index"
 
-    override fun createLookupTable(
-        parameterNames: Set<String>
-    ): Map<String, (PathFormatParameters) -> String> = parameterNames
-        .filter { it.startsWith("key:") }
-        .associateWith { name ->
-            val index = name.removePrefix("key:").split('.')
-            return@associateWith { params ->
-                RecordPathFactory.sanitizeId(params.key.lookup(index), "unknown-key")
-            }
+    override fun lookup(parameterContents: String): PathFormatParameters.() -> String {
+        val index = parameterContents.split('.')
+        require(index.none { it.isBlank() }) { "Cannot format key record with index $parameterContents" }
+        return {
+            sanitizeId(key.lookup(index), "unknown-key")
         }
+    }
 }
diff --git a/src/main/java/org/radarbase/output/path/PathFormatter.kt b/src/main/java/org/radarbase/output/path/PathFormatter.kt
@@ -16,6 +16,7 @@
 
 package org.radarbase.output.path
 
+import org.slf4j.LoggerFactory
 import java.nio.file.Path
 import java.nio.file.Paths
 
@@ -33,7 +34,14 @@ class PathFormatter(
 
         parameterLookups = buildMap {
             plugins.forEach { plugin ->
-                putAll(plugin.createLookupTable(foundParameters))
+                putAll(
+                    try {
+                        plugin.createLookupTable(foundParameters)
+                    } catch (ex: IllegalArgumentException) {
+                        logger.error("Cannot parse path format {}, illegal format parameter found by plugin {}", format, plugin.javaClass, ex)
+                        throw ex
+                    }
+                )
             }
         }
         val unsupportedParameters = foundParameters - parameterLookups.keys
@@ -61,4 +69,8 @@ class PathFormatter(
 
         return Paths.get(path)
     }
+
+    companion object {
+        private val logger = LoggerFactory.getLogger(PathFormatter::class.java)
+    }
 }
diff --git a/src/main/java/org/radarbase/output/path/PathFormatterPlugin.kt b/src/main/java/org/radarbase/output/path/PathFormatterPlugin.kt
@@ -3,9 +3,47 @@ package org.radarbase.output.path
 import org.radarbase.output.Plugin
 
 abstract class PathFormatterPlugin : Plugin {
+    /**
+     * Prefix for parameter names covered by this plugin. If null, [extractParamContents] must be
+     * overridden to cover only supported parameters.
+     */
+    open val prefix: String? = null
+
+    /** Textual format of formats allowed to be represented. */
     abstract val allowedFormats: String
 
-    abstract fun createLookupTable(
-        parameterNames: Set<String>
-    ): Map<String, PathFormatParameters.() -> String>
+    /**
+     * Create a lookup table from parameter names to
+     * its value for a given record. Only parameter names supported by this plugin will be mapped.
+     * @throws IllegalArgumentException if any of the parameter contents are invalid.
+     */
+    fun createLookupTable(
+        parameterNames: Collection<String>
+    ): Map<String, PathFormatParameters.() -> String> = buildMap {
+        parameterNames.forEach { paramName ->
+            val paramContents = extractParamContents(paramName)
+            if (paramContents != null) {
+                put(paramName, lookup(paramContents))
+            }
+        }
+    }
+
+    /**
+     * Validate a parameter name and extract its contents to use in the lookup.
+     *
+     * @return name to use in the lookup or null if the parameter is not supported by this plugin
+     */
+    protected open fun extractParamContents(paramName: String): String? {
+        val prefixString = prefix?.let { "$it:" } ?: return null
+        if (!paramName.startsWith(prefixString)) return null
+        val parameterContents = paramName.removePrefix(prefixString).trim()
+        require(parameterContents.isNotEmpty()) { "Parameter contents of '$paramName' are empty" }
+        return parameterContents
+    }
+
+    /**
+     * Create a lookup function from a record to formatted value, based on parameter contents.
+     * @throws IllegalArgumentException if the parameter contents are invalid.
+     */
+    protected abstract fun lookup(parameterContents: String): PathFormatParameters.() -> String
 }
diff --git a/src/main/java/org/radarbase/output/path/TimePathFormatterPlugin.kt b/src/main/java/org/radarbase/output/path/TimePathFormatterPlugin.kt
@@ -5,23 +5,19 @@ import java.time.ZoneOffset
 import java.time.format.DateTimeFormatter
 
 class TimePathFormatterPlugin : PathFormatterPlugin() {
+    override val prefix: String = "time"
+
     override val allowedFormats: String = "time:YYYY-mm-dd"
 
-    override fun createLookupTable(
-        parameterNames: Set<String>
-    ): Map<String, PathFormatParameters.() -> String> {
-        return parameterNames
-            .filter { it.startsWith("time:") }
-            .associateWith { p ->
-                val dateFormatter = DateTimeFormatter
-                    .ofPattern(p.removePrefix("time:"))
-                    .withZone(ZoneOffset.UTC)
-                return@associateWith {
-                    sanitizeId(
-                        time?.let { dateFormatter.format(it) },
-                        "unknown-time",
-                    )
-                }
-            }
+    override fun lookup(parameterContents: String): PathFormatParameters.() -> String {
+        val dateFormatter = DateTimeFormatter
+            .ofPattern(parameterContents)
+            .withZone(ZoneOffset.UTC)
+        return {
+            sanitizeId(
+                time?.let { dateFormatter.format(it) },
+                "unknown-time",
+            )
+        }
     }
 }
diff --git a/src/main/java/org/radarbase/output/path/ValuePathFormatterPlugin.kt b/src/main/java/org/radarbase/output/path/ValuePathFormatterPlugin.kt
@@ -4,18 +4,17 @@ import org.apache.avro.generic.GenericRecord
 import org.radarbase.output.path.RecordPathFactory.Companion.sanitizeId
 
 class ValuePathFormatterPlugin : PathFormatterPlugin() {
+    override val prefix: String = "value"
+
     override val allowedFormats: String = "value:my.value.index"
 
-    override fun createLookupTable(
-        parameterNames: Set<String>
-    ): Map<String, (PathFormatParameters) -> String> = parameterNames
-        .filter { it.startsWith("value:") }
-        .associateWith { name ->
-            val index = name.removePrefix("value:").split('.')
-            return@associateWith { params ->
-                sanitizeId(params.value.lookup(index), "unknown-value")
-            }
+    override fun lookup(parameterContents: String): PathFormatParameters.() -> String {
+        val index = parameterContents.split('.')
+        require(index.none { it.isBlank() }) { "Cannot format value record with index $parameterContents" }
+        return {
+            sanitizeId(value.lookup(index), "unknown-value")
         }
+    }
 
     companion object {
         fun GenericRecord.lookup(index: List<String>): Any? =

Original file line number	Diff line number	Diff line change
`@@ -35,12 +35,12 @@ open class FormattedPathFactory : RecordPathFactory() {`
`35`	`35`
`36`	`36`	`format = properties["format"]`
`37`	`37`	`?: run {`
`38`		`- logger.warn("Path format not provided, using {} instead", DEFAULT_FORMAT)`
	`38`	`+ logger.warn("Path format not provided, using '{}' instead", DEFAULT_FORMAT)`
`39`	`39`	`DEFAULT_FORMAT`
`40`	`40`	`}`
`41`	`41`	`val pluginClassNames = properties["plugins"]`
`42`	`42`	`?: run {`
`43`		`- logger.warn("Path format plugins not provided, using {} instead", DEFAULT_FORMAT_PLUGINS)`
	`43`	`+ logger.warn("Path format plugins not provided, using '{}' instead", DEFAULT_FORMAT_PLUGINS)`
`44`	`44`	`DEFAULT_FORMAT_PLUGINS`
`45`	`45`	`}`
`46`	`46`