@Override
  public int deduceNumPartitions() {
    // for source rdd, the partitioner is None
    final Optional<Partitioner> partitioner = rddData.partitioner();
    if (partitioner.isPresent()) {
      int partPartitions = partitioner.get().numPartitions();
      if (partPartitions > 0) {
        return partPartitions;
      }
    }

    if (SQLConf.get().contains(SQLConf.SHUFFLE_PARTITIONS().key())) {
      return Integer.parseInt(SQLConf.get().getConfString(SQLConf.SHUFFLE_PARTITIONS().key()));
    } else if (rddData.context().conf().contains("spark.default.parallelism")) {
      return rddData.context().defaultParallelism();
    } else {
      return rddData.getNumPartitions();
    }
  }
@Test
  public void testDeduceNumPartitions() {
    int numPartitions = 100;
    jsc.sc().conf().remove("spark.default.parallelism");
    SQLConf.get().unsetConf("spark.sql.shuffle.partitions");

    // rdd parallelize
    SQLConf.get().setConfString("spark.sql.shuffle.partitions", "5");
    HoodieData<Integer> rddData = HoodieJavaRDD.of(jsc.parallelize(
        IntStream.rangeClosed(0, 100).boxed().collect(Collectors.toList()), numPartitions));
    assertEquals(5, rddData.deduceNumPartitions());

    // sql parallelize
    SQLConf.get().unsetConf("spark.sql.shuffle.partitions");
    jsc.sc().conf().set("spark.default.parallelism", "6");
    rddData = HoodieJavaRDD.of(jsc.parallelize(
        IntStream.rangeClosed(0, 100).boxed().collect(Collectors.toList()), numPartitions));
    assertEquals(6, rddData.deduceNumPartitions());

    // use partitioner num
    HoodiePairData<Integer, Integer> shuffleRDD = rddData.mapToPair(key -> Pair.of(key, 1))
        .reduceByKey((p1, p2) -> p1, 11);
    assertEquals(11, shuffleRDD.deduceNumPartitions());
  }