You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by mi...@apache.org on 2018/10/18 21:32:43 UTC
[4/4] impala git commit: IMPALA-7272: Fix crash in StringMinMaxFilter
IMPALA-7272: Fix crash in StringMinMaxFilter
StringMinMaxFilter uses a MemPool to allocate space for StringBuffers.
Previously, the MemPool was created by RuntimeFilterBank and passed to
each StringMinMaxFilter. In queries with multiple StringMinMaxFilters
being generated by the same fragment instance, this leads to
overlapping use of the MemPool by different threads, which is
incorrect as MemPools are not thread-safe.
The solution is to have each StringMinMaxFilter create its own
MemPool.
This patch also documents MemPool as not thread-safe and introduces a
DFAKE_MUTEX to help enforce correct usage. Doing this requires
modifying our CMakeLists.txt to pass '-DNDEBUG' to clang only in
RELEASE builds, so that the DFAKE_MUTEX will be present in the
compiled IR for DEBUG builds.
Testing:
- I have been unable to repro the actual crash despite trying a large
variety of different things. However, with additional logging added
its clear that the MemPool is being used concurrently, which is
incorrect.
- Added an e2e test that covers the potential issue. It hits the
DFAKE_MUTEX with a sleep added to MemPool::Allocate.
- Ran a full exhaustive build in both DEBUG and RELEASE.
Change-Id: I751cad7e6b75c9d95d7ad029bbd1e52fe09e8a29
Reviewed-on: http://gerrit.cloudera.org:8080/11650
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/9f5c5e6d
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/9f5c5e6d
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/9f5c5e6d
Branch: refs/heads/master
Commit: 9f5c5e6df03824cba292fe5a619153462c11669c
Parents: 48fb490
Author: Thomas Tauber-Marshall <tm...@cloudera.com>
Authored: Wed Oct 10 11:43:50 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Thu Oct 18 20:56:52 2018 +0000
----------------------------------------------------------------------
be/CMakeLists.txt | 9 ++--
be/src/runtime/mem-pool.cc | 5 ++
be/src/runtime/mem-pool.h | 11 +++++
be/src/runtime/runtime-filter-bank.cc | 11 +++--
be/src/runtime/runtime-filter-bank.h | 9 ++--
be/src/util/min-max-filter-test.cc | 49 ++++++++++++--------
be/src/util/min-max-filter.cc | 18 ++++---
be/src/util/min-max-filter.h | 26 +++++++----
.../queries/QueryTest/min_max_filters.test | 24 ++++++++++
9 files changed, 113 insertions(+), 49 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/9f5c5e6d/be/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/be/CMakeLists.txt b/be/CMakeLists.txt
index 2188294..99cbf81 100644
--- a/be/CMakeLists.txt
+++ b/be/CMakeLists.txt
@@ -222,15 +222,16 @@ add_definitions(-DKUDU_HEADERS_USE_SHORT_STATUS_MACROS -DKUDU_HEADERS_USE_RICH_S
# -Wno-return-type-c-linkage: UDFs return C++ classes but use C linkage to prevent
# mangling
# -DBOOST_NO_EXCEPTIONS: call a custom error handler for exceptions in codegen'd code.
-set(CLANG_IR_CXX_FLAGS "-emit-llvm" "-c" "-std=c++14" "-DIR_COMPILE" "-DNDEBUG"
- "-DHAVE_INTTYPES_H" "-DHAVE_NETINET_IN_H" "-DBOOST_DATE_TIME_POSIX_TIME_STD_CONFIG"
- "-DBOOST_NO_EXCEPTIONS" "-fcolor-diagnostics" "-Wno-deprecated"
- "-Wno-return-type-c-linkage" "-O1")
+set(CLANG_IR_CXX_FLAGS "-emit-llvm" "-c" "-std=c++14" "-DIR_COMPILE" "-DHAVE_INTTYPES_H"
+ "-DHAVE_NETINET_IN_H" "-DBOOST_DATE_TIME_POSIX_TIME_STD_CONFIG" "-DBOOST_NO_EXCEPTIONS"
+ "-fcolor-diagnostics" "-Wno-deprecated" "-Wno-return-type-c-linkage" "-O1")
# -Werror: compile warnings should be errors when using the toolchain compiler.
set(CLANG_IR_CXX_FLAGS "${CLANG_IR_CXX_FLAGS}" "-Werror")
if ("${CMAKE_BUILD_TYPE}" STREQUAL "ADDRESS_SANITIZER")
SET(CLANG_IR_CXX_FLAGS "${CLANG_IR_CXX_FLAGS}" "-DADDRESS_SANITIZER")
+elseif ("${CMAKE_BUILD_TYPE}" STREQUAL "RELEASE")
+ SET(CLANG_IR_CXX_FLAGS "${CLANG_IR_CXX_FLAGS}" "-DNDEBUG")
endif()
if (UBSAN_CODEGEN)
http://git-wip-us.apache.org/repos/asf/impala/blob/9f5c5e6d/be/src/runtime/mem-pool.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/mem-pool.cc b/be/src/runtime/mem-pool.cc
index 4d00d82..c0aa438 100644
--- a/be/src/runtime/mem-pool.cc
+++ b/be/src/runtime/mem-pool.cc
@@ -72,6 +72,7 @@ MemPool::~MemPool() {
}
void MemPool::Clear() {
+ DFAKE_SCOPED_LOCK(mutex_);
current_chunk_idx_ = -1;
for (auto& chunk: chunks_) {
chunk.allocated_bytes = 0;
@@ -82,6 +83,7 @@ void MemPool::Clear() {
}
void MemPool::FreeAll() {
+ DFAKE_SCOPED_LOCK(mutex_);
int64_t total_bytes_released = 0;
for (auto& chunk: chunks_) {
total_bytes_released += chunk.size;
@@ -162,6 +164,7 @@ bool MemPool::FindChunk(int64_t min_size, bool check_limits) noexcept {
}
void MemPool::AcquireData(MemPool* src, bool keep_current) {
+ DFAKE_SCOPED_LOCK(mutex_);
DCHECK(src->CheckIntegrity(false));
int num_acquired_chunks;
if (keep_current) {
@@ -213,6 +216,7 @@ void MemPool::AcquireData(MemPool* src, bool keep_current) {
}
void MemPool::SetMemTracker(MemTracker* new_tracker) {
+ DFAKE_SCOPED_LOCK(mutex_);
mem_tracker_->TransferTo(new_tracker, total_reserved_bytes_);
mem_tracker_ = new_tracker;
}
@@ -236,6 +240,7 @@ string MemPool::DebugString() {
}
int64_t MemPool::GetTotalChunkSizes() const {
+ DFAKE_SCOPED_LOCK(mutex_);
int64_t result = 0;
for (int i = 0; i < chunks_.size(); ++i) {
result += chunks_[i].size;
http://git-wip-us.apache.org/repos/asf/impala/blob/9f5c5e6d/be/src/runtime/mem-pool.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/mem-pool.h b/be/src/runtime/mem-pool.h
index 64a769e..1d2b633 100644
--- a/be/src/runtime/mem-pool.h
+++ b/be/src/runtime/mem-pool.h
@@ -28,6 +28,7 @@
#include "common/logging.h"
#include "gutil/dynamic_annotations.h"
+#include "gutil/threading/thread_collision_warner.h"
#include "util/bit-util.h"
namespace impala {
@@ -88,6 +89,8 @@ class MemTracker;
/// At this point p.total_allocated_bytes_ would be 0.
/// The one remaining (empty) chunk is released:
/// delete p;
+//
+/// This class is not thread-safe. A DFAKE_MUTEX is used to help enforce correct usage.
class MemPool {
public:
@@ -104,6 +107,7 @@ class MemPool {
/// of the the current chunk. Creates a new chunk if there aren't any chunks
/// with enough capacity.
uint8_t* Allocate(int64_t size) noexcept {
+ DFAKE_SCOPED_LOCK(mutex_);
return Allocate<false>(size, DEFAULT_ALIGNMENT);
}
@@ -112,12 +116,14 @@ class MemPool {
/// The caller must handle the NULL case. This should be used for allocations
/// where the size can be very big to bound the amount by which we exceed mem limits.
uint8_t* TryAllocate(int64_t size) noexcept {
+ DFAKE_SCOPED_LOCK(mutex_);
return Allocate<true>(size, DEFAULT_ALIGNMENT);
}
/// Same as TryAllocate() except a non-default alignment can be specified. It
/// should be a power-of-two in [1, alignof(std::max_align_t)].
uint8_t* TryAllocateAligned(int64_t size, int alignment) noexcept {
+ DFAKE_SCOPED_LOCK(mutex_);
DCHECK_GE(alignment, 1);
DCHECK_LE(alignment, alignof(std::max_align_t));
DCHECK_EQ(BitUtil::RoundUpToPowerOfTwo(alignment), alignment);
@@ -126,6 +132,7 @@ class MemPool {
/// Same as TryAllocate() except returned memory is not aligned at all.
uint8_t* TryAllocateUnaligned(int64_t size) noexcept {
+ DFAKE_SCOPED_LOCK(mutex_);
// Call templated implementation directly so that it is inlined here and the
// alignment logic can be optimised out.
return Allocate<true>(size, 1);
@@ -135,6 +142,7 @@ class MemPool {
/// only be used to return either all or part of the previous allocation returned
/// by Allocate().
void ReturnPartialAllocation(int64_t byte_size) {
+ DFAKE_SCOPED_LOCK(mutex_);
DCHECK_GE(byte_size, 0);
DCHECK(current_chunk_idx_ != -1);
ChunkInfo& info = chunks_[current_chunk_idx_];
@@ -208,6 +216,9 @@ class MemPool {
/// TryAllocateAligned().
static uint32_t zero_length_region_ alignas(std::max_align_t);
+ /// Ensures a MemPool is not used by two threads concurrently.
+ DFAKE_MUTEX(mutex_);
+
/// chunk from which we served the last Allocate() call;
/// always points to the last chunk that contains allocated data;
/// chunks 0..current_chunk_idx_ - 1 are guaranteed to contain data
http://git-wip-us.apache.org/repos/asf/impala/blob/9f5c5e6d/be/src/runtime/runtime-filter-bank.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/runtime-filter-bank.cc b/be/src/runtime/runtime-filter-bank.cc
index ffb0a22..85c9625 100644
--- a/be/src/runtime/runtime-filter-bank.cc
+++ b/be/src/runtime/runtime-filter-bank.cc
@@ -50,7 +50,6 @@ RuntimeFilterBank::RuntimeFilterBank(const TQueryCtx& query_ctx, RuntimeState* s
: state_(state),
filter_mem_tracker_(
new MemTracker(-1, "Runtime Filter Bank", state->instance_mem_tracker(), false)),
- mem_pool_(filter_mem_tracker_.get()),
closed_(false),
total_bloom_filter_mem_required_(total_filter_mem_required) {
bloom_memory_allocated_ =
@@ -206,7 +205,8 @@ void RuntimeFilterBank::PublishGlobalFilter(const TPublishFilterParams& params)
DCHECK(it->second->is_min_max_filter());
DCHECK(params.__isset.min_max_filter);
min_max_filter = MinMaxFilter::Create(
- params.min_max_filter, it->second->type(), &obj_pool_, &mem_pool_);
+ params.min_max_filter, it->second->type(), &obj_pool_, filter_mem_tracker_.get());
+ min_max_filters_.push_back(min_max_filter);
}
it->second->SetFilter(bloom_filter, min_max_filter);
state_->runtime_profile()->AddInfoString(
@@ -247,7 +247,10 @@ MinMaxFilter* RuntimeFilterBank::AllocateScratchMinMaxFilter(
RuntimeFilterMap::iterator it = produced_filters_.find(filter_id);
DCHECK(it != produced_filters_.end()) << "Filter ID " << filter_id << " not registered";
- return MinMaxFilter::Create(type, &obj_pool_, &mem_pool_);
+ MinMaxFilter* min_max_filter =
+ MinMaxFilter::Create(type, &obj_pool_, filter_mem_tracker_.get());
+ min_max_filters_.push_back(min_max_filter);
+ return min_max_filter;
}
bool RuntimeFilterBank::FpRateTooHigh(int64_t filter_size, int64_t observed_ndv) {
@@ -260,8 +263,8 @@ void RuntimeFilterBank::Close() {
lock_guard<mutex> l(runtime_filter_lock_);
closed_ = true;
for (BloomFilter* filter : bloom_filters_) filter->Close();
+ for (MinMaxFilter* filter : min_max_filters_) filter->Close();
obj_pool_.Clear();
- mem_pool_.FreeAll();
if (buffer_pool_client_.is_registered()) {
VLOG_FILE << "RuntimeFilterBank (Fragment Id: "
<< PrintId(state_->fragment_instance_id())
http://git-wip-us.apache.org/repos/asf/impala/blob/9f5c5e6d/be/src/runtime/runtime-filter-bank.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/runtime-filter-bank.h b/be/src/runtime/runtime-filter-bank.h
index ead7e82..907d6c4 100644
--- a/be/src/runtime/runtime-filter-bank.h
+++ b/be/src/runtime/runtime-filter-bank.h
@@ -152,9 +152,6 @@ class RuntimeFilterBank {
/// MemTracker to track Bloom filter memory.
boost::scoped_ptr<MemTracker> filter_mem_tracker_;
- // Mem pool to track allocations made by filters.
- MemPool mem_pool_;
-
/// True iff Close() has been called. Used to prevent races between
/// AllocateScratchBloomFilter() and Close().
bool closed_;
@@ -166,9 +163,13 @@ class RuntimeFilterBank {
long total_bloom_filter_mem_required_;
/// Contains references to all the bloom filters generated. Used in Close() to safely
- /// release all memory allocated for Bloomfilters.
+ /// release all memory allocated for BloomFilters.
vector<BloomFilter*> bloom_filters_;
+ /// Contains references to all the min-max filters generated. Used in Close() to safely
+ /// release all memory allocated for MinMaxFilters.
+ vector<MinMaxFilter*> min_max_filters_;
+
/// Buffer pool client for the filter bank. Initialized with the required reservation
/// in ClaimBufferReservation(). Reservations are returned to the initial reservations
/// pool in Close().
http://git-wip-us.apache.org/repos/asf/impala/blob/9f5c5e6d/be/src/util/min-max-filter-test.cc
----------------------------------------------------------------------
diff --git a/be/src/util/min-max-filter-test.cc b/be/src/util/min-max-filter-test.cc
index 23712e6..37d6f83 100644
--- a/be/src/util/min-max-filter-test.cc
+++ b/be/src/util/min-max-filter-test.cc
@@ -31,11 +31,10 @@ using namespace impala;
// inserted into it, and that MinMaxFilter::Or works for bools.
TEST(MinMaxFilterTest, TestBoolMinMaxFilter) {
MemTracker mem_tracker;
- MemPool mem_pool(&mem_tracker);
ObjectPool obj_pool;
- MinMaxFilter* filter =
- MinMaxFilter::Create(ColumnType(PrimitiveType::TYPE_BOOLEAN), &obj_pool, &mem_pool);
+ MinMaxFilter* filter = MinMaxFilter::Create(
+ ColumnType(PrimitiveType::TYPE_BOOLEAN), &obj_pool, &mem_tracker);
EXPECT_TRUE(filter->AlwaysFalse());
bool b1 = true;
filter->Insert(&b1);
@@ -58,6 +57,8 @@ TEST(MinMaxFilterTest, TestBoolMinMaxFilter) {
MinMaxFilter::Or(tFilter1, &tFilter2);
EXPECT_FALSE(tFilter2.min.bool_val);
EXPECT_TRUE(tFilter2.max.bool_val);
+
+ filter->Close();
}
void CheckIntVals(MinMaxFilter* filter, int32_t min, int32_t max) {
@@ -73,11 +74,10 @@ void CheckIntVals(MinMaxFilter* filter, int32_t min, int32_t max) {
// generated with maxcros and the logic is identical.
TEST(MinMaxFilterTest, TestNumericMinMaxFilter) {
MemTracker mem_tracker;
- MemPool mem_pool(&mem_tracker);
ObjectPool obj_pool;
ColumnType int_type(PrimitiveType::TYPE_INT);
- MinMaxFilter* int_filter = MinMaxFilter::Create(int_type, &obj_pool, &mem_pool);
+ MinMaxFilter* int_filter = MinMaxFilter::Create(int_type, &obj_pool, &mem_tracker);
// Test the behavior of an empty filter.
EXPECT_TRUE(int_filter->AlwaysFalse());
@@ -89,7 +89,7 @@ TEST(MinMaxFilterTest, TestNumericMinMaxFilter) {
EXPECT_FALSE(tFilter.min.__isset.int_val);
EXPECT_FALSE(tFilter.max.__isset.int_val);
MinMaxFilter* empty_filter =
- MinMaxFilter::Create(tFilter, int_type, &obj_pool, &mem_pool);
+ MinMaxFilter::Create(tFilter, int_type, &obj_pool, &mem_tracker);
EXPECT_TRUE(empty_filter->AlwaysFalse());
EXPECT_FALSE(empty_filter->AlwaysTrue());
@@ -113,7 +113,7 @@ TEST(MinMaxFilterTest, TestNumericMinMaxFilter) {
EXPECT_EQ(tFilter.min.int_val, i4);
EXPECT_EQ(tFilter.max.int_val, i2);
MinMaxFilter* int_filter2 =
- MinMaxFilter::Create(tFilter, int_type, &obj_pool, &mem_pool);
+ MinMaxFilter::Create(tFilter, int_type, &obj_pool, &mem_tracker);
CheckIntVals(int_filter2, i4, i2);
// Check the behavior of Or.
@@ -126,6 +126,10 @@ TEST(MinMaxFilterTest, TestNumericMinMaxFilter) {
MinMaxFilter::Or(tFilter1, &tFilter2);
EXPECT_EQ(tFilter2.min.int_val, 2);
EXPECT_EQ(tFilter2.max.int_val, 8);
+
+ int_filter->Close();
+ empty_filter->Close();
+ int_filter2->Close();
}
void CheckStringVals(MinMaxFilter* filter, const string& min, const string& max) {
@@ -146,10 +150,9 @@ void CheckStringVals(MinMaxFilter* filter, const string& min, const string& max)
TEST(MinMaxFilterTest, TestStringMinMaxFilter) {
ObjectPool obj_pool;
MemTracker mem_tracker;
- MemPool mem_pool(&mem_tracker);
ColumnType string_type(PrimitiveType::TYPE_STRING);
- MinMaxFilter* filter = MinMaxFilter::Create(string_type, &obj_pool, &mem_pool);
+ MinMaxFilter* filter = MinMaxFilter::Create(string_type, &obj_pool, &mem_tracker);
// Test the behavior of an empty filter.
EXPECT_TRUE(filter->AlwaysFalse());
@@ -163,7 +166,7 @@ TEST(MinMaxFilterTest, TestStringMinMaxFilter) {
EXPECT_FALSE(tFilter.always_true);
MinMaxFilter* empty_filter =
- MinMaxFilter::Create(tFilter, string_type, &obj_pool, &mem_pool);
+ MinMaxFilter::Create(tFilter, string_type, &obj_pool, &mem_tracker);
EXPECT_TRUE(empty_filter->AlwaysFalse());
EXPECT_FALSE(empty_filter->AlwaysTrue());
@@ -229,7 +232,7 @@ TEST(MinMaxFilterTest, TestStringMinMaxFilter) {
EXPECT_EQ(tFilter.max.string_val, truncTrailMaxChar);
MinMaxFilter* filter2 =
- MinMaxFilter::Create(tFilter, string_type, &obj_pool, &mem_pool);
+ MinMaxFilter::Create(tFilter, string_type, &obj_pool, &mem_tracker);
CheckStringVals(filter2, b1024, truncTrailMaxChar);
// Check that if the entire string is the max char and therefore after truncating for
@@ -249,15 +252,12 @@ TEST(MinMaxFilterTest, TestStringMinMaxFilter) {
EXPECT_TRUE(tFilter.always_true);
MinMaxFilter* always_true_filter =
- MinMaxFilter::Create(tFilter, string_type, &obj_pool, &mem_pool);
+ MinMaxFilter::Create(tFilter, string_type, &obj_pool, &mem_tracker);
EXPECT_FALSE(always_true_filter->AlwaysFalse());
EXPECT_TRUE(always_true_filter->AlwaysTrue());
- mem_pool.FreeAll();
-
// Check that a filter that hits the mem limit is disabled.
MemTracker limit_mem_tracker(1);
- MemPool limit_mem_pool(&limit_mem_tracker);
// We do not want to start the webserver.
FLAGS_enable_webserver = false;
std::unique_ptr<TestEnv> env;
@@ -265,7 +265,7 @@ TEST(MinMaxFilterTest, TestStringMinMaxFilter) {
ASSERT_OK(env->Init());
MinMaxFilter* limit_filter =
- MinMaxFilter::Create(string_type, &obj_pool, &limit_mem_pool);
+ MinMaxFilter::Create(string_type, &obj_pool, &limit_mem_tracker);
EXPECT_FALSE(limit_filter->AlwaysTrue());
limit_filter->Insert(&cVal);
limit_filter->MaterializeValues();
@@ -288,6 +288,12 @@ TEST(MinMaxFilterTest, TestStringMinMaxFilter) {
MinMaxFilter::Or(tFilter1, &tFilter2);
EXPECT_EQ(tFilter2.min.string_val, "a");
EXPECT_EQ(tFilter2.max.string_val, "e");
+
+ filter->Close();
+ empty_filter->Close();
+ filter2->Close();
+ limit_filter->Close();
+ always_true_filter->Close();
}
void CheckTimestampVals(
@@ -303,9 +309,8 @@ void CheckTimestampVals(
TEST(MinMaxFilterTest, TestTimestampMinMaxFilter) {
ObjectPool obj_pool;
MemTracker mem_tracker;
- MemPool mem_pool(&mem_tracker);
ColumnType timestamp_type(PrimitiveType::TYPE_TIMESTAMP);
- MinMaxFilter* filter = MinMaxFilter::Create(timestamp_type, &obj_pool, &mem_pool);
+ MinMaxFilter* filter = MinMaxFilter::Create(timestamp_type, &obj_pool, &mem_tracker);
// Test the behavior of an empty filter.
EXPECT_TRUE(filter->AlwaysFalse());
@@ -317,7 +322,7 @@ TEST(MinMaxFilterTest, TestTimestampMinMaxFilter) {
EXPECT_FALSE(tFilter.min.__isset.timestamp_val);
EXPECT_FALSE(tFilter.max.__isset.timestamp_val);
MinMaxFilter* empty_filter =
- MinMaxFilter::Create(tFilter, timestamp_type, &obj_pool, &mem_pool);
+ MinMaxFilter::Create(tFilter, timestamp_type, &obj_pool, &mem_tracker);
EXPECT_TRUE(empty_filter->AlwaysFalse());
EXPECT_FALSE(empty_filter->AlwaysTrue());
@@ -341,7 +346,7 @@ TEST(MinMaxFilterTest, TestTimestampMinMaxFilter) {
EXPECT_EQ(TimestampValue::FromTColumnValue(tFilter.min), t2);
EXPECT_EQ(TimestampValue::FromTColumnValue(tFilter.max), t3);
MinMaxFilter* filter2 =
- MinMaxFilter::Create(tFilter, timestamp_type, &obj_pool, &mem_pool);
+ MinMaxFilter::Create(tFilter, timestamp_type, &obj_pool, &mem_tracker);
CheckTimestampVals(filter2, t2, t3);
// Check the behavior of Or.
@@ -354,6 +359,10 @@ TEST(MinMaxFilterTest, TestTimestampMinMaxFilter) {
MinMaxFilter::Or(tFilter1, &tFilter2);
EXPECT_EQ(TimestampValue::FromTColumnValue(tFilter2.min), t2);
EXPECT_EQ(TimestampValue::FromTColumnValue(tFilter2.max), t3);
+
+ filter->Close();
+ empty_filter->Close();
+ filter2->Close();
}
int main(int argc, char** argv) {
http://git-wip-us.apache.org/repos/asf/impala/blob/9f5c5e6d/be/src/util/min-max-filter.cc
----------------------------------------------------------------------
diff --git a/be/src/util/min-max-filter.cc b/be/src/util/min-max-filter.cc
index f50f896..5a8bfc7 100644
--- a/be/src/util/min-max-filter.cc
+++ b/be/src/util/min-max-filter.cc
@@ -199,8 +199,11 @@ NUMERIC_MIN_MAX_FILTER_NO_CAST(Double);
const char* StringMinMaxFilter::LLVM_CLASS_NAME = "class.impala::StringMinMaxFilter";
const int StringMinMaxFilter::MAX_BOUND_LENGTH = 1024;
-StringMinMaxFilter::StringMinMaxFilter(const TMinMaxFilter& thrift, MemPool* mem_pool)
- : min_buffer_(mem_pool), max_buffer_(mem_pool) {
+StringMinMaxFilter::StringMinMaxFilter(
+ const TMinMaxFilter& thrift, MemTracker* mem_tracker)
+ : mem_pool_(mem_tracker),
+ min_buffer_(&mem_pool_),
+ max_buffer_(&mem_pool_) {
always_false_ = thrift.always_false;
always_true_ = thrift.always_true;
if (!always_true_ && !always_false_) {
@@ -388,7 +391,8 @@ bool MinMaxFilter::GetCastIntMinMax(
return true;
}
-MinMaxFilter* MinMaxFilter::Create(ColumnType type, ObjectPool* pool, MemPool* mem_pool) {
+MinMaxFilter* MinMaxFilter::Create(
+ ColumnType type, ObjectPool* pool, MemTracker* mem_tracker) {
switch (type.type) {
case PrimitiveType::TYPE_BOOLEAN:
return pool->Add(new BoolMinMaxFilter());
@@ -405,7 +409,7 @@ MinMaxFilter* MinMaxFilter::Create(ColumnType type, ObjectPool* pool, MemPool* m
case PrimitiveType::TYPE_DOUBLE:
return pool->Add(new DoubleMinMaxFilter());
case PrimitiveType::TYPE_STRING:
- return pool->Add(new StringMinMaxFilter(mem_pool));
+ return pool->Add(new StringMinMaxFilter(mem_tracker));
case PrimitiveType::TYPE_TIMESTAMP:
return pool->Add(new TimestampMinMaxFilter());
default:
@@ -414,8 +418,8 @@ MinMaxFilter* MinMaxFilter::Create(ColumnType type, ObjectPool* pool, MemPool* m
return nullptr;
}
-MinMaxFilter* MinMaxFilter::Create(
- const TMinMaxFilter& thrift, ColumnType type, ObjectPool* pool, MemPool* mem_pool) {
+MinMaxFilter* MinMaxFilter::Create(const TMinMaxFilter& thrift, ColumnType type,
+ ObjectPool* pool, MemTracker* mem_tracker) {
switch (type.type) {
case PrimitiveType::TYPE_BOOLEAN:
return pool->Add(new BoolMinMaxFilter(thrift));
@@ -432,7 +436,7 @@ MinMaxFilter* MinMaxFilter::Create(
case PrimitiveType::TYPE_DOUBLE:
return pool->Add(new DoubleMinMaxFilter(thrift));
case PrimitiveType::TYPE_STRING:
- return pool->Add(new StringMinMaxFilter(thrift, mem_pool));
+ return pool->Add(new StringMinMaxFilter(thrift, mem_tracker));
case PrimitiveType::TYPE_TIMESTAMP:
return pool->Add(new TimestampMinMaxFilter(thrift));
default:
http://git-wip-us.apache.org/repos/asf/impala/blob/9f5c5e6d/be/src/util/min-max-filter.h
----------------------------------------------------------------------
diff --git a/be/src/util/min-max-filter.h b/be/src/util/min-max-filter.h
index 556f5fa..7982500 100644
--- a/be/src/util/min-max-filter.h
+++ b/be/src/util/min-max-filter.h
@@ -27,7 +27,7 @@
namespace impala {
-class MemPool;
+class MemTacker;
class ObjectPool;
/// A MinMaxFilter tracks the min and max currently seen values in a data set for use in
@@ -42,6 +42,7 @@ class ObjectPool;
class MinMaxFilter {
public:
virtual ~MinMaxFilter() {}
+ virtual void Close() {}
/// Returns the min/max values in the tuple slot representation. It is not valid to call
/// these functions if AlwaysFalse() returns true.
@@ -77,13 +78,13 @@ class MinMaxFilter {
virtual std::string DebugString() const = 0;
- /// Returns a new MinMaxFilter with the given type, allocated from 'pool'.
- static MinMaxFilter* Create(ColumnType type, ObjectPool* pool, MemPool* mem_pool);
+ /// Returns a new MinMaxFilter with the given type, allocated from 'mem_tracker'.
+ static MinMaxFilter* Create(ColumnType type, ObjectPool* pool, MemTracker* mem_tracker);
/// Returns a new MinMaxFilter created from the thrift representation, allocated from
- /// 'pool'.
- static MinMaxFilter* Create(
- const TMinMaxFilter& thrift, ColumnType type, ObjectPool* pool, MemPool* mem_pool);
+ /// 'mem_tracker'.
+ static MinMaxFilter* Create(const TMinMaxFilter& thrift, ColumnType type,
+ ObjectPool* pool, MemTracker* mem_tracker);
/// Computes the logical OR of 'in' with 'out' and stores the result in 'out'.
static void Or(const TMinMaxFilter& in, TMinMaxFilter* out);
@@ -139,13 +140,15 @@ NUMERIC_MIN_MAX_FILTER(Double, double);
class StringMinMaxFilter : public MinMaxFilter {
public:
- StringMinMaxFilter(MemPool* mem_pool)
- : min_buffer_(mem_pool),
- max_buffer_(mem_pool),
+ StringMinMaxFilter(MemTracker* mem_tracker)
+ : mem_pool_(mem_tracker),
+ min_buffer_(&mem_pool_),
+ max_buffer_(&mem_pool_),
always_false_(true),
always_true_(false) {}
- StringMinMaxFilter(const TMinMaxFilter& thrift, MemPool* mem_pool);
+ StringMinMaxFilter(const TMinMaxFilter& thrift, MemTracker* mem_tracker);
virtual ~StringMinMaxFilter() {}
+ virtual void Close() override { mem_pool_.FreeAll(); }
virtual void* GetMin() override { return &min_; }
virtual void* GetMax() override { return &max_; }
@@ -182,6 +185,9 @@ class StringMinMaxFilter : public MinMaxFilter {
/// into this filter that are longer than this will be truncated.
static const int MAX_BOUND_LENGTH;
+ /// MemPool that 'min_buffer_'/'max_buffer_' are allocated from.
+ MemPool mem_pool_;
+
/// The min/max values. After a call to MaterializeValues() these will point to
/// 'min_buffer_'/'max_buffer_'.
StringValue min_;
http://git-wip-us.apache.org/repos/asf/impala/blob/9f5c5e6d/testdata/workloads/functional-query/queries/QueryTest/min_max_filters.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/min_max_filters.test b/testdata/workloads/functional-query/queries/QueryTest/min_max_filters.test
index c87cc74..5e2ec29 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/min_max_filters.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/min_max_filters.test
@@ -119,3 +119,27 @@ where a.tinyint_col = v.x
---- RESULTS
730
====
+
+
+---- QUERY
+####################################################
+# Test case 4: Two StringMinMaxFilters generated in the same fragment (IMPALA-7272).
+###################################################
+select straight_join a.l_orderkey
+from tpch_kudu.lineitem a
+ inner join /* +shuffle */ tpch_kudu.lineitem b on a.l_comment = b.l_comment
+ inner join /* +shuffle */ tpch_kudu.lineitem c on b.l_comment = c.l_comment
+where a.l_orderkey = 1
+---- TYPES
+BIGINT
+---- RESULTS
+1
+1
+1
+1
+1
+1
+1
+1
+1
+====
\ No newline at end of file