Convert to Qbeast

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

@@ -4,7 +4,8 @@ import com.fasterxml.jackson.annotation.{JsonCreator, JsonValue}
 import com.fasterxml.jackson.databind.annotation.JsonSerialize
 import com.fasterxml.jackson.databind.annotation.JsonSerialize.Typing
 import io.qbeast.IISeq
-import io.qbeast.core.transform.{Transformation, Transformer}
+import io.qbeast.core.model.Revision.isStaging
+import io.qbeast.core.transform.{EmptyTransformer, Transformation, Transformer}
 
 import scala.collection.immutable.SortedMap
 
@@ -41,6 +42,17 @@ final class QTableID(_id: String) extends Serializable {
  * Companion object for Revision
  */
 object Revision {
+  val stagingID: RevisionID = 0
+
+  def isStaging(revisionID: RevisionID): Boolean =
+    revisionID == stagingID
+
+  def isStaging(revision: Revision): Boolean =
+    isStaging(revision.revisionID) &&
+      revision.columnTransformers.forall {
+        case _: EmptyTransformer => true
+        case _ => false
+      }
 
   /**
    * Create a new first revision for a table
@@ -60,7 +72,27 @@ object Revision {
       desiredCubeSize,
       columnTransformers,
       Vector.empty)
+  }
 
+  /**
+   * Initialize Revision for table conversion. The RevisionID for a converted table is 0.
+   * EmptyTransformers and EmptyTransformations are used. This Revision should always be
+   * superseded.
+   */
+  def emptyRevision(
+      tableID: QTableID,
+      desiredCubeSize: Int,
+      columnsToIndex: Seq[String]): Revision = {
+    val transformers = columnsToIndex.map(s => EmptyTransformer(s)).toIndexedSeq
+    val transformations = transformers.map(_.makeTransformation(r => r))
+
+    Revision(
+      stagingID,
+      System.currentTimeMillis(),
+      tableID,
+      desiredCubeSize,
+      transformers,
+      transformations)
   }
 
 }
@@ -89,8 +121,7 @@ final case class Revision(
   assert(columnTransformers != null || transformations != null)
 
   /**
-   * *
-   * Controls that the this revision indexes all and only the provided columns.
+   * Controls that this revision indexes all and only the provided columns.
    *
    * @param columnsToIndex the column names to check.
    * @return true if the revision indexes all and only the provided columns.
@@ -117,7 +148,7 @@ final case class Revision(
 
   /**
    * returns the normalized values
-   * @param values
+   * @param values row values for the indexing columns
    * @return the normalized values
    */
   def transform(values: IISeq[_]): IISeq[Double] = {
@@ -193,8 +224,10 @@ case class IndexStatus(
     cubesStatuses: SortedMap[CubeId, CubeStatus] = SortedMap.empty)
     extends Serializable {
 
-  def addAnnouncements(newAnnouncedSet: Set[CubeId]): IndexStatus =
-    copy(announcedSet = announcedSet ++ newAnnouncedSet)
+  def addAnnouncements(newAnnouncedSet: Set[CubeId]): IndexStatus = {
+    if (isStaging(revision)) this
+    else copy(announcedSet = announcedSet ++ newAnnouncedSet)
+  }
 
   def cubesToOptimize: Set[CubeId] = announcedSet.diff(replicatedSet)
 

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

@@ -0,0 +1,13 @@
+package io.qbeast.core.transform
+
+/**
+ * An empty Transformation meant for empty revisions
+ */
+case class EmptyTransformation() extends Transformation {
+
+  override def transform(value: Any): Double = 0d
+
+  override def isSupersededBy(newTransformation: Transformation): Boolean = true
+
+  override def merge(other: Transformation): Transformation = other
+}

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

@@ -0,0 +1,22 @@
+package io.qbeast.core.transform
+
+import io.qbeast.core.model.QDataType
+
+object EmptyTransformer extends TransformerType {
+  override def transformerSimpleName: String = "empty"
+
+  override def apply(columnName: String, dataType: QDataType): Transformer =
+    EmptyTransformer(columnName)
+
+}
+
+/**
+ * An empty Transformer meant for empty revisions
+ */
+case class EmptyTransformer(columnName: String) extends Transformer {
+  override protected def transformerType: TransformerType = EmptyTransformer
+
+  override def stats: ColumnStats = NoColumnStats
+
+  override def makeTransformation(row: String => Any): Transformation = EmptyTransformation()
+}

docs/QbeastFormat.md

@@ -153,6 +153,36 @@ In Revision, you can find different information about the tree status and config
 
 In this case, we index columns `user_id` and `product_id` (which are both `Integers`) with a linear transformation. This means that they will not suffer any transformation besides the normalization.
 
+### Staging Revision and ConvertToQbeastConvertToQbeastCommand
+The introduction of the staging revision allows users to read a delta or a hybrid `qbeast + delta` table using qbeast format.
+The non-qbeast `AddFile`s are considered as part of this staging revision.
+
+Its RevisionID is fixed to `stagingID = 0`, and it has `EmptyTransformer`s and `EmptyTransformation`s.
+This Revision is normally created during the first write or when overwriting a table using qbeast.
+For a table that is entirely written in `delta` or `parquet`, partitioned or not, we can use the `ConvertToQbeastCommand` to create this revision:
+```scala
+import io.qbeast.spark.internal.commands.ConvertToQbeastCommand
+
+val path = "/pathToTable/"
+val tableIdentifier = s"parquet.`$path`"
+val columnsToIndex = Seq("col1", "col2", "col3")
+val desiredCubeSize = 50000
+
+ConvertToQbeastCommand(tableIdentifier, columnsToIndex, desiredCubeSize).run(spark)
+
+val qTable = spark.read.format("qbeast").load(path)
+```
+
+By doing so, we also enable appends using either delta or the qbeast format.
+
+`Compaction` is allowed to group small files in the staging revision:
+```scala
+import io.qbeast.spark.QbeastTable
+
+val table = QbeastTable.forPath(spark, "/pathToTable/")
+table.compact(0)
+```
+
 ### State changes in MetaData
 
 **Data de-normalization** is a crucial component behind our multi-dimensional index. Instead of storing an index in a separate tree-like data structure, we reorganize the data and their replications in an `OTree`, whose **hierarchical structure** is the actual index.

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

@@ -103,7 +103,7 @@ class QbeastTable private (
     val allCubeStatuses = qbeastSnapshot.loadLatestIndexStatus.cubesStatuses
 
     val cubeCount = allCubeStatuses.size
-    val depth = allCubeStatuses.map(_._1.depth).max
+    val depth = if (cubeCount == 0) 0 else allCubeStatuses.map(_._1.depth).max
     val rowCount = allCubeStatuses.flatMap(_._2.files.map(_.elementCount)).sum
 
     val dimensionCount = indexedColumns().size

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

@@ -168,7 +168,7 @@ private[delta] case class DeltaMetadataWriter(
       addFiles.map(_.copy(dataChange = !rearrangeOnly)) ++
         deletedFiles.map(_.copy(dataChange = !rearrangeOnly))
     } else {
-      newFiles ++ deletedFiles
+      addFiles ++ deletedFiles
     }
 
     if (isOptimizeOperation) {

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

@@ -4,7 +4,9 @@
 package io.qbeast.spark.delta
 
 import io.qbeast.IISeq
+import io.qbeast.core.model.Revision.{isStaging, stagingID}
 import io.qbeast.core.model._
+import io.qbeast.spark.utils.Staging.isStagingFile
 import io.qbeast.spark.utils.{MetadataConfig, TagColumns}
 import org.apache.spark.sql.delta.Snapshot
 import org.apache.spark.sql.delta.actions.AddFile
@@ -122,7 +124,8 @@ case class DeltaQbeastSnapshot(private val snapshot: Snapshot) extends QbeastSna
    *
    * @return an immutable Seq of Revision for qtable
    */
-  override def loadAllRevisions: IISeq[Revision] = revisionsMap.values.toVector
+  override def loadAllRevisions: IISeq[Revision] =
+    revisionsMap.values.toVector
 
   /**
    * Obtain the last Revisions
@@ -150,7 +153,7 @@ case class DeltaQbeastSnapshot(private val snapshot: Snapshot) extends QbeastSna
    * @return the latest Revision at a concrete timestamp
    */
   override def loadRevisionAt(timestamp: Long): Revision = {
-    revisionsMap.values.find(_.timestamp <= timestamp).getOrElse {
+    (revisionsMap - stagingID).values.find(_.timestamp <= timestamp).getOrElse {
       throw AnalysisExceptionFactory.create(s"No space revision available before $timestamp")
     }
   }
@@ -161,8 +164,8 @@ case class DeltaQbeastSnapshot(private val snapshot: Snapshot) extends QbeastSna
    * @return the Dataset of QbeastBlocks
    */
   def loadRevisionBlocks(revisionID: RevisionID): Dataset[AddFile] = {
-    snapshot.allFiles
-      .where(TagColumns.revision === lit(revisionID.toString))
+    if (isStaging(revisionID)) snapshot.allFiles.where(isStagingFile)
+    else snapshot.allFiles.where(TagColumns.revision === lit(revisionID.toString))
   }
 
 }

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

@@ -3,8 +3,10 @@
  */
 package io.qbeast.spark.delta
 
+import io.qbeast.core.model.Revision.isStaging
 import io.qbeast.core.model._
 import io.qbeast.spark.delta.QbeastMetadataSQL._
+import io.qbeast.spark.utils.State.FLOODED
 import io.qbeast.spark.utils.TagColumns
 import org.apache.spark.sql.delta.actions.AddFile
 import org.apache.spark.sql.functions.{col, collect_list, lit, min, sum}
@@ -35,41 +37,66 @@ private[delta] class IndexStatusBuilder(
     qbeastSnapshot.loadRevisionBlocks(revision.revisionID)
 
   def build(): IndexStatus = {
+    val cubeStatus =
+      if (isStaging(revision)) stagingCubeStatuses
+      else buildCubesStatuses
+
     IndexStatus(
       revision = revision,
       replicatedSet = replicatedSet,
       announcedSet = announcedSet,
-      cubesStatuses = buildCubesStatuses)
+      cubesStatuses = cubeStatus)
+  }
+
+  def stagingCubeStatuses: SortedMap[CubeId, CubeStatus] = {
+    val root = revision.createCubeIdRoot()
+    val maxWeight = Weight.MaxValue
+    val blocks = revisionFiles
+      .collect()
+      .map(addFile =>
+        QbeastBlock(
+          addFile.path,
+          revision.revisionID,
+          Weight.MinValue,
+          maxWeight,
+          FLOODED,
+          0,
+          addFile.size,
+          addFile.modificationTime))
+      .toIndexedSeq
+
+    SortedMap(root -> CubeStatus(root, maxWeight, maxWeight.fraction, blocks))
   }
 
   /**
    * Returns the index state for the given space revision
    * @return Dataset containing cube information
    */
   def buildCubesStatuses: SortedMap[CubeId, CubeStatus] = {
-
     val spark = SparkSession.active
+    import spark.implicits._
+
     val builder = SortedMap.newBuilder[CubeId, CubeStatus]
 
-    val rev = revision
+    val dimCount: Int = revision.transformations.size
+    val dcs = revision.desiredCubeSize
 
-    import spark.implicits._
-    val ndims: Int = rev.transformations.size
     revisionFiles
       .groupBy(TagColumns.cube)
       .agg(
         min(weight(TagColumns.maxWeight)).as("maxWeight"),
         sum(TagColumns.elementCount).as("elementCount"),
         collect_list(qblock).as("files"))
       .select(
-        createCube(col("cube"), lit(ndims)).as("cubeId"),
+        createCube(col("cube"), lit(dimCount)).as("cubeId"),
         col("maxWeight"),
-        normalizeWeight(col("maxWeight"), col("elementCount"), lit(rev.desiredCubeSize)).as(
-          "normalizedWeight"),
+        normalizeWeight(col("maxWeight"), col("elementCount"), lit(dcs))
+          .as("normalizedWeight"),
         col("files"))
       .as[CubeStatus]
       .collect()
       .foreach(row => builder += row.cubeId -> row)
+
     builder.result()
   }
 

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

@@ -5,16 +5,17 @@ package io.qbeast.spark.delta
 
 import io.qbeast.core.model.QbeastBlock
 import io.qbeast.spark.index.query.{QueryExecutor, QuerySpecBuilder}
+import io.qbeast.spark.utils.Staging.isStagingFile
 import org.apache.hadoop.fs.{FileStatus, Path}
+import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.expressions.{Expression, GenericInternalRow}
-import org.apache.spark.sql.delta.Snapshot
+import org.apache.spark.sql.delta.{DeltaLog, Snapshot}
+import org.apache.spark.sql.delta.actions.AddFile
 import org.apache.spark.sql.delta.files.TahoeLogFileIndex
 import org.apache.spark.sql.execution.datasources.{FileIndex, PartitionDirectory}
 import org.apache.spark.sql.types.StructType
-import org.apache.spark.sql.SparkSession
 
 import java.net.URI
-import org.apache.spark.sql.delta.DeltaLog
 
 /**
  * FileIndex to prune files
@@ -49,11 +50,28 @@ case class OTreeIndex(index: TahoeLogFileIndex) extends FileIndex {
     queryExecutor.execute()
   }
 
+  /**
+   * Collect Staging AddFiles from _delta_log and convert them into FileStatuses.
+   * The output is merged with those built from QbeastBlocks.
+   * @return
+   */
+  private def stagingFiles: Seq[FileStatus] = {
+    snapshot.allFiles.where(isStagingFile).collect().map { a: AddFile =>
+      new FileStatus(
+        /* length */ a.size,
+        /* isDir */ false,
+        /* blockReplication */ 0,
+        /* blockSize */ 1,
+        /* modificationTime */ a.modificationTime,
+        absolutePath(a.path))
+    }
+  }
+
   override def listFiles(
       partitionFilters: Seq[Expression],
       dataFilters: Seq[Expression]): Seq[PartitionDirectory] = {
 
-    val fileStats = matchingBlocks(partitionFilters, dataFilters).map { qbeastBlock =>
+    val qbeastFileStats = matchingBlocks(partitionFilters, dataFilters).map { qbeastBlock =>
       new FileStatus(
         /* length */ qbeastBlock.size,
         /* isDir */ false,
@@ -62,6 +80,8 @@ case class OTreeIndex(index: TahoeLogFileIndex) extends FileIndex {
         /* modificationTime */ qbeastBlock.modificationTime,
         absolutePath(qbeastBlock.path))
     }.toArray
+    val stagingStats = stagingFiles
+    val fileStats = qbeastFileStats ++ stagingStats
 
     Seq(PartitionDirectory(new GenericInternalRow(Array.empty[Any]), fileStats))
 
@@ -80,17 +100,17 @@ case class OTreeIndex(index: TahoeLogFileIndex) extends FileIndex {
   override def partitionSchema: StructType = index.partitionSchema
 }
 
-/**
- * Object OTreeIndex to create a new OTreeIndex
- * @param sparkSession the spark session
- * @param path the path to the delta log
- * @return the OTreeIndex
- */
 object OTreeIndex {
 
   def apply(spark: SparkSession, path: Path): OTreeIndex = {
     val deltaLog = DeltaLog.forTable(spark, path)
-    val tahoe = TahoeLogFileIndex(spark, deltaLog, path, deltaLog.snapshot, Seq.empty, false)
+    val tahoe = TahoeLogFileIndex(
+      spark,
+      deltaLog,
+      path,
+      deltaLog.snapshot,
+      Seq.empty,
+      isTimeTravelQuery = false)
     OTreeIndex(tahoe)
   }