remove unnecessary entities from deletes and add cmdline flag

Author: dszakallas

src/main/scala/ldbc/snb/datagen/spark/LdbcDatagen.scala

@@ -20,6 +20,7 @@ object LdbcDatagen extends SparkApp {
     bulkloadPortion: Double = 0.1,
     explodeEdges: Boolean = false,
     explodeAttrs: Boolean = false,
+    keepImplicitDeletes: Boolean = false,
     mode: String = "raw",
     batchPeriod: String = "day",
     numThreads: Option[Int] = None,
@@ -83,6 +84,12 @@ object LdbcDatagen extends SparkApp {
         .action((x, c) => args.format.set(c)(x))
         .text("Output format. Currently, Spark Datasource formats are supported, such as 'csv', 'parquet' or 'orc'.")
 
+      opt[Unit]("keep-implicit-deletes")
+        .action((x, c) => args.keepImplicitDeletes.set(c)(true))
+        .text("Keep implicit deletes. Only applicable to BI mode. By default the BI output doesn't contain dynamic entities that" +
+          "without the explicitlyDeleted attribute and filters dynamic entities where explicitlyDeleted is false. " +
+          "Setting this flag retains all deletes.")
+
       opt[Map[String,String]]("format-options")
         .action((x, c) => args.formatOptions.set(c)(x))
         .text("Output format options specified as key=value1[,key=value...]. See format options for specific formats " +
@@ -116,6 +123,7 @@ object LdbcDatagen extends SparkApp {
       outputDir = args.outputDir,
       explodeEdges = args.explodeEdges,
       explodeAttrs = args.explodeAttrs,
+      keepImplicitDeletes = args.keepImplicitDeletes,
       simulationStart = Dictionaries.dates.getSimulationStart,
       simulationEnd = Dictionaries.dates.getSimulationEnd,
       mode = args.mode match {

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

@@ -18,6 +18,7 @@ object TransformationStage extends SparkApp with Logging {
     outputDir: String = "out",
     explodeEdges: Boolean = false,
     explodeAttrs: Boolean = false,
+    keepImplicitDeletes: Boolean = false,
     simulationStart: Long = 0,
     simulationEnd: Long = 0,
     mode: Mode = Mode.Raw,
@@ -110,7 +111,7 @@ object TransformationStage extends SparkApp with Logging {
       .pipe[OutputTypes] {
         g =>
           args.mode match {
-            case bi@Mode.BI(_, _) => Inr(Inr(Inl(RawToBiTransform(bi, args.simulationStart, args.simulationEnd).transform(g))))
+            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(g)
           }

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

@@ -8,7 +8,7 @@ import ldbc.snb.datagen.util.Logging
 import org.apache.spark.sql.{Column, DataFrame}
 import org.apache.spark.sql.functions._
 
-case class RawToBiTransform(mode: BI, simulationStart: Long, simulationEnd: Long) extends Transform[Mode.Raw.type, Mode.BI] with Logging {
+case class RawToBiTransform(mode: BI, simulationStart: Long, simulationEnd: Long, keepImplicitDeletes: Boolean) extends Transform[Mode.Raw.type, Mode.BI] with Logging {
   log.debug(s"BI Transformation parameters: $mode")
 
   val bulkLoadThreshold = Interactive.calculateBulkLoadThreshold(mode.bulkloadPortion, simulationStart, simulationEnd)
@@ -52,7 +52,7 @@ case class RawToBiTransform(mode: BI, simulationStart: Long, simulationEnd: Long
       val idColumns = tpe.primaryKey.map(qcol)
       df
         .filter(inBatch($"deletionDate", batchStart, batchEnd))
-        .filter(if (df.columns.contains("explicitlyDeleted")) col("explicitlyDeleted") else lit(false))
+        .filter(if (df.columns.contains("explicitlyDeleted")) col("explicitlyDeleted") else lit(true))
         .pipe(batched)
         .select(Seq($"delete_batch_id".as("batch_id"), $"deletionDate") ++ idColumns: _*)
         .repartitionByRange($"batch_id")
@@ -65,7 +65,10 @@ case class RawToBiTransform(mode: BI, simulationStart: Long, simulationEnd: Long
         tpe -> BatchedEntity(
           Interactive.snapshotPart(tpe, v, bulkLoadThreshold, filterDeletion = false),
           Some(Batched(insertBatchPart(tpe, v, bulkLoadThreshold, simulationEnd), Seq("batch_id"))),
-          Some(Batched(deleteBatchPart(tpe, v, bulkLoadThreshold, simulationEnd), Seq("batch_id")))
+          if (keepImplicitDeletes || v.columns.contains("explicitlyDeleted"))
+            Some(Batched(deleteBatchPart(tpe, v, bulkLoadThreshold, simulationEnd), Seq("batch_id")))
+          else
+            None
       )
     }
     Graph[Mode.BI, DataFrame](isAttrExploded = input.isAttrExploded, isEdgesExploded = input.isEdgesExploded, mode, entities)