protected void createBucketForFileIdx(FSPaths fsp, int filesIdx)
      throws HiveException {
    try {
      if (isCompactionTable) {
        fsp.initializeBucketPaths(filesIdx, AcidUtils.BUCKET_PREFIX + String.format(AcidUtils.BUCKET_DIGITS, bucketId),
            isNativeTable(), isSkewedStoredAsSubDirectories);
      } else {
        fsp.initializeBucketPaths(filesIdx, taskId, isNativeTable(), isSkewedStoredAsSubDirectories);
      }
      if (Utilities.FILE_OP_LOGGER.isTraceEnabled()) {
        Utilities.FILE_OP_LOGGER.trace("createBucketForFileIdx " + filesIdx + ": final path " + fsp.finalPaths[filesIdx]
          + "; out path " + fsp.outPaths[filesIdx] +" (spec path " + specPath + ", tmp path "
          + fsp.buildTmpPath() + ", task " + taskId + ")");
      }
      LOG.info("New Final Path: FS " + fsp.finalPaths[filesIdx]);

      if (isNativeTable() && !conf.isMmTable() && !conf.isDirectInsert()) {
        // in recent hadoop versions, use deleteOnExit to clean tmp files.
        autoDelete = fs.deleteOnExit(fsp.outPaths[filesIdx]);
      }

      updateDPCounters(fsp, filesIdx);

      Utilities.copyTableJobPropertiesToConf(conf.getTableInfo(), jc);
      // only create bucket files only if no dynamic partitions,
      // buckets of dynamic partitions will be created for each newly created partition
      //todo IOW integration. Full Acid uses the else if block to create Acid's RecordUpdater (HiveFileFormatUtils)
      // and that will set writingBase(conf.getInsertOverwrite())
      // If MM wants to create a new base for IOW (instead of delta dir), it should specify it here
      if (conf.getWriteType() == AcidUtils.Operation.NOT_ACID || conf.isMmTable() || isCompactionTable) {
        Path outPath = fsp.outPaths[filesIdx];
        if (conf.isMmTable()
            && !FileUtils.mkdir(fs, outPath.getParent(), hconf)) {
          LOG.warn("Unable to create directory with inheritPerms: " + outPath);
        }
        fsp.outWriters[filesIdx] = HiveFileFormatUtils.getHiveRecordWriter(jc, conf.getTableInfo(),
            outputClass, conf, outPath, reporter);
        // If the record writer provides stats, get it from there instead of the serde
        statsFromRecordWriter[filesIdx] = fsp.outWriters[filesIdx] instanceof
            StatsProvidingRecordWriter;
        // increment the CREATED_FILES counter
      } else if (conf.getWriteType() == AcidUtils.Operation.INSERT) {
        Path outPath = fsp.outPaths[filesIdx];
        if (conf.isDirectInsert()
            && !FileUtils.mkdir(fs, outPath.getParent(), hconf)) {
          LOG.warn("Unable to create directory: " + outPath);
        }
        // Only set up the updater for insert.  For update and delete we don't know unitl we see
        // the row.
        ObjectInspector inspector = bDynParts ? subSetOI : outputObjInspector;
        int acidBucketNum = Integer.parseInt(Utilities.getTaskIdFromFilename(taskId));
        Integer attemptId = getAttemptIdFromTaskId(taskId);
        fsp.updaters[filesIdx] = HiveFileFormatUtils.getAcidRecordUpdater(jc, conf.getTableInfo(),
            acidBucketNum, conf, fsp.outPaths[filesIdx], inspector, reporter, -1, attemptId); // outPath.getParent()
      }

      if (reporter != null) {
        reporter.incrCounter(counterGroup, Operator.HIVE_COUNTER_CREATED_FILES, 1);
      }

    } catch (IOException e) {
      throw new HiveException(e);
    }
  }
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