Use update() for retrieving Snapshot and add Timestamp_NTZ type

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

@@ -92,6 +92,12 @@ object TimestampDataType extends OrderedDataType {
 
 }
 
+object TimestampNTZType extends OrderedDataType {
+  override def name: String = "TimestampNTZType"
+  override val ordering: Numeric[Any] = implicitly[Numeric[Long]].asInstanceOf[Numeric[Any]]
+
+}
+
 object DateDataType extends OrderedDataType {
   override def name: String = "DateDataType"
   override val ordering: Numeric[Any] = implicitly[Numeric[Long]].asInstanceOf[Numeric[Any]]

core/src/main/scala/io/qbeast/core/transform/LinearTransformation.scala

@@ -69,8 +69,9 @@ case class LinearTransformation(
 
   /**
    * Merges two transformations. The domain of the resulting transformation is the union of this
-   * and the other transformation. The range of the resulting transformation is the intersection of
-   * this and the other transformation, which can be a LinearTransformation or IdentityTransformation
+   * and the other transformation. The range of the resulting transformation
+   * is the intersection of this and the other transformation,
+   * which can be a LinearTransformation or IdentityTransformation
    * @param other
    * @return a new Transformation that contains both this and other.
    */

src/main/scala/io/qbeast/spark/QbeastTable.scala

@@ -34,7 +34,7 @@ class QbeastTable private (
   private def deltaLog: DeltaLog = DeltaLog.forTable(sparkSession, tableID.id)
 
   private def qbeastSnapshot: DeltaQbeastSnapshot =
-    delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+    delta.DeltaQbeastSnapshot(deltaLog.update())
 
   private def indexedTable: IndexedTable = indexedTableFactory.getIndexedTable(tableID)
 

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

@@ -104,7 +104,9 @@ private[delta] case class DeltaMetadataWriter(
 
     val cubeStrings = deltaReplicatedSet.map(_.string)
     val cubeBlocks =
-      deltaLog.unsafeVolatileSnapshot.allFiles
+      deltaLog
+        .update()
+        .allFiles
         .where(TagColumns.revision === lit(revision.revisionID.toString) &&
           TagColumns.cube.isInCollection(cubeStrings))
         .collect()

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

@@ -118,7 +118,7 @@ object OTreeIndex {
 
   def apply(spark: SparkSession, path: Path): OTreeIndex = {
     val deltaLog = DeltaLog.forTable(spark, path)
-    val unsafeVolatileSnapshot = deltaLog.unsafeVolatileSnapshot
+    val unsafeVolatileSnapshot = deltaLog.update()
     val tahoe = TahoeLogFileIndex(spark, deltaLog, path, unsafeVolatileSnapshot, Seq.empty, false)
     OTreeIndex(tahoe)
   }

src/main/scala/io/qbeast/spark/internal/commands/ConvertToQbeastCommand.scala

@@ -54,7 +54,7 @@ case class ConvertToQbeastCommand(
     val (fileFormat, tableId) = resolveTableFormat(spark)
 
     val deltaLog = DeltaLog.forTable(spark, tableId.table)
-    val unsafeVolatileSnapshot = deltaLog.unsafeVolatileSnapshot
+    val unsafeVolatileSnapshot = deltaLog.update()
     val qbeastSnapshot = DeltaQbeastSnapshot(unsafeVolatileSnapshot)
     val isQbeast = qbeastSnapshot.loadAllRevisions.nonEmpty
 
@@ -79,7 +79,7 @@ case class ConvertToQbeastCommand(
 
       // Convert delta to qbeast through metadata modification
       val tableID = QTableID(tableId.table)
-      val schema = deltaLog.unsafeVolatileSnapshot.schema
+      val schema = deltaLog.update().schema
 
       SparkDeltaMetadataManager.updateMetadataWithTransaction(tableID, schema) {
         val convRevision = stagingRevision(tableID, cubeSize, columnsToIndex)

src/main/scala/io/qbeast/spark/internal/sources/v2/QbeastWriteBuilder.scala

@@ -59,6 +59,8 @@ class QbeastWriteBuilder(
           // Passing the options in the query plan plus the properties
           // because columnsToIndex needs to be included in the contract
           val writeOptions = info.options().toMap ++ properties
+          // scalastyle:off
+          println("data schema " + data.schema)
           indexedTable.save(data, writeOptions, append)
         }
       }

src/main/scala/io/qbeast/spark/utils/SparkToQTypesUtils.scala

@@ -17,6 +17,7 @@ object SparkToQTypesUtils {
     case _: DecimalType => qmodel.DecimalDataType
     case _: TimestampType => qmodel.TimestampDataType
     case _: DateType => qmodel.DateDataType
+    case _: TimestampNTZType => qmodel.TimestampNTZType
     case _ => throw new RuntimeException(s"${sparkType.typeName} is not supported yet")
     // TODO add more types
   }

src/main/scala/org/apache/spark/sql/delta/DeltaStatsCollectionUtils.scala

@@ -28,7 +28,7 @@ trait DeltaStatsCollectionUtils {
       val deltaLog = DeltaLog.forTable(sparkSession, tableID.id)
       val metadata = deltaLog.unsafeVolatileMetadata
       val outputPath = deltaLog.dataPath
-      val deltaProtocol = deltaLog.unsafeVolatileSnapshot.protocol
+      val deltaProtocol = deltaLog.update().protocol
 
       val indexedCols = DeltaConfigs.DATA_SKIPPING_NUM_INDEXED_COLS.fromMetaData(metadata)
 

src/test/scala/io/qbeast/docs/DocumentationTests.scala

@@ -87,7 +87,7 @@ class DocumentationTests extends QbeastIntegrationTestSpec {
       val qbeast_df = spark.read.format("qbeast").load(qbeast_table_path)
 
       val deltaLog = DeltaLog.forTable(spark, qbeast_table_path)
-      val totalNumberOfFiles = deltaLog.unsafeVolatileSnapshot.allFiles.count()
+      val totalNumberOfFiles = deltaLog.update().allFiles.count()
 
       totalNumberOfFiles should be > 1L withClue
         "Total number of files in pushdown notebook changes to " + totalNumberOfFiles

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

@@ -20,7 +20,7 @@ class IndexStatusBuilderTest extends QbeastIntegrationTestSpec {
 
       val deltaLog = DeltaLog.forTable(spark, tmpDir)
       val indexStatus =
-        DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot).loadLatestIndexStatus
+        DeltaQbeastSnapshot(deltaLog.update()).loadLatestIndexStatus
 
       indexStatus.revision.revisionID shouldBe 1
       indexStatus.cubesStatuses.foreach(_._2.files.size shouldBe 1)
@@ -44,7 +44,7 @@ class IndexStatusBuilderTest extends QbeastIntegrationTestSpec {
       .save(tmpDir)
     val deltaLog = DeltaLog.forTable(spark, tmpDir)
     val firstIndexStatus =
-      DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot).loadLatestIndexStatus
+      DeltaQbeastSnapshot(deltaLog.update()).loadLatestIndexStatus
     data.write
       .format("qbeast")
       .mode("append")

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

@@ -45,17 +45,11 @@ class OTreeIndexTest extends QbeastIntegrationTestSpec {
 
     val deltaLog = DeltaLog.forTable(spark, tmpdir)
     val tahoeFileIndex = {
-      TahoeLogFileIndex(
-        spark,
-        deltaLog,
-        deltaLog.dataPath,
-        deltaLog.unsafeVolatileSnapshot,
-        Seq.empty,
-        false)
+      TahoeLogFileIndex(spark, deltaLog, deltaLog.dataPath, deltaLog.update(), Seq.empty, false)
     }
     val oTreeIndex = new OTreeIndexTest(tahoeFileIndex)
 
-    val allFiles = deltaLog.unsafeVolatileSnapshot.allFiles.collect().map(_.path)
+    val allFiles = deltaLog.update().allFiles.collect().map(_.path)
 
     val matchFiles = oTreeIndex.matchingBlocks(Seq.empty, Seq.empty).map(_.path)
 
@@ -79,17 +73,13 @@ class OTreeIndexTest extends QbeastIntegrationTestSpec {
 
     val deltaLog = DeltaLog.forTable(spark, tmpdir)
     val tahoeFileIndex = {
-      TahoeLogFileIndex(
-        spark,
-        deltaLog,
-        deltaLog.dataPath,
-        deltaLog.unsafeVolatileSnapshot,
-        Seq.empty,
-        false)
+      TahoeLogFileIndex(spark, deltaLog, deltaLog.dataPath, deltaLog.update(), Seq.empty, false)
     }
     val oTreeIndex = new OTreeIndexTest(tahoeFileIndex)
 
-    oTreeIndex.inputFiles shouldBe deltaLog.unsafeVolatileSnapshot.allFiles
+    oTreeIndex.inputFiles shouldBe deltaLog
+      .update()
+      .allFiles
       .collect()
       .map(file => new Path(deltaLog.dataPath, file.path).toString)
   })
@@ -107,16 +97,10 @@ class OTreeIndexTest extends QbeastIntegrationTestSpec {
 
       val deltaLog = DeltaLog.forTable(spark, tmpdir)
       val tahoeFileIndex = {
-        TahoeLogFileIndex(
-          spark,
-          deltaLog,
-          deltaLog.dataPath,
-          deltaLog.unsafeVolatileSnapshot,
-          Seq.empty,
-          false)
+        TahoeLogFileIndex(spark, deltaLog, deltaLog.dataPath, deltaLog.update(), Seq.empty, false)
       }
       val oTreeIndex = new OTreeIndexTest(tahoeFileIndex)
-      val allFiles = deltaLog.unsafeVolatileSnapshot.allFiles.collect().map(_.path)
+      val allFiles = deltaLog.update().allFiles.collect().map(_.path)
 
       oTreeIndex.matchingBlocks(Seq.empty, Seq.empty).map(_.path).toSet shouldBe allFiles.toSet
     })
@@ -133,17 +117,11 @@ class OTreeIndexTest extends QbeastIntegrationTestSpec {
 
     val deltaLog = DeltaLog.forTable(spark, tmpdir)
     val tahoeFileIndex = {
-      TahoeLogFileIndex(
-        spark,
-        deltaLog,
-        deltaLog.dataPath,
-        deltaLog.unsafeVolatileSnapshot,
-        Seq.empty,
-        false)
+      TahoeLogFileIndex(spark, deltaLog, deltaLog.dataPath, deltaLog.update(), Seq.empty, false)
     }
     val oTreeIndex = new OTreeIndexTest(tahoeFileIndex)
 
-    val sizeInBytes = deltaLog.unsafeVolatileSnapshot.allFiles.collect().map(_.size).sum
+    val sizeInBytes = deltaLog.update().allFiles.collect().map(_.size).sum
     oTreeIndex.sizeInBytes shouldBe sizeInBytes
   })
 

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

@@ -36,7 +36,7 @@ class QbeastSnapshotTest extends QbeastIntegrationTestSpec {
           .save(tmpDir)
 
         val deltaLog = DeltaLog.forTable(spark, tmpDir)
-        val qbeastSnapshot = DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+        val qbeastSnapshot = DeltaQbeastSnapshot(deltaLog.update())
         val indexStatus = qbeastSnapshot.loadLatestIndexStatus
         val revision = indexStatus.revision
 
@@ -62,7 +62,7 @@ class QbeastSnapshotTest extends QbeastIntegrationTestSpec {
           .save(tmpDir)
 
         val deltaLog = DeltaLog.forTable(spark, tmpDir)
-        val qbeastSnapshot = DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+        val qbeastSnapshot = DeltaQbeastSnapshot(deltaLog.update())
         val columnTransformers = SparkRevisionFactory
           .createNewRevision(QTableID(tmpDir), df.schema, options)
           .columnTransformers
@@ -92,7 +92,7 @@ class QbeastSnapshotTest extends QbeastIntegrationTestSpec {
           .save(tmpDir)
 
         val deltaLog = DeltaLog.forTable(spark, tmpDir)
-        val qbeastSnapshot = DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+        val qbeastSnapshot = DeltaQbeastSnapshot(deltaLog.update())
         val timestamp = System.currentTimeMillis()
         qbeastSnapshot.loadRevisionAt(timestamp) shouldBe qbeastSnapshot.loadLatestRevision
 
@@ -117,7 +117,7 @@ class QbeastSnapshotTest extends QbeastIntegrationTestSpec {
           .save(tmpDir)
 
         val deltaLog = DeltaLog.forTable(spark, tmpDir)
-        val qbeastSnapshot = DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+        val qbeastSnapshot = DeltaQbeastSnapshot(deltaLog.update())
         an[AnalysisException] shouldBe thrownBy(
           qbeastSnapshot.loadRevisionAt(invalidRevisionTimestamp))
 
@@ -140,7 +140,7 @@ class QbeastSnapshotTest extends QbeastIntegrationTestSpec {
             .save(tmpDir)
 
           val deltaLog = DeltaLog.forTable(spark, tmpDir)
-          val qbeastSnapshot = DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+          val qbeastSnapshot = DeltaQbeastSnapshot(deltaLog.update())
           val builder =
             new IndexStatusBuilder(
               qbeastSnapshot,

src/test/scala/io/qbeast/spark/index/AnalyzeAndOptimizeTest.scala

@@ -55,7 +55,7 @@ class AnalyzeAndOptimizeTest
     qbeastTable.optimize()
 
     val deltaLog = DeltaLog.forTable(spark, tmpDir)
-    val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+    val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.update())
     val replicatedCubes = qbeastSnapshot.loadLatestIndexStatus.replicatedSet
 
     val announcedCubes = qbeastTable.analyze()
@@ -72,7 +72,7 @@ class AnalyzeAndOptimizeTest
         val announcedCubes = qbeastTable.analyze()
         qbeastTable.optimize()
         val deltaLog = DeltaLog.forTable(spark, tmpDir)
-        val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+        val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.update())
         val replicatedCubes =
           qbeastSnapshot.loadLatestIndexStatus.replicatedSet.map(_.string)
 

src/test/scala/io/qbeast/spark/index/CubeWeightsIntegrationTest.scala

@@ -40,7 +40,7 @@ class CubeWeightsIntegrationTest extends QbeastIntegrationTestSpec with PrivateM
             .save(tmpDir)
 
           val deltaLog = DeltaLog.forTable(spark, tmpDir)
-          val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+          val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.update())
           val commitLogWeightMap = qbeastSnapshot.loadLatestIndexStatus.cubesStatuses
 
           // commitLogWeightMap shouldBe weightMap
@@ -62,7 +62,7 @@ class CubeWeightsIntegrationTest extends QbeastIntegrationTestSpec with PrivateM
       .save(tmpDir)
 
     val deltaLog = DeltaLog.forTable(spark, tmpDir)
-    val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+    val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.update())
     val cubeWeights = qbeastSnapshot.loadLatestIndexStatus.cubesStatuses
 
     cubeWeights.values.foreach { case CubeStatus(_, weight, _, _) =>

src/test/scala/io/qbeast/spark/index/DataStagingTest.scala

@@ -21,7 +21,7 @@ class DataStagingTest
 
   def getQbeastSnapshot(spark: SparkSession, dir: String): DeltaQbeastSnapshot = {
     val deltaLog = DeltaLog.forTable(spark, dir)
-    DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+    DeltaQbeastSnapshot(deltaLog.update())
   }
 
   private val getCurrentStagingSize: PrivateMethod[Long] =

src/test/scala/io/qbeast/spark/index/IndexTest.scala

@@ -127,7 +127,7 @@ class IndexTest
           .save(tmpDir)
 
         val deltaLog = DeltaLog.forTable(spark, tmpDir)
-        val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+        val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.update())
 
         val offset = 0.5
         val appendData = df
@@ -190,7 +190,7 @@ class IndexTest
 
         val deltaLog = DeltaLog.forTable(spark, tmpDir)
 
-        deltaLog.unsafeVolatileSnapshot.allFiles.collect() foreach (f =>
+        deltaLog.update().allFiles.collect() foreach (f =>
           {
             val cubeId = CubeId(2, f.tags("cube"))
             cubeId.parent match {

src/test/scala/io/qbeast/spark/index/NewRevisionTest.scala

@@ -40,7 +40,7 @@ class NewRevisionTest
       spaceMultipliers.foreach(i => appendNewRevision(spark, tmpDir, i))
 
       val deltaLog = DeltaLog.forTable(spark, tmpDir)
-      val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+      val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.update())
       val spaceRevisions = qbeastSnapshot.loadAllRevisions
 
       // Including the staging revision
@@ -55,7 +55,7 @@ class NewRevisionTest
         appendNewRevision(spark, tmpDir, 3)
 
         val deltaLog = DeltaLog.forTable(spark, tmpDir)
-        val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+        val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.update())
 
         val revisions = qbeastSnapshot.loadAllRevisions
         val allWM =
@@ -88,7 +88,7 @@ class NewRevisionTest
             .save(tmpDir)
 
           val deltaLog = DeltaLog.forTable(spark, tmpDir)
-          val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+          val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.update())
 
           qbeastSnapshot.loadLatestRevision.desiredCubeSize shouldBe cubeSize
 
@@ -121,7 +121,7 @@ class NewRevisionTest
         .save(tmpDir)
 
       val deltaLog = DeltaLog.forTable(spark, tmpDir)
-      val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+      val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.update())
 
       // Including the staging revision
       qbeastSnapshot.loadAllRevisions.size shouldBe 3
@@ -150,7 +150,7 @@ class NewRevisionTest
           .save(tmpDir)
 
         val deltaLog = DeltaLog.forTable(spark, tmpDir)
-        val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+        val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.update())
         val transformation = qbeastSnapshot.loadLatestRevision.transformations.head
 
         qbeastSnapshot.loadLatestRevision.revisionID shouldBe 1
@@ -182,7 +182,7 @@ class NewRevisionTest
           .save(tmpDir)
 
         val deltaLog = DeltaLog.forTable(spark, tmpDir)
-        val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+        val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.update())
         val transformation = qbeastSnapshot.loadLatestRevision.transformations.head
 
         qbeastSnapshot.loadLatestRevision.revisionID shouldBe 1
@@ -215,7 +215,7 @@ class NewRevisionTest
           .save(tmpDir)
 
         val deltaLog = DeltaLog.forTable(spark, tmpDir)
-        val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+        val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.update())
         val revision = qbeastSnapshot.loadLatestRevision
         val transformation = revision.transformations.head
 
@@ -253,7 +253,7 @@ class NewRevisionTest
         .save(tmpDir)
 
       val deltaLog = DeltaLog.forTable(spark, tmpDir)
-      val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+      val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.update())
       val allRevisions = qbeastSnapshot.loadAllRevisions.sortBy(_.revisionID)
 
       val firstWriteTransformation =
@@ -303,7 +303,7 @@ class NewRevisionTest
           .save(tmpDir)
 
         val deltaLog = DeltaLog.forTable(spark, tmpDir)
-        val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.unsafeVolatileSnapshot)
+        val qbeastSnapshot = delta.DeltaQbeastSnapshot(deltaLog.update())
         val allRevisions = qbeastSnapshot.loadAllRevisions.sortBy(_.revisionID)
 
         val firstWriteTransformation =