refactor date conversion

Author: dszakallas

src/main/scala/ldbc/snb/datagen/transformation/TransformationStage.scala

@@ -4,7 +4,7 @@ import ldbc.snb.datagen.io.graphs.{GraphSink, GraphSource}
 import ldbc.snb.datagen.model
 import ldbc.snb.datagen.model.{BatchedEntity, Graph, Mode}
 import ldbc.snb.datagen.syntax._
-import ldbc.snb.datagen.transformation.transform.{ExplodeAttrs, ExplodeEdges, IrToRawTransform, RawToBiTransform, RawToInteractiveTransform}
+import ldbc.snb.datagen.transformation.transform.{ExplodeAttrs, ExplodeEdges, ConvertDates, RawToBiTransform, RawToInteractiveTransform}
 import ldbc.snb.datagen.util.{DatagenStage, Logging}
 import org.apache.spark.sql.{DataFrame, SparkSession}
 import shapeless._
@@ -44,11 +44,12 @@ object TransformationStage extends DatagenStage with Logging {
     GraphSource(model.graphs.Raw.graphDef, args.outputDir, "parquet").read
       .pipeFoldLeft(args.explodeAttrs.fork)((graph, _: Unit) => ExplodeAttrs.transform(graph))
       .pipeFoldLeft(args.explodeEdges.fork)((graph, _: Unit) => ExplodeEdges.transform(graph))
+      .pipe(ConvertDates.transform)
       .pipe[OutputTypes] { g =>
         args.mode match {
           case bi @ Mode.BI(_, _) => Inr(Inr(Inl(RawToBiTransform(bi, args.simulationStart, args.simulationEnd, args.keepImplicitDeletes).transform(g))))
           case interactive @ Mode.Interactive(_) => Inr(Inl(RawToInteractiveTransform(interactive, args.simulationStart, args.simulationEnd).transform(g)))
-          case Mode.Raw                          => Inl(IrToRawTransform.transform(g))
+          case Mode.Raw                          => Inl(g)
         }
       }
       .map(write)

src/main/scala/ldbc/snb/datagen/transformation/transform/IrToRawTransform.scala renamed to src/main/scala/ldbc/snb/datagen/transformation/transform/ConvertDates.scala

@@ -7,7 +7,7 @@ import org.apache.spark.sql.functions.lit
 import org.apache.spark.sql.types.{DateType, TimestampType}
 import shapeless._
 
-object IrToRawTransform extends Transform[Mode.Raw.type, Mode.Raw.type] {
+object ConvertDates extends Transform[Mode.Raw.type, Mode.Raw.type] {
 
   def convertDates(tpe: EntityType, df: DataFrame): DataFrame = {
     tpe match {

src/main/scala/ldbc/snb/datagen/transformation/transform/RawToBiTransform.scala

@@ -63,9 +63,6 @@ case class RawToBiTransform(mode: BI, simulationStart: Long, simulationEnd: Long
     }
 
     val entities = input.entities
-      .map { case (tpe, v) =>
-        tpe -> IrToRawTransform.convertDates(tpe, v)
-      }
       .map {
         case (tpe, v) if tpe.isStatic => tpe -> BatchedEntity(v, None, None)
         case (tpe, v) =>

src/main/scala/ldbc/snb/datagen/transformation/transform/RawToInteractiveTransform.scala

@@ -18,9 +18,6 @@ case class RawToInteractiveTransform(mode: Mode.Interactive, simulationStart: Lo
 
   override def transform(input: In): Out = {
     val entities = input.entities
-      .map { case (tpe, v) =>
-        tpe -> IrToRawTransform.convertDates(tpe, v)
-      }
       .map { case (tpe, v) =>
         tpe -> RawToInteractiveTransform.snapshotPart(tpe, v, bulkLoadThreshold, filterDeletion = true)
       }