@Override
  public void process(Object row, int tag) throws HiveException {
    runTimeNumRows++;
    /* Create list bucketing sub-directory only if stored-as-directories is on. */
    String lbDirName = null;
    lbDirName = (lbCtx == null) ? null : generateListBucketingDirName(row);

    if (!bDynParts && (!filesCreated || isCompactionTable)) {
      if (lbDirName != null) {
        if (valToPaths.get(lbDirName) == null) {
          createNewPaths(null, lbDirName);
        }
      } else if (isCompactionTable) {
        if (conf.isRebalanceRequested()) {
          //For rebalancing compaction, the unencoded bucket id comes in the bucketproperty. It must be encoded before
          //writing the data out
          bucketId = getBucketProperty(row);
          setBucketProperty(hconf, row, bucketId);
        } else {
          int bucketProperty = getBucketProperty(row);
          bucketId = BucketCodec.determineVersion(bucketProperty).decodeWriterId(bucketProperty);
        }
        if (!filesCreatedPerBucket.get(bucketId)) {
          createBucketFilesForCompaction(fsp);
        }
      } else {
        createBucketFiles(fsp);
      }
    }

    try {
      updateProgress();

      // if DP is enabled, get the final output writers and prepare the real output row
      assert inputObjInspectors[0].getCategory() == ObjectInspector.Category.STRUCT
          : "input object inspector is not struct";

      if (bDynParts) {

        // we need to read bucket number which is the last column in value (after partition columns)
        if (conf.getDpSortState().equals(DPSortState.PARTITION_BUCKET_SORTED)) {
          numDynParts += 1;
        }

        // copy the DP column values from the input row to dpVals
        dpVals.clear();
        dpWritables.clear();
        ObjectInspectorUtils.partialCopyToStandardObject(dpWritables, row, dpStartCol,numDynParts,
            (StructObjectInspector) inputObjInspectors[0],ObjectInspectorCopyOption.WRITABLE);

        // get a set of RecordWriter based on the DP column values
        // pass the null value along to the escaping process to determine what the dir should be
        for (Object o : dpWritables) {
          if (o == null || o.toString().length() == 0) {
            dpVals.add(dpCtx.getDefaultPartitionName());
          } else {
            dpVals.add(o.toString());
          }
        }

        String invalidPartitionVal;
        if((invalidPartitionVal = HiveStringUtils.getPartitionValWithInvalidCharacter(dpVals, dpCtx.getWhiteListPattern()))!=null) {
          throw new HiveFatalException("Partition value '" + invalidPartitionVal +
              "' contains a character not matched by whitelist pattern '" +
              dpCtx.getWhiteListPattern().toString() + "'.  " + "(configure with " +
              HiveConf.ConfVars.METASTORE_PARTITION_NAME_WHITELIST_PATTERN.varname + ")");
        }
        fpaths = getDynOutPaths(dpVals, lbDirName);
        dynamicPartitionSpecs.add(fpaths.dpDirForCounters);

        // use SubStructObjectInspector to serialize the non-partitioning columns in the input row
        recordValue = serializer.serialize(row, subSetOI);
      } else {
        if (lbDirName != null) {
          fpaths = valToPaths.get(lbDirName);
          if (fpaths == null) {
            fpaths = createNewPaths(null, lbDirName);
          }
        } else {
          fpaths = fsp;
        }
        recordValue = serializer.serialize(row, inputObjInspectors[0]);
        // if serializer is ThriftJDBCBinarySerDe, then recordValue is null if the buffer is not full (the size of buffer
        // is kept track of in the SerDe)
        if (recordValue == null) {
          return;
        }
      }

      rowOutWriters = fpaths.outWriters;
      // check if all record writers implement statistics. if at least one RW
      // doesn't implement stats interface we will fall back to conventional way
      // of gathering stats
      isCollectRWStats = areAllTrue(statsFromRecordWriter);
      if (conf.isGatherStats() && !isCollectRWStats) {
        SerDeStats stats = serializer.getSerDeStats();
        if (stats != null) {
          fpaths.addToStat(StatsSetupConst.RAW_DATA_SIZE, stats.getRawDataSize());
        }
        fpaths.addToStat(StatsSetupConst.ROW_COUNT, 1);
      }

      if ((++numRows == cntr) && LOG.isInfoEnabled()) {
        cntr = logEveryNRows == 0 ? cntr * 10 : numRows + logEveryNRows;
        if (cntr < 0 || numRows < 0) {
          cntr = 0;
          numRows = 1;
        }
        LOG.info("{}: {} written - {}",
                this, conf.isDeleteOfSplitUpdate() ? "delete delta records" : "records", numRows);
      }

      int writerOffset;
      // This if/else chain looks ugly in the inner loop, but given that it will be 100% the same
      // for a given operator branch prediction should work quite nicely on it.
      // RecordUpdater expects to get the actual row, not a serialized version of it.  Thus we
      // pass the row rather than recordValue.
      if (conf.getWriteType() == AcidUtils.Operation.NOT_ACID || conf.isMmTable() || isCompactionTable) {
        writerOffset = bucketId;
        if (!isCompactionTable) {
          writerOffset = findWriterOffset(row);
        }
        rowOutWriters[writerOffset].write(recordValue);
      } else if (conf.getWriteType() == AcidUtils.Operation.INSERT) {
        fpaths.updaters[findWriterOffset(row)].insert(conf.getTableWriteId(), row);
      } else {
        // TODO I suspect we could skip much of the stuff above this in the function in the case
        // of update and delete.  But I don't understand all of the side effects of the above
        // code and don't want to skip over it yet.

        // Find the bucket id, and switch buckets if need to
        ObjectInspector rowInspector = bDynParts ? subSetOI : outputObjInspector;
        Object recId = ((StructObjectInspector)rowInspector).getStructFieldData(row, recIdField);
        int bucketProperty =
            bucketInspector.get(recIdInspector.getStructFieldData(recId, bucketField));
        int bucketNum =
          BucketCodec.determineVersion(bucketProperty).decodeWriterId(bucketProperty);
        writerOffset = 0;
        if (multiFileSpray) {
          //bucket_num_reducers_acid.q, TestTxnCommands.testMoreBucketsThanReducers()
          if (!bucketMap.containsKey(bucketNum)) {
            String extraMsg = "  (no path info/)" + recId;
            if (fpaths != null && fpaths.finalPaths != null && fpaths.finalPaths.length > 0) {
              extraMsg = "  (finalPaths[0]=" + fpaths.finalPaths[0] + ")/" + recId;
            }
            throw new IllegalStateException("Found bucketNum=" + bucketNum +
              " from data but no mapping in 'bucketMap'." + extraMsg);
          }
          writerOffset = bucketMap.get(bucketNum);
        } else if (!isBucketed) {
          writerOffset = fpaths.createDynamicBucket(bucketNum);
        }
        if (fpaths.updaters[writerOffset] == null) {
          Integer attemptId = getAttemptIdFromTaskId(taskId);
          fpaths.updaters[writerOffset] = HiveFileFormatUtils.getAcidRecordUpdater(jc, conf.getTableInfo(), bucketNum,
              conf, fpaths.outPaths[writerOffset], rowInspector, reporter, 0, attemptId);
          if (LOG.isDebugEnabled()) {
            LOG.debug("Created updater for bucket number " + bucketNum + " using file " +
              fpaths.outPaths[writerOffset]);
          }
        }
        if (conf.getWriteType() == AcidUtils.Operation.UPDATE) {
          fpaths.updaters[writerOffset].update(conf.getTableWriteId(), row);
        } else if (conf.getWriteType() == AcidUtils.Operation.DELETE) {
          fpaths.updaters[writerOffset].delete(conf.getTableWriteId(), row);
        } else {
          throw new HiveException("Unknown write type " + conf.getWriteType().toString());
        }
      }
    } catch (IOException e) {
      LOG.error("Trying to close the writers as an IOException occurred: " + e.getMessage());
      closeWriters(true);
      throw new HiveException(e);
    } catch (SerDeException e) {
      closeWriters(true);
      throw new HiveException(e);
    }
  }
Report status to JT so that JT won't kill this task if closing takes too long
due to too many files to close and the NN is overloaded.

@return true if a new progress update is reported, false otherwise.
ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java