Keep dates in epocMillis in the output Row

Do not rely anymore on ElasticSearch auto-mapping feature and pass
dates as epocMillis without post-processing.

Cutting a post-processing unneeded step is going to be better
and safer, with less code and heap allocated.

Change-Id: Ibbc85b44c4e250fc63a792b41162394a38fd8a31

src/main/scala/com/gerritforge/analytics/engine/GerritAnalyticsTransformations.scala

diff --git a/src/main/scala/com/gerritforge/analytics/engine/GerritAnalyticsTransformations.scala b/src/main/scala/com/gerritforge/analytics/engine/GerritAnalyticsTransformations.scala
index 8aa89d2..a34e702 100644
--- a/src/main/scala/com/gerritforge/analytics/engine/GerritAnalyticsTransformations.scala
+++ b/src/main/scala/com/gerritforge/analytics/engine/GerritAnalyticsTransformations.scala
@@ -139,11 +139,6 @@
         .getOrElse(df)
     }
 
-
-    def convertDates(columnName: String)(implicit spark: SparkSession): DataFrame = {
-      df.withColumn(columnName, longDateToISOUdf(col(columnName)))
-    }
-
     def dropCommits(implicit spark: SparkSession): DataFrame = {
       df.drop("commits")
     }
@@ -159,7 +154,6 @@
       df
         .addOrganization()
         .handleAliases(aliasesDFMaybe)
-        .convertDates("last_commit_date")
         .dropCommits
     }
   }

src/test/scala/com/gerritforge/analytics/GerritAnalyticsTransformationsSpec.scala

diff --git a/src/test/scala/com/gerritforge/analytics/GerritAnalyticsTransformationsSpec.scala b/src/test/scala/com/gerritforge/analytics/GerritAnalyticsTransformationsSpec.scala
index 31d37c4..187b5ce 100644
--- a/src/test/scala/com/gerritforge/analytics/GerritAnalyticsTransformationsSpec.scala
+++ b/src/test/scala/com/gerritforge/analytics/GerritAnalyticsTransformationsSpec.scala
@@ -202,13 +202,14 @@
 
   it should "return correct columns when alias DF is not defined" in {
     import spark.implicits._
-    val inputSampleDF = sc.parallelize(Seq(
-      ("author_name", "email@mail.com", "an_organization")
-    )).toDF("author", "email", "organization")
+    val expectedTuple = ("author_name", "email@mail.com", "an_organization")
+    val inputSampleDF = sc.parallelize(Seq(expectedTuple)).toDF("author", "email", "organization")
+    val expectedRow = Row.fromTuple(expectedTuple)
 
     val df = inputSampleDF.handleAliases(None)
 
     df.schema.fields.map(_.name) should contain allOf("author", "email", "organization")
+    df.collect().head should be(expectedRow)
   }
 
   it should "lowercase aliased organizations" in {
@@ -265,27 +266,6 @@
     )
   }
 
-  "convertDates" should "process specific column from Long to ISO date" in {
-    // some notable dates converted in UnixMillisecs and ISO format
-    val DATES = Map(
-      0L -> "1970-01-01T00:00:00Z",
-      1500000000000L -> "2017-07-14T02:40:00Z",
-      1600000000000L -> "2020-09-13T12:26:40Z")
-    import sql.implicits._
-    val df = sc.parallelize(Seq(
-      ("a", 0L, 1),
-      ("b", 1500000000000L, 2),
-      ("c", 1600000000000L, 3))).toDF("name", "date", "num")
-
-    val converted = df
-      .convertDates("date")
-
-    converted.collect should contain allOf(
-      Row("a", DATES(0), 1),
-      Row("b", DATES(1500000000000L), 2),
-      Row("c", DATES(1600000000000L), 3)
-    )
-  }
 
   "extractCommitsPerProject" should "generate a Dataset with the all the SHA of commits with associated project" in {
     import sql.implicits._

src/test/scala/com/gerritforge/analytics/engine/events/GerritEventsTransformationsSpec.scala

diff --git a/src/test/scala/com/gerritforge/analytics/engine/events/GerritEventsTransformationsSpec.scala b/src/test/scala/com/gerritforge/analytics/engine/events/GerritEventsTransformationsSpec.scala
index bde0705..05b98dd 100644
--- a/src/test/scala/com/gerritforge/analytics/engine/events/GerritEventsTransformationsSpec.scala
+++ b/src/test/scala/com/gerritforge/analytics/engine/events/GerritEventsTransformationsSpec.scala
@@ -164,21 +164,20 @@
         ("stefano_alias", "stefano@galarraga-org.com", "")
       )).toDF("author", "email", "organization")
 
-      val expectedDate : ZonedDateTime = ZonedDateTime.now(ZoneId.of("UTC")).`with`(MILLI_OF_SECOND, 0).`with`(NANO_OF_SECOND, 0)
+      val expectedDate = System.currentTimeMillis
 
       val analyticsJobOutput =
         sc.parallelize(Seq(
-          "project1" -> UserActivitySummary(2018, 1, 20, 10, "Stefano", "stefano@galarraga-org.com", 1, 2, 1, 10, 4, Array(CommitInfo("sha1", expectedDate.toInstant.toEpochMilli, false)),
-            expectedDate.toInstant.toEpochMilli, false)
+          "project1" -> UserActivitySummary(2018, 1, 20, 10, "Stefano", "stefano@galarraga-org.com", 1, 2, 1, 10, 4, Array(CommitInfo("sha1", expectedDate, false)),
+            expectedDate, false)
         ))
           .asEtlDataFrame(sql)
           .addOrganization()
           .handleAliases(Some(aliasDF))
-          .convertDates("last_commit_date")
           .dropCommits
 
       val expected = sc.parallelize(Seq(
-        ("project1", "stefano_alias", "stefano@galarraga-org.com", 2018, 1, 20, 10, 2, 1, 10, 4, 1, expectedDate.format(DateTimeFormatter.ISO_OFFSET_DATE_TIME), false, "galarraga-org")
+        ("project1", "stefano_alias", "stefano@galarraga-org.com", 2018, 1, 20, 10, 2, 1, 10, 4, 1, expectedDate, false, "galarraga-org")
       )).toDF("project", "author", "email", "year", "month", "day", "hour", "num_files", "num_distinct_files",
         "added_lines", "deleted_lines", "num_commits", "last_commit_date", "is_merge", "organization")