| 45 | } |
| 46 | |
| 47 | Status TableSinkBase::Prepare(RuntimeState* state, MemTracker* parent_mem_tracker) { |
| 48 | RETURN_IF_ERROR(DataSink::Prepare(state, parent_mem_tracker)); |
| 49 | partitions_created_counter_ = ADD_COUNTER(profile(), "PartitionsCreated", TUnit::UNIT); |
| 50 | files_created_counter_ = ADD_COUNTER(profile(), "FilesCreated", TUnit::UNIT); |
| 51 | rows_inserted_counter_ = ADD_COUNTER(profile(), "RowsInserted", TUnit::UNIT); |
| 52 | bytes_written_counter_ = ADD_COUNTER(profile(), "BytesWritten", TUnit::BYTES); |
| 53 | encode_timer_ = ADD_TIMER(profile(), "EncodeTimer"); |
| 54 | hdfs_write_timer_ = ADD_TIMER(profile(), "HdfsWriteTimer"); |
| 55 | compress_timer_ = ADD_TIMER(profile(), "CompressTimer"); |
| 56 | |
| 57 | RETURN_IF_ERROR(ScalarExprEvaluator::Create(partition_key_exprs_, state, |
| 58 | state->obj_pool(), expr_perm_pool_.get(), expr_results_pool_.get(), |
| 59 | &partition_key_expr_evals_)); |
| 60 | |
| 61 | // Prepare partition key exprs and gather dynamic partition key exprs. |
| 62 | for (size_t i = 0; i < partition_key_expr_evals_.size(); ++i) { |
| 63 | // Remember non-constant partition key exprs for building hash table of Hdfs files. |
| 64 | if (!partition_key_expr_evals_[i]->root().is_constant()) { |
| 65 | dynamic_partition_key_expr_evals_.push_back(partition_key_expr_evals_[i]); |
| 66 | } |
| 67 | } |
| 68 | |
| 69 | return Status::OK(); |
| 70 | } |
| 71 | |
| 72 | Status TableSinkBase::Open(RuntimeState* state) { |
| 73 | RETURN_IF_ERROR(DataSink::Open(state)); |