Method FinalizeHdfsDml

be/src/runtime/coordinator.cc:947–991

}

Status Coordinator::FinalizeHdfsDml() {
  // All instances must have reported their final statuses before finalization, which is a
  // post-condition of Wait. If the query was not successful, still try to clean up the
  // staging directory.
  DCHECK(has_called_wait_.Load());
  DCHECK(finalize_params() != nullptr);
  bool is_hive_acid = finalize_params()->__isset.write_id;
  bool is_iceberg_table = finalize_params()->__isset.iceberg_params;

  VLOG_QUERY << "Finalizing query: " << PrintId(query_id());
  SCOPED_TIMER(finalization_timer_);
  Status return_status = UpdateExecState(Status::OK(), nullptr, FLAGS_hostname);
  HdfsTableDescriptor* hdfs_table = nullptr;
  if (return_status.ok()) {
    DCHECK(query_ctx().__isset.desc_tbl_serialized);
    RETURN_IF_ERROR(DescriptorTbl::CreateHdfsTblDescriptor(
            query_ctx().desc_tbl_serialized, finalize_params()->table_id, obj_pool(),
            &hdfs_table));
    DCHECK(hdfs_table != nullptr)
        << "INSERT target table not known in descriptor table: "
        << finalize_params()->table_id;
    if (is_iceberg_table) {
      // No-op. The data files are written by the table sink operators to their final
      // places. We append them to the table in UpdateCatalog().
    }
    // There is no need for finalization for Hive ACID transactional inserts.
    else if (!is_hive_acid) {
      // 'write_id' is NOT set, therefore we need to do some finalization, e.g. moving
      // files or delete old files in case of INSERT OVERWRITE.
      return_status = dml_exec_state_.FinalizeHdfsInsert(*finalize_params(),
          query_ctx().client_request.query_options.s3_skip_insert_staging,
          hdfs_table, query_profile_);
    }
    hdfs_table->ReleaseResources();
  } else if (is_hive_acid) {
    parent_request_state_->AbortTransaction();
  }
  // Cleanup after DML operation.
  if (is_hive_acid || is_iceberg_table) {
    DCHECK(!finalize_params()->__isset.staging_dir);
  } else {
    RETURN_IF_ERROR(DeleteQueryLevelStagingDir());
  }
  return return_status;
}

Status Coordinator::DeleteQueryLevelStagingDir() {
  stringstream staging_dir;