Method GetOutputPartition

be/src/exec/hdfs-table-sink.cc:325–366

}

inline Status HdfsTableSink::GetOutputPartition(RuntimeState* state, const TupleRow* row,
    const string& key, PartitionPair** partition_pair, bool no_more_rows) {
  DCHECK(row != nullptr || key == ROOT_PARTITION_KEY);
  PartitionMap::iterator existing_partition;
  existing_partition = partition_keys_to_output_partitions_.find(key);
  if (existing_partition == partition_keys_to_output_partitions_.end()) {
    // Create a new OutputPartition, and add it to partition_keys_to_output_partitions.
    const HdfsPartitionDescriptor* partition_descriptor = GetPartitionDescriptor(key);
    std::unique_ptr<OutputPartition> partition(new OutputPartition());
    // Build the unique name for this partition from the partition keys, e.g. "j=1/f=foo/"
    // etc.
    RETURN_IF_ERROR(ConstructPartitionInfo(row, partition.get()));
    Status status =
        InitOutputPartition(state, *partition_descriptor, partition.get(),
            no_more_rows);
    if (!status.ok()) {
      // We failed to create the output partition successfully. Clean it up now
      // as it is not added to partition_keys_to_output_partitions_ so won't be
      // cleaned up in Close().
      if (partition->writer.get() != nullptr) partition->writer->Close();
      return status;
    }

    // Indicate that temporary directory is to be deleted after execution.
    bool clean_up_staging_dir =
        !no_more_rows && !ShouldSkipStaging(state, partition.get());

    // Save the partition name so that the coordinator can create the partition
    // directory structure if needed.
    state->dml_exec_state()->AddPartition(
        partition->partition_name, partition_descriptor->id(),
        &table_desc_->hdfs_base_dir(),
        clean_up_staging_dir ? &partition->tmp_hdfs_dir_name : nullptr);

    partition_keys_to_output_partitions_[key].first = std::move(partition);
    *partition_pair = &partition_keys_to_output_partitions_[key];
  } else {
    // Use existing output_partition partition.
    *partition_pair = &existing_partition->second;
  }
  return Status::OK();
}

Status HdfsTableSink::Send(RuntimeState* state, RowBatch* batch) {
  SCOPED_TIMER(profile()->total_time_counter());