Method InitOutputPartition

be/src/exec/table-sink-base.cc:201–276

}

Status TableSinkBase::InitOutputPartition(RuntimeState* state,
    const HdfsPartitionDescriptor& partition_descriptor,
    OutputPartition* output_partition, bool empty_partition) {
  BuildHdfsFileNames(partition_descriptor, output_partition);

  if (ShouldSkipStaging(state, output_partition)) {
    // We will be writing to the final file if we're skipping staging, so get a connection
    // to its filesystem.
    RETURN_IF_ERROR(HdfsFsCache::instance()->GetConnection(
        output_partition->final_hdfs_file_name_prefix,
        &output_partition->hdfs_connection));
  } else {
    // Else get a connection to the filesystem of the tmp file.
    RETURN_IF_ERROR(HdfsFsCache::instance()->GetConnection(
        output_partition->tmp_hdfs_file_name_prefix, &output_partition->hdfs_connection));
  }

  output_partition->partition_descriptor = &partition_descriptor;

  if (partition_descriptor.file_format() == THdfsFileFormat::SEQUENCE_FILE ||
      partition_descriptor.file_format() == THdfsFileFormat::AVRO) {
    stringstream error_msg;
    map<int, const char*>::const_iterator i =
        _THdfsFileFormat_VALUES_TO_NAMES.find(partition_descriptor.file_format());
    error_msg << "Writing to table format " << i->second << " is not supported.";
    return Status(error_msg.str());
  }
  if (partition_descriptor.file_format() == THdfsFileFormat::TEXT &&
      state->query_options().__isset.compression_codec &&
      state->query_options().compression_codec.codec != THdfsCompression::NONE) {
    stringstream error_msg;
    error_msg << "Writing to compressed text table is not supported. ";
    return Status(error_msg.str());
  }

  // It is incorrect to initialize a writer if there are no rows to feed it. The writer
  // could incorrectly create an empty file or empty partition.
  // However, for transactional tables we should create a new empty base directory in
  // case of INSERT OVERWRITEs.
  if (empty_partition && (!is_overwrite() || !IsTransactional())) return Status::OK();

  switch (partition_descriptor.file_format()) {
    case THdfsFileFormat::TEXT:
      output_partition->writer.reset(
          new HdfsTextTableWriter(
              this, state, output_partition, &partition_descriptor, table_desc_));
      break;
    case THdfsFileFormat::ICEBERG:
    case THdfsFileFormat::PARQUET:
      output_partition->writer.reset(
          new HdfsParquetTableWriter(
              this, state, output_partition, &partition_descriptor, table_desc_));
      break;
    case THdfsFileFormat::PUFFIN:
      output_partition->writer.reset(
          new PuffinWriter(
              this, state, output_partition, &partition_descriptor, table_desc_));
      break;