You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2017/11/08 22:38:42 UTC
[1/3] incubator-impala git commit: IMPALA-2281: Replace FNV with
FastHash in exchange nodes
Repository: incubator-impala
Updated Branches:
refs/heads/master 1ca3adf46 -> a772f8456
IMPALA-2281: Replace FNV with FastHash in exchange nodes
FNV is not a good enough hash function. This patch introduces FastHash
into the codebase and uses it in exchange nodes.
Testing: Two test cases involving arbitrary ordering are changed.
Single node performance benchmark shows no performance difference.
Change-Id: I778317d982dcdb94173a369a65b39f32b4f7ded2
Reviewed-on: http://gerrit.cloudera.org:8080/8417
Reviewed-by: Jim Apple <jb...@apache.org>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/73d1bc3a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/73d1bc3a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/73d1bc3a
Branch: refs/heads/master
Commit: 73d1bc3ad448181626c559c36f9d86b23e036fef
Parents: 1ca3adf
Author: Tianyi Wang <tw...@cloudera.com>
Authored: Thu Oct 26 15:26:59 2017 -0700
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Wed Nov 8 07:39:02 2017 +0000
----------------------------------------------------------------------
LICENSE.txt | 26 ++++++
be/src/benchmarks/hash-benchmark.cc | 98 +++++++++++++++-----
be/src/codegen/gen_ir_descriptions.py | 1 -
be/src/codegen/llvm-codegen.cc | 4 -
be/src/runtime/data-stream-sender.cc | 17 ++--
be/src/runtime/data-stream-sender.h | 3 +
be/src/runtime/data-stream-test.cc | 4 +-
be/src/runtime/raw-value-test.cc | 44 ++++-----
be/src/runtime/raw-value.cc | 42 ++++-----
be/src/runtime/raw-value.h | 10 +-
be/src/util/hash-util-ir.cc | 5 -
be/src/util/hash-util.h | 42 +++++++++
.../queries/QueryTest/nested-types-runtime.test | 12 +--
13 files changed, 207 insertions(+), 101 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/73d1bc3a/LICENSE.txt
----------------------------------------------------------------------
diff --git a/LICENSE.txt b/LICENSE.txt
index 01359d4..419a726 100644
--- a/LICENSE.txt
+++ b/LICENSE.txt
@@ -920,3 +920,29 @@ be/src/kudu/util/x509_check_host.*: OpenSSL software license:
[including the GNU Public Licence.]
--------------------------------------------------------------------------------
+
+be/src/util/hash-util.h (some portions): MIT license
+
+Some portions of this module are derived from code from FastHash
+(https://code.google.com/archive/p/fast-hash/).
+
+ Copyright (C) 2012 Zilong Tan (eric.zltan@gmail.com)
+
+ Permission is hereby granted, free of charge, to any person obtaining a copy of this
+ software and associated documentation files (the "Software"), to deal in the Software
+ without restriction, including without limitation the rights to use, copy, modify,
+ merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+ permit persons to whom the Software is furnished to do so, subject to the following
+ conditions:
+
+ The above copyright notice and this permission notice shall be included in all copies or
+ substantial portions of the Software.
+
+ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
+ INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
+ PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+ LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT
+ OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+ OTHER DEALINGS IN THE SOFTWARE.
+
+--------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/73d1bc3a/be/src/benchmarks/hash-benchmark.cc
----------------------------------------------------------------------
diff --git a/be/src/benchmarks/hash-benchmark.cc b/be/src/benchmarks/hash-benchmark.cc
index b183915..d0ecc82 100644
--- a/be/src/benchmarks/hash-benchmark.cc
+++ b/be/src/benchmarks/hash-benchmark.cc
@@ -16,8 +16,6 @@
// under the License.
#include <iostream>
-#include <stdlib.h>
-#include <stdio.h>
#include <vector>
#include <boost/functional/hash.hpp>
@@ -26,12 +24,9 @@
#include "common/init.h"
#include "experiments/data-provider.h"
#include "runtime/mem-tracker.h"
-#include "runtime/string-value.h"
#include "runtime/test-env.h"
#include "service/fe-support.h"
#include "util/benchmark.h"
-#include "util/cpu-info.h"
-#include "util/hash-util.h"
#include "common/names.h"
@@ -50,6 +45,7 @@ using namespace llvm;
// The different hash functions benchmarked:
// 1. FNV Hash: Fowler-Noll-Vo hash function
// 2. FNV Hash with empty string handling: FNV with special-case for empty string
+// 3. FastHash64: 64-bit FastHash function
// 3. Murmur2_64 Hash: Murmur2 hash function
// 4. Boost Hash: boost hash function
// 5. Crc: hash using sse4 crc hash instruction
@@ -62,24 +58,30 @@ using namespace llvm;
// = lim n->inf n(1 - 1/n) ^n
// = n / e
// = 367
-
-// Machine Info: Intel(R) Core(TM) i7-4790 CPU @ 3.60GHz
-// Int Hash: Function Rate (iters/ms) Comparison
-// ----------------------------------------------------------------------
-// Fnv 114.6 1X
-// Murmur2_64 129.9 1.134X
-// Boost 294.1 2.567X
-// Crc 536.2 4.68X
-// Codegen 1413 12.33X
-//
-// Mixed Hash: Function Rate (iters/ms) Comparison
-// ----------------------------------------------------------------------
-// Fnv 90.88 1X
-// FnvEmpty 91.58 1.008X
-// Murmur2_64 124.9 1.374X
-// Boost 133.5 1.469X
-// Crc 435.8 4.795X
-// Codegen 379.3 4.174X
+/*
+Machine Info: Intel(R) Core(TM) i7-6700 CPU @ 3.40GHz
+Int Hash:
+Function 10%ile 50%ile 90%ile 10%ile 50%ile 90%ile
+ (relative) (relative) (relative)
+----------- ----------------------------------------------------------
+ Fnv 88.5 109 111 1X 1X 1X
+ FastHash64 96.9 110 112 1.1X 1.01X 1.01X
+ Murmur2_64 90.7 124 126 1.03X 1.14X 1.14X
+ Boost 203 277 282 2.3X 2.55X 2.55X
+ Crc 415 536 540 4.68X 4.93X 4.89X
+ Codegen 1.5e+03 1.85e+03 1.88e+03 16.9X 17X 17X
+Mixed Hash:
+Function 10%ile 50%ile 90%ile 10%ile 50%ile 90%ile
+ (relative) (relative) (relative)
+----------------------------------------------------------------------
+ Fnv 75.3 78 78.9 1X 1X 1X
+ FastHash64 91.2 95.4 96.3 1.21X 1.22X 1.22X
+ FnvEmpty 82.7 86.3 86.9 1.1X 1.11X 1.1X
+ Murmur2_64 109 113 114 1.45X 1.45X 1.45X
+ Boost 109 112 113 1.45X 1.43X 1.44X
+ Crc 467 481 489 6.21X 6.17X 6.21X
+ Codegen 292 312 318 3.88X 4X 4.03X
+*/
typedef uint32_t (*CodegenHashFn)(int rows, char* data, int32_t* results);
@@ -125,6 +127,23 @@ void TestCrcIntHash(int batch, void* d) {
}
}
+void TestFastHashIntHash(int batch, void* d) {
+ TestData* data = reinterpret_cast<TestData*>(d);
+ int rows = data->num_rows;
+ int cols = data->num_cols;
+ for (int i = 0; i < batch; ++i) {
+ int32_t* values = reinterpret_cast<int32_t*>(data->data);
+ for (int j = 0; j < rows; ++j) {
+ uint64_t hash = HashUtil::FNV_SEED;
+ for (int k = 0; k < cols; ++k) {
+ hash = HashUtil::FastHash64(&values[k], sizeof(uint32_t), hash);
+ }
+ data->results[j] = hash;
+ values += cols;
+ }
+ }
+}
+
void TestBoostIntHash(int batch, void* d) {
TestData* data = reinterpret_cast<TestData*>(d);
int rows = data->num_rows;
@@ -235,6 +254,32 @@ void TestCrcMixedHash(int batch, void* d) {
}
}
+void TestFastHashMixedHash(int batch, void* d) {
+ TestData* data = reinterpret_cast<TestData*>(d);
+ int rows = data->num_rows;
+ for (int i = 0; i < batch; ++i) {
+ char* values = reinterpret_cast<char*>(data->data);
+ for (int j = 0; j < rows; ++j) {
+ uint64_t hash = HashUtil::FNV_SEED;
+
+ hash = HashUtil::FastHash64(values, sizeof(int8_t), hash);
+ values += sizeof(int8_t);
+
+ hash = HashUtil::FastHash64(values, sizeof(int32_t), hash);
+ values += sizeof(int32_t);
+
+ hash = HashUtil::FastHash64(values, sizeof(int64_t), hash);
+ values += sizeof(int64_t);
+
+ StringValue* str = reinterpret_cast<StringValue*>(values);
+ hash = HashUtil::FastHash64(str->ptr, str->len, hash);
+ values += sizeof(StringValue);
+
+ data->results[j] = hash;
+ }
+ }
+}
+
void TestCodegenMixedHash(int batch, void* d) {
TestData* data = reinterpret_cast<TestData*>(d);
CodegenHashFn fn = reinterpret_cast<CodegenHashFn>(data->jitted_fn);
@@ -416,9 +461,8 @@ Function* CodegenCrcHash(LlvmCodeGen* codegen, bool mixed) {
}
int main(int argc, char **argv) {
- CpuInfo::Init();
- cout << Benchmark::GetMachineInfo() << endl;
impala::InitCommonRuntime(argc, argv, true, impala::TestInfo::BE_TEST);
+ cout << Benchmark::GetMachineInfo() << endl;
impala::InitFeSupport();
ABORT_IF_ERROR(LlvmCodeGen::InitializeLlvm());
@@ -506,6 +550,7 @@ int main(int argc, char **argv) {
Benchmark int_suite("Int Hash");
int_suite.AddBenchmark("Fnv", TestFnvIntHash, &int_data);
+ int_suite.AddBenchmark("FastHash64", TestFastHashIntHash, &int_data);
int_suite.AddBenchmark("Murmur2_64", TestMurmur2_64IntHash, &int_data);
int_suite.AddBenchmark("Boost", TestBoostIntHash, &int_data);
int_suite.AddBenchmark("Crc", TestCrcIntHash, &int_data);
@@ -514,6 +559,7 @@ int main(int argc, char **argv) {
Benchmark mixed_suite("Mixed Hash");
mixed_suite.AddBenchmark("Fnv", TestFnvMixedHash, &mixed_data);
+ mixed_suite.AddBenchmark("FastHash64", TestFastHashMixedHash, &mixed_data);
mixed_suite.AddBenchmark("FnvEmpty", TestFnvEmptyMixedHash, &mixed_data);
mixed_suite.AddBenchmark("Murmur2_64", TestMurmur2_64MixedHash, &mixed_data);
mixed_suite.AddBenchmark("Boost", TestBoostMixedHash, &mixed_data);
@@ -521,5 +567,7 @@ int main(int argc, char **argv) {
mixed_suite.AddBenchmark("Codegen", TestCodegenMixedHash, &mixed_data);
cout << mixed_suite.Measure();
+ codegen->Close();
+ mem_pool.FreeAll();
return 0;
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/73d1bc3a/be/src/codegen/gen_ir_descriptions.py
----------------------------------------------------------------------
diff --git a/be/src/codegen/gen_ir_descriptions.py b/be/src/codegen/gen_ir_descriptions.py
index 1044a40..28e0a71 100755
--- a/be/src/codegen/gen_ir_descriptions.py
+++ b/be/src/codegen/gen_ir_descriptions.py
@@ -96,7 +96,6 @@ ir_functions = [
["SCALAR_EXPR_GET_DECIMAL_VAL",
"_ZN6impala10ScalarExpr13GetDecimalValEPS0_PNS_19ScalarExprEvaluatorEPKNS_8TupleRowE"],
["HASH_CRC", "IrCrcHash"],
- ["HASH_FNV", "IrFnvHash"],
["HASH_MURMUR", "IrMurmurHash"],
["PHJ_PROCESS_BUILD_BATCH",
"_ZN6impala10PhjBuilder17ProcessBuildBatchEPNS_8RowBatchEPNS_12HashTableCtxEbb"],
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/73d1bc3a/be/src/codegen/llvm-codegen.cc
----------------------------------------------------------------------
diff --git a/be/src/codegen/llvm-codegen.cc b/be/src/codegen/llvm-codegen.cc
index 777ca0a..77a45c6 100644
--- a/be/src/codegen/llvm-codegen.cc
+++ b/be/src/codegen/llvm-codegen.cc
@@ -1584,10 +1584,6 @@ static Function* GetLenOptimizedHashFn(
return codegen->FinalizeFunction(fn);
}
-Function* LlvmCodeGen::GetFnvHashFunction(int len) {
- return GetLenOptimizedHashFn(this, IRFunction::HASH_FNV, len);
-}
-
Function* LlvmCodeGen::GetMurmurHashFunction(int len) {
return GetLenOptimizedHashFn(this, IRFunction::HASH_MURMUR, len);
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/73d1bc3a/be/src/runtime/data-stream-sender.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/data-stream-sender.cc b/be/src/runtime/data-stream-sender.cc
index ed156dc..cf749cd 100644
--- a/be/src/runtime/data-stream-sender.cc
+++ b/be/src/runtime/data-stream-sender.cc
@@ -464,17 +464,16 @@ Status DataStreamSender::Send(RuntimeState* state, RowBatch* batch) {
int num_channels = channels_.size();
for (int i = 0; i < batch->num_rows(); ++i) {
TupleRow* row = batch->GetRow(i);
- uint32_t hash_val = HashUtil::FNV_SEED;
- for (int i = 0; i < partition_exprs_.size(); ++i) {
- ScalarExprEvaluator* eval = partition_expr_evals_[i];
+ uint64_t hash_val = EXCHANGE_HASH_SEED;
+ for (int j = 0; j < partition_exprs_.size(); ++j) {
+ ScalarExprEvaluator* eval = partition_expr_evals_[j];
void* partition_val = eval->GetValue(row);
- // We can't use the crc hash function here because it does not result
- // in uncorrelated hashes with different seeds. Instead we must use
- // fnv hash.
+ // We can't use the crc hash function here because it does not result in
+ // uncorrelated hashes with different seeds. Instead we use FastHash.
// TODO: fix crc hash/GetHashValue()
- DCHECK(&partition_expr_evals_[i]->root() == partition_exprs_[i]);
- hash_val = RawValue::GetHashValueFnv(
- partition_val, partition_exprs_[i]->type(), hash_val);
+ DCHECK(&(eval->root()) == partition_exprs_[j]);
+ hash_val = RawValue::GetHashValueFastHash(
+ partition_val, partition_exprs_[j]->type(), hash_val);
}
RETURN_IF_ERROR(channels_[hash_val % num_channels]->AddRow(row));
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/73d1bc3a/be/src/runtime/data-stream-sender.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/data-stream-sender.h b/be/src/runtime/data-stream-sender.h
index 95a83e5..8ed4bd0 100644
--- a/be/src/runtime/data-stream-sender.h
+++ b/be/src/runtime/data-stream-sender.h
@@ -153,6 +153,9 @@ class DataStreamSender : public DataSink {
/// Used for Kudu partitioning to round-robin rows that don't correspond to a partition
/// or when errors are encountered.
int next_unknown_partition_;
+
+ /// An arbitrary hash seed used for exchanges.
+ static constexpr uint64_t EXCHANGE_HASH_SEED = 0x66bd68df22c3ef37;
};
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/73d1bc3a/be/src/runtime/data-stream-test.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/data-stream-test.cc b/be/src/runtime/data-stream-test.cc
index 79a97b6..3f4844c 100644
--- a/be/src/runtime/data-stream-test.cc
+++ b/be/src/runtime/data-stream-test.cc
@@ -425,8 +425,8 @@ class DataStreamTest : public testing::Test {
} else if (stream_type == TPartitionType::HASH_PARTITIONED) {
// hash-partitioned streams send values to the right partition
int64_t value = *j;
- uint32_t hash_val =
- RawValue::GetHashValueFnv(&value, TYPE_BIGINT, HashUtil::FNV_SEED);
+ uint64_t hash_val = RawValue::GetHashValueFastHash(&value, TYPE_BIGINT,
+ DataStreamSender::EXCHANGE_HASH_SEED);
EXPECT_EQ(hash_val % receiver_info_.size(), info.receiver_num);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/73d1bc3a/be/src/runtime/raw-value-test.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/raw-value-test.cc b/be/src/runtime/raw-value-test.cc
index 9c71822..bced866 100644
--- a/be/src/runtime/raw-value-test.cc
+++ b/be/src/runtime/raw-value-test.cc
@@ -79,35 +79,37 @@ TEST_F(RawValueTest, TypeChar) {
// IMPALA-2270: "", false, and NULL should hash to distinct values.
TEST_F(RawValueTest, HashEmptyAndNull) {
uint32_t seed = 12345;
- uint32_t null_hash = RawValue::GetHashValue(NULL, TYPE_STRING, seed);
- uint32_t null_hash_fnv = RawValue::GetHashValueFnv(NULL, TYPE_STRING, seed);
- StringValue empty(NULL, 0);
- uint32_t empty_hash = RawValue::GetHashValue(&empty, TYPE_STRING, seed);
- uint32_t empty_hash_fnv = RawValue::GetHashValueFnv(&empty, TYPE_STRING, seed);
+ uint32_t hash_null = RawValue::GetHashValue(nullptr, TYPE_STRING, seed);
+ uint64_t fast_hash_null = RawValue::GetHashValueFastHash(nullptr, TYPE_STRING, seed);
+ StringValue empty(nullptr, 0);
+ uint32_t hash_empty = RawValue::GetHashValue(&empty, TYPE_STRING, seed);
+ uint64_t fast_hash_empty =
+ RawValue::GetHashValueFastHash(&empty, TYPE_STRING, seed);
bool false_val = false;
- uint32_t false_hash = RawValue::GetHashValue(&false_val, TYPE_BOOLEAN, seed);
- uint32_t false_hash_fnv = RawValue::GetHashValue(&false_val, TYPE_BOOLEAN, seed);
- EXPECT_NE(seed, null_hash);
- EXPECT_NE(seed, empty_hash);
- EXPECT_NE(seed, false_hash);
- EXPECT_NE(seed, null_hash_fnv);
- EXPECT_NE(seed, empty_hash_fnv);
- EXPECT_NE(seed, false_hash_fnv);
- EXPECT_NE(null_hash, empty_hash);
- EXPECT_NE(null_hash_fnv, empty_hash_fnv);
- EXPECT_NE(null_hash, false_hash);
- EXPECT_NE(false_hash, null_hash_fnv);
+ uint32_t hash_false = RawValue::GetHashValue(&false_val, TYPE_BOOLEAN, seed);
+ uint64_t fast_hash_false =
+ RawValue::GetHashValueFastHash(&false_val, TYPE_BOOLEAN, seed);
+ EXPECT_NE(seed, hash_null);
+ EXPECT_NE(seed, hash_empty);
+ EXPECT_NE(seed, hash_false);
+ EXPECT_NE(seed, fast_hash_null);
+ EXPECT_NE(seed, fast_hash_empty);
+ EXPECT_NE(seed, fast_hash_false);
+ EXPECT_NE(hash_null, hash_empty);
+ EXPECT_NE(fast_hash_null, fast_hash_empty);
+ EXPECT_NE(hash_null, hash_false);
+ EXPECT_NE(hash_false, fast_hash_null);
}
-/// IMPALA-2270: Test that FNV hash of (int, "") is not skewed.
+/// IMPALA-2270: Test that FastHash of (int, "") is not skewed.
TEST(HashUtil, IntNullSkew) {
int num_values = 100000;
int num_buckets = 16;
vector<int> buckets(num_buckets, 0);
for (int32_t i = 0; i < num_values; ++i) {
- uint32_t hash = RawValue::GetHashValueFnv(&i, TYPE_INT, 9999);
- StringValue empty(NULL, 0);
- hash = RawValue::GetHashValueFnv(&empty, TYPE_STRING, hash);
+ uint64_t hash = RawValue::GetHashValueFastHash(&i, TYPE_INT, 9999);
+ StringValue empty(nullptr, 0);
+ hash = RawValue::GetHashValueFastHash(&empty, TYPE_STRING, hash);
++buckets[hash % num_buckets];
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/73d1bc3a/be/src/runtime/raw-value.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/raw-value.cc b/be/src/runtime/raw-value.cc
index d7f4fd5..1db9741 100644
--- a/be/src/runtime/raw-value.cc
+++ b/be/src/runtime/raw-value.cc
@@ -60,7 +60,7 @@ void RawValue::PrintValueAsBytes(const void* value, const ColumnType& type,
case TYPE_STRING:
case TYPE_VARCHAR:
string_val = reinterpret_cast<const StringValue*>(value);
- stream->write(static_cast<char*>(string_val->ptr), string_val->len);
+ stream->write(string_val->ptr, string_val->len);
break;
case TYPE_TIMESTAMP:
stream->write(chars, TimestampValue::Size());
@@ -98,7 +98,7 @@ void RawValue::PrintValue(const void* value, const ColumnType& type, int scale,
case TYPE_STRING:
case TYPE_VARCHAR:
string_val = reinterpret_cast<const StringValue*>(value);
- tmp.assign(static_cast<char*>(string_val->ptr), string_val->len);
+ tmp.assign(string_val->ptr, string_val->len);
str->swap(tmp);
return;
case TYPE_CHAR:
@@ -193,31 +193,29 @@ void RawValue::Write(const void* value, Tuple* tuple, const SlotDescriptor* slot
}
}
-uint32_t RawValue::GetHashValueFnv(const void* v, const ColumnType& type, uint32_t seed) {
- // Use HashCombine with arbitrary constant to ensure we don't return seed.
- if (v == NULL) return HashUtil::HashCombine32(HASH_VAL_NULL, seed);
-
+uint64_t RawValue::GetHashValueFastHash(const void* v, const ColumnType& type,
+ uint64_t seed) {
+ // Hash with an arbitrary constant to ensure we don't return seed.
+ if (v == nullptr) {
+ return HashUtil::FastHash64(&HASH_VAL_NULL, sizeof(HASH_VAL_NULL), seed);
+ }
switch (type.type) {
case TYPE_STRING:
case TYPE_VARCHAR: {
const StringValue* string_value = reinterpret_cast<const StringValue*>(v);
- if (string_value->len == 0) {
- return HashUtil::HashCombine32(HASH_VAL_EMPTY, seed);
- }
- return HashUtil::FnvHash64to32(string_value->ptr, string_value->len, seed);
+ return HashUtil::FastHash64(string_value->ptr,
+ static_cast<size_t>(string_value->len), seed);
}
- case TYPE_BOOLEAN:
- return HashUtil::HashCombine32(*reinterpret_cast<const bool*>(v), seed);
- case TYPE_TINYINT: return HashUtil::FnvHash64to32(v, 1, seed);
- case TYPE_SMALLINT: return HashUtil::FnvHash64to32(v, 2, seed);
- case TYPE_INT: return HashUtil::FnvHash64to32(v, 4, seed);
- case TYPE_BIGINT: return HashUtil::FnvHash64to32(v, 8, seed);
- case TYPE_FLOAT: return HashUtil::FnvHash64to32(v, 4, seed);
- case TYPE_DOUBLE: return HashUtil::FnvHash64to32(v, 8, seed);
- case TYPE_TIMESTAMP: return HashUtil::FnvHash64to32(v, 12, seed);
- case TYPE_CHAR:
- return HashUtil::FnvHash64to32(v, type.len, seed);
- case TYPE_DECIMAL: return HashUtil::FnvHash64to32(v, type.GetByteSize(), seed);
+ case TYPE_BOOLEAN: return HashUtil::FastHash64(v, 1, seed);
+ case TYPE_TINYINT: return HashUtil::FastHash64(v, 1, seed);
+ case TYPE_SMALLINT: return HashUtil::FastHash64(v, 2, seed);
+ case TYPE_INT: return HashUtil::FastHash64(v, 4, seed);
+ case TYPE_BIGINT: return HashUtil::FastHash64(v, 8, seed);
+ case TYPE_FLOAT: return HashUtil::FastHash64(v, 4, seed);
+ case TYPE_DOUBLE: return HashUtil::FastHash64(v, 8, seed);
+ case TYPE_TIMESTAMP: return HashUtil::FastHash64(v, 12, seed);
+ case TYPE_CHAR: return HashUtil::FastHash64(v, type.len, seed);
+ case TYPE_DECIMAL: return HashUtil::FastHash64(v, type.GetByteSize(), seed);
default: DCHECK(false); return 0;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/73d1bc3a/be/src/runtime/raw-value.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/raw-value.h b/be/src/runtime/raw-value.h
index 89ae323..6ee9191 100644
--- a/be/src/runtime/raw-value.h
+++ b/be/src/runtime/raw-value.h
@@ -71,12 +71,10 @@ class RawValue {
static inline uint32_t GetHashValueNonNull(const T* v, const ColumnType& type,
uint32_t seed = 0);
- /// Get a 32-bit hash value using the FNV hash function.
- /// Using different seeds with FNV results in different hash functions.
- /// GetHashValue() does not have this property and cannot be safely used as the first
- /// step in data repartitioning. However, GetHashValue() can be significantly faster.
- /// TODO: fix GetHashValue
- static uint32_t GetHashValueFnv(const void* v, const ColumnType& type, uint32_t seed);
+ /// Get a 64-bit hash value using the FastHash function.
+ /// https://code.google.com/archive/p/fast-hash/
+ static uint64_t GetHashValueFastHash(const void* v, const ColumnType& type,
+ uint64_t seed);
/// Compares both values.
/// Return value is < 0 if v1 < v2, 0 if v1 == v2, > 0 if v1 > v2.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/73d1bc3a/be/src/util/hash-util-ir.cc
----------------------------------------------------------------------
diff --git a/be/src/util/hash-util-ir.cc b/be/src/util/hash-util-ir.cc
index cf812eb..da3271b 100644
--- a/be/src/util/hash-util-ir.cc
+++ b/be/src/util/hash-util-ir.cc
@@ -23,11 +23,6 @@
using namespace impala;
extern "C"
-uint32_t IrFnvHash(const void* data, int32_t bytes, uint32_t hash) {
- return HashUtil::FnvHash64to32(data, bytes, hash);
-}
-
-extern "C"
uint32_t IrMurmurHash(const void* data, int32_t bytes, uint32_t hash) {
return HashUtil::MurmurHash2_64(data, bytes, hash);
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/73d1bc3a/be/src/util/hash-util.h
----------------------------------------------------------------------
diff --git a/be/src/util/hash-util.h b/be/src/util/hash-util.h
index d0c791a..054cb8a 100644
--- a/be/src/util/hash-util.h
+++ b/be/src/util/hash-util.h
@@ -231,6 +231,48 @@ class HashUtil {
// Rehash32to32(hash2) is minimal.
return (static_cast<uint64_t>(hash) * m + a) >> 32;
}
+
+ // The FastHash64 implementation is adapted from
+ // https://code.google.com/archive/p/fast-hash/
+ // available under MIT license.
+ static inline uint64_t FastHashMix(uint64_t h) {
+ h ^= h >> 23;
+ h *= 0x2127599bf4325c37ULL;
+ h ^= h >> 47;
+ return h;
+ }
+
+ static inline uint64_t FastHash64(const void* buf, int64_t len, uint64_t seed)
+ {
+ const uint64_t m = 0x880355f21e6d1965ULL;
+ const uint64_t* pos = reinterpret_cast<const uint64_t*>(buf);
+ const uint64_t* end = pos + (len / 8);
+ uint64_t h = seed ^ (len * m);
+ uint64_t v;
+
+ while (pos != end) {
+ v = *pos++;
+ h ^= FastHashMix(v);
+ h *= m;
+ }
+
+ const uint8_t* pos2 = reinterpret_cast<const uint8_t*>(pos);
+ v = 0;
+
+ switch (len & 7) {
+ case 7: v ^= static_cast<uint64_t>(pos2[6]) << 48;
+ case 6: v ^= static_cast<uint64_t>(pos2[5]) << 40;
+ case 5: v ^= static_cast<uint64_t>(pos2[4]) << 32;
+ case 4: v ^= static_cast<uint64_t>(pos2[3]) << 24;
+ case 3: v ^= static_cast<uint64_t>(pos2[2]) << 16;
+ case 2: v ^= static_cast<uint64_t>(pos2[1]) << 8;
+ case 1: v ^= static_cast<uint64_t>(pos2[0]);
+ h ^= FastHashMix(v);
+ h *= m;
+ }
+
+ return FastHashMix(h);
+ }
};
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/73d1bc3a/testdata/workloads/functional-query/queries/QueryTest/nested-types-runtime.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/nested-types-runtime.test b/testdata/workloads/functional-query/queries/QueryTest/nested-types-runtime.test
index 35e27c5..c72558a 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/nested-types-runtime.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/nested-types-runtime.test
@@ -139,16 +139,16 @@ bigint,string,bigint
---- QUERY
# Test several analytic functions with incompatible partition by and order by clauses
# on top of a subplan that flattens a map.
-select id, key, value, max(key) over(partition by id), row_number() over (order by value)
-from complextypestbl t, t.int_map
+select id, key, value, max(key) over(partition by id), row_number()
+over (order by value, key) from complextypestbl t, t.int_map
---- RESULTS: VERIFY_IS_EQUAL_SORTED
+8,'k1',-1,'k1',1
1,'k1',1,'k2',2
-1,'k2',100,'k2',4
2,'k1',2,'k2',3
-2,'k2',NULL,'k2',5
-7,'k1',NULL,'k3',6
+1,'k2',100,'k2',4
+7,'k1',NULL,'k3',5
+2,'k2',NULL,'k2',6
7,'k3',NULL,'k3',7
-8,'k1',-1,'k1',1
---- TYPES
bigint,string,int,string,bigint
====
[2/3] incubator-impala git commit: IMPALA-5307: Part 4: copy out
uncompressed text and seq
Posted by ta...@apache.org.
IMPALA-5307: Part 4: copy out uncompressed text and seq
This is the final patch for IMPALA-5307.
Text and Seq scanners are converted to use the same approach
as Avro.
contains_tuple_data is now false so a bunch of dead code in
ScannerContext can be removed. We also no longer attach I/O
buffers to row batches so that logic can be removed.
Testing:
Ran core ASAN tests.
Perf:
I reran the same benchmarks as in Part 2. There was no measurable
difference before and after - for both text and seq processing time
is dominated by text parsing.
Change-Id: I304fd002b61bfedf41c8b1405cd7eb7b492bb941
Reviewed-on: http://gerrit.cloudera.org:8080/8172
Reviewed-by: Tim Armstrong <ta...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/19c17e64
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/19c17e64
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/19c17e64
Branch: refs/heads/master
Commit: 19c17e64b54fcc0a599eadaa14d35943ee703ed4
Parents: 73d1bc3
Author: Tim Armstrong <ta...@cloudera.com>
Authored: Tue Sep 26 15:27:30 2017 -0700
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Wed Nov 8 10:09:25 2017 +0000
----------------------------------------------------------------------
be/src/codegen/gen_ir_descriptions.py | 2 +-
be/src/exec/base-sequence-scanner.cc | 6 +--
be/src/exec/exec-node.cc | 9 +----
be/src/exec/exec-node.h | 8 +---
be/src/exec/hdfs-avro-scanner-ir.cc | 4 +-
be/src/exec/hdfs-avro-scanner.cc | 5 ---
be/src/exec/hdfs-avro-scanner.h | 6 ---
be/src/exec/hdfs-parquet-scanner.cc | 9 ++---
be/src/exec/hdfs-rcfile-scanner.cc | 4 --
be/src/exec/hdfs-scan-node-base.h | 6 ---
be/src/exec/hdfs-scan-node-mt.cc | 2 -
be/src/exec/hdfs-scan-node.cc | 9 +----
be/src/exec/hdfs-scanner-ir.cc | 16 ++++++--
be/src/exec/hdfs-scanner.cc | 20 +++++++++-
be/src/exec/hdfs-scanner.h | 27 +++++++++-----
be/src/exec/hdfs-sequence-scanner.cc | 22 ++++++++---
be/src/exec/hdfs-text-scanner.cc | 28 ++++++--------
be/src/exec/parquet-column-readers.cc | 2 +-
be/src/exec/scanner-context.cc | 32 +++-------------
be/src/exec/scanner-context.h | 47 +++++++++--------------
be/src/exec/text-converter.cc | 4 +-
be/src/runtime/row-batch.cc | 28 +++-----------
be/src/runtime/row-batch.h | 60 +++++++++++++-----------------
23 files changed, 144 insertions(+), 212 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/codegen/gen_ir_descriptions.py
----------------------------------------------------------------------
diff --git a/be/src/codegen/gen_ir_descriptions.py b/be/src/codegen/gen_ir_descriptions.py
index 28e0a71..b3ad25d 100755
--- a/be/src/codegen/gen_ir_descriptions.py
+++ b/be/src/codegen/gen_ir_descriptions.py
@@ -174,7 +174,7 @@ ir_functions = [
["HDFS_SCANNER_INIT_TUPLE",
"_ZN6impala11HdfsScanner9InitTupleEPNS_5TupleES2_"],
["HDFS_SCANNER_WRITE_ALIGNED_TUPLES",
- "_ZN6impala11HdfsScanner18WriteAlignedTuplesEPNS_7MemPoolEPNS_8TupleRowEiPNS_13FieldLocationEiiii"],
+ "_ZN6impala11HdfsScanner18WriteAlignedTuplesEPNS_7MemPoolEPNS_8TupleRowEPNS_13FieldLocationEiiiib"],
["PROCESS_SCRATCH_BATCH",
"_ZN6impala18HdfsParquetScanner19ProcessScratchBatchEPNS_8RowBatchE"],
["PARQUET_SCANNER_EVAL_RUNTIME_FILTER",
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/base-sequence-scanner.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/base-sequence-scanner.cc b/be/src/exec/base-sequence-scanner.cc
index c22f18d..fcf58c6 100644
--- a/be/src/exec/base-sequence-scanner.cc
+++ b/be/src/exec/base-sequence-scanner.cc
@@ -98,9 +98,6 @@ Status BaseSequenceScanner::Open(ScannerContext* context) {
only_parsing_header_ = true;
return Status::OK();
}
-
- // If the file is compressed, the buffers in the stream_ are not used directly.
- if (header_->is_compressed) stream_->set_contains_tuple_data(false);
RETURN_IF_ERROR(InitNewRange());
// Skip to the first record
@@ -128,7 +125,6 @@ void BaseSequenceScanner::Close(RowBatch* row_batch) {
if (row_batch != nullptr) {
row_batch->tuple_data_pool()->AcquireData(data_buffer_pool_.get(), false);
row_batch->tuple_data_pool()->AcquireData(template_tuple_pool_.get(), false);
- context_->ReleaseCompletedResources(row_batch, true);
if (scan_node_->HasRowBatchQueue()) {
static_cast<HdfsScanNode*>(scan_node_)->AddMaterializedRowBatch(
unique_ptr<RowBatch>(row_batch));
@@ -136,8 +132,8 @@ void BaseSequenceScanner::Close(RowBatch* row_batch) {
} else {
data_buffer_pool_->FreeAll();
template_tuple_pool_->FreeAll();
- context_->ReleaseCompletedResources(nullptr, true);
}
+ context_->ReleaseCompletedResources(true);
// Verify all resources (if any) have been transferred.
DCHECK_EQ(template_tuple_pool_.get()->total_allocated_bytes(), 0);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/exec-node.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/exec-node.cc b/be/src/exec/exec-node.cc
index aaca8be..afb6249 100644
--- a/be/src/exec/exec-node.cc
+++ b/be/src/exec/exec-node.cc
@@ -103,21 +103,14 @@ unique_ptr<RowBatch> ExecNode::RowBatchQueue::GetBatch() {
return unique_ptr<RowBatch>();
}
-int ExecNode::RowBatchQueue::Cleanup() {
- int num_io_buffers = 0;
-
+void ExecNode::RowBatchQueue::Cleanup() {
unique_ptr<RowBatch> batch = NULL;
while ((batch = GetBatch()) != NULL) {
- num_io_buffers += batch->num_io_buffers();
batch.reset();
}
lock_guard<SpinLock> l(lock_);
- for (const unique_ptr<RowBatch>& row_batch: cleanup_queue_) {
- num_io_buffers += row_batch->num_io_buffers();
- }
cleanup_queue_.clear();
- return num_io_buffers;
}
ExecNode::ExecNode(ObjectPool* pool, const TPlanNode& tnode, const DescriptorTbl& descs)
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/exec-node.h
----------------------------------------------------------------------
diff --git a/be/src/exec/exec-node.h b/be/src/exec/exec-node.h
index a0ed352..187d9a7 100644
--- a/be/src/exec/exec-node.h
+++ b/be/src/exec/exec-node.h
@@ -81,7 +81,6 @@ class ExecNode {
virtual void Codegen(RuntimeState* state);
/// Performs any preparatory work prior to calling GetNext().
- /// Caller must not be holding any io buffers. This will cause deadlock.
/// If overridden in subclass, must first call superclass's Open().
/// Open() is called after Prepare() or Reset(), i.e., possibly multiple times
/// throughout the lifetime of this node.
@@ -91,7 +90,7 @@ class ExecNode {
/// child before acquiring their own resources to reduce the peak resource requirement.
/// This is particularly important if there are multiple blocking ExecNodes in a
/// pipeline because the lower nodes will release resources in Close() before the
- /// Open() of their parent retuns. The resource profile calculation in the frontend
+ /// Open() of their parent returns. The resource profile calculation in the frontend
/// relies on this when computing the peak resources required for a query.
virtual Status Open(RuntimeState* state) WARN_UNUSED_RESULT;
@@ -106,8 +105,6 @@ class ExecNode {
/// In other words, if the memory holding the tuple data will be referenced
/// by the callee in subsequent GetNext() calls, it must *not* be attached to the
/// row_batch's tuple_data_pool.
- /// Caller must not be holding any io buffers. This will cause deadlock.
- /// TODO: AggregationNode and HashJoinNode cannot be "re-opened" yet.
virtual Status GetNext(
RuntimeState* state, RowBatch* row_batch, bool* eos) WARN_UNUSED_RESULT = 0;
@@ -274,8 +271,7 @@ class ExecNode {
/// Deletes all row batches in cleanup_queue_. Not valid to call AddBatch()
/// after this is called.
- /// Returns the number of io buffers that were released (for debug tracking)
- int Cleanup();
+ void Cleanup();
private:
/// Lock protecting cleanup_queue_
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/hdfs-avro-scanner-ir.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-avro-scanner-ir.cc b/be/src/exec/hdfs-avro-scanner-ir.cc
index a2bf606..ffd5774 100644
--- a/be/src/exec/hdfs-avro-scanner-ir.cc
+++ b/be/src/exec/hdfs-avro-scanner-ir.cc
@@ -37,7 +37,7 @@ int HdfsAvroScanner::DecodeAvroData(int max_tuples, MemPool* pool, uint8_t** dat
// If the file is uncompressed, StringValues will have pointers into the I/O buffers.
// We don't attach I/O buffers to output batches so need to copy out data referenced
// by tuples that survive conjunct evaluation.
- const bool copy_out_strings = !header_->is_compressed && !string_slot_offsets_.empty();
+ const bool copy_strings = !header_->is_compressed && !string_slot_offsets_.empty();
int num_to_commit = 0;
for (int i = 0; i < max_tuples; ++i) {
InitTuple(template_tuple_, tuple);
@@ -47,7 +47,7 @@ int HdfsAvroScanner::DecodeAvroData(int max_tuples, MemPool* pool, uint8_t** dat
}
tuple_row->SetTuple(scan_node_->tuple_idx(), tuple);
if (EvalConjuncts(tuple_row)) {
- if (copy_out_strings) {
+ if (copy_strings) {
if (UNLIKELY(!tuple->CopyStrings("HdfsAvroScanner::DecodeAvroData()",
state_, string_slot_offsets_.data(), string_slot_offsets_.size(), pool,
&parse_status_))) {
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/hdfs-avro-scanner.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-avro-scanner.cc b/be/src/exec/hdfs-avro-scanner.cc
index cad74e6..dea51a9 100644
--- a/be/src/exec/hdfs-avro-scanner.cc
+++ b/be/src/exec/hdfs-avro-scanner.cc
@@ -66,10 +66,6 @@ static Status CheckSchema(const AvroSchemaElement& avro_schema) {
HdfsAvroScanner::HdfsAvroScanner(HdfsScanNodeBase* scan_node, RuntimeState* state)
: BaseSequenceScanner(scan_node, state) {
- for (SlotDescriptor* string_slot : scan_node_->tuple_desc()->string_slots()) {
- string_slot_offsets_.push_back(
- {string_slot->null_indicator_offset(), string_slot->tuple_offset()});
- }
}
HdfsAvroScanner::HdfsAvroScanner()
@@ -80,7 +76,6 @@ HdfsAvroScanner::HdfsAvroScanner()
Status HdfsAvroScanner::Open(ScannerContext* context) {
RETURN_IF_ERROR(BaseSequenceScanner::Open(context));
RETURN_IF_ERROR(CheckSchema(scan_node_->avro_schema()));
- stream_->set_contains_tuple_data(false); // Avro scanner always copies out data.
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/hdfs-avro-scanner.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-avro-scanner.h b/be/src/exec/hdfs-avro-scanner.h
index ab5746a..e2260a6 100644
--- a/be/src/exec/hdfs-avro-scanner.h
+++ b/be/src/exec/hdfs-avro-scanner.h
@@ -130,12 +130,6 @@ class HdfsAvroScanner : public BaseSequenceScanner {
bool use_codegend_decode_avro_data;
};
- /// Offsets of string slots in the result tuple that may need to be copied as part of
- /// tuple materialization. Populated in constructor. This is redundant with offset
- /// information stored in the TupleDescriptor but storing only the required metadata
- /// in a simple array of struct simplifies codegen and speeds up interpretation.
- std::vector<SlotOffsets> string_slot_offsets_;
-
AvroFileHeader* avro_header_ = nullptr;
/// Current data block after decompression with its end and length.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/hdfs-parquet-scanner.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-parquet-scanner.cc b/be/src/exec/hdfs-parquet-scanner.cc
index e35f64a..2897287 100644
--- a/be/src/exec/hdfs-parquet-scanner.cc
+++ b/be/src/exec/hdfs-parquet-scanner.cc
@@ -164,7 +164,6 @@ HdfsParquetScanner::HdfsParquetScanner(HdfsScanNodeBase* scan_node, RuntimeState
Status HdfsParquetScanner::Open(ScannerContext* context) {
RETURN_IF_ERROR(HdfsScanner::Open(context));
- stream_->set_contains_tuple_data(false);
metadata_range_ = stream_->scan_range();
num_cols_counter_ =
ADD_COUNTER(scan_node_->runtime_profile(), "NumColumns", TUnit::UNIT);
@@ -228,7 +227,7 @@ Status HdfsParquetScanner::Open(ScannerContext* context) {
// Release I/O buffers immediately to make sure they are cleaned up
// in case we return a non-OK status anywhere below.
- context_->ReleaseCompletedResources(nullptr, true);
+ context_->ReleaseCompletedResources(true);
RETURN_IF_ERROR(footer_status);
// Parse the file schema into an internal representation for schema resolution.
@@ -265,7 +264,7 @@ void HdfsParquetScanner::Close(RowBatch* row_batch) {
} else {
template_tuple_pool_->FreeAll();
dictionary_pool_.get()->FreeAll();
- context_->ReleaseCompletedResources(nullptr, true);
+ context_->ReleaseCompletedResources(true);
for (ParquetColumnReader* col_reader : column_readers_) col_reader->Close(nullptr);
// The scratch batch may still contain tuple data. We can get into this case if
// Open() fails or if the query is cancelled.
@@ -731,7 +730,7 @@ void HdfsParquetScanner::FlushRowGroupResources(RowBatch* row_batch) {
DCHECK(row_batch != nullptr);
row_batch->tuple_data_pool()->AcquireData(dictionary_pool_.get(), false);
scratch_batch_->ReleaseResources(row_batch->tuple_data_pool());
- context_->ReleaseCompletedResources(nullptr, true);
+ context_->ReleaseCompletedResources(true);
for (ParquetColumnReader* col_reader : column_readers_) {
col_reader->Close(row_batch);
}
@@ -1676,8 +1675,6 @@ Status HdfsParquetScanner::InitColumns(
DCHECK(stream != NULL);
RETURN_IF_ERROR(scalar_reader->Reset(&col_chunk.meta_data, stream));
- // Parquet column readers never return tuple data with pointers into I/O buffers.
- stream->set_contains_tuple_data(false);
}
DCHECK_EQ(col_ranges.size(), num_scalar_readers);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/hdfs-rcfile-scanner.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-rcfile-scanner.cc b/be/src/exec/hdfs-rcfile-scanner.cc
index 2ea8229..a706c3d 100644
--- a/be/src/exec/hdfs-rcfile-scanner.cc
+++ b/be/src/exec/hdfs-rcfile-scanner.cc
@@ -77,10 +77,6 @@ Status HdfsRCFileScanner::InitNewRange() {
// ptrs into the decompressed data).
reuse_row_group_buffer_ = scan_node_->tuple_desc()->string_slots().empty();
- // The scanner currently copies all the column data out of the io buffer so the
- // stream never contains any tuple data.
- stream_->set_contains_tuple_data(false);
-
if (header_->is_compressed) {
RETURN_IF_ERROR(Codec::CreateDecompressor(nullptr,
reuse_row_group_buffer_, header_->codec, &decompressor_));
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/hdfs-scan-node-base.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-scan-node-base.h b/be/src/exec/hdfs-scan-node-base.h
index f6e010a..6a0abde 100644
--- a/be/src/exec/hdfs-scan-node-base.h
+++ b/be/src/exec/hdfs-scan-node-base.h
@@ -421,12 +421,6 @@ class HdfsScanNodeBase : public ScanNode {
AtomicInt32 num_scanners_codegen_enabled_;
AtomicInt32 num_scanners_codegen_disabled_;
- /// This is the number of io buffers that are owned by the scan node and the scanners.
- /// This is used just to help debug leaked io buffers to determine if the leak is
- /// happening in the scanners vs other parts of the execution.
- /// TODO: Remove this counter when deprecating the multi-threaded scan node.
- AtomicInt32 num_owned_io_buffers_;
-
/// If true, counters are actively running and need to be reported in the runtime
/// profile.
bool counters_running_ = false;
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/hdfs-scan-node-mt.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-scan-node-mt.cc b/be/src/exec/hdfs-scan-node-mt.cc
index 8d4efec..6803b69 100644
--- a/be/src/exec/hdfs-scan-node-mt.cc
+++ b/be/src/exec/hdfs-scan-node-mt.cc
@@ -103,7 +103,6 @@ Status HdfsScanNodeMt::GetNext(RuntimeState* state, RowBatch* row_batch, bool* e
if (!status.ok()) {
scanner_->Close(row_batch);
scanner_.reset();
- num_owned_io_buffers_.Add(-row_batch->num_io_buffers());
return status;
}
InitNullCollectionValues(row_batch);
@@ -119,7 +118,6 @@ Status HdfsScanNodeMt::GetNext(RuntimeState* state, RowBatch* row_batch, bool* e
*eos = true;
}
COUNTER_SET(rows_returned_counter_, num_rows_returned_);
- num_owned_io_buffers_.Add(-row_batch->num_io_buffers());
if (*eos) StopAndFinalizeCounters();
return Status::OK();
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/hdfs-scan-node.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-scan-node.cc b/be/src/exec/hdfs-scan-node.cc
index 8cc85ff..8be191f 100644
--- a/be/src/exec/hdfs-scan-node.cc
+++ b/be/src/exec/hdfs-scan-node.cc
@@ -115,7 +115,6 @@ Status HdfsScanNode::GetNextInternal(
*eos = false;
unique_ptr<RowBatch> materialized_batch = materialized_row_batches_->GetBatch();
if (materialized_batch != NULL) {
- num_owned_io_buffers_.Add(-materialized_batch->num_io_buffers());
row_batch->AcquireState(materialized_batch.get());
// Update the number of materialized rows now instead of when they are materialized.
// This means that scanners might process and queue up more rows than are necessary
@@ -133,7 +132,6 @@ Status HdfsScanNode::GetNextInternal(
*eos = true;
SetDone();
}
- DCHECK_EQ(materialized_batch->num_io_buffers(), 0);
materialized_batch.reset();
return Status::OK();
}
@@ -229,16 +227,11 @@ Status HdfsScanNode::Open(RuntimeState* state) {
void HdfsScanNode::Close(RuntimeState* state) {
if (is_closed()) return;
SetDone();
-
if (thread_avail_cb_id_ != -1) {
state->resource_pool()->RemoveThreadAvailableCb(thread_avail_cb_id_);
}
-
scanner_threads_.JoinAll();
-
- num_owned_io_buffers_.Add(-materialized_row_batches_->Cleanup());
- DCHECK_EQ(num_owned_io_buffers_.Load(), 0) << "ScanNode has leaked io buffers";
-
+ materialized_row_batches_->Cleanup();
HdfsScanNodeBase::Close(state);
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/hdfs-scanner-ir.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-scanner-ir.cc b/be/src/exec/hdfs-scanner-ir.cc
index 867ce1e..d7c2d81 100644
--- a/be/src/exec/hdfs-scanner-ir.cc
+++ b/be/src/exec/hdfs-scanner-ir.cc
@@ -19,6 +19,7 @@
#include "runtime/row-batch.h"
#include "util/string-parser.h"
#include "runtime/string-value.inline.h"
+#include "runtime/tuple.h"
#include "common/names.h"
@@ -33,9 +34,9 @@ using namespace impala;
// This function takes more arguments than are strictly necessary (they could be
// computed inside this function) but this is done to minimize the clang dependencies,
// specifically, calling function on the scan node.
-int HdfsScanner::WriteAlignedTuples(MemPool* pool, TupleRow* tuple_row, int row_size,
+int HdfsScanner::WriteAlignedTuples(MemPool* pool, TupleRow* tuple_row,
FieldLocation* fields, int num_tuples, int max_added_tuples,
- int slots_per_tuple, int row_idx_start) {
+ int slots_per_tuple, int row_idx_start, bool copy_strings) {
DCHECK(tuple_ != NULL);
uint8_t* tuple_row_mem = reinterpret_cast<uint8_t*>(tuple_row);
uint8_t* tuple_mem = reinterpret_cast<uint8_t*>(tuple_);
@@ -53,9 +54,16 @@ int HdfsScanner::WriteAlignedTuples(MemPool* pool, TupleRow* tuple_row, int row_
// function.
if (WriteCompleteTuple(pool, fields, tuple, tuple_row, template_tuple_,
error, &error_in_row)) {
+ if (copy_strings) {
+ if (UNLIKELY(!tuple->CopyStrings("HdfsScanner::WriteAlignedTuples()",
+ state_, string_slot_offsets_.data(), string_slot_offsets_.size(), pool,
+ &parse_status_))) {
+ return -1;
+ }
+ }
++tuples_returned;
- tuple_mem += tuple_byte_size_;
- tuple_row_mem += row_size;
+ tuple_mem += tuple_byte_size();
+ tuple_row_mem += sizeof(Tuple*);
tuple = reinterpret_cast<Tuple*>(tuple_mem);
tuple_row = reinterpret_cast<TupleRow*>(tuple_row_mem);
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/hdfs-scanner.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-scanner.cc b/be/src/exec/hdfs-scanner.cc
index 4107e42..959aa91 100644
--- a/be/src/exec/hdfs-scanner.cc
+++ b/be/src/exec/hdfs-scanner.cc
@@ -48,6 +48,12 @@ HdfsScanner::HdfsScanner(HdfsScanNodeBase* scan_node, RuntimeState* state)
template_tuple_pool_(new MemPool(scan_node->mem_tracker())),
tuple_byte_size_(scan_node->tuple_desc()->byte_size()),
data_buffer_pool_(new MemPool(scan_node->mem_tracker())) {
+ DCHECK_EQ(1, scan_node->row_desc()->tuple_descriptors().size())
+ << "All HDFS scanners assume one tuple per row";
+ for (SlotDescriptor* string_slot : scan_node_->tuple_desc()->string_slots()) {
+ string_slot_offsets_.push_back(
+ {string_slot->null_indicator_offset(), string_slot->tuple_offset()});
+ }
}
HdfsScanner::HdfsScanner()
@@ -188,7 +194,7 @@ Status HdfsScanner::CommitRows(int num_rows, RowBatch* row_batch) {
// which can happen if the query is very selective. We need to release memory even
// if no rows passed predicates.
if (row_batch->AtCapacity() || context_->num_completed_io_buffers() > 0) {
- context_->ReleaseCompletedResources(row_batch, /* done */ false);
+ context_->ReleaseCompletedResources(/* done */ false);
}
if (context_->cancelled()) return Status::CANCELLED;
// Check for UDF errors.
@@ -524,6 +530,18 @@ Status HdfsScanner::CodegenWriteAlignedTuples(const HdfsScanNodeBase* node,
"WriteCompleteTuple");
DCHECK_EQ(replaced, 1);
+ Function* copy_strings_fn;
+ RETURN_IF_ERROR(Tuple::CodegenCopyStrings(
+ codegen, *node->tuple_desc(), ©_strings_fn));
+ replaced = codegen->ReplaceCallSites(
+ write_tuples_fn, copy_strings_fn, "CopyStrings");
+ DCHECK_EQ(replaced, 1);
+
+ int tuple_byte_size = node->tuple_desc()->byte_size();
+ replaced = codegen->ReplaceCallSitesWithValue(write_tuples_fn,
+ codegen->GetIntConstant(TYPE_INT, tuple_byte_size), "tuple_byte_size");
+ DCHECK_EQ(replaced, 1);
+
*write_aligned_tuples_fn = codegen->FinalizeFunction(write_tuples_fn);
if (*write_aligned_tuples_fn == NULL) {
return Status("Failed to finalize write_aligned_tuples_fn.");
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/hdfs-scanner.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-scanner.h b/be/src/exec/hdfs-scanner.h
index cb69bbe..c603593 100644
--- a/be/src/exec/hdfs-scanner.h
+++ b/be/src/exec/hdfs-scanner.h
@@ -96,7 +96,7 @@ struct FieldLocation {
//
/// This class also encapsulates row batch management. Subclasses should call CommitRows()
/// after writing to the current row batch, which handles creating row batches, attaching
-/// resources (IO buffers and mem pools) to the current row batch, and passing row batches
+/// resources (buffers and mem pools) to the current row batch, and passing row batches
/// up to the scan node. Subclasses can also use GetMemory() to help with per-row memory
/// management.
/// TODO: Have a pass over all members and move them out of the base class if sensible
@@ -137,7 +137,7 @@ class HdfsScanner {
/// queue. Only valid to call if HasRowBatchQueue().
void Close();
- /// Transfers the ownership of memory backing returned tuples such as IO buffers
+ /// Transfers the ownership of memory backing returned tuples such as buffers
/// and memory in mem pools to the given row batch. If the row batch is NULL,
/// those resources are released instead. In any case, releases all other resources
/// that are not backing returned rows (e.g. temporary decompression buffers).
@@ -271,13 +271,19 @@ class HdfsScanner {
/// decompressor and any other per data block allocations.
boost::scoped_ptr<MemPool> data_buffer_pool_;
+ /// Offsets of string slots in the result tuple that may need to be copied as part of
+ /// tuple materialization. Populated in constructor. This is redundant with offset
+ /// information stored in the TupleDescriptor but storing only the required metadata
+ /// in a simple array of struct simplifies codegen and speeds up interpretation.
+ std::vector<SlotOffsets> string_slot_offsets_;
+
/// Time spent decompressing bytes.
RuntimeProfile::Counter* decompress_timer_ = nullptr;
/// Matching typedef for WriteAlignedTuples for codegen. Refer to comments for
/// that function.
- typedef int (*WriteTuplesFn)(HdfsScanner*, MemPool*, TupleRow*, int, FieldLocation*,
- int, int, int, int);
+ typedef int (*WriteTuplesFn)(HdfsScanner*, MemPool*, TupleRow*, FieldLocation*,
+ int, int, int, int, bool);
/// Jitted write tuples function pointer. Null if codegen is disabled.
WriteTuplesFn write_tuples_fn_ = nullptr;
@@ -332,15 +338,18 @@ class HdfsScanner {
/// - 'fields' must start at the beginning of a tuple.
/// - 'num_tuples' number of tuples to process
/// - 'max_added_tuples' the maximum number of tuples that should be added to the batch.
- /// - 'row_start_index' is the number of rows that have already been processed
+ /// - 'row_idx_start' is the number of rows that have already been processed
/// as part of WritePartialTuple.
+ /// - 'copy_strings': if true, strings in returned tuples that pass conjuncts are
+ /// copied into 'pool'
/// Returns the number of tuples added to the row batch. This can be less than
/// num_tuples/tuples_till_limit because of failed conjuncts.
- /// Returns -1 if parsing should be aborted due to parse errors.
+ /// Returns -1 if an error is encountered, e.g. a parse error or a memory allocation
+ /// error.
/// Only valid to call if the parent scan node is multi-threaded.
- int WriteAlignedTuples(MemPool* pool, TupleRow* tuple_row_mem, int row_size,
- FieldLocation* fields, int num_tuples,
- int max_added_tuples, int slots_per_tuple, int row_start_indx);
+ int WriteAlignedTuples(MemPool* pool, TupleRow* tuple_row_mem, FieldLocation* fields,
+ int num_tuples, int max_added_tuples, int slots_per_tuple, int row_idx_start,
+ bool copy_strings);
/// Update the decompressor_ object given a compression type or codec name. Depending on
/// the old compression type and the new one, it may close the old decompressor and/or
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/hdfs-sequence-scanner.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-sequence-scanner.cc b/be/src/exec/hdfs-sequence-scanner.cc
index 9b66432..1c248bb 100644
--- a/be/src/exec/hdfs-sequence-scanner.cc
+++ b/be/src/exec/hdfs-sequence-scanner.cc
@@ -266,6 +266,9 @@ Status HdfsSequenceScanner::ProcessDecompressedBlock(RowBatch* row_batch) {
// Materialize parsed cols to tuples
SCOPED_TIMER(scan_node_->materialize_tuple_timer());
+
+ // Need to copy out strings if they may reference the original I/O buffer.
+ const bool copy_strings = !header_->is_compressed && !string_slot_offsets_.empty();
// Call jitted function if possible
int tuples_returned;
if (write_tuples_fn_ != nullptr) {
@@ -275,12 +278,12 @@ Status HdfsSequenceScanner::ProcessDecompressedBlock(RowBatch* row_batch) {
delimited_text_parser_->escape_char() == '\0');
// last argument: seq always starts at record_location[0]
tuples_returned = write_tuples_fn_(this, row_batch->tuple_data_pool(), tuple_row,
- row_batch->row_byte_size(), field_locations_.data(), num_to_process,
- max_added_tuples, scan_node_->materialized_slots().size(), 0);
+ field_locations_.data(), num_to_process,
+ max_added_tuples, scan_node_->materialized_slots().size(), 0, copy_strings);
} else {
tuples_returned = WriteAlignedTuples(row_batch->tuple_data_pool(), tuple_row,
- row_batch->row_byte_size(), field_locations_.data(), num_to_process,
- max_added_tuples, scan_node_->materialized_slots().size(), 0);
+ field_locations_.data(), num_to_process,
+ max_added_tuples, scan_node_->materialized_slots().size(), 0, copy_strings);
}
if (tuples_returned == -1) return parse_status_;
@@ -301,6 +304,7 @@ Status HdfsSequenceScanner::ProcessRange(RowBatch* row_batch) {
SCOPED_TIMER(scan_node_->materialize_tuple_timer());
int64_t num_rows_read = 0;
+ const bool copy_strings = !seq_header->is_compressed && !string_slot_offsets_.empty();
const bool has_materialized_slots = !scan_node_->materialized_slots().empty();
while (!eos_) {
DCHECK_GT(record_locations_.size(), 0);
@@ -323,12 +327,20 @@ Status HdfsSequenceScanner::ProcessRange(RowBatch* row_batch) {
uint8_t error_in_row = false;
uint8_t errors[num_fields];
memset(errors, 0, num_fields);
- add_row = WriteCompleteTuple(row_batch->tuple_data_pool(), field_locations_.data(),
+ MemPool* pool = row_batch->tuple_data_pool();
+ add_row = WriteCompleteTuple(pool, field_locations_.data(),
tuple_, tuple_row_mem, template_tuple_, &errors[0], &error_in_row);
if (UNLIKELY(error_in_row)) {
ReportTupleParseError(field_locations_.data(), errors);
RETURN_IF_ERROR(parse_status_);
}
+ if (add_row && copy_strings) {
+ if (UNLIKELY(!tuple_->CopyStrings("HdfsSequenceScanner::ProcessRange()",
+ state_, string_slot_offsets_.data(), string_slot_offsets_.size(), pool,
+ &parse_status_))) {
+ return parse_status_;
+ }
+ }
} else {
add_row = WriteTemplateTuples(tuple_row_mem, 1);
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/hdfs-text-scanner.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-text-scanner.cc b/be/src/exec/hdfs-text-scanner.cc
index f1c1ff5..1cbc6f5 100644
--- a/be/src/exec/hdfs-text-scanner.cc
+++ b/be/src/exec/hdfs-text-scanner.cc
@@ -167,7 +167,6 @@ void HdfsTextScanner::Close(RowBatch* row_batch) {
if (row_batch != nullptr) {
row_batch->tuple_data_pool()->AcquireData(template_tuple_pool_.get(), false);
row_batch->tuple_data_pool()->AcquireData(data_buffer_pool_.get(), false);
- context_->ReleaseCompletedResources(row_batch, true);
if (scan_node_->HasRowBatchQueue()) {
static_cast<HdfsScanNode*>(scan_node_)->AddMaterializedRowBatch(
unique_ptr<RowBatch>(row_batch));
@@ -175,8 +174,8 @@ void HdfsTextScanner::Close(RowBatch* row_batch) {
} else {
template_tuple_pool_->FreeAll();
data_buffer_pool_->FreeAll();
- context_->ReleaseCompletedResources(nullptr, true);
}
+ context_->ReleaseCompletedResources(true);
// Verify all resources (if any) have been transferred or freed.
DCHECK_EQ(template_tuple_pool_.get()->total_allocated_bytes(), 0);
@@ -192,12 +191,6 @@ void HdfsTextScanner::Close(RowBatch* row_batch) {
Status HdfsTextScanner::InitNewRange() {
DCHECK_EQ(scan_state_, CONSTRUCTED);
- // Compressed text does not reference data in the io buffers directly. In such case, we
- // can recycle the buffers in the stream_ more promptly.
- if (stream_->file_desc()->file_compression != THdfsCompression::NONE) {
- stream_->set_contains_tuple_data(false);
- }
-
// Update the decompressor based on the compression type of the file in the context.
DCHECK(stream_->file_desc()->file_compression != THdfsCompression::SNAPPY)
<< "FE should have generated SNAPPY_BLOCKED instead.";
@@ -591,7 +584,7 @@ Status HdfsTextScanner::FillByteBufferCompressedStream(MemPool* pool, bool* eosr
if (*eosr) {
DCHECK(stream_->eosr());
- context_->ReleaseCompletedResources(nullptr, true);
+ context_->ReleaseCompletedResources(true);
}
return Status::OK();
@@ -637,7 +630,7 @@ Status HdfsTextScanner::FillByteBufferCompressedFile(bool* eosr) {
&decompressed_buffer));
// Inform 'stream_' that the buffer with the compressed text can be released.
- context_->ReleaseCompletedResources(nullptr, true);
+ context_->ReleaseCompletedResources(true);
VLOG_FILE << "Decompressed " << byte_buffer_read_size_ << " to " << decompressed_len;
byte_buffer_ptr_ = reinterpret_cast<char*>(decompressed_buffer);
@@ -837,6 +830,9 @@ int HdfsTextScanner::WriteFields(int num_fields, int num_tuples, MemPool* pool,
// Write complete tuples. The current field, if any, is at the start of a tuple.
if (num_tuples > 0) {
+ // Need to copy out strings if they may reference the original I/O buffer.
+ const bool copy_strings = !string_slot_offsets_.empty() &&
+ stream_->file_desc()->file_compression == THdfsCompression::NONE;
int max_added_tuples = (scan_node_->limit() == -1) ?
num_tuples : scan_node_->limit() - scan_node_->rows_returned();
int tuples_returned = 0;
@@ -846,13 +842,13 @@ int HdfsTextScanner::WriteFields(int num_fields, int num_tuples, MemPool* pool,
// slots and escape characters. TextConverter::WriteSlot() will be used instead.
DCHECK(scan_node_->tuple_desc()->string_slots().empty() ||
delimited_text_parser_->escape_char() == '\0');
- tuples_returned = write_tuples_fn_(this, pool, row, sizeof(Tuple*), fields,
- num_tuples, max_added_tuples, scan_node_->materialized_slots().size(),
- num_tuples_processed);
+ tuples_returned = write_tuples_fn_(this, pool, row, fields, num_tuples,
+ max_added_tuples, scan_node_->materialized_slots().size(),
+ num_tuples_processed, copy_strings);
} else {
- tuples_returned = WriteAlignedTuples(pool, row, sizeof(Tuple*), fields,
- num_tuples, max_added_tuples, scan_node_->materialized_slots().size(),
- num_tuples_processed);
+ tuples_returned = WriteAlignedTuples(pool, row, fields, num_tuples,
+ max_added_tuples, scan_node_->materialized_slots().size(),
+ num_tuples_processed, copy_strings);
}
if (tuples_returned == -1) return 0;
DCHECK_EQ(slot_idx_, 0);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/parquet-column-readers.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/parquet-column-readers.cc b/be/src/exec/parquet-column-readers.cc
index 15c963d..04127e3 100644
--- a/be/src/exec/parquet-column-readers.cc
+++ b/be/src/exec/parquet-column-readers.cc
@@ -996,7 +996,7 @@ Status BaseScalarColumnReader::ReadDataPage() {
}
// We don't hold any pointers to earlier pages in the stream - we can safely free
// any accumulated I/O or boundary buffers.
- stream_->ReleaseCompletedResources(nullptr, false);
+ stream_->ReleaseCompletedResources(false);
// Read the next data page, skipping page types we don't care about.
// We break out of this loop on the non-error case (a data page was found or we read all
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/scanner-context.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/scanner-context.cc b/be/src/exec/scanner-context.cc
index 3ed8b4a..8cb195d 100644
--- a/be/src/exec/scanner-context.cc
+++ b/be/src/exec/scanner-context.cc
@@ -58,9 +58,9 @@ ScannerContext::~ScannerContext() {
DCHECK(streams_.empty());
}
-void ScannerContext::ReleaseCompletedResources(RowBatch* batch, bool done) {
+void ScannerContext::ReleaseCompletedResources(bool done) {
for (int i = 0; i < streams_.size(); ++i) {
- streams_[i]->ReleaseCompletedResources(batch, done);
+ streams_[i]->ReleaseCompletedResources(done);
}
}
@@ -87,13 +87,11 @@ ScannerContext::Stream* ScannerContext::AddStream(DiskIoMgr::ScanRange* range) {
stream->output_buffer_pos_ = NULL;
stream->output_buffer_bytes_left_ =
const_cast<int64_t*>(&OUTPUT_BUFFER_BYTES_LEFT_INIT);
- stream->contains_tuple_data_ = scan_node_->tuple_desc()->ContainsStringData();
streams_.push_back(std::move(stream));
return streams_.back().get();
}
-void ScannerContext::Stream::ReleaseCompletedResources(RowBatch* batch, bool done) {
- DCHECK(batch != nullptr || done || !contains_tuple_data_);
+void ScannerContext::Stream::ReleaseCompletedResources(bool done) {
if (done) {
// Mark any pending resources as completed
if (io_buffer_ != nullptr) {
@@ -108,24 +106,11 @@ void ScannerContext::Stream::ReleaseCompletedResources(RowBatch* batch, bool don
}
for (unique_ptr<DiskIoMgr::BufferDescriptor>& buffer : completed_io_buffers_) {
- if (contains_tuple_data_ && batch != nullptr) {
- batch->AddIoBuffer(move(buffer));
- // TODO: We can do row batch compaction here. This is the only place io buffers are
- // queued. A good heuristic is to check the number of io buffers queued and if
- // there are too many, we should compact.
- } else {
- ExecEnv::GetInstance()->disk_io_mgr()->ReturnBuffer(move(buffer));
- parent_->scan_node_->num_owned_io_buffers_.Add(-1);
- }
+ ExecEnv::GetInstance()->disk_io_mgr()->ReturnBuffer(move(buffer));
}
parent_->num_completed_io_buffers_ -= completed_io_buffers_.size();
completed_io_buffers_.clear();
- if (contains_tuple_data_ && batch != nullptr) {
- // If we're not done, keep using the last chunk allocated in boundary_pool_ so we
- // don't have to reallocate. If we are done, transfer it to the row batch.
- batch->tuple_data_pool()->AcquireData(boundary_pool_.get(), /* keep_current */ !done);
- }
if (done) boundary_pool_->FreeAll();
}
@@ -196,7 +181,6 @@ Status ScannerContext::Stream::GetNextBuffer(int64_t read_past_size) {
filename(), offset));
}
- parent_->scan_node_->num_owned_io_buffers_.Add(1);
io_buffer_pos_ = reinterpret_cast<uint8_t*>(io_buffer_->buffer());
io_buffer_bytes_left_ = io_buffer_->len();
if (io_buffer_->len() == 0) {
@@ -262,13 +246,7 @@ Status ScannerContext::Stream::GetBytesInternal(int64_t requested_len,
DCHECK_GT(requested_len, boundary_buffer_bytes_left_);
*out_buffer = NULL;
- if (boundary_buffer_bytes_left_ == 0) {
- if (contains_tuple_data_) {
- boundary_buffer_->Reset();
- } else {
- boundary_buffer_->Clear();
- }
- }
+ if (boundary_buffer_bytes_left_ == 0) boundary_buffer_->Clear();
DCHECK(ValidateBufferPointers());
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/scanner-context.h
----------------------------------------------------------------------
diff --git a/be/src/exec/scanner-context.h b/be/src/exec/scanner-context.h
index 3fe14aa..216209f 100644
--- a/be/src/exec/scanner-context.h
+++ b/be/src/exec/scanner-context.h
@@ -100,13 +100,6 @@ class ScannerContext {
/// If we are past the end of the scan range, no bytes are returned.
Status GetBuffer(bool peek, uint8_t** buffer, int64_t* out_len);
- /// Sets whether of not the resulting tuples contain ptrs into memory owned by
- /// the scanner context. This by default, is inferred from the scan_node tuple
- /// descriptor (i.e. contains string slots) but can be overridden. If possible,
- /// this should be set to false to reduce memory usage as resources can be reused
- /// and recycled more quickly.
- void set_contains_tuple_data(bool v) { contains_tuple_data_ = v; }
-
/// Callback that returns the buffer size to use when reading past the end of the scan
/// range. Reading past the end of the scan range is likely a remote read, so we want
/// find a good trade-off between io requests and data volume. Scanners that have
@@ -175,13 +168,10 @@ class ScannerContext {
/// Skip this text object.
bool SkipText(Status*);
- /// If 'batch' is not NULL and 'contains_tuple_data_' is true, attaches all completed
- /// io buffers and the boundary mem pool to 'batch'. If 'done' is set, all in-flight
- /// resources are also attached or released.
- /// If 'batch' is NULL then 'done' must be true or 'contains_tuple_data_' false. Such
- /// a call will release all completed resources. If 'done' is true all in-flight
- /// resources are also freed.
- void ReleaseCompletedResources(RowBatch* batch, bool done);
+ /// Release all completed resources in the context, i.e. I/O and boundary buffers
+ /// that the caller has finished reading. If 'done' is true all resources are
+ /// freed, even if the caller has not read that data yet.
+ void ReleaseCompletedResources(bool done);
private:
friend class ScannerContext;
@@ -189,10 +179,6 @@ class ScannerContext {
DiskIoMgr::ScanRange* scan_range_;
const HdfsFileDesc* file_desc_;
- /// If true, tuples will contain pointers into memory contained in this object.
- /// That memory (io buffers or boundary buffers) must be attached to the row batch.
- bool contains_tuple_data_;
-
/// Total number of bytes returned from GetBytes()
int64_t total_bytes_returned_;
@@ -240,7 +226,7 @@ class ScannerContext {
/// List of buffers that are completed but still have bytes referenced by the caller.
/// On the next GetBytes() call, these buffers are released (the caller by calling
/// GetBytes() signals it is done with its previous bytes). At this point the
- /// buffers are either returned to the io mgr or attached to the current row batch.
+ /// buffers are returned to the I/O manager.
std::deque<std::unique_ptr<DiskIoMgr::BufferDescriptor>> completed_io_buffers_;
Stream(ScannerContext* parent);
@@ -280,22 +266,23 @@ class ScannerContext {
return streams_[idx].get();
}
- /// If a non-NULL 'batch' is passed, attaches completed io buffers and boundary mem pools
- /// from all streams to 'batch'. Attaching only completed resources ensures that buffers
- /// (and their cleanup) trail the rows that reference them (row batches are consumed and
- /// cleaned up in order by the rest of the query).
- /// If 'done' is true, this is the final call for the current streams and any pending
- /// resources in each stream are also passed to the row batch. Callers which want to
- /// clear the streams from the context should also call ClearStreams().
- ///
- /// A NULL 'batch' may be passed to free all resources. It is only valid to pass a NULL
- /// 'batch' when also passing 'done'.
+ /// Returns completed I/O buffers to the I/O manager. If 'done' is true, this is the
+ /// final call for the current streams and any pending resources in each stream are
+ /// also freed. Callers which want to clear the streams from the context should also
+ /// call ClearStreams().
///
/// This must be called with 'done' set when the scanner is complete and no longer needs
/// any resources (e.g. tuple memory, io buffers) returned from the current streams.
/// After calling with 'done' set, this should be called again if new streams are
/// created via AddStream().
- void ReleaseCompletedResources(RowBatch* batch, bool done);
+ void ReleaseCompletedResources(bool done);
+
+ /// Overload with the signature expected by Impala-lzo to enable easier staging of
+ /// the API change. TODO: remove this once Impala-lzo is updated to use the new
+ /// signature.
+ void ReleaseCompletedResources(RowBatch* batch, bool done) {
+ ReleaseCompletedResources(done);
+ }
/// Releases all the Stream objects in the vector 'streams_' and reduces the vector's
/// size to 0.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/exec/text-converter.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/text-converter.cc b/be/src/exec/text-converter.cc
index 0cad05c..3bb65f2 100644
--- a/be/src/exec/text-converter.cc
+++ b/be/src/exec/text-converter.cc
@@ -103,8 +103,8 @@ void TextConverter::UnescapeString(const char* src, char* dest, int* len,
// store i8 %null_bit_set, i8* %null_byte_ptr
// ret i1 false
//}
-
-
+// TODO: convert this function to use cross-compilation + constant substitution in whole
+// or part. It is currently too complex and doesn't implement the full functionality.
Status TextConverter::CodegenWriteSlot(LlvmCodeGen* codegen,
TupleDescriptor* tuple_desc, SlotDescriptor* slot_desc, Function** fn,
const char* null_col_val, int len, bool check_null, bool strict_mode) {
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/runtime/row-batch.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/row-batch.cc b/be/src/runtime/row-batch.cc
index beed671..a6e935a 100644
--- a/be/src/runtime/row-batch.cc
+++ b/be/src/runtime/row-batch.cc
@@ -45,7 +45,7 @@ RowBatch::RowBatch(const RowDescriptor* row_desc, int capacity, MemTracker* mem_
flush_(FlushMode::NO_FLUSH_RESOURCES),
needs_deep_copy_(false),
num_tuples_per_row_(row_desc->tuple_descriptors().size()),
- auxiliary_mem_usage_(0),
+ attached_buffer_bytes_(0),
tuple_data_pool_(mem_tracker),
row_desc_(row_desc),
mem_tracker_(mem_tracker) {
@@ -72,7 +72,7 @@ RowBatch::RowBatch(
flush_(FlushMode::NO_FLUSH_RESOURCES),
needs_deep_copy_(false),
num_tuples_per_row_(input_batch.row_tuples.size()),
- auxiliary_mem_usage_(0),
+ attached_buffer_bytes_(0),
tuple_data_pool_(mem_tracker),
row_desc_(row_desc),
mem_tracker_(mem_tracker) {
@@ -144,9 +144,6 @@ RowBatch::RowBatch(
RowBatch::~RowBatch() {
tuple_data_pool_.FreeAll();
- for (int i = 0; i < io_buffers_.size(); ++i) {
- ExecEnv::GetInstance()->disk_io_mgr()->ReturnBuffer(move(io_buffers_[i]));
- }
for (BufferInfo& buffer_info : buffers_) {
ExecEnv::GetInstance()->buffer_pool()->FreeBuffer(
buffer_info.client, &buffer_info.buffer);
@@ -285,16 +282,9 @@ void RowBatch::SerializeInternal(int64_t size, DedupMap* distinct_tuples,
DCHECK_EQ(offset, size);
}
-void RowBatch::AddIoBuffer(unique_ptr<DiskIoMgr::BufferDescriptor> buffer) {
- DCHECK(buffer != NULL);
- auxiliary_mem_usage_ += buffer->buffer_len();
- buffer->TransferOwnership(mem_tracker_);
- io_buffers_.emplace_back(move(buffer));
-}
-
void RowBatch::AddBuffer(BufferPool::ClientHandle* client,
BufferPool::BufferHandle&& buffer, FlushMode flush) {
- auxiliary_mem_usage_ += buffer.len();
+ attached_buffer_bytes_ += buffer.len();
BufferInfo buffer_info;
buffer_info.client = client;
buffer_info.buffer = std::move(buffer);
@@ -307,26 +297,18 @@ void RowBatch::Reset() {
capacity_ = tuple_ptrs_size_ / (num_tuples_per_row_ * sizeof(Tuple*));
// TODO: Change this to Clear() and investigate the repercussions.
tuple_data_pool_.FreeAll();
- for (int i = 0; i < io_buffers_.size(); ++i) {
- ExecEnv::GetInstance()->disk_io_mgr()->ReturnBuffer(move(io_buffers_[i]));
- }
- io_buffers_.clear();
for (BufferInfo& buffer_info : buffers_) {
ExecEnv::GetInstance()->buffer_pool()->FreeBuffer(
buffer_info.client, &buffer_info.buffer);
}
buffers_.clear();
- auxiliary_mem_usage_ = 0;
+ attached_buffer_bytes_ = 0;
flush_ = FlushMode::NO_FLUSH_RESOURCES;
needs_deep_copy_ = false;
}
void RowBatch::TransferResourceOwnership(RowBatch* dest) {
dest->tuple_data_pool_.AcquireData(&tuple_data_pool_, false);
- for (int i = 0; i < io_buffers_.size(); ++i) {
- dest->AddIoBuffer(move(io_buffers_[i]));
- }
- io_buffers_.clear();
for (BufferInfo& buffer_info : buffers_) {
dest->AddBuffer(
buffer_info.client, std::move(buffer_info.buffer), FlushMode::NO_FLUSH_RESOURCES);
@@ -362,7 +344,7 @@ void RowBatch::AcquireState(RowBatch* src) {
// The destination row batch should be empty.
DCHECK(!needs_deep_copy_);
DCHECK_EQ(num_rows_, 0);
- DCHECK_EQ(auxiliary_mem_usage_, 0);
+ DCHECK_EQ(attached_buffer_bytes_, 0);
num_rows_ = src->num_rows_;
capacity_ = src->capacity_;
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/19c17e64/be/src/runtime/row-batch.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/row-batch.h b/be/src/runtime/row-batch.h
index 49dd066..5a7edd4 100644
--- a/be/src/runtime/row-batch.h
+++ b/be/src/runtime/row-batch.h
@@ -43,16 +43,18 @@ class TupleDescriptor;
/// A RowBatch encapsulates a batch of rows, each composed of a number of tuples.
/// The maximum number of rows is fixed at the time of construction.
-/// The row batch reference a few different sources of memory.
-/// 1. TupleRow ptrs - may be malloc'd and owned by the RowBatch or allocated from
-/// the tuple pool, depending on whether legacy joins and aggs are enabled.
-/// See the comment on tuple_ptrs_ for more details.
-/// 2. Tuple memory - this is allocated (or transferred to) the row batches tuple pool.
-/// 3. Auxiliary tuple memory (e.g. string data) - this can either be stored externally
-/// (don't copy strings) or from the tuple pool (strings are copied). If external,
-/// the data is in an io buffer that may not be attached to this row batch. The
-/// creator of that row batch has to make sure that the io buffer is not recycled
-/// until all batches that reference the memory have been consumed.
+/// The row batch can reference various types of memory.
+/// 1. TupleRow ptrs - malloc'd and owned by the RowBatch. See the comment on
+/// tuple_ptrs_ for more details.
+/// 2. Fixed and variable-length tuple data. This memory may be directly attached to
+/// the batch: either in the batch's MemPool or in an attached buffer. Or it may
+/// live elsewhere - either in a subsequent batch returned by an ExecNode or
+/// still be owned by the ExecNode that produced the batch. In those cases the
+/// owner of this RowBatch must be careful not to close the producing ExecNode
+/// or free resources from trailing batches while the batch's data is still being
+/// used.
+/// TODO: IMPALA-4179: simplify the ownership transfer model.
+///
/// In order to minimize memory allocations, RowBatches and TRowBatches that have been
/// serialized and sent over the wire should be reused (this prevents compression_scratch_
/// from being needlessly reallocated).
@@ -62,13 +64,13 @@ class TupleDescriptor;
/// and reference memory outside of the row batch. This results in most row batches
/// having a very small memory footprint and in some row batches having a very large
/// one (it contains all the memory that other row batches are referencing). An example
-/// is IoBuffers which are only attached to one row batch. Only when the row batch reaches
+/// is buffers which are only attached to one row batch. Only when the row batch reaches
/// a blocking operator or the root of the fragment is the row batch memory freed.
/// This means that in some cases (e.g. very selective queries), we still need to
/// pass the row batch through the exec nodes (even if they have no rows) to trigger
-/// memory deletion. AtCapacity() encapsulates the check that we are not accumulating
-/// excessive memory.
-//
+/// memory deletion. AtCapacity() encapsulates the check that the batch does not have
+/// excessive memory attached to it.
+///
/// A row batch is considered at capacity if all the rows are full or it has accumulated
/// auxiliary memory up to a soft cap. (See at_capacity_mem_usage_ comment).
class RowBatch {
@@ -138,7 +140,7 @@ class RowBatch {
// MarkFlushResources().
DCHECK((!needs_deep_copy_ && flush_ == FlushMode::NO_FLUSH_RESOURCES)
|| num_rows_ == capacity_);
- int64_t mem_usage = auxiliary_mem_usage_ + tuple_data_pool_.total_allocated_bytes();
+ int64_t mem_usage = attached_buffer_bytes_ + tuple_data_pool_.total_allocated_bytes();
return num_rows_ == capacity_ || mem_usage >= AT_CAPACITY_MEM_USAGE;
}
@@ -203,24 +205,18 @@ class RowBatch {
};
int num_tuples_per_row() { return num_tuples_per_row_; }
- int row_byte_size() { return num_tuples_per_row_ * sizeof(Tuple*); }
MemPool* tuple_data_pool() { return &tuple_data_pool_; }
- int num_io_buffers() const { return io_buffers_.size(); }
int num_buffers() const { return buffers_.size(); }
/// Resets the row batch, returning all resources it has accumulated.
void Reset();
- /// Add io buffer to this row batch.
- void AddIoBuffer(std::unique_ptr<DiskIoMgr::BufferDescriptor> buffer);
-
/// Adds a buffer to this row batch. The buffer is deleted when freeing resources.
/// The buffer's memory remains accounted against the original owner, even when the
/// ownership of batches is transferred. If the original owner wants the memory to be
/// released, it should call this with 'mode' FLUSH_RESOURCES (see MarkFlushResources()
/// for further explanation).
- /// TODO: IMPALA-4179: after IMPALA-3200, simplify the ownership transfer model and
- /// make it consistent between buffers and I/O buffers.
+ /// TODO: IMPALA-4179: simplify the ownership transfer model.
void AddBuffer(BufferPool::ClientHandle* client, BufferPool::BufferHandle&& buffer,
FlushMode flush);
@@ -230,10 +226,10 @@ class RowBatch {
/// can be added. The "flush" mark is transferred by TransferResourceOwnership(). This
/// ensures that batches are flushed by streaming operators all the way up the operator
/// tree. Blocking operators can still accumulate batches with this flag.
- /// TODO: IMPALA-3200: blocking operators should acquire all memory resources including
- /// attached blocks/buffers, so that MarkFlushResources() can guarantee that the
+ /// TODO: IMPALA-4179: blocking operators should acquire all memory resources including
+ /// attached buffers, so that MarkFlushResources() can guarantee that the
/// resources will not be accounted against the original operator (this is currently
- /// not true for Blocks, which can't be transferred).
+ /// not true for buffers, which aren't transferred).
void MarkFlushResources() {
DCHECK_LE(num_rows_, capacity_);
capacity_ = num_rows_;
@@ -256,7 +252,7 @@ class RowBatch {
bool needs_deep_copy() { return needs_deep_copy_; }
/// Transfer ownership of resources to dest. This includes tuple data in mem
- /// pool and io buffers.
+ /// pool and buffers.
void TransferResourceOwnership(RowBatch* dest);
void CopyRow(TupleRow* src, TupleRow* dest) {
@@ -277,7 +273,7 @@ class RowBatch {
memset(row, 0, num_tuples_per_row_ * sizeof(Tuple*));
}
- /// Acquires state from the 'src' row batch into this row batch. This includes all IO
+ /// Acquires state from the 'src' row batch into this row batch. This includes all
/// buffers and tuple data.
/// This row batch must be empty and have the same row descriptor as the src batch.
/// This is used for scan nodes which produce RowBatches asynchronously. Typically,
@@ -399,9 +395,8 @@ class RowBatch {
int tuple_ptrs_size_;
Tuple** tuple_ptrs_;
- /// Sum of all auxiliary bytes. This includes IoBuffers and memory from
- /// TransferResourceOwnership().
- int64_t auxiliary_mem_usage_;
+ /// Total bytes of BufferPool buffers attached to this batch.
+ int64_t attached_buffer_bytes_;
/// holding (some of the) data referenced by rows
MemPool tuple_data_pool_;
@@ -415,11 +410,6 @@ class RowBatch {
MemTracker* mem_tracker_; // not owned
- /// IO buffers current owned by this row batch. Ownership of IO buffers transfer
- /// between row batches. Any IO buffer will be owned by at most one row batch
- /// (i.e. they are not ref counted) so most row batches don't own any.
- std::vector<std::unique_ptr<DiskIoMgr::BufferDescriptor>> io_buffers_;
-
struct BufferInfo {
BufferPool::ClientHandle* client;
BufferPool::BufferHandle buffer;
[3/3] incubator-impala git commit: IMPALA-6171: Revert "IMPALA-1575:
part 2: yield admission control resources"
Posted by ta...@apache.org.
IMPALA-6171: Revert "IMPALA-1575: part 2: yield admission control resources"
This reverts commit fe90867d890c71bfdcf8ff941f8ec51e36083f25.
Change-Id: I3eec4b5a6ff350933ffda0bb80949c5960ecdf25
Reviewed-on: http://gerrit.cloudera.org:8080/8499
Reviewed-by: Thomas Tauber-Marshall <tm...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/a772f845
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/a772f845
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/a772f845
Branch: refs/heads/master
Commit: a772f845628f19b816935cd4770714c980c553a6
Parents: 19c17e6
Author: Tim Armstrong <ta...@cloudera.com>
Authored: Wed Nov 8 10:33:52 2017 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Wed Nov 8 22:03:59 2017 +0000
----------------------------------------------------------------------
be/src/runtime/coordinator.cc | 29 +---
be/src/runtime/coordinator.h | 38 +----
be/src/runtime/mem-tracker.cc | 6 +-
be/src/runtime/mem-tracker.h | 16 +-
be/src/runtime/query-state.cc | 3 -
be/src/scheduling/admission-controller.cc | 13 +-
be/src/scheduling/admission-controller.h | 15 +-
be/src/service/client-request-state.cc | 11 ++
.../custom_cluster/test_admission_controller.py | 155 +++++++------------
9 files changed, 94 insertions(+), 192 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/runtime/coordinator.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/coordinator.cc b/be/src/runtime/coordinator.cc
index ae0c9ad..02c0ca1 100644
--- a/be/src/runtime/coordinator.cc
+++ b/be/src/runtime/coordinator.cc
@@ -37,7 +37,6 @@
#include "runtime/coordinator-backend-state.h"
#include "runtime/debug-options.h"
#include "runtime/query-state.h"
-#include "scheduling/admission-controller.h"
#include "scheduling/scheduler.h"
#include "util/bloom-filter.h"
#include "util/counting-barrier.h"
@@ -80,9 +79,6 @@ Coordinator::Coordinator(
Coordinator::~Coordinator() {
DCHECK(released_exec_resources_)
<< "ReleaseExecResources() must be called before Coordinator is destroyed";
- DCHECK(released_admission_control_resources_)
- << "ReleaseAdmissionControlResources() must be called before Coordinator is "
- << "destroyed";
if (query_state_ != nullptr) {
ExecEnv::GetInstance()->query_exec_mgr()->ReleaseQueryState(query_state_);
}
@@ -833,16 +829,15 @@ Status Coordinator::Wait() {
// Execution of query fragments has finished. We don't need to hold onto query execution
// resources while we finalize the query.
ReleaseExecResources();
+
// Query finalization is required only for HDFS table sinks
if (needs_finalization_) RETURN_IF_ERROR(FinalizeQuery());
- // Release admission control resources after we'd done the potentially heavyweight
- // finalization.
- ReleaseAdmissionControlResources();
query_profile_->AddInfoString(
"DML Stats", DataSink::OutputDmlStats(per_partition_status_, "\n"));
// For DML queries, when Wait is done, the query is complete.
ComputeQuerySummary();
+
return status;
}
@@ -873,11 +868,10 @@ Status Coordinator::GetNext(QueryResultSet* results, int max_rows, bool* eos) {
returned_all_results_ = true;
// release query execution resources here, since we won't be fetching more result rows
ReleaseExecResources();
+
// wait for all backends to complete before computing the summary
// TODO: relocate this so GetNext() won't have to wait for backends to complete?
RETURN_IF_ERROR(WaitForBackendCompletion());
- // Release admission control resources after backends are finished.
- ReleaseAdmissionControlResources();
// if the query completed successfully, compute the summary
if (query_status_.ok()) ComputeQuerySummary();
}
@@ -916,7 +910,7 @@ void Coordinator::CancelInternal() {
backend_completion_cv_.NotifyAll();
ReleaseExecResourcesLocked();
- ReleaseAdmissionControlResourcesLocked();
+
// Report the summary with whatever progress the query made before being cancelled.
ComputeQuerySummary();
}
@@ -1088,21 +1082,6 @@ void Coordinator::ReleaseExecResourcesLocked() {
// caching. The query MemTracker will be cleaned up later.
}
-void Coordinator::ReleaseAdmissionControlResources() {
- lock_guard<mutex> l(lock_);
- ReleaseAdmissionControlResourcesLocked();
-}
-
-void Coordinator::ReleaseAdmissionControlResourcesLocked() {
- if (released_admission_control_resources_) return;
- LOG(INFO) << "Release admssion control resources for query "
- << PrintId(query_ctx_.query_id);
- AdmissionController* admission_controller =
- ExecEnv::GetInstance()->admission_controller();
- if (admission_controller != nullptr) admission_controller->ReleaseQuery(schedule_);
- released_admission_control_resources_ = true;
-}
-
void Coordinator::UpdateFilter(const TUpdateFilterParams& params) {
DCHECK_NE(filter_mode_, TRuntimeFilterMode::OFF)
<< "UpdateFilter() called although runtime filters are disabled";
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/runtime/coordinator.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/coordinator.h b/be/src/runtime/coordinator.h
index e7ddee9..65ef678 100644
--- a/be/src/runtime/coordinator.h
+++ b/be/src/runtime/coordinator.h
@@ -189,8 +189,7 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
class FilterState;
class FragmentStats;
- /// owned by the ClientRequestState that owns this coordinator
- const QuerySchedule& schedule_;
+ const QuerySchedule schedule_;
/// copied from TQueryExecRequest; constant across all fragments
TQueryCtx query_ctx_;
@@ -347,9 +346,6 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
/// True if and only if ReleaseExecResources() has been called.
bool released_exec_resources_ = false;
- /// True if and only if ReleaseAdmissionControlResources() has been called.
- bool released_admission_control_resources_ = false;
-
/// Returns a local object pool.
ObjectPool* obj_pool() { return obj_pool_.get(); }
@@ -441,38 +437,6 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
/// Same as ReleaseExecResources() except the lock must be held by the caller.
void ReleaseExecResourcesLocked();
-
- /// Releases admission control resources for use by other queries.
- /// This should only be called if one of following preconditions is satisfied for each
- /// backend on which the query is executing:
- /// * The backend finished execution.
- /// Rationale: the backend isn't consuming resources.
- //
- /// * A cancellation RPC was delivered to the backend.
- /// Rationale: the backend will be cancelled and release resources soon. By the
- /// time a newly admitted query fragment starts up on the backend and starts consuming
- /// resources, the resources from this query will probably have been released.
- //
- /// * Sending the cancellation RPC to the backend failed
- /// Rationale: the backend is either down or will tear itself down when it next tries
- /// to send a status RPC to the coordinator. It's possible that the fragment will be
- /// slow to tear down and we could overadmit and cause query failures. However, given
- /// the communication errors, we need to proceed based on incomplete information about
- /// the state of the cluster. We choose to optimistically assume that the backend will
- /// tear itself down in a timely manner and admit more queries instead of
- /// pessimistically queueing queries while we wait for a response from a backend that
- /// may never come.
- ///
- /// Calling WaitForBackendCompletion() or CancelInternal() before this function is
- /// sufficient to satisfy the above preconditions. If the query has an expensive
- /// finalization step post query execution (e.g. a DML statement), then this should
- /// be called after that completes to avoid over-admitting queries.
- ///
- /// Acquires lock_. Idempotent.
- void ReleaseAdmissionControlResources();
-
- /// Same as ReleaseAdmissionControlResources() except lock must be held by caller.
- void ReleaseAdmissionControlResourcesLocked();
};
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/runtime/mem-tracker.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/mem-tracker.cc b/be/src/runtime/mem-tracker.cc
index 4162e8a..a8cb37e 100644
--- a/be/src/runtime/mem-tracker.cc
+++ b/be/src/runtime/mem-tracker.cc
@@ -141,13 +141,11 @@ int64_t MemTracker::GetPoolMemReserved() {
lock_guard<SpinLock> l(child_trackers_lock_);
for (MemTracker* child : child_trackers_) {
int64_t child_limit = child->limit();
- bool query_exec_finished = child->query_exec_finished_.Load() != 0;
- if (child_limit > 0 && !query_exec_finished) {
+ if (child_limit > 0) {
// Make sure we don't overflow if the query limits are set to ridiculous values.
mem_reserved += std::min(child_limit, MemInfo::physical_mem());
} else {
- DCHECK(query_exec_finished || child_limit == -1)
- << child->LogUsage(UNLIMITED_DEPTH);
+ DCHECK_EQ(child_limit, -1) << child->LogUsage(UNLIMITED_DEPTH);
mem_reserved += child->consumption();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/runtime/mem-tracker.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/mem-tracker.h b/be/src/runtime/mem-tracker.h
index 539f973..1260351 100644
--- a/be/src/runtime/mem-tracker.h
+++ b/be/src/runtime/mem-tracker.h
@@ -294,9 +294,8 @@ class MemTracker {
/// Returns the memory 'reserved' by this resource pool mem tracker, which is the sum
/// of the memory reserved by the queries in it (i.e. its child trackers). The mem
- /// reserved for a query that is currently executing is its limit_, if set (which
- /// should be the common case with admission control). Otherwise, if the query has
- /// no limit or the query is finished executing, the current consumption is used.
+ /// reserved for a query is its limit_, if set (which should be the common case with
+ /// admission control). Otherwise the current consumption is used.
int64_t GetPoolMemReserved();
/// Returns the memory consumed in bytes.
@@ -352,11 +351,6 @@ class MemTracker {
Status MemLimitExceeded(RuntimeState* state, const std::string& details,
int64_t failed_allocation = 0) WARN_UNUSED_RESULT;
- void set_query_exec_finished() {
- DCHECK(is_query_mem_tracker_);
- query_exec_finished_.Store(1);
- }
-
static const std::string COUNTER_NAME;
private:
@@ -392,12 +386,6 @@ class MemTracker {
/// True if this is a Query MemTracker returned from CreateQueryMemTracker().
bool is_query_mem_tracker_ = false;
- /// Only used if 'is_query_mem_tracker_' is true.
- /// 0 if the query is still executing or 1 if it has finished executing. Before
- /// it has finished executing, the tracker limit is treated as "reserved memory"
- /// for the purpose of admission control - see GetPoolMemReserved().
- AtomicInt32 query_exec_finished_{0};
-
/// Only valid for MemTrackers returned from CreateQueryMemTracker()
TUniqueId query_id_;
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/runtime/query-state.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/query-state.cc b/be/src/runtime/query-state.cc
index fd72f6a..6796c82 100644
--- a/be/src/runtime/query-state.cc
+++ b/be/src/runtime/query-state.cc
@@ -82,9 +82,6 @@ void QueryState::ReleaseExecResources() {
if (initial_reservations_ != nullptr) initial_reservations_->ReleaseResources();
if (buffer_reservation_ != nullptr) buffer_reservation_->Close();
if (desc_tbl_ != nullptr) desc_tbl_->ReleaseResources();
- // Mark the query as finished on the query MemTracker so that admission control will
- // not consider the whole query memory limit to be "reserved".
- query_mem_tracker_->set_query_exec_finished();
// At this point query execution should not be consuming any resources but some tracked
// memory may still be used by the ClientRequestState for result caching. The query
// MemTracker will be closed later when this QueryState is torn down.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/scheduling/admission-controller.cc
----------------------------------------------------------------------
diff --git a/be/src/scheduling/admission-controller.cc b/be/src/scheduling/admission-controller.cc
index 99f659a..a5d62b4 100644
--- a/be/src/scheduling/admission-controller.cc
+++ b/be/src/scheduling/admission-controller.cc
@@ -608,19 +608,20 @@ Status AdmissionController::AdmitQuery(QuerySchedule* schedule) {
}
}
-void AdmissionController::ReleaseQuery(const QuerySchedule& schedule) {
- if (!schedule.is_admitted()) return; // No-op if query was not admitted
- const string& pool_name = schedule.request_pool();
+Status AdmissionController::ReleaseQuery(QuerySchedule* schedule) {
+ if (!schedule->is_admitted()) return Status::OK(); // No-op if query was not admitted
+ const string& pool_name = schedule->request_pool();
{
lock_guard<mutex> lock(admission_ctrl_lock_);
PoolStats* stats = GetPoolStats(pool_name);
- stats->Release(schedule);
- UpdateHostMemAdmitted(schedule, -schedule.GetPerHostMemoryEstimate());
+ stats->Release(*schedule);
+ UpdateHostMemAdmitted(*schedule, -schedule->GetPerHostMemoryEstimate());
pools_for_updates_.insert(pool_name);
- VLOG_RPC << "Released query id=" << schedule.query_id() << " "
+ VLOG_RPC << "Released query id=" << schedule->query_id() << " "
<< stats->DebugString();
}
dequeue_cv_.NotifyOne();
+ return Status::OK();
}
// Statestore subscriber callback for IMPALA_REQUEST_QUEUE_TOPIC.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/scheduling/admission-controller.h
----------------------------------------------------------------------
diff --git a/be/src/scheduling/admission-controller.h b/be/src/scheduling/admission-controller.h
index 2830bee..86b8338 100644
--- a/be/src/scheduling/admission-controller.h
+++ b/be/src/scheduling/admission-controller.h
@@ -96,13 +96,12 @@ class ExecEnv;
/// a) Mem Reserved: the amount of memory that has been reported as reserved by all
/// backends, which come from the statestore topic updates. The values that are sent
/// come from the pool mem trackers in UpdateMemTrackerStats(), which reflects the
-/// memory reserved by fragments that have begun execution. For queries that are
-/// executing and have mem limits, the limit is considered to be its reserved memory
-/// because it may consume up to that limit. Otherwise the query's current consumption
-/// is used (see MemTracker::GetPoolMemReserved()). The per-pool and per-host
-/// aggregates are computed in UpdateClusterAggregates(). This state, once all updates
-/// are fully distributed and aggregated, provides enough information to make
-/// admission decisions by any impalad. However, this requires waiting for both
+/// memory reserved by fragments that have begun execution. For queries that have mem
+/// limits, the limit is considered to be its reserved memory, otherwise the current
+/// consumption is used (see MemTracker::GetPoolMemReserved()). The per-pool and
+/// per-host aggregates are computed in UpdateClusterAggregates(). This state, once
+/// all updates are fully distributed and aggregated, provides enough information to
+/// make admission decisions by any impalad. However, this requires waiting for both
/// admitted requests to start all remote fragments and then for the updated state to
/// be distributed via the statestore.
/// b) Mem Admitted: the amount of memory required (i.e. the value used in admission,
@@ -197,7 +196,7 @@ class AdmissionController {
/// been submitted via AdmitQuery(). (If the request was not admitted, this is
/// a no-op.)
/// This does not block.
- void ReleaseQuery(const QuerySchedule& schedule);
+ Status ReleaseQuery(QuerySchedule* schedule);
/// Registers the request queue topic with the statestore.
Status Init();
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/be/src/service/client-request-state.cc
----------------------------------------------------------------------
diff --git a/be/src/service/client-request-state.cc b/be/src/service/client-request-state.cc
index 5d00d24..38c556d 100644
--- a/be/src/service/client-request-state.cc
+++ b/be/src/service/client-request-state.cc
@@ -572,6 +572,17 @@ void ClientRequestState::Done() {
// Update result set cache metrics, and update mem limit accounting before tearing
// down the coordinator.
ClearResultCache();
+
+ if (coord_.get() != NULL) {
+ // Release any reserved resources.
+ if (exec_env_->admission_controller() != nullptr) {
+ Status status = exec_env_->admission_controller()->ReleaseQuery(schedule_.get());
+ if (!status.ok()) {
+ LOG(WARNING) << "Failed to release resources of query " << schedule_->query_id()
+ << " because of error: " << status.GetDetail();
+ }
+ }
+ }
}
Status ClientRequestState::Exec(const TMetadataOpRequest& exec_request) {
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/a772f845/tests/custom_cluster/test_admission_controller.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_admission_controller.py b/tests/custom_cluster/test_admission_controller.py
index 02fd6db..eb1e942 100644
--- a/tests/custom_cluster/test_admission_controller.py
+++ b/tests/custom_cluster/test_admission_controller.py
@@ -40,25 +40,24 @@ from TCLIService import TCLIService
LOG = logging.getLogger('admission_test')
-# The query used for testing. It is important that this query be able to fetch many
-# rows. This allows a thread to stay active by fetching one row at a time. The
-# where clause is for debugging purposes; each thread will insert its id so
+# We set a WAIT debug action so it doesn't complete the execution of this query. The
+# limit is a parameter for debugging purposes; each thread will insert its id so
# that running queries can be correlated with the thread that submitted them.
-QUERY = "select * from alltypes where id != %s"
+QUERY = "select * from alltypes where id != %s"# limit %s"
# Time to sleep (in milliseconds) between issuing queries. The default statestore
# heartbeat is 500ms, so the lower the delay the more we can submit before the global
# state is updated. When the delay is at least the statestore heartbeat frequency, all
# state should be visible by every impalad by the time the next query is submitted.
-SUBMISSION_DELAY_MS = [0, 50, 100, 600]
+SUBMISSION_DELAY_MS = [50]
# The number of queries to submit. The test does not support fewer queries than
# MAX_NUM_CONCURRENT_QUERIES + MAX_NUM_QUEUED_QUERIES to keep some validation logic
# simple.
-NUM_QUERIES = [15, 30, 50]
+NUM_QUERIES = [50]
# Whether we will submit queries to all available impalads (in a round-robin fashion)
-ROUND_ROBIN_SUBMISSION = [True, False]
+ROUND_ROBIN_SUBMISSION = [True]
# The query pool to use. The impalads should be configured to recognize this
# pool with the parameters below.
@@ -87,9 +86,6 @@ _STATESTORED_ARGS = "-statestore_heartbeat_frequency_ms=%s "\
# Key in the query profile for the query options.
PROFILE_QUERY_OPTIONS_KEY = "Query Options (set by configuration): "
-# The different ways that a query thread can end its query.
-QUERY_END_BEHAVIORS = ['EOS', 'CLIENT_CANCEL', 'QUERY_TIMEOUT', 'CLIENT_CLOSE']
-
def impalad_admission_ctrl_flags(max_requests, max_queued, pool_max_mem,
proc_mem_limit = None):
if proc_mem_limit is not None:
@@ -369,13 +365,12 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
class TestAdmissionControllerStress(TestAdmissionControllerBase):
"""Submits a number of queries (parameterized) with some delay between submissions
(parameterized) and the ability to submit to one impalad or many in a round-robin
- fashion. Each query is submitted on a separate thread. After admission, the query
- thread will block with the query open and wait for the main thread to notify it to
- end its query. The query thread can end its query by fetching to the end, cancelling
- itself, closing itself, or waiting for the query timeout to take effect. Depending
- on the test parameters a varying number of queries will be admitted, queued, and
- rejected. After the queries are admitted, the main thread will request each admitted
- query thread to end its query and allow queued queries to be admitted.
+ fashion. The queries are set with the WAIT debug action so that we have more control
+ over the state that the admission controller uses to make decisions. Each query is
+ submitted on a separate thread. Depending on the test parameters a varying number of
+ queries will be admitted, queued, and rejected. Once queries are admitted, the query
+ execution blocks and we can cancel the query in order to allow another queued query to
+ be admitted.
The test tracks the state of the admission controller using the metrics from each
impalad to do the following:
@@ -383,15 +378,13 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase):
queued, and rejected requests should sum to the number of queries and that the
values are reasonable given the test parameters.
(2) While there are running queries:
- * Request the currently running queries to end and wait for the queries to end.
- Verify the metric for the number of completed queries. The threads that
- submitted those queries will keep their connections open until the entire test
- completes. This verifies that admission control is tied to the end of the query
- and does not depend on closing the connection.
+ * Cancel the currently running queries (they are blocked with the WAIT debug action)
+ and verify the metric for the number of completed queries. The threads that
+ submitted those queries should complete.
* Check that queued requests are then dequeued and verify using the metric for the
number of dequeued requests. The threads that were waiting to submit the query
should then insert themselves into a list of currently running queries and then
- wait for a notification from the main thread.
+ fetch() the results (which will block).
(3) After all queries have completed, check that the final number of admitted,
queued, and rejected requests are reasonable given the test parameters. When
submitting to a single impalad, we know exactly what the values should be,
@@ -435,7 +428,6 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase):
self.executing_threads = list()
def teardown(self):
- # Set shutdown for all threads (cancel if needed)
for thread in self.all_threads:
try:
thread.lock.acquire()
@@ -450,9 +442,6 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase):
client.close()
finally:
thread.lock.release()
-
- # Wait for all threads to exit
- for thread in self.all_threads:
thread.join(5)
LOG.debug("Join thread for query num %s %s", thread.query_num,
"TIMED OUT" if thread.isAlive() else "")
@@ -548,39 +537,36 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase):
LOG.debug("Found all %s admitted threads after %s seconds", num_threads,
round(time() - start_time, 1))
- def end_admitted_queries(self, num_queries):
+ def cancel_admitted_queries(self, num_queries):
"""
- Requests each admitted query to end its query.
+ Cancels queries on threads that are currently blocked on query execution.
"""
assert len(self.executing_threads) >= num_queries
- LOG.debug("Requesting {0} clients to end queries".format(num_queries))
-
- # Request admitted clients to end their queries
- current_executing_queries = []
+ LOG.debug("Cancelling %s queries", num_queries)
for i in xrange(num_queries):
# pop() is thread-safe, it's OK if another thread is appending concurrently.
thread = self.executing_threads.pop(0)
LOG.debug("Cancelling query %s", thread.query_num)
+ # The other thread sets the query_state before appending itself to the list,
+ # and will not change its state until it is cancelled by this thread.
assert thread.query_state == 'ADMITTED'
- current_executing_queries.append(thread)
- thread.query_state = 'REQUEST_QUERY_END'
-
- # Wait for the queries to end
- start_time = time()
- while True:
- all_done = True
- for thread in self.all_threads:
- if thread.query_state == 'REQUEST_QUERY_END':
- all_done = False
- if all_done:
- break
- assert (time() - start_time < STRESS_TIMEOUT),\
- "Timed out waiting %s seconds for query end" % (STRESS_TIMEOUT,)
- sleep(1)
+ client = thread.impalad.service.create_beeswax_client()
+ try:
+ cancel_result = client.cancel(thread.query_handle)
+ assert cancel_result.status_code == 0,\
+ 'Unexpected status code from cancel request: %s' % cancel_result
+ # Wait for the query to be cancelled and return
+ thread.join(20)
+ LOG.debug("Cancelled admitted query %s %s",
+ thread.query_num, "TIMED OUT" if thread.isAlive() else "")
+ assert not thread.isAlive()
+ assert thread.query_state == 'COMPLETED'
+ finally:
+ client.close()
class SubmitQueryThread(threading.Thread):
def __init__(self, impalad, additional_query_options, vector, query_num,
- query_end_behavior, executing_threads):
+ executing_threads):
"""
executing_threads must be provided so that this thread can add itself when the
query is admitted and begins execution.
@@ -590,7 +576,6 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase):
self.vector = vector
self.additional_query_options = additional_query_options
self.query_num = query_num
- self.query_end_behavior = query_end_behavior
self.impalad = impalad
self.error = None
# query_state is defined and used only by the test code, not a property exposed by
@@ -614,6 +599,7 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase):
return
exec_options = self.vector.get_value('exec_option')
+ exec_options['debug_action'] = '0:GETNEXT:WAIT'
exec_options.update(self.additional_query_options)
query = QUERY % (self.query_num,)
self.query_state = 'SUBMITTING'
@@ -621,9 +607,6 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase):
ImpalaTestSuite.change_database(client, self.vector.get_value('table_format'))
client.set_configuration(exec_options)
- if self.query_end_behavior == 'QUERY_TIMEOUT':
- client.execute("SET QUERY_TIMEOUT_S=5")
-
LOG.debug("Submitting query %s", self.query_num)
self.query_handle = client.execute_async(query)
except ImpalaBeeswaxException as e:
@@ -644,22 +627,22 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase):
# The thread becomes visible to the main thread when it is added to the
# shared list of executing_threads. append() is atomic and thread-safe.
self.executing_threads.append(self)
-
- # Synchronize with the main thread. At this point, the thread is executing a
- # query. It needs to wait until the main thread requests it to end its query.
- while not self.shutdown:
- # The QUERY_TIMEOUT needs to stay active until the main thread requests it
- # to end. Otherwise, the query may get cancelled early. Fetch a row every
- # second to avoid going idle.
- if self.query_end_behavior == 'QUERY_TIMEOUT' and \
- self.query_state != 'COMPLETED':
- client.fetch(query, self.query_handle, 1)
- if self.query_state == 'REQUEST_QUERY_END':
- self._end_query(client, query)
- # The query has released admission control resources
+ try:
+ # fetch() will block until we cancel the query from the main thread
+ # (unless an unexpected error occurs). If an error occurs on the main therad,
+ # it is possible that teardown() cancels this query before we call fetch(). In
+ # that case a different exception is thrown and we handle it gracefully.
+ client.fetch(query, self.query_handle)
+ except ImpalaBeeswaxException as e:
+ if "Cancelled" in str(e):
+ LOG.debug("Query %s completed", self.query_num)
self.query_state = 'COMPLETED'
self.query_handle = None
- sleep(1)
+ elif "Invalid or unknown query handle" in str(e):
+ # May happen if the test is being torn down early (i.e. an error occurred).
+ LOG.debug("Query %s already cancelled in test shutdown.")
+ else:
+ raise e
except Exception as e:
LOG.exception(e)
# Unknown errors will be raised later
@@ -670,27 +653,6 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase):
if client is not None:
client.close()
- def _end_query(self, client, query):
- """Bring the query to the appropriate end state defined by self.query_end_behaviour.
- Returns once the query has reached that state."""
- if self.query_end_behavior == 'QUERY_TIMEOUT':
- # Sleep and wait for the query to be cancelled. The cancellation will
- # set the state to EXCEPTION.
- start_time = time()
- while (client.get_state(self.query_handle) != \
- client.QUERY_STATES['EXCEPTION']):
- assert (time() - start_time < STRESS_TIMEOUT),\
- "Timed out waiting %s seconds for query cancel" % (STRESS_TIMEOUT,)
- sleep(1)
- elif self.query_end_behavior == 'EOS':
- # Fetch all rows so we hit eos.
- client.fetch(query, self.query_handle)
- elif self.query_end_behavior == 'CLIENT_CANCEL':
- client.cancel(self.query_handle)
- else:
- assert self.query_end_behavior == 'CLIENT_CLOSE'
- client.close_query(self.query_handle)
-
def _check_queries_page_resource_pools(self):
"""Checks that all queries in the '/queries' webpage json have the correct resource
pool (this is called after all queries have been admitted, queued, or rejected, so
@@ -731,11 +693,14 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase):
initial_metrics = self.get_admission_metrics();
log_metrics("Initial metrics: ", initial_metrics);
- for query_num in xrange(num_queries):
+ # Want query_num to start at 1 because this gets used as the limit in the query to
+ # help debugging (we can associate a running query with a thread). If we start at 0,
+ # that query would be evaluated as a constant expression and never hit the WAIT debug
+ # action.
+ for query_num in xrange(1, num_queries + 1):
impalad = self.impalads[query_num % len(self.impalads)]
- query_end_behavior = QUERY_END_BEHAVIORS[query_num % len(QUERY_END_BEHAVIORS)]
thread = self.SubmitQueryThread(impalad, additional_query_options, vector,
- query_num, query_end_behavior, self.executing_threads)
+ query_num, self.executing_threads)
thread.start()
self.all_threads.append(thread)
sleep(submission_delay_ms / 1000.0)
@@ -770,10 +735,10 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase):
while len(self.executing_threads) > 0:
curr_metrics = self.get_admission_metrics();
log_metrics("Main loop, curr_metrics: ", curr_metrics);
- num_to_end = len(self.executing_threads)
- LOG.debug("Main loop, will request %s queries to end", num_to_end)
- self.end_admitted_queries(num_to_end)
- self.wait_for_metric_changes(['released'], curr_metrics, num_to_end)
+ num_to_cancel = len(self.executing_threads)
+ LOG.debug("Main loop, will cancel %s queries", num_to_cancel)
+ self.cancel_admitted_queries(num_to_cancel)
+ self.wait_for_metric_changes(['released'], curr_metrics, num_to_cancel)
num_queued_remaining =\
curr_metrics['queued'] - curr_metrics['dequeued'] - curr_metrics['timed-out']