| 1705 | } |
| 1706 | |
| 1707 | void Coordinator::FilterState::ApplyUpdate( |
| 1708 | const UpdateFilterParamsPB& params, Coordinator* coord, RpcContext* context) { |
| 1709 | DCHECK(enabled()); |
| 1710 | DCHECK_GT(pending_count_, 0); |
| 1711 | DCHECK(!has_completion_time()); |
| 1712 | if (!has_first_arrival_time()) { |
| 1713 | first_arrival_time_ = coord->query_events_->ElapsedTime(); |
| 1714 | } |
| 1715 | |
| 1716 | --pending_count_; |
| 1717 | if (is_bloom_filter()) { |
| 1718 | DCHECK(params.has_bloom_filter()); |
| 1719 | if (params.bloom_filter().always_true()) { |
| 1720 | // An always_true filter is received. We don't need to wait for other pending |
| 1721 | // backends. |
| 1722 | always_true_filter_received_ = true; |
| 1723 | DisableAndRelease(coord->filter_mem_tracker_, true); |
| 1724 | } else if (params.bloom_filter().always_false()) { |
| 1725 | if (!bloom_filter_.has_log_bufferpool_space()) { |
| 1726 | bloom_filter_ = BloomFilterPB(params.bloom_filter()); |
| 1727 | } |
| 1728 | } else { |
| 1729 | // If the incoming Bloom filter is neither an always true filter nor an |
| 1730 | // always false filter, then it must be the case that a non-empty sidecar slice |
| 1731 | // has been received. Refer to BloomFilter::ToProtobuf() for further details. |
| 1732 | DCHECK(params.bloom_filter().has_directory_sidecar_idx()); |
| 1733 | kudu::Slice sidecar_slice; |
| 1734 | kudu::Status status = context->GetInboundSidecar( |
| 1735 | params.bloom_filter().directory_sidecar_idx(), &sidecar_slice); |
| 1736 | if (!status.ok()) { |
| 1737 | LOG(ERROR) << "Cannot get inbound sidecar: " << status.message().ToString(); |
| 1738 | DisableAndRelease(coord->filter_mem_tracker_, false); |
| 1739 | } else if (bloom_filter_.always_false()) { |
| 1740 | int64_t heap_space = sidecar_slice.size(); |
| 1741 | if (!coord->filter_mem_tracker_->TryConsume(heap_space)) { |
| 1742 | VLOG_QUERY << "Not enough memory to allocate filter: " |
| 1743 | << PrettyPrinter::Print(heap_space, TUnit::BYTES) |
| 1744 | << " (query_id=" << PrintId(coord->query_id()) << ")"; |
| 1745 | // Disable, as one missing update means a correct filter cannot be produced. |
| 1746 | DisableAndRelease(coord->filter_mem_tracker_, false); |
| 1747 | } else { |
| 1748 | bloom_filter_ = params.bloom_filter(); |
| 1749 | bloom_filter_directory_ = sidecar_slice.ToString(); |
| 1750 | } |
| 1751 | } else { |
| 1752 | DCHECK_EQ(bloom_filter_directory_.size(), sidecar_slice.size()); |
| 1753 | BloomFilter::Or(params.bloom_filter(), sidecar_slice.data(), &bloom_filter_, |
| 1754 | reinterpret_cast<uint8_t*>(const_cast<char*>(bloom_filter_directory_.data())), |
| 1755 | sidecar_slice.size()); |
| 1756 | } |
| 1757 | } |
| 1758 | } else if (is_min_max_filter()) { |
| 1759 | DCHECK(params.has_min_max_filter()); |
| 1760 | ColumnType col_type = ColumnType::FromThrift(desc_.src_expr.nodes[0].type); |
| 1761 | VLOG(3) << "Coordinator::FilterState::ApplyUpdate() on minmax." |
| 1762 | << " Current accumulated filter=" << DebugString() |
| 1763 | << ". Incoming min/max param:" |
| 1764 | << " has_filter_id()=" << params.has_filter_id() |
no test coverage detected