You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by mi...@apache.org on 2018/01/08 17:04:02 UTC
[1/3] impala git commit: IMPALA-5522:Use tracked memory for
DictDecoder and DictEncoder
Repository: impala
Updated Branches:
refs/heads/master d25607d01 -> 6a87eb20a
IMPALA-5522:Use tracked memory for DictDecoder and DictEncoder
Currently DictDecoder class and DictEncoder class uses std::vector
to store the tables mapping codeword to value and vice-versa. It is
hard to detect the memory usage by these tables when they becomes
very large, since this memory is not accounted by Impala's memory
mangement infrastructure.
This patch uses the memory tracker of HdfsScanner to track the memory used
by dictionary in DictDecoder class. Similary it uses memory tracker of
HdfsTableSink to track the memory used by dictionary in DictEncoder class.
Memory for the dictionary, stored as std::vector is still allocated
from std:allocator but the amount allocated is accounted by
introducing a counter which is incremented and decremented as the
memory is consumed and released by vector.
Testing
-------
Ran all the backend and end-end tests with no failures.
Change-Id: I02a3b54f6c107d19b62ad9e1c49df94175964299
Reviewed-on: http://gerrit.cloudera.org:8080/8034
Reviewed-by: Tim Armstrong <ta...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/302ec25b
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/302ec25b
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/302ec25b
Branch: refs/heads/master
Commit: 302ec25b2e76398f11bb5dccbdad84e06c750f22
Parents: d25607d
Author: Pranay <ps...@cloudera.com>
Authored: Mon Sep 11 14:38:49 2017 -0700
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Sat Jan 6 01:30:36 2018 +0000
----------------------------------------------------------------------
be/src/exec/hdfs-parquet-table-writer.cc | 5 +-
be/src/exec/parquet-column-readers.cc | 1 +
be/src/exec/parquet-column-readers.h | 2 +
be/src/util/dict-encoding.h | 138 ++++++++++++++++++++++++--
be/src/util/dict-test.cc | 51 ++++++++--
5 files changed, 173 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/302ec25b/be/src/exec/hdfs-parquet-table-writer.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-parquet-table-writer.cc b/be/src/exec/hdfs-parquet-table-writer.cc
index 8c9d92e..952c78e 100644
--- a/be/src/exec/hdfs-parquet-table-writer.cc
+++ b/be/src/exec/hdfs-parquet-table-writer.cc
@@ -175,7 +175,7 @@ class HdfsParquetTableWriter::BaseColumnWriter {
// be added.
void Close() {
if (compressor_.get() != nullptr) compressor_->Close();
- if (dict_encoder_base_ != nullptr) dict_encoder_base_->ClearIndices();
+ if (dict_encoder_base_ != nullptr) dict_encoder_base_->Close();
}
const ColumnType& type() const { return expr_eval_->root().type(); }
@@ -315,7 +315,8 @@ class HdfsParquetTableWriter::ColumnWriter :
current_encoding_ = Encoding::PLAIN_DICTIONARY;
next_page_encoding_ = Encoding::PLAIN_DICTIONARY;
dict_encoder_.reset(
- new DictEncoder<T>(parent_->per_file_mem_pool_.get(), plain_encoded_value_size_));
+ new DictEncoder<T>(parent_->per_file_mem_pool_.get(), plain_encoded_value_size_,
+ parent_->parent_->mem_tracker()));
dict_encoder_base_ = dict_encoder_.get();
page_stats_.reset(
new ColumnStats<T>(parent_->per_file_mem_pool_.get(), plain_encoded_value_size_));
http://git-wip-us.apache.org/repos/asf/impala/blob/302ec25b/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 02d2950..1c1cf39 100644
--- a/be/src/exec/parquet-column-readers.cc
+++ b/be/src/exec/parquet-column-readers.cc
@@ -236,6 +236,7 @@ class ScalarColumnReader : public BaseScalarColumnReader {
ScalarColumnReader(HdfsParquetScanner* parent, const SchemaNode& node,
const SlotDescriptor* slot_desc)
: BaseScalarColumnReader(parent, node, slot_desc),
+ dict_decoder_(parent->scan_node_->mem_tracker()),
dict_decoder_init_(false),
needs_conversion_(false) {
if (!MATERIALIZED) {
http://git-wip-us.apache.org/repos/asf/impala/blob/302ec25b/be/src/exec/parquet-column-readers.h
----------------------------------------------------------------------
diff --git a/be/src/exec/parquet-column-readers.h b/be/src/exec/parquet-column-readers.h
index 17b8c0f..3a8ad70 100644
--- a/be/src/exec/parquet-column-readers.h
+++ b/be/src/exec/parquet-column-readers.h
@@ -372,6 +372,8 @@ class BaseScalarColumnReader : public ParquetColumnReader {
data_page_pool_->FreeAll();
}
if (decompressor_ != nullptr) decompressor_->Close();
+ DictDecoderBase* dict_decoder = GetDictionaryDecoder();
+ if (dict_decoder != nullptr) dict_decoder->Close();
}
int64_t total_len() const { return metadata_->total_compressed_size; }
http://git-wip-us.apache.org/repos/asf/impala/blob/302ec25b/be/src/util/dict-encoding.h
----------------------------------------------------------------------
diff --git a/be/src/util/dict-encoding.h b/be/src/util/dict-encoding.h
index fa5e798..d916bab 100644
--- a/be/src/util/dict-encoding.h
+++ b/be/src/util/dict-encoding.h
@@ -52,6 +52,15 @@ class DictEncoderBase {
public:
virtual ~DictEncoderBase() {
DCHECK(buffered_indices_.empty());
+ ReleaseBytes();
+ DCHECK_EQ(dict_bytes_cnt_, 0);
+ }
+
+ /// This function will clear the buffered_indices and
+ /// decrement the bytes used by dictionary.
+ void Close() {
+ ClearIndices();
+ ReleaseBytes();
}
/// Writes out the encoded dictionary to buffer. buffer must be preallocated to
@@ -86,26 +95,77 @@ class DictEncoderBase {
int dict_encoded_size() { return dict_encoded_size_; }
+ void UsedbyTest() { used_by_test_ = true;}
+
protected:
- DictEncoderBase(MemPool* pool)
- : dict_encoded_size_(0), pool_(pool) {
- }
+ DictEncoderBase(MemPool* pool, MemTracker* mem_tracker) :
+ dict_encoded_size_(0),
+ pool_(pool),
+ dict_bytes_cnt_(0),
+ dict_mem_tracker_(mem_tracker) { }
/// Indices that have not yet be written out by WriteData().
std::vector<int> buffered_indices_;
+ /// Memtracker Consume is called every ENC_MEM_TRACK_CNT times.
+ /// Periodicity of calling Memtracker Consume.
+ const int ENC_MEM_TRACK_CNT = 8192;
+
+ /// Number of times ConsumeBytes() was called.
+ int num_call_track_{0};
+
/// The number of bytes needed to encode the dictionary.
- int dict_encoded_size_;
+ int dict_encoded_size_{0};
/// Pool to store StringValue data. Not owned.
- MemPool* pool_;
+ MemPool* pool_{nullptr};
+
+ /// This will account for bytes consumed by nodes_
+ int dict_bytes_cnt_{0};
+
+ /// This will account for bytes consumed, last_accounted by memtracker
+ int dict_bytes_cnt_memtrack_{0};
+
+ /// This will track the memory used by nodes_
+ MemTracker* dict_mem_tracker_{nullptr};
+
+ /// Function to decrement the byte counter and decrease the bytes usage
+ /// of the memory tracker.
+ void ReleaseBytes() {
+ if (dict_mem_tracker_ != nullptr) {
+ dict_mem_tracker_->Release(dict_bytes_cnt_memtrack_);
+ dict_bytes_cnt_ = 0;
+ dict_bytes_cnt_memtrack_ = 0;
+ num_call_track_ = 0;
+ }
+ }
+
+ /// Used by dict-test.cc to check the usage of bytes by dictionary
+ /// is properly accounted.
+ bool used_by_test_{false};
+
+ /// Function to increment the byte counter and increase the bytes usage
+ /// of the memory tracker.
+ void ConsumeBytes(int num_bytes) {
+ if (dict_mem_tracker_ != nullptr) {
+ dict_bytes_cnt_ += num_bytes;
+ // Calling Memtracker frequently may be expensive so update
+ // the memtracker every ENC_MEM_TRACK_CNT times.
+ if (num_call_track_ % ENC_MEM_TRACK_CNT == 0 || used_by_test_ == true) {
+ // TODO: TryConsume() can be called to check if memory limit has been exceeded.
+ dict_mem_tracker_->Consume(dict_bytes_cnt_ - dict_bytes_cnt_memtrack_);
+ dict_bytes_cnt_memtrack_ = dict_bytes_cnt_;
+ }
+ num_call_track_++;
+ }
+ }
};
template<typename T>
class DictEncoder : public DictEncoderBase {
public:
- DictEncoder(MemPool* pool, int encoded_value_size) :
- DictEncoderBase(pool), buckets_(HASH_TABLE_SIZE, Node::INVALID_INDEX),
+ DictEncoder(MemPool* pool, int encoded_value_size, MemTracker* mem_tracker) :
+ DictEncoderBase(pool, mem_tracker), buckets_(HASH_TABLE_SIZE, Node::INVALID_INDEX),
encoded_value_size_(encoded_value_size) { }
/// Encode value. Returns the number of bytes added to the dictionary page length
@@ -115,6 +175,13 @@ class DictEncoder : public DictEncoderBase {
/// written later.
int Put(const T& value);
+ /// This function returns the size in bytes of the dictionary vector.
+ /// It is used by dict-test.cc for validation of bytes consumed against
+ /// memory tracked.
+ int DictByteSize() {
+ return sizeof(Node) * nodes_.size();
+ }
+
virtual void WriteDict(uint8_t* buffer);
virtual int num_entries() const { return nodes_.size(); }
@@ -168,6 +235,9 @@ class DictEncoder : public DictEncoderBase {
/// by the caller and valid as long as this object is.
class DictDecoderBase {
public:
+ DictDecoderBase(MemTracker* tracker) :
+ dict_bytes_cnt_(0), dict_mem_tracker_(tracker) { }
+
/// The rle encoded indices into the dictionary. Returns an error status if the buffer
/// is too short or the bit_width metadata in the buffer is invalid.
Status SetData(uint8_t* buffer, int buffer_len) {
@@ -187,7 +257,10 @@ class DictDecoderBase {
return Status::OK();
}
- virtual ~DictDecoderBase() {}
+ virtual ~DictDecoderBase() {
+ ReleaseBytes();
+ DCHECK_EQ(dict_bytes_cnt_, 0);
+ }
virtual int num_entries() const = 0;
@@ -195,6 +268,11 @@ class DictDecoderBase {
/// The buffer must be large enough to receive the datatype for this dictionary.
virtual void GetValue(int index, void* buffer) = 0;
+ /// This function will decrement the bytes used by dictionary, MemTracker
+ void Close() {
+ ReleaseBytes();
+ }
+
protected:
/// Number of decoded values to buffer at a time. A multiple of 32 is chosen to allow
/// efficient reading in batches from data_decoder_. Increasing the batch size up to
@@ -212,13 +290,37 @@ class DictDecoderBase {
/// The index of the next decoded value to return.
int next_literal_idx_ = 0;
+
+ /// This will account for bytes consumed by dict_
+ int dict_bytes_cnt_{0};
+
+ /// This will track the memory used by dict_
+ MemTracker* dict_mem_tracker_{nullptr};
+
+ /// Function to decrement the byte counter and decrease the bytes usage
+ /// of the memory tracker.
+ void ReleaseBytes() {
+ if (dict_mem_tracker_ != nullptr) {
+ dict_mem_tracker_->Release(dict_bytes_cnt_);
+ dict_bytes_cnt_ = 0;
+ }
+ }
+
+ /// Function to increment the byte counter and increase the bytes usage
+ /// of the memory tracker.
+ void ConsumeBytes(int num_bytes) {
+ if (dict_mem_tracker_ != nullptr) {
+ dict_bytes_cnt_ += num_bytes;
+ dict_mem_tracker_->Consume(num_bytes);
+ }
+ }
};
template<typename T>
class DictDecoder : public DictDecoderBase {
public:
/// Construct empty dictionary.
- DictDecoder() {}
+ DictDecoder(MemTracker* tracker):DictDecoderBase(tracker) {}
/// Initialize the decoder with an input buffer containing the dictionary.
/// 'dict_len' is the byte length of dict_buffer.
@@ -245,7 +347,15 @@ class DictDecoder : public DictDecoderBase {
/// the string data is from the dictionary buffer passed into the c'tor.
bool GetNextValue(T* value) WARN_UNUSED_RESULT;
+ /// This function returns the size in bytes of the dictionary vector.
+ /// It is used by dict-test.cc for validation of bytes consumed against
+ /// memory tracked.
+ int DictByteSize() {
+ return sizeof(T) * dict_.size();
+ }
+
private:
+ /// List of decoded values stored in the dict_
std::vector<T> dict_;
/// Decoded values, buffered to allow caller to consume one-by-one. If in the middle of
@@ -293,8 +403,10 @@ inline uint32_t DictEncoder<StringValue>::Hash(const StringValue& value) const {
template<typename T>
inline int DictEncoder<T>::AddToTable(const T& value, NodeIndex* bucket) {
DCHECK_GT(encoded_value_size_, 0);
+ Node node(value, *bucket);
+ ConsumeBytes(sizeof(node));
// Prepend the new node to this bucket's chain.
- nodes_.push_back(Node(value, *bucket));
+ nodes_.push_back(node);
*bucket = nodes_.size() - 1;
dict_encoded_size_ += encoded_value_size_;
return encoded_value_size_;
@@ -306,8 +418,10 @@ inline int DictEncoder<StringValue>::AddToTable(const StringValue& value,
char* ptr_copy = reinterpret_cast<char*>(pool_->Allocate(value.len));
memcpy(ptr_copy, value.ptr, value.len);
StringValue sv(ptr_copy, value.len);
+ Node node(sv, *bucket);
+ ConsumeBytes(sizeof(node));
// Prepend the new node to this bucket's chain.
- nodes_.push_back(Node(sv, *bucket));
+ nodes_.push_back(node);
*bucket = nodes_.size() - 1;
int bytes_added = ParquetPlainEncoder::ByteSize(sv);
dict_encoded_size_ += bytes_added;
@@ -386,6 +500,7 @@ template<parquet::Type::type PARQUET_TYPE>
inline bool DictDecoder<T>::Reset(uint8_t* dict_buffer, int dict_len,
int fixed_len_size) {
dict_.clear();
+ ReleaseBytes();
uint8_t* end = dict_buffer + dict_len;
while (dict_buffer < end) {
T value;
@@ -395,6 +510,7 @@ inline bool DictDecoder<T>::Reset(uint8_t* dict_buffer, int dict_len,
dict_buffer += decoded_len;
dict_.push_back(value);
}
+ ConsumeBytes(sizeof(T) * dict_.size());
return true;
}
http://git-wip-us.apache.org/repos/asf/impala/blob/302ec25b/be/src/util/dict-test.cc
----------------------------------------------------------------------
diff --git a/be/src/util/dict-test.cc b/be/src/util/dict-test.cc
index 11043f0..cf438f6 100644
--- a/be/src/util/dict-test.cc
+++ b/be/src/util/dict-test.cc
@@ -36,10 +36,15 @@ void ValidateDict(const vector<InternalType>& values,
const vector<InternalType>& dict_values, int fixed_buffer_byte_size) {
set<InternalType> values_set(values.begin(), values.end());
+ int bytes_alloc = 0;
+ MemTracker track_encoder;
MemTracker tracker;
MemPool pool(&tracker);
- DictEncoder<InternalType> encoder(&pool, fixed_buffer_byte_size);
+ DictEncoder<InternalType> encoder(&pool, fixed_buffer_byte_size, &track_encoder);
+ encoder.UsedbyTest();
for (InternalType i: values) encoder.Put(i);
+ bytes_alloc = encoder.DictByteSize();
+ EXPECT_EQ(track_encoder.consumption(), bytes_alloc);
EXPECT_EQ(encoder.num_entries(), values_set.size());
uint8_t dict_buffer[encoder.dict_encoded_size()];
@@ -51,9 +56,12 @@ void ValidateDict(const vector<InternalType>& values,
EXPECT_GT(data_len, 0);
encoder.ClearIndices();
- DictDecoder<InternalType> decoder;
+ MemTracker decode_tracker;
+ DictDecoder<InternalType> decoder(&decode_tracker);
ASSERT_TRUE(decoder.template Reset<PARQUET_TYPE>(dict_buffer,
encoder.dict_encoded_size(),fixed_buffer_byte_size));
+ bytes_alloc = decoder.DictByteSize();
+ EXPECT_EQ(decode_tracker.consumption(), bytes_alloc);
// Test direct access to the dictionary via indexes
for (int i = 0; i < dict_values.size(); ++i) {
@@ -180,7 +188,8 @@ TEST(DictTest, TestInvalidStrings) {
// Test a dictionary with a string encoded with negative length. Initializing
// the decoder should fail.
- DictDecoder<StringValue> decoder;
+ MemTracker tracker;
+ DictDecoder<StringValue> decoder(&tracker);
ASSERT_FALSE(decoder.template Reset<parquet::Type::BYTE_ARRAY>(buffer, sizeof(buffer),
0));
}
@@ -193,7 +202,8 @@ TEST(DictTest, TestStringBufferOverrun) {
// Initializing the dictionary should fail, since the string would reference
// invalid memory.
- DictDecoder<StringValue> decoder;
+ MemTracker tracker;
+ DictDecoder<StringValue> decoder(&tracker);
ASSERT_FALSE(decoder.template Reset<parquet::Type::BYTE_ARRAY>(buffer, sizeof(buffer),
0));
}
@@ -202,9 +212,13 @@ TEST(DictTest, TestStringBufferOverrun) {
// decoder to a clean state, even if the input is not fully consumed. The RLE decoder
// has various state that needs to be reset.
TEST(DictTest, SetDataAfterPartialRead) {
+ int bytes_alloc = 0;
MemTracker tracker;
+ MemTracker track_encoder;
+ MemTracker track_decoder;
MemPool pool(&tracker);
- DictEncoder<int> encoder(&pool, sizeof(int));
+ DictEncoder<int> encoder(&pool, sizeof(int), &track_encoder);
+ encoder.UsedbyTest();
// Literal run followed by a repeated run.
vector<int> values{1, 2, 3, 4, 5, 6, 7, 8, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9};
@@ -214,10 +228,12 @@ TEST(DictTest, SetDataAfterPartialRead) {
encoder.WriteDict(dict_buffer.data());
vector<uint8_t> data_buffer(encoder.EstimatedDataEncodedSize() * 2);
int data_len = encoder.WriteData(data_buffer.data(), data_buffer.size());
+ bytes_alloc = encoder.DictByteSize();
+ EXPECT_EQ(track_encoder.consumption(), bytes_alloc);
ASSERT_GT(data_len, 0);
encoder.ClearIndices();
- DictDecoder<int> decoder;
+ DictDecoder<int> decoder(&track_decoder);
ASSERT_TRUE(decoder.template Reset<parquet::Type::INT32>(
dict_buffer.data(), dict_buffer.size(), sizeof(int)));
@@ -231,13 +247,19 @@ TEST(DictTest, SetDataAfterPartialRead) {
EXPECT_EQ(values[i], val) << num_to_decode << " " << i;
}
}
+ bytes_alloc = decoder.DictByteSize();
+ EXPECT_EQ(track_decoder.consumption(), bytes_alloc);
}
// Test handling of decode errors from out-of-range values.
TEST(DictTest, DecodeErrors) {
+ int bytes_alloc = 0;
MemTracker tracker;
+ MemTracker track_encoder;
+ MemTracker track_decoder;
MemPool pool(&tracker);
- DictEncoder<int> small_dict_encoder(&pool, sizeof(int));
+ DictEncoder<int> small_dict_encoder(&pool, sizeof(int), &track_encoder);
+ small_dict_encoder.UsedbyTest();
// Generate a dictionary with 9 values (requires 4 bits to encode).
vector<int> small_dict_values{1, 2, 3, 4, 5, 6, 7, 8, 9};
@@ -245,9 +267,11 @@ TEST(DictTest, DecodeErrors) {
vector<uint8_t> small_dict_buffer(small_dict_encoder.dict_encoded_size());
small_dict_encoder.WriteDict(small_dict_buffer.data());
+ bytes_alloc = small_dict_encoder.DictByteSize();
+ EXPECT_EQ(track_encoder.consumption(), bytes_alloc);
small_dict_encoder.ClearIndices();
- DictDecoder<int> small_dict_decoder;
+ DictDecoder<int> small_dict_decoder(&track_decoder);
ASSERT_TRUE(small_dict_decoder.template Reset<parquet::Type::INT32>(
small_dict_buffer.data(), small_dict_buffer.size(), sizeof(int)));
@@ -266,17 +290,20 @@ TEST(DictTest, DecodeErrors) {
for (TestCase& test_case: test_cases) {
// Encode the values. This will produce a dictionary with more distinct values than
// the small dictionary that we'll use to decode it.
- DictEncoder<int> large_dict_encoder(&pool, sizeof(int));
+ DictEncoder<int> large_dict_encoder(&pool, sizeof(int), &track_encoder);
+ large_dict_encoder.UsedbyTest();
// Initialize the dictionary with the values already in the small dictionary.
for (int val : small_dict_values) large_dict_encoder.Put(val);
- large_dict_encoder.ClearIndices();
+ large_dict_encoder.Close();
for (int val: test_case.second) large_dict_encoder.Put(val);
vector<uint8_t> data_buffer(large_dict_encoder.EstimatedDataEncodedSize() * 2);
int data_len = large_dict_encoder.WriteData(data_buffer.data(), data_buffer.size());
ASSERT_GT(data_len, 0);
- large_dict_encoder.ClearIndices();
+ bytes_alloc = large_dict_encoder.DictByteSize();
+ EXPECT_EQ(track_encoder.consumption(), bytes_alloc);
+ large_dict_encoder.Close();
ASSERT_OK(small_dict_decoder.SetData(data_buffer.data(), data_buffer.size()));
bool failed = false;
@@ -285,6 +312,8 @@ TEST(DictTest, DecodeErrors) {
failed = !small_dict_decoder.GetNextValue(&val);
if (failed) break;
}
+ bytes_alloc = small_dict_decoder.DictByteSize();
+ EXPECT_EQ(track_decoder.consumption(), bytes_alloc);
EXPECT_TRUE(failed) << "Should have detected out-of-range dict-encoded value in test "
<< test_case.first;
}
[3/3] impala git commit: IMPALA-6348: Redact only sensitive fields in
runtime profiles
Posted by mi...@apache.org.
IMPALA-6348: Redact only sensitive fields in runtime profiles
Without this patch, redaction is applied to every field in the
runtime profile. This approach has an undesired side effect when
Kerberos auth + email redaction is in place.
Since the redaction applies to every field, even principals
(from Connected/Delegated User fields) are redacted, as the Kerberos
principal format generally pattern matches with an email redactor
template.
This is particularly problematic for monitoring tools that consume
runtime profiles and use these fields to group the queries by user.
This patch fixes the problem by redacting only the following sensitive
fields.
- Query Statement
- Error logs (since they can contain column references etc.)
- Query Status
- Query Plan
Other fields in the runtime profile are left unredacted.
Change-Id: Iae3b6726009bf458a7ec73131e5d659b12ab73cf
Reviewed-on: http://gerrit.cloudera.org:8080/8934
Reviewed-by: Bharath Vissapragada <bh...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/6a87eb20
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/6a87eb20
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/6a87eb20
Branch: refs/heads/master
Commit: 6a87eb20a5d55b0a2f6f9102375ff8da4b98ccba
Parents: ce65b43
Author: Bharath Vissapragada <bh...@cloudera.com>
Authored: Wed Jan 3 15:31:32 2018 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Sat Jan 6 22:54:17 2018 +0000
----------------------------------------------------------------------
be/src/service/client-request-state.cc | 7 ++++---
be/src/service/client-request-state.h | 10 ++++++++-
be/src/service/impala-server.cc | 4 ++--
be/src/util/runtime-profile.cc | 9 +++++---
be/src/util/runtime-profile.h | 7 ++++++-
tests/custom_cluster/test_redaction.py | 32 ++++++++++++++++++++++++++---
6 files changed, 56 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/6a87eb20/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 5e95bb2..87ac8ff 100644
--- a/be/src/service/client-request-state.cc
+++ b/be/src/service/client-request-state.cc
@@ -124,7 +124,8 @@ ClientRequestState::ClientRequestState(
summary_profile_->AddInfoString("Network Address",
lexical_cast<string>(session_->network_address));
summary_profile_->AddInfoString("Default Db", default_db());
- summary_profile_->AddInfoString("Sql Statement", query_ctx_.client_request.stmt);
+ summary_profile_->AddInfoStringRedacted(
+ "Sql Statement", query_ctx_.client_request.stmt);
summary_profile_->AddInfoString("Coordinator",
TNetworkAddressToString(exec_env->backend_address()));
}
@@ -419,7 +420,7 @@ Status ClientRequestState::ExecQueryOrDmlRequest(
plan_ss << "\n----------------\n"
<< query_exec_request.query_plan
<< "----------------";
- summary_profile_->AddInfoString("Plan", plan_ss.str());
+ summary_profile_->AddInfoStringRedacted("Plan", plan_ss.str());
}
// Add info strings consumed by CM: Estimated mem and tables missing stats.
if (query_exec_request.__isset.per_host_mem_estimate) {
@@ -741,7 +742,7 @@ Status ClientRequestState::UpdateQueryStatus(const Status& status) {
if (!status.ok() && query_status_.ok()) {
UpdateQueryState(beeswax::QueryState::EXCEPTION);
query_status_ = status;
- summary_profile_->AddInfoString("Query Status", query_status_.GetDetail());
+ summary_profile_->AddInfoStringRedacted("Query Status", query_status_.GetDetail());
}
return status;
http://git-wip-us.apache.org/repos/asf/impala/blob/6a87eb20/be/src/service/client-request-state.h
----------------------------------------------------------------------
diff --git a/be/src/service/client-request-state.h b/be/src/service/client-request-state.h
index 87f870a..d93512e 100644
--- a/be/src/service/client-request-state.h
+++ b/be/src/service/client-request-state.h
@@ -296,10 +296,18 @@ class ClientRequestState {
/// * server_profile_ tracks time spent inside the ImpalaServer,
/// but not inside fragment execution, i.e. the time taken to
/// register and set-up the query and for rows to be fetched.
- //
+ ///
/// There's a fourth profile which is not built here (but is a
/// child of profile_); the execution profile which tracks the
/// actual fragment execution.
+ ///
+ /// Redaction: Only the following info strings in the profile are redacted as they
+ /// are expected to contain sensitive information like schema/column references etc.
+ /// Other fields are left unredacted.
+ /// - Query Statement
+ /// - Query Plan
+ /// - Query Status
+ /// - Error logs
RuntimeProfile* const profile_;
RuntimeProfile* const server_profile_;
RuntimeProfile* const summary_profile_;
http://git-wip-us.apache.org/repos/asf/impala/blob/6a87eb20/be/src/service/impala-server.cc
----------------------------------------------------------------------
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index 49009b6..5b895ee 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -1030,8 +1030,8 @@ Status ImpalaServer::UnregisterQuery(const TUniqueId& query_id, bool check_infli
TExecSummary t_exec_summary;
request_state->coord()->GetTExecSummary(&t_exec_summary);
string exec_summary = PrintExecSummary(t_exec_summary);
- request_state->summary_profile()->AddInfoString("ExecSummary", exec_summary);
- request_state->summary_profile()->AddInfoString("Errors",
+ request_state->summary_profile()->AddInfoStringRedacted("ExecSummary", exec_summary);
+ request_state->summary_profile()->AddInfoStringRedacted("Errors",
request_state->coord()->GetErrorLog());
const PerBackendExecParams& per_backend_params =
http://git-wip-us.apache.org/repos/asf/impala/blob/6a87eb20/be/src/util/runtime-profile.cc
----------------------------------------------------------------------
diff --git a/be/src/util/runtime-profile.cc b/be/src/util/runtime-profile.cc
index abb691d..a05e55c 100644
--- a/be/src/util/runtime-profile.cc
+++ b/be/src/util/runtime-profile.cc
@@ -466,14 +466,17 @@ void RuntimeProfile::AddInfoString(const string& key, const string& value) {
return AddInfoStringInternal(key, value, false);
}
+void RuntimeProfile::AddInfoStringRedacted(const string& key, const string& value) {
+ return AddInfoStringInternal(key, value, false, true);
+}
+
void RuntimeProfile::AppendInfoString(const string& key, const string& value) {
return AddInfoStringInternal(key, value, true);
}
void RuntimeProfile::AddInfoStringInternal(
- const string& key, const string& value, bool append) {
- // Values may contain sensitive data, such as a query.
- const string& info = RedactCopy(value);
+ const string& key, const string& value, bool append, bool redact) {
+ const string& info = redact ? RedactCopy(value): value;
lock_guard<SpinLock> l(info_strings_lock_);
InfoStrings::iterator it = info_strings_.find(key);
if (it == info_strings_.end()) {
http://git-wip-us.apache.org/repos/asf/impala/blob/6a87eb20/be/src/util/runtime-profile.h
----------------------------------------------------------------------
diff --git a/be/src/util/runtime-profile.h b/be/src/util/runtime-profile.h
index 8348161..320edf3 100644
--- a/be/src/util/runtime-profile.h
+++ b/be/src/util/runtime-profile.h
@@ -209,6 +209,10 @@ class RuntimeProfile { // NOLINT: This struct is not packed, but there are not s
/// the value will be updated.
void AddInfoString(const std::string& key, const std::string& value);
+ /// Same as AddInfoString(), except that this method applies the redaction
+ /// rules on 'value' before adding it to the runtime profile.
+ void AddInfoStringRedacted(const std::string& key, const std::string& value);
+
/// Adds a string to the runtime profile. If a value already exists for 'key',
/// 'value' will be appended to the previous value, with ", " separating them.
void AppendInfoString(const std::string& key, const std::string& value);
@@ -473,8 +477,9 @@ class RuntimeProfile { // NOLINT: This struct is not packed, but there are not s
/// Implementation of AddInfoString() and AppendInfoString(). If 'append' is false,
/// implements AddInfoString(), otherwise implements AppendInfoString().
+ /// Redaction rules are applied on the info string if 'redact' is true.
void AddInfoStringInternal(
- const std::string& key, const std::string& value, bool append);
+ const std::string& key, const std::string& value, bool append, bool redact = false);
/// Name of the counter maintaining the total time.
static const std::string TOTAL_TIME_COUNTER_NAME;
http://git-wip-us.apache.org/repos/asf/impala/blob/6a87eb20/tests/custom_cluster/test_redaction.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_redaction.py b/tests/custom_cluster/test_redaction.py
index 46fef9b..729c2ca 100644
--- a/tests/custom_cluster/test_redaction.py
+++ b/tests/custom_cluster/test_redaction.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+import getpass
import logging
import os
import pytest
@@ -160,6 +161,14 @@ class TestRedaction(CustomClusterTestSuite, unittest.TestCase):
assert results, "Web page %s should contain '%s' but does not" \
% (url, search)
+ def assert_query_profile_contains(self, query_id, search):
+ ''' Asserts that the query profile for 'query_id' contains 'search' string'''
+ impala_service = self.create_impala_service()
+ url = 'query_profile?query_id=%s' % query_id
+ results = self.grep_file(impala_service.open_debug_webpage(url), search)
+ assert results, "Query profile %s should contain '%s' but does not" \
+ % (url, search)
+
def assert_file_in_dir_contains(self, dir, search):
'''Asserts that at least one file in the 'dir' contains the 'search' term.'''
results = self.grep_dir(dir,search)
@@ -271,6 +280,7 @@ class TestRedaction(CustomClusterTestSuite, unittest.TestCase):
'''Check that redaction rules prevent 'sensitive' data from leaking into the
logs and web ui.
'''
+ current_user = getpass.getuser()
self.start_cluster_using_rules(r"""
{
"version": 1,
@@ -285,17 +295,26 @@ class TestRedaction(CustomClusterTestSuite, unittest.TestCase):
"description": "Don't show credit cards numbers",
"search": "\\d{4}-\\d{4}-\\d{4}-\\d{4}",
"replace": "*credit card*"
+ },
+ {
+ "description": "Don't show current username in queries",
+ "search": "%s",
+ "replace": "redacted user"
}
]
- }""")
+ }""" % current_user)
email = 'FOO@bar.com'
# GROUP BY an expr containing the email so the expr will also be shown in the exec
# node summary, ie HASH(string_col = ...).
- self.execute_query_expect_success(self.client,
- "SELECT string_col = '%s', COUNT(*) FROM functional.alltypes GROUP BY 1" % email)
+ query_template =\
+ "SELECT string_col = '%s', COUNT(*) FROM functional.alltypes GROUP BY 1"
+ self.execute_query_expect_success(self.client, query_template % email)
+ user_profile_pattern = "User: %s" % current_user
# The email should be replaced with '*email*'.
self.assert_web_ui_redaction(self.find_last_query_id(), email, "*email*")
+ # User field should not be redacted from the query profile.
+ self.assert_query_profile_contains(self.find_last_query_id(), user_profile_pattern)
# Wait for the logs to be written.
sleep(5)
self.assert_log_redaction(email, "*email*")
@@ -306,10 +325,17 @@ class TestRedaction(CustomClusterTestSuite, unittest.TestCase):
# This assertion below relies on the fact that there is a syntax error be near
# the credit card number so the number would have appeared in the message.
self.assert_web_ui_redaction(self.find_last_query_id(), credit_card, "*credit card*")
+ # User field should not be redacted from the query profile.
+ self.assert_query_profile_contains(self.find_last_query_id(), user_profile_pattern)
sleep(5)
# Apparently an invalid query doesn't generate an audit log entry.
self.assert_log_redaction(credit_card, "*credit card*", expect_audit=False)
+ # Assert that the username in the query stmt is redacted but not from the user fields.
+ self.execute_query_expect_success(self.client, query_template % current_user)
+ self.assert_query_profile_contains(self.find_last_query_id(), user_profile_pattern)
+ self.assert_query_profile_contains(self.find_last_query_id(), "redacted user")
+
# Since all the tests passed, the log dir shouldn't be of interest and can be
# deleted.
shutil.rmtree(self.tmp_dir)
[2/3] impala git commit: IMPALA-2248: Make idle_session_timeout a
query option
Posted by mi...@apache.org.
IMPALA-2248: Make idle_session_timeout a query option
This commit makes idle_session_timeout a query option.
idle_session_timeout currently can be set as a command line
option, which will be the default timeout for sessions.
HS2 sessions can override it with a smaller value by setting
it in the configuration overlay of HS2 OpenSession().
However, we can't override idle_session_timeout for JDBC/ODBC
connections, because we cannot put this in the connection string.
This commit is a workaround for this problem, it allows JDBC/ODBC
connections to set the session timeout as a query option
with the SET statement.
After this commit, the session timeout can be overridden to
any value, i.e. the command line flag idle_session_timeout
doesn't limit this option anymore.
I created an automated test case in JdbcTest.java based on
test_hs2.py::test_concurrent_session_mixed_idle_timeout. I also
extended the test_session_expiration and test_set_and_unset
test suites.
Change-Id: I32e2775f80da387b0df4195fe2c5435b3f8e585e
Reviewed-on: http://gerrit.cloudera.org:8080/8490
Reviewed-by: Tim Armstrong <ta...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/ce65b43d
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/ce65b43d
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/ce65b43d
Branch: refs/heads/master
Commit: ce65b43d47d16840998eb5fab5333695050ac436
Parents: 302ec25
Author: Zoltan Borok-Nagy <bo...@cloudera.com>
Authored: Tue Nov 7 14:51:36 2017 +0100
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Sat Jan 6 01:47:47 2018 +0000
----------------------------------------------------------------------
be/src/service/client-request-state.cc | 18 +++-
be/src/service/impala-hs2-server.cc | 24 ++---
be/src/service/impala-server.cc | 47 +++++---
be/src/service/impala-server.h | 30 ++++--
be/src/service/query-options.cc | 23 +++-
be/src/service/query-options.h | 3 +-
common/thrift/ImpalaInternalService.thrift | 5 +
common/thrift/ImpalaService.thrift | 4 +
.../org/apache/impala/service/JdbcTest.java | 107 +++++++++++++++++--
.../java/org/apache/impala/util/Metrics.java | 82 ++++++++++++++
tests/custom_cluster/test_session_expiration.py | 49 ++++++++-
tests/custom_cluster/test_set_and_unset.py | 47 ++++++++
tests/hs2/hs2_test_suite.py | 10 +-
tests/hs2/test_hs2.py | 2 +-
14 files changed, 391 insertions(+), 60 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/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 76e2557..5e95bb2 100644
--- a/be/src/service/client-request-state.cc
+++ b/be/src/service/client-request-state.cc
@@ -17,6 +17,7 @@
#include "service/client-request-state.h"
+#include <boost/algorithm/string/predicate.hpp>
#include <limits>
#include <gutil/strings/substitute.h>
@@ -42,6 +43,7 @@
#include "common/names.h"
+using boost::algorithm::iequals;
using boost::algorithm::join;
using namespace apache::hive::service::cli::thrift;
using namespace apache::thrift;
@@ -200,12 +202,18 @@ Status ClientRequestState::Exec(TExecRequest* exec_request) {
if (exec_request_.set_query_option_request.__isset.key) {
// "SET key=value" updates the session query options.
DCHECK(exec_request_.set_query_option_request.__isset.value);
- RETURN_IF_ERROR(SetQueryOption(
- exec_request_.set_query_option_request.key,
- exec_request_.set_query_option_request.value,
- &session_->set_query_options,
- &session_->set_query_options_mask));
+ const auto& key = exec_request_.set_query_option_request.key;
+ const auto& value = exec_request_.set_query_option_request.value;
+ RETURN_IF_ERROR(SetQueryOption(key, value, &session_->set_query_options,
+ &session_->set_query_options_mask));
SetResultSet({}, {}, {});
+ if (iequals(key, "idle_session_timeout")) {
+ // IMPALA-2248: Session timeout is set as a query option
+ session_->last_accessed_ms = UnixMillis(); // do not expire session immediately
+ session_->UpdateTimeout();
+ VLOG_QUERY << "ClientRequestState::Exec() SET: idle_session_timeout="
+ << PrettyPrinter::Print(session_->session_timeout, TUnit::TIME_S);
+ }
} else {
// "SET" or "SET ALL"
bool is_set_all = exec_request_.set_query_option_request.__isset.is_set_all &&
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/be/src/service/impala-hs2-server.cc
----------------------------------------------------------------------
diff --git a/be/src/service/impala-hs2-server.cc b/be/src/service/impala-hs2-server.cc
index 20608b7..2f2ec40 100644
--- a/be/src/service/impala-hs2-server.cc
+++ b/be/src/service/impala-hs2-server.cc
@@ -296,7 +296,7 @@ void ImpalaServer::OpenSession(TOpenSessionResp& return_val,
// query options.
// TODO: put secret in session state map and check it
// TODO: Fix duplication of code between here and ConnectionStart().
- shared_ptr<SessionState> state = make_shared<SessionState>();
+ shared_ptr<SessionState> state = make_shared<SessionState>(this);
state->closed = false;
state->start_time_ms = UnixMillis();
state->session_type = TSessionType::HIVESERVER2;
@@ -316,8 +316,8 @@ void ImpalaServer::OpenSession(TOpenSessionResp& return_val,
// Process the supplied configuration map.
state->database = "default";
- state->session_timeout = FLAGS_idle_session_timeout;
state->server_default_query_options = &default_query_options_;
+ state->session_timeout = FLAGS_idle_session_timeout;
if (request.__isset.configuration) {
typedef map<string, string> ConfigurationMap;
for (const ConfigurationMap::value_type& v: request.configuration) {
@@ -330,22 +330,16 @@ void ImpalaServer::OpenSession(TOpenSessionResp& return_val,
HS2_RETURN_IF_ERROR(return_val, status, SQLSTATE_GENERAL_ERROR);
} else if (iequals(v.first, "use:database")) {
state->database = v.second;
- } else if (iequals(v.first, "idle_session_timeout")) {
- int32_t requested_timeout = atoi(v.second.c_str());
- if (requested_timeout > 0) {
- if (FLAGS_idle_session_timeout > 0) {
- state->session_timeout = min(FLAGS_idle_session_timeout, requested_timeout);
- } else {
- state->session_timeout = requested_timeout;
- }
- }
- VLOG_QUERY << "OpenSession(): idle_session_timeout="
- << PrettyPrinter::Print(state->session_timeout, TUnit::TIME_S);
} else {
// Normal configuration key. Use it to set session default query options.
// Ignore failure (failures will be logged in SetQueryOption()).
- discard_result(SetQueryOption(v.first, v.second, &state->set_query_options,
- &state->set_query_options_mask));
+ Status status = SetQueryOption(v.first, v.second, &state->set_query_options,
+ &state->set_query_options_mask);
+ if (status.ok() && iequals(v.first, "idle_session_timeout")) {
+ state->session_timeout = state->set_query_options.idle_session_timeout;
+ VLOG_QUERY << "OpenSession(): idle_session_timeout="
+ << PrettyPrinter::Print(state->session_timeout, TUnit::TIME_S);
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/be/src/service/impala-server.cc
----------------------------------------------------------------------
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index 2222802..49009b6 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -188,7 +188,8 @@ DEFINE_string(ssl_minimum_version, "tlsv1", SSL_MIN_VERSION_HELP.c_str());
DEFINE_int32(idle_session_timeout, 0, "The time, in seconds, that a session may be idle"
" for before it is closed (and all running queries cancelled) by Impala. If 0, idle"
- " sessions are never expired.");
+ " sessions are never expired. It can be overridden by the query option"
+ " 'idle_session_timeout' for specific sessions");
DEFINE_int32(idle_query_timeout, 0, "The time, in seconds, that a query may be idle for"
" (i.e. no processing work is done and no updates are received from the client) "
"before it is cancelled. If 0, idle queries are never expired. The query option "
@@ -1118,14 +1119,7 @@ Status ImpalaServer::CloseSessionInternal(const TUniqueId& session_id,
discard_result(UnregisterQuery(query_id, false, &status));
}
// Reconfigure the poll period of session_timeout_thread_ if necessary.
- int32_t session_timeout = session_state->session_timeout;
- if (session_timeout > 0) {
- lock_guard<mutex> l(session_timeout_lock_);
- multiset<int32_t>::const_iterator itr = session_timeout_set_.find(session_timeout);
- DCHECK(itr != session_timeout_set_.end());
- session_timeout_set_.erase(itr);
- session_timeout_cv_.NotifyOne();
- }
+ UnregisterSessionTimeout(session_state->session_timeout);
return Status::OK();
}
@@ -1203,6 +1197,9 @@ void ImpalaServer::TransmitData(
}
void ImpalaServer::InitializeConfigVariables() {
+ // Set idle_session_timeout here to let the SET command return the value of
+ // the command line option FLAGS_idle_session_timeout
+ default_query_options_.__set_idle_session_timeout(FLAGS_idle_session_timeout);
QueryOptionsMask set_query_options; // unused
Status status = ParseQueryOptions(FLAGS_default_query_options,
&default_query_options_, &set_query_options);
@@ -1227,6 +1224,20 @@ void ImpalaServer::InitializeConfigVariables() {
}
}
+void ImpalaServer::SessionState::UpdateTimeout() {
+ DCHECK(impala_server != nullptr);
+ int32_t old_timeout = session_timeout;
+ if (set_query_options.__isset.idle_session_timeout) {
+ session_timeout = set_query_options.idle_session_timeout;
+ } else {
+ session_timeout = server_default_query_options->idle_session_timeout;
+ }
+ if (old_timeout != session_timeout) {
+ impala_server->UnregisterSessionTimeout(old_timeout);
+ impala_server->RegisterSessionTimeout(session_timeout);
+ }
+}
+
void ImpalaServer::AddOptionLevelToConfig(ConfigVariable* config,
const string& option_key) const {
const auto query_option_level = query_option_levels_.find(option_key);
@@ -1742,8 +1753,7 @@ void ImpalaServer::ConnectionStart(
// Beeswax only allows for one session per connection, so we can share the session ID
// with the connection ID
const TUniqueId& session_id = connection_context.connection_id;
- shared_ptr<SessionState> session_state;
- session_state.reset(new SessionState);
+ shared_ptr<SessionState> session_state = make_shared<SessionState>(this);
session_state->closed = false;
session_state->start_time_ms = UnixMillis();
session_state->last_accessed_ms = UnixMillis();
@@ -1807,11 +1817,22 @@ void ImpalaServer::ConnectionEnd(
void ImpalaServer::RegisterSessionTimeout(int32_t session_timeout) {
if (session_timeout <= 0) return;
- lock_guard<mutex> l(session_timeout_lock_);
- session_timeout_set_.insert(session_timeout);
+ {
+ lock_guard<mutex> l(session_timeout_lock_);
+ session_timeout_set_.insert(session_timeout);
+ }
session_timeout_cv_.NotifyOne();
}
+void ImpalaServer::UnregisterSessionTimeout(int32_t session_timeout) {
+ if (session_timeout > 0) {
+ lock_guard<mutex> l(session_timeout_lock_);
+ auto itr = session_timeout_set_.find(session_timeout);
+ DCHECK(itr != session_timeout_set_.end());
+ session_timeout_set_.erase(itr);
+ }
+}
+
[[noreturn]] void ImpalaServer::ExpireSessions() {
while (true) {
{
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/be/src/service/impala-server.h
----------------------------------------------------------------------
diff --git a/be/src/service/impala-server.h b/be/src/service/impala-server.h
index 9e6c830..868ae54 100644
--- a/be/src/service/impala-server.h
+++ b/be/src/service/impala-server.h
@@ -350,11 +350,15 @@ class ImpalaServer : public ImpalaServiceIf,
/// run as children of Beeswax sessions) get results back in the expected format -
/// child queries inherit the HS2 version from their parents, and a Beeswax session
/// will never update the HS2 version from the default.
- SessionState() : closed(false), expired(false),
- hs2_version(apache::hive::service::cli::thrift::
+ SessionState(ImpalaServer* impala_server) : impala_server(impala_server),
+ closed(false), expired(false), hs2_version(apache::hive::service::cli::thrift::
TProtocolVersion::HIVE_CLI_SERVICE_PROTOCOL_V1), total_queries(0), ref_count(0) {
+ DCHECK(this->impala_server != nullptr);
}
+ /// Pointer to the Impala server of this session
+ ImpalaServer* impala_server;
+
TSessionType::type session_type;
/// Time the session was created, in ms since epoch (UTC).
@@ -421,9 +425,14 @@ class ImpalaServer : public ImpalaServiceIf,
uint32_t ref_count;
/// Per-session idle timeout in seconds. Default value is FLAGS_idle_session_timeout.
- /// It can be overridden with a smaller value via the option "idle_session_timeout"
- /// when opening a HS2 session.
- int32_t session_timeout;
+ /// It can be overridden with the query option "idle_session_timeout" when opening a
+ /// HS2 session, or using the SET command.
+ int32_t session_timeout = 0;
+
+ /// Updates the session timeout based on the query option idle_session_timeout.
+ /// It registers/unregisters the session timeout to the Impala server.
+ /// The lock must be owned by the caller of this function.
+ void UpdateTimeout();
/// Builds a Thrift representation of this SessionState for serialisation to
/// the frontend.
@@ -437,6 +446,7 @@ class ImpalaServer : public ImpalaServiceIf,
private:
friend class ChildQuery;
friend class ImpalaHttpHandler;
+ friend class SessionState;
boost::scoped_ptr<ImpalaHttpHandler> http_handler_;
@@ -766,11 +776,13 @@ class ImpalaServer : public ImpalaServiceIf,
const QueryOptionsMask& override_options_mask);
/// Register timeout value upon opening a new session. This will wake up
- /// session_timeout_thread_ to update its poll period.
+ /// session_timeout_thread_.
void RegisterSessionTimeout(int32_t timeout);
- /// To be run in a thread which wakes up every x / 2 seconds in which x is the minimum
- /// non-zero idle session timeout value of all sessions. This function checks all
+ /// Unregister timeout value.
+ void UnregisterSessionTimeout(int32_t timeout);
+
+ /// To be run in a thread which wakes up every second. This function checks all
/// sessions for their last-idle times. Those that have been idle for longer than
/// their configured timeout values are 'expired': they will no longer accept queries
/// and any running queries associated with those sessions are unregistered.
@@ -831,7 +843,7 @@ class ImpalaServer : public ImpalaServiceIf,
boost::mutex session_timeout_lock_;
/// session_timeout_thread_ relies on the following conditional variable to wake up
- /// on every poll period expiration or when the poll period changes.
+ /// when there are sessions that have a timeout.
ConditionVariable session_timeout_cv_;
/// map from query id to exec state; ClientRequestState is owned by us and referenced
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/be/src/service/query-options.cc
----------------------------------------------------------------------
diff --git a/be/src/service/query-options.cc b/be/src/service/query-options.cc
index 49f1698..e8e8c7b 100644
--- a/be/src/service/query-options.cc
+++ b/be/src/service/query-options.cc
@@ -40,6 +40,8 @@ using beeswax::TQueryOptionLevel;
using namespace impala;
using namespace strings;
+DECLARE_int32(idle_session_timeout);
+
// Utility method to wrap ParseUtil::ParseMemSpec() by returning a Status instead of an
// int.
static Status ParseMemValue(const string& value, const string& key, int64_t* result) {
@@ -94,8 +96,15 @@ static void ResetQueryOption(const int option, TQueryOptions* query_options) {
}
}
+static TQueryOptions DefaultQueryOptions() {
+ TQueryOptions defaults;
+ // default value of idle_session_timeout is set by a command line flag.
+ defaults.__set_idle_session_timeout(FLAGS_idle_session_timeout);
+ return defaults;
+}
+
string impala::DebugQueryOptions(const TQueryOptions& query_options) {
- const static TQueryOptions defaults;
+ const static TQueryOptions defaults = DefaultQueryOptions();
int i = 0;
stringstream ss;
#define QUERY_OPT_FN(NAME, ENUM, LEVEL)\
@@ -569,6 +578,18 @@ Status impala::SetQueryOption(const string& key, const string& value,
query_options->__set_max_row_size(max_row_size_bytes);
break;
}
+ case TImpalaQueryOptions::IDLE_SESSION_TIMEOUT: {
+ StringParser::ParseResult result;
+ const int32_t requested_timeout =
+ StringParser::StringToInt<int32_t>(value.c_str(), value.length(), &result);
+ if (result != StringParser::PARSE_SUCCESS || requested_timeout < 0) {
+ return Status(
+ Substitute("Invalid idle session timeout: '$0'. "
+ "Only positive numbers are allowed.", value));
+ }
+ query_options->__set_idle_session_timeout(requested_timeout);
+ break;
+ }
default:
// We hit this DCHECK(false) if we forgot to add the corresponding entry here
// when we add a new query option.
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/be/src/service/query-options.h
----------------------------------------------------------------------
diff --git a/be/src/service/query-options.h b/be/src/service/query-options.h
index 64e0647..b7065f4 100644
--- a/be/src/service/query-options.h
+++ b/be/src/service/query-options.h
@@ -41,7 +41,7 @@ typedef std::unordered_map<string, beeswax::TQueryOptionLevel::type>
// the DCHECK.
#define QUERY_OPTS_TABLE\
DCHECK_EQ(_TImpalaQueryOptions_VALUES_TO_NAMES.size(),\
- TImpalaQueryOptions::MAX_ROW_SIZE + 1);\
+ TImpalaQueryOptions::IDLE_SESSION_TIMEOUT + 1);\
QUERY_OPT_FN(abort_on_default_limit_exceeded, ABORT_ON_DEFAULT_LIMIT_EXCEEDED,\
TQueryOptionLevel::DEPRECATED)\
QUERY_OPT_FN(abort_on_error, ABORT_ON_ERROR, TQueryOptionLevel::REGULAR)\
@@ -129,6 +129,7 @@ typedef std::unordered_map<string, beeswax::TQueryOptionLevel::type>
QUERY_OPT_FN(min_spillable_buffer_size, MIN_SPILLABLE_BUFFER_SIZE,\
TQueryOptionLevel::ADVANCED)\
QUERY_OPT_FN(max_row_size, MAX_ROW_SIZE, TQueryOptionLevel::REGULAR)\
+ QUERY_OPT_FN(idle_session_timeout, IDLE_SESSION_TIMEOUT, TQueryOptionLevel::REGULAR)\
;
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/common/thrift/ImpalaInternalService.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/ImpalaInternalService.thrift b/common/thrift/ImpalaInternalService.thrift
index 1fd9c25..a584ce1 100644
--- a/common/thrift/ImpalaInternalService.thrift
+++ b/common/thrift/ImpalaInternalService.thrift
@@ -287,6 +287,11 @@ struct TQueryOptions {
// sophisticated strategies - e.g. reserving a small number of buffers large enough to
// fit maximum-sized rows.
60: optional i64 max_row_size = 524288;
+
+ // The time, in seconds, that a session may be idle for before it is closed (and all
+ // running queries cancelled) by Impala. If 0, idle sessions never expire.
+ // The default session timeout is set by the command line flag of the same name.
+ 61: optional i32 idle_session_timeout;
}
// Impala currently has two types of sessions: Beeswax and HiveServer2
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/common/thrift/ImpalaService.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/ImpalaService.thrift b/common/thrift/ImpalaService.thrift
index 061da00..53f6e81 100644
--- a/common/thrift/ImpalaService.thrift
+++ b/common/thrift/ImpalaService.thrift
@@ -288,6 +288,10 @@ enum TImpalaQueryOptions {
// The maximum row size that memory is reserved for, in bytes.
MAX_ROW_SIZE,
+
+ // The time, in seconds, that a session may be idle for before it is closed (and all
+ // running queries cancelled) by Impala. If 0, idle sessions never expire.
+ IDLE_SESSION_TIMEOUT,
}
// The summary of a DML statement.
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/fe/src/test/java/org/apache/impala/service/JdbcTest.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/service/JdbcTest.java b/fe/src/test/java/org/apache/impala/service/JdbcTest.java
index 8ee8fba..1a907f8 100644
--- a/fe/src/test/java/org/apache/impala/service/JdbcTest.java
+++ b/fe/src/test/java/org/apache/impala/service/JdbcTest.java
@@ -30,6 +30,8 @@ import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -38,11 +40,12 @@ import org.junit.After;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
-
import org.apache.impala.analysis.CreateTableStmt;
import org.apache.impala.analysis.SqlParser;
import org.apache.impala.analysis.SqlScanner;
import org.apache.impala.testutil.ImpalaJdbcClient;
+import org.apache.impala.util.Metrics;
+
import com.google.common.collect.Lists;
/**
@@ -60,13 +63,7 @@ public class JdbcTest {
@BeforeClass
public static void setUp() throws Exception {
- ImpalaJdbcClient client = ImpalaJdbcClient.createClientUsingHiveJdbcDriver();
- client.connect();
- con_ = client.getConnection();
- assertNotNull("Connection is null", con_);
- assertFalse("Connection should not be closed", con_.isClosed());
- Statement stmt = con_.createStatement();
- assertNotNull("Statement is null", stmt);
+ con_ = createConnection();
}
@AfterClass
@@ -85,6 +82,19 @@ public class JdbcTest {
expectedException);
}
+ private static Connection createConnection() throws Exception {
+ ImpalaJdbcClient client = ImpalaJdbcClient.createClientUsingHiveJdbcDriver();
+ client.connect();
+ Connection connection = client.getConnection();
+
+ assertNotNull("Connection is null", connection);
+ assertFalse("Connection should not be closed", connection.isClosed());
+ Statement stmt = connection.createStatement();
+ assertNotNull("Statement is null", stmt);
+
+ return connection;
+ }
+
protected void addTestTable(String createTableSql) throws Exception {
// Parse the stmt to extract the table name. We do this first to ensure
// that we do not execute arbitrary SQL here and pollute the test setup.
@@ -576,4 +586,85 @@ public class JdbcTest {
rs.close();
}
}
+
+ @Test
+ public void testConcurrentSessionMixedIdleTimeout() throws Exception {
+ // Test for concurrent idle sessions' expiration with mixed timeout durations.
+ Metrics metrics = new Metrics();
+
+ List<Integer> timeoutPeriods = Arrays.asList(0, 3, 15);
+ List<Connection> connections = new ArrayList<>();
+ List<Long> lastTimeSessionActive = new ArrayList<>();
+
+ for (int timeout : timeoutPeriods) {
+ connections.add(createConnection());
+ }
+
+ Long numOpenSessions = (Long)metrics.getMetric(
+ "impala-server.num-open-hiveserver2-sessions");
+ Long numExpiredSessions = (Long)metrics.getMetric(
+ "impala-server.num-sessions-expired");
+
+ for (int i = 0; i < connections.size(); ++i) {
+ Connection connection = connections.get(i);
+ Integer timeout = timeoutPeriods.get(i);
+
+ connection.createStatement().executeQuery("SELECT 1+2");
+ connection.createStatement().executeQuery("SET IDLE_SESSION_TIMEOUT=" +
+ Integer.toString(timeout));
+
+ lastTimeSessionActive.add(System.currentTimeMillis() / 1000);
+ }
+
+ assertEquals(numOpenSessions, (Long)metrics.getMetric(
+ "impala-server.num-open-hiveserver2-sessions"));
+ assertEquals(numExpiredSessions, (Long)metrics.getMetric(
+ "impala-server.num-sessions-expired"));
+
+ for (int timeout : timeoutPeriods) {
+ // Let's expire a session by sleeping,
+ // while renewing the remainders by issuing a query
+ // (except for the 0 timeout, which should never expire)
+ int timeoutToleranceMs = 1500;
+ int sleepPeriodMs = timeout * 1000 + timeoutToleranceMs;
+ Thread.sleep(sleepPeriodMs);
+
+ for (int i = 0; i < connections.size(); ++i) {
+ Connection connection = connections.get(i);
+
+ if (connection != null) {
+ Integer timeoutPeriod = timeoutPeriods.get(i);
+ long now = System.currentTimeMillis() / 1000;
+
+ boolean sessionIsValid = timeoutPeriod == 0 ||
+ timeoutPeriod > now - lastTimeSessionActive.get(i);
+
+ try (ResultSet rs = connection.createStatement().executeQuery("SELECT 1+2")) {
+ assertTrue(sessionIsValid);
+ lastTimeSessionActive.set(i, now);
+ }
+ catch (SQLException exception) {
+ assertFalse(sessionIsValid);
+ connection.close();
+ connections.set(i, null);
+ numOpenSessions -= 1;
+ numExpiredSessions += 1;
+ }
+ }
+ }
+
+ assertEquals(numOpenSessions, (Long)metrics.getMetric(
+ "impala-server.num-open-hiveserver2-sessions"));
+ assertEquals(numExpiredSessions, (Long)metrics.getMetric(
+ "impala-server.num-sessions-expired"));
+ }
+
+ assertNotNull("Connection with 0 timeout should not be null", connections.get(0));
+ connections.get(0).close();
+ connections.set(0, null);
+
+ for (Connection connection : connections) {
+ assertNull("Connection is not null", connection);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/fe/src/test/java/org/apache/impala/util/Metrics.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/util/Metrics.java b/fe/src/test/java/org/apache/impala/util/Metrics.java
new file mode 100644
index 0000000..537f406
--- /dev/null
+++ b/fe/src/test/java/org/apache/impala/util/Metrics.java
@@ -0,0 +1,82 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.util;
+
+import java.net.URL;
+import java.util.Scanner;
+
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+
+/**
+ * Utility class for retrieving metrics from the Impala webserver.
+ */
+public class Metrics {
+ private final static String DEFAULT_WEBSERVER_BASE_URL = "http://localhost:25000/";
+ private final static String JSON_METRICS = "jsonmetrics?json";
+
+ private String webserverBaseUrl;
+
+ public Metrics() {
+ this.webserverBaseUrl = DEFAULT_WEBSERVER_BASE_URL;
+ }
+
+ public Metrics(String webserverUrl) {
+ this.webserverBaseUrl = webserverUrl;
+ }
+
+ /**
+ * Returns the metric for the given metric id from the Impala web server.
+ * @param metricId identifier of the metric we want to retrieve
+ * @return A Java Object that holds the value for the given metric id.
+ * The caller needs to cast it to the appropriate type, e.g. Long, String, etc.
+ */
+ public Object getMetric(String metricId) throws Exception {
+ String content = readContent(new URL(this.webserverBaseUrl + JSON_METRICS));
+ if (content == null) return null;
+
+ JSONObject json = toJson(content);
+ if (json == null) return null;
+
+ return json.get(metricId);
+ }
+
+ private static String readContent(URL url) throws Exception {
+ String ret = null;
+
+ try (Scanner scanner = new Scanner(url.openStream(), "UTF-8")) {
+ scanner.useDelimiter("\\A");
+ if (scanner.hasNext()) {
+ ret = scanner.next();
+ }
+ }
+
+ return ret;
+ }
+
+ private static JSONObject toJson(String text) throws Exception {
+ JSONParser parser = new JSONParser();
+ Object obj = parser.parse(text);
+
+ if (!(obj instanceof JSONObject)) {
+ return null;
+ }
+
+ return (JSONObject)obj;
+ }
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/tests/custom_cluster/test_session_expiration.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_session_expiration.py b/tests/custom_cluster/test_session_expiration.py
index 8d47f9c..e89fffd 100644
--- a/tests/custom_cluster/test_session_expiration.py
+++ b/tests/custom_cluster/test_session_expiration.py
@@ -42,9 +42,50 @@ class TestSessionExpiration(CustomClusterTestSuite):
sleep(3)
assert num_expired == impalad.service.get_metric_value(
"impala-server.num-sessions-expired")
- # Wait for session expiration. Session timeout was set for 6 seconds, so Impala
- # will poll the session expiry queue every 3 seconds. So, as long as the sleep in
- # ImpalaSever::ExpireSessions() is not late, the session will expire in at most 9
- # seconds. The test has already waited 3 seconds.
+ # Wait for session expiration. Impala will poll the session expiry queue every second
impalad.service.wait_for_metric_value(
"impala-server.num-sessions-expired", num_expired + 1, 20)
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args("--idle_session_timeout=3")
+ def test_session_expiration_with_set(self, vector):
+ impalad = self.cluster.get_any_impalad()
+ self.client.close()
+ num_expired = impalad.service.get_metric_value("impala-server.num-sessions-expired")
+
+ # Test if we can set a shorter timeout than the process-wide option
+ client = impalad.service.create_beeswax_client()
+ client.execute("SET IDLE_SESSION_TIMEOUT=1")
+ sleep(2.5)
+ assert num_expired + 1 == impalad.service.get_metric_value(
+ "impala-server.num-sessions-expired")
+
+ # Test if we can set a longer timeout than the process-wide option
+ client = impalad.service.create_beeswax_client()
+ client.execute("SET IDLE_SESSION_TIMEOUT=10")
+ sleep(5)
+ assert num_expired + 1 == impalad.service.get_metric_value(
+ "impala-server.num-sessions-expired")
+
+
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args("--idle_session_timeout=5")
+ def test_unsetting_session_expiration(self, vector):
+ impalad = self.cluster.get_any_impalad()
+ self.client.close()
+ num_expired = impalad.service.get_metric_value("impala-server.num-sessions-expired")
+
+ # Test unsetting IDLE_SESSION_TIMEOUT
+ client = impalad.service.create_beeswax_client()
+ client.execute("SET IDLE_SESSION_TIMEOUT=1")
+
+ # Unset to 5 sec
+ client.execute('SET IDLE_SESSION_TIMEOUT=""')
+ sleep(2)
+ # client session should be alive at this point
+ assert num_expired == impalad.service.get_metric_value(
+ "impala-server.num-sessions-expired")
+ sleep(5)
+ # now client should have expired
+ assert num_expired + 1 == impalad.service.get_metric_value(
+ "impala-server.num-sessions-expired")
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/tests/custom_cluster/test_set_and_unset.py
----------------------------------------------------------------------
diff --git a/tests/custom_cluster/test_set_and_unset.py b/tests/custom_cluster/test_set_and_unset.py
index 8cc756c..cb596b4 100644
--- a/tests/custom_cluster/test_set_and_unset.py
+++ b/tests/custom_cluster/test_set_and_unset.py
@@ -84,6 +84,53 @@ class TestSetAndUnset(CustomClusterTestSuite, HS2TestSuite):
get_profile_req.sessionHandle = self.session_handle
assert "BATCH_SIZE=999" in self.hs2_client.GetRuntimeProfile(get_profile_req).profile
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ impalad_args="--idle_session_timeout=321")
+ @needs_session(TCLIService.TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V6)
+ def test_set_and_unset_session_timeout(self):
+ """
+ Starts Impala cluster with idle_session_timeout configured, and checks if
+ the SET query displays the correct value.
+
+ The Beeswax API and the HiveServer2 implementations are slightly different,
+ so the same test is run in both contexts.
+ """
+ # Beeswax API:
+ # Default value
+ result = self.execute_query_expect_success(self.client, "set")
+ assert "IDLE_SESSION_TIMEOUT\t321\tREGULAR" in result.data, "baseline"
+ # Session override
+ self.execute_query_expect_success(self.client, "set idle_session_timeout=123")
+ assert "IDLE_SESSION_TIMEOUT\t123\tREGULAR" in \
+ self.execute_query_expect_success(self.client, "set").data, "session override"
+ self.execute_query_expect_success(self.client, 'set idle_session_timeout=1000')
+ assert "IDLE_SESSION_TIMEOUT\t1000\tREGULAR" in \
+ self.execute_query_expect_success(self.client, "set").data, "session override"
+ self.execute_query_expect_success(self.client, 'set idle_session_timeout=0')
+ assert "IDLE_SESSION_TIMEOUT\t0\tREGULAR" in \
+ self.execute_query_expect_success(self.client, "set").data, "session override"
+ # Unset
+ self.execute_query_expect_success(self.client, 'set idle_session_timeout=""')
+ assert "IDLE_SESSION_TIMEOUT\t321\tREGULAR" in \
+ self.execute_query_expect_success(self.client, "set").data, "reset"
+ assert "IDLE_SESSION_TIMEOUT\t321\tREGULAR" in \
+ self.execute_query_expect_success(self.client, "set").data, "after errors"
+
+ # Same dance, but with HS2:
+ # Default value
+ assert ("IDLE_SESSION_TIMEOUT", "321") in self.get_set_results(), "baseline"
+ # Session override
+ self.execute_statement("set idle_session_timeout=123")
+ assert ("IDLE_SESSION_TIMEOUT", "123") in self.get_set_results(), "session override"
+ self.execute_statement("set idle_session_timeout=1000")
+ assert ("IDLE_SESSION_TIMEOUT", "1000") in self.get_set_results(), "session override"
+ self.execute_statement("set idle_session_timeout=0")
+ assert ("IDLE_SESSION_TIMEOUT", "0") in self.get_set_results(), "session override"
+ # Unset
+ self.execute_statement("set idle_session_timeout=''")
+ assert ("IDLE_SESSION_TIMEOUT", "321") in self.get_set_results(), "reset"
+
def get_set_results(self):
"""
Executes a "SET" HiveServer2 query and returns a list
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/tests/hs2/hs2_test_suite.py
----------------------------------------------------------------------
diff --git a/tests/hs2/hs2_test_suite.py b/tests/hs2/hs2_test_suite.py
index da9f68b..137bd46 100644
--- a/tests/hs2/hs2_test_suite.py
+++ b/tests/hs2/hs2_test_suite.py
@@ -228,13 +228,17 @@ class HS2TestSuite(ImpalaTestSuite):
assert False, 'Did not reach expected operation state %s in time, actual state was ' \
'%s' % (expected_state, get_operation_status_resp.operationState)
- def execute_statement(self, statement, conf_overlay=None):
- """Executes statement and returns response, which is checked."""
+ def execute_statement(self, statement, conf_overlay=None,
+ expected_status_code=TCLIService.TStatusCode.SUCCESS_STATUS,
+ expected_error_prefix=None):
+ """Executes statement and checks if the response meets the expectations.
+ If so, it returns the response."""
execute_statement_req = TCLIService.TExecuteStatementReq()
execute_statement_req.sessionHandle = self.session_handle
execute_statement_req.statement = statement
if conf_overlay:
execute_statement_req.confOverlay = conf_overlay
execute_statement_resp = self.hs2_client.ExecuteStatement(execute_statement_req)
- HS2TestSuite.check_response(execute_statement_resp)
+ HS2TestSuite.check_response(execute_statement_resp, expected_status_code,
+ expected_error_prefix)
return execute_statement_resp
http://git-wip-us.apache.org/repos/asf/impala/blob/ce65b43d/tests/hs2/test_hs2.py
----------------------------------------------------------------------
diff --git a/tests/hs2/test_hs2.py b/tests/hs2/test_hs2.py
index 5fa8fda..040d997 100644
--- a/tests/hs2/test_hs2.py
+++ b/tests/hs2/test_hs2.py
@@ -213,7 +213,7 @@ class TestHS2(HS2TestSuite):
"impala-server.num-sessions-expired")
for timeout in timeout_periods:
- sleep_period = timeout * 1.5
+ sleep_period = timeout + 1.5
time.sleep(sleep_period)
for i, session_handle in enumerate(session_handles):
if session_handle is not None: