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 2016/10/30 21:10:16 UTC
[3/3] incubator-impala git commit: IMPALA-3823: Add timer to measure
Parquet footer reads
IMPALA-3823: Add timer to measure Parquet footer reads
It's been observed that Parquet footer reads perform poorly especially
when reading from S3. This patch adds a timer "FooterProcessingTimer"
which keeps a track of the average time each split of each scan node
spends in reading and processing the parquet footer.
Added a new utility counter called SummaryStatsCounter which keeps
track of the min, max and average values seen so far from a set of
values. This counter is used to calculate the min, max and average
time taken to scan and process Parquet footers per query per node.
The RuntimeProfile has also been updated to keep a track of, display
and serialize this new counter to thrift.
BE tests have been added to verify that this counter works fine.
Change-Id: Icf87bad90037dd0cea63b10c537382ec0f980cbf
Reviewed-on: http://gerrit.cloudera.org:8080/4371
Reviewed-by: Sailesh Mukil <sa...@cloudera.com>
Tested-by: Marcel Kornacker <ma...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/29faca56
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/29faca56
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/29faca56
Branch: refs/heads/master
Commit: 29faca5680e34d9211eb8d551385b671198f626b
Parents: 9f3f4b7
Author: Sailesh Mukil <sa...@cloudera.com>
Authored: Tue Jul 5 13:11:35 2016 -0700
Committer: Marcel Kornacker <ma...@cloudera.com>
Committed: Sun Oct 30 19:36:13 2016 +0000
----------------------------------------------------------------------
be/src/exec/hdfs-parquet-scanner.cc | 17 +++-
be/src/exec/hdfs-parquet-scanner.h | 3 +
be/src/util/runtime-profile-counters.h | 74 ++++++++++++++---
be/src/util/runtime-profile-test.cc | 59 ++++++++++++++
be/src/util/runtime-profile.cc | 121 +++++++++++++++++++++++++++-
be/src/util/runtime-profile.h | 14 +++-
common/thrift/RuntimeProfile.thrift | 13 +++
tests/query_test/test_scanners.py | 18 +++++
8 files changed, 303 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/29faca56/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 542f4cc..79925a4 100644
--- a/be/src/exec/hdfs-parquet-scanner.cc
+++ b/be/src/exec/hdfs-parquet-scanner.cc
@@ -150,6 +150,7 @@ HdfsParquetScanner::HdfsParquetScanner(HdfsScanNodeBase* scan_node, RuntimeState
metadata_range_(NULL),
dictionary_pool_(new MemPool(scan_node->mem_tracker())),
assemble_rows_timer_(scan_node_->materialize_tuple_timer()),
+ process_footer_timer_stats_(NULL),
num_cols_counter_(NULL),
num_row_groups_counter_(NULL),
codegend_process_scratch_batch_fn_(NULL) {
@@ -164,6 +165,9 @@ Status HdfsParquetScanner::Open(ScannerContext* context) {
ADD_COUNTER(scan_node_->runtime_profile(), "NumColumns", TUnit::UNIT);
num_row_groups_counter_ =
ADD_COUNTER(scan_node_->runtime_profile(), "NumRowGroups", TUnit::UNIT);
+ process_footer_timer_stats_ =
+ ADD_SUMMARY_STATS_TIMER(
+ scan_node_->runtime_profile(), "FooterProcessingTime");
codegend_process_scratch_batch_fn_ = reinterpret_cast<ProcessScratchBatchFn>(
scan_node_->GetCodegenFn(THdfsFileFormat::PARQUET));
@@ -184,12 +188,21 @@ Status HdfsParquetScanner::Open(ScannerContext* context) {
DCHECK(parse_status_.ok()) << "Invalid parse_status_" << parse_status_.GetDetail();
+ // Each scan node can process multiple splits. Each split processes the footer once.
+ // We use a timer to measure the time taken to ProcessFooter() per split and add
+ // this time to the averaged timer.
+ MonotonicStopWatch single_footer_process_timer;
+ single_footer_process_timer.Start();
// First process the file metadata in the footer.
- Status status = ProcessFooter();
+ Status footer_status = ProcessFooter();
+ single_footer_process_timer.Stop();
+
+ process_footer_timer_stats_->UpdateCounter(single_footer_process_timer.ElapsedTime());
+
// Release I/O buffers immediately to make sure they are cleaned up
// in case we return a non-OK status anywhere below.
context_->ReleaseCompletedResources(NULL, true);
- RETURN_IF_ERROR(status);
+ RETURN_IF_ERROR(footer_status);
// Parse the file schema into an internal representation for schema resolution.
schema_resolver_.reset(new ParquetSchemaResolver(*scan_node_->hdfs_table(),
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/29faca56/be/src/exec/hdfs-parquet-scanner.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-parquet-scanner.h b/be/src/exec/hdfs-parquet-scanner.h
index 7707083..cc5795a 100644
--- a/be/src/exec/hdfs-parquet-scanner.h
+++ b/be/src/exec/hdfs-parquet-scanner.h
@@ -426,6 +426,9 @@ class HdfsParquetScanner : public HdfsScanner {
/// Timer for materializing rows. This ignores time getting the next buffer.
ScopedTimer<MonotonicStopWatch> assemble_rows_timer_;
+ /// Average and min/max time spent processing the footer by each split.
+ RuntimeProfile::SummaryStatsCounter* process_footer_timer_stats_;
+
/// Number of columns that need to be read.
RuntimeProfile::Counter* num_cols_counter_;
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/29faca56/be/src/util/runtime-profile-counters.h
----------------------------------------------------------------------
diff --git a/be/src/util/runtime-profile-counters.h b/be/src/util/runtime-profile-counters.h
index 77d7938..b37235f 100644
--- a/be/src/util/runtime-profile-counters.h
+++ b/be/src/util/runtime-profile-counters.h
@@ -46,6 +46,8 @@ namespace impala {
#define ADD_TIME_SERIES_COUNTER(profile, name, src_counter) \
(profile)->AddTimeSeriesCounter(name, src_counter)
#define ADD_TIMER(profile, name) (profile)->AddCounter(name, TUnit::TIME_NS)
+ #define ADD_SUMMARY_STATS_TIMER(profile, name) \
+ (profile)->AddSummaryStatsCounter(name, TUnit::TIME_NS)
#define ADD_CHILD_TIMER(profile, name, parent) \
(profile)->AddCounter(name, TUnit::TIME_NS, parent)
#define SCOPED_TIMER(c) \
@@ -65,6 +67,7 @@ namespace impala {
#define ADD_COUNTER(profile, name, unit) NULL
#define ADD_TIME_SERIES_COUNTER(profile, name, src_counter) NULL
#define ADD_TIMER(profile, name) NULL
+ #define ADD_SUMMARY_STATS_TIMER(profile, name) NULL
#define ADD_CHILD_TIMER(profile, name, parent) NULL
#define SCOPED_TIMER(c)
#define CANCEL_SAFE_SCOPED_TIMER(c)
@@ -181,17 +184,9 @@ class RuntimeProfile::AveragedCounter : public RuntimeProfile::Counter {
/// The value for this counter should be updated through UpdateCounter().
/// Set() and Add() should not be used.
- virtual void Set(double value) {
- DCHECK(false);
- }
-
- virtual void Set(int64_t value) {
- DCHECK(false);
- }
-
- virtual void Add(int64_t delta) {
- DCHECK(false);
- }
+ virtual void Set(double value) { DCHECK(false); }
+ virtual void Set(int64_t value) { DCHECK(false); }
+ virtual void Add(int64_t delta) { DCHECK(false); }
private:
/// Map from counters to their existing values. Modified via UpdateCounter().
@@ -204,6 +199,63 @@ class RuntimeProfile::AveragedCounter : public RuntimeProfile::Counter {
int64_t current_int_sum_;
};
+/// This counter records multiple values and keeps a track of the minimum, maximum and
+/// average value of all the values seen so far.
+/// Unlike the AveragedCounter, this only keeps track of statistics of raw values
+/// whereas the AveragedCounter maintains an average of counters.
+/// value() stores the average.
+class RuntimeProfile::SummaryStatsCounter : public RuntimeProfile::Counter {
+ public:
+ SummaryStatsCounter(TUnit::type unit, int32_t total_num_values,
+ int64_t min_value, int64_t max_value, int64_t sum)
+ : Counter(unit),
+ total_num_values_(total_num_values),
+ min_(min_value),
+ max_(max_value),
+ sum_(sum) {
+ value_.Store(total_num_values == 0 ? 0 : sum / total_num_values);
+ }
+
+ SummaryStatsCounter(TUnit::type unit)
+ : Counter(unit),
+ total_num_values_(0),
+ min_(numeric_limits<int64_t>::max()),
+ max_(numeric_limits<int64_t>::min()),
+ sum_(0) {
+ }
+
+ int64_t MinValue();
+ int64_t MaxValue();
+ int32_t TotalNumValues();
+
+ /// Update sum_ with the new value and also update the min and the max values
+ /// seen so far.
+ void UpdateCounter(int64_t new_value);
+
+ /// The value for this counter should be updated through UpdateCounter() or SetStats().
+ /// Set() and Add() should not be used.
+ virtual void Set(double value) { DCHECK(false); }
+ virtual void Set(int64_t value) { DCHECK(false); }
+ virtual void Add(int64_t delta) { DCHECK(false); }
+
+ /// Overwrites the existing counter with 'counter'
+ void SetStats(const TSummaryStatsCounter& counter);
+
+ void ToThrift(TSummaryStatsCounter* counter, const std::string& name);
+
+ private:
+ /// The total number of values seen so far.
+ int32_t total_num_values_;
+
+ /// Summary statistics of values seen so far.
+ int64_t min_;
+ int64_t max_;
+ int64_t sum_;
+
+ // Protects min_, max_, sum_, total_num_values_ and value_.
+ SpinLock lock_;
+};
+
/// A set of counters that measure thread info, such as total time, user time, sys time.
class RuntimeProfile::ThreadCounters {
private:
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/29faca56/be/src/util/runtime-profile-test.cc
----------------------------------------------------------------------
diff --git a/be/src/util/runtime-profile-test.cc b/be/src/util/runtime-profile-test.cc
index 6910ac9..fe9f3ae 100644
--- a/be/src/util/runtime-profile-test.cc
+++ b/be/src/util/runtime-profile-test.cc
@@ -258,6 +258,65 @@ TEST(CountersTest, HighWaterMarkCounters) {
EXPECT_EQ(bytes_counter->value(), 28);
}
+TEST(CountersTest, SummaryStatsCounters) {
+ ObjectPool pool;
+ RuntimeProfile profile1(&pool, "Profile 1");
+ RuntimeProfile::SummaryStatsCounter* summary_stats_counter_1 =
+ profile1.AddSummaryStatsCounter("summary_stats", TUnit::UNIT);
+
+ EXPECT_EQ(summary_stats_counter_1->value(), 0);
+ EXPECT_EQ(summary_stats_counter_1->MinValue(), numeric_limits<int64_t>::max());
+ EXPECT_EQ(summary_stats_counter_1->MaxValue(), numeric_limits<int64_t>::min());
+
+ summary_stats_counter_1->UpdateCounter(10);
+ EXPECT_EQ(summary_stats_counter_1->value(), 10);
+ EXPECT_EQ(summary_stats_counter_1->MinValue(), 10);
+ EXPECT_EQ(summary_stats_counter_1->MaxValue(), 10);
+
+ // Check that the average stays the same when updating with the same number.
+ summary_stats_counter_1->UpdateCounter(10);
+ EXPECT_EQ(summary_stats_counter_1->value(), 10);
+ EXPECT_EQ(summary_stats_counter_1->MinValue(), 10);
+ EXPECT_EQ(summary_stats_counter_1->MaxValue(), 10);
+
+ summary_stats_counter_1->UpdateCounter(40);
+ EXPECT_EQ(summary_stats_counter_1->value(), 20);
+ EXPECT_EQ(summary_stats_counter_1->MinValue(), 10);
+ EXPECT_EQ(summary_stats_counter_1->MaxValue(), 40);
+
+ // Verify an update with 0. This should still change the average as the number of
+ // samples increase
+ summary_stats_counter_1->UpdateCounter(0);
+ EXPECT_EQ(summary_stats_counter_1->value(), 15);
+ EXPECT_EQ(summary_stats_counter_1->MinValue(), 0);
+ EXPECT_EQ(summary_stats_counter_1->MaxValue(), 40);
+
+ // Verify a negative update..
+ summary_stats_counter_1->UpdateCounter(-40);
+ EXPECT_EQ(summary_stats_counter_1->value(), 4);
+ EXPECT_EQ(summary_stats_counter_1->MinValue(), -40);
+ EXPECT_EQ(summary_stats_counter_1->MaxValue(), 40);
+
+ RuntimeProfile profile2(&pool, "Profile 2");
+ RuntimeProfile::SummaryStatsCounter* summary_stats_counter_2 =
+ profile2.AddSummaryStatsCounter("summary_stats", TUnit::UNIT);
+
+ summary_stats_counter_2->UpdateCounter(100);
+ EXPECT_EQ(summary_stats_counter_2->value(), 100);
+ EXPECT_EQ(summary_stats_counter_2->MinValue(), 100);
+ EXPECT_EQ(summary_stats_counter_2->MaxValue(), 100);
+
+ TRuntimeProfileTree tprofile1;
+ profile1.ToThrift(&tprofile1);
+
+ // Merge profile1 and profile2 and check that profile2 is overwritten.
+ profile2.Update(tprofile1);
+ EXPECT_EQ(summary_stats_counter_2->value(), 4);
+ EXPECT_EQ(summary_stats_counter_2->MinValue(), -40);
+ EXPECT_EQ(summary_stats_counter_2->MaxValue(), 40);
+
+}
+
TEST(CountersTest, DerivedCounters) {
ObjectPool pool;
RuntimeProfile profile(&pool, "Profile");
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/29faca56/be/src/util/runtime-profile.cc
----------------------------------------------------------------------
diff --git a/be/src/util/runtime-profile.cc b/be/src/util/runtime-profile.cc
index b19ba33..a9b46b9 100644
--- a/be/src/util/runtime-profile.cc
+++ b/be/src/util/runtime-profile.cc
@@ -133,6 +133,14 @@ RuntimeProfile* RuntimeProfile::CreateFromThrift(ObjectPool* pool,
}
}
+ if (node.__isset.summary_stats_counters) {
+ for (const TSummaryStatsCounter& val: node.summary_stats_counters) {
+ profile->summary_stats_map_[val.name] =
+ pool->Add(new SummaryStatsCounter(
+ val.unit, val.total_num_values, val.min_value, val.max_value, val.sum));
+ }
+ }
+
profile->child_counter_map_ = node.child_counters_map;
profile->info_strings_ = node.info_strings;
profile->info_strings_display_order_ = node.info_strings_display_order;
@@ -286,6 +294,21 @@ void RuntimeProfile::Update(const vector<TRuntimeProfileNode>& nodes, int* idx)
}
}
+ {
+ lock_guard<SpinLock> l(summary_stats_map_lock_);
+ for (int i = 0; i < node.summary_stats_counters.size(); ++i) {
+ const TSummaryStatsCounter& c = node.summary_stats_counters[i];
+ SummaryStatsCounterMap::iterator it = summary_stats_map_.find(c.name);
+ if (it == summary_stats_map_.end()) {
+ summary_stats_map_[c.name] =
+ pool_->Add(new SummaryStatsCounter(
+ c.unit, c.total_num_values, c.min_value, c.max_value, c.sum));
+ } else {
+ it->second->SetStats(c);
+ }
+ }
+ }
+
++*idx;
{
lock_guard<SpinLock> l(children_lock_);
@@ -651,6 +674,28 @@ void RuntimeProfile::PrettyPrint(ostream* s, const string& prefix) const {
}
}
+ {
+ lock_guard<SpinLock> l(summary_stats_map_lock_);
+ // Print all SummaryStatsCounters as following:
+ // <Name>: (Avg: <value> ; Min: <min_value> ; Max: <max_value> ;
+ // Number of samples: <total>)
+ for (const SummaryStatsCounterMap::value_type& v: summary_stats_map_) {
+ if (v.second->TotalNumValues() == 0) {
+ // No point printing all the stats if number of samples is zero.
+ stream << prefix << " - " << v.first << ": "
+ << PrettyPrinter::Print(v.second->value(), v.second->unit(), true)
+ << " (Number of samples: " << v.second->TotalNumValues() << ")" << endl;
+ } else {
+ stream << prefix << " - " << v.first << ": (Avg: "
+ << PrettyPrinter::Print(v.second->value(), v.second->unit(), true)
+ << " ; Min: "
+ << PrettyPrinter::Print(v.second->MinValue(), v.second->unit(), true)
+ << " ; Max: "
+ << PrettyPrinter::Print(v.second->MaxValue(), v.second->unit(), true)
+ << " ; Number of samples: " << v.second->TotalNumValues() << ")" << endl;
+ }
+ }
+ }
RuntimeProfile::PrintChildCounters(
prefix, ROOT_COUNTER, counter_map, child_counter_map, s);
@@ -760,8 +805,8 @@ void RuntimeProfile::ToThrift(vector<TRuntimeProfileNode>* nodes) const {
{
lock_guard<SpinLock> l(time_series_counter_map_lock_);
if (time_series_counter_map_.size() != 0) {
- node.__set_time_series_counters(vector<TTimeSeriesCounter>());
- node.time_series_counters.resize(time_series_counter_map_.size());
+ node.__set_time_series_counters(
+ vector<TTimeSeriesCounter>(time_series_counter_map_.size()));
int idx = 0;
for (const TimeSeriesCounterMap::value_type& val: time_series_counter_map_) {
val.second->ToThrift(&node.time_series_counters[idx++]);
@@ -769,6 +814,18 @@ void RuntimeProfile::ToThrift(vector<TRuntimeProfileNode>* nodes) const {
}
}
+ {
+ lock_guard<SpinLock> l(summary_stats_map_lock_);
+ if (summary_stats_map_.size() != 0) {
+ node.__set_summary_stats_counters(
+ vector<TSummaryStatsCounter>(summary_stats_map_.size()));
+ int idx = 0;
+ for (const SummaryStatsCounterMap::value_type& val: summary_stats_map_) {
+ val.second->ToThrift(&node.summary_stats_counters[idx++], val.first);
+ }
+ }
+ }
+
ChildVector children;
{
lock_guard<SpinLock> l(children_lock_);
@@ -896,6 +953,18 @@ void RuntimeProfile::PrintChildCounters(const string& prefix,
}
}
+RuntimeProfile::SummaryStatsCounter* RuntimeProfile::AddSummaryStatsCounter(
+ const string& name, TUnit::type unit, const std::string& parent_counter_name) {
+ DCHECK_EQ(is_averaged_profile_, false);
+ lock_guard<SpinLock> l(summary_stats_map_lock_);
+ if (summary_stats_map_.find(name) != summary_stats_map_.end()) {
+ return summary_stats_map_[name];
+ }
+ SummaryStatsCounter* counter = pool_->Add(new SummaryStatsCounter(unit));
+ summary_stats_map_[name] = counter;
+ return counter;
+}
+
RuntimeProfile::TimeSeriesCounter* RuntimeProfile::AddTimeSeriesCounter(
const string& name, TUnit::type unit, DerivedCounterFunction fn) {
DCHECK(fn != NULL);
@@ -945,4 +1014,52 @@ void RuntimeProfile::EventSequence::ToThrift(TEventSequence* seq) const {
}
}
+void RuntimeProfile::SummaryStatsCounter::ToThrift(TSummaryStatsCounter* counter,
+ const std::string& name) {
+ lock_guard<SpinLock> l(lock_);
+ counter->name = name;
+ counter->unit = unit_;
+ counter->sum = sum_;
+ counter->total_num_values = total_num_values_;
+ counter->min_value = min_;
+ counter->max_value = max_;
+}
+
+void RuntimeProfile::SummaryStatsCounter::UpdateCounter(int64_t new_value) {
+ lock_guard<SpinLock> l(lock_);
+
+ ++total_num_values_;
+ sum_ += new_value;
+ value_.Store(sum_ / total_num_values_);
+
+ if (new_value < min_) min_ = new_value;
+ if (new_value > max_) max_ = new_value;
+}
+
+void RuntimeProfile::SummaryStatsCounter::SetStats(const TSummaryStatsCounter& counter) {
+ lock_guard<SpinLock> l(lock_);
+ unit_ = counter.unit;
+ sum_ = counter.sum;
+ total_num_values_ = counter.total_num_values;
+ min_ = counter.min_value;
+ max_ = counter.max_value;
+
+ value_.Store(sum_ / total_num_values_);
+}
+
+int64_t RuntimeProfile::SummaryStatsCounter::MinValue() {
+ lock_guard<SpinLock> l(lock_);
+ return min_;
+}
+
+int64_t RuntimeProfile::SummaryStatsCounter::MaxValue() {
+ lock_guard<SpinLock> l(lock_);
+ return max_;
+}
+
+int32_t RuntimeProfile::SummaryStatsCounter::TotalNumValues() {
+ lock_guard<SpinLock> l(lock_);
+ return total_num_values_;
+}
+
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/29faca56/be/src/util/runtime-profile.h
----------------------------------------------------------------------
diff --git a/be/src/util/runtime-profile.h b/be/src/util/runtime-profile.h
index 513f39d..c409d62 100644
--- a/be/src/util/runtime-profile.h
+++ b/be/src/util/runtime-profile.h
@@ -94,6 +94,7 @@ class RuntimeProfile {
class DerivedCounter;
class EventSequence;
class HighWaterMarkCounter;
+ class SummaryStatsCounter;
class ThreadCounters;
class TimeSeriesCounter;
@@ -159,8 +160,13 @@ class RuntimeProfile {
Counter* AddCounter(const std::string& name, TUnit::type unit,
const std::string& parent_counter_name = "");
+ /// Adds a counter that tracks the min, max and average values to the runtime profile.
+ /// Otherwise, same behavior as AddCounter().
+ SummaryStatsCounter* AddSummaryStatsCounter(const std::string& name, TUnit::type unit,
+ const std::string& parent_counter_name = "");
+
/// Adds a high water mark counter to the runtime profile. Otherwise, same behavior
- /// as AddCounter()
+ /// as AddCounter().
HighWaterMarkCounter* AddHighWaterMarkCounter(const std::string& name,
TUnit::type unit, const std::string& parent_counter_name = "");
@@ -396,6 +402,12 @@ class RuntimeProfile {
/// Protects time_series_counter_map_.
mutable SpinLock time_series_counter_map_lock_;
+ typedef std::map<std::string, SummaryStatsCounter*> SummaryStatsCounterMap;
+ SummaryStatsCounterMap summary_stats_map_;
+
+ /// Protects summary_stats_map_.
+ mutable SpinLock summary_stats_map_lock_;
+
Counter counter_total_time_;
/// Total time spent waiting (on non-children) that should not be counted when
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/29faca56/common/thrift/RuntimeProfile.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/RuntimeProfile.thrift b/common/thrift/RuntimeProfile.thrift
index 068108b..8751b17 100644
--- a/common/thrift/RuntimeProfile.thrift
+++ b/common/thrift/RuntimeProfile.thrift
@@ -52,6 +52,16 @@ struct TTimeSeriesCounter {
4: required list<i64> values
}
+// Thrift version of RuntimeProfile::SummaryStatsCounter.
+struct TSummaryStatsCounter {
+ 1: required string name
+ 2: required Metrics.TUnit unit
+ 3: required i64 sum
+ 4: required i64 total_num_values
+ 5: required i64 min_value
+ 6: required i64 max_value
+}
+
// A single runtime profile
struct TRuntimeProfileNode {
1: required string name
@@ -80,6 +90,9 @@ struct TRuntimeProfileNode {
// List of time series counters
10: optional list<TTimeSeriesCounter> time_series_counters
+
+ // List of summary stats counters
+ 11: optional list<TSummaryStatsCounter> summary_stats_counters
}
// A flattened tree of runtime profiles, obtained by an
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/29faca56/tests/query_test/test_scanners.py
----------------------------------------------------------------------
diff --git a/tests/query_test/test_scanners.py b/tests/query_test/test_scanners.py
index 6dce0af..0231261 100644
--- a/tests/query_test/test_scanners.py
+++ b/tests/query_test/test_scanners.py
@@ -337,6 +337,14 @@ class TestParquet(ImpalaTestSuite):
'ScanRangesComplete: ([0-9]*)', runtime_profile)
num_rows_read_list = re.findall('RowsRead: [0-9.K]* \(([0-9]*)\)', runtime_profile)
+ REGEX_UNIT_SECOND = "[0-9]*[s]*[0-9]*[.]*[0-9]*[nm]*[s]*"
+ REGEX_MIN_MAX_FOOTER_PROCESSING_TIME = \
+ ("FooterProcessingTime: \(Avg: %s ; \(Min: (%s) ; Max: (%s) ; "
+ "Number of samples: %s\)" % (REGEX_UNIT_SECOND, REGEX_UNIT_SECOND,
+ REGEX_UNIT_SECOND, "[0-9]*"))
+ footer_processing_time_list = re.findall(
+ REGEX_MIN_MAX_FOOTER_PROCESSING_TIME, runtime_profile)
+
# This will fail if the number of impalads != 3
# The fourth fragment is the "Averaged Fragment"
assert len(num_row_groups_list) == 4
@@ -359,6 +367,16 @@ class TestParquet(ImpalaTestSuite):
for scan_ranges_complete in scan_ranges_complete_list:
assert int(scan_ranges_complete) == ranges_per_node
+ # This checks if the SummaryStatsCounter works correctly. When there is one scan
+ # range per node, we verify that the FooterProcessingTime counter has the min, max
+ # and average values as the same since we have only one sample (i.e. only one range)
+ # TODO: Test this for multiple ranges per node as well. This requires parsing the
+ # stat times as strings and comparing if min <= avg <= max.
+ if ranges_per_node == 1:
+ for min_max_time in footer_processing_time_list:
+ # Assert that (min == avg == max)
+ assert min_max_time[0] == min_max_time[1] == min_max_time[2] != 0
+
def test_annotate_utf8_option(self, vector, unique_database):
if self.exploration_strategy() != 'exhaustive': pytest.skip("Only run in exhaustive")