Method SetUp

be/src/runtime/data-stream-test.cc:186–250

  ~DataStreamTest() { runtime_state_->ReleaseResources(); }

  virtual void SetUp() {
    exec_env_.reset(new ExecEnv());
    ABORT_IF_ERROR(exec_env_->InitForFeSupport());
    // Always use aggressive decommit for backend tests
    FLAGS_tcmalloc_aggressive_memory_decommit = true;
    ABORT_IF_ERROR(MallocUtil::GetInstance()->Init());
    exec_env_->InitBufferPool(32 * 1024, 1024 * 1024 * 1024, 32 * 1024);
    runtime_state_.reset(new RuntimeState(TQueryCtx(), exec_env_.get()));
    TPlanFragment* fragment = runtime_state_->obj_pool()->Add(new TPlanFragment());
    PlanFragmentCtxPB* fragment_ctx =
        runtime_state_->obj_pool()->Add(new PlanFragmentCtxPB());
    fragment_state_ = runtime_state_->obj_pool()->Add(
        new FragmentState(runtime_state_->query_state(), *fragment, *fragment_ctx));
    mem_pool_.reset(new MemPool(&tracker_));

    // Register a BufferPool client for allocating buffers for row batches.
    ABORT_IF_ERROR(exec_env_->buffer_pool()->RegisterClient(
        "DataStream Test Recvr", nullptr, exec_env_->buffer_reservation(), &tracker_,
        numeric_limits<int64_t>::max(), runtime_state_->runtime_profile(),
        &buffer_pool_client_));

    CreateRowDesc();

    SlotRef* lhs_slot = obj_pool_.Add(new SlotRef(ColumnType(TYPE_BIGINT), 0));
    ASSERT_OK(lhs_slot->Init(RowDescriptor(), true, fragment_state_));
    ordering_exprs_.push_back(lhs_slot);

    tsort_info_.sorting_order = TSortingOrder::LEXICAL;
    tsort_info_.is_asc_order.push_back(true);
    tsort_info_.nulls_first.push_back(true);

    next_instance_id_.set_lo(0);
    next_instance_id_.set_hi(0);
    stream_mgr_ = exec_env_->stream_mgr();

    broadcast_sink_.dest_node_id = DEST_NODE_ID;
    broadcast_sink_.output_partition.type = TPartitionType::UNPARTITIONED;

    random_sink_.dest_node_id = DEST_NODE_ID;
    random_sink_.output_partition.type = TPartitionType::RANDOM;

    hash_sink_.dest_node_id = DEST_NODE_ID;
    hash_sink_.output_partition.type = TPartitionType::HASH_PARTITIONED;
    // there's only one column to partition on
    TExprNode expr_node;
    expr_node.node_type = TExprNodeType::SLOT_REF;
    expr_node.type.types.push_back(TTypeNode());
    expr_node.type.types.back().__isset.scalar_type = true;
    expr_node.type.types.back().scalar_type.type = TPrimitiveType::BIGINT;
    expr_node.num_children = 0;
    TSlotRef slot_ref;
    slot_ref.slot_id = 0;
    expr_node.__set_slot_ref(slot_ref);
    TExpr expr;
    expr.nodes.push_back(expr_node);
    hash_sink_.output_partition.__isset.partition_exprs = true;
    hash_sink_.output_partition.partition_exprs.push_back(expr);