| 39 | : ScalarExpr(node), tkudu_partition_expr_(node.kudu_partition_expr) {} |
| 40 | |
| 41 | Status KuduPartitionExpr::Init( |
| 42 | const RowDescriptor& row_desc, bool is_entry_point, FragmentState* state) { |
| 43 | RETURN_IF_ERROR(ScalarExpr::Init(row_desc, is_entry_point, state)); |
| 44 | DCHECK_EQ(tkudu_partition_expr_.referenced_columns.size(), children_.size()); |
| 45 | |
| 46 | // Create the KuduPartitioner we'll use to get the partition index for each row. |
| 47 | TableDescriptor* table_desc = |
| 48 | state->desc_tbl().GetTableDescriptor(tkudu_partition_expr_.target_table_id); |
| 49 | DCHECK(table_desc != nullptr); |
| 50 | DCHECK(dynamic_cast<KuduTableDescriptor*>(table_desc)) |
| 51 | << "Target table for KuduPartitioner must be a Kudu table."; |
| 52 | table_desc_ = static_cast<KuduTableDescriptor*>(table_desc); |
| 53 | RETURN_IF_ERROR(ExecEnv::GetInstance()->GetKuduClient( |
| 54 | table_desc_->kudu_master_addresses(), &client_)); |
| 55 | KUDU_RETURN_IF_ERROR(client_->OpenTable(table_desc_->table_name(), &table_), |
| 56 | "Failed to open Kudu table."); |
| 57 | return Status::OK(); |
| 58 | } |
| 59 | |
| 60 | Status KuduPartitionExpr::OpenEvaluator(FunctionContext::FunctionStateScope scope, |
| 61 | RuntimeState* state, ScalarExprEvaluator* eval) const { |
nothing calls this directly
no test coverage detected