fix(downsample): remove Option wrapper around export values

core/src/main/resources/filodb-defaults.conf

@@ -588,6 +588,7 @@ filodb {
 
     data-export {
       enabled = false
+      log-all-row-errors = true
       parallelism = 10
       # Catalog under Unified Catalog
       catalog = ""

spark-jobs/src/main/scala/filodb/downsampler/chunk/BatchExporter.scala

@@ -1,7 +1,7 @@
 package filodb.downsampler.chunk
 
 import java.security.MessageDigest
-import java.time.{Instant, LocalDateTime, ZoneId}
+import java.time.{Instant, ZoneId}
 
 import kamon.Kamon
 import org.apache.spark.rdd.RDD
@@ -37,7 +37,9 @@ case class ExportTableConfig(tableName: String,
 case class ExportRowData(metric: String,
                          labels: collection.Map[String, String],
                          epoch_timestamp: Long,
-                         timestamp: LocalDateTime,
+                         // IMPORTANT: a Spark-compatible value must be used here (something
+                         //   like LocalDateTime will throw exceptions).
+                         timestamp: java.sql.Timestamp,
                          value: Double,
                          year: Int,
                          month: Int,
@@ -55,6 +57,8 @@ case class BatchExporter(downsamplerSettings: DownsamplerSettings, userStartTime
 
   @transient lazy val numRowsExportPrepped = Kamon.counter("num-rows-export-prepped").withoutTags()
 
+  @transient lazy val numRowExportPrepErrors = Kamon.counter("num-row-export-prep-errors").withoutTags()
+
   val keyToRules = downsamplerSettings.exportKeyToRules
 
   /**
@@ -131,9 +135,21 @@ case class BatchExporter(downsamplerSettings: DownsamplerSettings, userStartTime
         getExportData(part, partKeyMap, rule.get)
       }
       .map { exportData =>
-        numRowsExportPrepped.increment()
-        exportDataToRow(exportData, exportTableConfig)
+        try {
+          val row = exportDataToRow(exportData, exportTableConfig)
+          numRowsExportPrepped.increment()
+          Some(row)
+        } catch {
+          case t: Throwable =>
+            if (downsamplerSettings.logAllRowErrors) {
+              DownsamplerContext.dsLogger.error(s"error during exportDataToRow: $exportData", t)
+            }
+            numRowExportPrepErrors.increment()
+            None
+        }
       }
+      .filter(_.isDefined)
+      .map(_.get)
   }
 
   /**
@@ -143,7 +159,11 @@ case class BatchExporter(downsamplerSettings: DownsamplerSettings, userStartTime
   private def exportDataToRow(exportData: ExportRowData, exportTableConfig: ExportTableConfig): Row = {
     val dataSeq = new mutable.ArrayBuffer[Any](exportTableConfig.tableSchema.fields.length)
     // append all dynamic column values
-    exportTableConfig.labelColumnMapping.foreach { pair => dataSeq.append(exportData.labels.get(pair._1)) }
+    exportTableConfig.labelColumnMapping.foreach { pair =>
+      val labelValue = exportData.labels.get(pair._1)
+      assert(labelValue.isDefined, s"${pair._1} label was expected but not found: ${exportData.labels}")
+      dataSeq.append(labelValue.get)
+    }
     // append all fixed column values
     dataSeq.append(
       exportData.metric,
@@ -311,11 +331,12 @@ case class BatchExporter(downsamplerSettings: DownsamplerSettings, userStartTime
       val metric = labels(LABEL_NAME)
       // to compute YYYY, MM, dd, hh
       // to compute readable timestamp from unix timestamp
-      val timestamp = Instant.ofEpochMilli(epoch_timestamp).atZone(ZoneId.systemDefault()).toLocalDateTime()
-      val year = timestamp.getYear()
-      val month = timestamp.getMonthValue()
-      val day = timestamp.getDayOfMonth()
-      val hour = timestamp.getHour()
+      val dateTime = Instant.ofEpochMilli(epoch_timestamp).atZone(ZoneId.systemDefault()).toLocalDateTime()
+      val timestamp = java.sql.Timestamp.valueOf(dateTime)
+      val year = dateTime.getYear()
+      val month = dateTime.getMonthValue()
+      val day = dateTime.getDayOfMonth()
+      val hour = dateTime.getHour()
       ExportRowData(metric, labels, epoch_timestamp, timestamp, value, year, month, day, hour)
     }
   }

spark-jobs/src/main/scala/filodb/downsampler/chunk/DownsamplerSettings.scala

@@ -154,6 +154,8 @@ class DownsamplerSettings(conf: Config = ConfigFactory.empty()) extends Serializ
 
   @transient lazy val exportDatabase = downsamplerConfig.getString("data-export.database")
 
+  @transient lazy val logAllRowErrors = downsamplerConfig.getBoolean("data-export.log-all-row-errors")
+
   /**
    * Two conditions should satisfy for eligibility:
    * (a) If allow list is nonEmpty partKey should match a filter in the allow list.

spark-jobs/src/test/scala/filodb/downsampler/DownsamplerMainSpec.scala

@@ -59,15 +59,14 @@ class DownsamplerMainSpec extends AnyFunSpec with Matchers with BeforeAndAfterAl
        |    "enabled": ${exportToFile.isDefined},
        |    "catalog": "",
        |    "database": "",
-       |    "key-labels": [],
        |    "format": "iceberg",
+       |    "key-labels": [_ns_],
        |    "groups": [
        |      {
-       |        "key": [],
+       |        "key": [my_ns],
        |        "table": "",
        |        "table-path": "${exportToFile.getOrElse("")}",
        |        "label-column-mapping": [
-       |         "_ws_", "workspace", "NOT NULL",
        |         "_ns_", "namespace", "NOT NULL"
        |        ]
        |        "partition-by-columns": ["namespace"]