Add file skipping with Delta

[osopardo1]

Right now, the filtering works as follows:

1. Apply Delta Filters for Staging Area files.
2. Apply Qbeast Filters on all the rest.
3. Union both sets of files.

We should apply Delta (or any other underlying format) file filtering in all the set of files, not just the subset belonging to the Staging Area.

This is the piece of code:

  override def listFiles(...) = {

    // FILTER FILES FROM QBEAST
    val qbeastFileStats = qbeastMatchingFiles(partitionFilters, dataFilters)
    // FILTER FILES FROM DELTA
    val stagingFileStats = stagingFiles(partitionFilters, dataFilters)
    // JOIN BOTH FILTERED FILES
    val fileStats = qbeastFileStats ++ stagingFileStats

And the code for the stagingFiles method:

  /**
   * Collect matching staging files from _delta_log and convert them into FileStatuses.
   * The output is merged with those built from QbeastBlocks.
   * @return
   */
  private def stagingFiles(
      partitionFilters: Seq[Expression],
      dataFilters: Seq[Expression]): Seq[FileStatus] = {

    index
      .matchingFiles(partitionFilters, dataFilters)
      .filter(isStagingFile)
      .map { f =>
        new FileStatus(
          /* length */ f.size,
          /* isDir */ false,
          /* blockReplication */ 0,
          /* blockSize */ 1,
          /* modificationTime */ f.modificationTime,
          absolutePath(f.path))
      }
  }

This should not apply .filter(isStagingFile) and return all the set of files filtered by the underlying format. Probably also need to update the method's name.

[osopardo1]

I have a question regarding this task:

If we filter all the files with Delta, does still make sense to filter again with Qbeast to filter by min/max?

For the Sampling of course is necessary. But to avoid replicated files in a WHERE predicate, only a filter is needed after applying Delta Data Skipping. No need to rebuild the index structure from scratch.

What do you think? @cugni @Jiaweihu08 @alexeiakimov

[osopardo1]

And can you @alexeiakimov take care of this task?

Thank you!

[osopardo1]

After discussion, agreed on:

• When applying WHERE file filtering, let min/max Delta Skipping filter the set of files.
• When applying SAMPLING, join both sets of files. This allows Qbeast tree navigation to filter more files.

[alexeiakimov]

The target version for this feature is 1.0.0, to support it in both 0.x and 1.x is too complex, because the Qbeast metadata format was changed in version 1.0.0.

[alexeiakimov]

The approach with more details is the following. There are two main cases:

1. The query has SAMPLING clause
2. The query has no SAMPLING clause

If query defines the SAMPLING clause, then apply Delta filtering to the staging area and Qbeast filtering to the normal revisions. It is not necessary to apply Delta filtering to the files returned by Qbeast filtering.

If query does not define SAMPLING clause, then apply Delta filtering to all the files, and then exclude the replicated files. Qbeast index is not used in this case.