protected void createBucketFiles(FSPaths fsp) throws HiveException {
    try {
      int filesIdx = 0;
      Set<Integer> seenBuckets = new HashSet<Integer>();
      for (int idx = 0; idx < totalFiles; idx++) {
        if (this.getExecContext() != null && this.getExecContext().getFileId() != null) {
          LOG.info("replace taskId from execContext");

          taskId = Utilities.replaceTaskIdFromFilename(taskId, this.getExecContext().getFileId());

          LOG.info("new taskId: FS " + taskId);

          assert !multiFileSpray;
          assert totalFiles == 1;
        }

        int bucketNum = 0;
        if (multiFileSpray) {
          key.setHashCode(idx);

          // Does this hashcode belong to this reducer
          int numReducers = totalFiles / numFiles;

          if (numReducers > 1) {
            int currReducer = Integer.parseInt(Utilities.getTaskIdFromFilename(Utilities
                .getTaskId(hconf)));

            int reducerIdx = prtner.getPartition(key, null, numReducers);
            if (currReducer != reducerIdx) {
              continue;
            }
          }

          bucketNum = prtner.getBucket(key, null, totalFiles);
          if (seenBuckets.contains(bucketNum)) {
            continue;
          }
          seenBuckets.add(bucketNum);

          bucketMap.put(bucketNum, filesIdx);
          taskId = Utilities.replaceTaskIdFromFilename(Utilities.getTaskId(hconf), bucketNum);
        }
        createBucketForFileIdx(fsp, filesIdx);
        filesIdx++;
      }
      assert filesIdx == numFiles;
    } catch (Exception e) {
      throw new HiveException(e);
    }

    filesCreated = true;
  }
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