Method CreateNewTmpFile

be/src/exec/table-sink-base.cc:278–374

}

Status TableSinkBase::CreateNewTmpFile(RuntimeState* state,
    OutputPartition* output_partition) {
  SCOPED_TIMER(ADD_TIMER(profile(), "TmpFileCreateTimer"));
  string file_name_pattern =
      output_partition->writer->file_extension().empty() ? "$0.$1" : "$0.$1.$2";
  string final_location = Substitute(file_name_pattern,
      output_partition->final_hdfs_file_name_prefix, output_partition->num_files,
      output_partition->writer->file_extension());

  // If ShouldSkipStaging() is true, then the table sink will write the file(s) for this
  // partition to the final location directly. If it is false, the file(s) will be written
  // to a temporary staging location which will be moved by the coordinator to the final
  // location.
  if (ShouldSkipStaging(state, output_partition)) {
    output_partition->current_file_name = final_location;
    output_partition->current_file_final_name = "";
  } else {
    output_partition->current_file_name = Substitute(file_name_pattern,
        output_partition->tmp_hdfs_file_name_prefix, output_partition->num_files,
        output_partition->writer->file_extension());
    // Save the ultimate destination for this file (it will be moved by the coordinator).
    output_partition->current_file_final_name = final_location;
  }
  // Check if tmp_hdfs_file_name exists.
  const char* tmp_hdfs_file_name_cstr =
      output_partition->current_file_name.c_str();

  if (hdfsExists(output_partition->hdfs_connection, tmp_hdfs_file_name_cstr) == 0) {
    return Status(GetHdfsErrorMsg("Temporary HDFS file already exists: ",
        output_partition->current_file_name));
  }

  // This is the block size from the HDFS partition metadata.
  uint64_t block_size = output_partition->partition_descriptor->block_size();
  // hdfsOpenFile takes a 4 byte integer as the block size.
  if (block_size > numeric_limits<int32_t>::max()) {
    return Status(Substitute("HDFS block size must be smaller than 2GB but is configured "
        "in the HDFS partition to $0.", block_size));
  }

  if (block_size == 0) block_size = output_partition->writer->default_block_size();
  if (block_size > numeric_limits<int32_t>::max()) {
    return Status(Substitute("HDFS block size must be smaller than 2GB but the target "
        "table requires $0.", block_size));
  }

  DCHECK_LE(block_size, numeric_limits<int32_t>::max());
  output_partition->tmp_hdfs_file = hdfsOpenFile(output_partition->hdfs_connection,
      tmp_hdfs_file_name_cstr, O_WRONLY, 0, 0, block_size);

  VLOG_FILE << "hdfsOpenFile() file=" << tmp_hdfs_file_name_cstr;
  if (output_partition->tmp_hdfs_file == nullptr) {
    return Status(GetHdfsErrorMsg("Failed to open HDFS file for writing: ",
        output_partition->current_file_name));
  }

  if (IsS3APath(tmp_hdfs_file_name_cstr) ||
      IsABFSPath(tmp_hdfs_file_name_cstr) ||