You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2017/03/27 23:13:50 UTC
[1/5] incubator-impala git commit: IMPALA-4859: Push down IS NULL /
IS NOT NULL to Kudu
Repository: incubator-impala
Updated Branches:
refs/heads/master 878fcf5a7 -> 874d20d0f
IMPALA-4859: Push down IS NULL / IS NOT NULL to Kudu
This detects IS NULL / IS NOT NULL and creates a Kudu
predicate to push this to Kudu.
For testing, there are planner tests to verify that the
predicate is pushed to Kudu. There are also end-to-end
tests for correctness.
Change-Id: I9c96fec8d41f77222879c0ffdd6940b168e47e65
Reviewed-on: http://gerrit.cloudera.org:8080/5958
Reviewed-by: Marcel Kornacker <ma...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/077c07ee
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/077c07ee
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/077c07ee
Branch: refs/heads/master
Commit: 077c07eec77c05f97d5e45704454b788ed1b45e0
Parents: 878fcf5
Author: Joe McDonnell <jo...@cloudera.com>
Authored: Wed Feb 8 10:48:55 2017 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Sat Mar 25 04:51:36 2017 +0000
----------------------------------------------------------------------
.../org/apache/impala/planner/KuduScanNode.java | 33 +++++++++++++++++++-
.../queries/PlannerTest/kudu-selectivity.test | 14 +++++++++
.../queries/PlannerTest/kudu.test | 6 ++--
.../queries/QueryTest/kudu-scan-node.test | 30 ++++++++++++++++++
4 files changed, 80 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/077c07ee/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java b/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
index 7cd6e7c..64e5fde 100644
--- a/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
@@ -27,6 +27,7 @@ import org.apache.impala.analysis.BinaryPredicate;
import org.apache.impala.analysis.BoolLiteral;
import org.apache.impala.analysis.Expr;
import org.apache.impala.analysis.InPredicate;
+import org.apache.impala.analysis.IsNullPredicate;
import org.apache.impala.analysis.LiteralExpr;
import org.apache.impala.analysis.NullLiteral;
import org.apache.impala.analysis.NumericLiteral;
@@ -322,7 +323,8 @@ public class KuduScanNode extends ScanNode {
while (it.hasNext()) {
Expr predicate = it.next();
if (tryConvertBinaryKuduPredicate(analyzer, rpcTable, predicate) ||
- tryConvertInListKuduPredicate(analyzer, rpcTable, predicate)) {
+ tryConvertInListKuduPredicate(analyzer, rpcTable, predicate) ||
+ tryConvertIsNullKuduPredicate(analyzer, rpcTable, predicate)) {
it.remove();
}
}
@@ -436,6 +438,35 @@ public class KuduScanNode extends ScanNode {
}
/**
+ * If IS NULL/IS NOT NULL 'expr' can be converted to a KuduPredicate,
+ * returns true and updates kuduPredicates_ and kuduConjuncts_.
+ */
+ private boolean tryConvertIsNullKuduPredicate(Analyzer analyzer,
+ org.apache.kudu.client.KuduTable table, Expr expr) {
+ if (!(expr instanceof IsNullPredicate)) return false;
+ IsNullPredicate predicate = (IsNullPredicate) expr;
+
+ // Do not convert if expression is more than a SlotRef
+ // This is true even for casts, as certain casts can take a non-NULL
+ // value and produce a NULL. For example, CAST('test' as tinyint)
+ // is NULL.
+ if (!(predicate.getChild(0) instanceof SlotRef)) return false;
+ SlotRef ref = (SlotRef) predicate.getChild(0);
+
+ String colName = ref.getDesc().getColumn().getName();
+ ColumnSchema column = table.getSchema().getColumn(colName);
+ KuduPredicate kuduPredicate = null;
+ if (predicate.isNotNull()) {
+ kuduPredicate = KuduPredicate.newIsNotNullPredicate(column);
+ } else {
+ kuduPredicate = KuduPredicate.newIsNullPredicate(column);
+ }
+ kuduConjuncts_.add(predicate);
+ kuduPredicates_.add(kuduPredicate);
+ return true;
+ }
+
+ /**
* Return the value of the InList child expression 'e' as an Object that can be
* added to a KuduPredicate. If the Expr is not supported by Kudu or the type doesn't
* match the expected PrimitiveType 'type', null is returned.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/077c07ee/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test b/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
index eba7741..b064d2b 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
@@ -130,3 +130,17 @@ F00:PLAN FRAGMENT [UNPARTITIONED]
hosts=3 per-host-mem=unavailable
tuple-ids=0 row-size=126B cardinality=4
====
+select * from functional_kudu.alltypes where
+tinyint_col is not null and
+smallint_col is null and
+cast(date_string_col as tinyint) is null
+---- PLAN
+F00:PLAN FRAGMENT [UNPARTITIONED]
+ PLAN-ROOT SINK
+ |
+ 00:SCAN KUDU [functional_kudu.alltypes]
+ predicates: CAST(date_string_col AS TINYINT) IS NULL
+ kudu predicates: smallint_col IS NULL, tinyint_col IS NOT NULL
+ hosts=3 per-host-mem=unavailable
+ tuple-ids=0 row-size=126B cardinality=730
+====
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/077c07ee/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test b/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
index 63db8ba..c99de37 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
@@ -166,7 +166,8 @@ where cast(sin(id) as boolean) = true and name is null
PLAN-ROOT SINK
|
00:SCAN KUDU [functional_kudu.testtbl]
- predicates: name IS NULL, CAST(sin(id) AS BOOLEAN) = TRUE
+ predicates: CAST(sin(id) AS BOOLEAN) = TRUE
+ kudu predicates: name IS NULL
---- SCANRANGELOCATIONS
NODE 0:
ScanToken{table=impala::functional_kudu.testtbl, range-partition: [(int64 id=1004), (int64 id=1008))}
@@ -178,7 +179,8 @@ PLAN-ROOT SINK
01:EXCHANGE [UNPARTITIONED]
|
00:SCAN KUDU [functional_kudu.testtbl]
- predicates: name IS NULL, CAST(sin(id) AS BOOLEAN) = TRUE
+ predicates: CAST(sin(id) AS BOOLEAN) = TRUE
+ kudu predicates: name IS NULL
====
# IMPALA-3856: KuduScanNode crash when pushing predicates including a cast
select o_orderkey from tpch_kudu.orders where o_orderkey < 10.0 order by 1
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/077c07ee/testdata/workloads/functional-query/queries/QueryTest/kudu-scan-node.test
----------------------------------------------------------------------
diff --git a/testdata/workloads/functional-query/queries/QueryTest/kudu-scan-node.test b/testdata/workloads/functional-query/queries/QueryTest/kudu-scan-node.test
index 0e7bf03..243b7e5 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/kudu-scan-node.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/kudu-scan-node.test
@@ -55,3 +55,33 @@ select count(int_col) from functional_kudu.tinyinttable
---- TYPES
BIGINT
====
+---- QUERY
+# IMPALA-4859: Test Kudu IS NULL/IS NOT NULL pushdown
+select count(*) from functional_kudu.alltypesagg where id < 10 and float_col is null;
+---- RESULTS
+2
+---- TYPES
+BIGINT
+====
+---- QUERY
+select count(*) from functional_kudu.alltypesagg where id < 10 and float_col is not null;
+---- RESULTS
+9
+---- TYPES
+BIGINT
+====
+---- QUERY
+# alltypes.id is primary key/not nullable, verify IS NOT NULL/IS NULL pushdown works
+select count(*) from functional_kudu.alltypes where id is not null;
+---- RESULTS
+7300
+---- TYPES
+BIGINT
+====
+---- QUERY
+select count(*) from functional_kudu.alltypes where id is null;
+---- RESULTS
+0
+---- TYPES
+BIGINT
+====
\ No newline at end of file
[2/5] incubator-impala git commit: IMPALA-5115: Handle status from
HdfsTableSink::WriteClusteredRowBatch
Posted by ta...@apache.org.
IMPALA-5115: Handle status from HdfsTableSink::WriteClusteredRowBatch
Change-Id: Id6f5d87ae5e46d6019d9988011d3f2657e81ae15
Reviewed-on: http://gerrit.cloudera.org:8080/6486
Reviewed-by: Marcel Kornacker <ma...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/8e45c4ae
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/8e45c4ae
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/8e45c4ae
Branch: refs/heads/master
Commit: 8e45c4ae9c65f316072e395602648fe36081377a
Parents: 077c07e
Author: Lars Volker <lv...@cloudera.com>
Authored: Sat Mar 25 17:50:09 2017 +0100
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Sun Mar 26 15:36:44 2017 +0000
----------------------------------------------------------------------
be/src/exec/hdfs-table-sink.cc | 2 +-
be/src/exec/hdfs-table-sink.h | 21 +++++++++++++--------
2 files changed, 14 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8e45c4ae/be/src/exec/hdfs-table-sink.cc
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-table-sink.cc b/be/src/exec/hdfs-table-sink.cc
index 4aa09cd..ef983e2 100644
--- a/be/src/exec/hdfs-table-sink.cc
+++ b/be/src/exec/hdfs-table-sink.cc
@@ -627,7 +627,7 @@ Status HdfsTableSink::Send(RuntimeState* state, RowBatch* batch) {
GetOutputPartition(state, nullptr, ROOT_PARTITION_KEY, &partition_pair, false));
RETURN_IF_ERROR(WriteRowsToPartition(state, batch, partition_pair));
} else if (input_is_clustered_) {
- WriteClusteredRowBatch(state, batch);
+ RETURN_IF_ERROR(WriteClusteredRowBatch(state, batch));
} else {
for (int i = 0; i < batch->num_rows(); ++i) {
const TupleRow* current_row = batch->GetRow(i);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/8e45c4ae/be/src/exec/hdfs-table-sink.h
----------------------------------------------------------------------
diff --git a/be/src/exec/hdfs-table-sink.h b/be/src/exec/hdfs-table-sink.h
index 45d64c5..d452f3d 100644
--- a/be/src/exec/hdfs-table-sink.h
+++ b/be/src/exec/hdfs-table-sink.h
@@ -169,14 +169,15 @@ class HdfsTableSink : public DataSink {
/// added to it, empty_partition must be true.
Status InitOutputPartition(RuntimeState* state,
const HdfsPartitionDescriptor& partition_descriptor, const TupleRow* row,
- OutputPartition* output_partition, bool empty_partition);
+ OutputPartition* output_partition, bool empty_partition) WARN_UNUSED_RESULT;
/// Add a temporary file to an output partition. Files are created in a
/// temporary directory and then moved to the real partition directory by the
/// coordinator in a finalization step. The temporary file's current location
/// and final destination are recorded in the state parameter.
/// If this function fails, the tmp file is cleaned up.
- Status CreateNewTmpFile(RuntimeState* state, OutputPartition* output_partition);
+ Status CreateNewTmpFile(RuntimeState* state, OutputPartition* output_partition)
+ WARN_UNUSED_RESULT;
/// Key is the concatenation of the evaluated dynamic_partition_key_exprs_ generated by
/// GetHashTblKey(). Maps to an OutputPartition and a vector of indices of the rows in
@@ -199,10 +200,11 @@ class HdfsTableSink : public DataSink {
/// the 'partition_keys_to_output_partitions_'. 'no_more_rows' indicates that no more
/// rows will be added to the partition.
Status GetOutputPartition(RuntimeState* state, const TupleRow* row,
- const std::string& key, PartitionPair** partition_pair, bool no_more_rows);
+ const std::string& key, PartitionPair** partition_pair, bool no_more_rows)
+ WARN_UNUSED_RESULT;
/// Initialise and prepare select and partition key expressions
- Status PrepareExprs(RuntimeState* state);
+ Status PrepareExprs(RuntimeState* state) WARN_UNUSED_RESULT;
/// Sets hdfs_file_name and tmp_hdfs_file_name of given output partition.
/// The Hdfs directory is created from the target table's base Hdfs dir,
@@ -214,18 +216,21 @@ class HdfsTableSink : public DataSink {
/// Writes all rows referenced by the row index vector in 'partition_pair' to the
/// partition's writer and clears the row index vector afterwards.
Status WriteRowsToPartition(
- RuntimeState* state, RowBatch* batch, PartitionPair* partition_pair);
+ RuntimeState* state, RowBatch* batch, PartitionPair* partition_pair)
+ WARN_UNUSED_RESULT;
/// Maps all rows in 'batch' to partitions and appends them to their temporary Hdfs
/// files. The input must be ordered by the partition key expressions.
- Status WriteClusteredRowBatch(RuntimeState* state, RowBatch* batch);
+ Status WriteClusteredRowBatch(RuntimeState* state, RowBatch* batch) WARN_UNUSED_RESULT;
/// Updates runtime stats of HDFS with rows written, then closes the file associated
/// with the partition by calling ClosePartitionFile()
- Status FinalizePartitionFile(RuntimeState* state, OutputPartition* partition);
+ Status FinalizePartitionFile(RuntimeState* state, OutputPartition* partition)
+ WARN_UNUSED_RESULT;
/// Closes the hdfs file for this partition as well as the writer.
- Status ClosePartitionFile(RuntimeState* state, OutputPartition* partition);
+ Status ClosePartitionFile(RuntimeState* state, OutputPartition* partition)
+ WARN_UNUSED_RESULT;
// Returns TRUE if the staging step should be skipped for this partition. This allows
// for faster INSERT query completion time for the S3A filesystem as the coordinator
[4/5] incubator-impala git commit: IMPALA-5110: Add deb support to
dump_breakpad_symbols.py
Posted by ta...@apache.org.
IMPALA-5110: Add deb support to dump_breakpad_symbols.py
Change-Id: I524d2fe4660551c2fe4ff190b7e5bbb33d986b10
Reviewed-on: http://gerrit.cloudera.org:8080/6462
Reviewed-by: Tim Armstrong <ta...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/e98c88f2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/e98c88f2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/e98c88f2
Branch: refs/heads/master
Commit: e98c88f233f17335b551ad53e4541491e07c5913
Parents: 77dbdc3
Author: Lars Volker <lv...@cloudera.com>
Authored: Thu Mar 23 14:26:46 2017 +0100
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Mon Mar 27 22:24:21 2017 +0000
----------------------------------------------------------------------
bin/dump_breakpad_symbols.py | 93 +++++++++++++++++++++++----------------
1 file changed, 56 insertions(+), 37 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/e98c88f2/bin/dump_breakpad_symbols.py
----------------------------------------------------------------------
diff --git a/bin/dump_breakpad_symbols.py b/bin/dump_breakpad_symbols.py
index a0d35e1..aee0418 100755
--- a/bin/dump_breakpad_symbols.py
+++ b/bin/dump_breakpad_symbols.py
@@ -23,10 +23,11 @@
# - Scan an Impala build dir for ELF files
# - Read files from stdin
# - Process a list of one or multiple explicitly specified files
-# - Extract an Impala rpm and corresponding debuginfo rpm file, scan for ELF files, and
-# process them together with their respective .debug file.
+# - Extract an Impala rpm/deb and corresponding debuginfo rpm/deb file, scan for ELF
+# files, and process them together with their respective .debug file.
#
# Dependencies:
+# - dpkg (sudo apt-get -y install dpkg)
# - rpm2cpio (sudo apt-get -y install rpm2cpio)
# - cpio (sudo apt-get -y install cpio)
# - Google Breakpad, either installed via the Impala toolchain or separately
@@ -67,7 +68,7 @@ from collections import namedtuple
logging.basicConfig(level=logging.INFO)
-BinaryDebugInfo = namedtuple('BinaryDebugInfo', 'path, debug_path')
+BinarySymbolInfo = namedtuple('BinarySymbolInfo', 'path, debug_path')
def die(msg=''):
@@ -111,18 +112,18 @@ def parse_args():
help='List of binary files to process')
parser.add_argument('-i', '--stdin_files', action='store_true', help="""Read the list
of files to process from stdin""")
- parser.add_argument('-r', '--rpm', help="""RPM file containing the binaries to process,
- use with --debuginfo_rpm""")
- parser.add_argument('-s', '--debuginfo_rpm', help="""RPM file containing the debug
- symbols matching the binaries in --rpm""")
+ parser.add_argument('-r', '--pkg', '--rpm', help="""RPM/DEB file containing the binaries
+ to process, use with -s""")
+ parser.add_argument('-s', '--symbol_pkg', '--debuginfo_rpm', help="""RPM/DEB file
+ containing the debug symbols matching the binaries in -r""")
args = parser.parse_args()
# Post processing checks
- # Check that either both rpm and debuginfo_rpm are specified, or none.
- if bool(args.rpm) != bool(args.debuginfo_rpm):
+ # Check that either both pkg and debuginfo_rpm/deb are specified, or none.
+ if bool(args.pkg) != bool(args.symbol_pkg):
parser.print_usage()
- die('Either both --rpm and --debuginfo_rpm have to be specified, or none')
- input_flags = [args.build_dir, args.binary_files, args.stdin_files, args.rpm]
+ die('Either both -r and -s have to be specified, or none')
+ input_flags = [args.build_dir, args.binary_files, args.stdin_files, args.pkg]
if sum(1 for flag in input_flags if flag) != 1:
die('You need to specify exactly one way to locate input files (-b/-f/-i/-r,-s)')
@@ -166,57 +167,75 @@ def extract_rpm(rpm, out_dir):
subprocess.check_call(cmd, shell=True, cwd=out_dir)
+def extract_deb(deb, out_dir):
+ """Extract 'deb' into 'out_dir'."""
+ assert os.path.isdir(out_dir)
+ cmd = 'dpkg -x %s %s' % (deb, out_dir)
+ subprocess.check_call(cmd, shell=True)
+
+
+def extract_pkg(pkg, out_dir):
+ """Autodetect type of 'pkg' and extract it to 'out_dir'."""
+ pkg_magic = magic.from_file(pkg)
+ if 'RPM' in pkg_magic:
+ return extract_rpm(pkg, out_dir)
+ elif 'Debian' in pkg_magic:
+ return extract_deb(pkg, out_dir)
+ else:
+ die('Unsupported package type: %s' % pkg_magic)
+
+
def assert_file_exists(path):
if not os.path.isfile(path):
die('File does not exists: %s' % path)
-def enumerate_rpm_files(rpm, debuginfo_rpm):
- """Return a generator over BinaryDebugInfo tuples for all ELF files in 'rpm'.
+def enumerate_pkg_files(pkg, symbol_pkg):
+ """Return a generator over BinarySymbolInfo tuples for all ELF files in 'pkg'.
- This function extracts both RPM files, then walks the binary rpm directory to enumerate
- all ELF files, matches them to the location of their respective .debug file and yields
- all tuples thereof. We use a generator here to keep the temporary directory and its
- contents around until the consumer of the generator has finished its processing.
+ This function extracts both RPM/DEB files, then walks the binary pkg directory to
+ enumerate all ELF files, matches them to the location of their respective .debug files
+ and yields all tuples thereof. We use a generator here to keep the temporary directory
+ and its contents around until the consumer of the generator has finished its processing.
"""
IMPALA_BINARY_BASE = os.path.join('usr', 'lib', 'impala')
- IMPALA_DEBUGINFO_BASE = os.path.join('usr', 'lib', 'debug', IMPALA_BINARY_BASE)
- assert_file_exists(rpm)
- assert_file_exists(debuginfo_rpm)
+ IMPALA_SYMBOL_BASE = os.path.join('usr', 'lib', 'debug', IMPALA_BINARY_BASE)
+ assert_file_exists(pkg)
+ assert_file_exists(symbol_pkg)
tmp_dir = tempfile.mkdtemp()
try:
- # Extract rpm
- logging.info('Extracting: %s' % rpm)
- extract_rpm(os.path.abspath(rpm), tmp_dir)
- # Extract debuginfo_rpm
- logging.info('Extracting: %s' % debuginfo_rpm)
- extract_rpm(os.path.abspath(debuginfo_rpm), tmp_dir)
- # Walk rpm path and find elf files
+ # Extract pkg
+ logging.info('Extracting to %s: %s' % (tmp_dir, pkg))
+ extract_pkg(os.path.abspath(pkg), tmp_dir)
+ # Extract symbol_pkg
+ logging.info('Extracting to %s: %s' % (tmp_dir, symbol_pkg))
+ extract_pkg(os.path.abspath(symbol_pkg), tmp_dir)
+ # Walk pkg path and find elf files
binary_base = os.path.join(tmp_dir, IMPALA_BINARY_BASE)
- debuginfo_base = os.path.join(tmp_dir, IMPALA_DEBUGINFO_BASE)
- # Find folder with .debug file in debuginfo_rpm path
+ symbol_base = os.path.join(tmp_dir, IMPALA_SYMBOL_BASE)
+ # Find folder with .debug file in symbol_pkg path
for binary_path in find_elf_files(binary_base):
# Add tuple to output
rel_dir = os.path.relpath(os.path.dirname(binary_path), binary_base)
- debug_dir = os.path.join(debuginfo_base, rel_dir)
- yield BinaryDebugInfo(binary_path, debug_dir)
+ debug_dir = os.path.join(symbol_base, rel_dir)
+ yield BinarySymbolInfo(binary_path, debug_dir)
finally:
shutil.rmtree(tmp_dir)
def enumerate_binaries(args):
- """Enumerate all BinaryDebugInfo tuples, from which symbols should be extracted.
+ """Enumerate all BinarySymbolInfo tuples, from which symbols should be extracted.
This function returns iterables, either lists or generators.
"""
if args.binary_files:
- return (BinaryDebugInfo(f, None) for f in args.binary_files)
+ return (BinarySymbolInfo(f, None) for f in args.binary_files)
elif args.stdin_files:
- return (BinaryDebugInfo(f, None) for f in sys.stdin.read().splitlines())
- elif args.rpm:
- return enumerate_rpm_files(args.rpm, args.debuginfo_rpm)
+ return (BinarySymbolInfo(f, None) for f in sys.stdin.read().splitlines())
+ elif args.pkg:
+ return enumerate_pkg_files(args.pkg, args.symbol_pkg)
elif args.build_dir:
- return (BinaryDebugInfo(f, None) for f in find_elf_files(args.build_dir))
+ return (BinarySymbolInfo(f, None) for f in find_elf_files(args.build_dir))
die('No input method provided')
[5/5] incubator-impala git commit: IMPALA-5113: fix dirty unpinned
invariant
Posted by ta...@apache.org.
IMPALA-5113: fix dirty unpinned invariant
There were two bugs:
* The invariant was too strict and didn't take into account multiple
pins of pages (which don't use buffers and therefore shouldn't count).
* The invariant wasn't enforced when reclaiming a clean page.
Change the logic so that it's implemented in terms of pages/buffers in
various states (this avoids the reservation double-counting and more
directly expresses the intent). To aid in this, refactor the page lists
to use a wrapper that tracks the # of bytes of pages in each list.
Testing:
Added a unit test that reproduces the issue and added stricter DCHECKs
to detect the issue in future.
Change-Id: I07e08acb6cf6839bfccbd09258c093b1c8252b25
Reviewed-on: http://gerrit.cloudera.org:8080/6469
Reviewed-by: Tim Armstrong <ta...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/874d20d0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/874d20d0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/874d20d0
Branch: refs/heads/master
Commit: 874d20d0fabaf0572202a6e65dc55d334229ca6b
Parents: e98c88f
Author: Tim Armstrong <ta...@cloudera.com>
Authored: Fri Mar 24 08:52:56 2017 -0700
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Mon Mar 27 22:52:33 2017 +0000
----------------------------------------------------------------------
.../runtime/bufferpool/buffer-pool-internal.h | 242 ++++++++++++-------
be/src/runtime/bufferpool/buffer-pool-test.cc | 32 +++
be/src/runtime/bufferpool/buffer-pool.cc | 130 +++++-----
be/src/runtime/bufferpool/buffer-pool.h | 1 +
be/src/util/internal-queue.h | 10 +-
5 files changed, 263 insertions(+), 152 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/874d20d0/be/src/runtime/bufferpool/buffer-pool-internal.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/bufferpool/buffer-pool-internal.h b/be/src/runtime/bufferpool/buffer-pool-internal.h
index 04d76bb..f0dfa09 100644
--- a/be/src/runtime/bufferpool/buffer-pool-internal.h
+++ b/be/src/runtime/bufferpool/buffer-pool-internal.h
@@ -56,15 +56,14 @@
/// reservations are not overcommitted (they shouldn't be), this global invariant can be
/// maintained by enforcing a local invariant for every client:
///
-/// unused reservation >= dirty unpinned pages
+/// reservation >= BufferHandles returned to client
+// + pinned pages + dirty pages (dirty unpinned or write in flight)
///
/// The local invariant is maintained by writing pages to disk as the first step of any
-/// operation that uses reservation. I.e. the R.H.S. of the invariant must be decreased
-/// before the L.H.S. can be decreased. These operations block waiting for enough writes
-/// to complete to satisfy the invariant.
-/// TODO: this invariant can be broken if a client calls DecreaseReservation() on the
-/// ReservationTracker. We should refactor so that DecreaseReservation() goes through
-/// the client before closing IMPALA-3202.
+/// operation that allocates a new buffer or reclaims buffers from clean pages. I.e.
+/// "dirty pages" must be decreased before one of the other values on the R.H.S. of the
+/// invariant can be increased. Operations block waiting for enough writes to complete
+/// to satisfy the invariant.
#ifndef IMPALA_RUNTIME_BUFFER_POOL_INTERNAL_H
#define IMPALA_RUNTIME_BUFFER_POOL_INTERNAL_H
@@ -79,8 +78,106 @@
#include "runtime/bufferpool/reservation-tracker.h"
#include "util/condition-variable.h"
+// Ensure that DCheckConsistency() function calls get removed in release builds.
+#ifndef NDEBUG
+#define DCHECK_CONSISTENCY() DCheckConsistency()
+#else
+#define DCHECK_CONSISTENCY()
+#endif
+
namespace impala {
+/// The internal representation of a page, which can be pinned or unpinned. See the
+/// class comment for explanation of the different page states.
+struct BufferPool::Page : public InternalList<Page>::Node {
+ Page(Client* client, int64_t len) : client(client), len(len), pin_count(0) {}
+
+ std::string DebugString();
+
+ // Helper for BufferPool::DebugString().
+ static bool DebugStringCallback(std::stringstream* ss, BufferPool::Page* page);
+
+ /// The client that the page belongs to.
+ Client* const client;
+
+ /// The length of the page in bytes.
+ const int64_t len;
+
+ /// The pin count of the page. Only accessed in contexts that are passed the associated
+ /// PageHandle, so it cannot be accessed by multiple threads concurrently.
+ int pin_count;
+
+ /// Non-null if there is a write in flight, the page is clean, or the page is evicted.
+ std::unique_ptr<TmpFileMgr::WriteHandle> write_handle;
+
+ /// Condition variable signalled when a write for this page completes. Protected by
+ /// client->lock_.
+ ConditionVariable write_complete_cv_;
+
+ /// This lock must be held when accessing 'buffer' if the page is unpinned and not
+ /// evicted (i.e. it is safe to access 'buffer' if the page is pinned or evicted).
+ SpinLock buffer_lock;
+
+ /// Buffer with the page's contents. Closed only iff page is evicted. Open otherwise.
+ BufferHandle buffer;
+};
+
+/// Wrapper around InternalList<Page> that tracks the # of bytes in the list.
+class BufferPool::PageList {
+ public:
+ PageList() : bytes_(0) {}
+ ~PageList() {
+ // Clients always empty out their list before destruction.
+ DCHECK(list_.empty());
+ DCHECK_EQ(0, bytes_);
+ }
+
+ void Enqueue(Page* page) {
+ list_.Enqueue(page);
+ bytes_ += page->len;
+ }
+
+ bool Remove(Page* page) {
+ if (list_.Remove(page)) {
+ bytes_ -= page->len;
+ return true;
+ }
+ return false;
+ }
+
+ Page* Dequeue() {
+ Page* page = list_.Dequeue();
+ if (page != nullptr) {
+ bytes_ -= page->len;
+ }
+ return page;
+ }
+
+ Page* PopBack() {
+ Page* page = list_.PopBack();
+ if (page != nullptr) {
+ bytes_ -= page->len;
+ }
+ return page;
+ }
+
+ void Iterate(boost::function<bool(Page*)> fn) { list_.Iterate(fn); }
+ bool Contains(Page* page) { return list_.Contains(page); }
+ Page* tail() { return list_.tail(); }
+ bool empty() const { return list_.empty(); }
+ int size() const { return list_.size(); }
+ int64_t bytes() const { return bytes_; }
+
+ void DCheckConsistency() {
+ DCHECK_GE(bytes_, 0);
+ DCHECK_EQ(list_.empty(), bytes_ == 0);
+ }
+
+ private:
+ InternalList<Page> list_;
+ int64_t bytes_;
+};
+
/// The internal state for the client.
class BufferPool::Client {
public:
@@ -90,18 +187,15 @@ class BufferPool::Client {
~Client() {
DCHECK_EQ(0, num_pages_);
- DCHECK_EQ(0, pinned_pages_.size());
- DCHECK_EQ(0, dirty_unpinned_pages_.size());
- DCHECK_EQ(0, in_flight_write_pages_.size());
+ DCHECK_EQ(0, buffers_allocated_bytes_);
}
/// Release reservation for this client.
void Close() { reservation_.Close(); }
- /// Add a new pinned page 'page' to the pinned pages list. 'page' must not be in any
- /// other lists. Neither the client's lock nor page->buffer_lock should be held by the
- /// caller.
- void AddNewPinnedPage(Page* page);
+ /// Create a pinned page using 'buffer', which was allocated using AllocateBuffer().
+ /// No client or page locks should be held by the caller.
+ Page* CreatePinnedPage(BufferHandle&& buffer);
/// Reset 'handle', clean up references to handle->page and release any resources
/// associated with handle->page. If the page is pinned, 'out_buffer' can be passed in
@@ -113,7 +207,7 @@ class BufferPool::Client {
/// Updates client state to reflect that 'page' is now a dirty unpinned page. May
/// initiate writes for this or other dirty unpinned pages.
/// Neither the client's lock nor page->buffer_lock should be held by the caller.
- void MoveToDirtyUnpinned(int64_t unused_reservation, Page* page);
+ void MoveToDirtyUnpinned(Page* page);
/// Move an unpinned page to the pinned state, moving between data structures and
/// reading from disk if necessary. Returns once the page's buffer is allocated
@@ -121,26 +215,24 @@ class BufferPool::Client {
/// handle->page_->buffer_lock should be held by the caller.
Status MoveToPinned(ClientHandle* client, PageHandle* handle);
- /// Must be called before allocating a buffer to ensure that the client can allocate
- /// 'allocation_len' bytes without pinned bytes plus dirty unpinned bytes exceeding the
- /// client's reservation. No page or client locks should be held by the caller.
- Status CleanPagesBeforeAllocation(
- ReservationTracker* reservation, int64_t allocation_len);
-
- /// Same as CleanPagesBeforeAllocation(), except 'lock_' must be held by 'client_lock'.
- /// 'client_lock' may be released temporarily while waiting for writes to complete.
- Status CleanPagesBeforeAllocationLocked(boost::unique_lock<boost::mutex>* client_lock,
- ReservationTracker* reservation, int64_t allocation_len);
-
- /// Initiates asynchronous writes of dirty unpinned pages to disk. Ensures that at
- /// least 'min_bytes_to_write' bytes of writes will be written asynchronously. May
- /// start writes more aggressively so that I/O and compute can be overlapped. If
- /// any errors are encountered, 'write_status_' is set. 'write_status_' must therefore
- /// be checked before reading back any pages. 'lock_' must be held by the caller.
- void WriteDirtyPagesAsync(int64_t min_bytes_to_write = 0);
+ /// Must be called once before allocating a buffer of 'len' via the AllocateBuffer()
+ /// API to deduct from the client's reservation and update internal accounting. Cleans
+ /// dirty pages if needed to satisfy the buffer pool's internal invariants. No page or
+ /// client locks should be held by the caller.
+ Status PrepareToAllocateBuffer(int64_t len);
+
+ /// Called after a buffer of 'len' is freed via the FreeBuffer() API to update
+ /// internal accounting and release the buffer to the client's reservation. No page or
+ /// client locks should be held by the caller.
+ void FreedBuffer(int64_t len) {
+ boost::lock_guard<boost::mutex> cl(lock_);
+ reservation_.ReleaseTo(len);
+ buffers_allocated_bytes_ -= len;
+ DCHECK_CONSISTENCY();
+ }
/// Wait for the in-flight write for 'page' to complete.
- /// 'lock_' must be held by the caller via 'client_lock'. page->bufffer_lock should
+ /// 'lock_' must be held by the caller via 'client_lock'. page->buffer_lock should
/// not be held.
void WaitForWrite(boost::unique_lock<boost::mutex>* client_lock, Page* page);
@@ -158,17 +250,32 @@ class BufferPool::Client {
private:
// Check consistency of client, DCHECK if inconsistent. 'lock_' must be held.
void DCheckConsistency() {
- DCHECK_GE(in_flight_write_bytes_, 0);
- DCHECK_LE(in_flight_write_bytes_, dirty_unpinned_bytes_);
+ DCHECK_GE(buffers_allocated_bytes_, 0);
+ pinned_pages_.DCheckConsistency();
+ dirty_unpinned_pages_.DCheckConsistency();
+ in_flight_write_pages_.DCheckConsistency();
DCHECK_LE(pinned_pages_.size() + dirty_unpinned_pages_.size()
+ in_flight_write_pages_.size(),
num_pages_);
- if (in_flight_write_pages_.empty()) DCHECK_EQ(0, in_flight_write_bytes_);
- if (in_flight_write_pages_.empty() && dirty_unpinned_pages_.empty()) {
- DCHECK_EQ(0, dirty_unpinned_bytes_);
- }
+ // Check that we flushed enough pages to disk given our eviction policy.
+ DCHECK_GE(reservation_.GetReservation(), buffers_allocated_bytes_
+ + pinned_pages_.bytes() + dirty_unpinned_pages_.bytes()
+ + in_flight_write_pages_.bytes());
}
+ /// Must be called once before allocating or reclaiming a buffer of 'len'. Ensures that
+ /// enough dirty pages are flushed to disk to satisfy the buffer pool's internal
+ /// invariants after the allocation. 'lock_' should be held by the caller via
+ /// 'client_lock'
+ Status CleanPages(boost::unique_lock<boost::mutex>* client_lock, int64_t len);
+
+ /// Initiates asynchronous writes of dirty unpinned pages to disk. Ensures that at
+ /// least 'min_bytes_to_write' bytes of writes will be written asynchronously. May
+ /// start writes more aggressively so that I/O and compute can be overlapped. If
+ /// any errors are encountered, 'write_status_' is set. 'write_status_' must therefore
+ /// be checked before reading back any pages. 'lock_' must be held by the caller.
+ void WriteDirtyPagesAsync(int64_t min_bytes_to_write = 0);
+
/// Called when a write for 'page' completes.
void WriteCompleteCallback(Page* page, const Status& write_status);
@@ -217,61 +324,20 @@ class BufferPool::Client {
/// pages are destroyed before the client.
int64_t num_pages_;
- /// All pinned pages for this client. Only used for debugging.
- InternalList<Page> pinned_pages_;
+ /// Total bytes of buffers in BufferHandles returned to clients (i.e. obtained from
+ /// AllocateBuffer() or ExtractBuffer()).
+ int64_t buffers_allocated_bytes_;
+
+ /// All pinned pages for this client.
+ PageList pinned_pages_;
/// Dirty unpinned pages for this client for which writes are not in flight. Page
/// writes are started in LIFO order, because operators typically have sequential access
/// patterns where the most recently evicted page will be last to be read.
- InternalList<Page> dirty_unpinned_pages_;
+ PageList dirty_unpinned_pages_;
/// Dirty unpinned pages for this client for which writes are in flight.
- InternalList<Page> in_flight_write_pages_;
-
- /// Total bytes of dirty unpinned pages for this client.
- int64_t dirty_unpinned_bytes_;
-
- /// Total bytes of in-flight writes for dirty unpinned pages. Bytes accounted here
- /// are also accounted in 'dirty_unpinned_bytes_'.
- int64_t in_flight_write_bytes_;
-};
-
-/// The internal representation of a page, which can be pinned or unpinned. See the
-/// class comment for explanation of the different page states.
-///
-/// Code manipulating the page is responsible for acquiring 'lock' when reading or
-/// modifying the page.
-struct BufferPool::Page : public InternalList<Page>::Node {
- Page(Client* client, int64_t len) : client(client), len(len), pin_count(0) {}
-
- std::string DebugString();
-
- // Helper for BufferPool::DebugString().
- static bool DebugStringCallback(std::stringstream* ss, BufferPool::Page* page);
-
- /// The client that the page belongs to.
- Client* const client;
-
- /// The length of the page in bytes.
- const int64_t len;
-
- /// The pin count of the page. Only accessed in contexts that are passed the associated
- /// PageHandle, so it cannot be accessed by multiple threads concurrently.
- int pin_count;
-
- /// Non-null if there is a write in flight, the page is clean, or the page is evicted.
- std::unique_ptr<TmpFileMgr::WriteHandle> write_handle;
-
- /// Condition variable signalled when a write for this page completes. Protected by
- /// client->lock_.
- ConditionVariable write_complete_cv_;
-
- /// This lock must be held when accessing 'buffer' if the page is unpinned and not
- /// evicted (i.e. it is safe to access 'buffer' if the page is pinned or evicted).
- SpinLock buffer_lock;
-
- /// Buffer with the page's contents. Closed only iff page is evicted. Open otherwise.
- BufferHandle buffer;
+ PageList in_flight_write_pages_;
};
}
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/874d20d0/be/src/runtime/bufferpool/buffer-pool-test.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/bufferpool/buffer-pool-test.cc b/be/src/runtime/bufferpool/buffer-pool-test.cc
index d4640ac..c71ab60 100644
--- a/be/src/runtime/bufferpool/buffer-pool-test.cc
+++ b/be/src/runtime/bufferpool/buffer-pool-test.cc
@@ -636,6 +636,38 @@ TEST_F(BufferPoolTest, EvictPageDifferentClient) {
pool.FreeBuffer(&clients[1], &buffer);
for (BufferPool::ClientHandle& client : clients) pool.DeregisterClient(&client);
}
+
+/// Regression test for IMPALA-5113 where the page flushing invariant didn't correctly
+/// take multiply pinned pages into account.
+TEST_F(BufferPoolTest, MultiplyPinnedPageAccounting) {
+ const int NUM_BUFFERS = 3;
+ const int64_t TOTAL_BYTES = NUM_BUFFERS * TEST_BUFFER_LEN;
+ global_reservations_.InitRootTracker(NULL, TOTAL_BYTES);
+ BufferPool pool(TEST_BUFFER_LEN, TOTAL_BYTES);
+
+ BufferPool::ClientHandle client;
+ RuntimeProfile* profile = NewProfile();
+ ASSERT_OK(pool.RegisterClient("test client", NewFileGroup(), &global_reservations_,
+ NULL, TOTAL_BYTES, profile, &client));
+ ASSERT_TRUE(client.IncreaseReservation(TOTAL_BYTES));
+
+ BufferPool::PageHandle handle1, handle2;
+ BufferPool::BufferHandle buffer;
+ ASSERT_OK(pool.CreatePage(&client, TEST_BUFFER_LEN, &handle1));
+ ASSERT_OK(pool.CreatePage(&client, TEST_BUFFER_LEN, &handle2));
+ pool.Unpin(&client, &handle1);
+ ASSERT_OK(pool.Pin(&client, &handle2));
+ ASSERT_OK(pool.AllocateBuffer(&client, TEST_BUFFER_LEN, &buffer));
+
+ // We shouldn't need to flush anything to disk since we have only three pages/buffers in
+ // memory. Rely on DCHECKs to check invariants and check we didn't evict the page.
+ EXPECT_FALSE(IsEvicted(&handle1)) << handle1.DebugString();
+
+ pool.DestroyPage(&client, &handle1);
+ pool.DestroyPage(&client, &handle2);
+ pool.FreeBuffer(&client, &buffer);
+ pool.DeregisterClient(&client);
+}
}
int main(int argc, char** argv) {
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/874d20d0/be/src/runtime/bufferpool/buffer-pool.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/bufferpool/buffer-pool.cc b/be/src/runtime/bufferpool/buffer-pool.cc
index 2503beb..482db10 100644
--- a/be/src/runtime/bufferpool/buffer-pool.cc
+++ b/be/src/runtime/bufferpool/buffer-pool.cc
@@ -127,12 +127,8 @@ Status BufferPool::CreatePage(ClientHandle* client, int64_t len, PageHandle* han
BufferHandle buffer;
// No changes have been made to state yet, so we can cleanly return on error.
RETURN_IF_ERROR(AllocateBuffer(client, len, &buffer));
-
- Page* page = new Page(client->impl_, len);
- page->buffer = std::move(buffer);
+ Page* page = client->impl_->CreatePinnedPage(move(buffer));
handle->Open(page, client);
- page->pin_count++;
- client->impl_->AddNewPinnedPage(page);
return Status::OK();
}
@@ -178,7 +174,7 @@ void BufferPool::Unpin(ClientHandle* client, PageHandle* handle) {
reservation->ReleaseTo(page->len);
if (--page->pin_count > 0) return;
- client->impl_->MoveToDirtyUnpinned(reservation->GetUnusedReservation(), page);
+ client->impl_->MoveToDirtyUnpinned(page);
COUNTER_ADD(client->impl_->counters().total_unpinned_bytes, handle->len());
COUNTER_ADD(client->impl_->counters().peak_unpinned_bytes, handle->len());
}
@@ -199,10 +195,13 @@ void BufferPool::ExtractBuffer(
Status BufferPool::AllocateBuffer(
ClientHandle* client, int64_t len, BufferHandle* handle) {
- ReservationTracker* reservation = client->impl_->reservation();
- RETURN_IF_ERROR(client->impl_->CleanPagesBeforeAllocation(reservation, len));
- reservation->AllocateFrom(len);
- return AllocateBufferInternal(client, len, handle);
+ RETURN_IF_ERROR(client->impl_->PrepareToAllocateBuffer(len));
+ Status status = AllocateBufferInternal(client, len, handle);
+ if (!status.ok()) {
+ // Allocation failed - update client's accounting to reflect the failure.
+ client->impl_->FreedBuffer(len);
+ }
+ return status;
}
Status BufferPool::AllocateBufferInternal(
@@ -232,8 +231,9 @@ Status BufferPool::AllocateBufferInternal(
void BufferPool::FreeBuffer(ClientHandle* client, BufferHandle* handle) {
if (!handle->is_open()) return; // Should be idempotent.
DCHECK_EQ(client, handle->client_);
- client->impl_->reservation()->ReleaseTo(handle->len_);
+ int64_t len = handle->len_;
FreeBufferInternal(handle);
+ client->impl_->FreedBuffer(len);
}
void BufferPool::FreeBufferInternal(BufferHandle* handle) {
@@ -281,9 +281,7 @@ void BufferPool::AddCleanPage(const unique_lock<mutex>& client_lock, Page* page)
bool BufferPool::RemoveCleanPage(const unique_lock<mutex>& client_lock, Page* page) {
page->client->DCheckHoldsLock(client_lock);
lock_guard<SpinLock> cpl(clean_pages_lock_);
- bool found = clean_pages_.Contains(page);
- if (found) clean_pages_.Remove(page);
- return found;
+ return clean_pages_.Remove(page);
}
Status BufferPool::EvictCleanPages(int64_t bytes_to_evict) {
@@ -347,8 +345,7 @@ BufferPool::Client::Client(BufferPool* pool, TmpFileMgr::FileGroup* file_group,
file_group_(file_group),
name_(name),
num_pages_(0),
- dirty_unpinned_bytes_(0),
- in_flight_write_bytes_(0) {
+ buffers_allocated_bytes_(0) {
reservation_.InitChildTracker(
profile, parent_reservation, mem_tracker, reservation_limit);
counters_.get_buffer_time = ADD_TIMER(profile, "BufferPoolGetBufferTime");
@@ -364,11 +361,19 @@ BufferPool::Client::Client(BufferPool* pool, TmpFileMgr::FileGroup* file_group,
ADD_COUNTER(profile, "BufferPoolTotalUnpinnedBytes", TUnit::BYTES);
}
-void BufferPool::Client::AddNewPinnedPage(Page* page) {
- DCHECK_GT(page->pin_count, 0);
+BufferPool::Page* BufferPool::Client::CreatePinnedPage(BufferHandle&& buffer) {
+ Page* page = new Page(this, buffer.len());
+ page->buffer = move(buffer);
+ page->pin_count = 1;
+
boost::lock_guard<boost::mutex> lock(lock_);
+ // The buffer is transferred to the page so will be accounted for in
+ // pinned_pages_.bytes() instead of buffers_allocated_bytes_.
+ buffers_allocated_bytes_ -= page->len;
pinned_pages_.Enqueue(page);
++num_pages_;
+ DCHECK_CONSISTENCY();
+ return page;
}
void BufferPool::Client::DestroyPageInternal(
@@ -378,12 +383,8 @@ void BufferPool::Client::DestroyPageInternal(
// Remove the page from the list that it is currently present in (if any).
{
unique_lock<mutex> cl(lock_);
- if (pinned_pages_.Contains(page)) {
- pinned_pages_.Remove(page);
- } else if (dirty_unpinned_pages_.Contains(page)) {
- dirty_unpinned_pages_.Remove(page);
- dirty_unpinned_bytes_ -= page->len;
- } else {
+ // First try to remove from the pinned or dirty unpinned lists.
+ if (!pinned_pages_.Remove(page) && !dirty_unpinned_pages_.Remove(page)) {
// The page either has a write in flight, is clean, or is evicted.
// Let the write complete, if in flight.
WaitForWrite(&cl, page);
@@ -401,6 +402,7 @@ void BufferPool::Client::DestroyPageInternal(
if (out_buffer != NULL) {
DCHECK(page->buffer.is_open());
*out_buffer = std::move(page->buffer);
+ buffers_allocated_bytes_ += out_buffer->len();
} else if (page->buffer.is_open()) {
pool_->FreeBufferInternal(&page->buffer);
}
@@ -408,15 +410,15 @@ void BufferPool::Client::DestroyPageInternal(
handle->Reset();
}
-void BufferPool::Client::MoveToDirtyUnpinned(int64_t unused_reservation, Page* page) {
+void BufferPool::Client::MoveToDirtyUnpinned(Page* page) {
// Only valid to unpin pages if spilling is enabled.
DCHECK(spilling_enabled());
DCHECK_EQ(0, page->pin_count);
unique_lock<mutex> lock(lock_);
+ DCHECK_CONSISTENCY();
DCHECK(pinned_pages_.Contains(page));
pinned_pages_.Remove(page);
dirty_unpinned_pages_.Enqueue(page);
- dirty_unpinned_bytes_ += page->len;
// Check if we should initiate writes for this (or another) dirty page.
WriteDirtyPagesAsync();
@@ -425,6 +427,7 @@ void BufferPool::Client::MoveToDirtyUnpinned(int64_t unused_reservation, Page* p
Status BufferPool::Client::MoveToPinned(ClientHandle* client, PageHandle* handle) {
Page* page = handle->page_;
unique_lock<mutex> cl(lock_);
+ DCHECK_CONSISTENCY();
// Propagate any write errors that occurred for this client.
RETURN_IF_ERROR(write_status_);
@@ -436,13 +439,15 @@ Status BufferPool::Client::MoveToPinned(ClientHandle* client, PageHandle* handle
lock_guard<SpinLock> pl(page->buffer_lock);
evicted = !page->buffer.is_open();
}
- if (evicted) return MoveEvictedToPinned(&cl, client, handle);
+ if (evicted) {
+ // We may need to clean some pages to allocate a buffer for the evicted page.
+ RETURN_IF_ERROR(CleanPages(&cl, page->len));
+ return MoveEvictedToPinned(&cl, client, handle);
+ }
- if (dirty_unpinned_pages_.Contains(page)) {
+ if (dirty_unpinned_pages_.Remove(page)) {
// No writes were initiated for the page - just move it back to the pinned state.
- dirty_unpinned_pages_.Remove(page);
pinned_pages_.Enqueue(page);
- dirty_unpinned_bytes_ -= page->len;
return Status::OK();
}
if (in_flight_write_pages_.Contains(page)) {
@@ -451,6 +456,10 @@ Status BufferPool::Client::MoveToPinned(ClientHandle* client, PageHandle* handle
WaitForWrite(&cl, page);
RETURN_IF_ERROR(write_status_); // The write may have set 'write_status_'.
}
+
+ // At this point we need to either reclaim a clean page or allocate a new buffer.
+ // We may need to clean some pages to do so.
+ RETURN_IF_ERROR(CleanPages(&cl, page->len));
if (pool_->RemoveCleanPage(cl, page)) {
// The clean page still has an associated buffer. Just clean up the write, restore
// the data, and move the page back to the pinned state.
@@ -471,8 +480,6 @@ Status BufferPool::Client::MoveEvictedToPinned(
unique_lock<mutex>* client_lock, ClientHandle* client, PageHandle* handle) {
Page* page = handle->page_;
DCHECK(!page->buffer.is_open());
- RETURN_IF_ERROR(CleanPagesBeforeAllocationLocked(
- client_lock, client->impl_->reservation(), page->len));
// Don't hold any locks while allocating or reading back the data. It is safe to modify
// the page's buffer handle without holding any locks because no concurrent operations
@@ -490,35 +497,43 @@ Status BufferPool::Client::MoveEvictedToPinned(
file_group_->DestroyWriteHandle(move(page->write_handle));
client_lock->lock();
pinned_pages_.Enqueue(page);
+ DCHECK_CONSISTENCY();
return Status::OK();
}
-Status BufferPool::Client::CleanPagesBeforeAllocation(
- ReservationTracker* reservation, int64_t allocation_len) {
+Status BufferPool::Client::PrepareToAllocateBuffer(int64_t len) {
unique_lock<mutex> lock(lock_);
- return CleanPagesBeforeAllocationLocked(&lock, reservation, allocation_len);
+ // Clean enough pages to allow allocation to proceed without violating our eviction
+ // policy. This can fail, so only update the accounting once success is ensured.
+ RETURN_IF_ERROR(CleanPages(&lock, len));
+ reservation_.AllocateFrom(len);
+ buffers_allocated_bytes_ += len;
+ DCHECK_CONSISTENCY();
+ return Status::OK();
}
-Status BufferPool::Client::CleanPagesBeforeAllocationLocked(
- unique_lock<mutex>* client_lock, ReservationTracker* reservation,
- int64_t allocation_len) {
+Status BufferPool::Client::CleanPages(unique_lock<mutex>* client_lock, int64_t len) {
DCheckHoldsLock(*client_lock);
- int64_t unused_reservation = reservation->GetUnusedReservation();
- DCHECK_LE(allocation_len, unused_reservation);
- int64_t unused_reservation_after_alloc = unused_reservation - allocation_len;
+ DCHECK_CONSISTENCY();
+ // Work out what we need to get bytes of dirty unpinned + in flight pages down to
+ // in order to satisfy the eviction policy.
+ int64_t target_dirty_bytes = reservation_.GetReservation() - buffers_allocated_bytes_
+ - pinned_pages_.bytes() - len;
// Start enough writes to ensure that the loop condition below will eventually become
// false (or a write error will be encountered).
- int64_t min_in_flight_bytes = dirty_unpinned_bytes_ - unused_reservation_after_alloc;
- WriteDirtyPagesAsync(max<int64_t>(0, min_in_flight_bytes - in_flight_write_bytes_));
+ int64_t min_bytes_to_write =
+ max<int64_t>(0, dirty_unpinned_pages_.bytes() - target_dirty_bytes);
+ WriteDirtyPagesAsync(min_bytes_to_write);
// One of the writes we initiated, or an earlier in-flight write may have hit an error.
RETURN_IF_ERROR(write_status_);
- // Wait until enough writes have finished that the allocation plus dirty pages won't
- // exceed our reservation. I.e. so that other clients can immediately get the allocated
+ // Wait until enough writes have finished so that we can make the allocation without
+ // violating the eviction policy. I.e. so that other clients can immediately get the
// memory they're entitled to without waiting for this client's write to complete.
- DCHECK_GE(in_flight_write_bytes_, min_in_flight_bytes);
- while (dirty_unpinned_bytes_ > unused_reservation_after_alloc) {
+ DCHECK_GE(in_flight_write_pages_.bytes(), min_bytes_to_write);
+ while (dirty_unpinned_pages_.bytes() + in_flight_write_pages_.bytes()
+ > target_dirty_bytes) {
SCOPED_TIMER(counters().write_wait_time);
write_complete_cv_.Wait(*client_lock);
RETURN_IF_ERROR(write_status_); // Check if error occurred while waiting.
@@ -528,19 +543,17 @@ Status BufferPool::Client::CleanPagesBeforeAllocationLocked(
void BufferPool::Client::WriteDirtyPagesAsync(int64_t min_bytes_to_write) {
DCHECK_GE(min_bytes_to_write, 0);
- DCheckConsistency();
+ DCHECK_LE(min_bytes_to_write, dirty_unpinned_pages_.bytes());
if (file_group_ == NULL) {
// Spilling disabled - there should be no unpinned pages to write.
DCHECK_EQ(0, min_bytes_to_write);
- DCHECK_EQ(0, dirty_unpinned_bytes_);
+ DCHECK_EQ(0, dirty_unpinned_pages_.bytes());
return;
}
// No point in starting writes if an error occurred because future operations for the
// client will fail regardless.
if (!write_status_.ok()) return;
- const int64_t writeable_bytes = dirty_unpinned_bytes_ - in_flight_write_bytes_;
- DCHECK_LE(min_bytes_to_write, writeable_bytes);
// Compute the ideal amount of writes to start. We use a simple heuristic based on the
// total number of writes. The FileGroup's allocation should spread the writes across
// disks somewhat, but doesn't guarantee we're fully using all available disks. In
@@ -549,7 +562,7 @@ void BufferPool::Client::WriteDirtyPagesAsync(int64_t min_bytes_to_write) {
* file_group_->tmp_file_mgr()->NumActiveTmpDevices();
int64_t bytes_written = 0;
- while (bytes_written < writeable_bytes
+ while (!dirty_unpinned_pages_.empty()
&& (bytes_written < min_bytes_to_write
|| in_flight_write_pages_.size() < target_writes)) {
Page* page = dirty_unpinned_pages_.tail(); // LIFO.
@@ -577,7 +590,6 @@ void BufferPool::Client::WriteDirtyPagesAsync(int64_t min_bytes_to_write) {
DCHECK_EQ(tmp, page);
in_flight_write_pages_.Enqueue(page);
bytes_written += page->len;
- in_flight_write_bytes_ += page->len;
}
}
@@ -593,8 +605,6 @@ void BufferPool::Client::WriteCompleteCallback(Page* page, const Status& write_s
// repurposed by other clients and 'write_status_' must be checked by this client
// before reading back the bad data.
pool_->AddCleanPage(cl, page);
- dirty_unpinned_bytes_ -= page->len;
- in_flight_write_bytes_ -= page->len;
WriteDirtyPagesAsync(); // Start another asynchronous write if needed.
// Notify before releasing lock to avoid race with Page and Client destruction.
@@ -614,10 +624,12 @@ void BufferPool::Client::WaitForWrite(unique_lock<mutex>* client_lock, Page* pag
string BufferPool::Client::DebugString() {
lock_guard<mutex> lock(lock_);
stringstream ss;
- ss << Substitute("<BufferPool::Client> $0 name: $1 write_status: $2 num_pages: $3 "
- "dirty_unpinned_bytes: $4 in_flight_write_bytes: $5 reservation: {$6}",
- this, name_, write_status_.GetDetail(), num_pages_, dirty_unpinned_bytes_,
- in_flight_write_bytes_, reservation_.DebugString());
+ ss << Substitute("<BufferPool::Client> $0 name: $1 write_status: $2 "
+ "buffers allocated $3 num_pages: $4 pinned_bytes: $5 "
+ "dirty_unpinned_bytes: $6 in_flight_write_bytes: $7 reservation: {$8}",
+ this, name_, write_status_.GetDetail(), buffers_allocated_bytes_, num_pages_,
+ pinned_pages_.bytes(), dirty_unpinned_pages_.bytes(),
+ in_flight_write_pages_.bytes(), reservation_.DebugString());
ss << "\n " << pinned_pages_.size() << " pinned pages: ";
pinned_pages_.Iterate(bind<bool>(Page::DebugStringCallback, &ss, _1));
ss << "\n " << dirty_unpinned_pages_.size() << " dirty unpinned pages: ";
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/874d20d0/be/src/runtime/bufferpool/buffer-pool.h
----------------------------------------------------------------------
diff --git a/be/src/runtime/bufferpool/buffer-pool.h b/be/src/runtime/bufferpool/buffer-pool.h
index b199637..da8ef3c 100644
--- a/be/src/runtime/bufferpool/buffer-pool.h
+++ b/be/src/runtime/bufferpool/buffer-pool.h
@@ -248,6 +248,7 @@ class BufferPool : public CacheLineAligned {
private:
DISALLOW_COPY_AND_ASSIGN(BufferPool);
+ class PageList;
struct Page;
/// Allocate a buffer of length 'len'. Assumes that the client's reservation has already
/// been consumed for the buffer. Returns an error if the pool is unable to fulfill the
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/874d20d0/be/src/util/internal-queue.h
----------------------------------------------------------------------
diff --git a/be/src/util/internal-queue.h b/be/src/util/internal-queue.h
index 5e32116..b295b1f 100644
--- a/be/src/util/internal-queue.h
+++ b/be/src/util/internal-queue.h
@@ -154,11 +154,10 @@ class InternalQueueBase {
}
/// Removes 'node' from the queue. This is O(1). No-op if node is
- /// not on the list.
- void Remove(T* n) {
+ /// not on the list. Returns true if removed
+ bool Remove(T* n) {
Node* node = (Node*)n;
- if (node->parent_queue == NULL) return;
- DCHECK(node->parent_queue == this);
+ if (node->parent_queue != this) return false;
{
boost::lock_guard<LockType> lock(lock_);
if (node->next == NULL && node->prev == NULL) {
@@ -168,7 +167,7 @@ class InternalQueueBase {
head_ = tail_ = NULL;
--size_;
node->parent_queue = NULL;
- return;
+ return true;
}
if (head_ == node) {
@@ -189,6 +188,7 @@ class InternalQueueBase {
}
node->next = node->prev = NULL;
node->parent_queue = NULL;
+ return true;
}
/// Clears all elements in the list.
[3/5] incubator-impala git commit: IMPALA-4673: Use
--local_library_dir for tzdb startup scratch space
Posted by ta...@apache.org.
IMPALA-4673: Use --local_library_dir for tzdb startup scratch space
Currently we hardcode the the path to /tmp which may not be
accessible on certain environments.
Change-Id: I7ab0238e6ce65ca39c167349e3b79aa70fb39d2f
Reviewed-on: http://gerrit.cloudera.org:8080/5523
Reviewed-by: Bharath Vissapragada <bh...@cloudera.com>
Tested-by: Impala Public Jenkins
Project: http://git-wip-us.apache.org/repos/asf/incubator-impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-impala/commit/77dbdc3f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-impala/tree/77dbdc3f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-impala/diff/77dbdc3f
Branch: refs/heads/master
Commit: 77dbdc3f572260a90af174dcb9112cb991ec296c
Parents: 8e45c4a
Author: Bharath Vissapragada <bh...@cloudera.com>
Authored: Thu Dec 15 12:03:16 2016 -0800
Committer: Impala Public Jenkins <im...@gerrit.cloudera.org>
Committed: Mon Mar 27 21:53:04 2017 +0000
----------------------------------------------------------------------
be/src/common/init.cc | 6 ++++--
be/src/exprs/expr-test.cc | 2 ++
be/src/exprs/timezone_db.cc | 15 ++++++++++-----
be/src/exprs/timezone_db.h | 2 +-
be/src/runtime/lib-cache.cc | 3 +--
be/src/service/impalad-main.cc | 2 ++
6 files changed, 20 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/77dbdc3f/be/src/common/init.cc
----------------------------------------------------------------------
diff --git a/be/src/common/init.cc b/be/src/common/init.cc
index 1e90f53..16f0f41 100644
--- a/be/src/common/init.cc
+++ b/be/src/common/init.cc
@@ -24,7 +24,6 @@
#include "common/status.h"
#include "exec/kudu-util.h"
#include "exprs/expr.h"
-#include "exprs/timezone_db.h"
#include "gutil/atomicops.h"
#include "rpc/authentication.h"
#include "rpc/thrift-util.h"
@@ -76,6 +75,10 @@ DEFINE_int64(pause_monitor_warn_threshold_ms, 10000, "If the pause monitor sleep
"more than this time period, a warning is logged. If set to 0 or less, pause monitor"
" is disabled.");
+DEFINE_string(local_library_dir, "/tmp",
+ "Scratch space for local fs operations. Currently used for copying "
+ "UDF binaries locally from HDFS and also for initializing the timezone db");
+
// Defined by glog. This allows users to specify the log level using a glob. For
// example -vmodule=*scanner*=3 would enable full logging for scanners. If redaction
// is enabled, this option won't be allowed because some logging dumps table data
@@ -198,7 +201,6 @@ void impala::InitCommonRuntime(int argc, char** argv, bool init_jvm,
impala::InitThreading();
impala::TimestampParser::Init();
impala::SeedOpenSSLRNG();
- ABORT_IF_ERROR(impala::TimezoneDatabase::Initialize());
ABORT_IF_ERROR(impala::InitAuth(argv[0]));
// Initialize maintenance_thread after InitGoogleLoggingSafe and InitThreading.
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/77dbdc3f/be/src/exprs/expr-test.cc
----------------------------------------------------------------------
diff --git a/be/src/exprs/expr-test.cc b/be/src/exprs/expr-test.cc
index 0ca2da2..0de6833 100644
--- a/be/src/exprs/expr-test.cc
+++ b/be/src/exprs/expr-test.cc
@@ -38,6 +38,7 @@
#include "exprs/null-literal.h"
#include "exprs/string-functions.h"
#include "exprs/timestamp-functions.h"
+#include "exprs/timezone_db.h"
#include "gen-cpp/Exprs_types.h"
#include "gen-cpp/hive_metastore_types.h"
#include "rpc/thrift-client.h"
@@ -7072,6 +7073,7 @@ TEST_F(ExprTest, UuidTest) {
int main(int argc, char **argv) {
::testing::InitGoogleTest(&argc, argv);
InitCommonRuntime(argc, argv, true, TestInfo::BE_TEST);
+ ABORT_IF_ERROR(TimezoneDatabase::Initialize());
InitFeSupport(false);
impala::LlvmCodeGen::InitializeLlvm();
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/77dbdc3f/be/src/exprs/timezone_db.cc
----------------------------------------------------------------------
diff --git a/be/src/exprs/timezone_db.cc b/be/src/exprs/timezone_db.cc
index c7ae888..110d23b 100644
--- a/be/src/exprs/timezone_db.cc
+++ b/be/src/exprs/timezone_db.cc
@@ -19,6 +19,7 @@
#include <boost/date_time/compiler_config.hpp>
#include <boost/date_time/local_time/posix_time_zone.hpp>
+#include <boost/filesystem.hpp>
#include "exprs/timestamp-functions.h"
#include "gutil/strings/substitute.h"
@@ -29,6 +30,8 @@ using boost::local_time::tz_database;
using boost::local_time::time_zone_ptr;
using boost::local_time::posix_time_zone;
+DECLARE_string(local_library_dir);
+
namespace impala {
tz_database TimezoneDatabase::tz_database_;
@@ -666,14 +669,16 @@ const char* TimezoneDatabase::TIMEZONE_DATABASE_STR = "\"ID\",\"STD ABBR\",\"STD
Status TimezoneDatabase::Initialize() {
// Create a temporary file and write the timezone information. The boost
- // interface only loads this format from a file. We don't want to raise
- // an error here since this is done when the backend is created and this
- // information might not actually get used by any queries.
- char filestr[] = "/tmp/impala.tzdb.XXXXXXX";
+ // interface only loads this format from a file. We abort the startup if
+ // this initialization fails for some reason.
+ char *filestr = const_cast<char*>((boost::filesystem::path(FLAGS_local_library_dir)
+ / string("impala.tzdb.XXXXXXX")).string().c_str());
FILE* file;
int fd;
if ((fd = mkstemp(filestr)) == -1) {
- return Status(Substitute("Could not create temporary timezone file: $0", filestr));
+ return Status(Substitute("Could not create temporary timezone file: $0. Check that "
+ "the directory $1 is writable by the user running Impala.", filestr,
+ FLAGS_local_library_dir));
}
if ((file = fopen(filestr, "w")) == NULL) {
unlink(filestr);
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/77dbdc3f/be/src/exprs/timezone_db.h
----------------------------------------------------------------------
diff --git a/be/src/exprs/timezone_db.h b/be/src/exprs/timezone_db.h
index 20ea205..3a1178a 100644
--- a/be/src/exprs/timezone_db.h
+++ b/be/src/exprs/timezone_db.h
@@ -32,7 +32,7 @@ namespace impala {
class TimezoneDatabase {
public:
/// Set up the static timezone database.
- static Status Initialize();
+ static Status Initialize() WARN_UNUSED_RESULT;
/// Converts the name of a timezone to a boost timezone object. In some cases, the
/// timestamp is required to determine the timezone because occasionally timezone
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/77dbdc3f/be/src/runtime/lib-cache.cc
----------------------------------------------------------------------
diff --git a/be/src/runtime/lib-cache.cc b/be/src/runtime/lib-cache.cc
index 9a8d873..d694c49 100644
--- a/be/src/runtime/lib-cache.cc
+++ b/be/src/runtime/lib-cache.cc
@@ -33,8 +33,7 @@
namespace filesystem = boost::filesystem;
-DEFINE_string(local_library_dir, "/tmp",
- "Local directory to copy UDF libraries from HDFS into");
+DECLARE_string(local_library_dir);
namespace impala {
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/77dbdc3f/be/src/service/impalad-main.cc
----------------------------------------------------------------------
diff --git a/be/src/service/impalad-main.cc b/be/src/service/impalad-main.cc
index 341399c..4848b45 100644
--- a/be/src/service/impalad-main.cc
+++ b/be/src/service/impalad-main.cc
@@ -27,6 +27,7 @@
#include "exec/hbase-table-scanner.h"
#include "exec/hbase-table-writer.h"
#include "exprs/hive-udf-call.h"
+#include "exprs/timezone_db.h"
#include "runtime/hbase-table.h"
#include "codegen/llvm-codegen.h"
#include "common/status.h"
@@ -59,6 +60,7 @@ DECLARE_bool(enable_rm);
int ImpaladMain(int argc, char** argv) {
InitCommonRuntime(argc, argv, true);
+ ABORT_IF_ERROR(TimezoneDatabase::Initialize());
ABORT_IF_ERROR(LlvmCodeGen::InitializeLlvm());
JniUtil::InitLibhdfs();
ABORT_IF_ERROR(HBaseTableScanner::Init());