public UpsertPartitioner(WorkloadProfile profile, HoodieEngineContext context, HoodieTable table,
                           HoodieWriteConfig config, WriteOperationType operationType) {
    super(profile, table);
    updateLocationToBucket = new HashMap<>();
    partitionPathToInsertBucketInfos = new HashMap<>();
    bucketInfoMap = new HashMap<>();
    this.config = config;
    this.operationType = operationType;
    assignUpdates(profile);
    long totalInserts = profile.getInputPartitionPathStatMap().values().stream().mapToLong(stat -> stat.getNumInserts()).sum();
    if (!WriteOperationType.isPreppedWriteOperation(operationType) || totalInserts > 0) { // skip if its prepped write operation. or if totalInserts = 0.
      assignInserts(profile, context);
    }

    LOG.info("Total Buckets: {}, bucketInfoMap size: {}, partitionPathToInsertBucketInfos size: {}, updateLocationToBucket size: {}",
        totalBuckets, bucketInfoMap.size(), partitionPathToInsertBucketInfos.size(), updateLocationToBucket.size());
    if (LOG.isDebugEnabled()) {
      LOG.debug("Buckets info => " + bucketInfoMap + ", \n"
              + "Partition to insert buckets => " + partitionPathToInsertBucketInfos + ", \n"
              + "UpdateLocations mapped to buckets =>" + updateLocationToBucket);
    }
  }
@Test
  public void testUpsertPartitioner() throws Exception {
    final String testPartitionPath = "2016/09/26";
    // Inserts + Updates... Check all updates go together & inserts subsplit
    UpsertPartitioner partitioner = getUpsertPartitioner(0, 200, 100, 1024, testPartitionPath, false);
    List<InsertBucketCumulativeWeightPair> insertBuckets = partitioner.getInsertBuckets(testPartitionPath);
    assertEquals(2, insertBuckets.size(), "Total of 2 insert buckets");
  }