Allow users to exclude fields from output

Excluding can be done per topic or for all topics

Author: blootsvoets

restructure.yml

@@ -93,6 +93,8 @@ format:
   # factory: org.radarbase.hdfs.format.FormatFactory
   # Additional format properties
   # properties: {}
+  # Do not write certain fields to file
+  # excludeFields: []
 
 # Worker settings. Each worker thread has its own cache and topic, so the
 # settings only apply to a single thread.
@@ -147,6 +149,11 @@ topics:
       enable: true
       # deduplicate this topic only using given fields.
       distinctFields: [key.sourceId, value.time]
+    # Do not write certain fields to file
+    # In this case, exclude user and project ID since they are always the same
+    excludeFields:
+      - key.userId
+      - key.projectId
   # topic name
   connect_fitbit_source2:
     # deduplicate this topic, regardless of the format settings

src/main/java/org/radarbase/output/config/TopicConfig.kt

@@ -15,6 +15,10 @@ data class TopicConfig(
      * [org.radarbase.output.path.FormattedPathFactory] format.
      */
     val pathProperties: Map<String, String> = emptyMap(),
+    /**
+     * Exclude given fields from output files.
+     */
+    val excludeFields: Set<String>? = null,
 ) {
     fun deduplication(deduplicationDefault: DeduplicationConfig): DeduplicationConfig =
         deduplication

src/main/java/org/radarbase/output/worker/FileCache.kt

@@ -63,12 +63,16 @@ class FileCache(
     private var lastUse: Long = 0
     private val hasError: AtomicBoolean = AtomicBoolean(false)
     private val deduplicate: DeduplicationConfig
+    private val excludeFields: Set<String>
 
     init {
         val topicConfig = factory.config.topics[topic]
         val defaultDeduplicate = factory.config.format.deduplication
         deduplicate = topicConfig?.deduplication(defaultDeduplicate) ?: defaultDeduplicate
 
+        val defaultExclude = factory.config.format.excludeFields
+        excludeFields = topicConfig?.excludeFields ?: defaultExclude
+
         this.tmpPath = createTempFile(tmpDir, fileName, ".tmp" + compression.extension)
     }
 
@@ -102,7 +106,7 @@ class FileCache(
 
         this.recordConverter = try {
             inputStream.reader().useSuspended { reader ->
-                converterFactory.converterFor(writer, record, fileIsNew, reader)
+                converterFactory.converterFor(writer, record, fileIsNew, reader, excludeFields)
             }
         } catch (ex: IOException) {
             coroutineScope {