[WIP]Try to unblock payments backfill

[yuli-han]

Summary

Why / Goal

Test Plan

• Added Unit Tests
• Covered by existing CI
• Integration tested

Checklist

• Documentation update

Reviewers

[hzding621]

Ideas for incorporating this change long term:

if (df.sparkSession.conf.getOption(SparkConstants.ChrononEnableRowCountBasedRepartitin).exists("true")) {
      wrapWithCache(s"repartition & write to $tableName", df) {
        logger.info(s"Repartitioning before writing...")
        repartitionAndWriteInternal(df, tableName, saveMode, stats, sortByCols)
      }.get  
    } else {
      val shuffleParallelism = df.sparkSession.conf
        .getOption(SparkConstants.ChrononOutputParallelismOverride)
        .map(_.toInt)
        .flatMap(value => if (value > 0) Some(value) else None)
      
      val dfRepartitioned = if (shuffleParallelism.isDefined) {
        df.repartition(shuffleParallelism.get)
      } else {
        df
      }
      dfRepartitioned
        .write
        .mode(saveMode)
        .insertInto(tableName)
    }