Ability to do small file handling without need for caching

As of now, in upsert path,
* hudi builds a workloadProfile to understand total inserts and updates(with location info) 
* Following which, small files info are populated
* Then buckets are populated with above info. 
* These buckets are later used when getPartition(Object key) is invoked in UpsertPartitioner.

In step1: to build global workload profile, we had to do an action on entire JavaRDDs in the driver and hudi does save the workload profile as well. 

For large write intensive batch jobs(COW types), caching this incurs additional overhead. So, this effort is trying to see if we can avoid doing this by some means. 

 

 

## JIRA info

- Link: https://issues.apache.org/jira/browse/HUDI-860
- Type: Improvement
- Epic: https://issues.apache.org/jira/browse/HUDI-1628

---

## Comments

06/Jul/20 13:16;shivnarayan;Here is a proposal on the New Partitioner.

High level idea is to avoid doing any record operation in the driver. 
* MapPartitions() with one per hudi partition to collect all bucket infos.
** This mapPartition call will take in Iterator and return an Iterator
** BucketInfo will contain \{PartitionPath, fileId, bucketType, totalRecords, weight(only for inserts)}
** All buckets for a partition can be populated within each mapPartitionCall itself and don't need any global stats. 
** I am not getting into impl of this mapPartition call since its mostly similar to what we have in UpsertPartitioner just that most of this calculation is done in the driver for now and we are moving it to executor with this proposal. 
* Once the mapPartition() call returns to driver, we have all bucketInfo from all partitions. 
* We need to generate few structures to assist in getPartition() call. 
** calculate total buckets
** Iterator through all buckets to do the following
*** Assign global bucketNumbers. And populate bucketInfoMap which maps from bucketNumber to BucketInfo
*** For update bucket, generate updateBucketFileIdToIndexMapping which maps fileId to bucketNumber.
*** For insert bucket, generate perPartitionInsertBucketWeights. Map>>. Key is partitionPath. Value is a List of meta info about insert buckets. i.e. resp. weight and bucketNumber.
*** For insert bucket, generate totalInsertsPerPartition (Map) , where key is partitionPath and value is totalInserts for the partition. 
* getPartition(Object key)
** if update, look up in updateBucketFileIdToIndexMapping and return bucketNumber
** if insert, fetch all target buckets from perPartitionInsertBucketWeights for the resp. partition where record belongs to. Mod by total inserts for the partition(from totalInsertsPerPartition) and find the right bucket based on perPartitionInsertBucketWeights;;;

---

16/Jul/20 11:05;shivnarayan;[~vinoth]: did you get a chance to look at this proposal. ;;;

---

17/Jul/20 03:44;vinoth;>MapPartitions() with one per hudi partition to collect all bucket infos.
>Once the mapPartition() call returns to driver, we have all bucketInfo from all partitions.

This means we cannot avoid caching, right? if you do RDD.mapPartitions().collect() -> List. is this what you are referring to?;;;

---

17/Jul/20 03:47;vinoth;What I mean is, we need to read the RDD again if we convert to Iteratore to Iterator.. We need to keep this within getPartition() and we build all the bucketInfo dynamically on the fly// ;;;

---

17/Jul/20 03:49;shivnarayan;we don't need to build global workload stats. with mapPartitions will calculate buckets required for the corresponding hoodie partition, and will return the info to driver. Driver will collect all such info for all partitions to find total parallelism and then distribute the work based. ;;;

---

17/Jul/20 03:50;shivnarayan;To determine total parallelism, driver needs to understand how many buckets are required in total right. not sure how we can completely delegate work to an executor and get it done w/o getting back to driver. 

 ;;;

---

17/Jul/20 03:58;vinoth;>and will return the info to driver. Driver will collect all such info for all partitions to find total parallelism and then distribute the work based.
This part needs a collect() correct? to return these to the driver? If we do this, then its all futile.. ;;;

---

06/Oct/21 13:47;vinoth;[~guoyihua] this is a good one to get started on the optimization pieces

 ;;;

---

06/Oct/21 23:59;guoyihua;Cool, I'll take a look.;;;

Start with the upsert path, workloadProfile, and UpsertPartitioner, then trace how small-file and bucket information is built and consumed. Review the mapPartitions proposal and the discussion about collect() and caching before choosing an approach. Done means the upsert path avoids the described driver-side workload-profile caching while preserving bucket assignment and parallelism.

Written by the indexing model from the issue text.