| 380 | } |
| 381 | |
| 382 | Status QueryState::InitFilterBank() { |
| 383 | const NetworkAddressPB& this_krpc_address = ExecEnv::GetInstance()->krpc_address(); |
| 384 | int64_t runtime_filters_reservation_bytes = 0; |
| 385 | int fragment_ctx_idx = -1; |
| 386 | const vector<TPlanFragment>& fragments = fragment_info_.fragments; |
| 387 | const vector<TPlanFragmentInstanceCtx>& instance_ctxs = |
| 388 | fragment_info_.fragment_instance_ctxs; |
| 389 | // Add entries for all produced and consumed filters. |
| 390 | unordered_map<int32_t, FilterRegistration> filters; |
| 391 | for (const TPlanFragment& fragment : fragments) { |
| 392 | for (const TPlanNode& plan_node : fragment.plan.nodes) { |
| 393 | if (!plan_node.__isset.runtime_filters) continue; |
| 394 | for (const TRuntimeFilterDesc& filter : plan_node.runtime_filters) { |
| 395 | // Add filter if not already present. |
| 396 | auto it = filters.emplace(filter.filter_id, FilterRegistration(filter)).first; |
| 397 | // Currently hash joins are the only filter sources. Otherwise it must be a filter |
| 398 | // consumer. 'num_producers' is computed later, so don't update that here. |
| 399 | if (!plan_node.__isset.join_node) it->second.has_consumer = true; |
| 400 | } |
| 401 | } |
| 402 | if (fragment.output_sink.__isset.join_build_sink) { |
| 403 | const TJoinBuildSink& join_sink = fragment.output_sink.join_build_sink; |
| 404 | for (const TRuntimeFilterDesc& filter : join_sink.runtime_filters) { |
| 405 | // Add filter if not already present. |
| 406 | filters.emplace(filter.filter_id, FilterRegistration(filter)); |
| 407 | } |
| 408 | } |
| 409 | } |
| 410 | DCHECK(VerifyFiltersProduced(instance_ctxs)) |
| 411 | << "Filters produced by all instances on the same backend should be the same"; |
| 412 | for (const TPlanFragmentInstanceCtx& instance_ctx : instance_ctxs) { |
| 413 | bool first_instance_of_fragment = fragment_ctx_idx == -1 |
| 414 | || fragments[fragment_ctx_idx].idx != instance_ctx.fragment_idx; |
| 415 | if (first_instance_of_fragment) { |
| 416 | ++fragment_ctx_idx; |
| 417 | DCHECK_EQ(fragments[fragment_ctx_idx].idx, instance_ctx.fragment_idx); |
| 418 | } |
| 419 | // TODO: this over-reserves memory a bit in a couple of cases: |
| 420 | // * if different fragments on this backend consume or produce the same filter. |
| 421 | // * if a finstance was chosen not to produce a global broadcast filter. |
| 422 | const TPlanFragment& fragment = fragments[fragment_ctx_idx]; |
| 423 | runtime_filters_reservation_bytes += |
| 424 | fragment.produced_runtime_filters_reservation_bytes; |
| 425 | if (first_instance_of_fragment) { |
| 426 | // Consumed filters are shared between all instances. |
| 427 | runtime_filters_reservation_bytes += |
| 428 | fragment.consumed_runtime_filters_reservation_bytes; |
| 429 | } |
| 430 | for (const TRuntimeFilterSource& produced_filter : instance_ctx.filters_produced) { |
| 431 | auto it = filters.find(produced_filter.filter_id); |
| 432 | DCHECK(it != filters.end()); |
| 433 | FilterRegistration& reg = it->second; |
| 434 | ++reg.num_producers; |
| 435 | |
| 436 | if (produced_filter.__isset.aggregator_desc) { |
| 437 | TRuntimeFilterAggDesc agg_desc = produced_filter.aggregator_desc; |
| 438 | if (reg.need_subaggregation) { |
| 439 | // This filter registration is already set before. |
nothing calls this directly
no test coverage detected