Avoid repeated reading of the DeltaLog

[osopardo1]

This PR fixes #61

The changes made are:

• OTreeIndex now extends FileIndex instead of TahoeLogFileIndex
• QueryExecutor doesn't need previouslyMatchedFiles anymore
• QbeastFile now it's named QbeastBlock
• Add more metadata to the QbeastBlock: size and modificationTime
• CubeStatus now contains QbeastBlock objects instead of the path

[cugni]

I don't understand why we do buildCubesStatuses in memory now, what was the need?
Also, we could change some class/method names to improve readability and be more consistent with the documentation.

[osopardo1]

#65 (comment)

In relation to this comment, I did the following comparisons using different approaches (collect before, collect after and the code in the main branch) on an indexed store_sales table of the 10gb TPC-DS dataset.

df.write
      .format("qbeast")
      .option("columnsToIndex", "ss_item_sk,ss_customer_sk")
      .option("cubeSize", "300000")
      .save(qbeastDir)

The times compared are from two different methods:

• Query. I compared running a count on the whole dataset both with Delta and Qbeast implementations

val qbeast = spark.read.format("qbeast").load(qbeastDir)
val delta = spark.read.format("delta").load(qbeastDir)

qbeast.count()
delta.count()

• ListFiles. This is the method used by the TahoeLogFileIndex and OTreeIndex for filtering the files needed to query. It is useful to know if we improved something by reading only once the log

val deltaLog = DeltaLog.forTable(spark, qbeastDir)
val tahoeLogFileIndex =
  TahoeLogFileIndex(spark, deltaLog, deltaLog.dataPath, deltaLog.snapshot)
val oTreeIndex = OTreeIndex(tahoeLogFileIndex)

tahoeLogFileIndex.listFiles(partitionFilters = Seq.empty, dataFilters = Seq.empty)
oTreeIndex.listFiles(partitionFilters = Seq.empty, dataFilters = Seq.empty)

These are the implementations tested:

• Delta → qbeast indexed but read as delta
• Qbeast main branch → qbeast indexed but read with the main version
• Qbeast PR65 commit f2445ec → Collect after groupBy in buildCubeStatuses
• Qbeast PR65 commit 36eea23 → Collect before groupBy in buildCubeStatuses

Operation	Delta	Qbeast Main	QBeast PR65 Collect After	Qbeast PR65 Collect Before
Query	1.38s	2.94s	2.36s	1.73s
ListFiles	0.18s	1.4s	0.9s	0.3s

The tests were run locally. Feel free to discuss this approach here, and to suggest reversion of the changes! Thank you!

[osopardo1]

Made more test on a distributed cluster, just to have more numbers. Here are the results:

mode: DISTRIBUTED, cluster: 4 node cluster, data:10gb store sales, columnsToIndex: ss_customer_sk, ss_item_sk, cubeSize default , num files: 5000+ files

Operation	Delta	Qbeast Main	Qbeast PR65 Collect After	Qbeast PR65 Collect Before
Query	8.2s	10.74s	10.05s	10.2s
ListFiles	0.43s	2.1s	0.91s	0.53s

mode: DISTRIBUTED, cluster: 4 node cluster, data:10gb store sales, columnsToIndex: ss_customer_sk, ss_item_sk, cubeSize 500000 , num files: 800+ files

Operation	Delta	Qbeast Main	Qbeast PR65 Collect After	Qbeast PR65 Collect Before
Query	1.64s	2.19s	3.14s	2.55s
ListFiles	0.32s	0.65s	0.96s	0.42s

mode: DISTRIBUTED, cluster: 4 node cluster, data:100gb store sales, columnsToIndex: ss_customer_sk, ss_item_sk, cubeSize 50000000 , num files: 86 files

Operation	Delta	Qbeast Main	Qbeast PR65 Collect After	Qbeast PR65 Collect Before
Query	0.96s	1.47s	1.46s	1.83s
ListFiles	0.24s	0.59s	0.59s	0.42s

We can see two things:

• Penalization of reading multiple files.
• For the largest DeltaLogs (for example the first one, with 5000+ files), this PR outputs better results on listFiles. In the other cases, the results are similar.

I will revert to a collect() after the groupBy because due it can give memory problems. delta-io/delta#779 .
We could think about optimization later.