Issue #441: Fix data change on optimize

core/src/main/scala/io/qbeast/core/model/IndexFileBuilder.scala

@@ -23,6 +23,7 @@ import scala.collection.immutable
 final class IndexFileBuilder {
   private var path: Option[String] = None
   private var size: Long = 0L
+  private var dataChange: Boolean = true
   private var modificationTime: Long = 0L
   private var revisionId: RevisionID = 0L
   private val blocks = immutable.Seq.newBuilder[VolatileBlock]
@@ -67,6 +68,19 @@ final class IndexFileBuilder {
     this
   }
 
+  /**
+   * Sets the data change.
+   *
+   * @param dataChange
+   *   whether this index file represents a data change
+   * @return
+   *   this instance
+   */
+  def setDataChange(dataChange: Boolean): IndexFileBuilder = {
+    this.dataChange = dataChange
+    this
+  }
+
   /**
    * Sets the modification time
    *
@@ -117,6 +131,7 @@ final class IndexFileBuilder {
     IndexFile(
       filePath,
       size,
+      dataChange,
       modificationTime,
       revisionId,
       blocks.result().map(_.toBlock(filePath)),

core/src/main/scala/io/qbeast/core/model/QbeastFiles.scala

@@ -23,8 +23,10 @@ sealed trait QbeastFile extends Serializable {
 
   def size: Long
 
+  def dataChange: Boolean
+
   override def toString: String = {
-    s"QbeastFile($path, $size)"
+    s"QbeastFile($path, $size, $dataChange)"
   }
 
 }
@@ -44,6 +46,7 @@ sealed trait QbeastFile extends Serializable {
 case class IndexFile(
     path: String,
     size: Long,
+    dataChange: Boolean,
     modificationTime: Long,
     revisionId: RevisionID,
     blocks: IISeq[Block],
@@ -84,7 +87,7 @@ case class IndexFile(
     val newBlocks = blocks.map { block =>
       if (cubeIds.contains(block.cubeId)) block.replicate() else block
     }
-    Some(IndexFile(path, size, newModificationTime, revisionId, newBlocks))
+    Some(IndexFile(path, size, dataChange, newModificationTime, revisionId, newBlocks))
   }
 
   override def toString: String = {
@@ -103,10 +106,11 @@ case class IndexFile(
  * @param deletionTimestamp
  *   the deletion timestamp
  */
-case class DeleteFile(path: String, size: Long, deletionTimestamp: Long) extends QbeastFile {
+case class DeleteFile(path: String, size: Long, dataChange: Boolean, deletionTimestamp: Long)
+    extends QbeastFile {
 
   override def toString: String = {
-    s"DeleteFile($path, $size, $deletionTimestamp)"
+    s"DeleteFile($path, $size, $dataChange, $deletionTimestamp)"
   }
 
 }

delta/src/main/scala/io/qbeast/spark/delta/DeltaMetadataWriter.scala

@@ -176,8 +176,8 @@ private[delta] case class DeltaMetadataWriter(
 
       // Execute write
       val (tableChanges, indexFiles, deleteFiles) = writer
-      val addFiles = indexFiles.map(DeltaQbeastFileUtils.toAddFile(dataChange = true))
-      val removeFiles = deleteFiles.map(DeltaQbeastFileUtils.toRemoveFile(dataChange = false))
+      val addFiles = indexFiles.map(DeltaQbeastFileUtils.toAddFile)
+      val removeFiles = deleteFiles.map(DeltaQbeastFileUtils.toRemoveFile)
 
       // Update Qbeast Metadata (replicated set, revision..)
       var actions = updateMetadata(txn, tableChanges, addFiles, removeFiles)
@@ -228,7 +228,10 @@ private[delta] case class DeltaMetadataWriter(
       .collect()
       .map(DeltaQbeastFileUtils.fromAddFile(dimensionCount))
       .flatMap(_.tryReplicateBlocks(deltaReplicatedSet))
-      .map(DeltaQbeastFileUtils.toAddFile(dataChange = false))
+      .map(file => {
+        val addFile = DeltaQbeastFileUtils.toAddFile(file)
+        addFile.copy(dataChange = false)
+      })
       .toSeq
   }
 

delta/src/main/scala/io/qbeast/spark/delta/DeltaQbeastFileUtils.scala

@@ -66,6 +66,7 @@ private[delta] object DeltaQbeastFileUtils {
       .setPath(addFile.path)
       .setSize(addFile.size)
       .setStats(stats)
+      .setDataChange(addFile.dataChange)
       .setModificationTime(addFile.modificationTime)
     addFile.getTag(TagUtils.revision) match {
       case Some(value) => builder.setRevisionId(value.toLong)
@@ -81,14 +82,12 @@ private[delta] object DeltaQbeastFileUtils {
   /**
    * Converts a given IndexFile instance to an AddFile instance.
    *
-   * @param dataChange
-   *   whether this index file represents a data change
    * @param indexFile
    *   the IndexFile instance
    * @return
    *   an AddFile instance
    */
-  def toAddFile(dataChange: Boolean)(indexFile: IndexFile): AddFile = {
+  def toAddFile(indexFile: IndexFile): AddFile = {
     val tags = Map(
       TagUtils.revision -> indexFile.revisionId.toString,
       TagUtils.blocks -> encodeBlocks(indexFile.blocks))
@@ -101,7 +100,7 @@ private[delta] object DeltaQbeastFileUtils {
       partitionValues = Map.empty[String, String],
       size = indexFile.size,
       modificationTime = indexFile.modificationTime,
-      dataChange = dataChange,
+      dataChange = indexFile.dataChange,
       stats = stats,
       tags = tags)
   }
@@ -110,22 +109,21 @@ private[delta] object DeltaQbeastFileUtils {
     DeleteFile(
       path = removeFile.path,
       size = removeFile.size.get,
+      dataChange = removeFile.dataChange,
       deletionTimestamp = removeFile.deletionTimestamp.get)
   }
 
   /**
    * Converts a given IndexFile instance to a RemoveFile instance.
    *
-   * @param dataChange
-   *   whether this file removal implies data change
    * @param deleteFile
    *   the DeleteFile instance
    */
-  def toRemoveFile(dataChange: Boolean)(deleteFile: DeleteFile): RemoveFile =
+  def toRemoveFile(deleteFile: DeleteFile): RemoveFile =
     RemoveFile(
       path = deleteFile.path,
       deletionTimestamp = Some(deleteFile.deletionTimestamp),
-      dataChange = dataChange,
+      dataChange = deleteFile.dataChange,
       partitionValues = Map.empty[String, String],
       size = Some(deleteFile.size))
 

delta/src/main/scala/io/qbeast/spark/delta/DeltaRollupDataWriter.scala

@@ -61,7 +61,7 @@ object DeltaRollupDataWriter extends RollupDataWriter with DeltaStatsCollectionU
     processStats(stats, statsTrackers, fileStatsTracker)
     filesAndStats
       .map(_._1)
-      .map(DeltaQbeastFileUtils.toAddFile(dataChange = true))
+      .map(DeltaQbeastFileUtils.toAddFile)
       .map(correctAddFileStats(fileStatsTracker))
       .map(DeltaQbeastFileUtils.fromAddFile(dimensionCount))
   }

src/main/scala/io/qbeast/table/IndexedTable.scala

@@ -515,14 +515,17 @@ private[table] class IndexedTableImpl(
               DeleteFile(
                 path = indexFile.path,
                 size = indexFile.size,
+                dataChange = false,
                 deletionTimestamp = currentTimeMillis())
             }
             .collect()
             .toIndexedSeq
           val data = snapshot.loadDataframeFromIndexFiles(indexFiles)
           val (dataExtended, tableChanges) =
             DoublePassOTreeDataAnalyzer.analyzeOptimize(data, indexStatus)
-          val addFiles = dataWriter.write(tableID, schema, dataExtended, tableChanges)
+          val addFiles = dataWriter
+            .write(tableID, schema, dataExtended, tableChanges)
+            .map(addFile => addFile.copy(dataChange = false))
           dataExtended.unpersist()
           (tableChanges, addFiles, deleteFiles)
         }

src/test/scala/io/qbeast/core/model/DenormalizedBlockTest.scala

@@ -86,8 +86,20 @@ class DenormalizedBlockTest extends QbeastIntegrationTestSpec {
 
       val fileSize = 10L
       val indexFilesDs = Vector(
-        IndexFile("f1.parquet", fileSize, 1, revision.revisionID, Vector(rootB1, c1B1)),
-        IndexFile("f2.parquet", fileSize, 2, revision.revisionID, Vector(rootB2, c1B2))).toDS
+        IndexFile(
+          "f1.parquet",
+          fileSize,
+          dataChange = true,
+          1,
+          revision.revisionID,
+          Vector(rootB1, c1B1)),
+        IndexFile(
+          "f2.parquet",
+          fileSize,
+          dataChange = true,
+          2,
+          revision.revisionID,
+          Vector(rootB2, c1B2))).toDS
 
       val denormalizedBlock =
         DenormalizedBlock.buildDataset(revision, cubeStatuses, indexFilesDs)

src/test/scala/io/qbeast/spark/delta/QbeastFileUtilsTest.scala

@@ -29,6 +29,7 @@ class QbeastFileUtilsTest extends QbeastIntegrationTestSpec {
     val indexFile = IndexFile(
       path = "path",
       size = 2L,
+      dataChange = true,
       modificationTime = 0L,
       revisionId = 1L,
       blocks = Seq(
@@ -40,7 +41,7 @@ class QbeastFileUtilsTest extends QbeastIntegrationTestSpec {
           1L,
           replicated = false)).toIndexedSeq)
 
-    val addFile = DeltaQbeastFileUtils.toAddFile(dataChange = true)(indexFile)
+    val addFile = DeltaQbeastFileUtils.toAddFile(indexFile)
     addFile.path shouldBe "path"
     addFile.size shouldBe 2L
     addFile.modificationTime shouldBe 0L
@@ -73,10 +74,9 @@ class QbeastFileUtilsTest extends QbeastIntegrationTestSpec {
   }
 
   it should "transform the DeleteFile to a RemoveFile" in withSpark { _ =>
-    val deleteFile = DeleteFile(path = "path", size = 2L, deletionTimestamp = 0L)
-
     val dataChange = false
-    val removeFile = DeltaQbeastFileUtils.toRemoveFile(dataChange = dataChange)(deleteFile)
+    val deleteFile = DeleteFile(path = "path", size = 2L, dataChange, deletionTimestamp = 0L)
+    val removeFile = DeltaQbeastFileUtils.toRemoveFile(deleteFile)
     removeFile.path shouldBe "path"
     removeFile.dataChange shouldBe dataChange
   }
@@ -93,13 +93,15 @@ class QbeastFileUtilsTest extends QbeastIntegrationTestSpec {
     val deleteFile = DeltaQbeastFileUtils.fromRemoveFile(removeFile)
     deleteFile.path shouldBe "path"
     deleteFile.size shouldBe 2L
+    deleteFile.dataChange shouldBe true
     deleteFile.deletionTimestamp shouldBe 0L
   }
 
   it should "be able to create a FileStatus from an IndexFile" in withSpark { _ =>
     val indexFile = IndexFile(
       path = "path",
       size = 2L,
+      dataChange = true,
       modificationTime = 0L,
       revisionId = 1L,
       blocks = Seq(
@@ -125,6 +127,7 @@ class QbeastFileUtilsTest extends QbeastIntegrationTestSpec {
     val indexFile = IndexFile(
       path = "path",
       size = 2L,
+      dataChange = true,
       modificationTime = 0L,
       revisionId = 1L,
       blocks = Seq(

src/test/scala/io/qbeast/spark/index/query/QueryExecutorTest.scala

@@ -243,6 +243,7 @@ class QueryExecutorTest extends QbeastIntegrationTestSpec with QueryTestSpec {
       val Seq(rb1, c1b1, c1b2) = IndexFile(
         path = tmpDir,
         size = 1L,
+        dataChange = true,
         modificationTime = 1L,
         revisionId = 1L,
         blocks = blocks).blocks