Hybrid scan operator for leveraging index alongside newly appended data - BucketUnion

[sezruby]

What changes were proposed in this pull request?

For full context on the "why" for this PR, please see the main issue: #150

Introducing a new BucketUnion operator that is useful for implementing hybrid scan, a technique that we propose that can leverage the index and appended data without the need to re-shuffle the index, thus preserving the benefits of the index.

As part of this, the following classes are being implemented:

1. BucketUnion: Logical Plan operator; Used during logical plan optimization when the newly appended data needs to be union-ed with the data being read from the index.
2. BucketUnionExec: SparkPlan (Physical operator); Calls into BucketUnionRDD
3. BucketUnionRDD: RDD operator
4. BucketUnionRDDPartition: Partition
5. BucketUnionStrategy: SparkStrategy that is used when the Logical Plan is being converted to SparkPlan; More specifically, converts BucketUnion to BucketUnionExec

Note: You can find more detailed information about Bucketing optimization in Bucketing 2.0: Improve Spark SQL Performance by Removing Shuffle

Why are the changes needed?

Spark does not support Union using PartitionSpecification, but just PartitionerAwareUnionRDD operation which does not retain outputPartitioning of result. Therefore, we define a new Union operation (being called the BucketUnion) which works when the following conditions are satisfied:

• input RDDs must have the same number of partitions.
• input RDDs must have the same partitioning keys.
• input RDDs must have the same column schema.

Unfortunately, since there is no explicit API to check Partitioning keys in RDD, we have to assure that on the caller side. Therefore, BucketUnionRDD is Hyperspace internal use only.

BucketUnion can be used to merge index data & newly appended data without losing the bucketing specification (outputPartitioning).

Does this PR introduce any user-facing change?

Existing experience:
When the underlying data changes, Hyperspace decides to not use the index anymore.

New experience:
When the underlying data changes and hybrid scan is enabled, Hyperspace utilizes the index to the extent possible and performs a linear scan on the new data.

How was this patch tested?

BucketUnionTest

[imback82]

Few minor comments, but LGTM, thanks @sezruby!

[imback82]

@apoorvedave1 @rapoth @pirz Can you review as well?

[imback82]

@sezruby Could you update the title to be a bit more descriptive? (it's used as a commit message).

[apoorvedave1]

a few minor comments, otherwise LGTM, thanks @sezruby

[pirz]

Some of these DF definitions are repeated per test; Do you think it is possible to define them once at the class level and initialize them in beforeAll? (similar to partitionedDataDF and nonPartitionedDataDF in CreateIndexTests.scala).

[imback82]

I actually prefer the current way. The dfs are simple enough and it makes it easier to read in this scope; I don't have to go back and forth to remember how it was defined.

[rapoth]

Thanks a lot for opening a short and consise PR! I really appreciate it!

[apoorvedave1]

LGTM 👍 , thanks @sezruby

[imback82]

LGTM, thanks @sezruby!