SNAP-656 Delink RDD partitions from buckets

[ymahajan]

What changes were proposed in this pull request?

Use child's numBuckets to decide partitions of shuffled relation in ShuffleExchange
Pass numBuckets to HashPartitioning and HashPartitioner

How was this patch tested?

Other PRs :
Store - TIBCOSoftware/snappy-store#85
Spark - #4
SnppyData - TIBCOSoftware/snappydata#297

[rishitesh]

Can we try out one thing. Leave everything as is @ catalyst layer.
a) The determination to shuffle or not to shuffle etc can be left alone. I think outputPartitioning is only used to determine shuffle. It should not have a bearing on actual scan. If we leave it as it is with numBuckets my best guess is we should be fine. Do you see any issues with that ?

b) Once catalyst determines what to to do we can put the num cpu logic at only RDD scan layer i.e. each relation's buildScan, which we already are doing.

[sumwale]

@rishitesh the issue is that we want to do shuffle even when partitions are equal and buckets are different. Spark layer will say no shuffle which is incorrect.

@ymahajan What we discussed about implementing new Partitioning did not work, either by marking Partitioning unsealed in snappy-spark or extending HashPartitioning?

[rishitesh]

@sumwale Correct. That's why I mentioned leave OrderlessHashPartitioning as that of today i.e. work on numBuckets. So if two tables have different num buckets then it will cause a shuffle. But if they are equal catalyst will say no to shuffle . Then RDD layer will take care of the scan and cpu logic will kick in.

[ymahajan]

@sumwale. Yes, I tried both your approaches of new partitioning or extending HashPartitioning. we had to do changes at the catalyst layer (In EnsureRequirements, ShuffleExchange) to get this working even for the embedded mode. And for split mode, all these changes were not possible so i had to disable this optimization for split mode.

@rishitesh. Wouldn't this optimization beneficial for both scans and joins as well? outputPartitioning also decides how many target partitions to be created. If that is same as numPartitions(in other cases), it simplifies. numBuckets will be a good input for catalyst in general also. Do you see any side effects of these catalyst related changes?

[sumwale]

@rishitesh Just using OrderlessHashPartitioning will not work. Partitioning will match due to buckets but then join will fail since the number of partitions may not match. I guess the number of partitions not matching will be a concern if server groups are being used, but otherwise they should always match? Also it means that to keep the number of partitions consistent, the total number of cores (and thus the partitions + their distribution) has to be determined only once at top-level query execution (perhaps in SnappySession just before query execution) and propagated to all RDDs below to handle stray cases of new nodes joining.

@ymahajan Why were additional changes required? Today OrderlessHashPartitioning works with no additional changes to other parts.

[ymahajan]

@sumwale Catalyst decides whether to use existing partitioning or go for shuffle by comparing(guarantees) with child's output partitioning. And to check the guarantees it creates HashPartitioning on the fly with expression and numPartitions) in EnsureRequirements#ensureDistributionAndOrdering. If we don't change the HashPartitioning to consider numBuckets, both the partitionings(our OrderlessHashPartitioning and HashPartitionig) satisfy guarantees and existing partitioning is used even for the two different numBuckets joins. So we need to provide numBuckets as an input to catalyst so that it will create HashPartitioning with numBuckets and compare against our OrderlessHashPartitioning

[sumwale]

@rishitesh Looking at the comment again, it seems you are saying return numBuckets in Partitioning, but use actual partitions for RDD execution? Unlikely to work because a lot of logic will depend on both being same (perhaps even assertions in code) with Spark assuming that the output will have numBuckets partitions, then try to apply further operators, including shuffle, which will likely fail. Can give it a shot but looks like a landmine to me.

@ymahajan Looking at the code there are two calls there: Partitioning.allCompatible and then inside child.outputPartitioning.guarantees. I guess you are saying that latter one won't be able to determine whether or not it guarantees since the argument is created on the fly. However, you actually don't need to. In fact for guarantees() you should only check if the number of buckets is equal to provided number of partitions (and if so then mark a flag in PartitionedPhysicalRDD to use number of buckets as partitions) because that is the only way to guarantee. For normal cases code will not even come inside there if Partitioning.allCompatible returns true (i.e. only in compatibleWith check the buckets on both sides).

[rishitesh]

Yes , just checked the code. There is an assertion checking RDD partition length with Spark plan partition length. My suggestion wont work here.

[sumwale]

@ymahajan @rishitesh I don't agree with merging these changes as noted before. It means that now for split mode the delinking will not work and it will be back to the old way.

Perhaps can keep for a moment for 0.6, but this should be reverted and better fix needed for 1.0