private void updateDPCounters(final FSPaths fsp, final int filesIdx) {
    // There are 2 cases where we increment CREATED_DYNAMIC_PARTITIONS counters
    // 1) Insert overwrite (all partitions are newly created)
    // 2) Insert into table which creates new partitions (some new partitions)

    if (bDynParts && destTablePath != null && fsp.dpDirForCounters != null) {
      Path destPartPath = new Path(destTablePath, fsp.dpDirForCounters);
      // For MM tables, directory structure is
      // <table-dir>/<partition-dir>/<delta-dir>/

      // For Non-MM tables, directory structure is
      // <table-dir>/<staging-dir>/<partition-dir>

      // if UNION ALL insert, for non-mm tables subquery creates another subdirectory at the end for each union queries
      // <table-dir>/<staging-dir>/<partition-dir>/<union-dir>

      // for non-MM tables, the final destination partition directory is created during move task via rename
      // for MM tables and ACID insert, the final destination partition directory is created by the tasks themselves
      try {
        if (conf.isMmTable() || conf.isDirectInsert()) {
          createDpDir(destPartPath);
        } else {
          // outPath will be
          // non-union case: <table-dir>/<staging-dir>/<partition-dir>/<taskid>
          // union case: <table-dir>/<staging-dir>/<partition-dir>/<union-dir>/<taskid>
          Path dpStagingDir = fsp.outPaths[filesIdx].getParent();
          if (isUnionDp) {
            dpStagingDir = dpStagingDir.getParent();
          }
          if (isInsertOverwrite) {
            createDpDir(dpStagingDir);
          } else {
            createDpDirCheckSrc(dpStagingDir, destPartPath);
          }
        }
      } catch (IOException e) {
        LOG.warn("Skipping to increment CREATED_DYNAMIC_PARTITIONS counter.Exception: {}", e.getMessage());
      }
    }
  }
There was an issue with the query-based MINOR compaction (HIVE-23763), that the row distribution between the FileSinkOperators
was not correlated correctly with the bucket numbers. So it could happen that rows from different buckets ended up in the same
FileSinkOperator and got written out into one file. This is not correct, one bucket file must contain rows from the same bucket.
Therefore the FileSinkOperator got extended with this method to be able to handle rows from different buckets.
In this case it will create separate files from each bucket. This logic is similar to the one in the createDynamicBucket method.
@param fsp
@throws HiveException
ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java