You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2021/01/13 01:08:53 UTC
[impala] 06/18: IMPALA-9643: fix runtime filter race for mt_dop
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch 3.x
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 0a1266ffca099c01b05425e9ce4d01473bfc69f0
Author: Tim Armstrong <ta...@cloudera.com>
AuthorDate: Fri Apr 10 16:36:45 2020 -0700
IMPALA-9643: fix runtime filter race for mt_dop
This patch avoids the race with registration of a
consumer filter by registering all filters upfront
when the filter bank is constructed. Then registration
of producers and consumers hands out references to the
pre-constructed filters.
A nice bonus of this change is that RegisterConsumer()
and RegisterProducer() don't mutate anything and
we can avoid lock acquisitions.
Also adds test infrastructure and fixes TestRuntimeRowFilters to
work with mt_dop=4 (it was accidentally not enabled before). That
mostly involved modifying the tests to use aggregates of counters
instead of picking out lines with regexes.
Testing:
Added a regression test that reliably failed before this
fix. This relies on extending debug actions to allow longer
delays, plus a minor extension to the RUNTIME_PROFILE .test
file parser to handle spaces in counter names.
Ran exhaustive tests.
Change-Id: I194c0d2515b6a0e5474e1c0c8647f0e54dc94397
Reviewed-on: http://gerrit.cloudera.org:8080/15715
Reviewed-by: Tim Armstrong <ta...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
(cherry picked from commit 76e4a17fb379bb232618dccb4ad3504dbe5c945c)
---
be/src/exec/exec-node.cc | 16 ++-
be/src/exec/partitioned-hash-join-builder.cc | 4 +-
be/src/exec/scan-node.cc | 2 +-
be/src/exec/scan-node.h | 4 +-
be/src/runtime/query-state.cc | 28 +++--
be/src/runtime/runtime-filter-bank.cc | 79 +++++++------
be/src/runtime/runtime-filter-bank.h | 79 ++++++++-----
common/thrift/PlanNodes.thrift | 3 +-
.../QueryTest/runtime_row_filter_reservations.test | 50 ++++++++
.../queries/QueryTest/runtime_row_filters.test | 131 +++++++++------------
tests/common/test_result_verifier.py | 2 +-
tests/query_test/test_runtime_filters.py | 12 +-
12 files changed, 252 insertions(+), 158 deletions(-)
diff --git a/be/src/exec/exec-node.cc b/be/src/exec/exec-node.cc
index 54e3e04..b8154b2 100644
--- a/be/src/exec/exec-node.cc
+++ b/be/src/exec/exec-node.cc
@@ -65,6 +65,7 @@
#include "runtime/runtime-state.h"
#include "util/debug-util.h"
#include "util/runtime-profile-counters.h"
+#include "util/string-parser.h"
#include "util/uid-util.h"
#include "common/names.h"
@@ -420,8 +421,19 @@ Status ExecNode::ExecDebugActionImpl(TExecNodePhase::type phase, RuntimeState* s
}
} else {
DCHECK_EQ(debug_options_.action, TDebugAction::DELAY);
- VLOG(1) << "DEBUG_ACTION: Sleeping";
- SleepForMs(100);
+ int64_t sleep_duration_ms = 100;
+ if (!debug_options_.action_param.empty()) {
+ const string& param = debug_options_.action_param;
+ StringParser::ParseResult result;
+ sleep_duration_ms =
+ StringParser::StringToInt<int64_t>(param.c_str(), param.length(), &result);
+ if (result != StringParser::PARSE_SUCCESS || sleep_duration_ms < 0) {
+ return Status(Substitute("Invalid sleep duration: '$0'. "
+ "Only non-negative numbers are allowed.", param));
+ }
+ }
+ VLOG(1) << "DEBUG_ACTION: Sleeping for " << sleep_duration_ms << "ms";
+ SleepForMs(sleep_duration_ms);
}
return Status::OK();
}
diff --git a/be/src/exec/partitioned-hash-join-builder.cc b/be/src/exec/partitioned-hash-join-builder.cc
index 29874f0..accadb3 100644
--- a/be/src/exec/partitioned-hash-join-builder.cc
+++ b/be/src/exec/partitioned-hash-join-builder.cc
@@ -176,7 +176,7 @@ PhjBuilder::PhjBuilder(
DCHECK_GT(sink_config.hash_seed_, 0);
for (const TRuntimeFilterDesc& filter_desc : sink_config.filter_descs_) {
filter_ctxs_.emplace_back();
- filter_ctxs_.back().filter = state->filter_bank()->RegisterFilter(filter_desc, true);
+ filter_ctxs_.back().filter = state->filter_bank()->RegisterProducer(filter_desc);
}
}
@@ -204,7 +204,7 @@ PhjBuilder::PhjBuilder(const PhjBuilderConfig& sink_config,
DCHECK_GT(sink_config.hash_seed_, 0);
for (const TRuntimeFilterDesc& filter_desc : sink_config.filter_descs_) {
filter_ctxs_.emplace_back();
- filter_ctxs_.back().filter = state->filter_bank()->RegisterFilter(filter_desc, true);
+ filter_ctxs_.back().filter = state->filter_bank()->RegisterProducer(filter_desc);
}
}
diff --git a/be/src/exec/scan-node.cc b/be/src/exec/scan-node.cc
index 267f624..a140492 100644
--- a/be/src/exec/scan-node.cc
+++ b/be/src/exec/scan-node.cc
@@ -152,7 +152,7 @@ Status ScanNode::Prepare(RuntimeState* state) {
for (const TRuntimeFilterDesc& filter_desc : plan_node_.tnode_->runtime_filters) {
filter_ctxs_.emplace_back();
FilterContext& filter_ctx = filter_ctxs_.back();
- filter_ctx.filter = state->filter_bank()->RegisterFilter(filter_desc, false);
+ filter_ctx.filter = state->filter_bank()->RegisterConsumer(filter_desc);
// TODO: Enable stats for min-max filters when Kudu exposes info about filters
// (KUDU-2162).
if (filter_ctx.filter->is_bloom_filter()) {
diff --git a/be/src/exec/scan-node.h b/be/src/exec/scan-node.h
index 23aaa88..a27d861 100644
--- a/be/src/exec/scan-node.h
+++ b/be/src/exec/scan-node.h
@@ -188,8 +188,8 @@ class ScanNode : public ExecNode {
/// Waits for runtime filters to arrive, checking every 20ms. Max wait time is specified
/// by the 'runtime_filter_wait_time_ms' flag, which is overridden by the query option
- /// of the same name. Returns true if all filters arrived within the time limit (as
- /// measured from the time of RuntimeFilterBank::RegisterFilter()), false otherwise.
+ /// of the same name. The wait starts from when this function is called. Returns
+ /// true if all filters arrived within the time limit, false otherwise.
bool WaitForRuntimeFilters();
/// Additional state only used by multi-threaded scan node implementations.
diff --git a/be/src/runtime/query-state.cc b/be/src/runtime/query-state.cc
index 432d43f..81beeee 100644
--- a/be/src/runtime/query-state.cc
+++ b/be/src/runtime/query-state.cc
@@ -263,13 +263,25 @@ Status QueryState::InitFilterBank() {
const vector<TPlanFragmentCtx>& fragment_ctxs = fragment_info_.fragment_ctxs;
const vector<TPlanFragmentInstanceCtx>& instance_ctxs =
fragment_info_.fragment_instance_ctxs;
- // Add entries for all filters.
- unordered_map<int32_t, int> produced_filter_counts;
+ // Add entries for all produced and consumed filters.
+ unordered_map<int32_t, FilterRegistration> filters;
for (const TPlanFragmentCtx& fragment_ctx : fragment_ctxs) {
- for (const TPlanNode& plan_node : fragment_ctx.fragment.plan.nodes) {
+ const TPlanFragment& fragment = fragment_ctx.fragment;
+ for (const TPlanNode& plan_node : fragment.plan.nodes) {
if (!plan_node.__isset.runtime_filters) continue;
for (const TRuntimeFilterDesc& filter : plan_node.runtime_filters) {
- produced_filter_counts.emplace(filter.filter_id, 0);
+ // Add filter if not already present.
+ auto it = filters.emplace(filter.filter_id, FilterRegistration(filter)).first;
+ // Currently hash joins are the only filter sources. Otherwise it must be a filter
+ // consumer. 'num_producers' is computed later, so don't update that here.
+ if (!plan_node.__isset.join_node) it->second.has_consumer = true;
+ }
+ }
+ if (fragment.output_sink.__isset.join_build_sink) {
+ const TJoinBuildSink& join_sink = fragment.output_sink.join_build_sink;
+ for (const TRuntimeFilterDesc& filter : join_sink.runtime_filters) {
+ // Add filter if not already present.
+ filters.emplace(filter.filter_id, FilterRegistration(filter));
}
}
}
@@ -292,11 +304,13 @@ Status QueryState::InitFilterBank() {
fragment.consumed_runtime_filters_reservation_bytes;
}
for (const TRuntimeFilterSource& produced_filter : instance_ctx.filters_produced) {
- ++produced_filter_counts[produced_filter.filter_id];
+ auto it = filters.find(produced_filter.filter_id);
+ DCHECK(it != filters.end());
+ ++it->second.num_producers;
}
}
- filter_bank_.reset(new RuntimeFilterBank(
- this, produced_filter_counts, runtime_filters_reservation_bytes));
+ filter_bank_.reset(
+ new RuntimeFilterBank(this, filters, runtime_filters_reservation_bytes));
return filter_bank_->ClaimBufferReservation();
}
diff --git a/be/src/runtime/runtime-filter-bank.cc b/be/src/runtime/runtime-filter-bank.cc
index 5b02ceb..cf1feb9 100644
--- a/be/src/runtime/runtime-filter-bank.cc
+++ b/be/src/runtime/runtime-filter-bank.cc
@@ -59,9 +59,9 @@ const int64_t RuntimeFilterBank::MIN_BLOOM_FILTER_SIZE;
const int64_t RuntimeFilterBank::MAX_BLOOM_FILTER_SIZE;
RuntimeFilterBank::RuntimeFilterBank(QueryState* query_state,
- const unordered_map<int32_t, int>& produced_filter_counts,
+ const unordered_map<int32_t, FilterRegistration>& filters,
long total_filter_mem_required)
- : filters_(BuildFilterMap(produced_filter_counts)),
+ : filters_(BuildFilterMap(filters, &obj_pool_)),
query_state_(query_state),
filter_mem_tracker_(query_state->obj_pool()->Add(new MemTracker(
-1, "Runtime Filter Bank", query_state->query_mem_tracker(), false))),
@@ -73,10 +73,23 @@ RuntimeFilterBank::~RuntimeFilterBank() {}
unordered_map<int32_t, unique_ptr<RuntimeFilterBank::PerFilterState>>
RuntimeFilterBank::BuildFilterMap(
- const unordered_map<int32_t, int>& produced_filter_counts) {
+ const unordered_map<int32_t, FilterRegistration>& filters, ObjectPool* obj_pool) {
unordered_map<int32_t, unique_ptr<PerFilterState>> result;
- for (auto& entry : produced_filter_counts) {
- result.emplace(entry.first, make_unique<PerFilterState>(entry.second));
+ for (auto& entry : filters) {
+ const FilterRegistration reg = entry.second;
+ RuntimeFilter* result_filter = nullptr;
+ RuntimeFilter* consumed_filter = nullptr;
+ if (reg.has_consumer) {
+ VLOG(3) << "registered consumer filter " << reg.desc.filter_id;
+ consumed_filter =
+ obj_pool->Add(new RuntimeFilter(reg.desc, reg.desc.filter_size_bytes));
+ }
+ if (reg.num_producers > 0) {
+ result_filter =
+ obj_pool->Add(new RuntimeFilter(reg.desc, reg.desc.filter_size_bytes));
+ }
+ result.emplace(entry.first,
+ make_unique<PerFilterState>(reg.num_producers, result_filter, consumed_filter));
}
return result;
}
@@ -96,41 +109,31 @@ Status RuntimeFilterBank::ClaimBufferReservation() {
return Status::OK();
}
-RuntimeFilter* RuntimeFilterBank::RegisterFilter(
- const TRuntimeFilterDesc& filter_desc, bool is_producer) {
+RuntimeFilter* RuntimeFilterBank::RegisterProducer(
+ const TRuntimeFilterDesc& filter_desc) {
auto it = filters_.find(filter_desc.filter_id);
DCHECK(it != filters_.end()) << "Filter ID " << filter_desc.filter_id
<< " not registered";
PerFilterState* fs = it->second.get();
- RuntimeFilter* ret = nullptr;
- lock_guard<SpinLock> l(fs->lock);
- if (is_producer) {
- if (fs->produced_filter.result_filter == nullptr) {
- ret = obj_pool_.Add(new RuntimeFilter(filter_desc, filter_desc.filter_size_bytes));
- fs->produced_filter.result_filter = ret;
- } else {
- ret = fs->produced_filter.result_filter;
- DCHECK_EQ(filter_desc.filter_size_bytes, ret->filter_size());
- }
- } else {
- if (fs->consumed_filter == nullptr) {
- ret = obj_pool_.Add(new RuntimeFilter(filter_desc, filter_desc.filter_size_bytes));
- // The filter bank may have already been cancelled. In that case, still allocate the
- // filter but cancel it immediately, so that callers of RuntimeFilterBank don't need
- // to have separate handling of that case.
- if (cancelled_) ret->Cancel();
- fs->consumed_filter = ret;
- VLOG(2) << "registered consumer filter " << filter_desc.filter_id;
- } else {
- // The filter has already been registered in this filter bank by another
- // fragment instance or target node.
- ret = fs->consumed_filter;
- VLOG_QUERY << "re-registered consumer filter " << filter_desc.filter_id;
- }
- }
+ DCHECK(fs->produced_filter.result_filter != nullptr);
+ RuntimeFilter* ret = fs->produced_filter.result_filter;
+ DCHECK_EQ(filter_desc.filter_size_bytes, ret->filter_size());
return ret;
}
+RuntimeFilter* RuntimeFilterBank::RegisterConsumer(
+ const TRuntimeFilterDesc& filter_desc) {
+ auto it = filters_.find(filter_desc.filter_id);
+ DCHECK(it != filters_.end()) << "Filter ID " << filter_desc.filter_id
+ << " not registered";
+ PerFilterState* fs = it->second.get();
+ DCHECK(fs->consumed_filter != nullptr)
+ << "Consumed filters must be created in constructor";
+ VLOG(3) << "Consumer registered for filter " << filter_desc.filter_id;
+ DCHECK_EQ(filter_desc.filter_size_bytes, fs->consumed_filter->filter_size());
+ return fs->consumed_filter;
+}
+
void RuntimeFilterBank::UpdateFilterCompleteCb(
const RpcController* rpc_controller, const UpdateFilterResultPB* res) {
const kudu::Status controller_status = rpc_controller->status();
@@ -463,8 +466,10 @@ void RuntimeFilterBank::Close() {
filter_mem_tracker_->Close();
}
-RuntimeFilterBank::ProducedFilter::ProducedFilter(int pending_producers)
- : pending_producers(pending_producers) {}
+RuntimeFilterBank::ProducedFilter::ProducedFilter(
+ int pending_producers, RuntimeFilter* result_filter)
+ : result_filter(result_filter), pending_producers(pending_producers) {}
-RuntimeFilterBank::PerFilterState::PerFilterState(int pending_producers)
- : produced_filter(pending_producers) {}
+RuntimeFilterBank::PerFilterState::PerFilterState(
+ int pending_producers, RuntimeFilter* result_filter, RuntimeFilter* consumed_filter)
+ : produced_filter(pending_producers, result_filter), consumed_filter(consumed_filter) {}
diff --git a/be/src/runtime/runtime-filter-bank.h b/be/src/runtime/runtime-filter-bank.h
index e8977e4..8532da4 100644
--- a/be/src/runtime/runtime-filter-bank.h
+++ b/be/src/runtime/runtime-filter-bank.h
@@ -52,21 +52,31 @@ class TBloomFilter;
class TRuntimeFilterDesc;
class TQueryCtx;
+/// Metadata about each filter required to initialize the RuntimeFilterBank for a query
+/// running on a backend.
+struct FilterRegistration {
+ FilterRegistration(const TRuntimeFilterDesc& desc) : desc(desc) {}
+
+ const TRuntimeFilterDesc& desc;
+
+ // Whether or not there is a consumer of the filter on this backend.
+ bool has_consumer = false;
+
+ // The number of producers of this filter executing on the backend.
+ int num_producers = 0;
+};
+
/// RuntimeFilters are produced and consumed by plan nodes at run time to propagate
/// predicates across the plan tree dynamically. Each query backend manages its
/// filters with a RuntimeFilterBank which provides low-synchronization access to filter
/// objects and data structures.
///
-/// A RuntimeFilterBank manages both production and consumption of filters. In the case
-/// where a given filter is both consumed and produced by the same backend, the
-/// RuntimeFilterBank treats each filter independently.
-///
-/// All filters must be registered with the filter bank via RegisterFilter(). Local plan
-/// fragments update the filters by calling UpdateFilterFromLocal(), with either a bloom
-/// filter or a min-max filter, depending on the filter's type. The 'bloom_filter' or
-/// 'min_max_filter' that is passed into UpdateFilterFromLocal() must have been allocated
-/// by AllocateScratch*Filter(); this allows RuntimeFilterBank to manage all memory
-/// associated with filters.
+/// All producers and consumers of filters must register via RegisterProducer() and
+/// RegisterConsumer(). Local plan fragments update the filters by calling
+/// UpdateFilterFromLocal(), with either a bloom filter or a min-max filter, depending
+/// on the filter's type. The 'bloom_filter' or 'min_max_filter' that is passed into
+/// UpdateFilterFromLocal() must have been allocated by AllocateScratch*Filter(); this
+/// allows RuntimeFilterBank to manage all memory associated with filters.
///
/// Filters are aggregated, first locally in this RuntimeFilterBank, if there are multiple
/// producers, and then made available to consumers after PublishGlobalFilter() has been
@@ -80,11 +90,9 @@ class TQueryCtx;
/// nor the thread that may call RuntimeFilter::Eval() need to coordinate in any way.
class RuntimeFilterBank {
public:
- /// 'produced_filter_counts': contains an entry for every filter produced or consumed
- /// on this backend, along with the number of producers (0 if there are only consumers
- /// on this backend).
+ /// 'filters': contains an entry for every filter produced or consumed on this backend.
RuntimeFilterBank(QueryState* query_state,
- const boost::unordered_map<int32_t, int>& produced_filter_counts,
+ const boost::unordered_map<int32_t, FilterRegistration>& filters,
long total_filter_mem_required);
// Define destructor in runtime-filter-bank.cc so that we can compile with only a
@@ -100,10 +108,15 @@ class RuntimeFilterBank {
/// before Close().
Status ClaimBufferReservation() WARN_UNUSED_RESULT;
- /// Registers a filter that will either be produced (is_producer == false) or consumed
- /// (is_producer == true) by fragments that share this QueryState. The filter
- /// bloom_filter itself is unallocated until the first call to PublishGlobalFilter().
- RuntimeFilter* RegisterFilter(const TRuntimeFilterDesc& filter_desc, bool is_producer);
+ /// Registers a producer of a filter. The filter must have been registered when
+ /// constructing the RuntimeFilterBank. The storage for the filter is not allocated;
+ /// the caller must call AllocateScratch*Filter() to allocate the actual filter.
+ RuntimeFilter* RegisterProducer(const TRuntimeFilterDesc& filter_desc);
+
+ /// Registers a consumer of a filter. The filter must have been registered when
+ /// constructing the RuntimeFilterBank. The consumer can use the returned RuntimeFilter
+ /// to check for the filter's arrival.
+ RuntimeFilter* RegisterConsumer(const TRuntimeFilterDesc& filter_desc);
/// Updates a filter's 'bloom_filter' or 'min_max_filter' which has been produced by
/// some operator in a local fragment instance. At most one of 'bloom_filter' and
@@ -125,8 +138,7 @@ class RuntimeFilterBank {
/// Returns a bloom_filter that can be used by an operator to produce a local filter,
/// which may then be used in UpdateFilterFromLocal(). The memory returned is owned by
/// the RuntimeFilterBank and should not be deleted by the caller. The filter identified
- /// by 'filter_id' must have been previously registered as a 'producer' by
- /// RegisterFilter().
+ /// by 'filter_id' must have been previously registered by RegisterProducer().
///
/// If memory allocation for the filter fails, or if Close() has been called first,
/// returns NULL.
@@ -152,7 +164,8 @@ class RuntimeFilterBank {
struct PerFilterState;
static boost::unordered_map<int32_t, std::unique_ptr<PerFilterState>> BuildFilterMap(
- const boost::unordered_map<int32_t, int>& produced_filter_counts);
+ const boost::unordered_map<int32_t, FilterRegistration>& filters,
+ ObjectPool* obj_pool);
/// Acquire locks for all filters, returning them to the caller.
std::vector<std::unique_lock<SpinLock>> LockAllFilters();
@@ -162,12 +175,11 @@ class RuntimeFilterBank {
/// Data tracked for each produced filter in the filter bank.
struct ProducedFilter {
- ProducedFilter(int pending_producers);
+ ProducedFilter(int pending_producers, RuntimeFilter* result_filter);
- /// The initial filter returned from RegisterFilter() with metadata about the filter.
- /// Initialised when RegisterFilter(is_producer=true) is called for this filter id.
- /// Not modified by producers.
- RuntimeFilter* result_filter = nullptr;
+ /// The initial filter returned from RegisterProducer() metadata about the filter.
+ /// Not modified by producers. Owned by 'obj_pool_'.
+ RuntimeFilter* const result_filter;
// The expected number of instances of the filter yet to arrive, i.e. additional
// UpdateFilterFromLocal() calls expected.
@@ -184,7 +196,14 @@ class RuntimeFilterBank {
/// separately to help with scalability. Aligned so that each lock is on a separate
/// cache line.
struct PerFilterState {
- PerFilterState(int pending_producers);
+ /// pending_producers: the number of producers that will call UpdateFilterFromLocal().
+ /// result_filter: the initial filter that will be returned to producers. Non-NULL if
+ /// there are any producers. Must be owned by 'obj_pool_'.
+ /// consumed_filter: the filter that will be returned to consumers. Non-NULL if there
+ /// are any consumers. Must be owned by 'obj_pool_'.
+ PerFilterState(int pending_producers, RuntimeFilter* result_filter,
+ RuntimeFilter* consumed_filter);
+
/// Lock protecting the structures in this PerFilterState. If multiple locks are
/// acquired, they must be acquired in the 'filters_' map iteration order.
SpinLock lock;
@@ -192,13 +211,13 @@ class RuntimeFilterBank {
/// State of a filter that will be produced by this filter bank.
ProducedFilter produced_filter;
- /// The filter that is returned to consumers that call RegisterFilter(producer=false).
- /// Initialised when RegisterFilter(producer=false) is called for this filter id.
+ /// The filter that is returned to consumers that call RegisterConsumer().
+ /// Initialised in the constructor if there are consumer filters on this backend.
///
/// For broadcast joins, SetFilter() must be called while holding 'lock' and after
/// checking HasFilter() to avoid SetFilter() being called multiple times for
/// broadcast join filters.
- RuntimeFilter* consumed_filter = nullptr;
+ RuntimeFilter* const consumed_filter;
/// Contains references to all the bloom filters generated. Used in Close() to safely
/// release all memory allocated for BloomFilters.
diff --git a/common/thrift/PlanNodes.thrift b/common/thrift/PlanNodes.thrift
index 17e353e..2d0f0e5 100644
--- a/common/thrift/PlanNodes.thrift
+++ b/common/thrift/PlanNodes.thrift
@@ -76,7 +76,8 @@ enum TDebugAction {
// A floating point number in range [0.0, 1.0] that gives the probability of denying
// each reservation increase request after the initial reservation.
SET_DENY_RESERVATION_PROBABILITY = 4
- // Delay for a short amount of time: 100ms
+ // Delay for a short amount of time: 100ms or the specified number of seconds in the
+ // optional parameter.
DELAY = 5
}
diff --git a/testdata/workloads/functional-query/queries/QueryTest/runtime_row_filter_reservations.test b/testdata/workloads/functional-query/queries/QueryTest/runtime_row_filter_reservations.test
new file mode 100644
index 0000000..0955a0e
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/runtime_row_filter_reservations.test
@@ -0,0 +1,50 @@
+====
+---- QUERY
+####################################################
+# Test case 1: Query is not admitted if it exceeds the
+# mem requirement after accounting for the memory
+# required by runtime filters.
+####################################################
+SET RUNTIME_FILTER_MODE=GLOBAL;
+SET RUNTIME_FILTER_WAIT_TIME_MS=$RUNTIME_FILTER_WAIT_TIME_MS;
+SET RUNTIME_FILTER_MIN_SIZE=128MB;
+SET RUNTIME_FILTER_MAX_SIZE=500MB;
+# Query would have been admitted if memory for runtime filters was not accounted for.
+SET BUFFER_POOL_LIMIT=290MB;
+select STRAIGHT_JOIN * from alltypes a join [SHUFFLE] alltypes b
+ on a.month = b.id and b.int_col = -3
+---- RESULTS
+====
+---- QUERY
+SET RUNTIME_FILTER_MODE=GLOBAL;
+SET RUNTIME_FILTER_WAIT_TIME_MS=$RUNTIME_FILTER_WAIT_TIME_MS;
+SET RUNTIME_FILTER_MIN_SIZE=128MB;
+SET RUNTIME_FILTER_MAX_SIZE=500MB;
+# Disable the estimation of cardinality for an hdfs table withot stats.
+SET DISABLE_HDFS_NUM_ROWS_ESTIMATE=1;
+# Query would have been admitted if memory for runtime filters was not accounted for.
+SET BUFFER_POOL_LIMIT=290MB;
+select STRAIGHT_JOIN * from alltypes a join [SHUFFLE] alltypes b
+ on a.month = b.id and b.int_col = -3
+---- RESULTS
+---- CATCH
+row_regex:.*minimum memory reservation on backend '.*' is greater than memory available to
+ the query for buffer reservations\. Increase the buffer_pool_limit to 290.17 MB\. See
+ the query profile for more information about the per-node memory requirements\.
+====
+---- QUERY
+# Confirm that with broadcast join, memory limit is not hit.
+SET RUNTIME_FILTER_MODE=GLOBAL;
+SET RUNTIME_FILTER_WAIT_TIME_MS=$RUNTIME_FILTER_WAIT_TIME_MS;
+SET RUNTIME_FILTER_MIN_SIZE=128MB;
+SET RUNTIME_FILTER_MAX_SIZE=500MB;
+# This would run perfectly with just enough memory provided by the buffer pool.
+SET BUFFER_POOL_LIMIT=295MB;
+select STRAIGHT_JOIN * from alltypes a join [SHUFFLE] alltypes b
+ on a.month = b.id and b.int_col = -3
+---- RESULTS
+---- RUNTIME_PROFILE
+row_regex: .*Filter 0 \(128.00 MB\).*
+aggregation(SUM, Files processed): 24
+aggregation(SUM, Files rejected): 24
+====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/runtime_row_filters.test b/testdata/workloads/functional-query/queries/QueryTest/runtime_row_filters.test
index 1365ef8..583e627 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/runtime_row_filters.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/runtime_row_filters.test
@@ -4,8 +4,6 @@
# Test case 1: broadcast join.
# Basic filtering use case: p's partition columns thanks to an implicit relationship
# between join column and build-side predicates.
-# Without filtering, expect 7300 / 3 = 2433 rows.
-# With filtering, expect 618 / 3 = 206 rows.
####################################################
SET RUNTIME_FILTER_WAIT_TIME_MS=$RUNTIME_FILTER_WAIT_TIME_MS;
@@ -15,10 +13,13 @@ on p.id = b.id and b.month = 1 and b.string_col = "1"
---- RESULTS
1
---- RUNTIME_PROFILE
-row_regex: .*RowsRead: 2.43K .*
+# 7300 rows will be read from alltypes and 2 will be read from alltypestiny.
+# None will be filtered because runtime filtering is disabled.
+aggregation(SUM, RowsRead): 7302
+aggregation(SUM, Rows rejected): 0
====
---- QUERY
-# Now turn on local filtering: we expect to see a reduction in scan volume.
+# Now turn on local filtering: we expect to see most rows filtered out.
SET RUNTIME_FILTER_WAIT_TIME_MS=$RUNTIME_FILTER_WAIT_TIME_MS;
SET RUNTIME_FILTER_MODE=LOCAL;
select STRAIGHT_JOIN count(*) from alltypes p join [BROADCAST] alltypestiny b
@@ -26,15 +27,14 @@ on p.id = b.id and b.month = 1 and b.string_col = "1"
---- RESULTS
1
---- RUNTIME_PROFILE
-row_regex: .*Rows rejected: 2.43K .*
+aggregation(SUM, RowsRead): 7302
+aggregation(SUM, Rows rejected): 7299
====
---- QUERY
####################################################
# Test case 2: shuffle join - test for filter propagation (or lack thereof in LOCAL mode).
-# Without filtering, expect 7300 / 3 = 2433 rows.
-# With filtering, expect 618 / 3 = 206 rows.
####################################################
# Local mode. Filters won't be propagated to scan, so scans will read all rows.
@@ -46,10 +46,14 @@ on p.id = b.int_col and b.month = 1 and b.string_col = "1"
---- RESULTS
1
---- RUNTIME_PROFILE
-row_regex: .*RowsRead: 2.43K .*
+# 7300 rows will be read from alltypes and 2 will be read from alltypestiny.
+# None will be filtered because the filter can't be propagated to the scan.
+aggregation(SUM, RowsRead): 7302
+aggregation(SUM, Rows rejected):0
====
---- QUERY
-# Shuffle join, global mode. Expect filters to be propagated.
+# Shuffle join, global mode. Expect filters to be propagated and rows to be
+# filtered in the scan.
SET RUNTIME_FILTER_WAIT_TIME_MS=$RUNTIME_FILTER_WAIT_TIME_MS;
SET RUNTIME_FILTER_MODE=GLOBAL;
select STRAIGHT_JOIN count(*) from alltypes p join [SHUFFLE] alltypestiny b
@@ -57,7 +61,8 @@ on p.id = b.int_col and b.month = 1 and b.string_col = "1"
---- RESULTS
1
---- RUNTIME_PROFILE
-row_regex: .*Rows rejected: 2.43K .*
+aggregation(SUM, RowsRead): 7302
+aggregation(SUM, Rows rejected): 7299
====
@@ -65,7 +70,7 @@ row_regex: .*Rows rejected: 2.43K .*
####################################################
# Test case 3: two-hop filter chain with BROADCAST
# joins.
-# Without filtering in left-most scan, expect 7300 / 3 = 2433 rows.
+# Without filtering in left-most scan, expect 7300 rows.
# With filtering, expect 0 rows.
####################################################
@@ -80,7 +85,7 @@ select STRAIGHT_JOIN count(*) from alltypes a
---- RESULTS
0
---- RUNTIME_PROFILE
-row_regex: .*Files rejected: 0 .*
+aggregation(SUM, Files rejected): 0
====
---- QUERY
# Global mode. Scan of 'b' will receive highly effective filter, and will propagate that
@@ -94,7 +99,7 @@ select STRAIGHT_JOIN count(*) from alltypes a
---- RESULTS
0
---- RUNTIME_PROFILE
-row_regex: .*Files rejected: 8 .*
+aggregation(SUM, Files rejected): 48
====
@@ -102,8 +107,7 @@ row_regex: .*Files rejected: 8 .*
####################################################
# Test case 4: complex filter expressions. The join predicate matches nothing, but
# isn't simplified by the planner before execution.
-# Without filtering in left-most scan, expect 7300 / 3 = 2433 rows.
-# With filtering, expect 0 rows.
+# Filtering in the scan of alltypes should reject all 7300 rows.
####################################################
SET RUNTIME_FILTER_WAIT_TIME_MS=$RUNTIME_FILTER_WAIT_TIME_MS;
@@ -114,7 +118,7 @@ select STRAIGHT_JOIN count(*) from alltypes a
---- RESULTS
0
---- RUNTIME_PROFILE
-row_regex: .*Rows rejected: 2.43K .*
+aggregation(SUM, Rows rejected): 7300
====
@@ -132,8 +136,8 @@ select STRAIGHT_JOIN count(*) from alltypes a
---- RESULTS
0
---- RUNTIME_PROFILE
-row_regex: .*FiltersReceived: 0 .*
-row_regex: .*Rows rejected: 2.43K .*
+aggregation(SUM, FiltersReceived): 0
+aggregation(SUM, Rows rejected): 7300
====
---- QUERY
# Global mode. Coordinator should report 0 filter updates received.
@@ -145,8 +149,8 @@ select STRAIGHT_JOIN count(*) from alltypes a
---- RESULTS
0
---- RUNTIME_PROFILE
-row_regex: .*FiltersReceived: 0 .*
-row_regex: .*Rows rejected: 2.43K .*
+aggregation(SUM, FiltersReceived): 0
+aggregation(SUM, Rows rejected): 7300
====
@@ -164,7 +168,7 @@ select STRAIGHT_JOIN count(*) from alltypes a
---- RESULTS
0
---- RUNTIME_PROFILE
-row_regex: .*FiltersReceived: 0 .*
+aggregation(SUM, FiltersReceived): 0
====
---- QUERY
# Global mode. Coordinator should report 1 filter updates per backend.
@@ -176,7 +180,7 @@ select STRAIGHT_JOIN count(*) from alltypes a
---- RESULTS
0
---- RUNTIME_PROFILE
-row_regex: .*FiltersReceived: 3 .*
+aggregation(SUM, FiltersReceived): 3
====
---- QUERY
@@ -192,7 +196,7 @@ select STRAIGHT_JOIN count(*) from alltypes a
---- RESULTS
4
---- RUNTIME_PROFILE
-row_regex: .*Rows rejected: 2.43K .*
+aggregation(SUM, Rows rejected): 7296
====
@@ -209,7 +213,8 @@ select STRAIGHT_JOIN count(*) from alltypes a
---- RESULTS
7300
---- RUNTIME_PROFILE
-row_regex: .*RowsReturned: 2.43K .*
+aggregation(SUM, ProbeRows): 7300
+aggregation(SUM, Rows rejected): 0
====
@@ -226,7 +231,7 @@ select STRAIGHT_JOIN count(*) from alltypes a
---- RESULTS
8
---- RUNTIME_PROFILE
-row_regex: .*Rows rejected: 2.43K .*
+aggregation(SUM, Rows rejected): 7296
====
@@ -243,7 +248,7 @@ select STRAIGHT_JOIN count(*) from alltypes a
---- RESULTS
7304
---- RUNTIME_PROFILE
-row_regex: .*RowsReturned: 2.43K .*
+aggregation(SUM, ProbeRows): 7300
====
@@ -342,7 +347,8 @@ where l_orderkey = 965 and l_extendedprice * l_tax = p_retailprice;
####################################################
# Test case 16: PHJ nodes that spill should still produce filters.
# Run this for Parquet only to avoid variable memory
-# consumption / spilling behaviour.
+# consumption / spilling behaviour. The filter is not effective and
+# will be disabled after processing 16k rows.
####################################################
SET BUFFER_POOL_LIMIT=80m;
@@ -359,7 +365,7 @@ BIGINT
---- RUNTIME_PROFILE
row_regex: .*SpilledPartitions: [1-9]\d* .*
row_regex: .*Rows processed: 16.38K.*
-row_regex: .*Rows rejected: 0 .*
+aggregation(SUM, Rows rejected): 0
row_regex: .*1 of 1 Runtime Filter Published.*
====
@@ -381,56 +387,33 @@ select straight_join count(1)
10
---- RUNTIME_PROFILE
row_regex: .*1 of 1 Runtime Filter Published.*
-row_regex: .*Rows rejected: 2.43K \(2432\).*
+aggregation(SUM, Rows rejected): 7297
====
-
-
---- QUERY
####################################################
-# Test case 18: Query is not admitted if it exceeds the
-# mem requirement after accounting for the memory
-# required by runtime filters.
+# Test case 17: Regression test for IMPALA-9643.
+# A LOCAL filter must be propagated correctly if the consuming SCAN registers
+# itself after it is produced by the join and set in the filter bank.
####################################################
-SET RUNTIME_FILTER_MODE=GLOBAL;
-SET RUNTIME_FILTER_WAIT_TIME_MS=$RUNTIME_FILTER_WAIT_TIME_MS;
-SET RUNTIME_FILTER_MIN_SIZE=128MB;
-SET RUNTIME_FILTER_MAX_SIZE=500MB;
-# Query would have been admitted if memory for runtime filters was not accounted for.
-SET BUFFER_POOL_LIMIT=290MB;
-select STRAIGHT_JOIN * from alltypes a join [SHUFFLE] alltypes b
- on a.month = b.id and b.int_col = -3
----- RESULTS
-====
----- QUERY
-SET RUNTIME_FILTER_MODE=GLOBAL;
-SET RUNTIME_FILTER_WAIT_TIME_MS=$RUNTIME_FILTER_WAIT_TIME_MS;
-SET RUNTIME_FILTER_MIN_SIZE=128MB;
-SET RUNTIME_FILTER_MAX_SIZE=500MB;
-# Disable the estimation of cardinality for an hdfs table withot stats.
-SET DISABLE_HDFS_NUM_ROWS_ESTIMATE=1;
-# Query would have been admitted if memory for runtime filters was not accounted for.
-SET BUFFER_POOL_LIMIT=290MB;
-select STRAIGHT_JOIN * from alltypes a join [SHUFFLE] alltypes b
- on a.month = b.id and b.int_col = -3
----- RESULTS
----- CATCH
-row_regex:.*minimum memory reservation on backend '.*' is greater than memory available to
- the query for buffer reservations\. Increase the buffer_pool_limit to 290.17 MB\. See
- the query profile for more information about the per-node memory requirements\.
-====
----- QUERY
-# Confirm that with broadcast join, memory limit is not hit.
-SET RUNTIME_FILTER_MODE=GLOBAL;
-SET RUNTIME_FILTER_WAIT_TIME_MS=$RUNTIME_FILTER_WAIT_TIME_MS;
-SET RUNTIME_FILTER_MIN_SIZE=128MB;
-SET RUNTIME_FILTER_MAX_SIZE=500MB;
-# This would run perfectly with just enough memory provided by the buffer pool.
-SET BUFFER_POOL_LIMIT=295MB;
-select STRAIGHT_JOIN * from alltypes a join [SHUFFLE] alltypes b
- on a.month = b.id and b.int_col = -3
+set RUNTIME_FILTER_WAIT_TIME_MS=$RUNTIME_FILTER_WAIT_TIME_MS;
+# Delay preparation of the leftmost scan node so that it registers itself as a filter
+# consumer late.
+set DEBUG_ACTION="0:PREPARE:DELAY@1000";
+select straight_join f.id, f.int_col
+from alltypesagg f
+ join /*+broadcast*/ alltypestiny d on f.id = d.id
---- RESULTS
+0,NULL
+0,NULL
+1,1
+2,2
+3,3
+4,4
+5,5
+6,6
+7,7
+---- TYPES
+INT,INT
---- RUNTIME_PROFILE
-row_regex: .*Filter 0 \(128.00 MB\).*
-row_regex: .*Files processed: 8.*
-row_regex: .*Files rejected: 8.*
+aggregation(SUM, Rows rejected): 10991
====
diff --git a/tests/common/test_result_verifier.py b/tests/common/test_result_verifier.py
index df3da73..875c730 100644
--- a/tests/common/test_result_verifier.py
+++ b/tests/common/test_result_verifier.py
@@ -505,7 +505,7 @@ def parse_result_rows(exec_result):
# Currently, the only implemented function is SUM and only integers are supported.
AGGREGATION_PREFIX_PATTERN = 'aggregation\('
AGGREGATION_PREFIX = re.compile(AGGREGATION_PREFIX_PATTERN)
-AGGREGATION_SYNTAX_MATCH_PATTERN = 'aggregation\((\w+)[ ]*,[ ]*(\w+)\):[ ]*(\d+)'
+AGGREGATION_SYNTAX_MATCH_PATTERN = 'aggregation\((\w+)[ ]*,[ ]*([^)]+)\):[ ]*(\d+)'
def try_compile_aggregation(row_string):
"""
diff --git a/tests/query_test/test_runtime_filters.py b/tests/query_test/test_runtime_filters.py
index 5e53c6e..5e4e506 100644
--- a/tests/query_test/test_runtime_filters.py
+++ b/tests/query_test/test_runtime_filters.py
@@ -241,5 +241,15 @@ class TestRuntimeRowFilters(ImpalaTestSuite):
def test_row_filters(self, vector):
new_vector = deepcopy(vector)
new_vector.get_value('exec_option')['mt_dop'] = vector.get_value('mt_dop')
- self.run_test_case('QueryTest/runtime_row_filters', vector,
+ self.run_test_case('QueryTest/runtime_row_filters', new_vector,
+ test_file_vars={'$RUNTIME_FILTER_WAIT_TIME_MS': str(WAIT_TIME_MS)})
+
+ def test_row_filter_reservation(self, vector):
+ """Test handling of runtime filter memory reservations. Tuned for mt_dop=0."""
+ mt_dop = vector.get_value('mt_dop')
+ if mt_dop != 0:
+ pytest.skip("Memory reservations tuned for mt_dop=0")
+ new_vector = deepcopy(vector)
+ new_vector.get_value('exec_option')['mt_dop'] = mt_dop
+ self.run_test_case('QueryTest/runtime_row_filter_reservations', new_vector,
test_file_vars={'$RUNTIME_FILTER_WAIT_TIME_MS' : str(WAIT_TIME_MS)})