@Override
  public void closeOp(boolean abort) throws HiveException {

    row_count.set(conf.isDeleteOfSplitUpdate() ? 0 : numRows);

    LOG.info("{}: {} written - {}",
            this, conf.isDeleteOfSplitUpdate() ? "delete delta records" : "records", numRows);

    if (!bDynParts && !filesCreated) {
      boolean isTez = "tez".equalsIgnoreCase(
          HiveConf.getVar(hconf, ConfVars.HIVE_EXECUTION_ENGINE));
      Class<?> clazz = conf.getTableInfo().getOutputFileFormatClass();
      boolean isStreaming = StreamingOutputFormat.class.isAssignableFrom(clazz);

      // let empty file generation for mm/acid table as a quick and dirty workaround for HIVE-22941
      if (!isTez || isStreaming || (this.isInsertOverwrite && (conf.isMmTable() || conf.isFullAcidTable()))) {
        createBucketFiles(fsp);
      }
    }

    lastProgressReport = System.currentTimeMillis();
    if (!abort) {
      // If serializer is ThriftJDBCBinarySerDe, then it buffers rows to a certain limit (hive.server2.thrift.resultset.max.fetch.size)
      // and serializes the whole batch when the buffer is full. The serialize returns null if the buffer is not full
      // (the size of buffer is kept track of in the ThriftJDBCBinarySerDe).
      if (conf.isUsingBatchingSerDe()) {
        try {
          recordValue = serializer.serialize(null, inputObjInspectors[0]);
          if (null != fpaths) {
            rowOutWriters = fpaths.outWriters;
            rowOutWriters[0].write(recordValue);
          }
        } catch (SerDeException | IOException e) {
          throw new HiveException(e);
        }
      }
      List<Path> commitPaths = new ArrayList<>();
      for (FSPaths fsp : valToPaths.values()) {
        List<Path> deleteDeltas = new ArrayList<Path>();
        fsp.closeWriters(abort, deleteDeltas);
        // before closing the operator check if statistics gathering is requested
        // and is provided by record writer. this is different from the statistics
        // gathering done in processOp(). In processOp(), for each row added
        // serde statistics about the row is gathered and accumulated in hashmap.
        // this adds more overhead to the actual processing of row. But if the
        // record writer already gathers the statistics, it can simply return the
        // accumulated statistics which will be aggregated in case of spray writers
        if (conf.isGatherStats() && isCollectRWStats) {
          if (conf.getWriteType() == AcidUtils.Operation.NOT_ACID || conf.isMmTable() || isCompactionTable) {
            for (int idx = 0; idx < fsp.outWriters.length; idx++) {
              RecordWriter outWriter = fsp.outWriters[idx];
              if (outWriter != null) {
                SerDeStats stats = ((StatsProvidingRecordWriter) outWriter).getStats();
                if (stats != null) {
                  fsp.addToStat(StatsSetupConst.RAW_DATA_SIZE, stats.getRawDataSize());
                  fsp.addToStat(StatsSetupConst.ROW_COUNT, stats.getRowCount());
                }
              }
            }
          } else {
            for (int i = 0; i < fsp.updaters.length; i++) {
              if (fsp.updaters[i] != null) {
                SerDeStats stats = fsp.updaters[i].getStats();
                if (stats != null) {
                  fsp.addToStat(StatsSetupConst.RAW_DATA_SIZE, stats.getRawDataSize());
                  fsp.addToStat(StatsSetupConst.ROW_COUNT, stats.getRowCount());
                  fsp.addToStat(StatsSetupConst.INSERT_COUNT, stats.getInsertCount());
                  fsp.addToStat(StatsSetupConst.UPDATE_COUNT, stats.getUpdateCount());
                  fsp.addToStat(StatsSetupConst.DELETE_COUNT, stats.getDeleteCount());
                }
              }
            }
          }
        }

        if (isNativeTable()) {
          fsp.commit(fs, commitPaths, deleteDeltas);
        }
      }
      if (conf.isMmTable() || conf.isDirectInsert()) {
        boolean isDelete = AcidUtils.Operation.DELETE.equals(conf.getAcidOperation());
        Utilities.writeCommitManifest(commitPaths, specPath, fs, originalTaskId, conf.getTableWriteId(),
            conf.getStatementId(), unionPath, conf.getInsertOverwrite(), bDynParts, dynamicPartitionSpecs,
            conf.getStaticSpec(), isDelete);
      }
      // Only publish stats if this operator's flag was set to gather stats
      if (conf.isGatherStats()) {
        publishStats();
      }
    } else {
      // Will come here if an Exception was thrown in map() or reduce().
      // Hadoop always call close() even if an Exception was thrown in map() or
      // reduce().
      for (FSPaths fsp : valToPaths.values()) {
        fsp.abortWritersAndUpdaters(fs, abort,
            !autoDelete && isNativeTable() && !conf.isMmTable() && !conf.isDirectInsert());
      }
    }
    fsp = prevFsp = null;
    super.closeOp(abort);
  }
Generate list bucketing directory name from a row.
@param row row to process.
@return directory name.
ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java