You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by jr...@apache.org on 2018/01/19 21:41:30 UTC
[1/8] impala git commit: IMPALA-6419: Revert "IMPALA-6383: free
memory after skipping parquet row groups"
Repository: impala
Updated Branches:
refs/heads/2.x 35a3e186d -> 579e33207
IMPALA-6419: Revert "IMPALA-6383: free memory after skipping parquet row groups"
This reverts commit 10fb24afb966c567adcf632a314f6af1826f19fc.
Change-Id: I4dd62380d02b61ca46f856b4eb40670b71e28140
Reviewed-on: http://gerrit.cloudera.org:8080/9054
Reviewed-by: Alex Behm <al...@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/f5d73f5e
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/f5d73f5e
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/f5d73f5e
Branch: refs/heads/2.x
Commit: f5d73f5e76d477ae47e02df4fb69ad590363c0d6
Parents: 35a3e18
Author: Tim Armstrong <ta...@cloudera.com>
Authored: Thu Jan 18 09:43:30 2018 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Thu Jan 18 21:25:28 2018 +0000
----------------------------------------------------------------------
be/src/exec/hdfs-parquet-scanner.cc | 26 +++++---------------------
be/src/exec/hdfs-parquet-scanner.h | 5 -----
be/src/exec/scanner-context.h | 8 ++++----
3 files changed, 9 insertions(+), 30 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/f5d73f5e/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 6380722..f0f280d 100644
--- a/be/src/exec/hdfs-parquet-scanner.cc
+++ b/be/src/exec/hdfs-parquet-scanner.cc
@@ -228,7 +228,6 @@ 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(true);
- context_->ClearStreams();
RETURN_IF_ERROR(footer_status);
// Parse the file schema into an internal representation for schema resolution.
@@ -264,7 +263,7 @@ void HdfsParquetScanner::Close(RowBatch* row_batch) {
}
} else {
template_tuple_pool_->FreeAll();
- dictionary_pool_->FreeAll();
+ dictionary_pool_.get()->FreeAll();
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
@@ -479,6 +478,7 @@ Status HdfsParquetScanner::GetNextInternal(RowBatch* row_batch) {
// Transfer resources and clear streams if there is any leftover from the previous
// row group. We will create new streams for the next row group.
FlushRowGroupResources(row_batch);
+ context_->ClearStreams();
if (!advance_row_group_) {
Status status =
ValidateEndOfRowGroup(column_readers_, row_group_idx_, row_group_rows_read_);
@@ -619,9 +619,6 @@ Status HdfsParquetScanner::NextRowGroup() {
while (true) {
// Reset the parse status for the next row group.
parse_status_ = Status::OK();
- // Make sure that we don't have leftover resources from the file metadata scan range
- // or previous row groups.
- DCHECK_EQ(0, context_->NumStreams());
++row_group_idx_;
if (row_group_idx_ >= file_metadata_.row_groups.size()) {
@@ -672,9 +669,6 @@ Status HdfsParquetScanner::NextRowGroup() {
// of the column.
RETURN_IF_ERROR(InitColumns(row_group_idx_, dict_filterable_readers_));
- // InitColumns() may have allocated resources to scan columns. If we skip this row
- // group below, we must call ReleaseSkippedRowGroupResources() before continuing.
-
// If there is a dictionary-encoded column where every value is eliminated
// by a conjunct, the row group can be eliminated. This initializes dictionaries
// for all columns visited.
@@ -683,12 +677,10 @@ Status HdfsParquetScanner::NextRowGroup() {
if (!status.ok()) {
// Either return an error or skip this row group if it is ok to ignore errors
RETURN_IF_ERROR(state_->LogOrReturnError(status.msg()));
- ReleaseSkippedRowGroupResources();
continue;
}
if (skip_row_group_on_dict_filters) {
COUNTER_ADD(num_dict_filtered_row_groups_counter_, 1);
- ReleaseSkippedRowGroupResources();
continue;
}
@@ -700,7 +692,6 @@ Status HdfsParquetScanner::NextRowGroup() {
if (!status.ok()) {
// Either return an error or skip this row group if it is ok to ignore errors
RETURN_IF_ERROR(state_->LogOrReturnError(status.msg()));
- ReleaseSkippedRowGroupResources();
continue;
}
@@ -739,16 +730,9 @@ void HdfsParquetScanner::FlushRowGroupResources(RowBatch* row_batch) {
row_batch->tuple_data_pool()->AcquireData(dictionary_pool_.get(), false);
scratch_batch_->ReleaseResources(row_batch->tuple_data_pool());
context_->ReleaseCompletedResources(true);
- for (ParquetColumnReader* col_reader : column_readers_) col_reader->Close(row_batch);
- context_->ClearStreams();
-}
-
-void HdfsParquetScanner::ReleaseSkippedRowGroupResources() {
- dictionary_pool_->FreeAll();
- scratch_batch_->ReleaseResources(nullptr);
- context_->ReleaseCompletedResources(true);
- for (ParquetColumnReader* col_reader : column_readers_) col_reader->Close(nullptr);
- context_->ClearStreams();
+ for (ParquetColumnReader* col_reader : column_readers_) {
+ col_reader->Close(row_batch);
+ }
}
bool HdfsParquetScanner::IsDictFilterable(ParquetColumnReader* col_reader) {
http://git-wip-us.apache.org/repos/asf/impala/blob/f5d73f5e/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 cea06ed..99b5a60 100644
--- a/be/src/exec/hdfs-parquet-scanner.h
+++ b/be/src/exec/hdfs-parquet-scanner.h
@@ -642,11 +642,6 @@ class HdfsParquetScanner : public HdfsScanner {
/// Should be called after completing a row group and when returning the last batch.
void FlushRowGroupResources(RowBatch* row_batch);
- /// Releases resources associated with a row group that was skipped and closes all
- /// column readers. Should be called after skipping a row group from which no rows
- /// were returned.
- void ReleaseSkippedRowGroupResources();
-
/// Evaluates whether the column reader is eligible for dictionary predicates
bool IsDictFilterable(ParquetColumnReader* col_reader);
http://git-wip-us.apache.org/repos/asf/impala/blob/f5d73f5e/be/src/exec/scanner-context.h
----------------------------------------------------------------------
diff --git a/be/src/exec/scanner-context.h b/be/src/exec/scanner-context.h
index 09a4bdc..e316063 100644
--- a/be/src/exec/scanner-context.h
+++ b/be/src/exec/scanner-context.h
@@ -89,6 +89,7 @@ class ScannerContext {
ScannerContext(RuntimeState*, HdfsScanNodeBase*, HdfsPartitionDescriptor*,
io::ScanRange* scan_range, const std::vector<FilterContext>& filter_ctxs,
MemPool* expr_results_pool);
+
/// Destructor verifies that all stream objects have been released.
~ScannerContext();
@@ -337,8 +338,6 @@ class ScannerContext {
return streams_[idx].get();
}
- int NumStreams() const { return streams_.size(); }
-
/// Release completed resources for all streams, e.g. the last buffer in each stream if
/// the current read position is at the end of the buffer. If 'done' is true all
/// resources are freed, even if the caller has not read that data yet. After calling
@@ -355,8 +354,8 @@ class ScannerContext {
/// size to 0.
void ClearStreams();
- /// Add a stream to this ScannerContext for 'range'. The stream is owned by this
- /// context.
+ /// Add a stream to this ScannerContext for 'range'. Returns the added stream.
+ /// The stream is created in the runtime state's object pool
Stream* AddStream(io::ScanRange* range);
/// Returns false if scan_node_ is multi-threaded and has been cancelled.
@@ -371,6 +370,7 @@ class ScannerContext {
RuntimeState* state_;
HdfsScanNodeBase* scan_node_;
+
HdfsPartitionDescriptor* partition_desc_;
/// Vector of streams. Non-columnar formats will always have one stream per context.
[8/8] impala git commit: IMPALA-6368: make test_chars parallel
Posted by jr...@apache.org.
IMPALA-6368: make test_chars parallel
Previously it had to be executed serially because it modified tables in
the functional database.
This change separates out tests that use temporary tables and runs those
in a unique_database.
Testing:
Ran locally in a loop with parallelism of 4 for a while.
Change-Id: I2f62ede90f619b8cebbb1276bab903e7555d9744
Reviewed-on: http://gerrit.cloudera.org:8080/9022
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/579e3320
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/579e3320
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/579e3320
Branch: refs/heads/2.x
Commit: 579e33207b3bbe8b6a26cf1fe1e7fd8d26021475
Parents: 3f00d10
Author: Tim Armstrong <ta...@cloudera.com>
Authored: Fri Jan 12 17:10:51 2018 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Fri Jan 19 09:55:52 2018 +0000
----------------------------------------------------------------------
.../queries/QueryTest/chars-tmp-tables.test | 275 +++++++++++++++++++
.../queries/QueryTest/chars.test | 262 +-----------------
tests/query_test/test_chars.py | 50 +---
3 files changed, 286 insertions(+), 301 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/579e3320/testdata/workloads/functional-query/queries/QueryTest/chars-tmp-tables.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/chars-tmp-tables.test b/testdata/workloads/functional-query/queries/QueryTest/chars-tmp-tables.test
new file mode 100644
index 0000000..f6dc4c4
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/chars-tmp-tables.test
@@ -0,0 +1,275 @@
+====
+---- QUERY
+create table test_char_tmp (c char(5))
+---- RESULTS
+====
+---- QUERY
+insert into test_char_tmp select cast("hello" as char(5))
+---- RESULTS
+: 1
+====
+---- QUERY
+select * from test_char_tmp
+---- TYPES
+char
+---- RESULTS
+'hello'
+====
+---- QUERY
+# Regression test for IMPALA-1248
+insert into test_char_tmp
+values (cast("hel" as char(5))),
+ (cast(cast("hello000" as VARCHAR(8)) as char(5)))
+====
+---- QUERY
+select * from test_char_tmp where c = cast('hel' as char(5))
+---- TYPES
+char
+---- RESULTS
+'hel '
+====
+---- QUERY
+insert into test_char_tmp values (NULL)
+====
+---- QUERY
+select * from test_char_tmp as A CROSS JOIN test_char_tmp as B
+where B.c = cast('hel' as CHAR(5))
+ORDER BY A.c
+---- TYPES
+char, char
+---- RESULTS
+'hel ','hel '
+'hello','hel '
+'hello','hel '
+'NULL','hel '
+====
+---- QUERY
+select * from test_char_tmp as A, test_char_tmp as B
+where A.c = B.c AND A.c != 'hello'
+---- TYPES
+char, char
+---- RESULTS
+'hel ','hel '
+====
+---- QUERY
+select lower(c) from test_char_tmp ORDER BY c
+---- TYPES
+string
+---- RESULTS
+'hel '
+'hello'
+'hello'
+'NULL'
+====
+---- QUERY
+create table test_varchar_tmp (vc varchar(5))
+---- RESULTS
+====
+---- QUERY
+insert into test_varchar_tmp values (cast("hello" as varchar(5)))
+====
+---- QUERY
+select * from test_varchar_tmp
+---- TYPES
+string
+---- RESULTS
+'hello'
+====
+---- QUERY
+insert into test_varchar_tmp values (cast("xyzzzzz12" as varchar(7)))
+---- CATCH
+would need to be cast to VARCHAR(5)
+====
+---- QUERY
+select cast("xyzzzzz12" as varchar(-1))
+---- CATCH
+Syntax error
+====
+====
+---- QUERY
+insert into test_varchar_tmp values (cast("hel" as varchar(4)))
+====
+---- QUERY
+select * from test_varchar_tmp
+---- TYPES
+string
+---- RESULTS
+'hello'
+'hel'
+====
+---- QUERY
+create table allchars
+(cshort char(5), clong char(140), vc varchar(5))
+---- RESULTS
+====
+---- QUERY
+insert into allchars values (cast("123456" as char(5)), cast("123456" as char(140)),
+cast("123456" as varchar(5)))
+====
+---- QUERY
+select cshort, clong, vc from allchars
+---- TYPES
+char,char,string
+---- RESULTS
+'12345','123456 ','12345'
+====
+---- QUERY
+create table allchars_par
+(cshort char(5), clong char(140), vc varchar(5)) stored as parquet
+---- RESULTS
+====
+---- QUERY
+insert into allchars_par values (cast("123456" as char(5)), cast("123456" as char(140)),
+cast("123456" as varchar(5)))
+====
+---- QUERY
+select cshort, clong, vc from allchars_par
+---- TYPES
+char,char,string
+---- RESULTS
+'12345','123456 ','12345'
+====
+---- QUERY
+create table char_parts (vc varchar(32)) partitioned by
+(csp char(5), clp char(140), vcp varchar(32))
+====
+---- QUERY
+insert into char_parts (csp, clp, vcp, vc) select cs, cl, vc, vc from functional.chars_tiny
+====
+---- QUERY
+select csp, clp, vcp from char_parts where csp != cast('dne' as char(5)) order by csp
+---- TYPES
+char, char, string
+---- RESULTS
+'1aaaa','1bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb','1cccc'
+'2aaaa','2bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb','2cccccc'
+'3aaa ','3bbbbb ','3ccc'
+'4aa ','4bbbb ','4cc'
+'5a ','5bbb ','5c'
+'6a ','6b ','6c'
+'6a ','6b ','6c'
+'a ','b ','c'
+====
+---- QUERY
+insert into char_parts partition (csp=cast('foo' as char(5)),
+clp=cast('01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789xxxxxxx' as char(140)),
+vcp=cast('myvar' as varchar(32))) select cast('val' as varchar(32));
+====
+---- QUERY
+select csp, clp, vcp from char_parts where csp = cast('foo' as char(5))
+---- TYPES
+char, char, string
+---- RESULTS
+'foo ','01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789','myvar'
+====
+---- QUERY
+# Regression test for IMPALA-1322
+create table t_1822 (c10 char(10), c100 char(100), v100 varchar(100), v200 varchar(200), s string);
+====
+---- QUERY
+# Regression test for IMPALA-1322
+insert into t_1822 values (cast('a' as char(1)), cast('a' as char(1)),
+cast('a' as varchar(1)), cast('a' as varchar(1)), 'a');
+====
+---- QUERY
+# Regression test for IMPALA-1316
+select count(*) from t_1822 as t join t_1822 as tt
+on cast(tt.s as char(129)) = t.c10 and
+cast(tt.s as char(129)) = t.c100 and tt.c10 = t.c100;
+---- TYPES
+bigint
+---- RESULTS
+1
+====
+---- QUERY
+create table
+test_char_nulls ( c20 char(20),
+ c40 char(40),
+ c60 char(60),
+ c80 char(80),
+ c81 char(81),
+ c82 char(82),
+ c100 char(100),
+ c120 char(120),
+ c140 char(140))
+---- RESULTS
+====
+---- QUERY
+insert into test_char_nulls
+values (NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL),
+ (NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL)
+---- RESULTS
+: 2
+====
+---- QUERY
+# Regression test for IMPALA-1339
+select c20 from test_char_nulls group by c20;
+---- TYPES
+char
+---- RESULTS
+'NULL'
+====
+---- QUERY
+# Regression test for IMPALA-1339
+select c40 from test_char_nulls group by c40;
+---- TYPES
+char
+---- RESULTS
+'NULL'
+====
+---- QUERY
+# Regression test for IMPALA-1339
+select c60 from test_char_nulls group by c60;
+---- TYPES
+char
+---- RESULTS
+'NULL'
+====
+---- QUERY
+# Regression test for IMPALA-1339
+select c80 from test_char_nulls group by c80;
+---- TYPES
+char
+---- RESULTS
+'NULL'
+====
+---- QUERY
+# Regression test for IMPALA-1339
+select c81 from test_char_nulls group by c81;
+---- TYPES
+char
+---- RESULTS
+'NULL'
+====
+---- QUERY
+# Regression test for IMPALA-1339
+select c82 from test_char_nulls group by c82;
+---- TYPES
+char
+---- RESULTS
+'NULL'
+====
+---- QUERY
+# Regression test for IMPALA-1339
+select c100 from test_char_nulls group by c100;
+---- TYPES
+char
+---- RESULTS
+'NULL'
+====
+---- QUERY
+# Regression test for IMPALA-1339
+select c120 from test_char_nulls group by c120;
+---- TYPES
+char
+---- RESULTS
+'NULL'
+====
+---- QUERY
+# Regression test for IMPALA-1339
+select c140 from test_char_nulls group by c140;
+---- TYPES
+char
+---- RESULTS
+'NULL'
+====
http://git-wip-us.apache.org/repos/asf/impala/blob/579e3320/testdata/workloads/functional-query/queries/QueryTest/chars.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/chars.test b/testdata/workloads/functional-query/queries/QueryTest/chars.test
index cd1519e..cd915ce 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/chars.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/chars.test
@@ -1,122 +1,10 @@
====
---- QUERY
-# TODO: string literals should start as CHAR(N) and analysis
-# should promote as necessary
-insert into test_char_tmp select cast("hello" as char(5))
-====
----- QUERY
-select * from test_char_tmp
----- TYPES
-char
----- RESULTS
-'hello'
-====
----- QUERY
-# Regression test for IMPALA-1248
-insert into test_char_tmp values (cast("hel" as char(5)))
-====
----- QUERY
-insert into test_char_tmp select cast(cast("hello000" as VARCHAR(8)) as char(5))
-====
----- QUERY
-select * from test_char_tmp where c = cast('hel' as char(5))
----- TYPES
-char
----- RESULTS
-'hel '
-====
----- QUERY
-insert into test_char_tmp values (NULL)
-====
----- QUERY
-select * from test_char_tmp as A CROSS JOIN test_char_tmp as B
-where B.c = cast('hel' as CHAR(5))
-ORDER BY A.c
----- TYPES
-char, char
----- RESULTS
-'hel ','hel '
-'hello','hel '
-'hello','hel '
-'NULL','hel '
-====
----- QUERY
-select * from test_char_tmp as A, test_char_tmp as B
-where A.c = B.c AND A.c != 'hello'
----- TYPES
-char, char
----- RESULTS
-'hel ','hel '
-====
----- QUERY
-select lower(c) from test_char_tmp ORDER BY c
----- TYPES
-string
----- RESULTS
-'hel '
-'hello'
-'hello'
-'NULL'
-====
----- QUERY
-insert into test_varchar_tmp values (cast("hello" as varchar(5)))
-====
----- QUERY
-select * from test_varchar_tmp
----- TYPES
-string
----- RESULTS
-'hello'
-====
----- QUERY
-insert into test_varchar_tmp values (cast("xyzzzzz12" as varchar(7)))
----- CATCH
-would need to be cast to VARCHAR(5)
-====
----- QUERY
-select cast("xyzzzzz12" as varchar(-1))
----- CATCH
-Syntax error
-====
----- QUERY
select (cast("xyzzzzz12" as char(-1)))
---- CATCH
Syntax error
====
---- QUERY
-insert into test_varchar_tmp values (cast("hel" as varchar(4)))
-====
----- QUERY
-select * from test_varchar_tmp
----- TYPES
-string
----- RESULTS
-'hello'
-'hel'
-====
----- QUERY
-insert into allchars values (cast("123456" as char(5)), cast("123456" as char(140)),
-cast("123456" as varchar(5)))
-====
----- QUERY
-select cshort, clong, vc from allchars
----- TYPES
-char,char,string
----- RESULTS
-'12345','123456 ','12345'
-====
----- QUERY
-insert into allchars_par values (cast("123456" as char(5)), cast("123456" as char(140)),
-cast("123456" as varchar(5)))
-====
----- QUERY
-select cshort, clong, vc from allchars_par
----- TYPES
-char,char,string
----- RESULTS
-'12345','123456 ','12345'
-====
----- QUERY
select count(*), count(cs), count(cl), count(vc) from chars_tiny
---- TYPES
bigint,bigint,bigint,bigint
@@ -160,14 +48,14 @@ bigint
1
====
---- QUERY
-select cs, count(cl) from functional.chars_tiny group by cs having count(vc) > 1
+select cs, count(cl) from chars_tiny group by cs having count(vc) > 1
---- TYPES
char, bigint
---- RESULTS
'6a ',2
====
---- QUERY
-select A.cs from functional.chars_tiny as A, functional.chars_tiny as B where
+select A.cs from chars_tiny as A, chars_tiny as B where
cast(A.cs as char(1)) = cast(B.cl as char(1)) order by A.cs
---- TYPES
char
@@ -183,47 +71,8 @@ char
'6a '
====
---- QUERY
-drop table if exists char_parts
-====
----- QUERY
-create table if not exists char_parts (vc varchar(32)) partitioned by
-(csp char(5), clp char(140), vcp varchar(32))
-====
----- QUERY
-insert into char_parts (csp, clp, vcp, vc) select cs, cl, vc, vc from chars_tiny
-====
----- QUERY
-select csp, clp, vcp from char_parts where csp != cast('dne' as char(5)) order by csp
----- TYPES
-char, char, string
----- RESULTS
-'1aaaa','1bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb','1cccc'
-'2aaaa','2bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb','2cccccc'
-'3aaa ','3bbbbb ','3ccc'
-'4aa ','4bbbb ','4cc'
-'5a ','5bbb ','5c'
-'6a ','6b ','6c'
-'6a ','6b ','6c'
-'a ','b ','c'
-====
----- QUERY
-insert into char_parts partition (csp=cast('foo' as char(5)),
-clp=cast('01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789xxxxxxx' as char(140)),
-vcp=cast('myvar' as varchar(32))) select cast('val' as varchar(32));
-====
----- QUERY
-select csp, clp, vcp from char_parts where csp = cast('foo' as char(5))
----- TYPES
-char, char, string
----- RESULTS
-'foo ','01234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789','myvar'
-====
----- QUERY
-drop table if exists char_parts
-====
----- QUERY
# Regression test for IMPALA-1316
-select A.vc from functional.chars_tiny as A join functional.chars_tiny using (vc) order by A.vc
+select A.vc from chars_tiny as A join chars_tiny using (vc) order by A.vc
---- TYPES
string
---- RESULTS
@@ -240,7 +89,7 @@ string
====
---- QUERY
# Regression test for IMPALA-1322
-select count(*) from functional.chars_tiny as A, functional.chars_tiny as B
+select count(*) from chars_tiny as A, chars_tiny as B
where cast(A.cs as CHAR(1)) = cast(B.vc as CHAR(1));
---- TYPES
bigint
@@ -249,40 +98,13 @@ bigint
====
---- QUERY
select min(cs), max(vc), ndv(cl), ndv(vc), appx_median(cs), appx_median(vc)
-from functional.chars_tiny
+from chars_tiny
---- TYPES
string, string, bigint, bigint, string, string
---- RESULTS
'1aaaa','c',7,7,'5a ','5c'
====
---- QUERY
-# Regression test for IMPALA-1322
-drop table if exists functional.t_1822;
-====
----- QUERY
-# Regression test for IMPALA-1322
-create table functional.t_1822 (c10 char(10), c100 char(100), v100 varchar(100), v200 varchar(200), s string);
-====
----- QUERY
-# Regression test for IMPALA-1322
-insert into functional.t_1822 values (cast('a' as char(1)), cast('a' as char(1)),
-cast('a' as varchar(1)), cast('a' as varchar(1)), 'a');
-====
----- QUERY
-# Regression test for IMPALA-1316
-select count(*) from functional.t_1822 as t join functional.t_1822 as tt
-on cast(tt.s as char(129)) = t.c10 and
-cast(tt.s as char(129)) = t.c100 and tt.c10 = t.c100;
----- TYPES
-bigint
----- RESULTS
-1
-====
----- QUERY
-# Regression test for IMPALA-1322
-drop table if exists functional.t_1822;
-====
----- QUERY
# Regression test for IMPALA-1316
select t1.vc, COUNT(1) FROM chars_tiny t1 GROUP BY 1 ORDER BY t1.vc
---- TYPES
@@ -313,81 +135,9 @@ char, bigint
'NULL',1
====
---- QUERY
-# Regression test for IMPALA-1339
-select c20 from test_char_nulls group by c20;
----- TYPES
-char
----- RESULTS
-'NULL'
-====
----- QUERY
-# Regression test for IMPALA-1339
-select c40 from test_char_nulls group by c40;
----- TYPES
-char
----- RESULTS
-'NULL'
-====
----- QUERY
-# Regression test for IMPALA-1339
-select c60 from test_char_nulls group by c60;
----- TYPES
-char
----- RESULTS
-'NULL'
-====
----- QUERY
-# Regression test for IMPALA-1339
-select c80 from test_char_nulls group by c80;
----- TYPES
-char
----- RESULTS
-'NULL'
-====
----- QUERY
-# Regression test for IMPALA-1339
-select c81 from test_char_nulls group by c81;
----- TYPES
-char
----- RESULTS
-'NULL'
-====
----- QUERY
-# Regression test for IMPALA-1339
-select c82 from test_char_nulls group by c82;
----- TYPES
-char
----- RESULTS
-'NULL'
-====
----- QUERY
-# Regression test for IMPALA-1339
-select c100 from test_char_nulls group by c100;
----- TYPES
-char
----- RESULTS
-'NULL'
-====
----- QUERY
-# Regression test for IMPALA-1339
-select c120 from test_char_nulls group by c120;
----- TYPES
-char
----- RESULTS
-'NULL'
-====
----- QUERY
-# Regression test for IMPALA-1339
-select c140 from test_char_nulls group by c140;
----- TYPES
-char
----- RESULTS
-'NULL'
-====
----- QUERY
# Regression test for IMPALA-1344
select cs, LAST_VALUE(cs) OVER (ORDER BY cs rows between unbounded preceding and
-current row) FROM functional.chars_tiny;
+current row) FROM chars_tiny;
---- TYPES
char, string
---- RESULTS
http://git-wip-us.apache.org/repos/asf/impala/blob/579e3320/tests/query_test/test_chars.py
----------------------------------------------------------------------
diff --git a/tests/query_test/test_chars.py b/tests/query_test/test_chars.py
index ec06da8..eaa744a 100644
--- a/tests/query_test/test_chars.py
+++ b/tests/query_test/test_chars.py
@@ -25,49 +25,6 @@ class TestStringQueries(ImpalaTestSuite):
def get_workload(cls):
return 'functional-query'
- def setup_method(self, method):
- self.__cleanup_char_tables()
- self.__create_char_tables()
-
- def teardown_method(self, method):
- self.__cleanup_char_tables()
-
- def __cleanup_char_tables(self):
- self.client.execute('drop table if exists functional.test_char_tmp');
- self.client.execute('drop table if exists functional.test_varchar_tmp');
- self.client.execute('drop table if exists functional.allchars');
- self.client.execute('drop table if exists functional.allchars_par');
- self.client.execute('drop table if exists functional.test_char_nulls');
-
- def __create_char_tables(self):
- self.client.execute(
- 'create table if not exists ' +
- 'functional.test_varchar_tmp (vc varchar(5))')
- self.client.execute(
- 'create table if not exists functional.test_char_tmp (c char(5))')
- self.client.execute(
- 'create table if not exists functional.allchars ' +
- '(cshort char(5), clong char(140), vc varchar(5))')
- self.client.execute(
- 'create table if not exists functional.allchars_par ' +
- '(cshort char(5), clong char(140), vc varchar(5)) stored as parquet')
-
- # Regression test for IMPALA-1339
- self.client.execute('create table if not exists ' +
- '''functional.test_char_nulls ( c20 char(20),
- c40 char(40),
- c60 char(60),
- c80 char(80),
- c81 char(81),
- c82 char(82),
- c100 char(100),
- c120 char(120),
- c140 char(140))''')
- self.client.execute('insert into functional.test_char_nulls ' +
- 'values (NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL)');
- self.client.execute('insert into functional.test_char_nulls ' +
- 'values (NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL)');
-
@classmethod
def add_test_dimensions(cls):
super(TestStringQueries, cls).add_test_dimensions()
@@ -77,10 +34,13 @@ class TestStringQueries(ImpalaTestSuite):
v.get_value('table_format').file_format in ['text'] and
v.get_value('table_format').compression_codec in ['none'])
- @pytest.mark.execute_serially
- def test_varchar(self, vector):
+ def test_chars(self, vector):
self.run_test_case('QueryTest/chars', vector)
+ def test_chars_tmp_tables(self, vector, unique_database):
+ # Tests that create temporary tables and require a unique database.
+ self.run_test_case('QueryTest/chars-tmp-tables', vector, unique_database)
+
class TestCharFormats(ImpalaTestSuite):
@classmethod
def get_workload(cls):
[6/8] impala git commit: IMPALA-6268:
KerberosOnAndOff/RpcMgrKerberizedTest.MultipleServices failing
Posted by jr...@apache.org.
IMPALA-6268: KerberosOnAndOff/RpcMgrKerberizedTest.MultipleServices failing
On systems that have Kerberos 1.11 or earlier, service principals with
IP addresses are not supported due to a bug:
http://krbdev.mit.edu/rt/Ticket/Display.html?id=7603
Since our BE tests use such principals, they fail on older platforms with the
above mentioned kerberos versions.
Kudu fixed this by adding a workaround which overrides krb5_realm_override.
https://github.com/cloudera/kudu/commit/ba2ae3de4a7c43ff2f5873e822410e066ea99667
However, when we moved Kudu's security library into Impala, we did not
add the appropriate build flags that allow it to be used. This patch fixes
that.
Testing: Verified that the failing test runs successfully on CentOs 6.4
with Kerberos 1.10.3
Change-Id: I60e291e8aa1b59b645b856d33c658471f314c221
Reviewed-on: http://gerrit.cloudera.org:8080/9006
Reviewed-by: Sailesh Mukil <sa...@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/d8ae8801
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/d8ae8801
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/d8ae8801
Branch: refs/heads/2.x
Commit: d8ae8801ae668f6ba4771c5794b80f7c9262cd65
Parents: e714f2b
Author: Sailesh Mukil <sa...@apache.org>
Authored: Tue Jan 9 14:58:38 2018 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Fri Jan 19 01:21:45 2018 +0000
----------------------------------------------------------------------
CMakeLists.txt | 5 +++++
be/src/rpc/CMakeLists.txt | 1 +
be/src/rpc/rpc-mgr-test.cc | 4 +++-
3 files changed, 9 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/d8ae8801/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 37a6324..612e00c 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -307,6 +307,11 @@ IMPALA_ADD_THIRDPARTY_LIB(krb5 ${KERBEROS_INCLUDE_DIR} "" ${KERBEROS_LIBRARY})
# testing.
find_package(KerberosPrograms REQUIRED)
+# Tests that run any security related tests need to link this in to override the
+# krb5_realm_override() implementation in krb5.
+# See be/src/kudu/security/krb5_realm_override.cc for more information.
+set(KRB5_REALM_OVERRIDE -Wl,--undefined=krb5_realm_override_loaded krb5_realm_override)
+
###################################################################
# System dependencies
http://git-wip-us.apache.org/repos/asf/impala/blob/d8ae8801/be/src/rpc/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/be/src/rpc/CMakeLists.txt b/be/src/rpc/CMakeLists.txt
index 4234e2b..7beb80d 100644
--- a/be/src/rpc/CMakeLists.txt
+++ b/be/src/rpc/CMakeLists.txt
@@ -50,6 +50,7 @@ ADD_BE_TEST(rpc-mgr-test)
add_dependencies(rpc-mgr-test rpc_test_proto)
target_link_libraries(rpc-mgr-test rpc_test_proto)
target_link_libraries(rpc-mgr-test security-test-for-impala)
+target_link_libraries(rpc-mgr-test ${KRB5_REALM_OVERRIDE})
add_library(rpc_test_proto ${RPC_TEST_PROTO_SRCS})
add_dependencies(rpc_test_proto rpc_test_proto_tgt krpc)
http://git-wip-us.apache.org/repos/asf/impala/blob/d8ae8801/be/src/rpc/rpc-mgr-test.cc
----------------------------------------------------------------------
diff --git a/be/src/rpc/rpc-mgr-test.cc b/be/src/rpc/rpc-mgr-test.cc
index 7e3cb25..441619b 100644
--- a/be/src/rpc/rpc-mgr-test.cc
+++ b/be/src/rpc/rpc-mgr-test.cc
@@ -195,7 +195,9 @@ class ScanMemServiceImpl : public ScanMemServiceIf {
// Reenable after fixing.
INSTANTIATE_TEST_CASE_P(KerberosOnAndOff,
RpcMgrKerberizedTest,
- ::testing::Values(KERBEROS_OFF));
+ ::testing::Values(KERBEROS_OFF,
+ USE_KUDU_KERBEROS,
+ USE_IMPALA_KERBEROS));
TEST_P(RpcMgrKerberizedTest, MultipleServices) {
// Test that a service can be started, and will respond to requests.
[2/8] impala git commit: [DOCS] Minor editorial change
Posted by jr...@apache.org.
[DOCS] Minor editorial change
Turn "royal we" into imperative statement.
Change-Id: Ib78e851761796a1751e6adaaffa049b1fbb58b88
Reviewed-on: http://gerrit.cloudera.org:8080/9064
Reviewed-by: Alex Rodoni <ar...@cloudera.com>
Reviewed-by: John Russell <jr...@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/ca7d03cf
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/ca7d03cf
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/ca7d03cf
Branch: refs/heads/2.x
Commit: ca7d03cfe94bd4961587b918ba6ef6bdb91bb2a6
Parents: f5d73f5
Author: John Russell <jr...@cloudera.com>
Authored: Thu Jan 18 13:18:52 2018 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Thu Jan 18 21:53:40 2018 +0000
----------------------------------------------------------------------
docs/topics/impala_union.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/ca7d03cf/docs/topics/impala_union.xml
----------------------------------------------------------------------
diff --git a/docs/topics/impala_union.xml b/docs/topics/impala_union.xml
index 2fbba3e..8fd5fd0 100644
--- a/docs/topics/impala_union.xml
+++ b/docs/topics/impala_union.xml
@@ -67,7 +67,7 @@ all results from both queries, even if there are duplicates.
<p conref="../shared/impala_common.xml#common/example_blurb"/>
<p>
- First, we set up some sample data, including duplicate <codeph>1</codeph> values.
+ First, set up some sample data, including duplicate <codeph>1</codeph> values:
</p>
<codeblock rev="obwl">[localhost:21000] > create table few_ints (x int);
[7/8] impala git commit: IMPALA-4886: Expose table metrics in the
catalog web UI.
Posted by jr...@apache.org.
IMPALA-4886: Expose table metrics in the catalog web UI.
The following changes are included in this commit:
* Adds a lightweight framework for registering metrics in the JVM.
* Adds table-level metrics and enables these metrics to be exposed
through the catalog web UI.
* Adds a CatalogUsageMonitor class that monitors and reports the catalog
usage in terms of the tables with the highest memory requirements and
the tables with the highest number of metadata operations. The catalog
usage information is exposed in the /catalog page of the catalog web UI.
Change-Id: I37d407979e6d3b1a444b6b6265900b148facde9e
Reviewed-on: http://gerrit.cloudera.org:8080/8529
Reviewed-by: Dimitris Tsirogiannis <dt...@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/3f00d10e
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/3f00d10e
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/3f00d10e
Branch: refs/heads/2.x
Commit: 3f00d10e1b6c5785990c1a73835f82e3821f839e
Parents: d8ae880
Author: Dimitris Tsirogiannis <dt...@cloudera.com>
Authored: Thu Oct 12 16:27:20 2017 -0700
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Fri Jan 19 09:25:01 2018 +0000
----------------------------------------------------------------------
be/src/catalog/catalog-server.cc | 115 +++++++-
be/src/catalog/catalog-server.h | 29 ++
be/src/catalog/catalog.cc | 16 ++
be/src/catalog/catalog.h | 11 +
common/thrift/CatalogObjects.thrift | 11 +
common/thrift/Frontend.thrift | 11 +
common/thrift/JniCatalog.thrift | 22 ++
fe/pom.xml | 6 +
.../java/org/apache/impala/catalog/Catalog.java | 7 +-
.../impala/catalog/CatalogServiceCatalog.java | 62 ++++-
.../impala/catalog/CatalogUsageMonitor.java | 72 +++++
.../org/apache/impala/catalog/HBaseTable.java | 6 +
.../apache/impala/catalog/HdfsPartition.java | 8 +-
.../org/apache/impala/catalog/HdfsTable.java | 262 ++++++++++++++-----
.../org/apache/impala/catalog/KuduTable.java | 51 ++--
.../java/org/apache/impala/catalog/Table.java | 59 +++++
.../java/org/apache/impala/common/Metrics.java | 149 +++++++++++
.../impala/service/CatalogOpExecutor.java | 14 +-
.../org/apache/impala/service/JniCatalog.java | 19 ++
.../java/org/apache/impala/util/TopNCache.java | 108 ++++++++
.../org/apache/impala/util/TestTopNCache.java | 130 +++++++++
tests/webserver/test_web_pages.py | 9 +
www/catalog.tmpl | 117 ++++++++-
www/table_metrics.tmpl | 23 ++
24 files changed, 1205 insertions(+), 112 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/be/src/catalog/catalog-server.cc
----------------------------------------------------------------------
diff --git a/be/src/catalog/catalog-server.cc b/be/src/catalog/catalog-server.cc
index b004b22..4bf26c0 100644
--- a/be/src/catalog/catalog-server.cc
+++ b/be/src/catalog/catalog-server.cc
@@ -55,6 +55,8 @@ const string CATALOG_WEB_PAGE = "/catalog";
const string CATALOG_TEMPLATE = "catalog.tmpl";
const string CATALOG_OBJECT_WEB_PAGE = "/catalog_object";
const string CATALOG_OBJECT_TEMPLATE = "catalog_object.tmpl";
+const string TABLE_METRICS_WEB_PAGE = "/table_metrics";
+const string TABLE_METRICS_TEMPLATE = "table_metrics.tmpl";
// Implementation for the CatalogService thrift interface.
class CatalogServiceThriftIf : public CatalogServiceIf {
@@ -200,16 +202,14 @@ Status CatalogServer::Start() {
}
void CatalogServer::RegisterWebpages(Webserver* webserver) {
- Webserver::UrlCallback catalog_callback =
- bind<void>(mem_fn(&CatalogServer::CatalogUrlCallback), this, _1, _2);
webserver->RegisterUrlCallback(CATALOG_WEB_PAGE, CATALOG_TEMPLATE,
- catalog_callback);
-
- Webserver::UrlCallback catalog_objects_callback =
- bind<void>(mem_fn(&CatalogServer::CatalogObjectsUrlCallback), this, _1, _2);
+ [this](const auto& args, auto* doc) { this->CatalogUrlCallback(args, doc); });
webserver->RegisterUrlCallback(CATALOG_OBJECT_WEB_PAGE, CATALOG_OBJECT_TEMPLATE,
- catalog_objects_callback, false);
-
+ [this](const auto& args, auto* doc) { this->CatalogObjectsUrlCallback(args, doc); },
+ false);
+ webserver->RegisterUrlCallback(TABLE_METRICS_WEB_PAGE, TABLE_METRICS_TEMPLATE,
+ [this](const auto& args, auto* doc) { this->TableMetricsUrlCallback(args, doc); },
+ false);
RegisterLogLevelCallbacks(webserver, true);
}
@@ -335,11 +335,12 @@ void CatalogServer::BuildTopicUpdates(const vector<TCatalogObject>& catalog_obje
void CatalogServer::CatalogUrlCallback(const Webserver::ArgumentMap& args,
Document* document) {
+ GetCatalogUsage(document);
TGetDbsResult get_dbs_result;
Status status = catalog_->GetDbs(NULL, &get_dbs_result);
if (!status.ok()) {
Value error(status.GetDetail().c_str(), document->GetAllocator());
- document->AddMember("error", error, document->GetAllocator());
+ document->AddMember("error", error, document->GetAllocator());
return;
}
Value databases(kArrayType);
@@ -364,15 +365,76 @@ void CatalogServer::CatalogUrlCallback(const Webserver::ArgumentMap& args,
table_obj.AddMember("fqtn", fq_name, document->GetAllocator());
Value table_name(table.c_str(), document->GetAllocator());
table_obj.AddMember("name", table_name, document->GetAllocator());
+ Value has_metrics;
+ has_metrics.SetBool(true);
+ table_obj.AddMember("has_metrics", has_metrics, document->GetAllocator());
table_array.PushBack(table_obj, document->GetAllocator());
}
database.AddMember("num_tables", table_array.Size(), document->GetAllocator());
database.AddMember("tables", table_array, document->GetAllocator());
+ Value has_metrics;
+ has_metrics.SetBool(true);
+ database.AddMember("has_metrics", has_metrics, document->GetAllocator());
databases.PushBack(database, document->GetAllocator());
}
document->AddMember("databases", databases, document->GetAllocator());
}
+void CatalogServer::GetCatalogUsage(Document* document) {
+ TGetCatalogUsageResponse catalog_usage_result;
+ Status status = catalog_->GetCatalogUsage(&catalog_usage_result);
+ if (!status.ok()) {
+ Value error(status.GetDetail().c_str(), document->GetAllocator());
+ document->AddMember("error", error, document->GetAllocator());
+ return;
+ }
+ // Collect information about the largest tables in terms of memory requirements
+ Value large_tables(kArrayType);
+ for (int i = 0; i < catalog_usage_result.large_tables.size(); ++i) {
+ Value tbl_obj(kObjectType);
+ const auto& large_table = catalog_usage_result.large_tables[i];
+ Value tbl_name(Substitute("$0.$1", large_table.table_name.db_name,
+ large_table.table_name.table_name).c_str(), document->GetAllocator());
+ tbl_obj.AddMember("name", tbl_name, document->GetAllocator());
+ DCHECK(large_table.__isset.memory_estimate_bytes);
+ Value memory_estimate(PrettyPrinter::Print(large_table.memory_estimate_bytes,
+ TUnit::BYTES).c_str(), document->GetAllocator());
+ tbl_obj.AddMember("mem_estimate", memory_estimate, document->GetAllocator());
+ large_tables.PushBack(tbl_obj, document->GetAllocator());
+ }
+ Value has_large_tables;
+ has_large_tables.SetBool(true);
+ document->AddMember("has_large_tables", has_large_tables, document->GetAllocator());
+ document->AddMember("large_tables", large_tables, document->GetAllocator());
+ Value num_large_tables;
+ num_large_tables.SetInt(catalog_usage_result.large_tables.size());
+ document->AddMember("num_large_tables", num_large_tables, document->GetAllocator());
+
+ // Collect information about the most frequently accessed tables.
+ Value frequent_tables(kArrayType);
+ for (int i = 0; i < catalog_usage_result.frequently_accessed_tables.size(); ++i) {
+ Value tbl_obj(kObjectType);
+ const auto& frequent_table = catalog_usage_result.frequently_accessed_tables[i];
+ Value tbl_name(Substitute("$0.$1", frequent_table.table_name.db_name,
+ frequent_table.table_name.table_name).c_str(), document->GetAllocator());
+ tbl_obj.AddMember("name", tbl_name, document->GetAllocator());
+ Value num_metadata_operations;
+ DCHECK(frequent_table.__isset.num_metadata_operations);
+ num_metadata_operations.SetInt64(frequent_table.num_metadata_operations);
+ tbl_obj.AddMember("num_metadata_ops", num_metadata_operations,
+ document->GetAllocator());
+ frequent_tables.PushBack(tbl_obj, document->GetAllocator());
+ }
+ Value has_frequent_tables;
+ has_frequent_tables.SetBool(true);
+ document->AddMember("has_frequent_tables", has_frequent_tables,
+ document->GetAllocator());
+ document->AddMember("frequent_tables", frequent_tables, document->GetAllocator());
+ Value num_frequent_tables;
+ num_frequent_tables.SetInt(catalog_usage_result.frequently_accessed_tables.size());
+ document->AddMember("num_frequent_tables", num_frequent_tables,
+ document->GetAllocator());
+}
void CatalogServer::CatalogObjectsUrlCallback(const Webserver::ArgumentMap& args,
Document* document) {
@@ -384,7 +446,8 @@ void CatalogServer::CatalogObjectsUrlCallback(const Webserver::ArgumentMap& args
// Get the object type and name from the topic entry key
TCatalogObject request;
- Status status = TCatalogObjectFromObjectName(object_type, object_name_arg->second, &request);
+ Status status =
+ TCatalogObjectFromObjectName(object_type, object_name_arg->second, &request);
// Get the object and dump its contents.
TCatalogObject result;
@@ -402,3 +465,35 @@ void CatalogServer::CatalogObjectsUrlCallback(const Webserver::ArgumentMap& args
document->AddMember("error", error, document->GetAllocator());
}
}
+
+void CatalogServer::TableMetricsUrlCallback(const Webserver::ArgumentMap& args,
+ Document* document) {
+ // TODO: Enable json view of table metrics
+ Webserver::ArgumentMap::const_iterator object_name_arg = args.find("name");
+ if (object_name_arg != args.end()) {
+ // Parse the object name to extract database and table names
+ const string& full_tbl_name = object_name_arg->second;
+ int pos = full_tbl_name.find(".");
+ if (pos == string::npos || pos >= full_tbl_name.size() - 1) {
+ stringstream error_msg;
+ error_msg << "Invalid table name: " << full_tbl_name;
+ Value error(error_msg.str().c_str(), document->GetAllocator());
+ document->AddMember("error", error, document->GetAllocator());
+ return;
+ }
+ string metrics;
+ Status status = catalog_->GetTableMetrics(
+ full_tbl_name.substr(0, pos), full_tbl_name.substr(pos + 1), &metrics);
+ if (status.ok()) {
+ Value metrics_str(metrics.c_str(), document->GetAllocator());
+ document->AddMember("table_metrics", metrics_str, document->GetAllocator());
+ } else {
+ Value error(status.GetDetail().c_str(), document->GetAllocator());
+ document->AddMember("error", error, document->GetAllocator());
+ }
+ } else {
+ Value error("Please specify the value of the name parameter.",
+ document->GetAllocator());
+ document->AddMember("error", error, document->GetAllocator());
+ }
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/be/src/catalog/catalog-server.h
----------------------------------------------------------------------
diff --git a/be/src/catalog/catalog-server.h b/be/src/catalog/catalog-server.h
index 78a3f20..0b6b220 100644
--- a/be/src/catalog/catalog-server.h
+++ b/be/src/catalog/catalog-server.h
@@ -185,6 +185,35 @@ class CatalogServer {
/// <host>:25020/catalog_objects?object_type=TABLE&object_name=foo.bar
void CatalogObjectsUrlCallback(const Webserver::ArgumentMap& args,
rapidjson::Document* document);
+
+ /// Retrieves from the FE information about the current catalog usage and populates
+ /// the /catalog debug webpage. The catalog usage includes information about the TOP-N
+ /// frequently used (in terms of number of metadata operations) tables as well as the
+ /// TOP-N tables with the highest memory requirements.
+ ///
+ /// Example output:
+ /// "large_tables": [
+ /// {
+ /// "name": "functional.alltypesagg",
+ /// "mem_estimate": 212434233
+ /// }
+ /// ]
+ /// "frequent_tables": [
+ /// {
+ /// "name": "functional.alltypestiny",
+ /// "frequency": 10
+ /// }
+ /// ]
+ void GetCatalogUsage(rapidjson::Document* document);
+
+ /// Debug webpage handler that is used to dump all the registered metrics of a
+ /// table. The caller specifies the "name" parameter which is the fully
+ /// qualified table name and this function retrieves all the metrics of that
+ /// table. For example, to get the table metrics of table "bar" in database
+ /// "foo":
+ /// <host>:25020/table_metrics?name=foo.bar
+ void TableMetricsUrlCallback(const Webserver::ArgumentMap& args,
+ rapidjson::Document* document);
};
}
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/be/src/catalog/catalog.cc
----------------------------------------------------------------------
diff --git a/be/src/catalog/catalog.cc b/be/src/catalog/catalog.cc
index b6dd86a..d96d23e 100644
--- a/be/src/catalog/catalog.cc
+++ b/be/src/catalog/catalog.cc
@@ -58,11 +58,13 @@ Catalog::Catalog() {
{"execDdl", "([B)[B", &exec_ddl_id_},
{"resetMetadata", "([B)[B", &reset_metadata_id_},
{"getTableNames", "([B)[B", &get_table_names_id_},
+ {"getTableMetrics", "([B)Ljava/lang/String;", &get_table_metrics_id_},
{"getDbs", "([B)[B", &get_dbs_id_},
{"getFunctions", "([B)[B", &get_functions_id_},
{"checkUserSentryAdmin", "([B)V", &sentry_admin_check_id_},
{"getCatalogObject", "([B)[B", &get_catalog_object_id_},
{"getCatalogDelta", "([B)[B", &get_catalog_delta_id_},
+ {"getCatalogUsage", "()[B", &get_catalog_usage_id_},
{"getCatalogVersion", "()J", &get_catalog_version_id_},
{"prioritizeLoad", "([B)V", &prioritize_load_id_}};
@@ -131,6 +133,20 @@ Status Catalog::GetTableNames(const string& db, const string* pattern,
return JniUtil::CallJniMethod(catalog_, get_table_names_id_, params, table_names);
}
+Status Catalog::GetTableMetrics(const string& db, const string& tbl,
+ string* table_metrics) {
+ TGetTableMetricsParams params;
+ TTableName tblName;
+ tblName.__set_db_name(db);
+ tblName.__set_table_name(tbl);
+ params.__set_table_name(tblName);
+ return JniUtil::CallJniMethod(catalog_, get_table_metrics_id_, params, table_metrics);
+}
+
+Status Catalog::GetCatalogUsage(TGetCatalogUsageResponse* response) {
+ return JniUtil::CallJniMethod(catalog_, get_catalog_usage_id_, response);
+}
+
Status Catalog::GetFunctions(const TGetFunctionsRequest& request,
TGetFunctionsResponse *response) {
return JniUtil::CallJniMethod(catalog_, get_functions_id_, request, response);
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/be/src/catalog/catalog.h
----------------------------------------------------------------------
diff --git a/be/src/catalog/catalog.h b/be/src/catalog/catalog.h
index 3119d60..13e4529 100644
--- a/be/src/catalog/catalog.h
+++ b/be/src/catalog/catalog.h
@@ -84,6 +84,15 @@ class Catalog {
Status GetTableNames(const std::string& db, const std::string* pattern,
TGetTablesResult* table_names);
+ /// Returns the collected metrics of a table. The response contains a
+ /// pretty-printed string representation of table metrics.
+ Status GetTableMetrics(const std::string& db, const std::string& tbl,
+ std::string* metrics);
+
+ /// Returns the current catalog usage that includes the most frequently accessed
+ /// tables as well as the tables with the highest memory requirements.
+ Status GetCatalogUsage(TGetCatalogUsageResponse* response);
+
/// Gets all functions in the catalog matching the parameters in the given
/// TFunctionsRequest.
Status GetFunctions(const TGetFunctionsRequest& request,
@@ -109,8 +118,10 @@ class Catalog {
jmethodID get_catalog_object_id_; // JniCatalog.getCatalogObject()
jmethodID get_catalog_delta_id_; // JniCatalog.getCatalogDelta()
jmethodID get_catalog_version_id_; // JniCatalog.getCatalogVersion()
+ jmethodID get_catalog_usage_id_; // JniCatalog.getCatalogUsage()
jmethodID get_dbs_id_; // JniCatalog.getDbs()
jmethodID get_table_names_id_; // JniCatalog.getTableNames()
+ jmethodID get_table_metrics_id_; // JniCatalog.getTableMetrics()
jmethodID get_functions_id_; // JniCatalog.getFunctions()
jmethodID prioritize_load_id_; // JniCatalog.prioritizeLoad()
jmethodID sentry_admin_check_id_; // JniCatalog.checkUserSentryAdmin()
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/common/thrift/CatalogObjects.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/CatalogObjects.thrift b/common/thrift/CatalogObjects.thrift
index 7894f75..0f1ea5d 100644
--- a/common/thrift/CatalogObjects.thrift
+++ b/common/thrift/CatalogObjects.thrift
@@ -266,6 +266,17 @@ struct THdfsPartition {
// (key,value) pairs stored in the Hive Metastore.
15: optional map<string, string> hms_parameters
+
+ // The following fields store stats about this partition
+ // which are collected when toThrift() is called.
+ // Total number of blocks in this partition.
+ 16: optional i64 num_blocks
+
+ // Total file size in bytes of this partition.
+ 17: optional i64 total_file_size_bytes
+
+ // True, if this partition has incremental stats
+ 18: optional bool has_incremental_stats
}
struct THdfsTable {
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/common/thrift/Frontend.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/Frontend.thrift b/common/thrift/Frontend.thrift
index f856871..ba21605 100644
--- a/common/thrift/Frontend.thrift
+++ b/common/thrift/Frontend.thrift
@@ -87,6 +87,17 @@ struct TGetTablesResult {
1: list<string> tables
}
+// Arguments to getTableMetrics, which returns the metrics of a specific table.
+struct TGetTableMetricsParams {
+ 1: required CatalogObjects.TTableName table_name
+}
+
+// Response to a getTableMetrics request. The response contains all the collected metrics
+// pretty-printed into a string.
+struct TGetTableMetricsResponse {
+ 1: required string metrics
+}
+
// Arguments to getDbs, which returns a list of dbs that match an optional pattern
struct TGetDbsParams {
// If not set, match every database
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/common/thrift/JniCatalog.thrift
----------------------------------------------------------------------
diff --git a/common/thrift/JniCatalog.thrift b/common/thrift/JniCatalog.thrift
index 939e276..4edf4d2 100644
--- a/common/thrift/JniCatalog.thrift
+++ b/common/thrift/JniCatalog.thrift
@@ -605,3 +605,25 @@ struct TDropFunctionParams {
// the signature generated by the FE should just be plumbed through).
4: optional string signature
}
+
+// Stores metrics of a catalog table.
+struct TTableUsageMetrics {
+ 1: required CatalogObjects.TTableName table_name
+
+ // Estimated memory usage of that table.
+ 2: optional i64 memory_estimate_bytes
+
+ // Number of metadata operations performed on the table since it was loaded.
+ 3: optional i64 num_metadata_operations
+}
+
+// Response to a GetCatalogUsage request.
+struct TGetCatalogUsageResponse{
+ // List of the largest (in terms of memory requirements) tables.
+ 1: required list<TTableUsageMetrics> large_tables
+
+ // List of the most frequently accessed (in terms of number of metadata operations)
+ // tables.
+ 2: required list<TTableUsageMetrics> frequently_accessed_tables
+}
+
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/fe/pom.xml
----------------------------------------------------------------------
diff --git a/fe/pom.xml b/fe/pom.xml
index 646820f..135ec77 100644
--- a/fe/pom.xml
+++ b/fe/pom.xml
@@ -312,6 +312,12 @@ under the License.
<version>1.6.0.1</version>
</dependency>
+ <dependency>
+ <groupId>io.dropwizard.metrics</groupId>
+ <artifactId>metrics-core</artifactId>
+ <version>3.2.2</version>
+ </dependency>
+
</dependencies>
<reporting>
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/fe/src/main/java/org/apache/impala/catalog/Catalog.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/Catalog.java b/fe/src/main/java/org/apache/impala/catalog/Catalog.java
index 9ed8133..4548c2b 100644
--- a/fe/src/main/java/org/apache/impala/catalog/Catalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/Catalog.java
@@ -35,6 +35,7 @@ import org.apache.impala.thrift.TPartitionKeyValue;
import org.apache.impala.thrift.TTable;
import org.apache.impala.thrift.TTableName;
import org.apache.impala.util.PatternMatcher;
+
import org.apache.log4j.Logger;
/**
@@ -167,7 +168,11 @@ public abstract class Catalog {
// Remove the old table name from the cache and add the new table.
Db db = getDb(tableName.getDb_name());
if (db == null) return null;
- return db.removeTable(tableName.getTable_name());
+ Table tbl = db.removeTable(tableName.getTable_name());
+ if (tbl != null && !tbl.isStoredInImpaladCatalogCache()) {
+ CatalogUsageMonitor.INSTANCE.removeTable(tbl);
+ }
+ return tbl;
}
/**
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
index f75b0a8..8f75a16 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
@@ -63,10 +63,12 @@ import org.apache.impala.thrift.TCatalogObjectType;
import org.apache.impala.thrift.TCatalogUpdateResult;
import org.apache.impala.thrift.TFunction;
import org.apache.impala.thrift.TGetCatalogDeltaResponse;
+import org.apache.impala.thrift.TGetCatalogUsageResponse;
import org.apache.impala.thrift.TPartitionKeyValue;
import org.apache.impala.thrift.TPrivilege;
import org.apache.impala.thrift.TTable;
import org.apache.impala.thrift.TTableName;
+import org.apache.impala.thrift.TTableUsageMetrics;
import org.apache.impala.thrift.TUniqueId;
import org.apache.impala.util.PatternMatcher;
import org.apache.impala.util.SentryProxy;
@@ -74,6 +76,7 @@ import org.apache.log4j.Logger;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TCompactProtocol;
+import com.codahale.metrics.Timer;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
@@ -1434,11 +1437,12 @@ public class CatalogServiceCatalog extends Catalog {
Preconditions.checkState(!(tbl instanceof IncompleteTable));
String dbName = tbl.getDb().getName();
String tblName = tbl.getName();
-
if (!tryLockTable(tbl)) {
throw new CatalogException(String.format("Error refreshing metadata for table " +
"%s due to lock contention", tbl.getFullName()));
}
+ final Timer.Context context =
+ tbl.getMetrics().getTimer(Table.REFRESH_DURATION_METRIC).time();
try {
long newCatalogVersion = incrementAndGetCatalogVersion();
versionLock_.writeLock().unlock();
@@ -1456,6 +1460,7 @@ public class CatalogServiceCatalog extends Catalog {
LOG.info(String.format("Refreshed table metadata: %s", tbl.getFullName()));
return tbl.toTCatalogObject();
} finally {
+ context.stop();
Preconditions.checkState(!versionLock_.isWriteLockedByCurrentThread());
tbl.getLock().unlock();
}
@@ -1904,4 +1909,59 @@ public class CatalogServiceCatalog extends Catalog {
}
return versionToWaitFor;
}
+
+ /**
+ * Retrieves information about the current catalog usage including the most frequently
+ * accessed tables as well as the tables with the highest memory requirements.
+ */
+ public TGetCatalogUsageResponse getCatalogUsage() {
+ TGetCatalogUsageResponse usage = new TGetCatalogUsageResponse();
+ usage.setLarge_tables(Lists.<TTableUsageMetrics>newArrayList());
+ usage.setFrequently_accessed_tables(Lists.<TTableUsageMetrics>newArrayList());
+ for (Table largeTable: CatalogUsageMonitor.INSTANCE.getLargestTables()) {
+ TTableUsageMetrics tableUsageMetrics =
+ new TTableUsageMetrics(largeTable.getTableName().toThrift());
+ tableUsageMetrics.setMemory_estimate_bytes(largeTable.getEstimatedMetadataSize());
+ usage.addToLarge_tables(tableUsageMetrics);
+ }
+ for (Table frequentTable:
+ CatalogUsageMonitor.INSTANCE.getFrequentlyAccessedTables()) {
+ TTableUsageMetrics tableUsageMetrics =
+ new TTableUsageMetrics(frequentTable.getTableName().toThrift());
+ tableUsageMetrics.setNum_metadata_operations(frequentTable.getMetadataOpsCount());
+ usage.addToFrequently_accessed_tables(tableUsageMetrics);
+ }
+ return usage;
+ }
+
+ /**
+ * Retrieves the stored metrics of the specified table and returns a pretty-printed
+ * string representation. Throws an exception if table metrics were not available
+ * because the table was not loaded or because another concurrent operation was holding
+ * the table lock.
+ */
+ public String getTableMetrics(TTableName tTableName) throws CatalogException {
+ String dbName = tTableName.db_name;
+ String tblName = tTableName.table_name;
+ Table tbl = getTable(dbName, tblName);
+ if (tbl == null) {
+ throw new CatalogException("Table " + dbName + "." + tblName + " was not found.");
+ }
+ String result;
+ if (tbl instanceof IncompleteTable) {
+ result = "No metrics available for table " + dbName + "." + tblName +
+ ". Table not yet loaded.";
+ return result;
+ }
+ if (!tbl.getLock().tryLock()) {
+ result = "Metrics for table " + dbName + "." + tblName + "are not available " +
+ "because the table is currently modified by another operation.";
+ return result;
+ }
+ try {
+ return tbl.getMetrics().toString();
+ } finally {
+ tbl.getLock().unlock();
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/fe/src/main/java/org/apache/impala/catalog/CatalogUsageMonitor.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogUsageMonitor.java b/fe/src/main/java/org/apache/impala/catalog/CatalogUsageMonitor.java
new file mode 100644
index 0000000..a2e8d7e
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogUsageMonitor.java
@@ -0,0 +1,72 @@
+// 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.catalog;
+
+import java.util.List;
+
+import org.apache.impala.util.TopNCache;
+
+import com.google.common.base.Function;
+
+/**
+ * Singleton class that monitors catalog usage. Currently, it tracks the most
+ * frequently accessed tables (in terms of number of metadata operations) as well as
+ * the tables with the highest (estimated) memory requirements. This class is
+ * thread-safe.
+ */
+public final class CatalogUsageMonitor {
+
+ public final static CatalogUsageMonitor INSTANCE = new CatalogUsageMonitor();
+
+ private final TopNCache<Table, Long> frequentlyAccessedTables_;
+
+ private final TopNCache<Table, Long> largestTables_;
+
+ private CatalogUsageMonitor() {
+ final int num_tables_tracked = Integer.getInteger(
+ "org.apache.impala.catalog.CatalogUsageMonitor.NUM_TABLES_TRACKED", 25);
+ frequentlyAccessedTables_ = new TopNCache<Table, Long>(
+ new Function<Table, Long>() {
+ @Override
+ public Long apply(Table tbl) { return tbl.getMetadataOpsCount(); }
+ }, num_tables_tracked, true);
+
+ largestTables_ = new TopNCache<Table, Long>(
+ new Function<Table, Long>() {
+ @Override
+ public Long apply(Table tbl) { return tbl.getEstimatedMetadataSize(); }
+ }, num_tables_tracked, false);
+ }
+
+ public void updateFrequentlyAccessedTables(Table tbl) {
+ frequentlyAccessedTables_.putOrUpdate(tbl);
+ }
+
+ public void updateLargestTables(Table tbl) { largestTables_.putOrUpdate(tbl); }
+
+ public void removeTable(Table tbl) {
+ frequentlyAccessedTables_.remove(tbl);
+ largestTables_.remove(tbl);
+ }
+
+ public List<Table> getFrequentlyAccessedTables() {
+ return frequentlyAccessedTables_.listEntries();
+ }
+
+ public List<Table> getLargestTables() { return largestTables_.listEntries(); }
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/fe/src/main/java/org/apache/impala/catalog/HBaseTable.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/HBaseTable.java b/fe/src/main/java/org/apache/impala/catalog/HBaseTable.java
index 6df7c28..cf36a89 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HBaseTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HBaseTable.java
@@ -64,6 +64,8 @@ import org.apache.impala.thrift.TTableDescriptor;
import org.apache.impala.thrift.TTableType;
import org.apache.impala.util.StatsHelper;
import org.apache.impala.util.TResultRowBuilder;
+
+import com.codahale.metrics.Timer;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
@@ -321,6 +323,8 @@ public class HBaseTable extends Table {
public void load(boolean reuseMetadata, IMetaStoreClient client,
org.apache.hadoop.hive.metastore.api.Table msTbl) throws TableLoadingException {
Preconditions.checkNotNull(getMetaStoreTable());
+ final Timer.Context context =
+ getMetrics().getTimer(Table.LOAD_DURATION_METRIC).time();
try {
msTable_ = msTbl;
hbaseTableName_ = getHBaseTableName(getMetaStoreTable());
@@ -414,6 +418,8 @@ public class HBaseTable extends Table {
} catch (Exception e) {
throw new TableLoadingException("Failed to load metadata for HBase table: " +
name_, e);
+ } finally {
+ context.stop();
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java b/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
index e78ce92..2179346 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
@@ -931,6 +931,7 @@ public class HdfsPartition implements Comparable<HdfsPartition> {
thriftHdfsPart.setAccess_level(accessLevel_);
thriftHdfsPart.setIs_marked_cached(isMarkedCached_);
thriftHdfsPart.setId(getId());
+ thriftHdfsPart.setHas_incremental_stats(hasIncrementalStats());
// IMPALA-4902: Shallow-clone the map to avoid concurrent modifications. One thread
// may try to serialize the returned THdfsPartition after releasing the table's lock,
// and another thread doing DDL may modify the map.
@@ -938,11 +939,16 @@ public class HdfsPartition implements Comparable<HdfsPartition> {
includeIncrementalStats ? hmsParameters_ : getFilteredHmsParameters()));
if (includeFileDesc) {
// Add block location information
+ long numBlocks = 0;
+ long totalFileBytes = 0;
for (FileDescriptor fd: fileDescriptors_) {
thriftHdfsPart.addToFile_desc(fd.toThrift());
+ numBlocks += fd.getNumFileBlocks();
+ totalFileBytes += fd.getFileLength();
}
+ thriftHdfsPart.setNum_blocks(numBlocks);
+ thriftHdfsPart.setTotal_file_size_bytes(totalFileBytes);
}
-
return thriftHdfsPart;
}
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
index 4de25fe..04599f5 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -95,6 +95,9 @@ import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Timer;
+
/**
* Internal representation of table-related metadata of a file-resident table on a
* Hadoop filesystem. The table data can be accessed through libHDFS (which is more of
@@ -124,6 +127,24 @@ public class HdfsTable extends Table {
// Table property key for skip.header.line.count
public static final String TBL_PROP_SKIP_HEADER_LINE_COUNT = "skip.header.line.count";
+ // Average memory requirements (in bytes) for storing the metadata of a partition.
+ private static final long PER_PARTITION_MEM_USAGE_BYTES = 2048;
+
+ // Average memory requirements (in bytes) for storing a file descriptor.
+ private static final long PER_FD_MEM_USAGE_BYTES = 500;
+
+ // Average memory requirements (in bytes) for storing a block.
+ private static final long PER_BLOCK_MEM_USAGE_BYTES = 150;
+
+ // Hdfs table specific metrics
+ public static final String CATALOG_UPDATE_DURATION_METRIC = "catalog-update-duration";
+ public static final String NUM_PARTITIONS_METRIC = "num-partitions";
+ public static final String NUM_FILES_METRIC = "num-files";
+ public static final String NUM_BLOCKS_METRIC = "num-blocks";
+ public static final String TOTAL_FILE_BYTES_METRIC = "total-file-size-bytes";
+ public static final String MEMORY_ESTIMATE_METRIC = "memory-estimate-bytes";
+ public static final String HAS_INCREMENTAL_STATS_METRIC = "has-incremental-stats";
+
// string to indicate NULL. set in load() from table properties
private String nullColumnValue_;
@@ -172,19 +193,14 @@ public class HdfsTable extends Table {
// replicas of the block.
private final ListMap<TNetworkAddress> hostIndex_ = new ListMap<TNetworkAddress>();
- private HdfsPartitionLocationCompressor partitionLocationCompressor_;
-
- // Total number of Hdfs files in this table. Accounted only in the Impalad catalog
- // cache. Set to -1 on Catalogd.
- private long numHdfsFiles_;
-
- // Sum of sizes of all Hdfs files in this table. Accounted only in the Impalad
- // catalog cache. Set to -1 on Catalogd.
- private long totalHdfsBytes_;
+ // True iff this table has incremental stats in any of its partitions.
+ private boolean hasIncrementalStats_ = false;
// True iff the table's partitions are located on more than one filesystem.
private boolean multipleFileSystems_ = false;
+ private HdfsPartitionLocationCompressor partitionLocationCompressor_;
+
// Base Hdfs directory where files of this table are stored.
// For unpartitioned tables it is simply the path where all files live.
// For partitioned tables it is the root directory
@@ -200,6 +216,50 @@ public class HdfsTable extends Table {
// for setAvroSchema().
private boolean isSchemaLoaded_ = false;
+ // Represents a set of storage-related statistics aggregated at the table or partition
+ // level.
+ public final static class FileMetadataStats {
+ // Nuber of files in a table/partition.
+ public long numFiles;
+ // Number of blocks in a table/partition.
+ public long numBlocks;
+ // Total size (in bytes) of all files in a table/partition.
+ public long totalFileBytes;
+
+ // Unsets the storage stats to indicate that their values are unknown.
+ public void unset() {
+ numFiles = -1;
+ numBlocks = -1;
+ totalFileBytes = -1;
+ }
+
+ // Initializes the values of the storage stats.
+ public void init() {
+ numFiles = 0;
+ numBlocks = 0;
+ totalFileBytes = 0;
+ }
+
+ public void set(FileMetadataStats stats) {
+ numFiles = stats.numFiles;
+ numBlocks = stats.numBlocks;
+ totalFileBytes = stats.totalFileBytes;
+ }
+ }
+
+ // Table level storage-related statistics. Depending on whether the table is stored in
+ // the catalog server or the impalad catalog cache, these statistics serve different
+ // purposes and, hence, are managed differently.
+ // Table stored in impalad catalog cache:
+ // - Used in planning.
+ // - Stats are modified real-time by the operations that modify table metadata
+ // (e.g. add partition).
+ // Table stored in the the catalog server:
+ // - Used for reporting through catalog web UI.
+ // - Stats are reset whenever the table is loaded (due to a metadata operation) and
+ // are set when the table is serialized to Thrift.
+ private FileMetadataStats fileMetadataStats_ = new FileMetadataStats();
+
private final static Logger LOG = LoggerFactory.getLogger(HdfsTable.class);
// Caching this configuration object makes calls to getFileSystem much quicker
@@ -311,17 +371,17 @@ public class HdfsTable extends Table {
}
/**
- * Updates numHdfsFiles_ and totalHdfsBytes_ based on the partition information.
+ * Updates the storage stats of this table based on the partition information.
* This is used only for the frontend tests that do not spawn a separate Catalog
* instance.
*/
public void computeHdfsStatsForTesting() {
- Preconditions.checkState(numHdfsFiles_ == -1 && totalHdfsBytes_ == -1);
- numHdfsFiles_ = 0;
- totalHdfsBytes_ = 0;
+ Preconditions.checkState(fileMetadataStats_.numFiles == -1
+ && fileMetadataStats_.totalFileBytes == -1);
+ fileMetadataStats_.init();
for (HdfsPartition partition: partitionMap_.values()) {
- numHdfsFiles_ += partition.getNumFileDescriptors();
- totalHdfsBytes_ += partition.getSize();
+ fileMetadataStats_.numFiles += partition.getNumFileDescriptors();
+ fileMetadataStats_.totalFileBytes += partition.getSize();
}
}
@@ -681,8 +741,7 @@ public class HdfsTable extends Table {
nullPartitionIds_.add(Sets.<Long>newHashSet());
}
}
- numHdfsFiles_ = 0;
- totalHdfsBytes_ = 0;
+ fileMetadataStats_.init();
}
/**
@@ -1023,8 +1082,8 @@ public class HdfsTable extends Table {
}
if (partition.getFileFormat() == HdfsFileFormat.AVRO) hasAvroData_ = true;
partitionMap_.put(partition.getId(), partition);
- totalHdfsBytes_ += partition.getSize();
- numHdfsFiles_ += partition.getNumFileDescriptors();
+ fileMetadataStats_.totalFileBytes += partition.getSize();
+ fileMetadataStats_.numFiles += partition.getNumFileDescriptors();
updatePartitionMdAndColStats(partition);
}
@@ -1078,8 +1137,8 @@ public class HdfsTable extends Table {
*/
private HdfsPartition dropPartition(HdfsPartition partition) {
if (partition == null) return null;
- totalHdfsBytes_ -= partition.getSize();
- numHdfsFiles_ -= partition.getNumFileDescriptors();
+ fileMetadataStats_.totalFileBytes -= partition.getSize();
+ fileMetadataStats_.numFiles -= partition.getNumFileDescriptors();
Preconditions.checkArgument(partition.getPartitionValues().size() ==
numClusteringCols_);
Long partitionId = partition.getId();
@@ -1176,49 +1235,54 @@ public class HdfsTable extends Table {
org.apache.hadoop.hive.metastore.api.Table msTbl,
boolean loadParitionFileMetadata, boolean loadTableSchema,
Set<String> partitionsToUpdate) throws TableLoadingException {
- // turn all exceptions into TableLoadingException
- msTable_ = msTbl;
+ final Timer.Context context =
+ getMetrics().getTimer(Table.LOAD_DURATION_METRIC).time();
try {
- if (loadTableSchema) loadSchema(client, msTbl);
- if (reuseMetadata && getCatalogVersion() == Catalog.INITIAL_CATALOG_VERSION) {
- // This is the special case of CTAS that creates a 'temp' table that does not
- // actually exist in the Hive Metastore.
- initializePartitionMetadata(msTbl);
- setTableStats(msTbl);
- return;
- }
- // Load partition and file metadata
- if (reuseMetadata) {
- // Incrementally update this table's partitions and file metadata
- LOG.info("Incrementally loading table metadata for: " + getFullName());
- Preconditions.checkState(
- partitionsToUpdate == null || loadParitionFileMetadata);
- updateMdFromHmsTable(msTbl);
- if (msTbl.getPartitionKeysSize() == 0) {
- if (loadParitionFileMetadata) updateUnpartitionedTableFileMd();
+ // turn all exceptions into TableLoadingException
+ msTable_ = msTbl;
+ try {
+ if (loadTableSchema) loadSchema(client, msTbl);
+ if (reuseMetadata && getCatalogVersion() == Catalog.INITIAL_CATALOG_VERSION) {
+ // This is the special case of CTAS that creates a 'temp' table that does not
+ // actually exist in the Hive Metastore.
+ initializePartitionMetadata(msTbl);
+ setTableStats(msTbl);
+ return;
+ }
+ // Load partition and file metadata
+ if (reuseMetadata) {
+ // Incrementally update this table's partitions and file metadata
+ LOG.info("Incrementally loading table metadata for: " + getFullName());
+ Preconditions.checkState(
+ partitionsToUpdate == null || loadParitionFileMetadata);
+ updateMdFromHmsTable(msTbl);
+ if (msTbl.getPartitionKeysSize() == 0) {
+ if (loadParitionFileMetadata) updateUnpartitionedTableFileMd();
+ } else {
+ updatePartitionsFromHms(
+ client, partitionsToUpdate, loadParitionFileMetadata);
+ }
+ LOG.info("Incrementally loaded table metadata for: " + getFullName());
} else {
- updatePartitionsFromHms(
- client, partitionsToUpdate, loadParitionFileMetadata);
+ // Load all partitions from Hive Metastore, including file metadata.
+ LOG.info("Fetching partition metadata from the Metastore: " + getFullName());
+ List<org.apache.hadoop.hive.metastore.api.Partition> msPartitions =
+ MetaStoreUtil.fetchAllPartitions(
+ client, db_.getName(), name_, NUM_PARTITION_FETCH_RETRIES);
+ LOG.info("Fetched partition metadata from the Metastore: " + getFullName());
+ loadAllPartitions(msPartitions, msTbl);
}
- LOG.info("Incrementally loaded table metadata for: " + getFullName());
- } else {
- // Load all partitions from Hive Metastore, including file metadata.
- LOG.info("Fetching partition metadata from the Metastore: " + getFullName());
- List<org.apache.hadoop.hive.metastore.api.Partition> msPartitions =
- MetaStoreUtil.fetchAllPartitions(
- client, db_.getName(), name_, NUM_PARTITION_FETCH_RETRIES);
- LOG.info("Fetched partition metadata from the Metastore: " + getFullName());
- loadAllPartitions(msPartitions, msTbl);
+ if (loadTableSchema) setAvroSchema(client, msTbl);
+ setTableStats(msTbl);
+ fileMetadataStats_.unset();
+ } catch (TableLoadingException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new TableLoadingException("Failed to load metadata for table: "
+ + getFullName(), e);
}
- if (loadTableSchema) setAvroSchema(client, msTbl);
- setTableStats(msTbl);
- numHdfsFiles_ = -1;
- totalHdfsBytes_ = -1;
- } catch (TableLoadingException e) {
- throw e;
- } catch (Exception e) {
- throw new TableLoadingException("Failed to load metadata for table: "
- + getFullName(), e);
+ } finally {
+ context.stop();
}
}
@@ -1648,25 +1712,49 @@ public class HdfsTable extends Table {
* partitions). To prevent the catalog from hitting an OOM error while trying to
* serialize large partition incremental stats, we estimate the stats size and filter
* the incremental stats data from partition objects if the estimate exceeds
- * --inc_stats_size_limit_bytes
+ * --inc_stats_size_limit_bytes. This function also collects storage related statistics
+ * (e.g. number of blocks, files, etc) in order to compute an estimate of the metadata
+ * size of this table.
*/
private THdfsTable getTHdfsTable(boolean includeFileDesc, Set<Long> refPartitions) {
// includeFileDesc implies all partitions should be included (refPartitions == null).
Preconditions.checkState(!includeFileDesc || refPartitions == null);
+ long memUsageEstimate = 0;
int numPartitions =
(refPartitions == null) ? partitionMap_.values().size() : refPartitions.size();
+ memUsageEstimate += numPartitions * PER_PARTITION_MEM_USAGE_BYTES;
long statsSizeEstimate =
numPartitions * getColumns().size() * STATS_SIZE_PER_COLUMN_BYTES;
boolean includeIncrementalStats =
(statsSizeEstimate < BackendConfig.INSTANCE.getIncStatsMaxSize());
+ FileMetadataStats stats = new FileMetadataStats();
Map<Long, THdfsPartition> idToPartition = Maps.newHashMap();
for (HdfsPartition partition: partitionMap_.values()) {
long id = partition.getId();
if (refPartitions == null || refPartitions.contains(id)) {
- idToPartition.put(id,
- partition.toThrift(includeFileDesc, includeIncrementalStats));
+ THdfsPartition tHdfsPartition =
+ partition.toThrift(includeFileDesc, includeIncrementalStats);
+ if (tHdfsPartition.isSetHas_incremental_stats() &&
+ tHdfsPartition.isHas_incremental_stats()) {
+ memUsageEstimate += getColumns().size() * STATS_SIZE_PER_COLUMN_BYTES;
+ hasIncrementalStats_ = true;
+ }
+ if (includeFileDesc) {
+ Preconditions.checkState(tHdfsPartition.isSetNum_blocks() &&
+ tHdfsPartition.isSetTotal_file_size_bytes());
+ stats.numBlocks += tHdfsPartition.getNum_blocks();
+ stats.numFiles +=
+ tHdfsPartition.isSetFile_desc() ? tHdfsPartition.getFile_desc().size() : 0;
+ stats.totalFileBytes += tHdfsPartition.getTotal_file_size_bytes();
+ }
+ idToPartition.put(id, tHdfsPartition);
}
}
+ if (includeFileDesc) fileMetadataStats_.set(stats);
+
+ memUsageEstimate += fileMetadataStats_.numFiles * PER_FD_MEM_USAGE_BYTES +
+ fileMetadataStats_.numBlocks * PER_BLOCK_MEM_USAGE_BYTES;
+ setEstimatedMetadataSize(memUsageEstimate);
THdfsTable hdfsTable = new THdfsTable(hdfsBaseDir_, getColumnNames(),
nullPartitionKeyValue_, nullColumnValue_, idToPartition);
hdfsTable.setAvroSchema(avroSchema_);
@@ -1680,7 +1768,7 @@ public class HdfsTable extends Table {
return hdfsTable;
}
- public long getTotalHdfsBytes() { return totalHdfsBytes_; }
+ public long getTotalHdfsBytes() { return fileMetadataStats_.totalFileBytes; }
public String getHdfsBaseDir() { return hdfsBaseDir_; }
public Path getHdfsBaseDirPath() { return new Path(hdfsBaseDir_); }
public boolean isAvroTable() { return avroSchema_ != null; }
@@ -1978,8 +2066,11 @@ public class HdfsTable extends Table {
// Compute and report the extrapolated row count because the set of files could
// have changed since we last computed stats for this partition. We also follow
// this policy during scan-cardinality estimation.
- if (statsExtrap) rowBuilder.add(getExtrapolatedNumRows(totalHdfsBytes_));
- rowBuilder.add(numHdfsFiles_).addBytes(totalHdfsBytes_)
+ if (statsExtrap) {
+ rowBuilder.add(getExtrapolatedNumRows(fileMetadataStats_.totalFileBytes));
+ }
+ rowBuilder.add(fileMetadataStats_.numFiles)
+ .addBytes(fileMetadataStats_.totalFileBytes)
.addBytes(totalCachedBytes).add("").add("").add("").add("");
result.addToRows(rowBuilder.get());
}
@@ -2072,13 +2163,13 @@ public class HdfsTable extends Table {
// Conservative max size for Java arrays. The actual maximum varies
// from JVM version and sometimes between configurations.
final long JVM_MAX_ARRAY_SIZE = Integer.MAX_VALUE - 10;
- if (numHdfsFiles_ > JVM_MAX_ARRAY_SIZE) {
+ if (fileMetadataStats_.numFiles > JVM_MAX_ARRAY_SIZE) {
throw new IllegalStateException(String.format(
"Too many files to generate a table sample. " +
"Table '%s' has %s files, but a maximum of %s files are supported.",
- getTableName().toString(), numHdfsFiles_, JVM_MAX_ARRAY_SIZE));
+ getTableName().toString(), fileMetadataStats_.numFiles, JVM_MAX_ARRAY_SIZE));
}
- int totalNumFiles = (int) numHdfsFiles_;
+ int totalNumFiles = (int) fileMetadataStats_.numFiles;
// Ensure a consistent ordering of files for repeatable runs. The files within a
// partition are already ordered based on how they are loaded in the catalog.
@@ -2134,4 +2225,37 @@ public class HdfsTable extends Table {
}
return result;
}
+
+ /**
+ * Registers table metrics.
+ */
+ @Override
+ public void initMetrics() {
+ super.initMetrics();
+ metrics_.addGauge(NUM_PARTITIONS_METRIC, new Gauge<Integer>() {
+ @Override
+ public Integer getValue() { return partitionMap_.values().size(); }
+ });
+ metrics_.addGauge(NUM_FILES_METRIC, new Gauge<Long>() {
+ @Override
+ public Long getValue() { return fileMetadataStats_.numFiles; }
+ });
+ metrics_.addGauge(NUM_BLOCKS_METRIC, new Gauge<Long>() {
+ @Override
+ public Long getValue() { return fileMetadataStats_.numBlocks; }
+ });
+ metrics_.addGauge(TOTAL_FILE_BYTES_METRIC, new Gauge<Long>() {
+ @Override
+ public Long getValue() { return fileMetadataStats_.totalFileBytes; }
+ });
+ metrics_.addGauge(MEMORY_ESTIMATE_METRIC, new Gauge<Long>() {
+ @Override
+ public Long getValue() { return getEstimatedMetadataSize(); }
+ });
+ metrics_.addGauge(HAS_INCREMENTAL_STATS_METRIC, new Gauge<Boolean>() {
+ @Override
+ public Boolean getValue() { return hasIncrementalStats_; }
+ });
+ metrics_.addTimer(CATALOG_UPDATE_DURATION_METRIC);
+ }
}
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/KuduTable.java b/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
index 7e13ac5..e9e1617 100644
--- a/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
@@ -58,6 +58,7 @@ import org.apache.kudu.client.PartitionSchema.RangeSchema;
import org.apache.log4j.Logger;
import org.apache.thrift.TException;
+import com.codahale.metrics.Timer;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
@@ -215,30 +216,36 @@ public class KuduTable extends Table {
@Override
public void load(boolean dummy /* not used */, IMetaStoreClient msClient,
org.apache.hadoop.hive.metastore.api.Table msTbl) throws TableLoadingException {
- msTable_ = msTbl;
- kuduTableName_ = msTable_.getParameters().get(KuduTable.KEY_TABLE_NAME);
- Preconditions.checkNotNull(kuduTableName_);
- kuduMasters_ = msTable_.getParameters().get(KuduTable.KEY_MASTER_HOSTS);
- Preconditions.checkNotNull(kuduMasters_);
- setTableStats(msTable_);
- // Load metadata from Kudu and HMS
+ final Timer.Context context =
+ getMetrics().getTimer(Table.LOAD_DURATION_METRIC).time();
try {
- loadSchemaFromKudu();
- loadAllColumnStats(msClient);
- } catch (ImpalaRuntimeException e) {
- throw new TableLoadingException("Error loading metadata for Kudu table " +
- kuduTableName_, e);
- }
+ msTable_ = msTbl;
+ kuduTableName_ = msTable_.getParameters().get(KuduTable.KEY_TABLE_NAME);
+ Preconditions.checkNotNull(kuduTableName_);
+ kuduMasters_ = msTable_.getParameters().get(KuduTable.KEY_MASTER_HOSTS);
+ Preconditions.checkNotNull(kuduMasters_);
+ setTableStats(msTable_);
+ // Load metadata from Kudu and HMS
+ try {
+ loadSchemaFromKudu();
+ loadAllColumnStats(msClient);
+ } catch (ImpalaRuntimeException e) {
+ throw new TableLoadingException("Error loading metadata for Kudu table " +
+ kuduTableName_, e);
+ }
- // Update the table schema in HMS.
- try {
- long lastDdlTime = CatalogOpExecutor.calculateDdlTime(msTable_);
- msTable_.putToParameters("transient_lastDdlTime", Long.toString(lastDdlTime));
- msTable_.putToParameters(StatsSetupConst.DO_NOT_UPDATE_STATS,
- StatsSetupConst.TRUE);
- msClient.alter_table(msTable_.getDbName(), msTable_.getTableName(), msTable_);
- } catch (TException e) {
- throw new TableLoadingException(e.getMessage());
+ // Update the table schema in HMS.
+ try {
+ long lastDdlTime = CatalogOpExecutor.calculateDdlTime(msTable_);
+ msTable_.putToParameters("transient_lastDdlTime", Long.toString(lastDdlTime));
+ msTable_.putToParameters(StatsSetupConst.DO_NOT_UPDATE_STATS,
+ StatsSetupConst.TRUE);
+ msClient.alter_table(msTable_.getDbName(), msTable_.getTableName(), msTable_);
+ } catch (TException e) {
+ throw new TableLoadingException(e.getMessage());
+ }
+ } finally {
+ context.stop();
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/fe/src/main/java/org/apache/impala/catalog/Table.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/Table.java b/fe/src/main/java/org/apache/impala/catalog/Table.java
index 50fe953..a6536ba 100644
--- a/fe/src/main/java/org/apache/impala/catalog/Table.java
+++ b/fe/src/main/java/org/apache/impala/catalog/Table.java
@@ -21,6 +21,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.hadoop.hive.common.StatsSetupConst;
@@ -30,6 +31,7 @@ import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.impala.analysis.TableName;
import org.apache.impala.common.ImpalaRuntimeException;
+import org.apache.impala.common.Metrics;
import org.apache.impala.common.Pair;
import org.apache.impala.common.RuntimeEnv;
import org.apache.impala.thrift.TAccessLevel;
@@ -73,6 +75,18 @@ public abstract class Table extends CatalogObjectImpl {
// values of -1 indicate an unknown statistic.
protected TTableStats tableStats_;
+ // Estimated size (in bytes) of this table metadata. Stored in an AtomicLong to allow
+ // this field to be accessed without holding the table lock.
+ protected AtomicLong estimatedMetadataSize_ = new AtomicLong(0);
+
+ // Number of metadata operations performed on that table since it was loaded.
+ // Stored in an AtomicLong to allow this field to be accessed without holding the
+ // table lock.
+ protected AtomicLong metadataOpsCount_ = new AtomicLong(0);
+
+ // Metrics for this table
+ protected final Metrics metrics_ = new Metrics();
+
// colsByPos[i] refers to the ith column in the table. The first numClusteringCols are
// the clustering columns.
protected final ArrayList<Column> colsByPos_ = Lists.newArrayList();
@@ -89,6 +103,12 @@ public abstract class Table extends CatalogObjectImpl {
// True if this object is stored in an Impalad catalog cache.
protected boolean storedInImpaladCatalogCache_ = false;
+ // Table metrics. These metrics are applicable to all table types. Each subclass of
+ // Table can define additional metrics specific to that table type.
+ public static final String REFRESH_DURATION_METRIC = "refresh-duration";
+ public static final String ALTER_DURATION_METRIC = "alter-duration";
+ public static final String LOAD_DURATION_METRIC = "load-duration";
+
protected Table(org.apache.hadoop.hive.metastore.api.Table msTable, Db db,
String name, String owner) {
msTable_ = msTable;
@@ -99,12 +119,36 @@ public abstract class Table extends CatalogObjectImpl {
CatalogServiceCatalog.getLastDdlTime(msTable_) : -1;
tableStats_ = new TTableStats(-1);
tableStats_.setTotal_file_bytes(-1);
+ initMetrics();
}
public ReentrantLock getLock() { return tableLock_; }
public abstract TTableDescriptor toThriftDescriptor(
int tableId, Set<Long> referencedPartitions);
public abstract TCatalogObjectType getCatalogObjectType();
+ public long getMetadataOpsCount() { return metadataOpsCount_.get(); }
+ public long getEstimatedMetadataSize() { return estimatedMetadataSize_.get(); }
+ public void setEstimatedMetadataSize(long estimatedMetadataSize) {
+ estimatedMetadataSize_.set(estimatedMetadataSize);
+ if (!isStoredInImpaladCatalogCache()) {
+ CatalogUsageMonitor.INSTANCE.updateLargestTables(this);
+ }
+ }
+
+ public void incrementMetadataOpsCount() {
+ metadataOpsCount_.incrementAndGet();
+ if (!isStoredInImpaladCatalogCache()) {
+ CatalogUsageMonitor.INSTANCE.updateFrequentlyAccessedTables(this);
+ }
+ }
+
+ public void initMetrics() {
+ metrics_.addTimer(REFRESH_DURATION_METRIC);
+ metrics_.addTimer(ALTER_DURATION_METRIC);
+ metrics_.addTimer(LOAD_DURATION_METRIC);
+ }
+
+ public Metrics getMetrics() { return metrics_; }
// Returns true if this table reference comes from the impalad catalog cache or if it
// is loaded from the testing framework. Returns false if this table reference points
@@ -527,4 +571,19 @@ public abstract class Table extends CatalogObjectImpl {
}
return new Pair<String, Short>(cachePoolName, cacheReplication);
}
+
+ /**
+ * The implementations of hashCode() and equals() functions are using table names as
+ * unique identifiers of tables. Hence, they should be used with caution and not in
+ * cases where truly unique table objects are needed.
+ */
+ @Override
+ public int hashCode() { return getFullName().hashCode(); }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == null) return false;
+ if (!(obj instanceof Table)) return false;
+ return getFullName().equals(((Table) obj).getFullName());
+ }
}
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/fe/src/main/java/org/apache/impala/common/Metrics.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/common/Metrics.java b/fe/src/main/java/org/apache/impala/common/Metrics.java
new file mode 100644
index 0000000..cf8621f
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/common/Metrics.java
@@ -0,0 +1,149 @@
+// 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.common;
+
+import java.util.Map.Entry;
+
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Histogram;
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.Snapshot;
+import com.codahale.metrics.Timer;
+
+/**
+ * Thin wrapper class around MetricRegisty. Allows users to register and access metrics of
+ * various types (counter, meter, histogram, and timer). This class is not thread-safe.
+ * TODO: Expose the metrics in Json format via a toJson() function.
+ */
+public final class Metrics {
+
+ private final MetricRegistry registry_ = new MetricRegistry();
+
+ public Metrics() {}
+
+ public void addCounter(String name) { registry_.counter(name); }
+ public void addMeter(String name) { registry_.meter(name); }
+ public void addHistogram(String name) { registry_.histogram(name); }
+ public void addTimer(String name) { registry_.timer(name); }
+
+ @SuppressWarnings("rawtypes")
+ public <T extends Gauge> void addGauge(String name, T gauge) {
+ registry_.register(name, gauge);
+ }
+
+ /**
+ * Returns a counter named 'name'. If the counter does not exist, it is registered in
+ * the metrics registry.
+ */
+ public Counter getCounter(String name) {
+ Counter counter = registry_.getCounters().get(name);
+ if (counter == null) counter = registry_.counter(name);
+ return counter;
+ }
+
+ /**
+ * Returns a meter named 'name'. If the meter does not exist, it is registered in the
+ * metrics registry.
+ */
+ public Meter getMeter(String name) {
+ Meter meter = registry_.getMeters().get(name);
+ if (meter == null) meter = registry_.meter(name);
+ return meter;
+ }
+
+ /**
+ * Returns a histogram named 'name'. If the histogram does not exist, it is registered
+ * in the metrics registry.
+ */
+ public Histogram getHistogram(String name) {
+ Histogram histogram = registry_.getHistograms().get(name);
+ if (histogram == null) histogram = registry_.histogram(name);
+ return histogram;
+ }
+
+ /**
+ * Returns a timer named 'name'. If the timer does not exist, it is registered in the
+ * metrics registry.
+ */
+ public Timer getTimer(String name) {
+ Timer timer = registry_.getTimers().get(name);
+ if (timer == null) timer = registry_.timer(name);
+ return timer;
+ }
+
+ @SuppressWarnings("rawtypes")
+ public Gauge getGauge(String name) { return registry_.getGauges().get(name); }
+
+ /**
+ * Returns a string representation of all registered metrics.
+ */
+ @Override
+ @SuppressWarnings("rawtypes")
+ public String toString() {
+ StringBuilder result = new StringBuilder();
+ for (Entry<String, Counter> entry: registry_.getCounters().entrySet()) {
+ result.append(entry.getKey() + ": " + String.valueOf(entry.getValue().getCount()));
+ result.append("\n");
+ }
+ for (Entry<String, Timer> entry: registry_.getTimers().entrySet()) {
+ result.append(entry.getKey() + ": " + timerToString(entry.getValue()));
+ result.append("\n");
+ }
+ for (Entry<String, Gauge> entry: registry_.getGauges().entrySet()) {
+ result.append(entry.getKey() + ": " + String.valueOf(entry.getValue().getValue()));
+ result.append("\n");
+ }
+ for (Entry<String, Histogram> entry: registry_.getHistograms().entrySet()) {
+ result.append(entry.getKey() + ": " +
+ snapshotToString(entry.getValue().getSnapshot()));
+ result.append("\n");
+ }
+ return result.toString();
+ }
+
+ /**
+ * Helper function that pretty prints the contents of a timer metric.
+ */
+ private String timerToString(Timer timer) {
+ StringBuilder builder = new StringBuilder();
+ return builder.append("\n\tCount: " + timer.getCount() + "\n")
+ .append("\tMean rate: " + timer.getMeanRate() + "\n")
+ .append("\t1min rate: " + timer.getOneMinuteRate() + "\n")
+ .append("\t5min rate: " + timer.getFiveMinuteRate() + "\n")
+ .append("\t15min rate: " + timer.getFifteenMinuteRate() + "\n")
+ .append(snapshotToString(timer.getSnapshot()))
+ .toString();
+ }
+
+ /**
+ * Helper function that pretty prints the contents of a metric snapshot.
+ */
+ private String snapshotToString(Snapshot snapshot) {
+ StringBuilder builder = new StringBuilder();
+ return builder.append("\n\tMin (msec): " + snapshot.getMin() / 1000000 + "\n")
+ .append("\tMax (msec): " + snapshot.getMax() / 1000000 + "\n")
+ .append("\tMean (msec): " + snapshot.getMean() / 1000000 + "\n")
+ .append("\tMedian (msec): " + snapshot.getMedian() / 1000000 + "\n")
+ .append("\t75th-% (msec): " + snapshot.get75thPercentile() / 1000000 + "\n")
+ .append("\t95th-% (msec): " + snapshot.get95thPercentile() / 1000000 + "\n")
+ .append("\t99th-% (msec): " + snapshot.get99thPercentile() / 1000000 + "\n")
+ .toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
index df3b10b..f1422db 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -56,6 +56,7 @@ import org.apache.impala.authorization.User;
import org.apache.impala.catalog.Catalog;
import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.CatalogServiceCatalog;
+import org.apache.impala.catalog.CatalogUsageMonitor;
import org.apache.impala.catalog.Column;
import org.apache.impala.catalog.ColumnNotFoundException;
import org.apache.impala.catalog.DataSource;
@@ -152,6 +153,7 @@ import org.apache.impala.util.MetaStoreUtil;
import org.apache.log4j.Logger;
import org.apache.thrift.TException;
+import com.codahale.metrics.Timer;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
@@ -368,6 +370,8 @@ public class CatalogOpExecutor {
throw new InternalException(String.format("Error altering table %s due to lock " +
"contention.", tbl.getFullName()));
}
+ final Timer.Context context
+ = tbl.getMetrics().getTimer(Table.ALTER_DURATION_METRIC).time();
try {
if (params.getAlter_type() == TAlterTableType.RENAME_VIEW
|| params.getAlter_type() == TAlterTableType.RENAME_TABLE) {
@@ -544,6 +548,7 @@ public class CatalogOpExecutor {
response.setResult_set(resultSet);
}
} finally {
+ context.stop();
Preconditions.checkState(!catalog_.getLock().isWriteLockedByCurrentThread());
tbl.getLock().unlock();
}
@@ -3165,6 +3170,8 @@ public class CatalogOpExecutor {
if (!catalog_.tryLockTable(table)) {
throw new InternalException("Error updating the catalog due to lock contention.");
}
+ final Timer.Context context
+ = table.getMetrics().getTimer(HdfsTable.CATALOG_UPDATE_DURATION_METRIC).time();
try {
long newCatalogVersion = catalog_.incrementAndGetCatalogVersion();
catalog_.getLock().writeLock().unlock();
@@ -3319,6 +3326,7 @@ public class CatalogOpExecutor {
loadTableMetadata(table, newCatalogVersion, true, false, partsToLoadMetadata);
addTableToCatalogUpdate(table, response.result);
} finally {
+ context.stop();
Preconditions.checkState(!catalog_.getLock().isWriteLockedByCurrentThread());
table.getLock().unlock();
}
@@ -3356,7 +3364,10 @@ public class CatalogOpExecutor {
* This is to help protect against certain scenarios where the table was
* modified or dropped between the time analysis completed and the the catalog op
* started executing. However, even with these checks it is possible the table was
- * modified or dropped/re-created without us knowing. TODO: Track object IDs to
+ * modified or dropped/re-created without us knowing. This function also updates the
+ * table usage counter.
+ *
+ * TODO: Track object IDs to
* know when a table has been dropped and re-created with the same name.
*/
private Table getExistingTable(String dbName, String tblName) throws CatalogException {
@@ -3364,6 +3375,7 @@ public class CatalogOpExecutor {
if (tbl == null) {
throw new TableNotFoundException("Table not found: " + dbName + "." + tblName);
}
+ tbl.incrementMetadataOpsCount();
if (!tbl.isLoaded()) {
throw new CatalogException(String.format("Table '%s.%s' was modified while " +
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/fe/src/main/java/org/apache/impala/service/JniCatalog.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/service/JniCatalog.java b/fe/src/main/java/org/apache/impala/service/JniCatalog.java
index e945a3b..ed5a51a 100644
--- a/fe/src/main/java/org/apache/impala/service/JniCatalog.java
+++ b/fe/src/main/java/org/apache/impala/service/JniCatalog.java
@@ -45,6 +45,7 @@ import org.apache.impala.thrift.TGetDbsResult;
import org.apache.impala.thrift.TGetFunctionsRequest;
import org.apache.impala.thrift.TGetFunctionsResponse;
import org.apache.impala.thrift.TGetTablesParams;
+import org.apache.impala.thrift.TGetTableMetricsParams;
import org.apache.impala.thrift.TGetTablesResult;
import org.apache.impala.thrift.TLogLevel;
import org.apache.impala.thrift.TPrioritizeLoadRequest;
@@ -197,6 +198,16 @@ public class JniCatalog {
}
/**
+ * Returns the collected metrics of a table.
+ */
+ public String getTableMetrics(byte[] getTableMetricsParams) throws ImpalaException,
+ TException {
+ TGetTableMetricsParams params = new TGetTableMetricsParams();
+ JniUtil.deserializeThrift(protocolFactory_, params, getTableMetricsParams);
+ return catalog_.getTableMetrics(params.table_name);
+ }
+
+ /**
* Gets the thrift representation of a catalog object.
*/
public byte[] getCatalogObject(byte[] thriftParams) throws ImpalaException,
@@ -262,4 +273,12 @@ public class JniCatalog {
TSerializer serializer = new TSerializer(protocolFactory_);
return serializer.serialize(catalogOpExecutor_.updateCatalog(request));
}
+
+ /**
+ * Returns information about the current catalog usage.
+ */
+ public byte[] getCatalogUsage() throws ImpalaException, TException {
+ TSerializer serializer = new TSerializer(protocolFactory_);
+ return serializer.serialize(catalog_.getCatalogUsage());
+ }
}
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/fe/src/main/java/org/apache/impala/util/TopNCache.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/util/TopNCache.java b/fe/src/main/java/org/apache/impala/util/TopNCache.java
new file mode 100644
index 0000000..9f6b972
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/util/TopNCache.java
@@ -0,0 +1,108 @@
+// 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.util.Comparator;
+import java.util.List;
+import java.util.PriorityQueue;
+
+import com.google.common.base.Function;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+/**
+ * Thread-safe class that represents a TOP-N cache of items. It stores the top-N items of
+ * a generic type T based on a user-specified ranking function that returns a numeric
+ * value (long).
+ *
+ * The cache has a maximum capacity (N) of stored items. The implementation allows two
+ * policies with respect to the way new items are handled when maximum capacity is
+ * reached:
+ * a) Always evict policy: A new item will always replace the item with the lowest rank
+ * according to the specified ranking function even if the rank of the newly added
+ * function is lower than the one to be replaced.
+ * b) Rank-based eviction policy: A new item will be added to the cache iff its rank is
+ * higher than the smallest rank in the cache and the item with that rank will be evicted.
+ *
+ * TODO: Replace these two policies with an LFU cache with dynamic aging.
+ */
+public final class TopNCache<T, R extends Long> {
+
+ // Function used to rank items stored in this cache.
+ private final Function<T, R> function_;
+ // Maximum capacity of this cache.
+ private final int maxCapacity_;
+ // The cache is stored as a priority queue.
+ private final PriorityQueue<T> heap_;
+ // Determines the eviction policy to apply when the cache reaches maximum capacity.
+ // TODO: Convert to enum?
+ private final boolean alwaysEvictAtCapacity_;
+
+ /**
+ * Compares the ranks of two T objects, returning 0 if they are equal, < 0 if the rank
+ * of the first is smaller, or > 0 if the rank of the first object is larger.
+ */
+ private int compareRanks(T t1, T t2) {
+ return function_.apply(t1).compareTo(function_.apply(t2));
+ }
+
+ public TopNCache(Function<T, R> f, int maxCapacity, boolean evictAtCapacity) {
+ Preconditions.checkNotNull(f);
+ Preconditions.checkState(maxCapacity > 0);
+ function_ = f;
+ maxCapacity_ = maxCapacity;
+ heap_ = new PriorityQueue<T>(maxCapacity_,
+ new Comparator<T>() {
+ @Override
+ public int compare(T t1, T t2) { return compareRanks(t1, t2); }
+ }
+ );
+ alwaysEvictAtCapacity_ = evictAtCapacity;
+ }
+
+ /**
+ * Adds or updates an item in the cache. If the item already exists, its rank position
+ * is refreshed by removing and adding the item back to the cache. If the item is not in
+ * the cache and maximum capacity hasn't been reached, the item is added to the cache.
+ * Otherwise, the eviction policy is applied and the item either replaces the cache item
+ * with the lowest rank or it is rejected from the cache if its rank is lower than the
+ * lowest rank in the cache.
+ */
+ public synchronized void putOrUpdate(T item) {
+ if (!heap_.remove(item)) {
+ if (heap_.size() == maxCapacity_) {
+ if (!alwaysEvictAtCapacity_ && compareRanks(item, heap_.peek()) <= 0) {
+ return;
+ }
+ heap_.poll();
+ }
+ }
+ heap_.add(item);
+ }
+
+ /**
+ * Removes an item from the cache.
+ */
+ public synchronized void remove(T item) { heap_.remove(item); }
+
+ /**
+ * Returns the list of all the items in the cache.
+ */
+ public synchronized List<T> listEntries() { return ImmutableList.copyOf(heap_); }
+}
+
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/fe/src/test/java/org/apache/impala/util/TestTopNCache.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/util/TestTopNCache.java b/fe/src/test/java/org/apache/impala/util/TestTopNCache.java
new file mode 100644
index 0000000..1b34599
--- /dev/null
+++ b/fe/src/test/java/org/apache/impala/util/TestTopNCache.java
@@ -0,0 +1,130 @@
+// 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 static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertEquals;
+import org.junit.Test;
+
+import com.google.common.base.Function;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Unit tests for the TopNCache class.
+ */
+public class TestTopNCache {
+
+ /**
+ * Create a TopNCache with 'capacity' max capacity and populate it with 'numEntries'
+ * entries where each entry is a number from 0 to 'numEntries'.
+ */
+ private static TopNCache<Long, Long> createAndPopulate(int capacity,
+ long numEntries, boolean policy) {
+ TopNCache<Long, Long> cache =
+ new TopNCache<Long, Long>(new Function<Long, Long>() {
+ @Override
+ public Long apply(Long element) { return element; }
+ }, capacity, policy);
+ for (long i = 0; i < numEntries; ++i) cache.putOrUpdate(i);
+ return cache;
+ }
+
+ @Test
+ public void testCreateAndPopulateCache() throws Exception {
+ int[] capacities = {1, 10, 1000};
+ boolean[] evictionPolicies = {true, false};
+ for (int capacity: capacities) {
+ for (boolean policy: evictionPolicies) {
+ TopNCache<Long, Long> cache =
+ createAndPopulate(capacity, 2 * capacity, policy);
+ assertEquals(cache.listEntries().size(), capacity);
+ for (long i = 0; i < capacity * 2; i++) cache.remove(i);
+ assertEquals(cache.listEntries().size(), 0);
+ }
+ }
+ }
+
+ @Test
+ public void testUpdateExistingElements() throws Exception {
+ final int capacity = 10;
+ TopNCache<Long, Long> cache = createAndPopulate(capacity, capacity / 2, true);
+ assertEquals(cache.listEntries().size(), capacity / 2);
+ // Adding the same elements should not alter the number of elements stored in the
+ // cache.
+ for (long i = 0; i < capacity / 2; i++) cache.putOrUpdate(i);
+ assertEquals(cache.listEntries().size(), capacity / 2);
+ }
+
+ @Test
+ public void testAlwaysEvictPolicy() throws Exception {
+ final int capacity = 10;
+ TopNCache<Long, Long> cache = createAndPopulate(capacity, capacity, true);
+ assertEquals(cache.listEntries().size(), capacity);
+ cache.putOrUpdate((long) capacity);
+ assertEquals(cache.listEntries().size(), capacity);
+ // Assert that the new element replaced the smallest element in the cache
+ assertTrue(!cache.listEntries().contains(Long.valueOf(0)));
+ cache.putOrUpdate((long) capacity + 1);
+ assertTrue(!cache.listEntries().contains(Long.valueOf(1)));
+ List<Long> cacheElements = cache.listEntries();
+ for (long i = 2; i < capacity + 2; i++) {
+ assertTrue(cacheElements.contains(Long.valueOf(i)));
+ }
+ }
+
+ @Test
+ public void testRankBasedEvictionPolicy() throws Exception {
+ final int capacity = 10;
+ TopNCache<Long, Long> cache = new TopNCache<Long, Long>(
+ new Function<Long, Long>() {
+ @Override
+ public Long apply(Long element) { return element; }
+ }, capacity, false);
+ for (long i = 1; i < capacity + 1; i++) cache.putOrUpdate(i);
+ assertEquals(cache.listEntries().size(), capacity);
+ cache.putOrUpdate((long) 0);
+ // 0 shouldn't be added to the cache because it's rank is smaller than the lowest rank
+ // in the cache.
+ assertTrue(!cache.listEntries().contains(Long.valueOf(0)));
+ cache.putOrUpdate((long) capacity + 1);
+ assertEquals(cache.listEntries().size(), capacity);
+ assertTrue(cache.listEntries().contains(Long.valueOf(capacity + 1)));
+ }
+
+ @Test
+ public void testRankBasedEvictionPolicyWithRandomInput() throws Exception {
+ final int capacity = 5;
+ long[] values = {10, 8, 1, 2, 5, 4, 3, 6, 9, 7};
+ TopNCache<Long, Long> cache = new TopNCache<Long, Long>(
+ new Function<Long, Long>() {
+ @Override
+ public Long apply(Long element) { return element; }
+ }, capacity, false);
+ for (Long entry: values) cache.putOrUpdate(entry);
+ List<Long> entries = cache.listEntries();
+ assertEquals(entries.size(), capacity);
+ // Make sure only the top-5 elements are in the cache
+ for (long i = 1; i <= capacity; ++i) {
+ assertTrue(!entries.contains(i));
+ assertTrue(entries.contains(i + capacity));
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/tests/webserver/test_web_pages.py
----------------------------------------------------------------------
diff --git a/tests/webserver/test_web_pages.py b/tests/webserver/test_web_pages.py
index 101f786..54163dc 100644
--- a/tests/webserver/test_web_pages.py
+++ b/tests/webserver/test_web_pages.py
@@ -29,6 +29,7 @@ class TestWebPage(ImpalaTestSuite):
RESET_GLOG_LOGLEVEL_URL = "http://localhost:{0}/reset_glog_level"
CATALOG_URL = "http://localhost:{0}/catalog"
CATALOG_OBJECT_URL = "http://localhost:{0}/catalog_object"
+ TABLE_METRICS_URL = "http://localhost:{0}/table_metrics"
QUERY_BACKENDS_URL = "http://localhost:{0}/query_backends"
THREAD_GROUP_URL = "http://localhost:{0}/thread-group"
# log4j changes do not apply to the statestore since it doesn't
@@ -37,6 +38,7 @@ class TestWebPage(ImpalaTestSuite):
# one with it.
TEST_PORTS_WITHOUT_SS = ["25000", "25020"]
TEST_PORTS_WITH_SS = ["25000", "25010", "25020"]
+ CATALOG_TEST_PORT = ["25020"]
def test_memz(self):
"""test /memz at impalad / statestored / catalogd"""
@@ -151,6 +153,8 @@ class TestWebPage(ImpalaTestSuite):
self.__test_catalog_object("functional_parquet", "alltypes")
self.__test_catalog_object("functional", "alltypesnopart")
self.__test_catalog_object("functional_kudu", "alltypes")
+ self.__test_table_metrics("functional", "alltypes", "total-file-size-bytes")
+ self.__test_table_metrics("functional_kudu", "alltypes", "alter-duration")
def __test_catalog_object(self, db_name, tbl_name):
"""Tests the /catalog_object endpoint for the given db/table. Runs
@@ -164,6 +168,11 @@ class TestWebPage(ImpalaTestSuite):
"?object_type=TABLE&object_name=%s.%s" % (db_name, tbl_name), tbl_name,
ports_to_test=self.TEST_PORTS_WITHOUT_SS)
+ def __test_table_metrics(self, db_name, tbl_name, metric):
+ self.client.execute("refresh %s.%s" % (db_name, tbl_name))
+ self.get_and_check_status(self.TABLE_METRICS_URL +
+ "?name=%s.%s" % (db_name, tbl_name), metric, ports_to_test=self.CATALOG_TEST_PORT)
+
def test_query_details(self, unique_database):
"""Test that /query_backends returns the list of backend states for DML or queries;
nothing for DDL statements"""
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/www/catalog.tmpl
----------------------------------------------------------------------
diff --git a/www/catalog.tmpl b/www/catalog.tmpl
index 5e271bf..ffe7c78 100644
--- a/www/catalog.tmpl
+++ b/www/catalog.tmpl
@@ -20,6 +20,87 @@ under the License.
<h2>Catalog</h2>
+{{?has_large_tables}}
+<div class="panel panel-info">
+ <div class="panel-heading">
+ <h2 class="panel-title">
+ Top-{{num_large_tables}} Tables with Highest Memory Requirements
+ </h2>
+ </div>
+ <div class="panel-body">
+ <table id="large-tables" class='table table-hover table-bordered'>
+ <thead>
+ <tr>
+ <th>Name</th>
+ <th>Estimated memory</th>
+ <th>Metrics</th>
+ </tr>
+ </thead>
+ <tbody>
+ {{#large_tables}}
+ <tr>
+ <td><a href="catalog_object?object_type=TABLE&object_name={{name}}">{{name}}</a>
+ </td>
+ <td>{{mem_estimate}}</td>
+ <td><a href="table_metrics?name={{name}}">{{name}}-metrics</a></td>
+ </tr>
+ {{/large_tables}}
+ </tbody>
+ </table>
+ </div>
+</div>
+
+<script>
+ $(document).ready(function() {
+ $('#large-tables').DataTable({
+ "order": [[ 0, "desc" ]],
+ "pageLength": 10
+ });
+ });
+</script>
+{{/has_large_tables}}
+
+{{?has_frequent_tables}}
+<div class="panel panel-info">
+ <div class="panel-heading">
+ <h2 class="panel-title">
+ Top-{{num_frequent_tables}} Tables with Highest Number of Metadata Operations
+ </h2>
+ </div>
+ <div class="panel-body">
+ <table id="frequent-tables" class='table table-hover table-bordered'>
+ <thead>
+ <tr>
+ <th>Name</th>
+ <th>Metadata Operations (since loaded)</th>
+ <th>Metrics</th>
+ </tr>
+ </thead>
+ <tbody>
+ {{#frequent_tables}}
+ <tr>
+ <td><a href="catalog_object?object_type=TABLE&object_name={{name}}">{{name}}</a>
+ </td>
+ <td>{{num_metadata_ops}}</td>
+ <td><a href="table_metrics?name={{name}}">{{name}}-metrics</a></td>
+ </tr>
+ {{/frequent_tables}}
+ </tbody>
+ </table>
+ </div>
+</div>
+
+<script>
+ $(document).ready(function() {
+ $('#frequent-tables').DataTable({
+ "order": [[ 0, "desc" ]],
+ "pageLength": 10
+ });
+ });
+</script>
+{{/has_frequent_tables}}
+
+<h3>Databases</h3>
<ol class="breadcrumb">
{{#databases}}
<li><a href='#{{name}}'>{{name}}</a></li>
@@ -36,16 +117,38 @@ under the License.
</a>
</div>
<div class="panel-body">
- <ul>
- {{#tables}}
- <li>
- <a href="catalog_object?object_type=TABLE&object_name={{fqtn}}">{{name}}</a>
- </li>
- {{/tables}}
- </ul>
+ <table id="{{name}}-tables" class='table table-hover table-bordered'>
+ <thead>
+ <tr>
+ <th>Name</th>
+ {{?has_metrics}}
+ <th>Metrics</th>
+ {{/has_metrics}}
+ </tr>
+ </thead>
+ <tbody>
+ {{#tables}}
+ <tr>
+ <td><a href="catalog_object?object_type=TABLE&object_name={{fqtn}}">{{name}}</a>
+ </td>
+ {{?has_metrics}}
+ <td><a href="table_metrics?name={{fqtn}}">{{name}}-metrics</a></td>
+ {{/has_metrics}}
+ </tr>
+ {{/tables}}
+ </tbody>
+ </table>
</div>
</div>
+<script>
+ $(document).ready(function() {
+ $('#{{name}}-tables').DataTable({
+ "pageLength": 5
+ });
+ });
+</script>
+
{{/databases}}
{{> www/common-footer.tmpl }}
http://git-wip-us.apache.org/repos/asf/impala/blob/3f00d10e/www/table_metrics.tmpl
----------------------------------------------------------------------
diff --git a/www/table_metrics.tmpl b/www/table_metrics.tmpl
new file mode 100644
index 0000000..5140309
--- /dev/null
+++ b/www/table_metrics.tmpl
@@ -0,0 +1,23 @@
+<!--
+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.
+-->
+{{> www/common-header.tmpl }}
+
+<pre>{{table_metrics}}</pre>
+
+{{> www/common-footer.tmpl }}
[3/8] impala git commit: IMPALA-6382: Cap spillable buffer size and
max row size query options
Posted by jr...@apache.org.
IMPALA-6382: Cap spillable buffer size and max row size query options
Currently the default and min spillable buffer size and max row size
query options accept any valid int64 value. Since the planner depends
on these values for memory estimations, if a very large value close to
the limits of int64 is set, the variables representing or relying on
these estimates can overflow during different phases of query execution.
This patch puts a reasonable upper limit of 1TB to these query options
to prevent such a situation.
Testing:
Added backend query option tests.
Change-Id: I36d3915f7019b13c3eb06f08bfdb38c71ec864f1
Reviewed-on: http://gerrit.cloudera.org:8080/9023
Reviewed-by: Bikramjeet Vig <bi...@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/028a83e6
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/028a83e6
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/028a83e6
Branch: refs/heads/2.x
Commit: 028a83e6543a18dd3b9161226355f1e8d36c4ed7
Parents: ca7d03c
Author: Bikramjeet Vig <bi...@cloudera.com>
Authored: Fri Jan 12 17:23:15 2018 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Thu Jan 18 23:08:26 2018 +0000
----------------------------------------------------------------------
be/src/service/query-options-test.cc | 12 +++++++++-
be/src/service/query-options.cc | 23 +++++++++++++++-----
be/src/service/query-options.h | 3 +++
.../functional-query/queries/QueryTest/set.test | 4 ++--
4 files changed, 34 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/028a83e6/be/src/service/query-options-test.cc
----------------------------------------------------------------------
diff --git a/be/src/service/query-options-test.cc b/be/src/service/query-options-test.cc
index efc5307..80c9866 100644
--- a/be/src/service/query-options-test.cc
+++ b/be/src/service/query-options-test.cc
@@ -140,7 +140,7 @@ TEST(QueryOptions, SetByteOptions) {
{MAKE_OPTIONDEF(max_scan_range_length), {-1, I64_MAX}},
{MAKE_OPTIONDEF(rm_initial_mem), {-1, I64_MAX}},
{MAKE_OPTIONDEF(buffer_pool_limit), {-1, I64_MAX}},
- {MAKE_OPTIONDEF(max_row_size), {1, I64_MAX}},
+ {MAKE_OPTIONDEF(max_row_size), {1, ROW_SIZE_LIMIT}},
{MAKE_OPTIONDEF(parquet_file_size), {-1, I32_MAX}}
};
vector<pair<OptionDef<int32_t>, Range<int32_t>>> case_set_i32 {
@@ -302,8 +302,18 @@ TEST(QueryOptions, SetSpecialOptions) {
TestOk("2MB", 2 * 1024 * 1024);
TestOk("32G", 32ll * 1024 * 1024 * 1024);
TestError("10MB");
+ TestOk(to_string(SPILLABLE_BUFFER_LIMIT).c_str(), SPILLABLE_BUFFER_LIMIT);
+ TestError(to_string(2 * SPILLABLE_BUFFER_LIMIT).c_str());
}
}
+ // MAX_ROW_SIZE should be between 1 and ROW_SIZE_LIMIT
+ {
+ OptionDef<int64_t> key_def = MAKE_OPTIONDEF(max_row_size);
+ auto TestError = MakeTestErrFn(options, key_def);
+ TestError("-1");
+ TestError("0");
+ TestError(to_string(ROW_SIZE_LIMIT + 1).c_str());
+ }
}
TEST(QueryOptions, ParseQueryOptions) {
http://git-wip-us.apache.org/repos/asf/impala/blob/028a83e6/be/src/service/query-options.cc
----------------------------------------------------------------------
diff --git a/be/src/service/query-options.cc b/be/src/service/query-options.cc
index e8e8c7b..e3b5a1f 100644
--- a/be/src/service/query-options.cc
+++ b/be/src/service/query-options.cc
@@ -552,7 +552,13 @@ Status impala::SetQueryOption(const string& key, const string& value,
ParseMemValue(value, "Default spillable buffer size", &buffer_size_bytes));
if (!BitUtil::IsPowerOf2(buffer_size_bytes)) {
return Status(
- Substitute("Buffer size must be a power of two: $0", buffer_size_bytes));
+ Substitute("Default spillable buffer size must be a power of two: $0",
+ buffer_size_bytes));
+ }
+ if (buffer_size_bytes > SPILLABLE_BUFFER_LIMIT) {
+ return Status(Substitute(
+ "Default spillable buffer size must be less than or equal to: $0",
+ SPILLABLE_BUFFER_LIMIT));
}
query_options->__set_default_spillable_buffer_size(buffer_size_bytes);
break;
@@ -563,7 +569,13 @@ Status impala::SetQueryOption(const string& key, const string& value,
ParseMemValue(value, "Minimum spillable buffer size", &buffer_size_bytes));
if (!BitUtil::IsPowerOf2(buffer_size_bytes)) {
return Status(
- Substitute("Buffer size must be a power of two: $0", buffer_size_bytes));
+ Substitute("Minimum spillable buffer size must be a power of two: $0",
+ buffer_size_bytes));
+ }
+ if (buffer_size_bytes > SPILLABLE_BUFFER_LIMIT) {
+ return Status(Substitute(
+ "Minimum spillable buffer size must be less than or equal to: $0",
+ SPILLABLE_BUFFER_LIMIT));
}
query_options->__set_min_spillable_buffer_size(buffer_size_bytes);
break;
@@ -571,9 +583,10 @@ Status impala::SetQueryOption(const string& key, const string& value,
case TImpalaQueryOptions::MAX_ROW_SIZE: {
int64_t max_row_size_bytes;
RETURN_IF_ERROR(ParseMemValue(value, "Max row size", &max_row_size_bytes));
- if (max_row_size_bytes <= 0) {
- return Status(Substitute(
- "Max row size must be a positive number of bytes: $0", value));
+ if (max_row_size_bytes <= 0 || max_row_size_bytes > ROW_SIZE_LIMIT) {
+ return Status(
+ Substitute("Invalid max row size of $0. Valid sizes are in [$1, $2]", value,
+ 1, ROW_SIZE_LIMIT));
}
query_options->__set_max_row_size(max_row_size_bytes);
break;
http://git-wip-us.apache.org/repos/asf/impala/blob/028a83e6/be/src/service/query-options.h
----------------------------------------------------------------------
diff --git a/be/src/service/query-options.h b/be/src/service/query-options.h
index b7065f4..be3607f 100644
--- a/be/src/service/query-options.h
+++ b/be/src/service/query-options.h
@@ -132,6 +132,9 @@ typedef std::unordered_map<string, beeswax::TQueryOptionLevel::type>
QUERY_OPT_FN(idle_session_timeout, IDLE_SESSION_TIMEOUT, TQueryOptionLevel::REGULAR)\
;
+/// Enforce practical limits on some query options to avoid undesired query state.
+ static const int64_t SPILLABLE_BUFFER_LIMIT = 1LL << 40; // 1 TB
+ static const int64_t ROW_SIZE_LIMIT = 1LL << 40; // 1 TB
/// Converts a TQueryOptions struct into a map of key, value pairs. Options that
/// aren't set and lack defaults in common/thrift/ImpalaInternalService.thrift are
http://git-wip-us.apache.org/repos/asf/impala/blob/028a83e6/testdata/workloads/functional-query/queries/QueryTest/set.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/set.test b/testdata/workloads/functional-query/queries/QueryTest/set.test
index 14ac629..32ad938 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/set.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/set.test
@@ -234,10 +234,10 @@ explain select count(distinct double_col) from functional.alltypesagg;
---- QUERY
set max_row_size=-1;
---- CATCH
-Max row size must be a positive number of bytes: -1
+Invalid max row size of -1. Valid sizes are in [1, 1099511627776]
====
---- QUERY
set max_row_size=0;
---- CATCH
-Max row size must be a positive number of bytes: 0
+Invalid max row size of 0. Valid sizes are in [1, 1099511627776]
====
[5/8] impala git commit: IMPALA-2397: Use atomics for IntGauge and
IntCounter
Posted by jr...@apache.org.
IMPALA-2397: Use atomics for IntGauge and IntCounter
This change removes the spinlock in IntGauge and IntCounter
and uses AtomicInt64 instead. As shown in IMPALA-2397, multiple
threads can be contending for the spinlocks of some global metrics
under concurrent queries.
This change also breaks up SimpleMetric is renamed to ScalarMetric
and broken into two subclasses:
- LockedMetric:
- a value store for any primitive type (int,float,string etc).
- atomic read and write via GetValue() and SetValue() respectively.
- AtomicMetric:
- the basis of IntGauge and IntCounter. Support atomic increment
of the metric value via Increment() interface.
- atomic read and write via GetValue() and SetValue() respectively.
- only support int64_t type.
Change-Id: I48dfa5443cd771916b53541a0ffeaf1bcc7e7606
Reviewed-on: http://gerrit.cloudera.org:8080/9012
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/e714f2b3
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/e714f2b3
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/e714f2b3
Branch: refs/heads/2.x
Commit: e714f2b33c5b64d5680dbc15e166759930f04560
Parents: b3d38b5
Author: Michael Ho <kw...@cloudera.com>
Authored: Wed Jan 10 19:28:09 2018 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Thu Jan 18 23:31:52 2018 +0000
----------------------------------------------------------------------
be/src/exec/external-data-source-executor.cc | 8 +-
be/src/rpc/TAcceptQueueServer.cpp | 2 +-
be/src/rpc/thrift-server.cc | 4 +-
be/src/runtime/client-cache.cc | 6 +-
be/src/runtime/data-stream-mgr.cc | 6 +-
be/src/runtime/exec-env.cc | 10 +-
be/src/runtime/io/scan-range.cc | 4 +-
be/src/runtime/krpc-data-stream-mgr.cc | 6 +-
be/src/runtime/mem-tracker-test.cc | 4 +-
be/src/runtime/mem-tracker.cc | 10 +-
be/src/runtime/mem-tracker.h | 4 +-
be/src/runtime/query-exec-mgr.cc | 2 +-
be/src/runtime/query-state.cc | 6 +-
be/src/runtime/tmp-file-mgr-test.cc | 2 +-
be/src/runtime/tmp-file-mgr.cc | 4 +-
be/src/scheduling/admission-controller.cc | 58 +++---
be/src/scheduling/scheduler.cc | 9 +-
be/src/service/impala-server.cc | 10 +-
be/src/service/session-expiry-test.cc | 12 +-
be/src/statestore/statestore-subscriber.cc | 14 +-
be/src/statestore/statestore.cc | 11 +-
be/src/util/common-metrics.cc | 2 +-
be/src/util/default-path-handlers.cc | 2 +-
be/src/util/impalad-metrics.cc | 66 +++----
be/src/util/memory-metrics.cc | 99 +++++-----
be/src/util/memory-metrics.h | 42 ++--
be/src/util/metrics-test.cc | 44 ++---
be/src/util/metrics.h | 222 ++++++++++++----------
be/src/util/thread.cc | 8 +-
common/thrift/metrics.json | 2 +-
30 files changed, 350 insertions(+), 329 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/exec/external-data-source-executor.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/external-data-source-executor.cc b/be/src/exec/external-data-source-executor.cc
index 7c810c6..7c54f39 100644
--- a/be/src/exec/external-data-source-executor.cc
+++ b/be/src/exec/external-data-source-executor.cc
@@ -76,9 +76,9 @@ class ExternalDataSourceExecutor::JniState {
"getNumClassCacheMisses", "()J");
RETURN_ERROR_IF_EXC(env);
- num_class_cache_hits_ = metrics->AddCounter<int64_t>(
+ num_class_cache_hits_ = metrics->AddCounter(
"external-data-source.class-cache.hits", 0);
- num_class_cache_misses_ = metrics->AddCounter<int64_t>(
+ num_class_cache_misses_ = metrics->AddCounter(
"external-data-source.class-cache.misses", 0);
return Status::OK();
}
@@ -92,11 +92,11 @@ class ExternalDataSourceExecutor::JniState {
int64_t num_cache_hits = env->CallStaticLongMethod(executor_class_,
get_num_cache_hits_id_);
RETURN_ERROR_IF_EXC(env);
- num_class_cache_hits_->set_value(num_cache_hits);
+ num_class_cache_hits_->SetValue(num_cache_hits);
int64_t num_cache_misses = env->CallStaticLongMethod(executor_class_,
get_num_cache_misses_id_);
RETURN_ERROR_IF_EXC(env);
- num_class_cache_misses_->set_value(num_cache_misses);
+ num_class_cache_misses_->SetValue(num_cache_misses);
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/rpc/TAcceptQueueServer.cpp
----------------------------------------------------------------------
diff --git a/be/src/rpc/TAcceptQueueServer.cpp b/be/src/rpc/TAcceptQueueServer.cpp
index 8a398a2..5c1b1da 100644
--- a/be/src/rpc/TAcceptQueueServer.cpp
+++ b/be/src/rpc/TAcceptQueueServer.cpp
@@ -286,7 +286,7 @@ void TAcceptQueueServer::InitMetrics(MetricGroup* metrics, const string& key_pre
DCHECK(metrics != NULL);
stringstream queue_size_ss;
queue_size_ss << key_prefix << ".connection-setup-queue-size";
- queue_size_metric_ = metrics->AddGauge<int64_t>(queue_size_ss.str(), 0);
+ queue_size_metric_ = metrics->AddGauge(queue_size_ss.str(), 0);
metrics_enabled_ = true;
}
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/rpc/thrift-server.cc
----------------------------------------------------------------------
diff --git a/be/src/rpc/thrift-server.cc b/be/src/rpc/thrift-server.cc
index ab51315..eaca699 100644
--- a/be/src/rpc/thrift-server.cc
+++ b/be/src/rpc/thrift-server.cc
@@ -342,10 +342,10 @@ ThriftServer::ThriftServer(const string& name,
metrics_enabled_ = true;
stringstream count_ss;
count_ss << "impala.thrift-server." << name << ".connections-in-use";
- num_current_connections_metric_ = metrics->AddGauge<int64_t>(count_ss.str(), 0);
+ num_current_connections_metric_ = metrics->AddGauge(count_ss.str(), 0);
stringstream max_ss;
max_ss << "impala.thrift-server." << name << ".total-connections";
- total_connections_metric_ = metrics->AddCounter<int64_t>(max_ss.str(), 0);
+ total_connections_metric_ = metrics->AddCounter(max_ss.str(), 0);
metrics_ = metrics;
} else {
metrics_enabled_ = false;
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/runtime/client-cache.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/client-cache.cc b/be/src/runtime/client-cache.cc
index 8c0b6aa..af530f7 100644
--- a/be/src/runtime/client-cache.cc
+++ b/be/src/runtime/client-cache.cc
@@ -94,7 +94,7 @@ Status ClientCacheHelper::ReopenClient(ClientFactory factory_method,
// CreateClient() will increment total_clients_metric_ if succeed.
if (metrics_enabled_) {
total_clients_metric_->Increment(-1);
- DCHECK_GE(total_clients_metric_->value(), 0);
+ DCHECK_GE(total_clients_metric_->GetValue(), 0);
}
lock_guard<mutex> lock(client_map_lock_);
client_map_.erase(client);
@@ -235,11 +235,11 @@ void ClientCacheHelper::InitMetrics(MetricGroup* metrics, const string& key_pref
lock_guard<mutex> lock(cache_lock_);
stringstream count_ss;
count_ss << key_prefix << ".client-cache.clients-in-use";
- clients_in_use_metric_ = metrics->AddGauge<int64_t>(count_ss.str(), 0);
+ clients_in_use_metric_ = metrics->AddGauge(count_ss.str(), 0);
stringstream max_ss;
max_ss << key_prefix << ".client-cache.total-clients";
- total_clients_metric_ = metrics->AddGauge<int64_t>(max_ss.str(), 0);
+ total_clients_metric_ = metrics->AddGauge(max_ss.str(), 0);
metrics_enabled_ = true;
}
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/runtime/data-stream-mgr.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/data-stream-mgr.cc b/be/src/runtime/data-stream-mgr.cc
index 93c524e..45eee7f 100644
--- a/be/src/runtime/data-stream-mgr.cc
+++ b/be/src/runtime/data-stream-mgr.cc
@@ -57,10 +57,10 @@ namespace impala {
DataStreamMgr::DataStreamMgr(MetricGroup* metrics) {
metrics_ = metrics->GetOrCreateChildGroup("datastream-manager");
num_senders_waiting_ =
- metrics_->AddGauge<int64_t>("senders-blocked-on-recvr-creation", 0L);
+ metrics_->AddGauge("senders-blocked-on-recvr-creation", 0L);
total_senders_waited_ =
- metrics_->AddCounter<int64_t>("total-senders-blocked-on-recvr-creation", 0L);
- num_senders_timedout_ = metrics_->AddCounter<int64_t>(
+ metrics_->AddCounter("total-senders-blocked-on-recvr-creation", 0L);
+ num_senders_timedout_ = metrics_->AddCounter(
"total-senders-timedout-waiting-for-recvr-creation", 0L);
}
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/runtime/exec-env.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/exec-env.cc b/be/src/runtime/exec-env.cc
index 6d9a857..f191921 100644
--- a/be/src/runtime/exec-env.cc
+++ b/be/src/runtime/exec-env.cc
@@ -333,9 +333,9 @@ Status ExecEnv::Init() {
// Also need a MemTracker for unused reservations as a negative value. Unused
// reservations are counted against queries but not against the process memory
// consumption. This accounts for that difference.
- IntGauge* negated_unused_reservation = obj_pool_->Add(new NegatedGauge<int64_t>(
- MakeTMetricDef("negated_unused_reservation", TMetricKind::GAUGE, TUnit::BYTES),
- BufferPoolMetric::UNUSED_RESERVATION_BYTES));
+ IntGauge* negated_unused_reservation = obj_pool_->Add(new NegatedGauge(
+ MakeTMetricDef("negated_unused_reservation", TMetricKind::GAUGE, TUnit::BYTES),
+ BufferPoolMetric::UNUSED_RESERVATION_BYTES));
obj_pool_->Add(new MemTracker(negated_unused_reservation, -1,
"Buffer Pool: Unused Reservation", mem_tracker_.get()));
#if !defined(ADDRESS_SANITIZER) && !defined(THREAD_SANITIZER)
@@ -350,13 +350,13 @@ Status ExecEnv::Init() {
// reserved (TcmallocMetric::PHYSICAL_BYTES_RESERVED) and the bytes in use
// (TcmallocMetrics::BYTES_IN_USE). This overhead accounts for all the cached freelists
// used by TCMalloc.
- IntGauge* negated_bytes_in_use = obj_pool_->Add(new NegatedGauge<int64_t>(
+ IntGauge* negated_bytes_in_use = obj_pool_->Add(new NegatedGauge(
MakeTMetricDef("negated_tcmalloc_bytes_in_use", TMetricKind::GAUGE, TUnit::BYTES),
TcmallocMetric::BYTES_IN_USE));
vector<IntGauge*> overhead_metrics;
overhead_metrics.push_back(negated_bytes_in_use);
overhead_metrics.push_back(TcmallocMetric::PHYSICAL_BYTES_RESERVED);
- SumGauge<int64_t>* tcmalloc_overhead = obj_pool_->Add(new SumGauge<int64_t>(
+ SumGauge* tcmalloc_overhead = obj_pool_->Add(new SumGauge(
MakeTMetricDef("tcmalloc_overhead", TMetricKind::GAUGE, TUnit::BYTES),
overhead_metrics));
obj_pool_->Add(
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/runtime/io/scan-range.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/io/scan-range.cc b/be/src/runtime/io/scan-range.cc
index dc14050..21daa96 100644
--- a/be/src/runtime/io/scan-range.cc
+++ b/be/src/runtime/io/scan-range.cc
@@ -335,8 +335,8 @@ void ScanRange::Close() {
struct hdfsHedgedReadMetrics* hedged_metrics;
int success = hdfsGetHedgedReadMetrics(fs_, &hedged_metrics);
if (success == 0) {
- ImpaladMetrics::HEDGED_READ_OPS->set_value(hedged_metrics->hedgedReadOps);
- ImpaladMetrics::HEDGED_READ_OPS_WIN->set_value(hedged_metrics->hedgedReadOpsWin);
+ ImpaladMetrics::HEDGED_READ_OPS->SetValue(hedged_metrics->hedgedReadOps);
+ ImpaladMetrics::HEDGED_READ_OPS_WIN->SetValue(hedged_metrics->hedgedReadOpsWin);
hdfsFreeHedgedReadMetrics(hedged_metrics);
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/runtime/krpc-data-stream-mgr.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/krpc-data-stream-mgr.cc b/be/src/runtime/krpc-data-stream-mgr.cc
index 348b9ab..86955c8 100644
--- a/be/src/runtime/krpc-data-stream-mgr.cc
+++ b/be/src/runtime/krpc-data-stream-mgr.cc
@@ -63,10 +63,10 @@ KrpcDataStreamMgr::KrpcDataStreamMgr(MetricGroup* metrics)
boost::bind(&KrpcDataStreamMgr::DeserializeThreadFn, this, _1, _2)) {
MetricGroup* dsm_metrics = metrics->GetOrCreateChildGroup("datastream-manager");
num_senders_waiting_ =
- dsm_metrics->AddGauge<int64_t>("senders-blocked-on-recvr-creation", 0L);
+ dsm_metrics->AddGauge("senders-blocked-on-recvr-creation", 0L);
total_senders_waited_ =
- dsm_metrics->AddCounter<int64_t>("total-senders-blocked-on-recvr-creation", 0L);
- num_senders_timedout_ = dsm_metrics->AddCounter<int64_t>(
+ dsm_metrics->AddCounter("total-senders-blocked-on-recvr-creation", 0L);
+ num_senders_timedout_ = dsm_metrics->AddCounter(
"total-senders-timedout-waiting-for-recvr-creation", 0L);
}
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/runtime/mem-tracker-test.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/mem-tracker-test.cc b/be/src/runtime/mem-tracker-test.cc
index 4aaac05..faeb6a9 100644
--- a/be/src/runtime/mem-tracker-test.cc
+++ b/be/src/runtime/mem-tracker-test.cc
@@ -62,13 +62,13 @@ TEST(MemTestTest, ConsumptionMetric) {
md.__set_units(TUnit::BYTES);
md.__set_kind(TMetricKind::GAUGE);
IntGauge metric(md, 0);
- EXPECT_EQ(metric.value(), 0);
+ EXPECT_EQ(metric.GetValue(), 0);
TMetricDef neg_md;
neg_md.__set_key("neg_test");
neg_md.__set_units(TUnit::BYTES);
neg_md.__set_kind(TMetricKind::GAUGE);
- NegatedGauge<int64_t> neg_metric(neg_md, &metric);
+ NegatedGauge neg_metric(neg_md, &metric);
MemTracker t(&metric, 100, "");
MemTracker neg_t(&neg_metric, 100, "");
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/runtime/mem-tracker.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/mem-tracker.cc b/be/src/runtime/mem-tracker.cc
index 98f45db..e5aa290 100644
--- a/be/src/runtime/mem-tracker.cc
+++ b/be/src/runtime/mem-tracker.cc
@@ -211,16 +211,16 @@ MemTracker::~MemTracker() {
}
void MemTracker::RegisterMetrics(MetricGroup* metrics, const string& prefix) {
- num_gcs_metric_ = metrics->AddCounter<int64_t>(Substitute("$0.num-gcs", prefix), 0);
+ num_gcs_metric_ = metrics->AddCounter(Substitute("$0.num-gcs", prefix), 0);
// TODO: Consider a total amount of bytes freed counter
- bytes_freed_by_last_gc_metric_ = metrics->AddGauge<int64_t>(
+ bytes_freed_by_last_gc_metric_ = metrics->AddGauge(
Substitute("$0.bytes-freed-by-last-gc", prefix), -1);
- bytes_over_limit_metric_ = metrics->AddGauge<int64_t>(
+ bytes_over_limit_metric_ = metrics->AddGauge(
Substitute("$0.bytes-over-limit", prefix), -1);
- limit_metric_ = metrics->AddGauge<int64_t>(Substitute("$0.limit", prefix), limit_);
+ limit_metric_ = metrics->AddGauge(Substitute("$0.limit", prefix), limit_);
}
// Calling this on the query tracker results in output like:
@@ -430,7 +430,7 @@ bool MemTracker::GcMemory(int64_t max_consumption) {
}
if (bytes_freed_by_last_gc_metric_ != NULL) {
- bytes_freed_by_last_gc_metric_->set_value(pre_gc_consumption - curr_consumption);
+ bytes_freed_by_last_gc_metric_->SetValue(pre_gc_consumption - curr_consumption);
}
return curr_consumption > max_consumption;
}
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/runtime/mem-tracker.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/mem-tracker.h b/be/src/runtime/mem-tracker.h
index fb1cd90..c582d72 100644
--- a/be/src/runtime/mem-tracker.h
+++ b/be/src/runtime/mem-tracker.h
@@ -250,7 +250,7 @@ class MemTracker {
bool LimitExceeded() {
if (UNLIKELY(CheckLimitExceeded())) {
if (bytes_over_limit_metric_ != NULL) {
- bytes_over_limit_metric_->set_value(consumption() - limit_);
+ bytes_over_limit_metric_->SetValue(consumption() - limit_);
}
return GcMemory(limit_);
}
@@ -274,7 +274,7 @@ class MemTracker {
/// call if this tracker has a consumption metric.
void RefreshConsumptionFromMetric() {
DCHECK(consumption_metric_ != nullptr);
- consumption_->Set(consumption_metric_->value());
+ consumption_->Set(consumption_metric_->GetValue());
}
int64_t limit() const { return limit_; }
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/runtime/query-exec-mgr.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/query-exec-mgr.cc b/be/src/runtime/query-exec-mgr.cc
index 4f30f4e..316b712 100644
--- a/be/src/runtime/query-exec-mgr.cc
+++ b/be/src/runtime/query-exec-mgr.cc
@@ -123,7 +123,7 @@ void QueryExecMgr::StartQueryHelper(QueryState* qs) {
#if !defined(ADDRESS_SANITIZER) && !defined(THREAD_SANITIZER)
// tcmalloc and address or thread sanitizer cannot be used together
if (FLAGS_log_mem_usage_interval > 0) {
- uint64_t num_complete = ImpaladMetrics::IMPALA_SERVER_NUM_FRAGMENTS->value();
+ uint64_t num_complete = ImpaladMetrics::IMPALA_SERVER_NUM_FRAGMENTS->GetValue();
if (num_complete % FLAGS_log_mem_usage_interval == 0) {
char buf[2048];
// This outputs how much memory is currently being used by this impalad
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/runtime/query-state.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/query-state.cc b/be/src/runtime/query-state.cc
index 10c8033..259cd34 100644
--- a/be/src/runtime/query-state.cc
+++ b/be/src/runtime/query-state.cc
@@ -381,11 +381,13 @@ void QueryState::ExecFInstance(FragmentInstanceState* fis) {
<< " fragment_idx=" << fis->instance_ctx().fragment_idx
<< " per_fragment_instance_idx=" << fis->instance_ctx().per_fragment_instance_idx
<< " coord_state_idx=" << rpc_params().coord_state_idx
- << " #in-flight=" << ImpaladMetrics::IMPALA_SERVER_NUM_FRAGMENTS_IN_FLIGHT->value();
+ << " #in-flight="
+ << ImpaladMetrics::IMPALA_SERVER_NUM_FRAGMENTS_IN_FLIGHT->GetValue();
Status status = fis->Exec();
ImpaladMetrics::IMPALA_SERVER_NUM_FRAGMENTS_IN_FLIGHT->Increment(-1L);
VLOG_QUERY << "Instance completed. instance_id=" << PrintId(fis->instance_id())
- << " #in-flight=" << ImpaladMetrics::IMPALA_SERVER_NUM_FRAGMENTS_IN_FLIGHT->value()
+ << " #in-flight="
+ << ImpaladMetrics::IMPALA_SERVER_NUM_FRAGMENTS_IN_FLIGHT->GetValue()
<< " status=" << status;
// initiate cancellation if nobody has done so yet
if (!status.ok()) Cancel();
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/runtime/tmp-file-mgr-test.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/tmp-file-mgr-test.cc b/be/src/runtime/tmp-file-mgr-test.cc
index fbc0a36..3091c58 100644
--- a/be/src/runtime/tmp-file-mgr-test.cc
+++ b/be/src/runtime/tmp-file-mgr-test.cc
@@ -78,7 +78,7 @@ class TmpFileMgrTest : public ::testing::Test {
vector<TmpFileMgr::DeviceId> active = tmp_file_mgr->ActiveTmpDevices();
IntGauge* active_metric =
metrics_->FindMetricForTesting<IntGauge>("tmp-file-mgr.active-scratch-dirs");
- EXPECT_EQ(active.size(), active_metric->value());
+ EXPECT_EQ(active.size(), active_metric->GetValue());
SetMetric<string>* active_set_metric =
metrics_->FindMetricForTesting<SetMetric<string>>(
"tmp-file-mgr.active-scratch-dirs.list");
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/runtime/tmp-file-mgr.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/tmp-file-mgr.cc b/be/src/runtime/tmp-file-mgr.cc
index 650af0b..d35d302 100644
--- a/be/src/runtime/tmp-file-mgr.cc
+++ b/be/src/runtime/tmp-file-mgr.cc
@@ -132,10 +132,10 @@ Status TmpFileMgr::InitCustom(const vector<string>& tmp_dirs, bool one_dir_per_d
DCHECK(metrics != nullptr);
num_active_scratch_dirs_metric_ =
- metrics->AddGauge<int64_t>(TMP_FILE_MGR_ACTIVE_SCRATCH_DIRS, 0);
+ metrics->AddGauge(TMP_FILE_MGR_ACTIVE_SCRATCH_DIRS, 0);
active_scratch_dirs_metric_ = SetMetric<string>::CreateAndRegister(
metrics, TMP_FILE_MGR_ACTIVE_SCRATCH_DIRS_LIST, set<string>());
- num_active_scratch_dirs_metric_->set_value(tmp_dirs_.size());
+ num_active_scratch_dirs_metric_->SetValue(tmp_dirs_.size());
for (int i = 0; i < tmp_dirs_.size(); ++i) {
active_scratch_dirs_metric_->Add(tmp_dirs_[i]);
}
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/scheduling/admission-controller.cc
----------------------------------------------------------------------
diff --git a/be/src/scheduling/admission-controller.cc b/be/src/scheduling/admission-controller.cc
index 99f659a..f43af2c 100644
--- a/be/src/scheduling/admission-controller.cc
+++ b/be/src/scheduling/admission-controller.cc
@@ -482,9 +482,9 @@ bool AdmissionController::RejectImmediately(QuerySchedule* schedule,
}
void AdmissionController::PoolStats::UpdateConfigMetrics(const TPoolConfig& pool_cfg) {
- metrics_.pool_max_mem_resources->set_value(pool_cfg.max_mem_resources);
- metrics_.pool_max_requests->set_value(pool_cfg.max_requests);
- metrics_.pool_max_queued->set_value(pool_cfg.max_queued);
+ metrics_.pool_max_mem_resources->SetValue(pool_cfg.max_mem_resources);
+ metrics_.pool_max_requests->SetValue(pool_cfg.max_requests);
+ metrics_.pool_max_queued->SetValue(pool_cfg.max_queued);
}
Status AdmissionController::AdmitQuery(QuerySchedule* schedule) {
@@ -734,18 +734,18 @@ void AdmissionController::PoolStats::UpdateAggregates(HostMemMap* host_mem_reser
if (agg_num_running_ == num_running && agg_num_queued_ == num_queued &&
agg_mem_reserved_ == mem_reserved) {
- DCHECK_EQ(num_running, metrics_.agg_num_running->value());
- DCHECK_EQ(num_queued, metrics_.agg_num_queued->value());
- DCHECK_EQ(mem_reserved, metrics_.agg_mem_reserved->value());
+ DCHECK_EQ(num_running, metrics_.agg_num_running->GetValue());
+ DCHECK_EQ(num_queued, metrics_.agg_num_queued->GetValue());
+ DCHECK_EQ(mem_reserved, metrics_.agg_mem_reserved->GetValue());
return;
}
VLOG_ROW << "Recomputed agg stats, previous: " << DebugString();
agg_num_running_ = num_running;
agg_num_queued_ = num_queued;
agg_mem_reserved_ = mem_reserved;
- metrics_.agg_num_running->set_value(num_running);
- metrics_.agg_num_queued->set_value(num_queued);
- metrics_.agg_mem_reserved->set_value(mem_reserved);
+ metrics_.agg_num_running->SetValue(num_running);
+ metrics_.agg_num_queued->SetValue(num_queued);
+ metrics_.agg_mem_reserved->SetValue(mem_reserved);
VLOG_ROW << "Updated: " << DebugString();
}
@@ -782,12 +782,12 @@ void AdmissionController::PoolStats::UpdateMemTrackerStats() {
if (current_reserved != local_stats_.backend_mem_reserved) {
parent_->pools_for_updates_.insert(name_);
local_stats_.backend_mem_reserved = current_reserved;
- metrics_.local_backend_mem_reserved->set_value(current_reserved);
+ metrics_.local_backend_mem_reserved->SetValue(current_reserved);
}
const int64_t current_usage =
tracker == nullptr ? static_cast<int64_t>(0) : tracker->consumption();
- metrics_.local_backend_mem_usage->set_value(current_usage);
+ metrics_.local_backend_mem_usage->SetValue(current_usage);
}
void AdmissionController::AddPoolUpdates(vector<TTopicDelta>* topic_updates) {
@@ -906,44 +906,44 @@ AdmissionController::GetPoolStats(const string& pool_name) {
}
void AdmissionController::PoolStats::InitMetrics() {
- metrics_.total_admitted = parent_->metrics_group_->AddCounter<int64_t>(
+ metrics_.total_admitted = parent_->metrics_group_->AddCounter(
TOTAL_ADMITTED_METRIC_KEY_FORMAT, 0, name_);
- metrics_.total_queued = parent_->metrics_group_->AddCounter<int64_t>(
+ metrics_.total_queued = parent_->metrics_group_->AddCounter(
TOTAL_QUEUED_METRIC_KEY_FORMAT, 0, name_);
- metrics_.total_dequeued = parent_->metrics_group_->AddCounter<int64_t>(
+ metrics_.total_dequeued = parent_->metrics_group_->AddCounter(
TOTAL_DEQUEUED_METRIC_KEY_FORMAT, 0, name_);
- metrics_.total_rejected = parent_->metrics_group_->AddCounter<int64_t>(
+ metrics_.total_rejected = parent_->metrics_group_->AddCounter(
TOTAL_REJECTED_METRIC_KEY_FORMAT, 0, name_);
- metrics_.total_timed_out = parent_->metrics_group_->AddCounter<int64_t>(
+ metrics_.total_timed_out = parent_->metrics_group_->AddCounter(
TOTAL_TIMED_OUT_METRIC_KEY_FORMAT, 0, name_);
- metrics_.total_released = parent_->metrics_group_->AddCounter<int64_t>(
+ metrics_.total_released = parent_->metrics_group_->AddCounter(
TOTAL_RELEASED_METRIC_KEY_FORMAT, 0, name_);
- metrics_.time_in_queue_ms = parent_->metrics_group_->AddCounter<int64_t>(
+ metrics_.time_in_queue_ms = parent_->metrics_group_->AddCounter(
TIME_IN_QUEUE_METRIC_KEY_FORMAT, 0, name_);
- metrics_.agg_num_running = parent_->metrics_group_->AddGauge<int64_t>(
+ metrics_.agg_num_running = parent_->metrics_group_->AddGauge(
AGG_NUM_RUNNING_METRIC_KEY_FORMAT, 0, name_);
- metrics_.agg_num_queued = parent_->metrics_group_->AddGauge<int64_t>(
+ metrics_.agg_num_queued = parent_->metrics_group_->AddGauge(
AGG_NUM_QUEUED_METRIC_KEY_FORMAT, 0, name_);
- metrics_.agg_mem_reserved = parent_->metrics_group_->AddGauge<int64_t>(
+ metrics_.agg_mem_reserved = parent_->metrics_group_->AddGauge(
AGG_MEM_RESERVED_METRIC_KEY_FORMAT, 0, name_);
- metrics_.local_mem_admitted = parent_->metrics_group_->AddGauge<int64_t>(
+ metrics_.local_mem_admitted = parent_->metrics_group_->AddGauge(
LOCAL_MEM_ADMITTED_METRIC_KEY_FORMAT, 0, name_);
- metrics_.local_num_admitted_running = parent_->metrics_group_->AddGauge<int64_t>(
+ metrics_.local_num_admitted_running = parent_->metrics_group_->AddGauge(
LOCAL_NUM_ADMITTED_RUNNING_METRIC_KEY_FORMAT, 0, name_);
- metrics_.local_num_queued = parent_->metrics_group_->AddGauge<int64_t>(
+ metrics_.local_num_queued = parent_->metrics_group_->AddGauge(
LOCAL_NUM_QUEUED_METRIC_KEY_FORMAT, 0, name_);
- metrics_.local_backend_mem_usage = parent_->metrics_group_->AddGauge<int64_t>(
+ metrics_.local_backend_mem_usage = parent_->metrics_group_->AddGauge(
LOCAL_BACKEND_MEM_USAGE_METRIC_KEY_FORMAT, 0, name_);
- metrics_.local_backend_mem_reserved = parent_->metrics_group_->AddGauge<int64_t>(
+ metrics_.local_backend_mem_reserved = parent_->metrics_group_->AddGauge(
LOCAL_BACKEND_MEM_RESERVED_METRIC_KEY_FORMAT, 0, name_);
- metrics_.pool_max_mem_resources = parent_->metrics_group_->AddGauge<int64_t>(
+ metrics_.pool_max_mem_resources = parent_->metrics_group_->AddGauge(
POOL_MAX_MEM_RESOURCES_METRIC_KEY_FORMAT, 0, name_);
- metrics_.pool_max_requests = parent_->metrics_group_->AddGauge<int64_t>(
+ metrics_.pool_max_requests = parent_->metrics_group_->AddGauge(
POOL_MAX_REQUESTS_METRIC_KEY_FORMAT, 0, name_);
- metrics_.pool_max_queued = parent_->metrics_group_->AddGauge<int64_t>(
+ metrics_.pool_max_queued = parent_->metrics_group_->AddGauge(
POOL_MAX_QUEUED_METRIC_KEY_FORMAT, 0, name_);
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/scheduling/scheduler.cc
----------------------------------------------------------------------
diff --git a/be/src/scheduling/scheduler.cc b/be/src/scheduling/scheduler.cc
index 5cf0f01..e924f50 100644
--- a/be/src/scheduling/scheduler.cc
+++ b/be/src/scheduling/scheduler.cc
@@ -97,11 +97,10 @@ Status Scheduler::Init(const TNetworkAddress& backend_address,
// This is after registering with the statestored, so we already have to synchronize
// access to the executors_config_ shared_ptr.
int num_backends = GetExecutorsConfig()->NumBackends();
- total_assignments_ = metrics_->AddCounter<int64_t>(ASSIGNMENTS_KEY, 0);
- total_local_assignments_ = metrics_->AddCounter<int64_t>(LOCAL_ASSIGNMENTS_KEY, 0);
+ total_assignments_ = metrics_->AddCounter(ASSIGNMENTS_KEY, 0);
+ total_local_assignments_ = metrics_->AddCounter(LOCAL_ASSIGNMENTS_KEY, 0);
initialized_ = metrics_->AddProperty(SCHEDULER_INIT_KEY, true);
- num_fragment_instances_metric_ =
- metrics_->AddGauge<int64_t>(NUM_BACKENDS_KEY, num_backends);
+ num_fragment_instances_metric_ = metrics_->AddGauge(NUM_BACKENDS_KEY, num_backends);
}
if (statestore_subscriber_ != nullptr) {
@@ -197,7 +196,7 @@ void Scheduler::UpdateMembership(
if (metrics_ != nullptr) {
/// TODO-MT: fix this (do we even need to report it?)
- num_fragment_instances_metric_->set_value(current_executors_.size());
+ num_fragment_instances_metric_->SetValue(current_executors_.size());
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/service/impala-server.cc
----------------------------------------------------------------------
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index 6358145..a62130c 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -1059,8 +1059,8 @@ Status ImpalaServer::UnregisterQuery(const TUniqueId& query_id, bool check_infli
Status ImpalaServer::UpdateCatalogMetrics() {
TGetDbsResult dbs;
RETURN_IF_ERROR(exec_env_->frontend()->GetDbs(nullptr, nullptr, &dbs));
- ImpaladMetrics::CATALOG_NUM_DBS->set_value(dbs.dbs.size());
- ImpaladMetrics::CATALOG_NUM_TABLES->set_value(0L);
+ ImpaladMetrics::CATALOG_NUM_DBS->SetValue(dbs.dbs.size());
+ ImpaladMetrics::CATALOG_NUM_TABLES->SetValue(0L);
for (const TDatabase& db: dbs.dbs) {
TGetTablesResult table_names;
RETURN_IF_ERROR(exec_env_->frontend()->GetTableNames(db.db_name, nullptr, nullptr,
@@ -1433,7 +1433,7 @@ void ImpalaServer::CatalogUpdateCallback(
TTopicDelta& update = subscriber_topic_updates->back();
update.topic_name = CatalogServer::IMPALA_CATALOG_TOPIC;
update.__set_from_version(0L);
- ImpaladMetrics::CATALOG_READY->set_value(false);
+ ImpaladMetrics::CATALOG_READY->SetValue(false);
// Dropped all cached lib files (this behaves as if all functions and data
// sources are dropped).
LibCache::instance()->DropCache();
@@ -1447,7 +1447,7 @@ void ImpalaServer::CatalogUpdateCallback(
LOG(INFO) << "Catalog topic update applied with version: " << new_catalog_version
<< " new min catalog object version: " << resp.min_catalog_object_version;
}
- ImpaladMetrics::CATALOG_READY->set_value(new_catalog_version > 0);
+ ImpaladMetrics::CATALOG_READY->SetValue(new_catalog_version > 0);
// TODO: deal with an error status
discard_result(UpdateCatalogMetrics());
// Remove all dropped objects from the library cache.
@@ -2130,7 +2130,7 @@ Status ImpalaServer::Start(int32_t thrift_be_port, int32_t beeswax_port,
LOG(INFO) << "Impala Beeswax Service listening on " << beeswax_server_->port();
}
services_started_ = true;
- ImpaladMetrics::IMPALA_SERVER_READY->set_value(true);
+ ImpaladMetrics::IMPALA_SERVER_READY->SetValue(true);
LOG(INFO) << "Impala has started.";
return Status::OK();
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/service/session-expiry-test.cc
----------------------------------------------------------------------
diff --git a/be/src/service/session-expiry-test.cc b/be/src/service/session-expiry-test.cc
index fa69476..a211701 100644
--- a/be/src/service/session-expiry-test.cc
+++ b/be/src/service/session-expiry-test.cc
@@ -58,8 +58,8 @@ TEST(SessionTest, TestExpiry) {
IntGauge* hs2_session_metric =
impala->metrics()->FindMetricForTesting<IntGauge>(
ImpaladMetricKeys::IMPALA_SERVER_NUM_OPEN_HS2_SESSIONS);
- EXPECT_EQ(expired_metric->value(), 0L);
- EXPECT_EQ(beeswax_session_metric->value(), 0L);
+ EXPECT_EQ(expired_metric->GetValue(), 0L);
+ EXPECT_EQ(beeswax_session_metric->GetValue(), 0L);
{
scoped_ptr<ThriftClient<ImpalaServiceClient>> beeswax_clients[NUM_SESSIONS];
@@ -80,16 +80,16 @@ TEST(SessionTest, TestExpiry) {
}
int64_t start = UnixMillis();
- while (expired_metric->value() != NUM_SESSIONS * 2 &&
+ while (expired_metric->GetValue() != NUM_SESSIONS * 2 &&
UnixMillis() - start < MAX_IDLE_TIMEOUT_MS) {
SleepForMs(100);
}
- ASSERT_EQ(expired_metric->value(), NUM_SESSIONS * 2)
+ ASSERT_EQ(expired_metric->GetValue(), NUM_SESSIONS * 2)
<< "Sessions did not expire within "<< MAX_IDLE_TIMEOUT_MS / 1000 <<" secs";
- ASSERT_EQ(beeswax_session_metric->value(), NUM_SESSIONS)
+ ASSERT_EQ(beeswax_session_metric->GetValue(), NUM_SESSIONS)
<< "Beeswax sessions unexpectedly closed after expiration";
- ASSERT_EQ(hs2_session_metric->value(), NUM_SESSIONS)
+ ASSERT_EQ(hs2_session_metric->GetValue(), NUM_SESSIONS)
<< "HiveServer2 sessions unexpectedly closed after expiration";
TPingImpalaServiceResp resp;
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/statestore/statestore-subscriber.cc
----------------------------------------------------------------------
diff --git a/be/src/statestore/statestore-subscriber.cc b/be/src/statestore/statestore-subscriber.cc
index 678236e..99da183 100644
--- a/be/src/statestore/statestore-subscriber.cc
+++ b/be/src/statestore/statestore-subscriber.cc
@@ -113,7 +113,7 @@ StatestoreSubscriber::StatestoreSubscriber(const std::string& subscriber_id,
metrics_(metrics->GetOrCreateChildGroup("statestore-subscriber")) {
connected_to_statestore_metric_ =
metrics_->AddProperty("statestore-subscriber.connected", false);
- last_recovery_duration_metric_ = metrics_->AddGauge(
+ last_recovery_duration_metric_ = metrics_->AddDoubleGauge(
"statestore-subscriber.last-recovery-duration", 0.0);
last_recovery_time_metric_ = metrics_->AddProperty<string>(
"statestore-subscriber.last-recovery-time", "N/A");
@@ -164,12 +164,12 @@ Status StatestoreSubscriber::Register() {
RETURN_IF_ERROR(client.DoRpc(&StatestoreServiceClientWrapper::RegisterSubscriber,
request, &response));
Status status = Status(response.status);
- if (status.ok()) connected_to_statestore_metric_->set_value(true);
+ if (status.ok()) connected_to_statestore_metric_->SetValue(true);
if (response.__isset.registration_id) {
lock_guard<mutex> l(registration_id_lock_);
registration_id_ = response.registration_id;
const string& registration_string = PrintId(registration_id_);
- registration_id_metric_->set_value(registration_string);
+ registration_id_metric_->SetValue(registration_string);
VLOG(1) << "Subscriber registration ID: " << registration_string;
} else {
VLOG(1) << "No subscriber registration ID received from statestore";
@@ -243,7 +243,7 @@ void StatestoreSubscriber::RecoveryModeChecker() {
lock_guard<mutex> l(lock_);
MonotonicStopWatch recovery_timer;
recovery_timer.Start();
- connected_to_statestore_metric_->set_value(false);
+ connected_to_statestore_metric_->SetValue(false);
LOG(INFO) << subscriber_id_
<< ": Connection with statestore lost, entering recovery mode";
uint32_t attempt_count = 1;
@@ -265,7 +265,7 @@ void StatestoreSubscriber::RecoveryModeChecker() {
<< status.GetDetail();
SleepForMs(SLEEP_INTERVAL_MS);
}
- last_recovery_duration_metric_->set_value(
+ last_recovery_duration_metric_->SetValue(
recovery_timer.ElapsedTime() / (1000.0 * 1000.0 * 1000.0));
}
// When we're successful in re-registering, we don't do anything
@@ -273,9 +273,9 @@ void StatestoreSubscriber::RecoveryModeChecker() {
// responsibility of individual clients to post missing updates
// back to the statestore. This saves a lot of complexity where
// we would otherwise have to cache updates here.
- last_recovery_duration_metric_->set_value(
+ last_recovery_duration_metric_->SetValue(
recovery_timer.ElapsedTime() / (1000.0 * 1000.0 * 1000.0));
- last_recovery_time_metric_->set_value(CurrentTimeString());
+ last_recovery_time_metric_->SetValue(CurrentTimeString());
}
SleepForMs(SLEEP_INTERVAL_MS);
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/statestore/statestore.cc
----------------------------------------------------------------------
diff --git a/be/src/statestore/statestore.cc b/be/src/statestore/statestore.cc
index d0a4851..b135e38 100644
--- a/be/src/statestore/statestore.cc
+++ b/be/src/statestore/statestore.cc
@@ -236,13 +236,12 @@ Statestore::Statestore(MetricGroup* metrics)
FLAGS_statestore_max_missed_heartbeats / 2)) {
DCHECK(metrics != NULL);
- num_subscribers_metric_ =
- metrics->AddGauge<int64_t>(STATESTORE_LIVE_SUBSCRIBERS, 0);
+ num_subscribers_metric_ = metrics->AddGauge(STATESTORE_LIVE_SUBSCRIBERS, 0);
subscriber_set_metric_ = SetMetric<string>::CreateAndRegister(metrics,
STATESTORE_LIVE_SUBSCRIBERS_LIST, set<string>());
- key_size_metric_ = metrics->AddGauge<int64_t>(STATESTORE_TOTAL_KEY_SIZE_BYTES, 0);
- value_size_metric_ = metrics->AddGauge<int64_t>(STATESTORE_TOTAL_VALUE_SIZE_BYTES, 0);
- topic_size_metric_ = metrics->AddGauge<int64_t>(STATESTORE_TOTAL_TOPIC_SIZE_BYTES, 0);
+ key_size_metric_ = metrics->AddGauge(STATESTORE_TOTAL_KEY_SIZE_BYTES, 0);
+ value_size_metric_ = metrics->AddGauge(STATESTORE_TOTAL_VALUE_SIZE_BYTES, 0);
+ topic_size_metric_ = metrics->AddGauge(STATESTORE_TOTAL_TOPIC_SIZE_BYTES, 0);
topic_update_duration_metric_ =
StatsMetric<double>::CreateAndRegister(metrics, STATESTORE_UPDATE_DURATION);
@@ -398,7 +397,7 @@ Status Statestore::RegisterSubscriber(const SubscriberId& subscriber_id,
subscribers_.insert(make_pair(subscriber_id, current_registration));
failure_detector_->UpdateHeartbeat(
PrintId(current_registration->registration_id()), true);
- num_subscribers_metric_->set_value(subscribers_.size());
+ num_subscribers_metric_->SetValue(subscribers_.size());
subscriber_set_metric_->Add(subscriber_id);
}
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/util/common-metrics.cc
----------------------------------------------------------------------
diff --git a/be/src/util/common-metrics.cc b/be/src/util/common-metrics.cc
index d147862..114e0e0 100644
--- a/be/src/util/common-metrics.cc
+++ b/be/src/util/common-metrics.cc
@@ -33,7 +33,7 @@ void CommonMetrics::InitCommonMetrics(MetricGroup* metric_group) {
KUDU_CLIENT_VERSION = metric_group->AddProperty<string>(
KUDU_CLIENT_VERSION_METRIC_NAME, kudu::client::GetShortVersionString());
- PROCESS_START_TIME->set_value(CurrentTimeString());
+ PROCESS_START_TIME->SetValue(CurrentTimeString());
}
}
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/util/default-path-handlers.cc
----------------------------------------------------------------------
diff --git a/be/src/util/default-path-handlers.cc b/be/src/util/default-path-handlers.cc
index 88d23f1..10966b4 100644
--- a/be/src/util/default-path-handlers.cc
+++ b/be/src/util/default-path-handlers.cc
@@ -211,7 +211,7 @@ void RootHandler(const Webserver::ArgumentMap& args, Document* document) {
document->GetAllocator());
if (CommonMetrics::PROCESS_START_TIME != nullptr) {
- Value process_start_time(CommonMetrics::PROCESS_START_TIME->value().c_str(),
+ Value process_start_time(CommonMetrics::PROCESS_START_TIME->GetValue().c_str(),
document->GetAllocator());
document->AddMember("process_start_time", process_start_time,
document->GetAllocator());
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/util/impalad-metrics.cc
----------------------------------------------------------------------
diff --git a/be/src/util/impalad-metrics.cc b/be/src/util/impalad-metrics.cc
index 1325f2e..18e96a8 100644
--- a/be/src/util/impalad-metrics.cc
+++ b/be/src/util/impalad-metrics.cc
@@ -162,70 +162,70 @@ void ImpaladMetrics::CreateMetrics(MetricGroup* m) {
IMPALA_SERVER_READY = m->AddProperty<bool>(
ImpaladMetricKeys::IMPALA_SERVER_READY, false);
- IMPALA_SERVER_NUM_QUERIES = m->AddCounter<int64_t>(
+ IMPALA_SERVER_NUM_QUERIES = m->AddCounter(
ImpaladMetricKeys::IMPALA_SERVER_NUM_QUERIES, 0);
- NUM_QUERIES_REGISTERED = m->AddGauge<int64_t>(ImpaladMetricKeys::NUM_QUERIES_REGISTERED, 0);
- NUM_QUERIES_EXPIRED = m->AddCounter<int64_t>(
+ NUM_QUERIES_REGISTERED = m->AddGauge(
+ ImpaladMetricKeys::NUM_QUERIES_REGISTERED, 0);
+ NUM_QUERIES_EXPIRED = m->AddCounter(
ImpaladMetricKeys::NUM_QUERIES_EXPIRED, 0);
- NUM_QUERIES_SPILLED = m->AddCounter<int64_t>(
+ NUM_QUERIES_SPILLED = m->AddCounter(
ImpaladMetricKeys::NUM_QUERIES_SPILLED, 0);
- IMPALA_SERVER_NUM_FRAGMENTS = m->AddCounter<int64_t>(
+ IMPALA_SERVER_NUM_FRAGMENTS = m->AddCounter(
ImpaladMetricKeys::IMPALA_SERVER_NUM_FRAGMENTS, 0);
IMPALA_SERVER_NUM_FRAGMENTS_IN_FLIGHT = m->AddGauge(
ImpaladMetricKeys::IMPALA_SERVER_NUM_FRAGMENTS_IN_FLIGHT, 0L);
- IMPALA_SERVER_NUM_OPEN_HS2_SESSIONS = m->AddGauge<int64_t>(
+ IMPALA_SERVER_NUM_OPEN_HS2_SESSIONS = m->AddGauge(
ImpaladMetricKeys::IMPALA_SERVER_NUM_OPEN_HS2_SESSIONS, 0);
- IMPALA_SERVER_NUM_OPEN_BEESWAX_SESSIONS = m->AddGauge<int64_t>(
+ IMPALA_SERVER_NUM_OPEN_BEESWAX_SESSIONS = m->AddGauge(
ImpaladMetricKeys::IMPALA_SERVER_NUM_OPEN_BEESWAX_SESSIONS, 0);
- NUM_SESSIONS_EXPIRED = m->AddCounter<int64_t>(
+ NUM_SESSIONS_EXPIRED = m->AddCounter(
ImpaladMetricKeys::NUM_SESSIONS_EXPIRED, 0);
- RESULTSET_CACHE_TOTAL_NUM_ROWS = m->AddGauge<int64_t>(
+ RESULTSET_CACHE_TOTAL_NUM_ROWS = m->AddGauge(
ImpaladMetricKeys::RESULTSET_CACHE_TOTAL_NUM_ROWS, 0);
- RESULTSET_CACHE_TOTAL_BYTES = m->AddGauge<int64_t>(
+ RESULTSET_CACHE_TOTAL_BYTES = m->AddGauge(
ImpaladMetricKeys::RESULTSET_CACHE_TOTAL_BYTES, 0);
// Initialize scan node metrics
- NUM_RANGES_PROCESSED = m->AddCounter<int64_t>(
+ NUM_RANGES_PROCESSED = m->AddCounter(
ImpaladMetricKeys::TOTAL_SCAN_RANGES_PROCESSED, 0);
- NUM_RANGES_MISSING_VOLUME_ID = m->AddCounter<int64_t>(
+ NUM_RANGES_MISSING_VOLUME_ID = m->AddCounter(
ImpaladMetricKeys::NUM_SCAN_RANGES_MISSING_VOLUME_ID, 0);
// Initialize memory usage metrics
- MEM_POOL_TOTAL_BYTES = m->AddGauge<int64_t>(
+ MEM_POOL_TOTAL_BYTES = m->AddGauge(
ImpaladMetricKeys::MEM_POOL_TOTAL_BYTES, 0);
- HASH_TABLE_TOTAL_BYTES = m->AddGauge<int64_t>(
+ HASH_TABLE_TOTAL_BYTES = m->AddGauge(
ImpaladMetricKeys::HASH_TABLE_TOTAL_BYTES, 0);
// Initialize insert metrics
- NUM_FILES_OPEN_FOR_INSERT = m->AddGauge<int64_t>(
+ NUM_FILES_OPEN_FOR_INSERT = m->AddGauge(
ImpaladMetricKeys::NUM_FILES_OPEN_FOR_INSERT, 0);
// Initialize IO mgr metrics
- IO_MGR_NUM_OPEN_FILES = m->AddGauge<int64_t>(
- ImpaladMetricKeys::IO_MGR_NUM_OPEN_FILES, 0);
- IO_MGR_NUM_BUFFERS = m->AddGauge<int64_t>(ImpaladMetricKeys::IO_MGR_NUM_BUFFERS, 0);
- IO_MGR_TOTAL_BYTES = m->AddGauge<int64_t>(ImpaladMetricKeys::IO_MGR_TOTAL_BYTES, 0);
- IO_MGR_NUM_UNUSED_BUFFERS = m->AddGauge<int64_t>(
+ IO_MGR_NUM_OPEN_FILES = m->AddGauge(ImpaladMetricKeys::IO_MGR_NUM_OPEN_FILES, 0);
+ IO_MGR_NUM_BUFFERS = m->AddGauge(ImpaladMetricKeys::IO_MGR_NUM_BUFFERS, 0);
+ IO_MGR_TOTAL_BYTES = m->AddGauge(ImpaladMetricKeys::IO_MGR_TOTAL_BYTES, 0);
+ IO_MGR_NUM_UNUSED_BUFFERS = m->AddGauge(
ImpaladMetricKeys::IO_MGR_NUM_UNUSED_BUFFERS, 0);
- IO_MGR_NUM_CACHED_FILE_HANDLES = m->AddGauge<int64_t>(
+ IO_MGR_NUM_CACHED_FILE_HANDLES = m->AddGauge(
ImpaladMetricKeys::IO_MGR_NUM_CACHED_FILE_HANDLES, 0);
- IO_MGR_NUM_FILE_HANDLES_OUTSTANDING = m->AddGauge<int64_t>(
+ IO_MGR_NUM_FILE_HANDLES_OUTSTANDING = m->AddGauge(
ImpaladMetricKeys::IO_MGR_NUM_FILE_HANDLES_OUTSTANDING, 0);
- IO_MGR_CACHED_FILE_HANDLES_HIT_COUNT = m->AddGauge<int64_t>(
+ IO_MGR_CACHED_FILE_HANDLES_HIT_COUNT = m->AddGauge(
ImpaladMetricKeys::IO_MGR_CACHED_FILE_HANDLES_HIT_COUNT, 0);
- IO_MGR_CACHED_FILE_HANDLES_MISS_COUNT = m->AddGauge<int64_t>(
+ IO_MGR_CACHED_FILE_HANDLES_MISS_COUNT = m->AddGauge(
ImpaladMetricKeys::IO_MGR_CACHED_FILE_HANDLES_MISS_COUNT, 0);
- IO_MGR_BYTES_READ = m->AddCounter<int64_t>(ImpaladMetricKeys::IO_MGR_BYTES_READ, 0);
- IO_MGR_LOCAL_BYTES_READ = m->AddCounter<int64_t>(
+ IO_MGR_BYTES_READ = m->AddCounter(ImpaladMetricKeys::IO_MGR_BYTES_READ, 0);
+ IO_MGR_LOCAL_BYTES_READ = m->AddCounter(
ImpaladMetricKeys::IO_MGR_LOCAL_BYTES_READ, 0);
- IO_MGR_CACHED_BYTES_READ = m->AddCounter<int64_t>(
+ IO_MGR_CACHED_BYTES_READ = m->AddCounter(
ImpaladMetricKeys::IO_MGR_CACHED_BYTES_READ, 0);
- IO_MGR_SHORT_CIRCUIT_BYTES_READ = m->AddCounter<int64_t>(
+ IO_MGR_SHORT_CIRCUIT_BYTES_READ = m->AddCounter(
ImpaladMetricKeys::IO_MGR_SHORT_CIRCUIT_BYTES_READ, 0);
- IO_MGR_BYTES_WRITTEN = m->AddCounter<int64_t>(
+ IO_MGR_BYTES_WRITTEN = m->AddCounter(
ImpaladMetricKeys::IO_MGR_BYTES_WRITTEN, 0);
IO_MGR_CACHED_FILE_HANDLES_HIT_RATIO =
@@ -233,8 +233,8 @@ void ImpaladMetrics::CreateMetrics(MetricGroup* m) {
ImpaladMetricKeys::IO_MGR_CACHED_FILE_HANDLES_HIT_RATIO);
// Initialize catalog metrics
- CATALOG_NUM_DBS = m->AddGauge<int64_t>(ImpaladMetricKeys::CATALOG_NUM_DBS, 0);
- CATALOG_NUM_TABLES = m->AddGauge<int64_t>(ImpaladMetricKeys::CATALOG_NUM_TABLES, 0);
+ CATALOG_NUM_DBS = m->AddGauge(ImpaladMetricKeys::CATALOG_NUM_DBS, 0);
+ CATALOG_NUM_TABLES = m->AddGauge(ImpaladMetricKeys::CATALOG_NUM_TABLES, 0);
CATALOG_READY = m->AddProperty<bool>(ImpaladMetricKeys::CATALOG_READY, false);
// Maximum duration to be tracked by the query durations metric. No particular reasoning
@@ -248,8 +248,8 @@ void ImpaladMetrics::CreateMetrics(MetricGroup* m) {
MetricDefs::Get(ImpaladMetricKeys::DDL_DURATIONS), FIVE_HOURS_IN_MS, 3));
// Initialize Hedged read metrics
- HEDGED_READ_OPS = m->AddCounter<int64_t>(ImpaladMetricKeys::HEDGED_READ_OPS, 0);
- HEDGED_READ_OPS_WIN = m->AddCounter<int64_t>(ImpaladMetricKeys::HEDGED_READ_OPS_WIN, 0);
+ HEDGED_READ_OPS = m->AddCounter(ImpaladMetricKeys::HEDGED_READ_OPS, 0);
+ HEDGED_READ_OPS_WIN = m->AddCounter(ImpaladMetricKeys::HEDGED_READ_OPS_WIN, 0);
}
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/util/memory-metrics.cc
----------------------------------------------------------------------
diff --git a/be/src/util/memory-metrics.cc b/be/src/util/memory-metrics.cc
index 3308bf4..fd78343 100644
--- a/be/src/util/memory-metrics.cc
+++ b/be/src/util/memory-metrics.cc
@@ -32,7 +32,7 @@ using namespace strings;
DECLARE_bool(mmap_buffers);
-SumGauge<int64_t>* AggregateMemoryMetrics::TOTAL_USED = nullptr;
+SumGauge* AggregateMemoryMetrics::TOTAL_USED = nullptr;
IntGauge* AggregateMemoryMetrics::NUM_MAPS = nullptr;
IntGauge* AggregateMemoryMetrics::MAPPED_BYTES = nullptr;
IntGauge* AggregateMemoryMetrics::RSS = nullptr;
@@ -110,19 +110,19 @@ Status impala::RegisterMemoryMetrics(MetricGroup* metrics, bool register_jvm_met
#endif
MetricGroup* aggregate_metrics = metrics->GetOrCreateChildGroup("memory");
AggregateMemoryMetrics::TOTAL_USED = aggregate_metrics->RegisterMetric(
- new SumGauge<int64_t>(MetricDefs::Get("memory.total-used"), used_metrics));
+ new SumGauge(MetricDefs::Get("memory.total-used"), used_metrics));
if (register_jvm_metrics) {
RETURN_IF_ERROR(JvmMetric::InitMetrics(metrics->GetOrCreateChildGroup("jvm")));
}
if (MemInfo::HaveSmaps()) {
AggregateMemoryMetrics::NUM_MAPS =
- aggregate_metrics->AddGauge<int64_t>("memory.num-maps", 0U);
+ aggregate_metrics->AddGauge("memory.num-maps", 0U);
AggregateMemoryMetrics::MAPPED_BYTES =
- aggregate_metrics->AddGauge<int64_t>("memory.mapped-bytes", 0U);
- AggregateMemoryMetrics::RSS = aggregate_metrics->AddGauge<int64_t>("memory.rss", 0U);
+ aggregate_metrics->AddGauge("memory.mapped-bytes", 0U);
+ AggregateMemoryMetrics::RSS = aggregate_metrics->AddGauge("memory.rss", 0U);
AggregateMemoryMetrics::ANON_HUGE_PAGE_BYTES =
- aggregate_metrics->AddGauge<int64_t>("memory.anon-huge-page-bytes", 0U);
+ aggregate_metrics->AddGauge("memory.anon-huge-page-bytes", 0U);
}
ThpConfig thp_config = MemInfo::ParseThpConfig();
AggregateMemoryMetrics::THP_ENABLED =
@@ -139,16 +139,16 @@ void AggregateMemoryMetrics::Refresh() {
if (NUM_MAPS != nullptr) {
// Only call ParseSmaps() if the metrics were created.
MappedMemInfo map_info = MemInfo::ParseSmaps();
- NUM_MAPS->set_value(map_info.num_maps);
- MAPPED_BYTES->set_value(map_info.size_kb * 1024);
- RSS->set_value(map_info.rss_kb * 1024);
- ANON_HUGE_PAGE_BYTES->set_value(map_info.anon_huge_pages_kb * 1024);
+ NUM_MAPS->SetValue(map_info.num_maps);
+ MAPPED_BYTES->SetValue(map_info.size_kb * 1024);
+ RSS->SetValue(map_info.rss_kb * 1024);
+ ANON_HUGE_PAGE_BYTES->SetValue(map_info.anon_huge_pages_kb * 1024);
}
ThpConfig thp_config = MemInfo::ParseThpConfig();
- THP_ENABLED->set_value(thp_config.enabled);
- THP_DEFRAG->set_value(thp_config.defrag);
- THP_KHUGEPAGED_DEFRAG->set_value(thp_config.khugepaged_defrag);
+ THP_ENABLED->SetValue(thp_config.enabled);
+ THP_DEFRAG->SetValue(thp_config.defrag);
+ THP_KHUGEPAGED_DEFRAG->SetValue(thp_config.khugepaged_defrag);
}
JvmMetric* JvmMetric::CreateAndRegister(MetricGroup* metrics, const string& key,
@@ -192,35 +192,36 @@ Status JvmMetric::InitMetrics(MetricGroup* metrics) {
return Status::OK();
}
-void JvmMetric::CalculateValue() {
+int64_t JvmMetric::GetValue() {
TGetJvmMetricsRequest request;
request.get_all = false;
request.__set_memory_pool(mempool_name_);
TGetJvmMetricsResponse response;
- if (!JniUtil::GetJvmMetrics(request, &response).ok()) return;
- if (response.memory_pools.size() != 1) return;
+ if (!JniUtil::GetJvmMetrics(request, &response).ok()) return 0;
+ if (response.memory_pools.size() != 1) return 0;
TJvmMemoryPool& pool = response.memory_pools[0];
DCHECK(pool.name == mempool_name_);
switch (metric_type_) {
- case MAX: value_ = pool.max;
- return;
- case INIT: value_ = pool.init;
- return;
- case CURRENT: value_ = pool.used;
- return;
- case COMMITTED: value_ = pool.committed;
- return;
- case PEAK_MAX: value_ = pool.peak_max;
- return;
- case PEAK_INIT: value_ = pool.peak_init;
- return;
- case PEAK_CURRENT: value_ = pool.peak_used;
- return;
- case PEAK_COMMITTED: value_ = pool.peak_committed;
- return;
+ case MAX:
+ return pool.max;
+ case INIT:
+ return pool.init;
+ case CURRENT:
+ return pool.used;
+ case COMMITTED:
+ return pool.committed;
+ case PEAK_MAX:
+ return pool.peak_max;
+ case PEAK_INIT:
+ return pool.peak_init;
+ case PEAK_CURRENT:
+ return pool.peak_used;
+ case PEAK_COMMITTED:
+ return pool.peak_committed;
default:
DCHECK(false) << "Unknown JvmMetricType: " << metric_type_;
}
+ return 0;
}
Status BufferPoolMetric::InitMetrics(MetricGroup* metrics,
@@ -263,47 +264,39 @@ BufferPoolMetric::BufferPoolMetric(const TMetricDef& def, BufferPoolMetricType t
global_reservations_(global_reservations),
buffer_pool_(buffer_pool) {}
-void BufferPoolMetric::CalculateValue() {
+int64_t BufferPoolMetric::GetValue() {
// IMPALA-6362: we have to be careful that none of the below calls to ReservationTracker
// methods acquire ReservationTracker::lock_ to avoid a potential circular dependency
// with MemTracker::child_trackers_lock_, which may be held when refreshing MemTracker
// consumption.
switch (type_) {
case BufferPoolMetricType::LIMIT:
- value_ = buffer_pool_->GetSystemBytesLimit();
- break;
+ return buffer_pool_->GetSystemBytesLimit();
case BufferPoolMetricType::SYSTEM_ALLOCATED:
- value_ = buffer_pool_->GetSystemBytesAllocated();
- break;
+ return buffer_pool_->GetSystemBytesAllocated();
case BufferPoolMetricType::RESERVED:
- value_ = global_reservations_->GetReservation();
- break;
+ return global_reservations_->GetReservation();
case BufferPoolMetricType::UNUSED_RESERVATION_BYTES: {
// Estimate the unused reservation based on other aggregate values, defined as
// the total bytes of reservation where there is no corresponding buffer in use
// by a client. Buffers are either in-use, free buffers, or attached to clean pages.
int64_t total_used_reservation = buffer_pool_->GetSystemBytesAllocated()
- - buffer_pool_->GetFreeBufferBytes()
- - buffer_pool_->GetCleanPageBytes();
- value_ = global_reservations_->GetReservation() - total_used_reservation;
- break;
+ - buffer_pool_->GetFreeBufferBytes()
+ - buffer_pool_->GetCleanPageBytes();
+ return global_reservations_->GetReservation() - total_used_reservation;
}
case BufferPoolMetricType::NUM_FREE_BUFFERS:
- value_ = buffer_pool_->GetNumFreeBuffers();
- break;
+ return buffer_pool_->GetNumFreeBuffers();
case BufferPoolMetricType::FREE_BUFFER_BYTES:
- value_ = buffer_pool_->GetFreeBufferBytes();
- break;
+ return buffer_pool_->GetFreeBufferBytes();
case BufferPoolMetricType::CLEAN_PAGES_LIMIT:
- value_ = buffer_pool_->GetCleanPageBytesLimit();
- break;
+ return buffer_pool_->GetCleanPageBytesLimit();
case BufferPoolMetricType::NUM_CLEAN_PAGES:
- value_ = buffer_pool_->GetNumCleanPages();
- break;
+ return buffer_pool_->GetNumCleanPages();
case BufferPoolMetricType::CLEAN_PAGE_BYTES:
- value_ = buffer_pool_->GetCleanPageBytes();
- break;
+ return buffer_pool_->GetCleanPageBytes();
default:
DCHECK(false) << "Unknown BufferPoolMetricType: " << static_cast<int>(type_);
}
+ return 0;
}
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/util/memory-metrics.h
----------------------------------------------------------------------
diff --git a/be/src/util/memory-metrics.h b/be/src/util/memory-metrics.h
index 3294c30..6c10e09 100644
--- a/be/src/util/memory-metrics.h
+++ b/be/src/util/memory-metrics.h
@@ -44,7 +44,7 @@ class AggregateMemoryMetrics {
/// including JVM memory), which is either in use by queries or cached by the BufferPool
/// or the malloc implementation.
/// TODO: IMPALA-691 - consider changing this to include JVM memory.
- static SumGauge<int64_t>* TOTAL_USED;
+ static SumGauge* TOTAL_USED;
/// The total number of virtual memory regions for the process.
/// The value must be refreshed by calling Refresh().
@@ -106,9 +106,8 @@ class TcmallocMetric : public IntGauge {
public:
PhysicalBytesMetric(const TMetricDef& def) : IntGauge(def, 0) { }
- private:
- virtual void CalculateValue() {
- value_ = TOTAL_BYTES_RESERVED->value() - PAGEHEAP_UNMAPPED_BYTES->value();
+ virtual int64_t GetValue() override {
+ return TOTAL_BYTES_RESERVED->GetValue() - PAGEHEAP_UNMAPPED_BYTES->GetValue();
}
};
@@ -117,20 +116,21 @@ class TcmallocMetric : public IntGauge {
static TcmallocMetric* CreateAndRegister(MetricGroup* metrics, const std::string& key,
const std::string& tcmalloc_var);
+ virtual int64_t GetValue() override {
+ int64_t retval = 0;
+#if !defined(ADDRESS_SANITIZER) && !defined(THREAD_SANITIZER)
+ MallocExtension::instance()->GetNumericProperty(tcmalloc_var_.c_str(),
+ reinterpret_cast<size_t*>(&retval));
+#endif
+ return retval;
+ }
+
private:
/// Name of the tcmalloc property this metric should fetch.
const std::string tcmalloc_var_;
TcmallocMetric(const TMetricDef& def, const std::string& tcmalloc_var)
- : IntGauge(def, 0), tcmalloc_var_(tcmalloc_var) { }
-
- virtual void CalculateValue() {
-#if !defined(ADDRESS_SANITIZER) && !defined(THREAD_SANITIZER)
- DCHECK_EQ(sizeof(size_t), sizeof(value_));
- MallocExtension::instance()->GetNumericProperty(tcmalloc_var_.c_str(),
- reinterpret_cast<size_t*>(&value_));
-#endif
- }
+ : IntGauge(def, 0), tcmalloc_var_(tcmalloc_var) { }
};
/// Alternative to TCMallocMetric if we're running under a sanitizer that replaces
@@ -138,12 +138,16 @@ class TcmallocMetric : public IntGauge {
class SanitizerMallocMetric : public IntGauge {
public:
SanitizerMallocMetric(const TMetricDef& def) : IntGauge(def, 0) {}
+
static SanitizerMallocMetric* BYTES_ALLOCATED;
- private:
- virtual void CalculateValue() override {
+
+ virtual int64_t GetValue() override {
#if defined(ADDRESS_SANITIZER) || defined(THREAD_SANITIZER)
- value_ = __sanitizer_get_current_allocated_bytes();
+ return __sanitizer_get_current_allocated_bytes();
+#else
+ return 0;
#endif
+
}
};
@@ -157,10 +161,9 @@ class JvmMetric : public IntGauge {
/// pool (usually ~5 pools plus a synthetic 'total' pool).
static Status InitMetrics(MetricGroup* metrics) WARN_UNUSED_RESULT;
- protected:
/// Searches through jvm_metrics_response_ for a matching memory pool and pulls out the
/// right value from that structure according to metric_type_.
- virtual void CalculateValue();
+ virtual int64_t GetValue() override;
private:
/// Each names one of the fields in TJvmMemoryPool.
@@ -206,8 +209,7 @@ class BufferPoolMetric : public IntGauge {
static BufferPoolMetric* NUM_CLEAN_PAGES;
static BufferPoolMetric* CLEAN_PAGE_BYTES;
- protected:
- virtual void CalculateValue();
+ virtual int64_t GetValue() override;
private:
friend class ReservationTrackerTest;
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/util/metrics-test.cc
----------------------------------------------------------------------
diff --git a/be/src/util/metrics-test.cc b/be/src/util/metrics-test.cc
index 0126281..bfbfdfe 100644
--- a/be/src/util/metrics-test.cc
+++ b/be/src/util/metrics-test.cc
@@ -36,7 +36,7 @@ namespace impala {
template <typename M, typename T>
void AssertValue(M* metric, const T& value,
const string& human_readable) {
- EXPECT_EQ(metric->value(), value);
+ EXPECT_EQ(metric->GetValue(), value);
if (!human_readable.empty()) {
EXPECT_EQ(metric->ToHumanReadable(), human_readable);
}
@@ -73,36 +73,36 @@ class MetricsTest : public testing::Test {
TEST_F(MetricsTest, CounterMetrics) {
MetricGroup metrics("CounterMetrics");
AddMetricDef("counter", TMetricKind::COUNTER, TUnit::UNIT);
- IntCounter* int_counter = metrics.AddCounter<int64_t>("counter", 0);
+ IntCounter* int_counter = metrics.AddCounter("counter", 0);
AssertValue(int_counter, 0, "0");
int_counter->Increment(1);
AssertValue(int_counter, 1, "1");
int_counter->Increment(10);
AssertValue(int_counter, 11, "11");
- int_counter->set_value(3456);
+ int_counter->SetValue(3456);
AssertValue(int_counter, 3456, "3.46K");
AddMetricDef("counter_with_units", TMetricKind::COUNTER, TUnit::BYTES);
IntCounter* int_counter_with_units =
- metrics.AddCounter<int64_t>("counter_with_units", 10);
+ metrics.AddCounter("counter_with_units", 10);
AssertValue(int_counter_with_units, 10, "10.00 B");
}
TEST_F(MetricsTest, GaugeMetrics) {
MetricGroup metrics("GaugeMetrics");
AddMetricDef("gauge", TMetricKind::GAUGE, TUnit::NONE);
- IntGauge* int_gauge = metrics.AddGauge<int64_t>("gauge", 0);
+ IntGauge* int_gauge = metrics.AddGauge("gauge", 0);
AssertValue(int_gauge, 0, "0");
int_gauge->Increment(-1);
AssertValue(int_gauge, -1, "-1");
int_gauge->Increment(10);
AssertValue(int_gauge, 9, "9");
- int_gauge->set_value(3456);
+ int_gauge->SetValue(3456);
AssertValue(int_gauge, 3456, "3456");
AddMetricDef("gauge_with_units", TMetricKind::GAUGE, TUnit::TIME_S);
IntGauge* int_gauge_with_units =
- metrics.AddGauge<int64_t>("gauge_with_units", 10);
+ metrics.AddGauge("gauge_with_units", 10);
AssertValue(int_gauge_with_units, 10, "10s000ms");
}
@@ -111,12 +111,12 @@ TEST_F(MetricsTest, SumGauge) {
AddMetricDef("gauge1", TMetricKind::GAUGE, TUnit::NONE);
AddMetricDef("gauge2", TMetricKind::GAUGE, TUnit::NONE);
AddMetricDef("sum", TMetricKind::GAUGE, TUnit::NONE);
- IntGauge* gauge1 = metrics.AddGauge<int64_t>("gauge1", 0);
- IntGauge* gauge2 = metrics.AddGauge<int64_t>("gauge2", 0);
+ IntGauge* gauge1 = metrics.AddGauge("gauge1", 0);
+ IntGauge* gauge2 = metrics.AddGauge("gauge2", 0);
vector<IntGauge*> gauges({gauge1, gauge2});
IntGauge* sum_gauge =
- metrics.RegisterMetric(new SumGauge<int64_t>(MetricDefs::Get("sum"), gauges));
+ metrics.RegisterMetric(new SumGauge(MetricDefs::Get("sum"), gauges));
AssertValue(sum_gauge, 0, "0");
gauge1->Increment(1);
@@ -132,14 +132,14 @@ TEST_F(MetricsTest, PropertyMetrics) {
AddMetricDef("bool_property", TMetricKind::PROPERTY, TUnit::NONE);
BooleanProperty* bool_property = metrics.AddProperty("bool_property", false);
AssertValue(bool_property, false, "false");
- bool_property->set_value(true);
+ bool_property->SetValue(true);
AssertValue(bool_property, true, "true");
AddMetricDef("string_property", TMetricKind::PROPERTY, TUnit::NONE);
StringProperty* string_property = metrics.AddProperty("string_property",
string("string1"));
AssertValue(string_property, "string1", "string1");
- string_property->set_value("string2");
+ string_property->SetValue("string2");
AssertValue(string_property, "string2", "string2");
}
@@ -147,11 +147,11 @@ TEST_F(MetricsTest, NonFiniteValues) {
MetricGroup metrics("NanValues");
AddMetricDef("inf_value", TMetricKind::GAUGE, TUnit::NONE);
double inf = numeric_limits<double>::infinity();
- DoubleGauge* gauge = metrics.AddGauge("inf_value", inf);
+ DoubleGauge* gauge = metrics.AddDoubleGauge("inf_value", inf);
AssertValue(gauge, inf, "inf");
double nan = numeric_limits<double>::quiet_NaN();
- gauge->set_value(nan);
- EXPECT_TRUE(std::isnan(gauge->value()));
+ gauge->SetValue(nan);
+ EXPECT_TRUE(std::isnan(gauge->GetValue()));
EXPECT_TRUE(gauge->ToHumanReadable() == "nan");
}
@@ -223,19 +223,19 @@ TEST_F(MetricsTest, MemMetric) {
metrics.FindMetricForTesting<IntGauge>("tcmalloc.bytes-in-use");
ASSERT_TRUE(bytes_in_use != NULL);
- uint64_t cur_in_use = bytes_in_use->value();
+ uint64_t cur_in_use = bytes_in_use->GetValue();
EXPECT_GT(cur_in_use, 0);
// Allocate 100MB to increase the number of bytes used. TCMalloc may also give up some
// bytes during this allocation, so this allocation is deliberately large to ensure that
// the bytes used metric goes up net.
scoped_ptr<vector<uint64_t>> chunk(new vector<uint64_t>(100 * 1024 * 1024));
- EXPECT_GT(bytes_in_use->value(), cur_in_use);
+ EXPECT_GT(bytes_in_use->GetValue(), cur_in_use);
IntGauge* total_bytes_reserved =
metrics.FindMetricForTesting<IntGauge>("tcmalloc.total-bytes-reserved");
ASSERT_TRUE(total_bytes_reserved != NULL);
- ASSERT_GT(total_bytes_reserved->value(), 0);
+ ASSERT_GT(total_bytes_reserved->GetValue(), 0);
IntGauge* pageheap_free_bytes =
metrics.FindMetricForTesting<IntGauge>("tcmalloc.pageheap-free-bytes");
@@ -254,12 +254,12 @@ TEST_F(MetricsTest, JvmMetrics) {
metrics.GetOrCreateChildGroup("jvm")->FindMetricForTesting<IntGauge>(
"jvm.total.current-usage-bytes");
ASSERT_TRUE(jvm_total_used != NULL);
- EXPECT_GT(jvm_total_used->value(), 0);
+ EXPECT_GT(jvm_total_used->GetValue(), 0);
IntGauge* jvm_peak_total_used =
metrics.GetOrCreateChildGroup("jvm")->FindMetricForTesting<IntGauge>(
"jvm.total.peak-current-usage-bytes");
ASSERT_TRUE(jvm_peak_total_used != NULL);
- EXPECT_GT(jvm_peak_total_used->value(), 0);
+ EXPECT_GT(jvm_peak_total_used->GetValue(), 0);
}
void AssertJson(const Value& val, const string& name, const string& value,
@@ -274,7 +274,7 @@ void AssertJson(const Value& val, const string& name, const string& value,
TEST_F(MetricsTest, CountersJson) {
MetricGroup metrics("CounterMetrics");
AddMetricDef("counter", TMetricKind::COUNTER, TUnit::UNIT, "description");
- metrics.AddCounter<int64_t>("counter", 0);
+ metrics.AddCounter("counter", 0);
Document document;
Value val;
metrics.ToJson(true, &document, &val);
@@ -286,7 +286,7 @@ TEST_F(MetricsTest, CountersJson) {
TEST_F(MetricsTest, GaugesJson) {
MetricGroup metrics("GaugeMetrics");
AddMetricDef("gauge", TMetricKind::GAUGE, TUnit::NONE);
- metrics.AddGauge<int64_t>("gauge", 10);
+ metrics.AddGauge("gauge", 10);
Document document;
Value val;
metrics.ToJson(true, &document, &val);
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/util/metrics.h
----------------------------------------------------------------------
diff --git a/be/src/util/metrics.h b/be/src/util/metrics.h
index 12d6df3..b513c1e 100644
--- a/be/src/util/metrics.h
+++ b/be/src/util/metrics.h
@@ -27,6 +27,7 @@
#include <boost/scoped_ptr.hpp>
#include <boost/thread/locks.hpp>
+#include "common/atomic.h"
#include "common/logging.h"
#include "common/object-pool.h"
#include "common/status.h"
@@ -118,59 +119,37 @@ class Metric {
void AddStandardFields(rapidjson::Document* document, rapidjson::Value* val);
};
-/// A SimpleMetric has a value which is a simple primitive type: e.g. integers, strings and
-/// floats. It is parameterised not only by the type of its value, but by both the unit
-/// (e.g. bytes/s), drawn from TUnit and the 'kind' of the metric itself. The kind
-/// can be one of: 'gauge', which may increase or decrease over time, a 'counter' which is
-/// increasing only over time, or a 'property' which is not numeric.
-//
-/// SimpleMetrics return their current value through the value() method. Access to value()
-/// is thread-safe.
-//
-/// TODO: We can use type traits to select a more efficient lock-free implementation of
-/// value() etc. where it is safe to do so.
-/// TODO: CalculateValue() can be returning a value, its current interface is not clean.
-template<typename T, TMetricKind::type metric_kind=TMetricKind::GAUGE>
-class SimpleMetric : public Metric {
+/// A ScalarMetric has a value which is a simple primitive type: e.g. integers, strings
+/// and floats. It is parameterised not only by the type of its value, but by both the
+/// unit (e.g. bytes/s), drawn from TUnit and the 'kind' of the metric itself.
+/// The kind can be one of:
+/// - 'gauge', which may increase or decrease over time,
+/// - 'counter' which can only increase over time
+/// - 'property' which is a value store which can be read and written only
+///
+/// Note that management software may use the metric kind as hint on how to display
+/// the value. ScalarMetrics return their current value through the GetValue() method
+/// and set/initialize the value with SetValue(). Both methods are thread safe.
+template<typename T, TMetricKind::type metric_kind_t>
+class ScalarMetric: public Metric {
public:
- SimpleMetric(const TMetricDef& metric_def, const T& initial_value)
- : Metric(metric_def), unit_(metric_def.units), value_(initial_value) {
- DCHECK_EQ(metric_kind, metric_def.kind) << "Metric kind does not match definition: "
+ ScalarMetric(const TMetricDef& metric_def)
+ : Metric(metric_def), unit_(metric_def.units) {
+ DCHECK_EQ(metric_kind_t, metric_def.kind) << "Metric kind does not match definition: "
<< metric_def.key;
}
- virtual ~SimpleMetric() { }
-
- /// Returns the current value, updating it if necessary. Thread-safe.
- T value() {
- boost::lock_guard<SpinLock> l(lock_);
- CalculateValue();
- return value_;
- }
-
- /// Sets the current value. Thread-safe.
- void set_value(const T& value) {
- boost::lock_guard<SpinLock> l(lock_);
- value_ = value;
- }
+ virtual ~ScalarMetric() { }
- /// Adds 'delta' to the current value atomically.
- void Increment(const T& delta) {
- DCHECK(kind() != TMetricKind::PROPERTY)
- << "Can't change value of PROPERTY metric: " << key();
- DCHECK(kind() != TMetricKind::COUNTER || delta >= 0)
- << "Can't decrement value of COUNTER metric: " << key();
- if (delta == 0) return;
- boost::lock_guard<SpinLock> l(lock_);
- value_ += delta;
- }
+ /// Returns the current value. Thread-safe.
+ virtual T GetValue() = 0;
- virtual void ToJson(rapidjson::Document* document, rapidjson::Value* val) {
+ virtual void ToJson(rapidjson::Document* document, rapidjson::Value* val) override {
rapidjson::Value container(rapidjson::kObjectType);
AddStandardFields(document, &container);
rapidjson::Value metric_value;
- ToJsonValue(value(), TUnit::NONE, document, &metric_value);
+ ToJsonValue(GetValue(), TUnit::NONE, document, &metric_value);
container.AddMember("value", metric_value, document->GetAllocator());
rapidjson::Value type_value(PrintTMetricKind(kind()).c_str(),
@@ -181,30 +160,46 @@ class SimpleMetric : public Metric {
*val = container;
}
- virtual std::string ToHumanReadable() {
- return PrettyPrinter::Print(value(), unit());
+ virtual std::string ToHumanReadable() override {
+ return PrettyPrinter::Print(GetValue(), unit());
}
- virtual void ToLegacyJson(rapidjson::Document* document) {
+ virtual void ToLegacyJson(rapidjson::Document* document) override {
rapidjson::Value val;
- ToJsonValue(value(), TUnit::NONE, document, &val);
+ ToJsonValue(GetValue(), TUnit::NONE, document, &val);
document->AddMember(key_.c_str(), val, document->GetAllocator());
}
TUnit::type unit() const { return unit_; }
- TMetricKind::type kind() const { return metric_kind; }
+ TMetricKind::type kind() const { return metric_kind_t; }
protected:
- /// Called to compute value_ if necessary during calls to value(). The more natural
- /// approach would be to have virtual T value(), but that's not possible in C++.
- //
- /// TODO: Should be cheap to have a blank implementation, but if required we can cause
- /// the compiler to avoid calling this entirely through a compile-time constant.
- virtual void CalculateValue() { }
-
/// Units of this metric.
const TUnit::type unit_;
+};
+/// An implementation of scalar metric with spinlock.
+template<typename T, TMetricKind::type metric_kind_t>
+class LockedMetric : public ScalarMetric<T, metric_kind_t> {
+ public:
+ LockedMetric(const TMetricDef& metric_def, const T& initial_value)
+ : ScalarMetric<T, metric_kind_t>(metric_def), value_(initial_value) {}
+
+ virtual ~LockedMetric() {}
+
+ /// Atomically reads the current value.
+ virtual T GetValue() override {
+ boost::lock_guard<SpinLock> l(lock_);
+ return value_;
+ }
+
+ /// Atomically sets the value.
+ void SetValue(const T& value) {
+ boost::lock_guard<SpinLock> l(lock_);
+ value_ = value;
+ }
+
+ protected:
/// Guards access to value_.
SpinLock lock_;
@@ -212,42 +207,81 @@ class SimpleMetric : public Metric {
T value_;
};
-// Gauge metric that computes the sum of several gauges.
-template <typename T>
-class SumGauge : public SimpleMetric<T, TMetricKind::GAUGE> {
+typedef class LockedMetric<bool, TMetricKind::PROPERTY> BooleanProperty;
+typedef class LockedMetric<std::string,TMetricKind::PROPERTY> StringProperty;
+typedef class LockedMetric<double, TMetricKind::GAUGE> DoubleGauge;
+
+/// An implementation of 'gauge' or 'counter' metric kind. The metric can be incremented
+/// atomically via the Increment() interface.
+template<TMetricKind::type metric_kind_t>
+class AtomicMetric : public ScalarMetric<int64_t, metric_kind_t> {
public:
- SumGauge(const TMetricDef& metric_def,
- const std::vector<SimpleMetric<T, TMetricKind::GAUGE>*>& metrics)
- : SimpleMetric<T, TMetricKind::GAUGE>(metric_def, 0), metrics_(metrics) {}
+ AtomicMetric(const TMetricDef& metric_def, const int64_t initial_value)
+ : ScalarMetric<int64_t, metric_kind_t>(metric_def), value_(initial_value) {
+ DCHECK(metric_kind_t == TMetricKind::GAUGE || metric_kind_t == TMetricKind::COUNTER);
+ }
+
+ virtual ~AtomicMetric() {}
+
+ /// Atomically reads the current value. May be overridden by derived classes.
+ /// The default implementation just atomically loads 'value_'. Derived classes
+ /// which derive the return value from mutliple sources other than 'value_'
+ /// need to take care of synchronization among sources.
+ virtual int64_t GetValue() override { return value_.Load(); }
+
+ /// Atomically sets the value.
+ void SetValue(const int64_t& value) { value_.Store(value); }
+
+ /// Adds 'delta' to the current value atomically.
+ void Increment(int64_t delta) {
+ DCHECK(metric_kind_t != TMetricKind::COUNTER || delta >= 0)
+ << "Can't decrement value of COUNTER metric: " << this->key();
+ value_.Add(delta);
+ }
+
+ protected:
+ /// The current value of the metric.
+ AtomicInt64 value_;
+};
+
+/// We write 'Int' as a placeholder for all integer types.
+typedef class AtomicMetric<TMetricKind::GAUGE> IntGauge;
+typedef class AtomicMetric<TMetricKind::COUNTER> IntCounter;
+
+/// Gauge metric that computes the sum of several gauges.
+class SumGauge : public IntGauge {
+ public:
+ SumGauge(const TMetricDef& metric_def, const std::vector<IntGauge*>& gauges)
+ : IntGauge(metric_def, 0), gauges_(gauges) {}
+
virtual ~SumGauge() {}
- private:
- virtual void CalculateValue() override {
- T sum = 0;
- for (SimpleMetric<T, TMetricKind::GAUGE>* metric : metrics_) sum += metric->value();
- this->value_ = sum;
+ virtual int64_t GetValue() override {
+ // Note that this doesn't hold the locks of all gauages before computing the sum so
+ // it's possible for one of the gauages to change after being read and added to sum.
+ int64_t sum = 0;
+ for (auto gauge : gauges_) sum += gauge->GetValue();
+ return sum;
}
- /// The metrics to be summed.
- std::vector<SimpleMetric<T, TMetricKind::GAUGE>*> metrics_;
+ private:
+ /// The gauges to be summed.
+ std::vector<IntGauge*> gauges_;
};
-// Gauge metric that negates another gauge.
-template <typename T>
-class NegatedGauge : public SimpleMetric<T, TMetricKind::GAUGE> {
+/// Gauge metric that negates another gauge.
+class NegatedGauge : public IntGauge {
public:
- NegatedGauge(const TMetricDef& metric_def,
- SimpleMetric<T, TMetricKind::GAUGE>* metric)
- : SimpleMetric<T, TMetricKind::GAUGE>(metric_def, 0), metric_(metric) {}
+ NegatedGauge(const TMetricDef& metric_def, IntGauge* gauge)
+ : IntGauge(metric_def, 0), gauge_(gauge) {}
+
virtual ~NegatedGauge() {}
- private:
- virtual void CalculateValue() override {
- this->value_ = -metric_->value();
- }
+ virtual int64_t GetValue() override { return -gauge_->GetValue(); }
+ private:
/// The metric to be negated.
- SimpleMetric<T, TMetricKind::GAUGE>* metric_;
+ IntGauge* gauge_;
};
/// Container for a set of metrics. A MetricGroup owns the memory for every metric
@@ -285,27 +319,28 @@ class MetricGroup {
}
/// Create a gauge metric object with given key and initial value (owned by this object)
- template<typename T>
- SimpleMetric<T>* AddGauge(const std::string& key, const T& value,
+ IntGauge* AddGauge(const std::string& key, const int64_t value,
const std::string& metric_def_arg = "") {
- return RegisterMetric(new SimpleMetric<T, TMetricKind::GAUGE>(
- MetricDefs::Get(key, metric_def_arg), value));
+ return RegisterMetric(new IntGauge(MetricDefs::Get(key, metric_def_arg), value));
}
- template<typename T>
- SimpleMetric<T, TMetricKind::PROPERTY>* AddProperty(const std::string& key,
- const T& value, const std::string& metric_def_arg = "") {
- return RegisterMetric(new SimpleMetric<T, TMetricKind::PROPERTY>(
- MetricDefs::Get(key, metric_def_arg), value));
+ DoubleGauge* AddDoubleGauge(const std::string& key, const double value,
+ const std::string& metric_def_arg = "") {
+ return RegisterMetric(new DoubleGauge(MetricDefs::Get(key, metric_def_arg), value));
}
template<typename T>
- SimpleMetric<T, TMetricKind::COUNTER>* AddCounter(const std::string& key,
+ LockedMetric<T, TMetricKind::PROPERTY>* AddProperty(const std::string& key,
const T& value, const std::string& metric_def_arg = "") {
- return RegisterMetric(new SimpleMetric<T, TMetricKind::COUNTER>(
+ return RegisterMetric(new LockedMetric<T, TMetricKind::PROPERTY>(
MetricDefs::Get(key, metric_def_arg), value));
}
+ IntCounter* AddCounter(const std::string& key, const int64_t value,
+ const std::string& metric_def_arg = "") {
+ return RegisterMetric(new IntCounter(MetricDefs::Get(key, metric_def_arg), value));
+ }
+
/// Returns a metric by key. All MetricGroups reachable from this group are searched in
/// depth-first order, starting with the root group. Returns NULL if there is no metric
/// with that key. This is not a very cheap operation; the result should be cached where
@@ -380,13 +415,6 @@ class MetricGroup {
rapidjson::Document* document);
};
-/// We write 'Int' as a placeholder for all integer types.
-typedef class SimpleMetric<int64_t, TMetricKind::GAUGE> IntGauge;
-typedef class SimpleMetric<double, TMetricKind::GAUGE> DoubleGauge;
-typedef class SimpleMetric<int64_t, TMetricKind::COUNTER> IntCounter;
-
-typedef class SimpleMetric<bool, TMetricKind::PROPERTY> BooleanProperty;
-typedef class SimpleMetric<std::string, TMetricKind::PROPERTY> StringProperty;
/// Convenience method to instantiate a TMetricDef with a subset of its fields defined.
/// Most externally-visible metrics should be defined in metrics.json and retrieved via
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/be/src/util/thread.cc
----------------------------------------------------------------------
diff --git a/be/src/util/thread.cc b/be/src/util/thread.cc
index 536119b..8397f35 100644
--- a/be/src/util/thread.cc
+++ b/be/src/util/thread.cc
@@ -194,10 +194,8 @@ Status ThreadMgr::StartInstrumentation(MetricGroup* metrics) {
DCHECK(metrics != NULL);
lock_guard<mutex> l(lock_);
metrics_enabled_ = true;
- total_threads_metric_ = metrics->AddGauge<int64_t>(
- "thread-manager.total-threads-created", 0L);
- current_num_threads_metric_ = metrics->AddGauge<int64_t>(
- "thread-manager.running-threads", 0L);
+ total_threads_metric_ = metrics->AddGauge("thread-manager.total-threads-created", 0L);
+ current_num_threads_metric_ = metrics->AddGauge("thread-manager.running-threads", 0L);
return Status::OK();
}
@@ -224,7 +222,7 @@ void ThreadMgr::RemoveThread(const thread::id& boost_id, const string& category)
void ThreadMgr::GetThreadOverview(Document* document) {
lock_guard<mutex> l(lock_);
if (metrics_enabled_) {
- document->AddMember("total_threads", current_num_threads_metric_->value(),
+ document->AddMember("total_threads", current_num_threads_metric_->GetValue(),
document->GetAllocator());
}
Value lst(kArrayType);
http://git-wip-us.apache.org/repos/asf/impala/blob/e714f2b3/common/thrift/metrics.json
----------------------------------------------------------------------
diff --git a/common/thrift/metrics.json b/common/thrift/metrics.json
index dafe986..f493d33 100644
--- a/common/thrift/metrics.json
+++ b/common/thrift/metrics.json
@@ -882,7 +882,7 @@
"IMPALAD"
],
"label": "StateStore Subscriber Last Recovery Duration",
- "units": "NONE",
+ "units": "TIME_S",
"kind": "GAUGE",
"key": "statestore-subscriber.last-recovery-duration"
},
[4/8] impala git commit: Revert "IMPALA-5528: Upgrade GPerfTools to
2.6.3 and tune TCMalloc for KRPC"
Posted by jr...@apache.org.
Revert "IMPALA-5528: Upgrade GPerfTools to 2.6.3 and tune TCMalloc for KRPC"
This reverts commit df3a440fff38225a03879955c99a87d8ced3b13a.
Apparently, linking Impalad against GPerfTools 2.6.3 caused Impalad to fail
on certain platforms (OLE6). The failure's symptom is SIGSEGV when trying to
exec Impalad binary. It's unclear which commit in GPerfTools could have caused
it so backing up this change to allow Impala to unbreak some platforms for now.
Change-Id: I97cccca74fb199d6ff0a42fe818f8789a0d66e83
Reviewed-on: http://gerrit.cloudera.org:8080/9057
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/b3d38b5c
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/b3d38b5c
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/b3d38b5c
Branch: refs/heads/2.x
Commit: b3d38b5c8650fb455ca556f321fb7aea35dbd5ee
Parents: 028a83e
Author: Michael Ho <kw...@cloudera.com>
Authored: Wed Jan 17 19:27:17 2018 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Thu Jan 18 23:25:09 2018 +0000
----------------------------------------------------------------------
.../runtime/bufferpool/buffer-allocator-test.cc | 7 -------
be/src/runtime/bufferpool/free-list-test.cc | 13 +-----------
be/src/runtime/bufferpool/suballocator-test.cc | 13 +-----------
be/src/runtime/exec-env.cc | 21 ++++++++++----------
bin/impala-config.sh | 4 ++--
5 files changed, 14 insertions(+), 44 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/b3d38b5c/be/src/runtime/bufferpool/buffer-allocator-test.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/bufferpool/buffer-allocator-test.cc b/be/src/runtime/bufferpool/buffer-allocator-test.cc
index 2b87278..21a9c08 100644
--- a/be/src/runtime/bufferpool/buffer-allocator-test.cc
+++ b/be/src/runtime/bufferpool/buffer-allocator-test.cc
@@ -22,8 +22,6 @@
#include "runtime/bufferpool/buffer-pool-internal.h"
#include "runtime/bufferpool/buffer-pool.h"
#include "runtime/bufferpool/system-allocator.h"
-#include "runtime/test-env.h"
-#include "service/fe-support.h"
#include "testutil/cpu-util.h"
#include "testutil/gtest-util.h"
#include "util/cpu-info.h"
@@ -41,8 +39,6 @@ using BufferHandle = BufferPool::BufferHandle;
class BufferAllocatorTest : public ::testing::Test {
public:
virtual void SetUp() {
- test_env_.reset(new TestEnv);
- ASSERT_OK(test_env_->Init());
dummy_pool_ = obj_pool_.Add(new BufferPool(1, 0, 0));
dummy_reservation_.InitRootTracker(nullptr, 0);
ASSERT_OK(dummy_pool_->RegisterClient("", nullptr, &dummy_reservation_, nullptr, 0,
@@ -63,8 +59,6 @@ class BufferAllocatorTest : public ::testing::Test {
/// The minimum buffer size used in most tests.
const static int64_t TEST_BUFFER_LEN = 1024;
- boost::scoped_ptr<TestEnv> test_env_;
-
ObjectPool obj_pool_;
/// Need a dummy pool and client to pass around. We bypass the reservation mechanisms
@@ -206,7 +200,6 @@ TEST_F(SystemAllocatorTest, LargeAllocFailure) {
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
impala::InitCommonRuntime(argc, argv, true, impala::TestInfo::BE_TEST);
- impala::InitFeSupport();
int result = 0;
for (bool mmap : {false, true}) {
for (bool madvise : {false, true}) {
http://git-wip-us.apache.org/repos/asf/impala/blob/b3d38b5c/be/src/runtime/bufferpool/free-list-test.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/bufferpool/free-list-test.cc b/be/src/runtime/bufferpool/free-list-test.cc
index d3c4c9a..7cb80b4 100644
--- a/be/src/runtime/bufferpool/free-list-test.cc
+++ b/be/src/runtime/bufferpool/free-list-test.cc
@@ -21,8 +21,6 @@
#include "common/object-pool.h"
#include "runtime/bufferpool/free-list.h"
#include "runtime/bufferpool/system-allocator.h"
-#include "runtime/test-env.h"
-#include "service/fe-support.h"
#include "testutil/gtest-util.h"
#include "testutil/rand-util.h"
@@ -33,8 +31,6 @@ namespace impala {
class FreeListTest : public ::testing::Test {
protected:
virtual void SetUp() override {
- test_env_.reset(new TestEnv);
- ASSERT_OK(test_env_->Init());
allocator_ = obj_pool_.Add(new SystemAllocator(MIN_BUFFER_LEN));
RandTestUtil::SeedRng("FREE_LIST_TEST_SEED", &rng_);
}
@@ -75,8 +71,6 @@ class FreeListTest : public ::testing::Test {
const static int MIN_BUFFER_LEN = 1024;
- boost::scoped_ptr<TestEnv> test_env_;
-
/// Per-test random number generator. Seeded before every test.
std::mt19937 rng_;
@@ -163,9 +157,4 @@ TEST_F(FreeListTest, ReturnOrder) {
}
}
-int main(int argc, char** argv) {
- ::testing::InitGoogleTest(&argc, argv);
- impala::InitCommonRuntime(argc, argv, true, impala::TestInfo::BE_TEST);
- impala::InitFeSupport();
- return RUN_ALL_TESTS();
-}
+IMPALA_TEST_MAIN();
http://git-wip-us.apache.org/repos/asf/impala/blob/b3d38b5c/be/src/runtime/bufferpool/suballocator-test.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/bufferpool/suballocator-test.cc b/be/src/runtime/bufferpool/suballocator-test.cc
index 470b065..6cd53fb 100644
--- a/be/src/runtime/bufferpool/suballocator-test.cc
+++ b/be/src/runtime/bufferpool/suballocator-test.cc
@@ -27,8 +27,6 @@
#include "common/object-pool.h"
#include "runtime/bufferpool/reservation-tracker.h"
#include "runtime/bufferpool/suballocator.h"
-#include "runtime/test-env.h"
-#include "service/fe-support.h"
#include "testutil/death-test-util.h"
#include "testutil/gtest-util.h"
#include "testutil/rand-util.h"
@@ -46,8 +44,6 @@ namespace impala {
class SuballocatorTest : public ::testing::Test {
public:
virtual void SetUp() override {
- test_env_.reset(new TestEnv);
- ASSERT_OK(test_env_->Init());
RandTestUtil::SeedRng("SUBALLOCATOR_TEST_SEED", &rng_);
profile_ = RuntimeProfile::Create(&obj_pool_, "test profile");
}
@@ -115,8 +111,6 @@ class SuballocatorTest : public ::testing::Test {
/// Clients for the buffer pool. Deregistered and freed after every test.
vector<unique_ptr<BufferPool::ClientHandle>> clients_;
- boost::scoped_ptr<TestEnv> test_env_;
-
/// Global profile - recreated for every test.
RuntimeProfile* profile_;
@@ -368,9 +362,4 @@ void SuballocatorTest::AssertMemoryValid(
}
}
-int main(int argc, char** argv) {
- ::testing::InitGoogleTest(&argc, argv);
- impala::InitCommonRuntime(argc, argv, true, impala::TestInfo::BE_TEST);
- impala::InitFeSupport();
- return RUN_ALL_TESTS();
-}
+IMPALA_TEST_MAIN();
http://git-wip-us.apache.org/repos/asf/impala/blob/b3d38b5c/be/src/runtime/exec-env.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/exec-env.cc b/be/src/runtime/exec-env.cc
index 94ca834..6d9a857 100644
--- a/be/src/runtime/exec-env.cc
+++ b/be/src/runtime/exec-env.cc
@@ -271,6 +271,16 @@ Status ExecEnv::Init() {
"bytes value or percentage: $0", FLAGS_mem_limit));
}
+#if !defined(ADDRESS_SANITIZER) && !defined(THREAD_SANITIZER)
+ // Aggressive decommit is required so that unused pages in the TCMalloc page heap are
+ // not backed by physical pages and do not contribute towards memory consumption.
+ // Enable it in TCMalloc before InitBufferPool().
+ if (!MallocExtension::instance()->SetNumericProperty(
+ "tcmalloc.aggressive_memory_decommit", 1)) {
+ return Status("Failed to enable TCMalloc aggressive decommit.");
+ }
+#endif
+
if (!BitUtil::IsPowerOf2(FLAGS_min_buffer_size)) {
return Status(Substitute(
"--min_buffer_size must be a power-of-two: $0", FLAGS_min_buffer_size));
@@ -310,11 +320,6 @@ Status ExecEnv::Init() {
FLAGS_datastream_service_num_svc_threads : CpuInfo::num_cores();
RETURN_IF_ERROR(rpc_mgr_->RegisterService(num_svc_threads,
FLAGS_datastream_service_queue_depth, move(data_svc)));
- // Bump thread cache to 1GB to reduce contention for TCMalloc central
- // list's spinlock.
- if (FLAGS_tcmalloc_max_total_thread_cache_bytes == 0) {
- FLAGS_tcmalloc_max_total_thread_cache_bytes = 1 << 30;
- }
}
mem_tracker_.reset(
@@ -427,12 +432,6 @@ Status ExecEnv::StartKrpcService() {
void ExecEnv::InitBufferPool(int64_t min_buffer_size, int64_t capacity,
int64_t clean_pages_limit) {
-#if !defined(ADDRESS_SANITIZER) && !defined(THREAD_SANITIZER)
- // Aggressive decommit is required so that unused pages in the TCMalloc page heap are
- // not backed by physical pages and do not contribute towards memory consumption.
- MallocExtension::instance()->SetNumericProperty(
- "tcmalloc.aggressive_memory_decommit", 1);
-#endif
buffer_pool_.reset(new BufferPool(min_buffer_size, capacity, clean_pages_limit));
buffer_reservation_.reset(new ReservationTracker());
buffer_reservation_->InitRootTracker(nullptr, capacity);
http://git-wip-us.apache.org/repos/asf/impala/blob/b3d38b5c/bin/impala-config.sh
----------------------------------------------------------------------
diff --git a/bin/impala-config.sh b/bin/impala-config.sh
index f5b67a9..058dc01 100755
--- a/bin/impala-config.sh
+++ b/bin/impala-config.sh
@@ -72,7 +72,7 @@ fi
# moving to a different build of the toolchain, e.g. when a version is bumped or a
# compile option is changed. The build id can be found in the output of the toolchain
# build jobs, it is constructed from the build number and toolchain git hash prefix.
-export IMPALA_TOOLCHAIN_BUILD_ID=4-7847dc86c4
+export IMPALA_TOOLCHAIN_BUILD_ID=482-c2361403fc
# Versions of toolchain dependencies.
# -----------------------------------
export IMPALA_AVRO_VERSION=1.7.4-p4
@@ -99,7 +99,7 @@ export IMPALA_GFLAGS_VERSION=2.2.0-p1
unset IMPALA_GFLAGS_URL
export IMPALA_GLOG_VERSION=0.3.4-p2
unset IMPALA_GLOG_URL
-export IMPALA_GPERFTOOLS_VERSION=2.6.3
+export IMPALA_GPERFTOOLS_VERSION=2.5
unset IMPALA_GPERFTOOLS_URL
export IMPALA_GTEST_VERSION=1.6.0
unset IMPALA_GTEST_URL